Changeset View
Changeset View
Standalone View
Standalone View
swh/core/db/tests/test_db.py
| Show First 20 Lines • Show All 284 Lines • ▼ Show 20 Lines | def test_db_copy_to_static(db_with_data): | ||||
| db_with_data.copy_to(items, "test_table", COLUMNS) | db_with_data.copy_to(items, "test_table", COLUMNS) | ||||
| with db_with_data.cursor() as cur: | with db_with_data.cursor() as cur: | ||||
| cur.execute("select * from test_table;") | cur.execute("select * from test_table;") | ||||
| output = convert_lines(cur) | output = convert_lines(cur) | ||||
| assert len(output) == 1 | assert len(output) == 1 | ||||
| assert EXPECTED_ROW_OUT == output[0] | assert EXPECTED_ROW_OUT == output[0] | ||||
| @settings(suppress_health_check=function_scoped_fixture_check) | @settings(suppress_health_check=function_scoped_fixture_check, max_examples=5) | ||||
| @given(db_rows) | @given(db_rows) | ||||
| def test_db_copy_to(db_with_data, data): | def test_db_copy_to(db_with_data, data): | ||||
| items = [dict(zip(COLUMNS, item)) for item in data] | items = [dict(zip(COLUMNS, item)) for item in data] | ||||
| with db_with_data.cursor() as cur: | with db_with_data.cursor() as cur: | ||||
| cur.execute("TRUNCATE TABLE test_table CASCADE") | cur.execute("TRUNCATE TABLE test_table CASCADE") | ||||
| db_with_data.copy_to(items, "test_table", COLUMNS) | db_with_data.copy_to(items, "test_table", COLUMNS) | ||||
| ▲ Show 20 Lines • Show All 165 Lines • Show Last 20 Lines | |||||