From 4546bfe204909220c86a9a0c996b393d5bc5849a Mon Sep 17 00:00:00 2001 From: gl-romil <129256325+gl-romil@users.noreply.github.com> Date: Fri, 3 Oct 2025 17:36:33 +0530 Subject: [PATCH 01/11] Meta data additon forced replcation key --- tap_eloqua/discover.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/tap_eloqua/discover.py b/tap_eloqua/discover.py index 782d620..d9c685a 100644 --- a/tap_eloqua/discover.py +++ b/tap_eloqua/discover.py @@ -1,4 +1,5 @@ from singer.catalog import Catalog, CatalogEntry, Schema +from singer import metadata as mdata # minimal addition from tap_eloqua.schema import get_schemas, get_pk @@ -8,7 +9,14 @@ def discover(client): for stream_name, schema_dict in schemas.items(): schema = Schema.from_dict(schema_dict) - metadata = field_metadata[stream_name] + + # existing field-level metadata from schema.py + md_list = field_metadata[stream_name] + m = mdata.to_map(md_list) + m = mdata.write(m, (), 'inclusion', 'available') + m = mdata.write(m, (), 'forced-replication-method', 'FULL_TABLE') + md_list = mdata.to_list(m) + pk = get_pk(stream_name) catalog.streams.append(CatalogEntry( @@ -16,7 +24,7 @@ def discover(client): tap_stream_id=stream_name, key_properties=pk, schema=schema, - metadata=metadata + metadata=md_list )) return catalog From c397e9665f53fcd4e6c462b289a97b27dc3ca19d Mon Sep 17 00:00:00 2001 From: Pagadala Narender Date: Fri, 28 Nov 2025 14:41:34 +0530 Subject: [PATCH 02/11] removing inclusion --- tap_eloqua/discover.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tap_eloqua/discover.py b/tap_eloqua/discover.py index d9c685a..0cfc9f6 100644 --- a/tap_eloqua/discover.py +++ b/tap_eloqua/discover.py @@ -13,7 +13,6 @@ def discover(client): # existing field-level metadata from schema.py md_list = field_metadata[stream_name] m = mdata.to_map(md_list) - m = mdata.write(m, (), 'inclusion', 'available') m = mdata.write(m, (), 'forced-replication-method', 'FULL_TABLE') md_list = mdata.to_list(m) From 05a751d71d00a28efb7b9b1511dded7d38bd1ba5 Mon Sep 17 00:00:00 2001 From: Pagadala Narender Date: Fri, 28 Nov 2025 14:56:22 +0530 Subject: [PATCH 03/11] changes for the dynamic method --- tap_eloqua/discover.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/tap_eloqua/discover.py b/tap_eloqua/discover.py index 0cfc9f6..2996ad3 100644 --- a/tap_eloqua/discover.py +++ b/tap_eloqua/discover.py @@ -9,15 +9,14 @@ def discover(client): for stream_name, schema_dict in schemas.items(): schema = Schema.from_dict(schema_dict) - - # existing field-level metadata from schema.py + properties = schema_dict.get("properties", {}) + replication_key = "updatedAt" if "UpdatedAt" in properties else None + replication_method = "INCREMENTAL" if replication_key else "FULL_TABLE" md_list = field_metadata[stream_name] m = mdata.to_map(md_list) - m = mdata.write(m, (), 'forced-replication-method', 'FULL_TABLE') + m = mdata.write(m, (), 'forced-replication-method', replication_method) md_list = mdata.to_list(m) - pk = get_pk(stream_name) - catalog.streams.append(CatalogEntry( stream=stream_name, tap_stream_id=stream_name, From 0b47da570e73100ebca34a535a1ed9bd27e8682c Mon Sep 17 00:00:00 2001 From: Pagadala Narender Date: Fri, 28 Nov 2025 18:09:10 +0530 Subject: [PATCH 04/11] added unittest case --- test/test_discovery_metadata.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) create mode 100644 test/test_discovery_metadata.py diff --git a/test/test_discovery_metadata.py b/test/test_discovery_metadata.py new file mode 100644 index 0000000..1b7ec60 --- /dev/null +++ b/test/test_discovery_metadata.py @@ -0,0 +1,26 @@ +import unittest +from singer import metadata as mdata +from tap_eloqua import discover +from unittest.mock import MagicMock + +class TestEloquaDiscoveryMetadata(unittest.TestCase): + + def test_metadata_updates(self): + client = MagicMock() + catalog = discover(client) + + for s in catalog.streams: + md_map = mdata.to_map(s.metadata) + self.assertIn((), md_map, msg=f"Root metadata missing for stream {s.stream}") + root = md_map[()] + + # compute expected replication method from schema properties + schema_props = s.schema.to_dict().get("properties", {}) + expected_rep_method = "INCREMENTAL" if "UpdatedAt" in schema_props else "FULL_TABLE" + + self.assertEqual(root.get("forced-replication-method"), + expected_rep_method, + msg=f"forced-replication-method mismatch for stream {s.stream}") + +if __name__ == "__main__": + unittest.main() From 7103c0c019d1ef7ac02dcfa62448428b14c79f1e Mon Sep 17 00:00:00 2001 From: Pagadala Narender Date: Fri, 28 Nov 2025 19:21:42 +0530 Subject: [PATCH 05/11] changes for the test cases --- tap_eloqua/discover.py | 2 +- test/test_discovery_metadata.py | 30 ++++++++++++++++++------------ 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/tap_eloqua/discover.py b/tap_eloqua/discover.py index 2996ad3..45f4578 100644 --- a/tap_eloqua/discover.py +++ b/tap_eloqua/discover.py @@ -10,7 +10,7 @@ def discover(client): for stream_name, schema_dict in schemas.items(): schema = Schema.from_dict(schema_dict) properties = schema_dict.get("properties", {}) - replication_key = "updatedAt" if "UpdatedAt" in properties else None + replication_key = next((k for k in properties if k.lower() == "updatedat"), None) replication_method = "INCREMENTAL" if replication_key else "FULL_TABLE" md_list = field_metadata[stream_name] m = mdata.to_map(md_list) diff --git a/test/test_discovery_metadata.py b/test/test_discovery_metadata.py index 1b7ec60..30f1fb9 100644 --- a/test/test_discovery_metadata.py +++ b/test/test_discovery_metadata.py @@ -4,23 +4,29 @@ from unittest.mock import MagicMock class TestEloquaDiscoveryMetadata(unittest.TestCase): - - def test_metadata_updates(self): + def test_replication_methods_for_known_streams(self): client = MagicMock() catalog = discover(client) + # Hardcoded expected replication methods for certain streams: + expected_methods = { + "accounts": "INCREMENTAL", + "contacts": "INCREMENTAL", + "visitors": "FULL_TABLE", + } for s in catalog.streams: + stream_name = s.stream.lower() + if stream_name not in expected_methods: + continue + md_map = mdata.to_map(s.metadata) - self.assertIn((), md_map, msg=f"Root metadata missing for stream {s.stream}") + self.assertIn((), md_map, msg=f"Root metadata missing for stream {stream_name}") root = md_map[()] - - # compute expected replication method from schema properties - schema_props = s.schema.to_dict().get("properties", {}) - expected_rep_method = "INCREMENTAL" if "UpdatedAt" in schema_props else "FULL_TABLE" - - self.assertEqual(root.get("forced-replication-method"), - expected_rep_method, - msg=f"forced-replication-method mismatch for stream {s.stream}") + self.assertEqual( + root.get("forced-replication-method"), + expected_methods[stream_name], + msg=f"Replication method mismatch for stream {stream_name}" + ) if __name__ == "__main__": - unittest.main() + unittest.main() \ No newline at end of file From 69f41f3b6fdaa29317fd263fe3d2f30d9f618dbc Mon Sep 17 00:00:00 2001 From: Pagadala Narender Date: Mon, 1 Dec 2025 08:00:11 +0530 Subject: [PATCH 06/11] changes for the version release details --- CHANGELOG.md | 3 +++ setup.py | 2 +- tap_eloqua/discover.py | 5 ++++- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b03e7c6..59b3756 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ # Changelog +## [1.4.0] 2025-11-28 +- Added replication method into metadata [#49](https://github.com/singer-io/tap-eloqua/pull/49) + ## [1.3.1] 2025-06-23 - Bump backoff to `1.10.0`, bump requests to `2.32.4`, bump singer-python to 5.13.2 diff --git a/setup.py b/setup.py index fe7565d..0f2848f 100644 --- a/setup.py +++ b/setup.py @@ -3,7 +3,7 @@ from setuptools import setup setup(name='tap-eloqua', - version='1.3.1', + version='1.4.0', description='Singer.io tap for extracting data from the Oracle Eloqua API', author='Stitch', url='https://singer.io', diff --git a/tap_eloqua/discover.py b/tap_eloqua/discover.py index 45f4578..93f7e42 100644 --- a/tap_eloqua/discover.py +++ b/tap_eloqua/discover.py @@ -10,7 +10,10 @@ def discover(client): for stream_name, schema_dict in schemas.items(): schema = Schema.from_dict(schema_dict) properties = schema_dict.get("properties", {}) - replication_key = next((k for k in properties if k.lower() == "updatedat"), None) + replication_key = next( + (k for k in properties if k.lower() == "updatedat" or k == "updatedAt"), + None + ) replication_method = "INCREMENTAL" if replication_key else "FULL_TABLE" md_list = field_metadata[stream_name] m = mdata.to_map(md_list) From c2094f742db9977fd048b596d24b5029e4f309e7 Mon Sep 17 00:00:00 2001 From: Mukesh Bhatt Date: Mon, 20 Apr 2026 11:47:47 +0530 Subject: [PATCH 07/11] upgrade python and added unittests (#50) * upgrade python and added unittests * Update tests/test_sync.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update tests/test_sync.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * review comments * resolve review comments * remove out of scope changes * request lib to latest * fix review comments * resolve review comments * resolve review comments * updated changelog --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Ranit Saha --- .circleci/config.yml | 10 +- CHANGELOG.md | 3 +- setup.py | 8 +- tests/__init__.py | 0 tests/base.py | 283 ++++++++++++++ tests/test_all_fields.py | 80 ++++ tests/test_automatic_fields.py | 44 +++ tests/test_bookmark.py | 86 +++++ tests/test_discovery_enhanced.py | 231 ++++++++++++ {test => tests}/test_discovery_metadata.py | 0 tests/test_pagination.py | 69 ++++ tests/test_start_date.py | 123 ++++++ tests/test_sync.py | 391 ++++++++++++++++++++ tests/unittests/__init__.py | 0 tests/unittests/test_discover_unit.py | 99 +++++ tests/unittests/test_schema_unit.py | 46 +++ tests/unittests/test_sync_helpers_unit.py | 81 ++++ tests/unittests/test_sync_selection_unit.py | 94 +++++ 18 files changed, 1641 insertions(+), 7 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/base.py create mode 100644 tests/test_all_fields.py create mode 100644 tests/test_automatic_fields.py create mode 100644 tests/test_bookmark.py create mode 100644 tests/test_discovery_enhanced.py rename {test => tests}/test_discovery_metadata.py (100%) create mode 100644 tests/test_pagination.py create mode 100644 tests/test_start_date.py create mode 100644 tests/test_sync.py create mode 100644 tests/unittests/__init__.py create mode 100644 tests/unittests/test_discover_unit.py create mode 100644 tests/unittests/test_schema_unit.py create mode 100644 tests/unittests/test_sync_helpers_unit.py create mode 100644 tests/unittests/test_sync_selection_unit.py diff --git a/.circleci/config.yml b/.circleci/config.yml index 8af28aa..a8ad341 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -8,7 +8,7 @@ jobs: - run: name: 'Setup virtual env' command: | - uv venv --python 3.9 /usr/local/share/virtualenvs/tap-eloqua + uv venv --python 3.12 /usr/local/share/virtualenvs/tap-eloqua source /usr/local/share/virtualenvs/tap-eloqua/bin/activate uv pip install -U pip setuptools uv pip install .[dev] @@ -16,7 +16,13 @@ jobs: name: 'JSON Validator' command: | source /usr/local/share/virtualenvs/tap-tester/bin/activate - stitch-validate-json /usr/local/share/virtualenvs/tap-eloqua/lib/python3.9/site-packages/tap_eloqua/schemas/*.json + stitch-validate-json /usr/local/share/virtualenvs/tap-eloqua/lib/python3.12/site-packages/tap_eloqua/schemas/*.json + - run: + name: 'Integration Tests' + command: | + source /usr/local/share/virtualenvs/tap-eloqua/bin/activate + python -m unittest discover -s tests -p 'test_*.py' --top-level-directory . + workflows: version: 2 commit: diff --git a/CHANGELOG.md b/CHANGELOG.md index 59b3756..ff467b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,8 @@ # Changelog -## [1.4.0] 2025-11-28 +## [1.4.0] 2026-04-08 - Added replication method into metadata [#49](https://github.com/singer-io/tap-eloqua/pull/49) +- Bump backoff, requests, pendulum, singer-python version [#50](https://github.com/singer-io/tap-eloqua/pull/50) ## [1.3.1] 2025-06-23 - Bump backoff to `1.10.0`, bump requests to `2.32.4`, bump diff --git a/setup.py b/setup.py index 0f2848f..5ccd5a2 100644 --- a/setup.py +++ b/setup.py @@ -10,10 +10,10 @@ classifiers=['Programming Language :: Python :: 3 :: Only'], py_modules=['tap_eloqua'], install_requires=[ - 'backoff==1.10.0', - 'requests==2.32.4', - 'pendulum==2.0.3', - 'singer-python==5.13.2' + 'backoff==2.2.1', + 'requests==2.33.1', + 'pendulum==3.2.0', + 'singer-python==6.8.0' ], extras_require={ 'dev': [ diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/base.py b/tests/base.py new file mode 100644 index 0000000..7f59ac4 --- /dev/null +++ b/tests/base.py @@ -0,0 +1,283 @@ +"""Base test class for tap-eloqua with common test utilities.""" +import unittest +from unittest.mock import MagicMock + +from singer.catalog import Catalog + +from tap_eloqua.schema import get_schemas + + +class EloquaBaseTest(unittest.TestCase): + """Base class for eloqua tapTests with common setup and utilities.""" + + # Metadata keys for organization + PRIMARY_KEYS = "primary_keys" + REPLICATION_METHOD = "replication_method" + REPLICATION_KEYS = "replication_keys" + OBEYS_START_DATE = "obeys_start_date" + + # Class-level schema cache shared across all subclasses + _schemas_cache = None + _field_metadata_cache = None + + @classmethod + def setUpClass(cls): + super().setUpClass() + if EloquaBaseTest._schemas_cache is None: + EloquaBaseTest._schemas_cache, EloquaBaseTest._field_metadata_cache = get_schemas( + MagicMock(get=lambda *a, **k: {"items": []}) + ) + + @classmethod + def expected_metadata(cls): + """ + Return expected metadata for known eloqua streams. + Maps stream name to expected properties. + """ + return { + # Built-in bulk objects (INCREMENTAL) + "accounts": { + cls.PRIMARY_KEYS: {"Id"}, + cls.REPLICATION_METHOD: "INCREMENTAL", + cls.REPLICATION_KEYS: {"UpdatedAt"}, + cls.OBEYS_START_DATE: True, + }, + "contacts": { + cls.PRIMARY_KEYS: {"Id"}, + cls.REPLICATION_METHOD: "INCREMENTAL", + cls.REPLICATION_KEYS: {"UpdatedAt"}, + cls.OBEYS_START_DATE: True, + }, + # Activity streams (FULL_TABLE - no primary key) + "activity_email_open": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_email_clickthrough": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_email_send": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_form_submit": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_subscribe": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_unsubscribe": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_bounceback": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_web_visit": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + "activity_page_view": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: False, + }, + # Static REST endpoints + "visitors": { + cls.PRIMARY_KEYS: set(), + cls.REPLICATION_METHOD: "FULL_TABLE", + cls.REPLICATION_KEYS: set(), + cls.OBEYS_START_DATE: True, + }, + "campaigns": { + cls.PRIMARY_KEYS: {"id"}, + cls.REPLICATION_METHOD: "INCREMENTAL", + cls.REPLICATION_KEYS: {"updatedAt"}, + cls.OBEYS_START_DATE: True, + }, + "emails": { + cls.PRIMARY_KEYS: {"id"}, + cls.REPLICATION_METHOD: "INCREMENTAL", + cls.REPLICATION_KEYS: {"updatedAt"}, + cls.OBEYS_START_DATE: True, + }, + "forms": { + cls.PRIMARY_KEYS: {"id"}, + cls.REPLICATION_METHOD: "INCREMENTAL", + cls.REPLICATION_KEYS: {"updatedAt"}, + cls.OBEYS_START_DATE: True, + }, + "assets": { + cls.PRIMARY_KEYS: {"id"}, + cls.REPLICATION_METHOD: "INCREMENTAL", + cls.REPLICATION_KEYS: {"updatedAt"}, + cls.OBEYS_START_DATE: True, + }, + "emailGroups": { + cls.PRIMARY_KEYS: {"id"}, + cls.REPLICATION_METHOD: "INCREMENTAL", + cls.REPLICATION_KEYS: {"updatedAt"}, + cls.OBEYS_START_DATE: True, + }, + } + + @staticmethod + def _schema_type(schema): + """Return concrete type when schema allows null union types.""" + schema_type = schema.get("type", "object") + if isinstance(schema_type, list): + # Handle ["null", "string"] type definitions + non_null = [item for item in schema_type if item != "null"] + return non_null[0] if non_null else "null" + return schema_type + + @staticmethod + def _generate_value(schema, date_value="2024-01-01T00:00:00Z"): + """Generate one valid mock value for a JSON-schema fragment.""" + if "enum" in schema and schema["enum"]: + return schema["enum"][0] + + schema_type = EloquaBaseTest._schema_type(schema) + + if schema_type == "object": + properties = schema.get("properties", {}) + required = set(schema.get("required", [])) + return { + key: EloquaBaseTest._generate_value(value, date_value=date_value) + for key, value in properties.items() + if key in required or EloquaBaseTest._schema_type(value) != "null" + } + + if schema_type == "array": + return [ + EloquaBaseTest._generate_value( + schema.get("items", {"type": "string"}), + date_value=date_value, + ) + ] + + if schema_type == "string": + fmt = schema.get("format") + return ( + date_value if fmt == "date-time" else "mock_string" + ) + + if schema_type == "integer": + return 1 + + if schema_type == "number": + return 1.0 + + if schema_type == "boolean": + return True + + return None + + @classmethod + def _generate_stream_record(cls, stream_name, date_value="2024-01-01T00:00:00Z"): + """Generate one schema-valid record for a stream.""" + schemas = EloquaBaseTest._schemas_cache + if schemas is None or stream_name not in schemas: + raise ValueError(f"Stream {stream_name} not found in schemas") + return cls._generate_value(schemas[stream_name], date_value=date_value) + + @classmethod + def _accounts_catalog(cls, name_selected=True, extra_properties=None, extra_metadata=None): + """ + Build a minimal accounts Catalog for use in tests. + + Args: + name_selected: Whether the 'Name' available field is marked selected. + extra_properties: Optional dict of additional property definitions. + extra_metadata: Optional list of additional metadata breadcrumb entries. + """ + properties = { + "Id": {"type": "string"}, + "UpdatedAt": {"type": "string", "format": "date-time"}, + "Name": {"type": ["null", "string"]}, + "Email": {"type": ["null", "string"]}, + } + md = [ + { + "breadcrumb": [], + "metadata": { + "selected": True, + "tap-eloqua.id": None, + "tap-eloqua.query-language-name": "Account", + }, + }, + { + "breadcrumb": ["properties", "Id"], + "metadata": { + "inclusion": "automatic", + "selected": False, + "tap-eloqua.statement": "{{Account.Id}}", + }, + }, + { + "breadcrumb": ["properties", "UpdatedAt"], + "metadata": { + "inclusion": "automatic", + "selected": False, + "tap-eloqua.statement": "{{Account.UpdatedAt}}", + }, + }, + { + "breadcrumb": ["properties", "Name"], + "metadata": { + "inclusion": "available", + "selected": name_selected, + "tap-eloqua.statement": "{{Account.Name}}", + }, + }, + { + "breadcrumb": ["properties", "Email"], + "metadata": { + "inclusion": "available", + "selected": True, + "tap-eloqua.statement": "{{Account.Email}}", + }, + }, + ] + if extra_properties: + properties.update(extra_properties) + if extra_metadata: + md.extend(extra_metadata) + return Catalog.from_dict( + { + "streams": [ + { + "tap_stream_id": "accounts", + "stream": "accounts", + "schema": {"type": "object", "properties": properties}, + "key_properties": ["Id"], + "metadata": md, + } + ] + } + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_all_fields.py b/tests/test_all_fields.py new file mode 100644 index 0000000..f3a87ed --- /dev/null +++ b/tests/test_all_fields.py @@ -0,0 +1,80 @@ +import unittest +from unittest.mock import MagicMock, patch + +from tap_eloqua.sync import sync_bulk_obj + +from .base import EloquaBaseTest + + +class AllFieldsTest(EloquaBaseTest): + def _catalog(self, include_many=False): + extra_properties = {} + extra_metadata = [] + if include_many: + for index in range(251): + field_name = f"Field{index}" + extra_properties[field_name] = {"type": ["null", "string"]} + extra_metadata.append( + { + "breadcrumb": ["properties", field_name], + "metadata": { + "inclusion": "available", + "selected": True, + "tap-eloqua.statement": "{{Account." + field_name + "}}", + }, + } + ) + return self._accounts_catalog( + name_selected=True, + extra_properties=extra_properties or None, + extra_metadata=extra_metadata or None, + ) + + @patch("tap_eloqua.sync.stream_export") + @patch("tap_eloqua.sync.write_bulk_bookmark") + def test_bulk_sync_includes_all_selected_and_automatic_fields(self, mock_write_bulk_bookmark, mock_stream_export): + catalog = self._catalog() + client = MagicMock() + + client.post.side_effect = [{"uri": "/exports/1"}, {"uri": "/syncs/7"}] + client.get.side_effect = [ + {"status": "success"}, + {"items": [{"message": "Successfully exported members to csv file.", "count": 2}]}, + ] + + sync_bulk_obj( + client, + catalog, + state={}, + start_date="2024-01-01T00:00:00Z", + stream_name="accounts", + bulk_page_size=500, + ) + + export_payload = client.post.call_args_list[0].kwargs["json"] + field_map = export_payload["fields"] + + self.assertIn("Id", field_map) + self.assertIn("UpdatedAt", field_map) + self.assertIn("Name", field_map) + self.assertIn("Email", field_map) + + def test_bulk_sync_raises_if_selected_fields_exceed_limit(self): + catalog = self._catalog(include_many=True) + client = MagicMock() + + with self.assertRaises(Exception) as ctx: + sync_bulk_obj( + client, + catalog, + state={}, + start_date="2024-01-01T00:00:00Z", + stream_name="accounts", + bulk_page_size=500, + ) + + self.assertIn("Exports can only have 250 fields selected", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_automatic_fields.py b/tests/test_automatic_fields.py new file mode 100644 index 0000000..cafb154 --- /dev/null +++ b/tests/test_automatic_fields.py @@ -0,0 +1,44 @@ +import unittest +from unittest.mock import MagicMock, patch + +from tap_eloqua.sync import sync_bulk_obj + +from .base import EloquaBaseTest + + +class AutomaticFieldsTest(EloquaBaseTest): + def _catalog(self): + return self._accounts_catalog(name_selected=False) + + @patch("tap_eloqua.sync.stream_export") + @patch("tap_eloqua.sync.write_bulk_bookmark") + def test_bulk_sync_includes_automatic_even_when_unselected(self, mock_write_bulk_bookmark, mock_stream_export): + catalog = self._catalog() + client = MagicMock() + + client.post.side_effect = [{"uri": "/exports/1"}, {"uri": "/syncs/7"}] + client.get.side_effect = [ + {"status": "success"}, + {"items": [{"message": "Successfully exported members to csv file.", "count": 2}]}, + ] + + sync_bulk_obj( + client, + catalog, + state={}, + start_date="2024-01-01T00:00:00Z", + stream_name="accounts", + bulk_page_size=500, + ) + + export_payload = client.post.call_args_list[0].kwargs["json"] + field_map = export_payload["fields"] + + self.assertIn("Id", field_map) + self.assertIn("UpdatedAt", field_map) + self.assertIn("Email", field_map) + self.assertNotIn("Name", field_map) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_bookmark.py b/tests/test_bookmark.py new file mode 100644 index 0000000..1490cdc --- /dev/null +++ b/tests/test_bookmark.py @@ -0,0 +1,86 @@ +import unittest +from unittest.mock import MagicMock, patch + +from singer.catalog import Catalog + +from tap_eloqua.sync import write_bookmark, write_bulk_bookmark, sync_static_endpoint + + +class BookmarkTest(unittest.TestCase): + @patch("tap_eloqua.sync.singer.write_state") + def test_write_bookmark_updates_state_and_emits(self, mock_write_state): + state = {} + write_bookmark(state, "visitors", "2024-01-01T00:00:00Z") + + # Validate overall bookmark state structure + self.assertIsInstance(state, dict) + self.assertIn("bookmarks", state) + self.assertIsInstance(state["bookmarks"], dict) + + self.assertEqual(state["bookmarks"]["visitors"], "2024-01-01T00:00:00Z") + mock_write_state.assert_called_once_with(state) + + @patch("tap_eloqua.sync.singer.write_state") + def test_write_bulk_bookmark_updates_state_and_emits(self, mock_write_state): + state = {} + write_bulk_bookmark( + state, + stream_name="accounts", + sync_id="12", + offset=200, + max_updated_at="2024-01-02T00:00:00Z", + ) + + self.assertEqual( + state["bookmarks"]["accounts"], + { + "sync_id": "12", + "offset": 200, + "datetime": "2024-01-02T00:00:00Z", + }, + ) + mock_write_state.assert_called_once_with(state) + + @patch("tap_eloqua.sync.persist_records") + @patch("tap_eloqua.sync.write_schema") + def test_static_sync_uses_bookmark_in_search(self, mock_write_schema, mock_persist_records): + catalog = Catalog.from_dict( + { + "streams": [ + { + "tap_stream_id": "visitors", + "stream": "visitors", + "schema": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "V_LastVisitDateAndTime": {"type": "string", "format": "date-time"}, + }, + }, + "key_properties": [], + "metadata": [{"breadcrumb": [], "metadata": {"selected": True}}], + } + ] + } + ) + + client = MagicMock() + client.get.return_value = {"elements": []} + + state = {"bookmarks": {"visitors": "2024-09-10T10:11:12Z"}} + sync_static_endpoint( + client, + catalog, + state, + start_date="2024-01-01T00:00:00Z", + stream_id="visitors", + path="data/visitors", + updated_at_col="V_LastVisitDateAndTime", + ) + + sent_params = client.get.call_args.kwargs["params"] + self.assertIn("2024-09-10 10:11:12", sent_params["search"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_discovery_enhanced.py b/tests/test_discovery_enhanced.py new file mode 100644 index 0000000..402ccba --- /dev/null +++ b/tests/test_discovery_enhanced.py @@ -0,0 +1,231 @@ +"""Enhanced discovery tests for tap-eloqua following Singer standards.""" +import json +import unittest +from unittest.mock import MagicMock + +from singer import metadata + +from tap_eloqua.discover import discover + +from .base import EloquaBaseTest + + +class DiscoveryTest(EloquaBaseTest): + """Tests for discovery mode - catalog generation and metadata.""" + + def _create_mock_client(self): + """Create a mock Eloqua API client for testing.""" + mock_client = MagicMock() + + # Mock custom objects endpoint + mock_client.get.side_effect = lambda path, params=None, endpoint=None: ( + {"items": []} + if path == "/api/bulk/2.0/customObjects" + else ( + # Activity/bulk fields + { + "items": [ + { + "internalName": "TestField", + "dataType": "string", + "statement": "{{TestField}}", + "uri": "/fields/1", + } + ] + } + if "fields" in path + else {"items": []} + ) + ) + return mock_client + + def test_discovery_returns_catalog(self): + """Test that discover() returns a valid catalog object.""" + client = self._create_mock_client() + catalog = discover(client) + + self.assertIsNotNone(catalog) + self.assertTrue(hasattr(catalog, "streams")) + self.assertGreater(len(catalog.streams), 0) + + def test_discovery_includes_all_stream_types(self): + """ + Test that discovery includes: + 1. Built-in bulk objects (accounts, contacts) + 2. Activity streams + 3. Static schema streams + """ + client = self._create_mock_client() + catalog = discover(client) + stream_names = {stream.stream for stream in catalog.streams} + + # Verify presence of key stream types + self.assertIn("accounts", stream_names) + self.assertIn("contacts", stream_names) + self.assertIn("activity_email_open", stream_names) + self.assertIn("visitors", stream_names) + self.assertIn("campaigns", stream_names) + + def test_stream_has_required_fields(self): + """Test that each stream entry has required Singer fields.""" + client = self._create_mock_client() + catalog = discover(client) + + required_stream_fields = {"stream", "tap_stream_id", "schema", "key_properties"} + + for stream in catalog.streams: + for field in required_stream_fields: + self.assertTrue( + hasattr(stream, field), + f"Stream '{stream.stream}' missing required field '{field}'", + ) + + def test_stream_metadata_format(self): + """Test that stream metadata follows Singer breadcrumb format.""" + client = self._create_mock_client() + catalog = discover(client) + + for stream in catalog.streams: + md_map = metadata.to_map(stream.metadata) + + # Check for root metadata (empty breadcrumb) + self.assertIn( + (), + md_map, + f"Stream '{stream.stream}' missing root metadata (empty breadcrumb)", + ) + + # Check that breadcrumbs are tuples + for breadcrumb in md_map.keys(): + self.assertIsInstance( + breadcrumb, tuple, f"Breadcrumb {breadcrumb} is not a tuple" + ) + + def test_known_streams_have_forced_replication_method(self): + """Test that known streams have forced-replication-method metadata.""" + client = self._create_mock_client() + catalog = discover(client) + expected = self.expected_metadata() + + for stream in catalog.streams: + if stream.stream in expected: + md_map = metadata.to_map(stream.metadata) + root_md = md_map[()] + + expected_method = expected[stream.stream][self.REPLICATION_METHOD] + actual_method = root_md.get("forced-replication-method") + + self.assertEqual( + actual_method, + expected_method, + f"Stream '{stream.stream}' replication method mismatch", + ) + + def test_bulk_streams_have_eloqua_metadata(self): + """Test that bulk streams have Eloqua-specific metadata.""" + client = self._create_mock_client() + catalog = discover(client) + + bulk_streams = {"accounts", "contacts"} + + for stream in catalog.streams: + if stream.stream in bulk_streams: + md_map = metadata.to_map(stream.metadata) + root_md = md_map[()] + + self.assertIn( + "tap-eloqua.query-language-name", + root_md, + f"Stream '{stream.stream}' missing tap-eloqua.query-language-name", + ) + + def test_static_streams_from_schema_files(self): + """Test that static schema streams are loaded correctly.""" + client = self._create_mock_client() + catalog = discover(client) + stream_names = {stream.stream for stream in catalog.streams} + + # These should come from /schemas/*.json files + expected_static = {"assets", "campaigns", "emails", "forms", "emailGroups", "visitors"} + for static_stream in expected_static: + self.assertIn( + static_stream, + stream_names, + f"Static stream '{static_stream}' not found in discovery", + ) + + def test_stream_schemas_are_valid_json_schema(self): + """Test that all stream schemas are valid JSON Schema objects.""" + client = self._create_mock_client() + catalog = discover(client) + + for stream in catalog.streams: + schema = stream.schema.to_dict() + + # Basic JSON Schema structure + self.assertIsInstance(schema, dict) + self.assertIn("type", schema) + self.assertEqual(schema["type"], "object") + + # Properties should be a dict + if "properties" in schema: + self.assertIsInstance(schema["properties"], dict) + + def test_key_properties_are_lists(self): + """Test that key_properties are properly formatted as lists.""" + client = self._create_mock_client() + catalog = discover(client) + + for stream in catalog.streams: + self.assertIsInstance( + stream.key_properties, + list, + f"Stream '{stream.stream}' key_properties is not a list", + ) + + # Check that all key properties exist in schema + schema = stream.schema.to_dict() + for key in stream.key_properties: + self.assertIn( + key, + schema.get("properties", {}), + f"Key property '{key}' not found in schema for stream '{stream.stream}'", + ) + + def test_field_level_metadata_inclusion(self): + """Test that fields have proper inclusion metadata.""" + client = self._create_mock_client() + catalog = discover(client) + + valid_inclusions = {"automatic", "available", "unsupported"} + + for stream in catalog.streams: + for md in stream.metadata: + breadcrumb = md.get("breadcrumb", []) + # Skip root metadata (which may not have inclusion) + if breadcrumb and len(breadcrumb) >= 2: + inclusion = md["metadata"].get("inclusion") + self.assertIn( + inclusion, + valid_inclusions, + f"Invalid inclusion '{inclusion}' in stream '{stream.stream}'", + ) + + def test_catalog_serialization(self): + """Test that catalog can be serialized to JSON.""" + client = self._create_mock_client() + catalog = discover(client) + catalog_dict = catalog.to_dict() + + # Should serialize properly + catalog_json = json.dumps(catalog_dict) + self.assertIsNotNone(catalog_json) + + # Should be deserializable + deserialized = json.loads(catalog_json) + self.assertIn("streams", deserialized) + self.assertEqual(len(deserialized["streams"]), len(catalog.streams)) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_discovery_metadata.py b/tests/test_discovery_metadata.py similarity index 100% rename from test/test_discovery_metadata.py rename to tests/test_discovery_metadata.py diff --git a/tests/test_pagination.py b/tests/test_pagination.py new file mode 100644 index 0000000..ea093d0 --- /dev/null +++ b/tests/test_pagination.py @@ -0,0 +1,69 @@ +import unittest +from unittest.mock import MagicMock, patch + +from singer.catalog import Catalog + +from tap_eloqua.sync import sync_static_endpoint + + +class PaginationTest(unittest.TestCase): + def _catalog(self): + return Catalog.from_dict( + { + "streams": [ + { + "tap_stream_id": "visitors", + "stream": "visitors", + "schema": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "V_LastVisitDateAndTime": {"type": "string", "format": "date-time"}, + }, + }, + "key_properties": [], + "metadata": [{"breadcrumb": [], "metadata": {"selected": True}}], + } + ] + } + ) + + @patch("tap_eloqua.sync.persist_records") + @patch("tap_eloqua.sync.write_bookmark") + @patch("tap_eloqua.sync.write_schema") + def test_static_endpoint_paginates_until_short_page(self, mock_write_schema, mock_write_bookmark, mock_persist_records): + catalog = self._catalog() + client = MagicMock() + + page_one = { + "elements": [ + {"id": str(i), "V_LastVisitDateAndTime": 1704067200 + i} for i in range(1000) + ] + } + page_two = {"elements": [{"id": "1001", "V_LastVisitDateAndTime": 1704069000}]} + + client.get.side_effect = [page_one, page_two] + state = {} + + sync_static_endpoint( + client, + catalog, + state, + start_date="2024-01-01T00:00:00Z", + stream_id="visitors", + path="data/visitors", + updated_at_col="V_LastVisitDateAndTime", + ) + + self.assertEqual(client.get.call_count, 2) + first_params = client.get.call_args_list[0].kwargs["params"] + second_params = client.get.call_args_list[1].kwargs["params"] + self.assertEqual(first_params["page"], 1) + self.assertEqual(second_params["page"], 2) + + self.assertEqual(mock_persist_records.call_count, 2) + self.assertEqual(mock_write_bookmark.call_count, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_start_date.py b/tests/test_start_date.py new file mode 100644 index 0000000..67574c9 --- /dev/null +++ b/tests/test_start_date.py @@ -0,0 +1,123 @@ +import unittest +from unittest.mock import MagicMock, patch + +from singer.catalog import Catalog + +from tap_eloqua.sync import sync_static_endpoint, sync_bulk_obj + + +class StartDateTest(unittest.TestCase): + def _static_catalog(self): + return Catalog.from_dict( + { + "streams": [ + { + "tap_stream_id": "visitors", + "stream": "visitors", + "schema": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "V_LastVisitDateAndTime": {"type": "string", "format": "date-time"}, + }, + }, + "key_properties": [], + "metadata": [{"breadcrumb": [], "metadata": {"selected": True}}], + } + ] + } + ) + + def _bulk_catalog(self): + return Catalog.from_dict( + { + "streams": [ + { + "tap_stream_id": "accounts", + "stream": "accounts", + "schema": { + "type": "object", + "properties": { + "Id": {"type": "string"}, + "UpdatedAt": {"type": "string", "format": "date-time"}, + }, + }, + "key_properties": ["Id"], + "metadata": [ + { + "breadcrumb": [], + "metadata": { + "selected": True, + "tap-eloqua.id": None, + "tap-eloqua.query-language-name": "Account", + }, + }, + { + "breadcrumb": ["properties", "Id"], + "metadata": { + "inclusion": "automatic", + "tap-eloqua.statement": "{{Account.Id}}", + }, + }, + { + "breadcrumb": ["properties", "UpdatedAt"], + "metadata": { + "inclusion": "automatic", + "tap-eloqua.statement": "{{Account.UpdatedAt}}", + }, + }, + ], + } + ] + } + ) + + @patch("tap_eloqua.sync.persist_records") + @patch("tap_eloqua.sync.write_schema") + def test_static_stream_uses_start_date_when_no_bookmark(self, mock_write_schema, mock_persist_records): + catalog = self._static_catalog() + client = MagicMock() + client.get.return_value = {"elements": []} + + sync_static_endpoint( + client, + catalog, + state={}, + start_date="2024-06-01T00:00:00Z", + stream_id="visitors", + path="data/visitors", + updated_at_col="V_LastVisitDateAndTime", + ) + + sent_params = client.get.call_args.kwargs["params"] + self.assertIn("2024-06-01 00:00:00", sent_params["search"]) + + @patch("tap_eloqua.sync.stream_export") + @patch("tap_eloqua.sync.write_bulk_bookmark") + def test_bulk_stream_uses_bookmark_over_start_date(self, mock_write_bulk_bookmark, mock_stream_export): + catalog = self._bulk_catalog() + client = MagicMock() + + client.post.side_effect = [{"uri": "/exports/1"}, {"uri": "/syncs/99"}] + client.get.side_effect = [ + {"status": "success"}, + {"items": [{"message": "Successfully exported members to csv file.", "count": 1}]}, + ] + + state = {"bookmarks": {"accounts": {"datetime": "2024-08-01T00:00:00Z"}}} + + sync_bulk_obj( + client, + catalog, + state=state, + start_date="2024-01-01T00:00:00Z", + stream_name="accounts", + bulk_page_size=500, + ) + + export_payload = client.post.call_args_list[0].kwargs["json"] + self.assertIn("2024-08-01 00:00:00", export_payload["filter"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sync.py b/tests/test_sync.py new file mode 100644 index 0000000..7802582 --- /dev/null +++ b/tests/test_sync.py @@ -0,0 +1,391 @@ +"""Sync tests for tap-eloqua following Singer standards.""" +import json +import unittest +from unittest.mock import MagicMock, patch, call + +from singer.catalog import Catalog + +from tap_eloqua.sync import sync, get_selected_streams, persist_records + +from .base import EloquaBaseTest + + +class SyncTest(EloquaBaseTest): + """Tests for sync mode - data extraction and transformation.""" + + def _create_mock_client(self): + """Create a mock Eloqua API client for sync testing.""" + mock_client = MagicMock() + + def get_side_effect(path, params=None, endpoint=None): + # Mock REST visitor endpoint + if "/api/REST/2.0/data/visitors" in path: + return { + "elements": [ + { + "id": "1", + "type": "Visitor", + "visitorId": "v1", + "createdAt": "1704067200", + "contactId": "c-100", + "V_IPAddress": "10.0.0.1", + "V_LastVisitDateAndTime": 1704067200, + }, + { + "id": "2", + "type": "Visitor", + "visitorId": "v2", + "createdAt": "1704067300", + "contactId": "c-101", + "V_IPAddress": "10.0.0.2", + "V_LastVisitDateAndTime": 1704067300, + }, + ] + } + + # Mock bulk export creation + if "bulk/2.0/accounts/exports" in path: + return {"uri": "/exports/100"} + + # Mock bulk sync creation + if "bulk/2.0/syncs" in path: + return {"uri": "/syncs/200"} + + # Mock bulk sync status (completed) + if "bulk/2.0/syncs/200" in path and endpoint == "export_sync_poll": + return {"status": "success"} + + # Mock bulk data retrieval + if "bulk/2.0/syncs/200/data" in path: + return { + "items": [ + {"Id": "a1", "CreatedAt": "2024-01-01T00:00:00Z", "UpdatedAt": "2024-01-01T00:00:00Z", "AccountName": "Acme Corp"}, + {"Id": "a2", "CreatedAt": "2024-01-02T00:00:00Z", "UpdatedAt": "2024-01-02T00:00:00Z", "AccountName": "Tech Inc"}, + ], + "hasMore": False, + } + + return {"items": []} + + mock_client.get.side_effect = get_side_effect + mock_client.post.return_value = {"uri": "/exports/100"} + + return mock_client + + def _create_catalog_with_selected_streams(self, selected_stream_names): + """Create a mock catalog with specific streams selected.""" + catalog_dict = { + "streams": [ + { + "tap_stream_id": "visitors", + "stream": "visitors", + "schema": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "type": {"type": "string"}, + "visitorId": {"type": "string"}, + "V_LastVisitDateAndTime": {"type": "string", "format": "date-time"}, + }, + }, + "key_properties": [], + "metadata": [ + { + "breadcrumb": [], + "metadata": { + "selected": "visitors" in selected_stream_names, + "inclusion": "available", + }, + } + ], + }, + { + "tap_stream_id": "accounts", + "stream": "accounts", + "schema": { + "type": "object", + "properties": { + "Id": {"type": "string"}, + "CreatedAt": {"type": "string", "format": "date-time"}, + "UpdatedAt": {"type": "string", "format": "date-time"}, + "AccountName": {"type": ["null", "string"]}, + }, + }, + "key_properties": ["Id"], + "metadata": [ + { + "breadcrumb": [], + "metadata": { + "selected": "accounts" in selected_stream_names, + "forced-replication-method": "INCREMENTAL", + "tap-eloqua.query-language-name": "Account", + }, + }, + { + "breadcrumb": ["properties", "Id"], + "metadata": {"inclusion": "automatic"}, + }, + { + "breadcrumb": ["properties", "UpdatedAt"], + "metadata": {"inclusion": "automatic"}, + }, + ], + }, + ] + } + return Catalog.from_dict(catalog_dict) + + def test_get_selected_streams_respects_selected_metadata(self): + """Test that get_selected_streams reads selected metadata correctly.""" + catalog = self._create_catalog_with_selected_streams(["visitors"]) + + selected = get_selected_streams(catalog) + + self.assertIn("visitors", selected) + self.assertNotIn("accounts", selected) + + def test_get_selected_streams_returns_list(self): + """Test that get_selected_streams returns a list.""" + catalog = self._create_catalog_with_selected_streams(["visitors"]) + + selected = get_selected_streams(catalog) + + self.assertIsInstance(selected, list) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_with_selected_streams(self, mock_write_state, mock_write_record, mock_write_schema): + """Test sync with selected streams.""" + client = self._create_mock_client() + catalog = self._create_catalog_with_selected_streams(["visitors"]) + state = {} + + sync(client, catalog, state, "2024-01-01T00:00:00Z", 1000) + + # Should have called write_schema for visitors + mock_write_schema.assert_any_call( + "visitors", + catalog.get_stream("visitors").schema.to_dict(), + catalog.get_stream("visitors").key_properties, + ) + + # Should have written records + self.assertGreater(mock_write_record.call_count, 0) + # Verify at least one record was written for visitors + calls = mock_write_record.call_args_list + visitor_calls = [c for c in calls if c[0][0] == "visitors"] + self.assertGreater(len(visitor_calls), 0) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_skips_unselected_streams(self, mock_write_state, mock_write_record, mock_write_schema): + """Test that sync skips streams not marked as selected.""" + client = self._create_mock_client() + # Select only visitors, not accounts + catalog = self._create_catalog_with_selected_streams(["visitors"]) + state = {} + + sync(client, catalog, state, "2024-01-01T00:00:00Z", 1000) + + # accounts should not have schema written + accounts_calls = [c for c in mock_write_schema.call_args_list if c[0][0] == "accounts"] + self.assertEqual(len(accounts_calls), 0) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_writes_state_messages(self, mock_write_state, mock_write_record, mock_write_schema): + """Test that sync writes state messages during execution.""" + client = self._create_mock_client() + catalog = self._create_catalog_with_selected_streams(["visitors"]) + state = {} + + sync(client, catalog, state, "2024-01-01T00:00:00Z", 1000) + + # Should have written state messages + self.assertGreater(mock_write_state.call_count, 0) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_returns_without_selected_streams(self, mock_write_state, mock_write_record, mock_write_schema): + """Test that sync returns early if no streams are selected.""" + client = self._create_mock_client() + # Create catalog with no selected streams + catalog = self._create_catalog_with_selected_streams([]) + state = {} + + sync(client, catalog, state, "2024-01-01T00:00:00Z", 1000) + + # Should not have written anything + mock_write_schema.assert_not_called() + mock_write_record.assert_not_called() + + @patch("tap_eloqua.sync.singer.write_record") + def test_persist_records_writes_records(self, mock_write_record): + """Test that persist_records writes records to stdout.""" + catalog = self._create_catalog_with_selected_streams(["visitors"]) + records = [ + { + "id": "1", + "type": "Visitor", + "visitorId": "v1", + "V_LastVisitDateAndTime": "2024-01-01T00:00:00Z", + } + ] + + persist_records(catalog, "visitors", records) + + # Should have written one record + self.assertEqual(mock_write_record.call_count, 1) + call_args = mock_write_record.call_args[0] + self.assertEqual(call_args[0], "visitors") + + def test_record_transformation_handles_datetime(self): + """Test that record transformation converts epoch timestamps to RFC3339.""" + catalog = self._create_catalog_with_selected_streams(["visitors"]) + + # Record with epoch timestamp + records = [ + { + "id": "1", + "V_LastVisitDateAndTime": 1704067200, # Epoch int + } + ] + + with patch("tap_eloqua.sync.singer.write_record") as mock_write: + persist_records(catalog, "visitors", records) + + # Check that written record has transformed datetime + written_record = mock_write.call_args[0][1] + # After transformation, should be ISO string format + self.assertIsNotNone(written_record.get("V_LastVisitDateAndTime")) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_respects_start_date(self, mock_write_state, mock_write_record, mock_write_schema): + """Test that sync passes start_date to API calls.""" + client = self._create_mock_client() + catalog = self._create_catalog_with_selected_streams(["visitors"]) + state = {} + start_date = "2024-06-01T00:00:00Z" + + sync(client, catalog, state, start_date, 1000) + + # Verify start_date appears in the search filter sent to the visitors endpoint + visitors_calls = [ + c for c in client.get.call_args_list + if "/api/REST/2.0/data/visitors" in c[0][0] + ] + self.assertTrue(visitors_calls, "client.get was never called for the visitors endpoint") + first_params = visitors_calls[0].kwargs["params"] + self.assertIn("2024-06-01 00:00:00", first_params["search"]) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_handles_pagination(self, mock_write_state, mock_write_record, mock_write_schema): + """Test that sync handles paginated responses correctly (multiple pages fetched).""" + # Build a client that returns a full page (1000 records) then a short page + paginating_client = MagicMock() + visitors_call_count = [0] + + def paginated_side_effect(path, params=None, endpoint=None): + if "/api/REST/2.0/data/visitors" in path: + visitors_call_count[0] += 1 + if visitors_call_count[0] == 1: + return { + "elements": [ + {"id": str(i), "V_LastVisitDateAndTime": 1704067200 + i} + for i in range(1000) + ] + } + return {"elements": [{"id": "1001", "V_LastVisitDateAndTime": 1704069000}]} + return {"items": []} + + paginating_client.get.side_effect = paginated_side_effect + paginating_client.post.return_value = {"uri": "/exports/100"} + + catalog = self._create_catalog_with_selected_streams(["visitors"]) + state = {} + + sync(paginating_client, catalog, state, "2024-01-01T00:00:00Z", 1000) + + # Pagination is exercised: at least 2 calls to the visitors endpoint + visitors_calls = [ + c for c in paginating_client.get.call_args_list + if "/api/REST/2.0/data/visitors" in c[0][0] + ] + self.assertGreaterEqual(len(visitors_calls), 2) + self.assertGreater(mock_write_record.call_count, 0) + + +class SyncStateManagementTest(EloquaBaseTest): + """Tests for state management in sync.""" + + def _create_mock_client(self): + """Create a minimal mock client.""" + mock_client = MagicMock() + mock_client.get.return_value = {"elements": []} + return mock_client + + def _create_catalog_with_selected_streams(self, selected_stream_names): + """Create a mock catalog with specific streams selected.""" + catalog_dict = { + "streams": [ + { + "tap_stream_id": "visitors", + "stream": "visitors", + "schema": { + "type": "object", + "properties": {"id": {"type": "string"}}, + }, + "key_properties": [], + "metadata": [ + { + "breadcrumb": [], + "metadata": {"selected": "visitors" in selected_stream_names}, + } + ], + } + ] + } + return Catalog.from_dict(catalog_dict) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_initializes_state_for_stream(self, mock_write_state, mock_write_record, mock_write_schema): + """Test that sync initializes state structure for streams.""" + client = self._create_mock_client() + catalog = self._create_catalog_with_selected_streams(["visitors"]) + state = {} + + sync(client, catalog, state, "2024-01-01T00:00:00Z", 1000) + + # write_state should have been called to track progress + self.assertGreater(mock_write_state.call_count, 0) + + @patch("tap_eloqua.sync.singer.write_schema") + @patch("tap_eloqua.sync.singer.write_record") + @patch("tap_eloqua.sync.singer.write_state") + def test_sync_resets_current_stream_on_completion(self, mock_write_state, mock_write_record, mock_write_schema): + """Test that sync sets current_stream to None when done.""" + client = self._create_mock_client() + catalog = self._create_catalog_with_selected_streams(["visitors"]) + state = {} + + sync(client, catalog, state, "2024-01-01T00:00:00Z", 1000) + + # Last state write should set current_stream to None + last_write = mock_write_state.call_args_list[-1] + final_state = last_write[0][0] + self.assertIn("current_stream", final_state) + self.assertIsNone(final_state["current_stream"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unittests/__init__.py b/tests/unittests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unittests/test_discover_unit.py b/tests/unittests/test_discover_unit.py new file mode 100644 index 0000000..d7d4474 --- /dev/null +++ b/tests/unittests/test_discover_unit.py @@ -0,0 +1,99 @@ +import unittest +from unittest.mock import MagicMock, patch + +from singer import metadata as mdata + +from tap_eloqua.discover import discover + + +class TestDiscoverUnit(unittest.TestCase): + @patch("tap_eloqua.discover.get_pk") + @patch("tap_eloqua.discover.get_schemas") + def test_discover_sets_forced_replication_method_and_keys(self, mock_get_schemas, mock_get_pk): + schemas = { + "accounts": { + "type": "object", + "additionalProperties": False, + "properties": { + "Id": {"type": "string"}, + "UpdatedAt": {"type": "string", "format": "date-time"}, + }, + }, + "visitors": { + "type": "object", + "additionalProperties": False, + "properties": { + "id": {"type": "string"}, + "createdAt": {"type": "string"}, + }, + }, + } + field_metadata = { + "accounts": [ + {"breadcrumb": [], "metadata": {"tap-eloqua.query-language-name": "Account"}}, + {"breadcrumb": ["properties", "Id"], "metadata": {"inclusion": "automatic"}}, + ], + "visitors": [ + {"breadcrumb": [], "metadata": {}}, + {"breadcrumb": ["properties", "id"], "metadata": {"inclusion": "automatic"}}, + ], + } + + mock_get_schemas.return_value = (schemas, field_metadata) + mock_get_pk.side_effect = lambda stream_name: ["Id"] if stream_name == "accounts" else [] + + catalog = discover(MagicMock()) + stream_map = {stream.stream: stream for stream in catalog.streams} + + accounts_md = mdata.to_map(stream_map["accounts"].metadata)[()] + visitors_md = mdata.to_map(stream_map["visitors"].metadata)[()] + + self.assertEqual(accounts_md.get("forced-replication-method"), "INCREMENTAL") + self.assertEqual(visitors_md.get("forced-replication-method"), "FULL_TABLE") + + self.assertEqual(stream_map["accounts"].key_properties, ["Id"]) + self.assertEqual(stream_map["visitors"].key_properties, []) + + @patch("tap_eloqua.discover.get_pk", return_value=["Id"]) + @patch("tap_eloqua.discover.get_schemas") + def test_discover_preserves_field_metadata(self, mock_get_schemas, mock_get_pk): + schemas = { + "accounts": { + "type": "object", + "additionalProperties": False, + "properties": { + "Id": {"type": "string"}, + "UpdatedAt": {"type": "string", "format": "date-time"}, + "Name": {"type": ["null", "string"]}, + }, + } + } + field_metadata = { + "accounts": [ + {"breadcrumb": [], "metadata": {"tap-eloqua.query-language-name": "Account"}}, + { + "breadcrumb": ["properties", "Name"], + "metadata": { + "inclusion": "available", + "tap-eloqua.statement": "{{Account.Name}}", + }, + }, + ] + } + + mock_get_schemas.return_value = (schemas, field_metadata) + + catalog = discover(MagicMock()) + accounts = next(stream for stream in catalog.streams if stream.stream == "accounts") + md_map = mdata.to_map(accounts.metadata) + + self.assertIn(("properties", "Name"), md_map) + self.assertEqual(md_map[("properties", "Name")]["inclusion"], "available") + self.assertEqual( + md_map[("properties", "Name")]["tap-eloqua.statement"], + "{{Account.Name}}", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unittests/test_schema_unit.py b/tests/unittests/test_schema_unit.py new file mode 100644 index 0000000..dc9232a --- /dev/null +++ b/tests/unittests/test_schema_unit.py @@ -0,0 +1,46 @@ +import unittest + +from tap_eloqua.schema import ( + camel_to_snake, + activity_type_to_stream, + get_pk, + get_type, +) + + +class TestSchemaUnit(unittest.TestCase): + def test_camel_to_snake(self): + self.assertEqual(camel_to_snake("EmailClickthrough"), "email_clickthrough") + self.assertEqual(camel_to_snake("UpdatedAt"), "updated_at") + + def test_activity_type_to_stream(self): + self.assertEqual(activity_type_to_stream("EmailOpen"), "activity_email_open") + self.assertEqual(activity_type_to_stream("FormSubmit"), "activity_form_submit") + + def test_get_pk_known_streams(self): + self.assertEqual(get_pk("accounts"), ["Id"]) + self.assertEqual(get_pk("assets"), ["id"]) + self.assertEqual(get_pk("activity_email_open"), []) + + def test_get_pk_unknown_stream_defaults_to_id(self): + self.assertEqual(get_pk("custom_unknown"), ["Id"]) + + def test_get_type_date_number_and_string(self): + date_field = {"dataType": "date", "internalName": "UpdatedAt"} + number_field = {"dataType": "number", "internalName": "Score"} + string_field = {"dataType": "string", "internalName": "Name"} + + self.assertEqual(get_type(date_field), (["null", "string"], "date-time")) + self.assertEqual(get_type(number_field), (["null", "number"], None)) + self.assertEqual(get_type(string_field), (["null", "string"], None)) + + def test_get_type_forces_numeric_ids_and_duration_to_string(self): + numeric_id = {"dataType": "number", "internalName": "CampaignId"} + duration = {"dataType": "number", "internalName": "Duration"} + + self.assertEqual(get_type(numeric_id), (["null", "string"], None)) + self.assertEqual(get_type(duration), (["null", "string"], None)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unittests/test_sync_helpers_unit.py b/tests/unittests/test_sync_helpers_unit.py new file mode 100644 index 0000000..b6f2f12 --- /dev/null +++ b/tests/unittests/test_sync_helpers_unit.py @@ -0,0 +1,81 @@ +import unittest +from unittest.mock import patch + +import pendulum + +from tap_eloqua.sync import ( + MIN_RETRY_INTERVAL, + MAX_RETRY_INTERVAL, + next_sleep_interval, + get_bookmark, + get_bulk_bookmark, +) + + +class TestSyncHelpersUnit(unittest.TestCase): + @patch("tap_eloqua.sync.random.randint", return_value=2) + def test_next_sleep_interval_uses_min_on_first_call(self, mock_randint): + value = next_sleep_interval(0) + self.assertEqual(value, MIN_RETRY_INTERVAL) + mock_randint.assert_called_once_with(MIN_RETRY_INTERVAL, MIN_RETRY_INTERVAL) + + @patch("tap_eloqua.sync.random.randint", return_value=7) + def test_next_sleep_interval_doubles_until_cap(self, mock_randint): + value = next_sleep_interval(5) + self.assertEqual(value, 7) + mock_randint.assert_called_once_with(5, 10) + + @patch("tap_eloqua.sync.random.randint", return_value=600) + def test_next_sleep_interval_caps_at_max(self, mock_randint): + value = next_sleep_interval(MAX_RETRY_INTERVAL) + self.assertEqual(value, MAX_RETRY_INTERVAL) + mock_randint.assert_called_once_with(MAX_RETRY_INTERVAL, MAX_RETRY_INTERVAL * 2) + + def test_get_bookmark_with_and_without_value(self): + state = {"bookmarks": {"accounts": "2024-01-01T00:00:00Z"}} + self.assertEqual(get_bookmark(state, "accounts", {}), "2024-01-01T00:00:00Z") + self.assertEqual(get_bookmark(state, "contacts", "default"), "default") + + def test_get_bulk_bookmark_supports_string_and_object(self): + string_state = {"bookmarks": {"accounts": "2024-01-01T00:00:00Z"}} + object_state = { + "bookmarks": { + "accounts": { + "datetime": "2024-01-02T00:00:00Z", + "sync_id": "12", + "offset": 100, + } + } + } + + self.assertEqual( + get_bulk_bookmark(string_state, "accounts"), + {"datetime": "2024-01-01T00:00:00Z"}, + ) + self.assertEqual( + get_bulk_bookmark(object_state, "accounts"), + { + "datetime": "2024-01-02T00:00:00Z", + "sync_id": "12", + "offset": 100, + }, + ) + + def test_pendulum_parse_to_datetime_string_from_string(self): + """pendulum.parse().to_datetime_string() produces YYYY-MM-DD HH:MM:SS + from an RFC-3339 string (replaces _to_eloqua_datetime for string inputs).""" + result = pendulum.parse("2024-06-01T00:00:00Z").to_datetime_string() + self.assertRegex(result, r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$") + self.assertEqual(result, "2024-06-01 00:00:00") + + def test_pendulum_datetime_to_datetime_string(self): + """pendulum DateTime.to_datetime_string() produces YYYY-MM-DD HH:MM:SS + (replaces _to_eloqua_datetime for end_date DateTime inputs).""" + dt = pendulum.parse("2024-08-15T12:30:45Z") + result = dt.to_datetime_string() + self.assertRegex(result, r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$") + self.assertEqual(result, "2024-08-15 12:30:45") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unittests/test_sync_selection_unit.py b/tests/unittests/test_sync_selection_unit.py new file mode 100644 index 0000000..1f6a5b4 --- /dev/null +++ b/tests/unittests/test_sync_selection_unit.py @@ -0,0 +1,94 @@ +import unittest + +from singer.catalog import Catalog + +from tap_eloqua.sync import ( + get_selected_streams, + get_custom_obj_streams, + should_sync_stream, + transform_export_row, +) + + +class TestSyncSelectionUnit(unittest.TestCase): + def _catalog(self): + return Catalog.from_dict( + { + "streams": [ + { + "tap_stream_id": "accounts", + "stream": "accounts", + "schema": {"type": "object", "properties": {"Id": {"type": "string"}}}, + "key_properties": ["Id"], + "metadata": [ + { + "breadcrumb": [], + "metadata": { + "selected": True, + "tap-eloqua.id": None, + }, + } + ], + }, + { + "tap_stream_id": "custom_lead_score", + "stream": "custom_lead_score", + "schema": {"type": "object", "properties": {"Id": {"type": "string"}}}, + "key_properties": ["Id"], + "metadata": [ + { + "breadcrumb": [], + "metadata": { + "selected": False, + "tap-eloqua.id": "42", + }, + } + ], + }, + { + "tap_stream_id": "visitors", + "stream": "visitors", + "schema": {"type": "object", "properties": {"id": {"type": "string"}}}, + "key_properties": [], + "metadata": [{"breadcrumb": [], "metadata": {"selected": True}}], + }, + ] + } + ) + + def test_get_selected_streams(self): + selected = set(get_selected_streams(self._catalog())) + self.assertEqual(selected, {"accounts", "visitors"}) + + def test_get_custom_obj_streams(self): + custom = set(get_custom_obj_streams(self._catalog())) + self.assertEqual(custom, {"custom_lead_score"}) + + def test_should_sync_stream_when_resuming(self): + should_stream, last_stream = should_sync_stream( + selected_streams=["accounts", "visitors"], + last_stream="accounts", + stream_name="accounts", + ) + self.assertTrue(should_stream) + self.assertIsNone(last_stream) + + def test_should_not_sync_stream_if_not_selected(self): + should_stream, last_stream = should_sync_stream( + selected_streams=["accounts"], + last_stream=None, + stream_name="visitors", + ) + self.assertFalse(should_stream) + self.assertIsNone(last_stream) + + def test_transform_export_row_empty_string_to_null(self): + row = {"Id": "1", "Name": "", "Title": "Engineer"} + out = transform_export_row(row) + self.assertEqual(out["Id"], "1") + self.assertIsNone(out["Name"]) + self.assertEqual(out["Title"], "Engineer") + + +if __name__ == "__main__": + unittest.main() From fcf75df5333aa9f1db0b0309c7aa940cf59c9d1b Mon Sep 17 00:00:00 2001 From: Ranit Saha Date: Mon, 20 Apr 2026 15:51:31 +0530 Subject: [PATCH 08/11] update circleci, changelog --- .circleci/config.yml | 12 +++++++++++- CHANGELOG.md | 1 + 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index a8ad341..42f7e1c 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -18,9 +18,19 @@ jobs: source /usr/local/share/virtualenvs/tap-tester/bin/activate stitch-validate-json /usr/local/share/virtualenvs/tap-eloqua/lib/python3.12/site-packages/tap_eloqua/schemas/*.json - run: - name: 'Integration Tests' + name: 'Pylint' + command: | + source /usr/local/share/virtualenvs/tap-eloqua/bin/activate + pylint tap_eloqua --disable=C,R,W + - run: + name: 'Unit Tests' command: | source /usr/local/share/virtualenvs/tap-eloqua/bin/activate + python -m unittest discover -s tests/unittests -p 'test_*.py' --top-level-directory . + - run: + name: 'Integration Tests' + command: | + source /usr/local/share/virtualenvs/tap-tester/bin/activate python -m unittest discover -s tests -p 'test_*.py' --top-level-directory . workflows: diff --git a/CHANGELOG.md b/CHANGELOG.md index ff467b3..fa29f84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## [1.4.0] 2026-04-08 - Added replication method into metadata [#49](https://github.com/singer-io/tap-eloqua/pull/49) - Bump backoff, requests, pendulum, singer-python version [#50](https://github.com/singer-io/tap-eloqua/pull/50) +- Added unit tests and integration tests ## [1.3.1] 2025-06-23 - Bump backoff to `1.10.0`, bump requests to `2.32.4`, bump From 55008eb005654187d2b20a8af19563f1001a4ceb Mon Sep 17 00:00:00 2001 From: Ranit Saha Date: Mon, 20 Apr 2026 16:17:08 +0530 Subject: [PATCH 09/11] added pylint --- setup.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/setup.py b/setup.py index 5ccd5a2..c39a088 100644 --- a/setup.py +++ b/setup.py @@ -17,7 +17,8 @@ ], extras_require={ 'dev': [ - 'ipdb' + 'ipdb', + 'pylint', ] }, entry_points=''' From f21a71ed329a9004ae47707a47cb4e16f758d971 Mon Sep 17 00:00:00 2001 From: Ranit Saha Date: Mon, 20 Apr 2026 16:31:40 +0530 Subject: [PATCH 10/11] Addressed pylint issue --- tap_eloqua/sync.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tap_eloqua/sync.py b/tap_eloqua/sync.py index f346301..b9310d5 100644 --- a/tap_eloqua/sync.py +++ b/tap_eloqua/sync.py @@ -378,6 +378,7 @@ def sync_activity_stream(client, finished = False sync_start = pendulum.now('UTC') end_date = sync_start + last_date = pendulum.parse(start_date) while not finished: try: # Get latest bookmark to adjust time window from, if needed From 4eb1a97ca1955ada8a07499ad1090d297361d485 Mon Sep 17 00:00:00 2001 From: Rushikesh Todkar <98420315+RushiT0122@users.noreply.github.com> Date: Mon, 8 Jun 2026 17:44:55 +0530 Subject: [PATCH 11/11] Update requests dependency version to 2.34.2 --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index c39a088..0df83ed 100644 --- a/setup.py +++ b/setup.py @@ -11,7 +11,7 @@ py_modules=['tap_eloqua'], install_requires=[ 'backoff==2.2.1', - 'requests==2.33.1', + 'requests==2.34.2', 'pendulum==3.2.0', 'singer-python==6.8.0' ],