From 152416540f2b67c7888564ae0efa80c8f096fd92 Mon Sep 17 00:00:00 2001 From: Kalan MacRow Date: Tue, 12 Aug 2025 12:10:36 -0400 Subject: [PATCH 1/7] more thorough integration tests for pg, redshift, snowflake, bq --- data-transfer/pontoon/Makefile | 3 + data-transfer/pontoon/pontoon/__init__.py | 4 + data-transfer/pontoon/pontoon/base.py | 56 ++- .../destination/bigquery_destination.py | 38 +- .../destination/postgres_destination.py | 45 ++- .../destination/redshift_destination.py | 17 +- .../destination/snowflake_destination.py | 17 +- .../snowflake_storage_destination.py | 20 +- .../pontoon/destination/sql_destination.py | 12 +- .../pontoon/pontoon/source/memory_source.py | 143 +++++++- .../pontoon/pontoon/source/sql_source.py | 48 ++- .../pontoon/tests/integration/common.py | 57 +++ .../integration/test_bigquery_connectors.py | 251 ++++++++++++++ .../tests/integration/test_core_connectors.py | 324 ------------------ .../integration/test_postgres_connectors.py | 259 ++++++++++++++ .../integration/test_redshift_connectors.py | 266 ++++++++++++++ .../integration/test_snowflake_connectors.py | 263 ++++++++++++++ 17 files changed, 1417 insertions(+), 406 deletions(-) create mode 100644 data-transfer/pontoon/tests/integration/common.py create mode 100644 data-transfer/pontoon/tests/integration/test_bigquery_connectors.py delete mode 100644 data-transfer/pontoon/tests/integration/test_core_connectors.py create mode 100644 data-transfer/pontoon/tests/integration/test_postgres_connectors.py create mode 100644 data-transfer/pontoon/tests/integration/test_redshift_connectors.py create mode 100644 data-transfer/pontoon/tests/integration/test_snowflake_connectors.py diff --git a/data-transfer/pontoon/Makefile b/data-transfer/pontoon/Makefile index e20f8b3..e4c9ec6 100644 --- a/data-transfer/pontoon/Makefile +++ b/data-transfer/pontoon/Makefile @@ -7,6 +7,9 @@ build-wheel: test: pytest -s +test-integration: + pytest -s tests/integration/ + install-dev: pip install -e . diff --git a/data-transfer/pontoon/pontoon/__init__.py b/data-transfer/pontoon/pontoon/__init__.py index 99f03d7..1bc452a 100644 --- a/data-transfer/pontoon/pontoon/__init__.py +++ b/data-transfer/pontoon/pontoon/__init__.py @@ -20,6 +20,10 @@ from pontoon.cache.memory_cache import MemoryCache from pontoon.cache.sqlite_cache import SqliteCache from pontoon.base import Namespace, Stream, Record, Dataset, Cache, Mode, Source, Destination +from pontoon.base import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon.base import DestinationConnectionFailed, DestinationStreamInvalidSchema +from pontoon.base import StreamMissingField + __sources = {} __destinations = {} diff --git a/data-transfer/pontoon/pontoon/base.py b/data-transfer/pontoon/pontoon/base.py index ec15731..34141ec 100644 --- a/data-transfer/pontoon/pontoon/base.py +++ b/data-transfer/pontoon/pontoon/base.py @@ -32,6 +32,16 @@ def __init__(self, data:List[Any]): self.data = data + +class StreamError(Exception): + """ Base class for all Stream related exceptions """ + pass + +class StreamMissingField(StreamError): + """ Raised when a stream schema is missing a field """ + pass + + class Stream: """ A class to represent a typed stream of records @@ -108,7 +118,7 @@ def _compute_checksum(self, row:List[Any]) -> str: def _missing_field(self, field_name:str): - raise Exception(f"Stream {self.schema_name}.{self.name} does not have field: {field_name}") + raise StreamMissingField(f"Stream {self.schema_name}.{self.name} does not have field: {field_name}") def with_field(self, field_name:str, field_type:Any, value:Any) -> 'Stream': @@ -135,7 +145,7 @@ def with_version(self, version:str, field_name='pontoon__version') -> 'Stream': def drop_field(self, field_name:str) -> 'Stream': if field_name not in self.schema.names: - raise Exception(f"Stream {self.schema_name}.{self.name} does not have field: {field_name}") + raise StreamMissingField(f"Stream {self.schema_name}.{self.name} does not have field: {field_name}") field_idx = self.schema.get_field_index(field_name) self.schema = self.schema.remove(field_idx) self._drop_fields.append(field_idx) @@ -418,6 +428,7 @@ def check_batch_volume(self, ds:Dataset): pass + class Source(ABC): """ Abstract base class to represent a data source connector """ @@ -441,6 +452,28 @@ def inspect_streams(self): def close(self): pass + +class SourceError(Exception): + """Base exception for all Source-related errors.""" + pass + + +class SourceConnectionFailed(SourceError): + """Raised when the connection to the source fails.""" + pass + + +class SourceStreamDoesNotExist(SourceError): + """Raised when a requested stream does not exist in the source.""" + pass + + +class SourceStreamInvalidSchema(SourceError): + """Raised when the source stream has an invalid or unexpected schema.""" + pass + + + class Destination(ABC): """ Abstract base class to represent a data destination connector """ @@ -459,4 +492,21 @@ def integrity(self) -> Integrity: @abstractmethod def close(self): - pass \ No newline at end of file + pass + + +class DestinationError(Exception): + """Base exception for all Destination-related errors.""" + pass + + +class DestinationConnectionFailed(DestinationError): + """Raised when the connection to the destination fails.""" + pass + + +class DestinationStreamInvalidSchema(DestinationError): + """Raised when the destination stream has an invalid or unexpected schema.""" + pass + + diff --git a/data-transfer/pontoon/pontoon/destination/bigquery_destination.py b/data-transfer/pontoon/pontoon/destination/bigquery_destination.py index 74f849e..1a13f6c 100644 --- a/data-transfer/pontoon/pontoon/destination/bigquery_destination.py +++ b/data-transfer/pontoon/pontoon/destination/bigquery_destination.py @@ -1,8 +1,10 @@ import json from typing import List, Dict, Tuple, Generator, Any -from sqlalchemy import create_engine, text +from sqlalchemy import create_engine, inspect, MetaData, Table, text from pontoon.base import Destination, Dataset, Stream, Record, Progress, Mode +from pontoon.base import DestinationConnectionFailed, DestinationStreamInvalidSchema + from pontoon.source.sql_source import SQLUtil from pontoon.destination.sql_destination import SQLDestination from pontoon.destination.gcs_destination import GCSDestination, GCSConfig @@ -72,6 +74,9 @@ def write(self, ds:Dataset, progress_callback = None): with self._connect() as conn: + insp = inspect(conn) + metadata_obj = MetaData() + for stream in ds.streams: # configure progress tracking @@ -83,6 +88,11 @@ def write(self, ds:Dataset, progress_callback = None): if callable(progress_callback): progress.subscribe(progress_callback) + # Check if there are any records to process + stream_size = ds.size(stream) + if stream_size == 0: + progress.message("No records to process for this stream") + continue # staging and target table names target_table_name = f"{stream.schema_name}.{stream.name}" @@ -100,6 +110,15 @@ def write(self, ds:Dataset, progress_callback = None): ) ) + # delete records depending on sync mode + if self._mode.type == Mode.FULL_REFRESH: + progress.message("Dropping target table") + SQLDestination.drop_table(conn, target_table_name) + + # Drop staging table if it happens to exist from previous failed load + # BQ does not support TEMP tables, so we use a real table - can't assume it will be cleaned up + SQLDestination.drop_table(conn, stage_table_name) + progress.message("Running LOAD from GCS") with conn.begin(): conn.execute(text(load_sql)) @@ -109,12 +128,19 @@ def write(self, ds:Dataset, progress_callback = None): with conn.begin(): conn.execute(text(create_target_sql)) - # delete records depending on sync mode - if self._mode.type == Mode.FULL_REFRESH: - progress.message("Truncating target table") - with conn.begin(): - conn.execute(text(f"DELETE FROM {target_table_name} WHERE 1=1")) + # Check that target and stage schemas are compatible + target_table = Table(stream.name, metadata_obj, schema=stream.schema_name, autoload_with=insp) + target_table_schema = SQLDestination.table_ddl_to_schema(target_table.columns) + + stage_table = Table(f"__temp_{stream.name}", metadata_obj, schema=stream.schema_name, autoload_with=insp) + stage_table_schema = SQLDestination.table_ddl_to_schema(stage_table.columns) + + # Use flexible schema comparison that ignores column order + if not SQLDestination.schemas_compatible(target_table_schema, stage_table_schema): + raise DestinationStreamInvalidSchema(f"Existing schema for stream {stream.name} does not match.") + + # sql to MERGE the staging table into the target table merge_sql = BigQuerySQLUtil.merge( target_table_name, diff --git a/data-transfer/pontoon/pontoon/destination/postgres_destination.py b/data-transfer/pontoon/pontoon/destination/postgres_destination.py index c645e9a..6928cbe 100644 --- a/data-transfer/pontoon/pontoon/destination/postgres_destination.py +++ b/data-transfer/pontoon/pontoon/destination/postgres_destination.py @@ -1,8 +1,14 @@ +import psycopg2 from psycopg2 import sql from psycopg2.extras import execute_values from typing import List, Dict, Tuple, Generator, Any +from sqlalchemy.exc import SQLAlchemyError, OperationalError, NoSuchTableError from pontoon.base import Destination, Dataset, Stream, Record, Progress, Mode +from pontoon.base import DestinationConnectionFailed, \ + DestinationStreamInvalidSchema + + from pontoon.destination.sql_destination import SQLDestination @@ -44,9 +50,9 @@ def create_temp_table(table_name:str, like_table_name:str) -> str: def drop_table(table_name:str): schema, table = PostgresSQLUtil.parse_table(table_name) if schema: - return sql.SQL("DROP TABLE {}.{}").format(schema, table) + return sql.SQL("DROP TABLE IF EXISTS {}.{}").format(schema, table) else: - return sql.SQL("DROP TABLE {}").format(table) + return sql.SQL("DROP TABLE IF EXISTS {}").format(table) @staticmethod @@ -147,17 +153,27 @@ def write(self, ds:Dataset, progress_callback = None): progress.message("No records to process for this stream") continue - with self._connect() as conn: - # create target table for the stream if it doesn't exist - table = SQLDestination.create_table_if_not_exists(conn, stream) - - - # using the raw psycopg2 connection for efficiency - conn = self._engine.raw_connection() - target_table_name = f"{stream.schema_name}.{stream.name}" stage_table_name = f"temp_{stream.schema_name}_{stream.name}" + # using the raw psycopg2 connection for efficiency + try: + conn = self._engine.raw_connection() + except OperationalError as e: + raise DestinationConnectionFailed("Could not connect to destination databse") from e + + # Drop existing table if needed + if self._mode.type == Mode.FULL_REFRESH: + with conn.cursor() as cur: + cur.execute(PostgresSQLUtil.drop_table(target_table_name)) + conn.commit() + + + with self._connect() as base_conn: + # create target table for the stream if it doesn't exist + table = SQLDestination.create_table_if_not_exists(base_conn, stream) + + # temporary staging table create_stage_sql = PostgresSQLUtil.create_temp_table(stage_table_name, target_table_name) @@ -193,15 +209,6 @@ def write(self, ds:Dataset, progress_callback = None): conn.commit() with conn.cursor() as cur: - # delete all records from the table - if self._mode.type == Mode.FULL_REFRESH: - progress.message("Truncating target table") - cur.execute( - sql.SQL("DELETE FROM {}.{}").format( - sql.Identifier(stream.schema_name), - sql.Identifier(stream.name) - ) - ) # upsert staging into target table progress.message("Upserting records into target table") diff --git a/data-transfer/pontoon/pontoon/destination/redshift_destination.py b/data-transfer/pontoon/pontoon/destination/redshift_destination.py index 869ea87..bdabf58 100644 --- a/data-transfer/pontoon/pontoon/destination/redshift_destination.py +++ b/data-transfer/pontoon/pontoon/destination/redshift_destination.py @@ -1,5 +1,5 @@ from typing import List, Dict, Tuple, Generator, Any -from sqlalchemy import text +from sqlalchemy import text, MetaData, Table from pontoon.base import Destination, Dataset, Stream, Record, Progress, Mode from pontoon.source.sql_source import SQLUtil @@ -76,16 +76,21 @@ def write(self, ds:Dataset, progress_callback = None): if callable(progress_callback): progress.subscribe(progress_callback) - # create a table for the stream if it doesn't exist - table = SQLDestination.create_table_if_not_exists(conn, stream) - + # Check if there are any records to process + stream_size = ds.size(stream) + if stream_size == 0: + progress.message("No records to process for this stream") + continue + target_table_name = f"{stream.schema_name}.{stream.name}" stage_table_name = f"temp_{stream.schema_name}_{stream.name}" if self._mode.type == Mode.FULL_REFRESH: with conn.begin(): - # delete all records from the table - conn.execute(table.delete()) + SQLDestination.drop_table(conn, target_table_name) + + # create a table for the stream if it doesn't exist + table = SQLDestination.create_table_if_not_exists(conn, stream) # temporary staging table create_stage_sql = RedshiftSQLUtil.create_temp_table(stage_table_name, target_table_name) diff --git a/data-transfer/pontoon/pontoon/destination/snowflake_destination.py b/data-transfer/pontoon/pontoon/destination/snowflake_destination.py index 9ec1cbf..95e6af3 100644 --- a/data-transfer/pontoon/pontoon/destination/snowflake_destination.py +++ b/data-transfer/pontoon/pontoon/destination/snowflake_destination.py @@ -82,16 +82,21 @@ def write(self, ds:Dataset, progress_callback = None): if callable(progress_callback): progress.subscribe(progress_callback) - # create a table for the stream if it doesn't exist - table = SQLDestination.create_table_if_not_exists(conn, stream) + # Check if there are any records to process + stream_size = ds.size(stream) + if stream_size == 0: + progress.message("No records to process for this stream") + continue target_table_name = f"{stream.schema_name}.{stream.name}" stage_table_name = f"{stream.schema_name}.__temp_{stream.name}" - - # delete records depending on sync mode + + # drop target depending on sync mode if self._mode.type == Mode.FULL_REFRESH: - with conn.begin(): - conn.execute(table.delete()) + SQLDestination.drop_table(conn, target_table_name) + + # create a table for the stream if it doesn't exist + SQLDestination.create_table_if_not_exists(conn, stream) # sql to create staging table stage_table_sql = SnowflakeSQLUtil.create_temp_table( diff --git a/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py b/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py index eaa3347..096fb30 100644 --- a/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py +++ b/data-transfer/pontoon/pontoon/destination/snowflake_storage_destination.py @@ -3,6 +3,7 @@ from datetime import datetime import snowflake.connector from pontoon.base import Namespace, Destination, Stream, Dataset, Record, Progress +from pontoon.base import DestinationConnectionFailed from pontoon.source.sql_source import SQLUtil from pontoon.destination import ObjectStoreBase from pontoon.destination.integrity import SMSIntegrity @@ -27,14 +28,17 @@ def __init__(self, config): def _get_snowflake_client(self): c = self._config.get('connect') - return snowflake.connector.connect( - user=c['user'], - password=c['access_token'], - account=c['account'], - warehouse=c['warehouse'], - database=c['database'], - schema=c['target_schema'] - ) + try: + return snowflake.connector.connect( + user=c['user'], + password=c['access_token'], + account=c['account'], + warehouse=c['warehouse'], + database=c['database'], + schema=c['target_schema'] + ) + except Exception as e: + raise DestinationConnectionFailed("Failed to connect to Snowflake") from e def _write_batch(self, stream:Stream, batch:List[Record], batch_index:int): diff --git a/data-transfer/pontoon/pontoon/destination/sql_destination.py b/data-transfer/pontoon/pontoon/destination/sql_destination.py index f67322b..7869b64 100644 --- a/data-transfer/pontoon/pontoon/destination/sql_destination.py +++ b/data-transfer/pontoon/pontoon/destination/sql_destination.py @@ -2,10 +2,13 @@ import pyarrow as pa from sqlalchemy import create_engine, inspect, MetaData, Table, Column, text, insert from sqlalchemy import Integer, BigInteger, SmallInteger, String, Text, Float, Numeric, Boolean, Date, Time, DateTime +from sqlalchemy.exc import SQLAlchemyError, OperationalError, DatabaseError, InterfaceError, NoSuchTableError from snowflake.sqlalchemy import TIMESTAMP_LTZ, TIMESTAMP_NTZ, TIMESTAMP_TZ from sqlalchemy.orm import sessionmaker from pontoon.base import Destination, Dataset, Stream, Record, Mode, Progress +from pontoon.base import DestinationConnectionFailed, DestinationStreamInvalidSchema + from pontoon.destination.integrity import SQLIntegrity @@ -143,7 +146,7 @@ def create_table_if_not_exists(conn, stream:Stream, override_name:str = None): # Use flexible schema comparison that ignores column order if not SQLDestination.schemas_compatible(stream.schema, existing_schema): - raise ValueError(f"Existing schema for stream {name} does not match.") + raise DestinationStreamInvalidSchema(f"Existing schema for stream {name} does not match.") else: @@ -167,7 +170,7 @@ def create_table_if_not_exists(conn, stream:Stream, override_name:str = None): def drop_table(conn, table_name:str): # drop a table with conn.begin(): - conn.execute(text(f"DROP TABLE {table_name}")) + conn.execute(text(f"DROP TABLE IF EXISTS {table_name}")) @@ -198,7 +201,10 @@ def __init__(self, config): def _connect(self): - return self._engine.connect() + try: + return self._engine.connect() + except (InterfaceError, DatabaseError, OperationalError) as e: + raise DestinationConnectionFailed("Failed to connect to destination database") from e def _batch_to_rows(self, stream:Stream, batch:List[Record]): diff --git a/data-transfer/pontoon/pontoon/source/memory_source.py b/data-transfer/pontoon/pontoon/source/memory_source.py index 29cbb9c..4b29633 100644 --- a/data-transfer/pontoon/pontoon/source/memory_source.py +++ b/data-transfer/pontoon/pontoon/source/memory_source.py @@ -1,4 +1,5 @@ -from datetime import datetime, timezone +import datetime +from datetime import timezone from pontoon.base import Source, Namespace, Stream, Dataset, Progress, Mode @@ -12,7 +13,7 @@ def __init__(self, config, cache_implementation, cache_config={}): self._mode = config.get('mode') self._with = config.get('with', {}) self._namespace = Namespace(config.get('connect', {}).get('namespace', 'memory')) - self._dt = datetime.now(timezone.utc) + self._dt = datetime.datetime.now(timezone.utc) self._batch_id = str(int(self._dt.timestamp()*1000)) self._progress_callback = None self._cache = cache_implementation(self._namespace, cache_config) @@ -38,6 +39,9 @@ def inspect_streams(self): {'name': 'name', 'type': 'string'}, {'name': 'email', 'type': 'string'}, {'name': 'score', 'type': 'int32'}, + {'name': 'total', 'type': 'float64'}, + {'name': 'open_date', 'type': 'date32'}, + {'name': 'prefs', 'type': 'str'}, {'name': 'notes', 'type': 'string'} ] }] @@ -55,12 +59,15 @@ def read(self, progress_callback=None) -> Dataset: filters={'customer_id': 'Customer5'}, schema=Stream.build_schema([ ('id',str), - ('created_at',datetime), - ('updated_at',datetime), + ('created_at',datetime.datetime), + ('updated_at',datetime.datetime), ('customer_id',str), ('name',str), ('email',str), ('score',int), + ('total', float), + ('open_date', datetime.date), + ('prefs', dict), ('notes',str)]) ) @@ -74,27 +81,129 @@ def read(self, progress_callback=None) -> Dataset: if self._with.get('last_sync'): stream.with_last_synced_at(self._dt) + # ignore any stream fields? + for field in self._config.get('streams', [{}])[0].get('drop_fields', []): + stream.drop_field(field) + self._streams.append(stream) batch = [ - stream.to_record(r) for r in [ - ['a', datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer5', 'User1', 'user1@example.com', 1, 'Notes for User1'], - ['b', datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer5', 'User2', 'user2@example.com', 1, 'Notes for User2'], - ['c', datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer5', 'User3', 'user3@example.com', 1, 'Notes for User3'], - ['d', datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer5', 'User4', 'user4@example.com', 1, 'Notes for User4'], - ['e', datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer5', 'User5', 'user5@example.com', 1, 'Notes for User5'], - ['f', datetime.fromisoformat('2025-01-02T00:00:00+00:00'), datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer5', 'User6', 'user6@example.com', 1, 'Notes for User6'], - ['g', datetime.fromisoformat('2025-01-02T00:00:00+00:00'), datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer5', 'User7', 'user7@example.com', 1, 'Notes for User7'], - ['h', datetime.fromisoformat('2025-01-02T00:00:00+00:00'), datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer5', 'User8', 'user8@example.com', 1, 'Notes for User8'], - ['i', datetime.fromisoformat('2025-01-02T00:00:00+00:00'), datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer5', 'User9', 'user9@example.com', 1, 'Notes for User9'], - ['j', datetime.fromisoformat('2025-01-02T00:00:00+00:00'), datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer5', 'User10', 'user10@example.com', 1, 'Notes for User10'] - ] + ['1', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer3', 'User1', 'user1@example.com', 9, 64.66, datetime.date(2024, 3, 31), {'theme': 'dark', 'notifications': False}, 'Notes for User1'], + ['2', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User2', 'user2@example.com', 8, 79.63, datetime.date(2024, 12, 29), {'theme': 'light', 'notifications': False}, 'Notes for User2'], + ['3', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer3', 'User3', 'user3@example.com', 10, 20.63, datetime.date(2024, 2, 18), {'theme': 'dark', 'notifications': False}, 'Notes for User3'], + ['4', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer2', 'User4', 'user4@example.com', 7, 18.1, datetime.date(2024, 9, 2), {'theme': 'light', 'notifications': False}, 'Notes for User4'], + ['5', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer1', 'User5', 'user5@example.com', 4, 17.56, datetime.date(2024, 8, 18), {'theme': 'light', 'notifications': False}, 'Notes for User5'], + ['6', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer3', 'User6', 'user6@example.com', 10, 63.73, datetime.date(2024, 9, 28), {'theme': 'dark', 'notifications': False}, 'Notes for User6'], + ['7', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User7', 'user7@example.com', 2, 44.05, datetime.date(2024, 4, 20), {'theme': 'dark', 'notifications': True}, 'Notes for User7'], + ['8', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User8', 'user8@example.com', 8, 41.91, datetime.date(2024, 5, 22), {'theme': 'dark', 'notifications': False}, 'Notes for User8'], + ['9', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer3', 'User9', 'user9@example.com', 9, 34.32, datetime.date(2024, 2, 29), {'theme': 'light', 'notifications': False}, 'Notes for User9'], + ['10', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User10', 'user10@example.com', 10, 97.53, datetime.date(2024, 9, 12), {'theme': 'dark', 'notifications': True}, 'Notes for User10'], + ['11', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer1', 'User11', 'user11@example.com', 2, 61.3, datetime.date(2024, 8, 23), {'theme': 'light', 'notifications': False}, 'Notes for User11'], + ['12', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User12', 'user12@example.com', 5, 66.97, datetime.date(2024, 6, 12), {'theme': 'light', 'notifications': False}, 'Notes for User12'], + ['13', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer3', 'User13', 'user13@example.com', 1, 59.07, datetime.date(2024, 10, 21), {'theme': 'dark', 'notifications': True}, 'Notes for User13'], + ['14', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User14', 'user14@example.com', 1, 80.21, datetime.date(2024, 10, 11), {'theme': 'dark', 'notifications': False}, 'Notes for User14'], + ['15', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer1', 'User15', 'user15@example.com', 10, 64.16, datetime.date(2024, 3, 10), {'theme': 'light', 'notifications': False}, 'Notes for User15'], + ['16', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer1', 'User16', 'user16@example.com', 2, 53.55, datetime.date(2024, 10, 30), {'theme': 'dark', 'notifications': False}, 'Notes for User16'], + ['17', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer1', 'User17', 'user17@example.com', 9, 10.62, datetime.date(2024, 12, 26), {'theme': 'dark', 'notifications': False}, 'Notes for User17'], + ['18', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User18', 'user18@example.com', 5, 24.21, datetime.date(2024, 12, 2), {'theme': 'light', 'notifications': False}, 'Notes for User18'], + ['19', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer1', 'User19', 'user19@example.com', 3, 84.07, datetime.date(2024, 1, 17), {'theme': 'light', 'notifications': False}, 'Notes for User19'], + ['20', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User20', 'user20@example.com', 3, 92.48, datetime.date(2024, 5, 30), {'theme': 'dark', 'notifications': False}, 'Notes for User20'], + ['21', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User21', 'user21@example.com', 10, 71.66, datetime.date(2024, 6, 1), {'theme': 'dark', 'notifications': True}, 'Notes for User21'], + ['22', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User22', 'user22@example.com', 10, 96.14, datetime.date(2024, 8, 19), {'theme': 'dark', 'notifications': True}, 'Notes for User22'], + ['23', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User23', 'user23@example.com', 5, 98.75, datetime.date(2024, 1, 19), {'theme': 'light', 'notifications': False}, 'Notes for User23'], + ['24', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer1', 'User24', 'user24@example.com', 6, 53.37, datetime.date(2024, 7, 6), {'theme': 'dark', 'notifications': True}, 'Notes for User24'], + ['25', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User25', 'user25@example.com', 1, 43.81, datetime.date(2024, 6, 26), {'theme': 'light', 'notifications': False}, 'Notes for User25'], + ['26', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User26', 'user26@example.com', 8, 96.87, datetime.date(2024, 1, 28), {'theme': 'dark', 'notifications': True}, 'Notes for User26'], + ['27', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User27', 'user27@example.com', 5, 31.58, datetime.date(2024, 6, 29), {'theme': 'dark', 'notifications': False}, 'Notes for User27'], + ['28', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer1', 'User28', 'user28@example.com', 7, 16.14, datetime.date(2024, 1, 13), {'theme': 'dark', 'notifications': True}, 'Notes for User28'], + ['29', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User29', 'user29@example.com', 3, 58.45, datetime.date(2024, 6, 10), {'theme': 'dark', 'notifications': False}, 'Notes for User29'], + ['30', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer3', 'User30', 'user30@example.com', 4, 18.52, datetime.date(2024, 7, 24), {'theme': 'dark', 'notifications': True}, 'Notes for User30'], + ['31', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer1', 'User31', 'user31@example.com', 5, 54.72, datetime.date(2024, 8, 2), {'theme': 'dark', 'notifications': False}, 'Notes for User31'], + ['32', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer1', 'User32', 'user32@example.com', 1, 89.11, datetime.date(2024, 8, 14), {'theme': 'dark', 'notifications': True}, 'Notes for User32'], + ['33', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer1', 'User33', 'user33@example.com', 7, 91.84, datetime.date(2024, 8, 20), {'theme': 'light', 'notifications': False}, 'Notes for User33'], + ['34', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User34', 'user34@example.com', 9, 84.8, datetime.date(2024, 9, 30), {'theme': 'light', 'notifications': False}, 'Notes for User34'], + ['35', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer3', 'User35', 'user35@example.com', 10, 80.43, datetime.date(2024, 1, 17), {'theme': 'light', 'notifications': False}, 'Notes for User35'], + ['36', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User36', 'user36@example.com', 6, 90.63, datetime.date(2024, 7, 10), {'theme': 'light', 'notifications': True}, 'Notes for User36'], + ['37', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer1', 'User37', 'user37@example.com', 7, 77.41, datetime.date(2024, 3, 30), {'theme': 'light', 'notifications': True}, 'Notes for User37'], + ['38', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User38', 'user38@example.com', 3, 46.25, datetime.date(2024, 12, 12), {'theme': 'light', 'notifications': True}, 'Notes for User38'], + ['39', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User39', 'user39@example.com', 6, 63.13, datetime.date(2024, 7, 7), {'theme': 'dark', 'notifications': False}, 'Notes for User39'], + ['40', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User40', 'user40@example.com', 10, 27.86, datetime.date(2024, 12, 19), {'theme': 'dark', 'notifications': False}, 'Notes for User40'], + ['41', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User41', 'user41@example.com', 2, 61.47, datetime.date(2024, 12, 2), {'theme': 'light', 'notifications': False}, 'Notes for User41'], + ['42', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User42', 'user42@example.com', 10, 47.69, datetime.date(2024, 10, 5), {'theme': 'light', 'notifications': False}, 'Notes for User42'], + ['43', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User43', 'user43@example.com', 7, 83.64, datetime.date(2024, 7, 28), {'theme': 'dark', 'notifications': True}, 'Notes for User43'], + ['44', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User44', 'user44@example.com', 5, 74.0, datetime.date(2024, 12, 28), {'theme': 'light', 'notifications': False}, 'Notes for User44'], + ['45', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer3', 'User45', 'user45@example.com', 3, 34.66, datetime.date(2024, 2, 4), {'theme': 'light', 'notifications': False}, 'Notes for User45'], + ['46', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer3', 'User46', 'user46@example.com', 6, 13.51, datetime.date(2024, 7, 13), {'theme': 'light', 'notifications': False}, 'Notes for User46'], + ['47', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer1', 'User47', 'user47@example.com', 8, 39.92, datetime.date(2024, 1, 24), {'theme': 'dark', 'notifications': False}, 'Notes for User47'], + ['48', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer1', 'User48', 'user48@example.com', 2, 73.54, datetime.date(2024, 4, 8), {'theme': 'dark', 'notifications': False}, 'Notes for User48'], + ['49', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer1', 'User49', 'user49@example.com', 5, 11.1, datetime.date(2024, 11, 26), {'theme': 'light', 'notifications': False}, 'Notes for User49'], + ['50', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer3', 'User50', 'user50@example.com', 10, 44.79, datetime.date(2024, 7, 29), {'theme': 'dark', 'notifications': True}, 'Notes for User50'], + ['51', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer1', 'User51', 'user51@example.com', 1, 57.32, datetime.date(2024, 10, 25), {'theme': 'dark', 'notifications': True}, 'Notes for User51'], + ['52', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User52', 'user52@example.com', 10, 80.22, datetime.date(2024, 6, 14), {'theme': 'light', 'notifications': True}, 'Notes for User52'], + ['53', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer3', 'User53', 'user53@example.com', 4, 42.95, datetime.date(2024, 2, 25), {'theme': 'dark', 'notifications': False}, 'Notes for User53'], + ['54', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User54', 'user54@example.com', 1, 96.79, datetime.date(2024, 7, 15), {'theme': 'light', 'notifications': True}, 'Notes for User54'], + ['55', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User55', 'user55@example.com', 4, 89.56, datetime.date(2024, 12, 5), {'theme': 'dark', 'notifications': False}, 'Notes for User55'], + ['56', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User56', 'user56@example.com', 6, 52.15, datetime.date(2024, 5, 10), {'theme': 'light', 'notifications': False}, 'Notes for User56'], + ['57', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer3', 'User57', 'user57@example.com', 10, 40.7, datetime.date(2024, 5, 8), {'theme': 'dark', 'notifications': True}, 'Notes for User57'], + ['58', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer1', 'User58', 'user58@example.com', 6, 96.51, datetime.date(2024, 6, 23), {'theme': 'light', 'notifications': True}, 'Notes for User58'], + ['59', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer1', 'User59', 'user59@example.com', 3, 22.93, datetime.date(2024, 1, 26), {'theme': 'dark', 'notifications': True}, 'Notes for User59'], + ['60', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User60', 'user60@example.com', 5, 66.02, datetime.date(2024, 7, 18), {'theme': 'dark', 'notifications': False}, 'Notes for User60'], + ['61', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer1', 'User61', 'user61@example.com', 4, 33.98, datetime.date(2024, 9, 26), {'theme': 'dark', 'notifications': False}, 'Notes for User61'], + ['62', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User62', 'user62@example.com', 6, 26.83, datetime.date(2024, 9, 21), {'theme': 'dark', 'notifications': False}, 'Notes for User62'], + ['63', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User63', 'user63@example.com', 7, 58.92, datetime.date(2024, 4, 12), {'theme': 'light', 'notifications': False}, 'Notes for User63'], + ['64', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User64', 'user64@example.com', 10, 36.37, datetime.date(2024, 12, 20), {'theme': 'dark', 'notifications': True}, 'Notes for User64'], + ['65', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User65', 'user65@example.com', 9, 59.98, datetime.date(2024, 12, 7), {'theme': 'dark', 'notifications': True}, 'Notes for User65'], + ['66', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User66', 'user66@example.com', 3, 43.79, datetime.date(2024, 10, 19), {'theme': 'light', 'notifications': True}, 'Notes for User66'], + ['67', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer1', 'User67', 'user67@example.com', 8, 29.1, datetime.date(2024, 2, 24), {'theme': 'dark', 'notifications': True}, 'Notes for User67'], + ['68', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User68', 'user68@example.com', 3, 91.13, datetime.date(2024, 10, 12), {'theme': 'light', 'notifications': True}, 'Notes for User68'], + ['69', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer1', 'User69', 'user69@example.com', 2, 87.22, datetime.date(2024, 3, 4), {'theme': 'dark', 'notifications': True}, 'Notes for User69'], + ['70', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User70', 'user70@example.com', 3, 69.95, datetime.date(2024, 6, 7), {'theme': 'dark', 'notifications': True}, 'Notes for User70'], + ['71', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer2', 'User71', 'user71@example.com', 2, 41.44, datetime.date(2024, 5, 12), {'theme': 'dark', 'notifications': True}, 'Notes for User71'], + ['72', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User72', 'user72@example.com', 1, 91.61, datetime.date(2024, 6, 4), {'theme': 'light', 'notifications': True}, 'Notes for User72'], + ['73', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer3', 'User73', 'user73@example.com', 2, 15.23, datetime.date(2024, 2, 14), {'theme': 'light', 'notifications': False}, 'Notes for User73'], + ['74', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer2', 'User74', 'user74@example.com', 8, 96.08, datetime.date(2024, 4, 9), {'theme': 'light', 'notifications': True}, 'Notes for User74'], + ['75', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer2', 'User75', 'user75@example.com', 4, 76.82, datetime.date(2024, 9, 17), {'theme': 'dark', 'notifications': True}, 'Notes for User75'], + ['76', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User76', 'user76@example.com', 2, 85.73, datetime.date(2024, 12, 28), {'theme': 'light', 'notifications': True}, 'Notes for User76'], + ['77', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer3', 'User77', 'user77@example.com', 4, 42.4, datetime.date(2024, 3, 10), {'theme': 'dark', 'notifications': True}, 'Notes for User77'], + ['78', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer3', 'User78', 'user78@example.com', 8, 65.35, datetime.date(2024, 4, 23), {'theme': 'dark', 'notifications': True}, 'Notes for User78'], + ['79', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer1', 'User79', 'user79@example.com', 5, 81.81, datetime.date(2024, 11, 8), {'theme': 'dark', 'notifications': False}, 'Notes for User79'], + ['80', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User80', 'user80@example.com', 4, 92.34, datetime.date(2024, 9, 3), {'theme': 'dark', 'notifications': True}, 'Notes for User80'], + ['81', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User81', 'user81@example.com', 3, 52.4, datetime.date(2024, 1, 28), {'theme': 'dark', 'notifications': False}, 'Notes for User81'], + ['82', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer2', 'User82', 'user82@example.com', 1, 16.41, datetime.date(2024, 4, 15), {'theme': 'light', 'notifications': True}, 'Notes for User82'], + ['83', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer1', 'User83', 'user83@example.com', 6, 70.29, datetime.date(2024, 8, 2), {'theme': 'light', 'notifications': False}, 'Notes for User83'], + ['84', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-02T00:00:00+00:00'), 'Customer1', 'User84', 'user84@example.com', 6, 39.54, datetime.date(2024, 12, 12), {'theme': 'light', 'notifications': True}, 'Notes for User84'], + ['85', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer1', 'User85', 'user85@example.com', 1, 55.65, datetime.date(2024, 10, 18), {'theme': 'light', 'notifications': False}, 'Notes for User85'], + ['86', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User86', 'user86@example.com', 4, 77.6, datetime.date(2024, 9, 24), {'theme': 'light', 'notifications': False}, 'Notes for User86'], + ['87', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer3', 'User87', 'user87@example.com', 1, 81.69, datetime.date(2024, 5, 12), {'theme': 'dark', 'notifications': True}, 'Notes for User87'], + ['88', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User88', 'user88@example.com', 1, 34.37, datetime.date(2024, 4, 6), {'theme': 'light', 'notifications': True}, 'Notes for User88'], + ['89', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer3', 'User89', 'user89@example.com', 3, 31.57, datetime.date(2024, 12, 27), {'theme': 'dark', 'notifications': True}, 'Notes for User89'], + ['90', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer2', 'User90', 'user90@example.com', 5, 73.71, datetime.date(2024, 6, 1), {'theme': 'light', 'notifications': False}, 'Notes for User90'], + ['91', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer1', 'User91', 'user91@example.com', 9, 12.74, datetime.date(2024, 12, 17), {'theme': 'light', 'notifications': False}, 'Notes for User91'], + ['92', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer2', 'User92', 'user92@example.com', 6, 90.76, datetime.date(2024, 1, 2), {'theme': 'light', 'notifications': False}, 'Notes for User92'], + ['93', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer1', 'User93', 'user93@example.com', 8, 49.29, datetime.date(2024, 9, 13), {'theme': 'light', 'notifications': False}, 'Notes for User93'], + ['94', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer3', 'User94', 'user94@example.com', 8, 22.58, datetime.date(2024, 6, 20), {'theme': 'dark', 'notifications': False}, 'Notes for User94'], + ['95', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer2', 'User95', 'user95@example.com', 3, 81.34, datetime.date(2024, 4, 12), {'theme': 'light', 'notifications': False}, 'Notes for User95'], + ['96', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-03T00:00:00+00:00'), 'Customer3', 'User96', 'user96@example.com', 10, 47.45, datetime.date(2024, 1, 1), {'theme': 'dark', 'notifications': False}, 'Notes for User96'], + ['97', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer1', 'User97', 'user97@example.com', 8, 85.96, datetime.date(2024, 10, 31), {'theme': 'light', 'notifications': True}, 'Notes for User97'], + ['98', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), 'Customer3', 'User98', 'user98@example.com', 6, 68.55, datetime.date(2024, 8, 16), {'theme': 'light', 'notifications': True}, 'Notes for User98'], + ['99', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-05T00:00:00+00:00'), 'Customer2', 'User99', 'user99@example.com', 9, 39.18, datetime.date(2024, 1, 29), {'theme': 'light', 'notifications': False}, 'Notes for User99'], + ['100', datetime.datetime.fromisoformat('2025-01-01T00:00:00+00:00'), datetime.datetime.fromisoformat('2025-01-04T00:00:00+00:00'), 'Customer1', 'User100', 'user100@example.com', 9, 35.7, datetime.date(2024, 8, 1), {'theme': 'light', 'notifications': False}, 'Notes for User100'] + ] + # Filter to a specific customer ID + if self._config.get('streams'): + customer_id = self._config.get('streams')[0].get('filters', {}).get('customer_id', None) + batch = [r for r in batch \ + if r[3] == customer_id] + + # Filter to a specific date range if self._mode.type == Mode.INCREMENTAL: batch = [r for r in batch \ - if r.data[1] >= self._mode.start and r.data[1] < self._mode.end] + if r[2] >= self._mode.start and r[2] < self._mode.end] + + batch = [stream.to_record(r) for r in batch] progress = Progress( f"source+memory://{self._namespace}/{stream.schema_name}/{stream.name}", diff --git a/data-transfer/pontoon/pontoon/source/sql_source.py b/data-transfer/pontoon/pontoon/source/sql_source.py index 555d7ad..d6bc6e9 100644 --- a/data-transfer/pontoon/pontoon/source/sql_source.py +++ b/data-transfer/pontoon/pontoon/source/sql_source.py @@ -4,9 +4,14 @@ from datetime import datetime, timezone, date from decimal import Decimal from sqlalchemy import create_engine, inspect, MetaData, Table, text, select, func +from sqlalchemy.exc import SQLAlchemyError, OperationalError, InterfaceError, DatabaseError, NoSuchTableError from pontoon import logger from pontoon.base import Source, Namespace, Stream, Dataset, Progress, Mode +from pontoon.base import StreamMissingField +from pontoon.base import SourceConnectionFailed, \ + SourceStreamDoesNotExist, \ + SourceStreamInvalidSchema class SQLUtil: @@ -158,7 +163,10 @@ def __init__(self, config, cache_implementation, cache_config={}): def _connect(self): - return self._engine.connect() + try: + return self._engine.connect() + except (OperationalError, InterfaceError, DatabaseError) as e: + raise SourceConnectionFailed("Failed to connect to source database") from e def test_connect(self): @@ -251,12 +259,13 @@ def read(self, progress_callback=None) -> Dataset: # determine the schema table_name = stream_config['table'] metadata = MetaData() - table = Table(table_name, metadata, schema=stream_config['schema'], autoload_with=conn) - - # table doesn't exist? - if not table.exists(conn): - raise Exception(f"SQLSource (source-sql) table does not exist: {stream_config['schema']}.{table_name}") + try: + # table doesn't exist? + table = Table(table_name, metadata, schema=stream_config['schema'], autoload_with=conn) + except NoSuchTableError as e: + raise SourceStreamDoesNotExist(f"SQLSource (source-sql) table does not exist: {stream_config['schema']}.{table_name}") from e + columns = [] for col in table.columns: try: @@ -266,14 +275,17 @@ def read(self, progress_callback=None) -> Dataset: columns.append((col.name, str(col.type))) # create a schema from the column field + types - stream = Stream( - name=stream_config['table'], - schema_name=stream_config['schema'], - primary_field=stream_config.get('primary_field', None), - cursor_field=stream_config.get('cursor_field', None), - filters=stream_config.get('filters', None), - schema=Stream.build_schema(columns) - ) + try: + stream = Stream( + name=stream_config['table'], + schema_name=stream_config['schema'], + primary_field=stream_config.get('primary_field', None), + cursor_field=stream_config.get('cursor_field', None), + filters=stream_config.get('filters', None), + schema=Stream.build_schema(columns) + ) + except StreamMissingField as e: + raise SourceStreamInvalidSchema(e) from e count_query = SQLUtil.build_select_query(stream, self._mode, count=True) select_query = SQLUtil.build_select_query(stream, self._mode) @@ -297,6 +309,7 @@ def read(self, progress_callback=None) -> Dataset: # configure progress tracking total_count = conn.execute(text(count_query)).scalar_one() + progress = Progress( f"source+sql://{self._namespace}/{stream.schema_name}/{stream.name}", total=total_count, @@ -305,6 +318,10 @@ def read(self, progress_callback=None) -> Dataset: if callable(progress_callback): progress.subscribe(progress_callback) + if total_count == 0: + progress.message("No records to process") + continue + # execute our main query result = conn.execution_options( stream_results=True, @@ -322,6 +339,9 @@ def read(self, progress_callback=None) -> Dataset: progress.update(self._chunk_size, increment=True) batch = [] + # close the server side cursor + result.close() + if len(batch) > 0: self._cache.write(stream, batch) progress.update(len(batch), increment=True) diff --git a/data-transfer/pontoon/tests/integration/common.py b/data-transfer/pontoon/tests/integration/common.py new file mode 100644 index 0000000..e2dd735 --- /dev/null +++ b/data-transfer/pontoon/tests/integration/common.py @@ -0,0 +1,57 @@ +import os +import json +import glob +import uuid +import psutil +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from dotenv import load_dotenv +load_dotenv() + +def clear_cache_files(): + for f in glob.glob("*_cache.db"): + os.remove(f) + + +def read_progress_handler(progress:Progress): + print(f"{progress.entity()}: {progress.processed}/{progress.total} records ({progress.percent}%)") + + +def write_progress_handler(progress:Progress): + print(f"{progress.entity()}: {progress.processed}/{progress.total} records ({progress.percent}%)") + + +def get_memory_source(mode_config={}, streams_config={}): + return get_source( + get_source_by_vendor('memory'), + config={ + 'with': { + 'batch_id': True, + 'last_sync': True + }, + 'mode': Mode({ + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config), + 'streams': [{ + 'filters': {'customer_id': 'Customer1'} + } | streams_config] + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_memory_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) diff --git a/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py b/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py new file mode 100644 index 0000000..fbee123 --- /dev/null +++ b/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py @@ -0,0 +1,251 @@ +import os +import json +import glob +import uuid +import psutil +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from tests.integration.common import clear_cache_files, get_memory_source, read_progress_handler, write_progress_handler + +from dotenv import load_dotenv +load_dotenv() + + +clear_cache_files() + +class TestBigQueryConnectors: + + def test_bigquery_source(self): + + def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) + } | mode_config + + test_with_config = { + 'batch_id': True, + 'last_sync': True + } | with_config + + test_connect_config = { + 'auth_type': 'service_account', + 'vendor_type': 'bigquery', + 'project_id': os.environ['BQ_PROJECT_ID'], + 'service_account': open(os.environ['GCP_SERVICE_ACCOUNT_FILE']).read() + } | connect_config + + test_streams_config = [{ + 'schema': 'pontoon', + 'table': 'leads_xs', + 'primary_field': 'id', + 'cursor_field': 'updated_at', + 'filters': {'customer_id': 'Customer1'}, + 'drop_fields': ['customer_id'] + } | streams_config] + + return get_source( + get_source_by_vendor('bigquery'), + config = { + 'mode': Mode(test_mode_config), + 'with': test_with_config, + 'connect': test_connect_config, + 'streams': test_streams_config + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_bigquery_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) + + + # Source stream does not exist + src = get_test_source(streams_config={'table': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamDoesNotExist): + src.read(progress_callback=read_progress_handler) + + # Source stream field does not exist + src = get_test_source(streams_config={'primary_field': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamInvalidSchema): + src.read(progress_callback=read_progress_handler) + + # Empty record set + src = get_test_source(mode_config={'start': datetime(2010, 1, 1, tzinfo=timezone.utc), 'end': datetime(2010, 1, 2, tzinfo=timezone.utc)}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 0 + + # Full refresh + src = get_test_source(mode_config={'type': Mode.FULL_REFRESH}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 329 + + # Incremental + src = get_test_source() + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 87 + + + def test_bigquery_destination(self): + + def get_test_destination(mode_config={}, connect_config={}): + + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config + + test_connect_config = { + 'auth_type': 'service_account', + 'vendor_type': 'bigquery', + 'gcs_bucket_name': os.environ['GCS_BUCKET'], + 'gcs_bucket_path': 'pontoon', + 'project_id': os.environ['BQ_PROJECT_ID'], + 'service_account': open(os.environ['GCP_SERVICE_ACCOUNT_FILE']).read(), + 'target_schema': 'target' + } | connect_config + + return get_destination( + get_destination_by_vendor('bigquery'), + config = { + 'mode': Mode(test_mode_config), + 'connect': test_connect_config + } + ) + + def connect(): + return create_engine( + f"bigquery://{os.environ['BQ_PROJECT_ID']}", + credentials_info=json.loads(open(os.environ['GCP_SERVICE_ACCOUNT_FILE']).read()), + arraysize=1024 + ).connect() + + def drop(): + with connect() as conn: + conn.execute(text("DROP TABLE IF EXISTS target.pontoon_transfer_test")) + conn.execute(text("DROP TABLE IF EXISTS target.__temp_pontoon_transfer_test")) + + + drop() + + # Cannot connect to destination + # dest = get_test_destination(connect_config={'host': 'doesnotexist'}) + # ds = get_memory_source().read(progress_callback=read_progress_handler) + # with pytest.raises(DestinationConnectionFailed): + # dest.write(ds, progress_callback=write_progress_handler) + + + # Existing destination schema is not compatible (destination changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + with connect() as conn: + conn.execute(text("ALTER TABLE target.pontoon_transfer_test DROP COLUMN customer_id")) + + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + drop() + + # Existing destination schema is not compatible (source changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + dest = get_test_destination() + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + # Full refresh should overwrite destination with different schema + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + drop() + + # Full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + # Full refresh followed by full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + + # Full refresh, then incremental load + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-02' AND updated_at < '2025-01-03'")).scalar() + assert count == 7 + + # Incremental reload same day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-02' AND updated_at < '2025-01-03'")).scalar() + assert count == 7 + + # Incremental load second day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-03' AND updated_at < '2025-01-04'")).scalar() + assert count == 7 + + # Incremental load third day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-04' AND updated_at < '2025-01-05'")).scalar() + assert count == 6 + + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + drop() + diff --git a/data-transfer/pontoon/tests/integration/test_core_connectors.py b/data-transfer/pontoon/tests/integration/test_core_connectors.py deleted file mode 100644 index 9bce7b7..0000000 --- a/data-transfer/pontoon/tests/integration/test_core_connectors.py +++ /dev/null @@ -1,324 +0,0 @@ -import os -import json -import glob -import uuid -import psutil -from datetime import datetime, timezone -from pontoon import configure_logging -from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor -from pontoon import SqliteCache -from pontoon import Progress, Mode - -from dotenv import load_dotenv - - -load_dotenv() - -for f in glob.glob("*_cache.db"): - os.remove(f) - - -def read_progress_handler(progress:Progress): - print(f"{progress.entity()}: {progress.processed}/{progress.total} records ({progress.percent}%)") - - -def write_progress_handler(progress:Progress): - print(f"{progress.entity()}: {progress.processed}/{progress.total} records ({progress.percent}%)") - - -def get_memory_source(): - return get_source( - get_source_by_vendor('memory'), - config={ - 'with': { - 'batch_id': True, - 'last_sync': True - }, - 'mode': Mode({ - 'type': Mode.INCREMENTAL, - 'period': Mode.DAILY, - 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), - 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) - }) - }, - cache_implementation = SqliteCache, - cache_config = { - 'db': f"_memory_{uuid.uuid4()}_cache.db", - 'chunk_size': 1024 - } - ) - - -mode_config = { - 'type': Mode.INCREMENTAL, - 'period': Mode.DAILY, - 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), - 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) -} - -with_config = { - 'batch_id': True, - 'last_sync': True -} - - -console_dest = get_destination( - get_destination_by_vendor('console'), - config = { - 'connect': { - 'limit': 3 - } - } -) - - -postgresql_src = get_source( - get_source_by_vendor('postgresql'), - config = { - 'mode': Mode(mode_config), - 'with': with_config, - 'connect': { - 'auth_type': 'basic', - 'vendor_type': 'postgresql', - 'host': os.environ['POSTGRES_HOST'], - 'port': '5432', - 'user': os.environ['POSTGRES_USER'], - 'password': os.environ['POSTGRES_PASSWORD'], - 'database': os.environ['POSTGRES_DATABASE'] - }, - 'streams': [{ - 'schema': 'source', - 'table': 'leads_xs', - 'primary_field': 'id', - 'cursor_field': 'updated_at', - 'filters': {'customer_id': 'Customer1'}, - 'drop_fields': ['customer_id'] - }] - }, - cache_implementation = SqliteCache, - cache_config = { - 'db': '_postgresql_cache.db', - 'chunk_size': 1024 - } -) - -redshift_src = get_source( - get_source_by_vendor('redshift'), - config = { - 'mode': Mode(mode_config), - 'with': with_config, - 'connect': { - 'auth_type': 'basic', - 'vendor_type': 'redshift', - 'host': os.environ['REDSHIFT_HOST'], - 'port': '5439', - 'user': os.environ['REDSHIFT_USER'], - 'password': os.environ['REDSHIFT_PASSWORD'], - 'database': os.environ['REDSHIFT_DATABASE'] - }, - 'streams': [{ - 'schema': 'source', - 'table': 'leads_xs', - 'primary_field': 'id', - 'cursor_field': 'updated_at', - 'filters': {'customer_id': 'Customer1'}, - 'drop_fields': ['customer_id'] - }] - }, - cache_implementation = SqliteCache, - cache_config = { - 'db': '_redshift_cache.db', - 'chunk_size': 1024 - } -) - -bigquery_src = get_source( - get_source_by_vendor('bigquery'), - config = { - 'mode': Mode(mode_config), - 'with': with_config, - 'connect': { - 'auth_type': 'service_account', - 'vendor_type': 'bigquery', - 'project_id': os.environ['BQ_PROJECT_ID'], - 'service_account': open(os.environ['GCP_SERVICE_ACCOUNT_FILE']).read() - }, - 'streams': [{ - 'schema': 'pontoon', - 'table': 'leads_xs', - 'primary_field': 'id', - 'cursor_field': 'updated_at', - 'filters': {'customer_id': 'Customer1'}, - 'drop_fields': ['customer_id'] - }] - }, - cache_implementation = SqliteCache, - cache_config = { - 'db': '_bigquery_cache.db', - 'chunk_size': 1024 - } -) - -snowflake_src = get_source( - get_source_by_vendor('snowflake'), - config = { - 'mode': Mode(mode_config), - 'with': with_config, - 'connect': { - 'auth_type': 'access_token', - 'vendor_type': 'snowflake', - 'user': os.environ['SNOWFLAKE_USER'], - 'access_token': os.environ['SNOWFLAKE_ACCESS_TOKEN'], - 'account': os.environ['SNOWFLAKE_ACCOUNT'], - 'database': os.environ['SNOWFLAKE_DATABASE'], - 'warehouse': os.environ['SNOWFLAKE_WAREHOUSE'] - }, - 'streams': [{ - 'schema': 'pontoon', - 'table': 'leads_xs', - 'primary_field': 'id', - 'cursor_field': 'updated_at', - 'filters': {'customer_id': 'Customer1'}, - 'drop_fields': ['customer_id'] - }] - }, - cache_implementation = SqliteCache, - cache_config = { - 'db': '_snowflake_cache.db', - 'chunk_size': 1024 - } -) - - -### -### ====================================================================== -### - - -postgresql_dest = get_destination( - get_destination_by_vendor('postgresql'), - config = { - 'mode': Mode(mode_config), - 'connect': { - 'auth_type': 'basic', - 'vendor_type': 'postgresql', - 'host': os.environ['POSTGRES_HOST'], - 'port': '5432', - 'user': os.environ['POSTGRES_USER'], - 'password': os.environ['POSTGRES_PASSWORD'], - 'database': os.environ['POSTGRES_DATABASE'], - 'target_schema': 'target' - } - } -) - - -redshift_dest = get_destination( - get_destination_by_vendor('redshift'), - config = { - 'mode': Mode(mode_config), - 'connect': { - 'auth_type': 'basic', - 'vendor_type': 'redshift', - 's3_bucket': os.environ['S3_BUCKET'], - 's3_prefix': 'pontoon', - 's3_region': os.environ['AWS_DEFAULT_REGION'], - 'iam_role': os.environ['REDSHIFT_IAM_ROLE'], - 'aws_access_key_id': os.environ['AWS_ACCESS_KEY_ID'], - 'aws_secret_access_key': os.environ['AWS_SECRET_ACCESS_KEY'], - 'host': os.environ['REDSHIFT_HOST'], - 'user': os.environ['REDSHIFT_USER'], - 'password': os.environ['REDSHIFT_PASSWORD'], - 'database': os.environ['REDSHIFT_DATABASE'], - 'port': '5439', - 'target_schema': 'target' - } - } -) - -bigquery_dest = get_destination( - get_destination_by_vendor('bigquery'), - config = { - 'mode': Mode(mode_config), - 'connect': { - 'auth_type': 'service_account', - 'vendor_type': 'bigquery', - 'gcs_bucket_name': os.environ['GCS_BUCKET'], - 'gcs_bucket_path': 'pontoon', - 'project_id': os.environ['BQ_PROJECT_ID'], - 'service_account': open(os.environ['GCP_SERVICE_ACCOUNT_FILE']).read(), - 'target_schema': 'target' - } - } -) - - -snowflake_dest = get_destination( - get_destination_by_vendor('snowflake'), - config = { - 'mode': Mode(mode_config), - 'connect': { - 'auth_type': 'access_token', - 'vendor_type': 'snowflake', - 'stage_name': 'pontoon', - 'create_stage': True, - 'delete_stage': True, - 'user': os.environ['SNOWFLAKE_USER'], - 'access_token': os.environ['SNOWFLAKE_ACCESS_TOKEN'], - 'account': os.environ['SNOWFLAKE_ACCOUNT'], - 'database': os.environ['SNOWFLAKE_DATABASE'], - 'warehouse': os.environ['SNOWFLAKE_WAREHOUSE'], - 'target_schema': 'target' - } - } -) - - -### -### ====================================================================== -### - - -class TestCoreConnectors: - - def test_memory_source(self): - ds = get_memory_source().read(progress_callback=read_progress_handler) - console_dest.write(ds, progress_callback=write_progress_handler) - - def test_postgres_source(self): - ds = postgresql_src.read(progress_callback=read_progress_handler) - console_dest.write(ds, progress_callback=write_progress_handler) - - def test_redshift_source(self): - ds = redshift_src.read(progress_callback=read_progress_handler) - console_dest.write(ds, progress_callback=write_progress_handler) - - def test_bigquery_source(self): - ds = bigquery_src.read(progress_callback=read_progress_handler) - console_dest.write(ds, progress_callback=write_progress_handler) - - def test_snowflake_source(self): - ds = snowflake_src.read(progress_callback=read_progress_handler) - console_dest.write(ds, progress_callback=write_progress_handler) - - def test_postgres_destination(self): - ds = get_memory_source().read(progress_callback=read_progress_handler) - postgresql_dest.write(ds, progress_callback=write_progress_handler) - postgresql_dest.integrity().check_batch_volume(ds) - - def test_redshift_destination(self): - ds = get_memory_source().read(progress_callback=read_progress_handler) - redshift_dest.write(ds, progress_callback=write_progress_handler) - redshift_dest.integrity().check_batch_volume(ds) - - def test_bigquery_destination(self): - ds = get_memory_source().read(progress_callback=read_progress_handler) - bigquery_dest.write(ds, progress_callback=write_progress_handler) - bigquery_dest.integrity().check_batch_volume(ds) - - def test_snowflake_destination(self): - ds = get_memory_source().read(progress_callback=read_progress_handler) - snowflake_dest.write(ds, progress_callback=write_progress_handler) - snowflake_dest.integrity().check_batch_volume(ds) - - diff --git a/data-transfer/pontoon/tests/integration/test_postgres_connectors.py b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py new file mode 100644 index 0000000..32022b3 --- /dev/null +++ b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py @@ -0,0 +1,259 @@ +import os +import json +import glob +import uuid +import psutil +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from tests.integration.common import clear_cache_files, get_memory_source, read_progress_handler, write_progress_handler + +from dotenv import load_dotenv +load_dotenv() + + +clear_cache_files() + +class TestPostgresConnectors: + + def test_postgres_source(self): + + def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) + } | mode_config + + test_with_config = { + 'batch_id': True, + 'last_sync': True + } | with_config + + test_connect_config = { + 'auth_type': 'basic', + 'vendor_type': 'postgresql', + 'host': os.environ['POSTGRES_HOST'], + 'port': '5432', + 'user': os.environ['POSTGRES_USER'], + 'password': os.environ['POSTGRES_PASSWORD'], + 'database': os.environ['POSTGRES_DATABASE'] + } | connect_config + + test_streams_config = [{ + 'schema': 'source', + 'table': 'leads_xs', + 'primary_field': 'id', + 'cursor_field': 'updated_at', + 'filters': {'customer_id': 'Customer1'}, + 'drop_fields': ['customer_id'] + } | streams_config] + + return get_source( + get_source_by_vendor('postgresql'), + config = { + 'mode': Mode(test_mode_config), + 'with': test_with_config, + 'connect': test_connect_config, + 'streams': test_streams_config + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_postgresql_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) + + # Cannot connect to source + src = get_test_source(connect_config={'host': 'doesnotexist'}) + with pytest.raises(SourceConnectionFailed): + src.test_connect() + with pytest.raises(SourceConnectionFailed): + src.read(progress_callback=read_progress_handler) + + + # Source stream does not exist + src = get_test_source(streams_config={'table': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamDoesNotExist): + src.read(progress_callback=read_progress_handler) + + # Source stream field does not exist + src = get_test_source(streams_config={'primary_field': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamInvalidSchema): + src.read(progress_callback=read_progress_handler) + + # Empty record set + src = get_test_source(mode_config={'start': datetime(2010, 1, 1, tzinfo=timezone.utc), 'end': datetime(2010, 1, 2, tzinfo=timezone.utc)}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 0 + + # Full refresh + src = get_test_source(mode_config={'type': Mode.FULL_REFRESH}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 329 + + # Incremental + src = get_test_source() + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 87 + + + def test_postgres_destination(self): + + def get_test_destination(mode_config={}, connect_config={}): + + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config + + test_connect_config = { + 'auth_type': 'basic', + 'vendor_type': 'postgresql', + 'host': os.environ['POSTGRES_HOST'], + 'port': '5432', + 'user': os.environ['POSTGRES_USER'], + 'password': os.environ['POSTGRES_PASSWORD'], + 'database': os.environ['POSTGRES_DATABASE'], + 'target_schema': 'target' + } | connect_config + + return get_destination( + get_destination_by_vendor('postgresql'), + config = { + 'mode': Mode(test_mode_config), + 'connect': test_connect_config + } + ) + + def connect(): + return create_engine( + f"postgresql+psycopg2://{os.environ['POSTGRES_USER']}:{os.environ['POSTGRES_PASSWORD']}@"\ + f"{os.environ['POSTGRES_HOST']}:5432/{os.environ['POSTGRES_DATABASE']}" + ).connect() + + def drop(): + with connect() as conn: + conn.execute(text("DROP TABLE IF EXISTS target.pontoon_transfer_test")) + + + drop() + + # Cannot connect to destination + dest = get_test_destination(connect_config={'host': 'doesnotexist'}) + ds = get_memory_source().read(progress_callback=read_progress_handler) + with pytest.raises(DestinationConnectionFailed): + dest.write(ds, progress_callback=write_progress_handler) + + + # Existing destination schema is not compatible (destination changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + with connect() as conn: + conn.execute(text("ALTER TABLE target.pontoon_transfer_test DROP COLUMN customer_id")) + + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + drop() + + # Existing destination schema is not compatible (source changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + dest = get_test_destination() + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + # Full refresh should overwrite destination with different schema + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + drop() + + # Full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + # Full refresh followed by full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + + # Full refresh, then incremental load + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= TIMESTAMPTZ '2025-01-02 00:00:00+00' AND updated_at < TIMESTAMPTZ '2025-01-03 00:00:00+00'")).scalar() + assert count == 7 + + # Incremental reload same day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= TIMESTAMPTZ '2025-01-02 00:00:00+00' AND updated_at < TIMESTAMPTZ '2025-01-03 00:00:00+00'")).scalar() + assert count == 7 + + # Incremental load second day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= TIMESTAMPTZ '2025-01-03 00:00:00+00' AND updated_at < TIMESTAMPTZ '2025-01-04 00:00:00+00'")).scalar() + assert count == 7 + + # Incremental load third day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= TIMESTAMPTZ '2025-01-04 00:00:00+00' AND updated_at < TIMESTAMPTZ '2025-01-05 00:00:00+00'")).scalar() + assert count == 6 + + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + drop() + diff --git a/data-transfer/pontoon/tests/integration/test_redshift_connectors.py b/data-transfer/pontoon/tests/integration/test_redshift_connectors.py new file mode 100644 index 0000000..9b82163 --- /dev/null +++ b/data-transfer/pontoon/tests/integration/test_redshift_connectors.py @@ -0,0 +1,266 @@ +import os +import json +import glob +import uuid +import psutil +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from tests.integration.common import clear_cache_files, get_memory_source, read_progress_handler, write_progress_handler + +from dotenv import load_dotenv +load_dotenv() + + +clear_cache_files() + +class TestRedshiftConnectors: + + def test_redshift_source(self): + + def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) + } | mode_config + + test_with_config = { + 'batch_id': True, + 'last_sync': True + } | with_config + + test_connect_config = { + 'auth_type': 'basic', + 'vendor_type': 'redshift', + 'host': os.environ['REDSHIFT_HOST'], + 'port': '5439', + 'user': os.environ['REDSHIFT_USER'], + 'password': os.environ['REDSHIFT_PASSWORD'], + 'database': os.environ['REDSHIFT_DATABASE'] + } | connect_config + + test_streams_config = [{ + 'schema': 'source', + 'table': 'leads_xs', + 'primary_field': 'id', + 'cursor_field': 'updated_at', + 'filters': {'customer_id': 'Customer1'}, + 'drop_fields': ['customer_id'] + } | streams_config] + + return get_source( + get_source_by_vendor('redshift'), + config = { + 'mode': Mode(test_mode_config), + 'with': test_with_config, + 'connect': test_connect_config, + 'streams': test_streams_config + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_redshift_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) + + # Cannot connect to source + src = get_test_source(connect_config={'host': 'doesnotexist'}) + with pytest.raises(SourceConnectionFailed): + src.test_connect() + with pytest.raises(SourceConnectionFailed): + src.read(progress_callback=read_progress_handler) + + + # Source stream does not exist + src = get_test_source(streams_config={'table': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamDoesNotExist): + src.read(progress_callback=read_progress_handler) + + # Source stream field does not exist + src = get_test_source(streams_config={'primary_field': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamInvalidSchema): + src.read(progress_callback=read_progress_handler) + + # Empty record set + src = get_test_source(mode_config={'start': datetime(2010, 1, 1, tzinfo=timezone.utc), 'end': datetime(2010, 1, 2, tzinfo=timezone.utc)}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 0 + + # Full refresh + src = get_test_source(mode_config={'type': Mode.FULL_REFRESH}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 329 + + # Incremental + src = get_test_source() + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 87 + + + def test_redshift_destination(self): + + def get_test_destination(mode_config={}, connect_config={}): + + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config + + test_connect_config = { + 'auth_type': 'basic', + 'vendor_type': 'redshift', + 's3_bucket': os.environ['S3_BUCKET'], + 's3_prefix': 'pontoon-test', + 's3_region': os.environ['AWS_DEFAULT_REGION'], + 'iam_role': os.environ['REDSHIFT_IAM_ROLE'], + 'aws_access_key_id': os.environ['AWS_ACCESS_KEY_ID'], + 'aws_secret_access_key': os.environ['AWS_SECRET_ACCESS_KEY'], + 'host': os.environ['REDSHIFT_HOST'], + 'user': os.environ['REDSHIFT_USER'], + 'password': os.environ['REDSHIFT_PASSWORD'], + 'database': os.environ['REDSHIFT_DATABASE'], + 'port': '5439', + 'target_schema': 'target' + } | connect_config + + return get_destination( + get_destination_by_vendor('redshift'), + config = { + 'mode': Mode(test_mode_config), + 'connect': test_connect_config + } + ) + + def connect(): + return create_engine( + f"postgresql+psycopg2://{os.environ['REDSHIFT_USER']}:{os.environ['REDSHIFT_PASSWORD']}@"\ + f"{os.environ['REDSHIFT_HOST']}:5439/{os.environ['REDSHIFT_DATABASE']}" + ).connect() + + def drop(): + with connect() as conn: + conn.execute(text("DROP TABLE IF EXISTS target.pontoon_transfer_test")) + + + drop() + + # Cannot connect to destination + dest = get_test_destination(connect_config={'host': 'doesnotexist'}) + ds = get_memory_source().read(progress_callback=read_progress_handler) + with pytest.raises(DestinationConnectionFailed): + dest.write(ds, progress_callback=write_progress_handler) + + + # Existing destination schema is not compatible (destination changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + with connect() as conn: + conn.execute(text("ALTER TABLE target.pontoon_transfer_test DROP COLUMN customer_id")) + + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + drop() + + # Existing destination schema is not compatible (source changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + dest = get_test_destination() + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + # Full refresh should overwrite destination with different schema + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + drop() + + # Full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + # Full refresh followed by full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + + # Full refresh, then incremental load + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-02' AND updated_at < '2025-01-03'")).scalar() + assert count == 7 + + # Incremental reload same day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-02' AND updated_at < '2025-01-03'")).scalar() + assert count == 7 + + # Incremental load second day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-03' AND updated_at < '2025-01-04'")).scalar() + assert count == 7 + + # Incremental load third day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-04' AND updated_at < '2025-01-05'")).scalar() + assert count == 6 + + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + drop() + diff --git a/data-transfer/pontoon/tests/integration/test_snowflake_connectors.py b/data-transfer/pontoon/tests/integration/test_snowflake_connectors.py new file mode 100644 index 0000000..58cd391 --- /dev/null +++ b/data-transfer/pontoon/tests/integration/test_snowflake_connectors.py @@ -0,0 +1,263 @@ +import os +import json +import glob +import uuid +import psutil +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, inspect, MetaData, Table, text + +from pontoon import configure_logging +from pontoon import get_source, get_destination, get_source_by_vendor, get_destination_by_vendor +from pontoon import SqliteCache +from pontoon import Progress, Mode +from pontoon import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema +from pontoon import DestinationConnectionFailed, DestinationStreamInvalidSchema + +from tests.integration.common import clear_cache_files, get_memory_source, read_progress_handler, write_progress_handler + +from dotenv import load_dotenv +load_dotenv() + + +clear_cache_files() + +class TestSnowflakeConnectors: + + def test_snowflake_source(self): + + def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 7, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 7, 2, tzinfo=timezone.utc) + } | mode_config + + test_with_config = { + 'batch_id': True, + 'last_sync': True + } | with_config + + test_connect_config = { + 'auth_type': 'access_token', + 'vendor_type': 'snowflake', + 'user': os.environ['SNOWFLAKE_USER'], + 'access_token': os.environ['SNOWFLAKE_ACCESS_TOKEN'], + 'account': os.environ['SNOWFLAKE_ACCOUNT'], + 'database': os.environ['SNOWFLAKE_DATABASE'], + 'warehouse': os.environ['SNOWFLAKE_WAREHOUSE'] + } | connect_config + + test_streams_config = [{ + 'schema': 'pontoon', + 'table': 'leads_xs', + 'primary_field': 'id', + 'cursor_field': 'updated_at', + 'filters': {'customer_id': 'Customer1'}, + 'drop_fields': ['customer_id'] + } | streams_config] + + return get_source( + get_source_by_vendor('snowflake'), + config = { + 'mode': Mode(test_mode_config), + 'with': test_with_config, + 'connect': test_connect_config, + 'streams': test_streams_config + }, + cache_implementation = SqliteCache, + cache_config = { + 'db': f"_snowflake_{uuid.uuid4()}_cache.db", + 'chunk_size': 1024 + } + ) + + # Cannot connect to source + src = get_test_source(connect_config={'account': 'doesnotexist'}) + with pytest.raises(SourceConnectionFailed): + src.test_connect() + with pytest.raises(SourceConnectionFailed): + src.read(progress_callback=read_progress_handler) + + + # Source stream does not exist + src = get_test_source(streams_config={'table': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamDoesNotExist): + src.read(progress_callback=read_progress_handler) + + # Source stream field does not exist + src = get_test_source(streams_config={'primary_field': 'doesnotexist'}) + assert src.test_connect() == True + with pytest.raises(SourceStreamInvalidSchema): + src.read(progress_callback=read_progress_handler) + + # Empty record set + src = get_test_source(mode_config={'start': datetime(2010, 1, 1, tzinfo=timezone.utc), 'end': datetime(2010, 1, 2, tzinfo=timezone.utc)}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 0 + + # Full refresh + src = get_test_source(mode_config={'type': Mode.FULL_REFRESH}) + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 329 + + # Incremental + src = get_test_source() + ds = src.read(progress_callback=read_progress_handler) + assert ds.size(ds.streams[0]) == 87 + + + def test_snowflake_destination(self): + + def get_test_destination(mode_config={}, connect_config={}): + + test_mode_config = { + 'type': Mode.INCREMENTAL, + 'period': Mode.DAILY, + 'start': datetime(2025, 1, 1, tzinfo=timezone.utc), + 'end': datetime(2025, 1, 2, tzinfo=timezone.utc) + } | mode_config + + test_connect_config = { + 'auth_type': 'access_token', + 'vendor_type': 'snowflake', + 'stage_name': 'pontoon', + 'create_stage': True, + 'delete_stage': True, + 'user': os.environ['SNOWFLAKE_USER'], + 'access_token': os.environ['SNOWFLAKE_ACCESS_TOKEN'], + 'account': os.environ['SNOWFLAKE_ACCOUNT'], + 'database': os.environ['SNOWFLAKE_DATABASE'], + 'warehouse': os.environ['SNOWFLAKE_WAREHOUSE'], + 'target_schema': 'target' + } | connect_config + + return get_destination( + get_destination_by_vendor('snowflake'), + config = { + 'mode': Mode(test_mode_config), + 'connect': test_connect_config + } + ) + + def connect(): + return create_engine( + f"snowflake://{os.environ['SNOWFLAKE_USER']}:{os.environ['SNOWFLAKE_ACCESS_TOKEN']}@"\ + f"{os.environ['SNOWFLAKE_ACCOUNT']}/{os.environ['SNOWFLAKE_DATABASE']}?warehouse={os.environ['SNOWFLAKE_WAREHOUSE']}" + ).connect() + + def drop(): + with connect() as conn: + conn.execute(text("DROP TABLE IF EXISTS target.pontoon_transfer_test")) + + + drop() + + # Cannot connect to destination + dest = get_test_destination(connect_config={'account': 'doesnotexist'}) + ds = get_memory_source().read(progress_callback=read_progress_handler) + with pytest.raises(DestinationConnectionFailed): + dest.write(ds, progress_callback=write_progress_handler) + + + # Existing destination schema is not compatible (destination changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + with connect() as conn: + conn.execute(text("ALTER TABLE target.pontoon_transfer_test DROP COLUMN customer_id")) + + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + drop() + + # Existing destination schema is not compatible (source changed) + dest = get_test_destination() + ds = get_memory_source().read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + + dest = get_test_destination() + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + with pytest.raises(DestinationStreamInvalidSchema): + dest.write(ds, progress_callback=write_progress_handler) + + # Full refresh should overwrite destination with different schema + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + drop() + + # Full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + # Full refresh followed by full refresh + dest = get_test_destination(mode_config={'type': Mode.FULL_REFRESH}) + ds = get_memory_source(mode_config={'type': Mode.FULL_REFRESH}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + + # Full refresh, then incremental load + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-02' AND updated_at < '2025-01-03'")).scalar() + assert count == 7 + + # Incremental reload same day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 2, tzinfo=timezone.utc), 'end': datetime(2025, 1, 3, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-02' AND updated_at < '2025-01-03'")).scalar() + assert count == 7 + + # Incremental load second day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 3, tzinfo=timezone.utc), 'end': datetime(2025, 1, 4, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-03' AND updated_at < '2025-01-04'")).scalar() + assert count == 7 + + # Incremental load third day + dest = get_test_destination(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}) + ds = get_memory_source(mode_config={'start': datetime(2025, 1, 4, tzinfo=timezone.utc), 'end': datetime(2025, 1, 5, tzinfo=timezone.utc)}).read(progress_callback=read_progress_handler) + dest.write(ds, progress_callback=write_progress_handler) + dest.integrity().check_batch_volume(ds) + + with connect() as conn: + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test WHERE customer_id='Customer1' AND updated_at >= '2025-01-04' AND updated_at < '2025-01-05'")).scalar() + assert count == 6 + + count = conn.execute(text("SELECT COUNT(1) FROM target.pontoon_transfer_test")).scalar() + assert count == 29 + + drop() + From 5e27ad97ba53db697765cac873ecc88205f05a8f Mon Sep 17 00:00:00 2001 From: Kalan MacRow Date: Tue, 12 Aug 2025 12:29:56 -0400 Subject: [PATCH 2/7] tidy and comments --- .../integration/test_bigquery_connectors.py | 6 ----- .../integration/test_postgres_connectors.py | 25 +++++++++++++++++++ 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py b/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py index fbee123..8ea3bd3 100644 --- a/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py +++ b/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py @@ -144,12 +144,6 @@ def drop(): drop() - # Cannot connect to destination - # dest = get_test_destination(connect_config={'host': 'doesnotexist'}) - # ds = get_memory_source().read(progress_callback=read_progress_handler) - # with pytest.raises(DestinationConnectionFailed): - # dest.write(ds, progress_callback=write_progress_handler) - # Existing destination schema is not compatible (destination changed) dest = get_test_destination() diff --git a/data-transfer/pontoon/tests/integration/test_postgres_connectors.py b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py index 32022b3..29c9183 100644 --- a/data-transfer/pontoon/tests/integration/test_postgres_connectors.py +++ b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py @@ -26,6 +26,17 @@ class TestPostgresConnectors: def test_postgres_source(self): + + """ + How this test works: + - Assumes you have a Postgresql instance with the test data loaded (see README) + - Assumes you have configured corresponding ENV vars in .env (see README) + - Instantiates the Postgresql source connector and checks: + - Bad connection info raises the expected errors + - Missing source tables and columns raise the expected errors + - Empty result sets are handled gracefully + - Full refresh and incremental replication modes return the expected results + """ def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_config={}): test_mode_config = { @@ -112,6 +123,20 @@ def get_test_source(mode_config={}, with_config={}, connect_config={}, streams_c def test_postgres_destination(self): + """ + How this test works: + - Assumes you have a Postgresql instance with database and "target" schema created + - Assumes you have configured corresponding ENV vars in .env (see README) + - Uses the "memory" source as a static data source + - Instantiates the Postgresql destination connector and checks: + - Bad connection info raises the expected errors + - Incompatible schemas between source stream and destination raise expected errors + - Full refresh overwrites existing target tables + - Back to back full refreshes + - Full refresh followed by sequence of incremental syncs + """ + + def get_test_destination(mode_config={}, connect_config={}): test_mode_config = { From bfac2f267aa1bc1c2ffd7b312ceb54591664c382 Mon Sep 17 00:00:00 2001 From: Kalan MacRow Date: Tue, 12 Aug 2025 13:47:52 -0400 Subject: [PATCH 3/7] add gh workflow to run pg integration tests on PRs --- .../data-transfer-integration-tests.yml | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 .github/workflows/data-transfer-integration-tests.yml diff --git a/.github/workflows/data-transfer-integration-tests.yml b/.github/workflows/data-transfer-integration-tests.yml new file mode 100644 index 0000000..ed44a60 --- /dev/null +++ b/.github/workflows/data-transfer-integration-tests.yml @@ -0,0 +1,72 @@ +name: Data Transfer Integration Tests + +on: + pull_request: + +jobs: + postgresql-connector: + runs-on: ubuntu-latest + + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: test + POSTGRES_PASSWORD: test + POSTGRES_DB: pontoon + ports: + - 5432:5432 + options: >- + --health-cmd="pg_isready -U test" + --health-interval=10s + --health-timeout=5s + --health-retries=5 + + env: + ENV: test + POSTGRES_USER: test + POSTGRES_PASSWORD: test + POSTGRES_DATABASE: pontoon + POSTGRES_HOST: localhost + POSTGRES_PORT: 5432 + ALLOW_ORIGIN: http://localhost:3000 + JWT_ALGORITHM: HS256 + JWT_SIGNING_KEY: test_key + SKIP_TRANSFERS: true + + steps: + - name: Checkout repo + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: 3.12 + + - name: Wait for DB to be ready + run: | + for i in {1..10}; do + pg_isready -h localhost -U test && break + echo "Waiting for postgres..." && sleep 2 + done + + - name: Run integration tests + working-directory: data-transfer/pontoon + run: | + python -m pip install --upgrade pip + pip install pytest + pip install . + + cat < .env + POSTGRES_HOST=localhost + POSTGRES_USER=test + POSTGRES_PASSWORD=test + POSTGRES_DATABASE=pontoon + EOF + + PGPASSWORD=test psql -h localhost -U test -d pontoon -c "CREATE SCHEMA source;" + PGPASSWORD=test psql -h localhost -U test -d pontoon -c "CREATE SCHEMA target;" + PGPASSWORD=test psql -h localhost -U test -d pontoon -c "CREATE TABLE source.leads_xs (id uuid NOT NULL,created_at timestamp NOT NULL,updated_at timestamp NOT NULL,customer_id text NOT NULL,name text NOT NULL,email text NOT NULL,score integer NOT NULL,notes text);" + PGPASSWORD=test psql -h localhost -U test -d pontoon -c "\COPY source.leads_xs(id,created_at,updated_at,customer_id,name,email,score,notes) FROM 'tests/data/leads_cx_20250701.csv' CSV HEADER;" + + pytest -s tests/integration/test_postgres_connectors.py From e9f4490df0460fb9bbbdab6cd5449e2d9bee6396 Mon Sep 17 00:00:00 2001 From: Kalan MacRow Date: Tue, 12 Aug 2025 13:50:19 -0400 Subject: [PATCH 4/7] fix csv name in workflow --- .github/workflows/data-transfer-integration-tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/data-transfer-integration-tests.yml b/.github/workflows/data-transfer-integration-tests.yml index ed44a60..50cc5eb 100644 --- a/.github/workflows/data-transfer-integration-tests.yml +++ b/.github/workflows/data-transfer-integration-tests.yml @@ -67,6 +67,6 @@ jobs: PGPASSWORD=test psql -h localhost -U test -d pontoon -c "CREATE SCHEMA source;" PGPASSWORD=test psql -h localhost -U test -d pontoon -c "CREATE SCHEMA target;" PGPASSWORD=test psql -h localhost -U test -d pontoon -c "CREATE TABLE source.leads_xs (id uuid NOT NULL,created_at timestamp NOT NULL,updated_at timestamp NOT NULL,customer_id text NOT NULL,name text NOT NULL,email text NOT NULL,score integer NOT NULL,notes text);" - PGPASSWORD=test psql -h localhost -U test -d pontoon -c "\COPY source.leads_xs(id,created_at,updated_at,customer_id,name,email,score,notes) FROM 'tests/data/leads_cx_20250701.csv' CSV HEADER;" + PGPASSWORD=test psql -h localhost -U test -d pontoon -c "\COPY source.leads_xs(id,created_at,updated_at,customer_id,name,email,score,notes) FROM 'tests/data/leads_xs_20250701.csv' CSV HEADER;" pytest -s tests/integration/test_postgres_connectors.py From c27a1b14026872ccdf199f0b3bca5d3e97558d32 Mon Sep 17 00:00:00 2001 From: Kalan MacRow Date: Tue, 12 Aug 2025 13:54:27 -0400 Subject: [PATCH 5/7] remove psutil dependency --- data-transfer/pontoon/tests/integration/common.py | 1 - .../pontoon/tests/integration/test_bigquery_connectors.py | 1 - .../pontoon/tests/integration/test_postgres_connectors.py | 1 - .../pontoon/tests/integration/test_redshift_connectors.py | 3 +-- .../pontoon/tests/integration/test_snowflake_connectors.py | 1 - 5 files changed, 1 insertion(+), 6 deletions(-) diff --git a/data-transfer/pontoon/tests/integration/common.py b/data-transfer/pontoon/tests/integration/common.py index e2dd735..97fc0c0 100644 --- a/data-transfer/pontoon/tests/integration/common.py +++ b/data-transfer/pontoon/tests/integration/common.py @@ -2,7 +2,6 @@ import json import glob import uuid -import psutil import pytest from datetime import datetime, timezone diff --git a/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py b/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py index 8ea3bd3..4e1c92c 100644 --- a/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py +++ b/data-transfer/pontoon/tests/integration/test_bigquery_connectors.py @@ -2,7 +2,6 @@ import json import glob import uuid -import psutil import pytest from datetime import datetime, timezone diff --git a/data-transfer/pontoon/tests/integration/test_postgres_connectors.py b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py index 29c9183..f973498 100644 --- a/data-transfer/pontoon/tests/integration/test_postgres_connectors.py +++ b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py @@ -2,7 +2,6 @@ import json import glob import uuid -import psutil import pytest from datetime import datetime, timezone diff --git a/data-transfer/pontoon/tests/integration/test_redshift_connectors.py b/data-transfer/pontoon/tests/integration/test_redshift_connectors.py index 9b82163..c10c073 100644 --- a/data-transfer/pontoon/tests/integration/test_redshift_connectors.py +++ b/data-transfer/pontoon/tests/integration/test_redshift_connectors.py @@ -2,7 +2,6 @@ import json import glob import uuid -import psutil import pytest from datetime import datetime, timezone @@ -195,7 +194,7 @@ def drop(): ds = get_memory_source(streams_config={'drop_fields': ['customer_id']}).read(progress_callback=read_progress_handler) dest.write(ds, progress_callback=write_progress_handler) dest.integrity().check_batch_volume(ds) - + drop() # Full refresh diff --git a/data-transfer/pontoon/tests/integration/test_snowflake_connectors.py b/data-transfer/pontoon/tests/integration/test_snowflake_connectors.py index 58cd391..d02042e 100644 --- a/data-transfer/pontoon/tests/integration/test_snowflake_connectors.py +++ b/data-transfer/pontoon/tests/integration/test_snowflake_connectors.py @@ -2,7 +2,6 @@ import json import glob import uuid -import psutil import pytest from datetime import datetime, timezone From b80fdd3131d5d11d61aadbe7914846dd5d61ed19 Mon Sep 17 00:00:00 2001 From: Kalan MacRow Date: Tue, 12 Aug 2025 13:57:32 -0400 Subject: [PATCH 6/7] add dotenv dependency --- .github/workflows/data-transfer-integration-tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/data-transfer-integration-tests.yml b/.github/workflows/data-transfer-integration-tests.yml index 50cc5eb..05fe902 100644 --- a/.github/workflows/data-transfer-integration-tests.yml +++ b/.github/workflows/data-transfer-integration-tests.yml @@ -54,7 +54,7 @@ jobs: working-directory: data-transfer/pontoon run: | python -m pip install --upgrade pip - pip install pytest + pip install pytest python-dotenv pip install . cat < .env From 1175eae0571ebf12afeae31be3b1bb05458d30f5 Mon Sep 17 00:00:00 2001 From: Kalan MacRow Date: Tue, 12 Aug 2025 14:27:22 -0400 Subject: [PATCH 7/7] update README and include path in comments --- data-transfer/pontoon/README.md | 3 ++- .../pontoon/tests/integration/test_postgres_connectors.py | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/data-transfer/pontoon/README.md b/data-transfer/pontoon/README.md index 54a41d7..1b94eaf 100644 --- a/data-transfer/pontoon/README.md +++ b/data-transfer/pontoon/README.md @@ -50,5 +50,6 @@ Create schemas named `source` and `target`. Load the `tests/data/leads_xs_*.csv` Create schemas named `pontoon` and `target`. Load the same test data into `pontoon.leads_xs` ## Running +Install test dependencies: `pytest` and `python-dotenv`. -`make tests` or use `pytest` directly to run specific sets of tests. +`make test`, `make test-integration` or use `pytest` directly to run specific sets of tests, e.g. `pytest -s tests/integration/test_postgres_connectors.py` diff --git a/data-transfer/pontoon/tests/integration/test_postgres_connectors.py b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py index f973498..3dc7b2c 100644 --- a/data-transfer/pontoon/tests/integration/test_postgres_connectors.py +++ b/data-transfer/pontoon/tests/integration/test_postgres_connectors.py @@ -28,8 +28,8 @@ def test_postgres_source(self): """ How this test works: - - Assumes you have a Postgresql instance with the test data loaded (see README) - - Assumes you have configured corresponding ENV vars in .env (see README) + - Assumes you have a Postgresql instance with the test data loaded (see data-transfer/pontoon/README.md) + - Assumes you have configured corresponding ENV vars in .env (see see data-transfer/pontoon/README.md) - Instantiates the Postgresql source connector and checks: - Bad connection info raises the expected errors - Missing source tables and columns raise the expected errors