SAC-29934: Refactored and fixed integration tests to support new streams (#43)

* Refactored and fixed integration tests to support new streams

* Excluded organization_actions stream as it doesn't have enough data

* Added organization_actions stream to untestable_streams

* Fixed bookmarks assertion

* Fixed duplicate records assertion error

* Fix automatic fields for child streams by calling modify_object in FullTableStream

* Fix package installation by using find_packages() to include streams submodule

* Fixed Field mismatch errors

* Addressed review comments

* Fix API rate limits

* Fix assertion error for card_custom_field_items stream

Author: MuralidharT03

setup.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-from setuptools import setup
+from setuptools import find_packages, setup
 
 setup(
     name="tap-trello",
@@ -27,9 +27,9 @@
     [console_scripts]
     tap-trello=tap_trello:main
     """,
-    packages=["tap_trello"],
+    packages=find_packages(),
     package_data = {
-        "tap_trello": ["tap_trello/schemas/*.json"]
+        "tap_trello": ["schemas/*.json"]
     },
     include_package_data=True,
 )

tap_trello/streams/abstracts.py

@@ -572,6 +572,7 @@ def sync(
         self.update_data_payload(parent_obj=parent_obj)
         with metrics.record_counter(self.tap_stream_id) as counter:
             for record in self.get_records():
+                record = self.modify_object(record, parent_obj)
                 transformed_record = transformer.transform(
                     record, self.schema, self.metadata
                 )

tap_trello/streams/board_custom_fields.py

@@ -6,3 +6,9 @@ class BoardCustomFields(FullTableStream):
     replication_method = "FULL_TABLE"
     path = "/boards/{id}/customFields"
     parent = "boards"
+
+    def modify_object(self, record, parent_record=None):
+        """Add boardId to board custom field records."""
+        if parent_record and 'id' in parent_record:
+            record["boardId"] = parent_record['id']
+        return record

tap_trello/streams/board_labels.py

@@ -6,3 +6,9 @@ class BoardLabels(FullTableStream):
     replication_method = "FULL_TABLE"
     path = "/boards/{id}/labels"
     parent = "boards"
+
+    def modify_object(self, record, parent_record=None):
+        """Add boardId to board label records."""
+        if parent_record and 'id' in parent_record:
+            record["boardId"] = parent_record['id']
+        return record

tap_trello/streams/card_attachments.py

@@ -6,3 +6,9 @@ class CardAttachments(FullTableStream):
     replication_method = "FULL_TABLE"
     path = "/cards/{id}/attachments"
     parent = "cards"
+
+    def modify_object(self, record, parent_record=None):
+        """Add card_id to card attachment records."""
+        if parent_record and 'id' in parent_record:
+            record["card_id"] = parent_record['id']
+        return record

tap_trello/streams/card_custom_field_items.py

@@ -6,3 +6,9 @@ class CardCustomFieldItems(FullTableStream):
     replication_method = "FULL_TABLE"
     path = "/cards/{id}/customFieldItems"
     parent = "cards"
+
+    def modify_object(self, record, parent_record=None):
+        """Add card_id to card custom field item records."""
+        if parent_record and 'id' in parent_record:
+            record["card_id"] = parent_record['id']
+        return record

tap_trello/streams/organization_actions.py

@@ -1,5 +1,8 @@
 import json
+from typing import Dict
+
 import singer
+from singer import Transformer
 
 from tap_trello.streams.abstracts import ChildBaseStream
 
@@ -13,14 +16,38 @@ class OrganizationActions(ChildBaseStream):
     replication_keys = ["date"]
     path = "/organizations/{id}/actions"
     parent = "organizations"
-    bookmark_value = None
+    params = {'limit': 1000}
+
+    def sync(
+        self,
+        state: Dict,
+        transformer: Transformer,
+        parent_obj: Dict = None,
+    ) -> Dict:
+        """Override sync to store state and add date filtering."""
+        self._sync_state = state
+        self._sync_parent_obj = parent_obj
 
+        return super().sync(state, transformer, parent_obj)
 
     def get_records(self):
-        url = self.get_url_endpoint(getattr(self, 'parent_obj', None)) if hasattr(self, 'parent_obj') else self.get_url_endpoint()
+        """Get records with date filtering for incremental replication."""
+        url = self.get_url_endpoint(getattr(self, '_sync_parent_obj', None))
         params = dict(self.params) if hasattr(self, 'params') else {}
         params.pop('page', None)
 
+        # Add date filtering for incremental replication
+        # Access state from the temporarily stored sync state
+        state = getattr(self, '_sync_state', {})
+        bookmark_date = self.get_bookmark(state, self.tap_stream_id)
+        if bookmark_date:
+            params['since'] = bookmark_date
+        else:
+            # Use start_date from config if no bookmark exists
+            start_date = self.client.config.get('start_date')
+            if start_date:
+                params['since'] = start_date
+
         response = self.client.make_request(
             self.http_method,
             url,

tap_trello/streams/organization_members.py

@@ -6,3 +6,9 @@ class OrganizationMembers(FullTableStream):
     replication_method = "FULL_TABLE"
     path = "/organizations/{id}/members"
     parent = "organizations"
+
+    def modify_object(self, record, parent_record=None):
+        """Add organization_id to organization member records."""
+        if parent_record and 'id' in parent_record:
+            record["organization_id"] = parent_record['id']
+        return record

tap_trello/streams/organization_memberships.py

@@ -6,3 +6,9 @@ class OrganizationMemberships(FullTableStream):
     replication_method = "FULL_TABLE"
     path = "/organizations/{id}/memberships"
     parent = "organizations"
+
+    def modify_object(self, record, parent_record=None):
+        """Add organization_id to organization membership records."""
+        if parent_record and 'id' in parent_record:
+            record["organization_id"] = parent_record['id']
+        return record

tests/base.py

@@ -0,0 +1,171 @@
+import os
+import unittest
+
+from datetime import datetime as dt
+
+import tap_tester.connections as connections
+import tap_tester.menagerie as menagerie
+
+
+class TrelloBaseTest(unittest.TestCase):
+    """Base test class with common setup and utility methods for Trello tap tests"""
+
+    START_DATE = ""
+    START_DATE_FORMAT = "%Y-%m-%dT00:00:00Z"
+    TEST_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
+
+    def setUp(self):
+        missing_envs = [x for x in [
+            "TAP_TRELLO_CONSUMER_KEY",
+            "TAP_TRELLO_CONSUMER_SECRET",
+            "TAP_TRELLO_ACCESS_TOKEN",
+            "TAP_TRELLO_ACCESS_TOKEN_SECRET",
+        ] if os.getenv(x) == None]
+        if len(missing_envs) != 0:
+            raise Exception("Missing environment variables: {}".format(missing_envs))
+
+    def name(self):
+        return "tap_tester_trello_base_test"
+
+    def get_type(self):
+        return "platform.trello"
+
+    def get_credentials(self):
+        return {
+            'consumer_key': os.getenv('TAP_TRELLO_CONSUMER_KEY'),
+            'consumer_secret': os.getenv('TAP_TRELLO_CONSUMER_SECRET'),
+            'access_token': os.getenv('TAP_TRELLO_ACCESS_TOKEN'),
+            'access_token_secret': os.getenv('TAP_TRELLO_ACCESS_TOKEN_SECRET'),
+        }
+
+    def tap_name(self):
+        return "tap-trello"
+
+    def get_properties(self):
+        return {
+            'start_date': dt.strftime(dt.utcnow(), self.START_DATE_FORMAT),
+        }
+
+    def testable_streams(self):
+        return self.expected_check_streams().difference(self.untestable_streams())
+
+    def untestable_streams(self):
+        return set()
+
+    def expected_check_streams(self):
+        return {
+            'actions',
+            'board_custom_fields',
+            'board_labels',
+            'board_memberships',
+            'boards',
+            'card_attachments',
+            'card_custom_field_items',
+            'cards',
+            'checklists',
+            'lists',
+            'members',
+            'organization_actions',
+            'organization_members',
+            'organization_memberships',
+            'organizations',
+            'users'
+        }
+
+    def expected_sync_streams(self):
+        return self.expected_check_streams()
+
+    def expected_full_table_streams(self):
+        return {
+            'boards',
+            'board_custom_fields',
+            'board_labels',
+            'board_memberships',
+            'cards',
+            'card_attachments',
+            'card_custom_field_items',
+            'checklists',
+            'lists',
+            'members',
+            'organizations',
+            'organization_members',
+            'organization_memberships',
+            'users',
+        }
+
+    def expected_full_table_sync_streams(self):
+        return self.expected_full_table_streams()
+
+    def expected_incremental_streams(self):
+        return {
+            'actions',
+            'organization_actions'
+        }
+
+    def expected_pks(self):
+        return {
+            'actions': {"id"},
+            'boards': {"id"},
+            'board_custom_fields': {"id", "boardId"},
+            'board_labels': {"id", "boardId"},
+            'board_memberships': {"id", "boardId"},
+            'cards': {'id'},
+            'card_attachments': {"id", "card_id"},
+            'card_custom_field_items': {"id", "card_id"},
+            'checklists': {'id'},
+            'lists': {"id"},
+            'members': {"id"},
+            'organizations': {"id"},
+            'organization_actions': {"id", "organization_id"},
+            'organization_members': {"id", "organization_id"},
+            'organization_memberships': {"id", "organization_id"},
+            'users': {"id", "boardId"}
+        }
+
+    def expected_automatic_fields(self):
+        return {
+            'actions': {"id", "date"},
+            'boards': {"id"},
+            'board_custom_fields': {"id", "boardId"},
+            'board_labels': {"id", "boardId"},
+            'board_memberships': {"id", "boardId"},
+            'cards': {'id'},
+            'card_attachments': {"id", "card_id"},
+            'card_custom_field_items': {"id", "card_id"},
+            'checklists': {'id'},
+            'lists': {"id"},
+            'members': {"id"},
+            'organizations': {"id"},
+            'organization_actions': {"id", "organization_id", "date"},
+            'organization_members': {"id", "organization_id"},
+            'organization_memberships': {"id", "organization_id"},
+            'users': {"id", "boardId"}
+        }
+
+    def select_all_streams_and_fields(self, conn_id, catalogs, select_all_fields: bool = True):
+        """
+        Select all streams and optionally all fields within streams.
+
+        Args:
+            conn_id: Connection ID
+            catalogs: List of catalogs to select
+            select_all_fields: If False, only select automatic fields
+        """
+        for catalog in catalogs:
+            schema = menagerie.get_annotated_schema(conn_id, catalog['stream_id'])
+
+            non_selected_properties = []
+            if not select_all_fields:
+                # get a list of all properties so that none are selected
+                non_selected_properties = schema.get('annotated-schema', {}).get(
+                    'properties', {})
+                # remove properties that are automatic
+                for prop in self.expected_automatic_fields().get(catalog['stream_name'], []):
+                    if prop in non_selected_properties:
+                        del non_selected_properties[prop]
+            additional_md = []
+
+            connections.select_catalog_and_fields_via_metadata(
+                conn_id, catalog, schema, additional_md=additional_md,
+                non_selected_fields=non_selected_properties.keys()
+            )