diff --git a/test/integration/export_test.py b/test/integration/export_test.py index cc9f9a0..11bc812 100644 --- a/test/integration/export_test.py +++ b/test/integration/export_test.py @@ -1,3 +1,4 @@ +import io import csv import pytest import pyexasol @@ -51,7 +52,7 @@ def table(connection, empty_table, data): name = empty_table stmt = "INSERT INTO {table} VALUES ({{col1}}, {{col2}});" for col1, col2 in data: - connection.execute(stmt.format(table=name), {"col1": col1, "col2":col2}) + connection.execute(stmt.format(table=name), {"col1": col1, "col2": col2}) connection.commit() yield name @@ -61,34 +62,60 @@ def export_file(tmp_path, data): yield tmp_path / "names.csv" -@pytest.mark.etl -def test_export_with_column_names(): - assert False +@pytest.fixture +def csv_dialect(): + class PyexasolCsvDialect(csv.Dialect): + lineterminator = "\n" + delimiter = "," + quoting = csv.QUOTE_MINIMAL + quotechar = '"' + + yield PyexasolCsvDialect() + + +@pytest.fixture +def expected_csv(csv_dialect): + def create_csv(table, data, **params): + csvfile = io.StringIO() + if "with_column_names" in params and params["with_column_names"]: + data = [("FIRSTNAME", "LASTNAME")] + data + writer = csv.writer(csvfile, csv_dialect) + writer.writerows(data) + return csvfile.getvalue() + + yield create_csv @pytest.mark.etl -def test_skip_rows_in_export(): - assert False +def test_export_with_column_names(connection, table, data, export_file, expected_csv): + params = {"with_column_names": True} + connection.export_to_file(export_file, table, export_params=params) + + expected = expected_csv(table, data, **params) + actual = export_file.read_text() + + assert actual == expected @pytest.mark.etl -def test_custom_export_callback(connection, table, data, export_file): +def test_custom_export_callback(connection, table, data, export_file, expected_csv): def export_cb(pipe, dst): dst.write_bytes(pipe.read()) connection.export_to_callback(export_cb, export_file, table) - expected = 0 # TBD - actual = None # TBD + expected = expected_csv(table, data) + actual = export_file.read_text() assert actual == expected @pytest.mark.etl -def test_export_with_reodered_columns(): - assert False +def test_export_csv_cols(connection, table, data, export_file, expected_csv): + params = {"csv_cols": ["1..2", "2 foo_LASTNAME"]} + connection.export_to_file(export_file, table, export_params=params) + expected = expected_csv(table, data, **params) + actual = export_file.read_text() -@pytest.mark.etl -def test_export_with_custom_csv_format(): - assert False + assert actual == expected