🐛 bugfix Source Okta: Fix endless loop while syncing data from logs stream (airbytehq#4464)

Co-authored-by: sabifranjo <[email protected]>
Co-authored-by: Sabolc Franjo <[email protected]>

Author: 3 people

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/1d4fdb25-64fc-4569-92da-fcdca79a8372.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "1d4fdb25-64fc-4569-92da-fcdca79a8372",
   "name": "Okta",
   "dockerRepository": "airbyte/source-okta",
-  "dockerImageTag": "0.1.1",
+  "dockerImageTag": "0.1.2",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/okta"
 }

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

@@ -347,5 +347,5 @@
 - sourceDefinitionId: 1d4fdb25-64fc-4569-92da-fcdca79a8372
   name: Okta
   dockerRepository: airbyte/source-okta
-  dockerImageTag: 0.1.1
+  dockerImageTag: 0.1.2
   documentationUrl: https://docs.airbyte.io/integrations/sources/okta

airbyte-integrations/connectors/source-okta/CHANGELOG.md

@@ -1,8 +1,3 @@
 # Changelog
 
-## 0.1.0
-Initial Release
-Added the following incremental streams:
-- Groups
-- Logs
-- Users
+See the [docs](https://docs.airbyte.io/integrations/sources/okta#changelog)

airbyte-integrations/connectors/source-okta/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.1
+LABEL io.airbyte.version=0.1.2
 LABEL io.airbyte.name=airbyte/source-okta

airbyte-integrations/connectors/source-okta/source_okta/source.py

@@ -23,7 +23,6 @@
 #
 
 
-import re
 from abc import ABC, abstractmethod
 from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Tuple
 from urllib import parse
@@ -51,13 +50,21 @@ def url_base(self) -> str:
     def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
         # Follow the next page cursor
         # https://developer.okta.com/docs/reference/api-overview/#pagination
-        link_regex = r'<(.+?)>; rel="(.+?)"[,\s]*'
-        raw_links = response.headers["link"]
-        for link, cursor_type in re.findall(link_regex, raw_links):
-            if cursor_type == "next":
-                parsed_link = parse.urlparse(link)
-                query_params = dict(parse.parse_qsl(parsed_link.query))
-                return query_params
+        links = response.links
+        if "next" in links:
+            next_url = links["next"]["url"]
+            parsed_link = parse.urlparse(next_url)
+            query_params = dict(parse.parse_qsl(parsed_link.query))
+
+            # Typically, the absence of the "next" link header indicates there are more pages to read
+            # However, some streams contain the "next" link header even when there are no more pages to read
+            # See https://developer.okta.com/docs/reference/api-overview/#link-header
+            if "self" in links:
+                if links["self"]["url"] == next_url:
+                    return None
+
+            return query_params
+
         return None
 
     def request_params(

airbyte-integrations/connectors/source-okta/unit_tests/unit_test.py

@@ -23,5 +23,45 @@
 #
 
 
-def test_example_method():
-    assert True
+import pytest
+import requests
+from source_okta.source import OktaStream
+
+
+class MockStream(OktaStream):
+    primary_key = "None"
+
+    def __init__(self):
+        super().__init__("")
+
+    def path(self, **kwargs) -> str:
+        return "path"
+
+
+class TestPagination:
+    @pytest.mark.parametrize(
+        ("_self_header", "_next_header", "expected_npt", "assert_msg"),
+        [
+            (None, "XYZ", {"after": "XYZ"}, "Expected to receive a new page token if next header is set and self is not set"),
+            ("ABC", "XYZ", {"after": "XYZ"}, "Expected to receive a new page token if next and self headers have different values"),
+            ("ABC", "ABC", None, "Expected no new page token if next and self headers are the same"),
+            ("ABC", None, None, "Expected no new page token if next header is not set"),
+        ],
+    )
+    def test_pagination(self, _self_header, _next_header, expected_npt, assert_msg):
+        stream = MockStream()
+
+        fake_response = requests.Response()
+        link_header = ""
+
+        if _self_header:
+            link_header += f'<https://somedomain.com/api/v1/users?after={_self_header}>; rel="self",'
+
+        if _next_header:
+            link_header += f'<https://somedomain.com/api/v1/users?after={_next_header}>; rel="next"'
+
+        fake_response.headers = {"link": link_header}
+
+        actual_npt = stream.next_page_token(fake_response)
+
+        assert actual_npt == expected_npt, assert_msg

docs/integrations/sources/okta.md

@@ -54,3 +54,13 @@ Different Okta APIs require different admin privilege levels. API tokens inherit
 5. Record the token value. This is the only opportunity to see it and record it.
 8. In Airbyte, create a Okta source.
 9. You can now pull data from your Okta instance!
+
+
+## Changelog
+
+| Version | Date       | Pull Request | Subject |
+| :------ | :--------  | :-----       | :------ |
+| 0.1.2 | 2021-07-01 | [4456](https://github.com/airbytehq/airbyte/pull/4456)| Bugfix infinite pagination in logs stream |
+| 0.1.1 | 2021-06-09 | [3937](https://github.com/airbytehq/airbyte/pull/3973) | Add `AIRBYTE_ENTRYPOINT` env variable for kubernetes support|
+| 0.1.0   | 2021-05-30 | [3563](https://github.com/airbytehq/airbyte/pull/3563) | Initial Release |
+