 (hugsql/def-db-fns "akvo/lumen/lib/job-execution.sql")
 (hugsql/def-db-fns "akvo/lumen/lib/transformation_test.sql")
 (hugsql/def-db-fns "akvo/lumen/lib/transformation.sql")
+
+(defn latest-data [dataset-id]
+  (let [table-name (:table-name
+                    (latest-dataset-version-by-dataset-id *tenant-conn* {:dataset-id dataset-id}))]
+    (get-data *tenant-conn* {:table-name table-name})))
 
 (def ops (vec (json/parse-string (slurp (io/resource "ops.json")))))
 


(deftest ^:functional test-import-and-transform
  (testing "Import CSV and transform"
    (let [t-log [(gen-transformation :core/trim {:akvo.lumen.specs.db.dataset-version.column/columnName "c5"
                                                 :akvo.lumen.specs.transformation.engine/onError "fail"})
                 (gen-transformation :core/change-datatype {:akvo.lumen.specs.db.dataset-version.column/columnName "c5"
                                                        :akvo.lumen.specs.transformation.change-datatype/defaultValue 0
                                                        :akvo.lumen.specs.transformation.change-datatype/newType "number"
                                                        :akvo.lumen.specs.transformation.engine/onError "default-value"})
                 (gen-transformation :core/filter-column {:akvo.lumen.specs.db.dataset-version.column/columnName "c4"
                                                      :akvo.lumen.specs.transformation.filter-column/expression {"contains" "broken"}
                                                      :akvo.lumen.specs.transformation.engine/onError "fail"})]
          dataset-id (import-file *tenant-conn* *error-tracker* {:name "GDP Test"
                                                                 :has-column-headers? false
                                                                 :file "GDP.csv"})]
      (let [[tag {:strs [datasetId]}] (last (for [transformation t-log]
                                              (transformation/apply {:tenant-conn *tenant-conn*}
                                                        dataset-id
                                                        {:type :transformation
                                                         :transformation transformation})))]

        (is (= ::lib/ok tag))

        (let [table-name (:table-name (latest-dataset-version-by-dataset-id *tenant-conn*
                                                                            {:dataset-id datasetId}))]
          (is (zero? (:c5 (get-val-from-table *tenant-conn*
                                              {:rnum 196
                                               :column-name "c5"
                                               :table-name table-name}))))
          (is (= "[Broken]" (:c4 (get-val-from-table *tenant-conn* {:rnum 196
                                                                    :column-name "c4"
                                                                    :table-name table-name}))))
          (is (= 1 (:total (get-row-count *tenant-conn* {:table-name table-name})))))))))

(deftest ^:functional test-undo

                                                       :akvo.lumen.specs.transformation.change-datatype/defaultValue 0
                                                       :akvo.lumen.specs.transformation.change-datatype/newType "date"
                                                       :akvo.lumen.specs.transformation.change-datatype/parseFormat format*
                                                       ::transformation.engine.s/onError "fail"})})
        data                (import.s/sample-imported-dataset [:text
                                                          [:text {::import.column.text.s/value (fn [] (import.column.s/date-format-gen
                                                                                         (fn [[y _ _ :as date]]
