🐛 Source Zendesk Support: add Subscription Plan check for available…

… streams (#15233)

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -1065,7 +1065,7 @@
 - name: Zendesk Support
   sourceDefinitionId: 79c1aa37-dae3-42ae-b333-d1c105477715
   dockerRepository: airbyte/source-zendesk-support
-  dockerImageTag: 0.2.14
+  dockerImageTag: 0.2.15
   documentationUrl: https://docs.airbyte.io/integrations/sources/zendesk-support
   icon: zendesk.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -10381,7 +10381,7 @@
               path_in_connector_config:
               - "credentials"
               - "client_secret"
-- dockerImage: "airbyte/source-zendesk-support:0.2.14"
+- dockerImage: "airbyte/source-zendesk-support:0.2.15"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/zendesk-support"
     connectionSpecification:

airbyte-integrations/connectors/source-zendesk-support/Dockerfile

@@ -25,5 +25,5 @@ COPY source_zendesk_support ./source_zendesk_support
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.2.14
+LABEL io.airbyte.version=0.2.15
 LABEL io.airbyte.name=airbyte/source-zendesk-support

airbyte-integrations/connectors/source-zendesk-support/source_zendesk_support/source.py

@@ -31,8 +31,13 @@
     Tickets,
     Users,
     UserSettingsStream,
+    UserSubscriptionStream,
 )
 
+# The Zendesk Subscription Plan gains complete access to all the streams
+FULL_ACCESS_PLAN = "Enterprise"
+FULL_ACCESS_ONLY_STREAMS = ["ticket_forms"]
+
 
 class BasicApiTokenAuthenticator(TokenAuthenticator):
     """basic Authorization header"""
@@ -104,24 +109,32 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         :param config: A Mapping of the user input configuration as defined in the connector spec.
         """
         args = self.convert_config2stream_args(config)
-        # sorted in alphabet order
-        return [
-            GroupMemberships(**args),
-            Groups(**args),
-            Macros(**args),
-            Organizations(**args),
-            SatisfactionRatings(**args),
-            SlaPolicies(**args),
-            Tags(**args),
-            TicketAudits(**args),
-            TicketComments(**args),
-            TicketFields(**args),
-            TicketForms(**args),
-            TicketMetrics(**args),
-            TicketMetricEvents(**args),
-            Tickets(**args),
-            Users(**args),
-            Brands(**args),
-            CustomRoles(**args),
-            Schedules(**args),
-        ]
+        all_streams_mapping = {
+            # sorted in alphabet order
+            "group_membership": GroupMemberships(**args),
+            "groups": Groups(**args),
+            "macros": Macros(**args),
+            "organizations": Organizations(**args),
+            "satisfaction_ratings": SatisfactionRatings(**args),
+            "sla_policies": SlaPolicies(**args),
+            "tags": Tags(**args),
+            "ticket_audits": TicketAudits(**args),
+            "ticket_comments": TicketComments(**args),
+            "ticket_fields": TicketFields(**args),
+            "ticket_forms": TicketForms(**args),
+            "ticket_metrics": TicketMetrics(**args),
+            "ticket_metric_events": TicketMetricEvents(**args),
+            "tickets": Tickets(**args),
+            "users": Users(**args),
+            "brands": Brands(**args),
+            "custom_roles": CustomRoles(**args),
+            "schedules": Schedules(**args),
+        }
+        # check the users Zendesk Subscription Plan
+        subscription_plan = UserSubscriptionStream(**args).get_subscription_plan()
+        if subscription_plan != FULL_ACCESS_PLAN:
+            # only those the streams that are not listed in FULL_ACCESS_ONLY_STREAMS should be available
+            return [stream_cls for stream_name, stream_cls in all_streams_mapping.items() if stream_name not in FULL_ACCESS_ONLY_STREAMS]
+        else:
+            # all streams should be available for user, otherwise
+            return [stream_cls for stream_cls in all_streams_mapping.values()]

airbyte-integrations/connectors/source-zendesk-support/source_zendesk_support/streams.py

@@ -47,7 +47,7 @@ class SourceZendeskException(Exception):
 
 class SourceZendeskSupportFuturesSession(FuturesSession):
     """
-    Check the docs at https://github.com/ross/requests-futures.
+    Check the docs at https://github.com/ross/requests-futures
     Used to async execute a set of requests.
     """
 
@@ -534,7 +534,7 @@ class TicketFields(SourceZendeskSupportStream):
 
 
 class TicketForms(SourceZendeskSupportCursorPaginationStream):
-    """TicketForms stream: https://developer.zendesk.com/api-reference/ticketing/tickets/ticket_forms/"""
+    """TicketForms stream: https://developer.zendesk.com/api-reference/ticketing/tickets/ticket_forms"""
 
 
 class TicketMetrics(SourceZendeskSupportCursorPaginationStream):
@@ -646,3 +646,23 @@ def get_settings(self) -> Mapping[str, Any]:
         for resp in self.read_records(SyncMode.full_refresh):
             return resp
         raise SourceZendeskException("not found settings")
+
+
+class UserSubscriptionStream(SourceZendeskSupportFullRefreshStream):
+    """Stream for checking read permissions for streams"""
+
+    def path(self, *args, **kwargs) -> str:
+        return "/api/v2/account/subscription.json"
+
+    def next_page_token(self, *args, **kwargs) -> Optional[Mapping[str, Any]]:
+        return None
+
+    def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
+        subscription_plan = response.json().get("subscription").get("plan_name")
+        if subscription_plan:
+            yield subscription_plan
+
+    def get_subscription_plan(self) -> Mapping[str, Any]:
+        for result in self.read_records(SyncMode.full_refresh):
+            return result
+        raise SourceZendeskException("Could not read User's Subscription Plan.")

airbyte-integrations/connectors/source-zendesk-support/unit_tests/unit_test.py

@@ -40,6 +40,7 @@
     Tickets,
     Users,
     UserSettingsStream,
+    UserSubscriptionStream,
 )
 from test_data.data import TICKET_EVENTS_STREAM_RESPONSE
 from utils import read_full_refresh
@@ -130,6 +131,22 @@ def test_check(response, check_passed):
         assert check_passed == result
 
 
+@pytest.mark.parametrize(
+    "response, expected_n_streams",
+    [
+        ("Enterprise", 18),
+        # if restricted, TicketForms stream will not be listed
+        ("Other", 17),
+    ],
+    ids=["full_access", "restricted_access"],
+)
+def test_full_access_streams(response, expected_n_streams):
+    with patch.object(UserSubscriptionStream, "get_subscription_plan", return_value=response) as mock_method:
+        result = SourceZendeskSupport().streams(config=TEST_CONFIG)
+        mock_method.assert_called()
+        assert len(result) == expected_n_streams
+
+
 @pytest.fixture(autouse=True)
 def time_sleep_mock(mocker):
     time_mock = mocker.patch("time.sleep", lambda x: None)
@@ -265,10 +282,12 @@ class TestAllStreams:
         ],
     )
     def test_streams(self, expected_stream_cls):
-        streams = SourceZendeskSupport().streams(TEST_CONFIG)
-        for stream in streams:
-            if expected_stream_cls in streams:
-                assert isinstance(stream, expected_stream_cls)
+        with patch.object(UserSubscriptionStream, "get_subscription_plan", return_value="Enterprise") as mock_method:
+            streams = SourceZendeskSupport().streams(TEST_CONFIG)
+            mock_method.assert_called()
+            for stream in streams:
+                if expected_stream_cls in streams:
+                    assert isinstance(stream, expected_stream_cls)
 
     @pytest.mark.parametrize(
         "stream_cls, expected",