Update export tests

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