diff --git a/CHANGELOG.md b/CHANGELOG.md index 107edff..cb9104e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ # Changelog +# 1.2.0 + * Streams returning 403 Forbidden during discovery are now excluded from the catalog; discovery fails only if none are accessible. [#52](https://github.com/singer-io/tap-gitlab/pull/52) + # 1.1.2 * `projects` now independently fetches, merges, and deduplicates project IDs from configured groups. [#55](https://github.com/singer-io/tap-gitlab/pull/55) diff --git a/setup.py b/setup.py index 346d5ec..0855014 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name='tap-gitlab', - version='1.1.2', + version='1.2.0', description='Singer.io tap for extracting data from the GitLab API', author='Stitch', url='https://singer.io', diff --git a/tap_gitlab/__init__.py b/tap_gitlab/__init__.py index 334d4ca..9fe1fc6 100644 --- a/tap_gitlab/__init__.py +++ b/tap_gitlab/__init__.py @@ -11,12 +11,12 @@ REQUIRED_CONFIG_KEYS = ["private_token", "start_date", "groups", "projects"] -def do_discover(): +def do_discover(client): """ Discover and emit the catalog to stdout """ LOGGER.info("Starting discover") - catalog = discover() + catalog = discover(client) json.dump(catalog.to_dict(), sys.stdout, indent=2) LOGGER.info("Finished discover") @@ -33,7 +33,7 @@ def main(): with Client(parsed_args.config) as client: if parsed_args.discover: - do_discover() + do_discover(client) elif parsed_args.catalog: sync( client=client, diff --git a/tap_gitlab/discover.py b/tap_gitlab/discover.py index 3a72408..64a14f8 100644 --- a/tap_gitlab/discover.py +++ b/tap_gitlab/discover.py @@ -2,15 +2,70 @@ from singer import metadata from singer.catalog import Catalog, CatalogEntry, Schema from tap_gitlab.schema import get_schemas +from tap_gitlab.streams import STREAMS +from tap_gitlab.exceptions import ForbiddenError LOGGER = singer.get_logger() -def discover() -> Catalog: +def _prune_inaccessible_children(schemas: dict, field_metadata: dict) -> None: + """ + Remove child streams from the catalog whose parent stream was excluded. + Mutates schemas and field_metadata in place. + """ + to_remove = [] + for name, stream_cls in list(STREAMS.items()): + parent = getattr(stream_cls, "parent", None) + if name in schemas and parent and parent not in schemas: + LOGGER.warning( + "Stream '%s' excluded from catalog because its parent stream '%s' is not accessible.", + name, parent, + ) + schemas.pop(name, None) + field_metadata.pop(name, None) + to_remove.append(name) + return to_remove + + +def _apply_access_checks(client, schemas: dict, field_metadata: dict) -> None: + """ + Probe each stream for read access and remove inaccessible streams + (and their children) from schemas and field_metadata in place. + Raises ForbiddenError if no parent streams are accessible. + """ + inaccessible_streams = [ + stream_name + for stream_name, stream_cls in STREAMS.items() + if stream_name in schemas + and not stream_cls(client=client).check_access() + ] + + for stream_name in inaccessible_streams: + schemas.pop(stream_name, None) + field_metadata.pop(stream_name, None) + + inaccessible_streams.extend(_prune_inaccessible_children(schemas, field_metadata)) + + if not schemas: + raise ForbiddenError( + "No streams are accessible. Ensure the credentials have read permission for at least one stream." + ) + elif inaccessible_streams: + LOGGER.warning( + "Unauthorized streams excluded from catalog: %s", + ", ".join(sorted(set(inaccessible_streams))), + ) + + +def discover(client) -> Catalog: """ Run the discovery mode, prepare the catalog file and return the catalog. + Access to each stream is verified using the provided client and streams + the credentials cannot read are excluded from the returned catalog. """ schemas, field_metadata = get_schemas() + _apply_access_checks(client, schemas, field_metadata) + catalog = Catalog([]) for stream_name, schema_dict in schemas.items(): diff --git a/tap_gitlab/schemas/branches.json b/tap_gitlab/schemas/branches.json index 0d52568..e3bd9df 100644 --- a/tap_gitlab/schemas/branches.json +++ b/tap_gitlab/schemas/branches.json @@ -7,6 +7,13 @@ "integer" ] }, + "projects_updated_at": { + "type": [ + "null", + "string" + ], + "format": "date-time" + }, "name": { "type": [ "null", diff --git a/tap_gitlab/schemas/users.json b/tap_gitlab/schemas/users.json index 21f2f66..f015aa1 100644 --- a/tap_gitlab/schemas/users.json +++ b/tap_gitlab/schemas/users.json @@ -13,6 +13,13 @@ "integer" ] }, + "projects_updated_at": { + "type": [ + "null", + "string" + ], + "format": "date-time" + }, "username": { "type": [ "null", diff --git a/tap_gitlab/streams/abstracts.py b/tap_gitlab/streams/abstracts.py index 452e783..2ee52a0 100644 --- a/tap_gitlab/streams/abstracts.py +++ b/tap_gitlab/streams/abstracts.py @@ -13,7 +13,10 @@ from dateutil import parser from datetime import datetime, timezone +from tap_gitlab.exceptions import ForbiddenError, UnauthorizedError + LOGGER = get_logger() +LOCAL_TIMEZONE = datetime.now().astimezone().tzinfo class BaseStream(ABC): @@ -31,8 +34,8 @@ class BaseStream(ABC): def __init__(self, client=None, catalog=None) -> None: self.client = client self.catalog = catalog - self.schema = self.catalog.schema.to_dict() if self.catalog else None - self.metadata = metadata.to_map(self.catalog.metadata) if self.catalog else None + self.schema = self.catalog.schema.to_dict() if self.catalog else {} + self.metadata = metadata.to_map(self.catalog.metadata) if self.catalog else {} self.child_to_sync = [] self.params = {} @@ -116,8 +119,33 @@ def modify_object(self, record: Dict, parent_record: Dict = None) -> Dict: def get_url_endpoint(self, parent_obj: Dict = None) -> str: return self.url_endpoint or f"{self.client.base_url}/{self.path}" + def check_access(self) -> bool: + """ + Verify that the API credentials have read access to this stream. + Returns True if accessible, False if a 403 Forbidden error is raised. + Child streams always return True (access is governed by the parent check). + """ + if self.parent: + return True + + url = self.get_url_endpoint() + params = {"per_page": 1} + + try: + self.client.get(url, params, self.headers, None) + return True + except (ForbiddenError, UnauthorizedError) as exc: + LOGGER.warning( + "Unauthorized Stream: %s, excluding from catalog. HTTP-Error-Message:'%s'", + self.tap_stream_id, + str(exc), + ) + return False + class IncrementalStream(BaseStream): + send_updated_since = True + def get_bookmark(self, state: dict, stream: str, key: Any = None) -> int: return get_bookmark( # pylint: disable=E1121 state, @@ -135,7 +163,15 @@ def update_bookmark_state(self, state: dict, stream: str, key: Any = None, value state, stream, bookmark_key, self.client.config["start_date"] ) try: - value = max(current_bookmark, value) + current_dt = self._to_utc_datetime(current_bookmark) + value_dt = self._to_utc_datetime(value) + + if current_dt and value_dt: + value = max(current_dt, value_dt).isoformat(timespec='seconds').replace('+00:00', 'Z') + elif value_dt: + value = value_dt.isoformat(timespec='seconds').replace('+00:00', 'Z') + else: + value = current_bookmark except Exception: LOGGER.warning("Failed to compare bookmark values. Keeping current bookmark.") value = current_bookmark @@ -153,14 +189,14 @@ def _to_utc_datetime(self, value): return None if isinstance(value, datetime): if value.tzinfo is None: - return value.replace(tzinfo=timezone.utc) + return value.replace(tzinfo=LOCAL_TIMEZONE).astimezone(timezone.utc) return value.astimezone(timezone.utc) if isinstance(value, (int, float)): - return datetime.fromtimestamp(value).replace(tzinfo=timezone.utc) + return datetime.fromtimestamp(value, tz=timezone.utc) if isinstance(value, str): dt = parser.parse(value) if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) + dt = dt.replace(tzinfo=LOCAL_TIMEZONE) return dt.astimezone(timezone.utc) LOGGER.warning(f"Unsupported timestamp type: {type(value)}") return None @@ -178,7 +214,8 @@ def sync(self, state: Dict, transformer: Transformer, parent_obj: Dict = None) - bookmark_date = self._to_utc_datetime(self.client.config["start_date"]) current_max_bookmark_date = bookmark_date - self.update_params(updated_since=bookmark_date.isoformat(timespec='seconds').replace('+00:00', 'Z')) + if self.send_updated_since: + self.update_params(updated_since=bookmark_date.isoformat(timespec='seconds').replace('+00:00', 'Z')) self.url_endpoint = self.get_url_endpoint(parent_obj) with metrics.record_counter(self.tap_stream_id) as counter: @@ -194,6 +231,12 @@ def sync(self, state: Dict, transformer: Transformer, parent_obj: Dict = None) - LOGGER.warning(f"Skipping record with invalid {self.replication_keys[0]}: {record_value}") continue + # Normalize replication key values to UTC with explicit timezone + # so emitted records and saved state are compared consistently. + transformed_record[self.replication_keys[0]] = ( + record_timestamp.isoformat(timespec='microseconds').replace('+00:00', 'Z') + ) + if record_timestamp >= bookmark_date: if self.is_selected(): write_record(self.tap_stream_id, transformed_record) @@ -215,6 +258,34 @@ def sync(self, state: Dict, transformer: Transformer, parent_obj: Dict = None) - ) return counter.value + +class ParentBaseStream(IncrementalStream): + """Incremental parent stream that owns child stream bookmarks.""" + + def get_bookmark(self, state: dict, stream: str, key: Any = None): + min_bookmark = super().get_bookmark(state, stream) if self.is_selected() else None + bookmark_key = f"{self.tap_stream_id}_{self.replication_keys[0]}" + + for child in self.child_to_sync: + child_bookmark = super().get_bookmark( + state, child.tap_stream_id, key=bookmark_key + ) + min_bookmark = min(min_bookmark, child_bookmark) if min_bookmark else child_bookmark + + return min_bookmark or self.client.config["start_date"] + + def update_bookmark_state(self, state: dict, stream: str, key: Any = None, value: Any = None) -> Dict: + if self.is_selected(): + super().update_bookmark_state(state, stream, key=key, value=value) + + bookmark_key = f"{self.tap_stream_id}_{self.replication_keys[0]}" + for child in self.child_to_sync: + super().update_bookmark_state( + state, child.tap_stream_id, key=bookmark_key, value=value + ) + + return state + class FullTableStream(BaseStream): """Base Class for FullTable Stream.""" @@ -245,6 +316,8 @@ def sync( class ChildBaseStream(IncrementalStream): """Base Class for Child Stream.""" + send_updated_since = False + def get_bookmark(self, state: Dict, stream: str, key: Any = None) -> int: """Singleton bookmark value for child streams.""" if not self.bookmark_value: diff --git a/tap_gitlab/streams/branches.py b/tap_gitlab/streams/branches.py index d5b8a18..19612cb 100644 --- a/tap_gitlab/streams/branches.py +++ b/tap_gitlab/streams/branches.py @@ -1,14 +1,14 @@ from typing import Dict, Any from urllib.parse import quote -from tap_gitlab.streams.abstracts import FullTableStream +from tap_gitlab.streams.abstracts import ChildBaseStream -class Branches(FullTableStream): +class Branches(ChildBaseStream): tap_stream_id = "branches" key_properties = ["project_id", "name"] - replication_method = "FULL_TABLE" + replication_method = "INCREMENTAL" parent = "projects" - replication_keys = None + replication_keys = ["projects_updated_at"] path = "projects/{}/repository/branches" data_key = None @@ -30,9 +30,9 @@ def get_url_endpoint(self, parent_obj: Dict = None) -> str: return endpoint def modify_object(self, record, parent_record=None): - """Adding project_id and last_committed_date to the record""" - if isinstance(record, dict): - if parent_record and isinstance(parent_record, dict): - record["project_id"] = parent_record.get("id") + """Add project_id and the parent project's update timestamp.""" + if isinstance(record, dict) and isinstance(parent_record, dict): + record["project_id"] = parent_record.get("id") + record["projects_updated_at"] = parent_record.get("updated_at") return record diff --git a/tap_gitlab/streams/projects.py b/tap_gitlab/streams/projects.py index 2a585c4..27bc660 100644 --- a/tap_gitlab/streams/projects.py +++ b/tap_gitlab/streams/projects.py @@ -2,11 +2,11 @@ from singer import get_logger from urllib.parse import quote -from tap_gitlab.streams.abstracts import IncrementalStream +from tap_gitlab.streams.abstracts import ParentBaseStream LOGGER = get_logger() -class Projects(IncrementalStream): +class Projects(ParentBaseStream): tap_stream_id = "projects" key_properties = ["id"] replication_method = "INCREMENTAL" diff --git a/tap_gitlab/streams/users.py b/tap_gitlab/streams/users.py index 2e8cfd1..788527f 100644 --- a/tap_gitlab/streams/users.py +++ b/tap_gitlab/streams/users.py @@ -1,14 +1,14 @@ from typing import Dict, Any from urllib.parse import quote -from tap_gitlab.streams.abstracts import FullTableStream +from tap_gitlab.streams.abstracts import ChildBaseStream -class Users(FullTableStream): +class Users(ChildBaseStream): tap_stream_id = "users" key_properties = ["id", "project_id"] - replication_method = "FULL_TABLE" + replication_method = "INCREMENTAL" parent = "projects" - replication_keys = None + replication_keys = ["projects_updated_at"] path = "projects/{}/users" data_key = None @@ -29,9 +29,10 @@ def get_url_endpoint(self, parent_obj: Dict = None) -> str: endpoint = f"{self.client.base_url}/{self.get_url(parent_obj)}" return endpoint - def modify_object(self, record, parent_record = None): - """Adding project_id to the record.""" - if isinstance(record, dict): + def modify_object(self, record, parent_record=None): + """Add project_id and the parent project's update timestamp.""" + if isinstance(record, dict) and isinstance(parent_record, dict): record["project_id"] = parent_record.get("id") + record["projects_updated_at"] = parent_record.get("updated_at") return record diff --git a/tests/base.py b/tests/base.py index 0cd00fc..c3edff1 100644 --- a/tests/base.py +++ b/tests/base.py @@ -18,6 +18,7 @@ class BaseTest(BaseCase): in tap-tester tests. Shared tap-specific methods (as needed). """ start_date = "2019-01-01T00:00:00Z" + IS_FORBIDDEN_STREAM = "is-forbidden-stream" @staticmethod def tap_name(): @@ -42,9 +43,9 @@ def expected_metadata(cls): }, "branches": { cls.PRIMARY_KEYS: {"project_id", "name"}, - cls.REPLICATION_METHOD: cls.FULL_TABLE, - cls.REPLICATION_KEYS: set(), - cls.OBEYS_START_DATE: False, + cls.REPLICATION_METHOD: cls.INCREMENTAL, + cls.REPLICATION_KEYS: {"projects_updated_at"}, + cls.OBEYS_START_DATE: True, cls.API_LIMIT: 5 }, "commits": { @@ -77,9 +78,9 @@ def expected_metadata(cls): }, "users": { cls.PRIMARY_KEYS: {"id", "project_id"}, - cls.REPLICATION_METHOD: cls.FULL_TABLE, - cls.REPLICATION_KEYS: set(), - cls.OBEYS_START_DATE: False, + cls.REPLICATION_METHOD: cls.INCREMENTAL, + cls.REPLICATION_KEYS: {"projects_updated_at"}, + cls.OBEYS_START_DATE: True, cls.API_LIMIT: 3 }, "groups": { @@ -91,6 +92,14 @@ def expected_metadata(cls): } } + def expected_stream_names(self): + """The expected stream names and exclude forbidden streams.""" + return { + stream_name + for stream_name, metadata in self.expected_metadata().items() + if not metadata.get(self.IS_FORBIDDEN_STREAM, False) + } + @staticmethod def get_credentials(): """Authentication information for the test account.""" diff --git a/tests/test_pagination.py b/tests/test_pagination.py index 0c0d0d3..25cb359 100644 --- a/tests/test_pagination.py +++ b/tests/test_pagination.py @@ -13,6 +13,7 @@ def name(): def streams_to_test(self): # don't have enough data to test pagination streams_to_exclude = { + "groups", "group_milestones", "issues", "project_milestones" diff --git a/tests/unittests/test_discovery.py b/tests/unittests/test_discovery.py new file mode 100644 index 0000000..85bee49 --- /dev/null +++ b/tests/unittests/test_discovery.py @@ -0,0 +1,240 @@ +import unittest +from unittest.mock import patch, MagicMock +from tap_gitlab.discover import discover, _apply_access_checks, _prune_inaccessible_children +from tap_gitlab.exceptions import ForbiddenError + + +class TestAccessChecks(unittest.TestCase): + """Tests for stream access check logic during discovery.""" + + @patch("tap_gitlab.discover._prune_inaccessible_children") + @patch("tap_gitlab.discover.STREAMS") + def test_all_streams_accessible(self, mock_streams, mock_prune): + """All streams accessible - none excluded.""" + mock_client = MagicMock() + + mock_stream_instance = MagicMock() + mock_stream_instance.check_access.return_value = True + mock_stream_cls = MagicMock(return_value=mock_stream_instance) + + mock_streams.items.return_value = [("projects", mock_stream_cls), ("groups", mock_stream_cls)] + + schemas = {"projects": {"properties": {}}, "groups": {"properties": {}}} + field_metadata = {"projects": [], "groups": []} + + _apply_access_checks(mock_client, schemas, field_metadata) + + self.assertIn("projects", schemas) + self.assertIn("groups", schemas) + + @patch("tap_gitlab.discover._prune_inaccessible_children") + @patch("tap_gitlab.discover.STREAMS") + def test_partial_access(self, mock_streams, mock_prune): + """Some streams inaccessible - those are excluded.""" + mock_client = MagicMock() + + accessible_instance = MagicMock() + accessible_instance.check_access.return_value = True + accessible_cls = MagicMock(return_value=accessible_instance) + + forbidden_instance = MagicMock() + forbidden_instance.check_access.return_value = False + forbidden_cls = MagicMock(return_value=forbidden_instance) + + mock_streams.items.return_value = [ + ("projects", accessible_cls), + ("groups", forbidden_cls), + ] + + schemas = {"projects": {"properties": {}}, "groups": {"properties": {}}} + field_metadata = {"projects": [], "groups": []} + + _apply_access_checks(mock_client, schemas, field_metadata) + + self.assertIn("projects", schemas) + self.assertNotIn("groups", schemas) + self.assertIn("projects", field_metadata) + self.assertNotIn("groups", field_metadata) + + @patch("tap_gitlab.discover._prune_inaccessible_children") + @patch("tap_gitlab.discover.STREAMS") + def test_no_streams_accessible_raises(self, mock_streams, mock_prune): + """All streams inaccessible - raises ForbiddenError.""" + mock_client = MagicMock() + + forbidden_instance = MagicMock() + forbidden_instance.check_access.return_value = False + forbidden_cls = MagicMock(return_value=forbidden_instance) + + mock_streams.items.return_value = [ + ("projects", forbidden_cls), + ("groups", forbidden_cls), + ] + + schemas = {"projects": {"properties": {}}, "groups": {"properties": {}}} + field_metadata = {"projects": [], "groups": []} + + with self.assertRaises(ForbiddenError): + _apply_access_checks(mock_client, schemas, field_metadata) + + @patch("tap_gitlab.discover._prune_inaccessible_children") + @patch("tap_gitlab.discover.STREAMS") + def test_no_streams_accessible_raises_with_message(self, mock_streams, mock_prune): + """All streams inaccessible - ForbiddenError has correct message.""" + mock_client = MagicMock() + + forbidden_instance = MagicMock() + forbidden_instance.check_access.return_value = False + forbidden_cls = MagicMock(return_value=forbidden_instance) + + mock_streams.items.return_value = [ + ("projects", forbidden_cls), + ("groups", forbidden_cls), + ] + + schemas = {"projects": {"properties": {}}, "groups": {"properties": {}}} + field_metadata = {"projects": [], "groups": []} + + with self.assertRaises(ForbiddenError) as context: + _apply_access_checks(mock_client, schemas, field_metadata) + + self.assertIn( + "No streams are accessible. Ensure the credentials have read permission for at least one stream.", + str(context.exception), + ) + + @patch("tap_gitlab.discover.LOGGER") + @patch("tap_gitlab.discover._prune_inaccessible_children") + @patch("tap_gitlab.discover.STREAMS") + def test_partial_access_logs_warning(self, mock_streams, mock_prune, mock_logger): + """Some streams inaccessible - logs warning listing excluded streams.""" + mock_client = MagicMock() + + accessible_instance = MagicMock() + accessible_instance.check_access.return_value = True + accessible_cls = MagicMock(return_value=accessible_instance) + + forbidden_instance = MagicMock() + forbidden_instance.check_access.return_value = False + forbidden_cls = MagicMock(return_value=forbidden_instance) + + mock_streams.items.return_value = [ + ("projects", accessible_cls), + ("groups", forbidden_cls), + ] + + schemas = {"projects": {"properties": {}}, "groups": {"properties": {}}} + field_metadata = {"projects": [], "groups": []} + + _apply_access_checks(mock_client, schemas, field_metadata) + + mock_logger.warning.assert_called_with( + "Unauthorized streams excluded from catalog: %s", + "groups", + ) + + def test_prune_inaccessible_children(self): + """Child streams are removed when parent is excluded.""" + schemas = { + "branches": {"properties": {}}, + "commits": {"properties": {}}, + "groups": {"properties": {}}, + "group_milestones": {"properties": {}}, + } + field_metadata = { + "branches": [], + "commits": [], + "groups": [], + "group_milestones": [], + } + + # branches and commits have parent="projects", which is not in schemas + # group_milestones has parent="groups", which IS in schemas + _prune_inaccessible_children(schemas, field_metadata) + + self.assertNotIn("branches", schemas) + self.assertNotIn("commits", schemas) + self.assertIn("groups", schemas) + self.assertIn("group_milestones", schemas) + + +class TestCheckAccessMethod(unittest.TestCase): + """Tests for BaseStream.check_access().""" + + def test_child_stream_always_returns_true(self): + """Child streams always return True without making API call.""" + from tap_gitlab.streams.branches import Branches + + mock_client = MagicMock() + stream = Branches(client=mock_client) + + result = stream.check_access() + + self.assertTrue(result) + mock_client.get.assert_not_called() + + def test_parent_stream_accessible(self): + """Parent stream returns True when API call succeeds.""" + from tap_gitlab.streams.groups import Groups + + mock_client = MagicMock() + mock_client.base_url = "https://gitlab.com/api/v4" + mock_client.get.return_value = [{"id": 1}] + stream = Groups(client=mock_client) + + result = stream.check_access() + + self.assertTrue(result) + + def test_parent_stream_forbidden(self): + """Parent stream returns False when 403 is raised.""" + from tap_gitlab.streams.groups import Groups + + mock_client = MagicMock() + mock_client.base_url = "https://gitlab.com/api/v4" + mock_client.get.side_effect = ForbiddenError("Forbidden") + stream = Groups(client=mock_client) + + result = stream.check_access() + + self.assertFalse(result) + + def test_parent_stream_forbidden_logs_warning(self): + """Parent stream logs warning with stream name and error message on 403.""" + from tap_gitlab.streams.groups import Groups + + mock_client = MagicMock() + mock_client.base_url = "https://gitlab.com/api/v4" + mock_client.get.side_effect = ForbiddenError("403 Access Denied") + stream = Groups(client=mock_client) + + with patch("tap_gitlab.streams.abstracts.LOGGER") as mock_logger: + result = stream.check_access() + + self.assertFalse(result) + mock_logger.warning.assert_called_once_with( + "Unauthorized Stream: %s, excluding from catalog. HTTP-Error-Message:'%s'", + stream.tap_stream_id, + "403 Access Denied", + ) + + +class TestDiscoverWithClient(unittest.TestCase): + """Tests for the discover() function accepting a client.""" + + @patch("tap_gitlab.discover._apply_access_checks") + @patch("tap_gitlab.discover.get_schemas") + def test_discover_calls_access_checks(self, mock_get_schemas, mock_access_checks): + """discover() calls _apply_access_checks with client.""" + mock_client = MagicMock() + mock_get_schemas.return_value = ({}, {}) + + discover(mock_client) + + mock_access_checks.assert_called_once() + args = mock_access_checks.call_args[0] + self.assertEqual(args[0], mock_client) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unittests/test_projects_stream.py b/tests/unittests/test_projects_stream.py index 944e91c..3e53bd1 100644 --- a/tests/unittests/test_projects_stream.py +++ b/tests/unittests/test_projects_stream.py @@ -203,6 +203,54 @@ def test_uses_correct_endpoint_per_project(self): self.assertEqual(called_endpoint, "https://gitlab.com/api/v4/projects/10") +class TestChildIncrementalSync(unittest.TestCase): + """Verify child streams use the parent timestamp without API filtering.""" + + def _make_stream(self, stream_class): + client = make_mock_client({ + "start_date": "2026-01-01T00:00:00Z", + "projects": "10", + }) + stream = stream_class(client=client, catalog=make_mock_catalog_entry()) + stream.is_selected = lambda: True + return stream, client + + def test_branches_do_not_send_updated_since(self): + from tap_gitlab.streams.branches import Branches + + stream, _ = self._make_stream(Branches) + stream.get_records = lambda: iter([]) + stream.sync({}, MagicMock(), {"id": 10, "updated_at": "2026-02-01T00:00:00Z"}) + + self.assertNotIn("updated_since", stream.params) + + def test_users_do_not_send_updated_since(self): + from tap_gitlab.streams.users import Users + + stream, _ = self._make_stream(Users) + stream.get_records = lambda: iter([]) + stream.sync({}, MagicMock(), {"id": 10, "updated_at": "2026-02-01T00:00:00Z"}) + + self.assertNotIn("updated_since", stream.params) + + def test_child_record_is_filtered_by_parent_timestamp(self): + from tap_gitlab.streams.branches import Branches + + stream, _ = self._make_stream(Branches) + stream.get_records = lambda: iter([{"name": "main"}]) + transformer = MagicMock() + transformer.transform.side_effect = lambda record, schema, metadata: record + + with unittest.mock.patch("tap_gitlab.streams.abstracts.write_record") as write_record: + stream.sync( + {"bookmarks": {"branches": {"projects_updated_at": "2026-03-01T00:00:00Z"}}}, + transformer, + {"id": 10, "updated_at": "2026-02-01T00:00:00Z"}, + ) + + write_record.assert_not_called() + + class TestGroupsStreamIndependence(unittest.TestCase): """Verify Groups stream no longer has project-syncing logic."""