🎉 Source Okta: return deprovisioned users

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

@@ -628,7 +628,7 @@
 - name: Okta
   sourceDefinitionId: 1d4fdb25-64fc-4569-92da-fcdca79a8372
   dockerRepository: airbyte/source-okta
-  dockerImageTag: 0.1.8
+  dockerImageTag: 0.1.9
   documentationUrl: https://docs.airbyte.io/integrations/sources/okta
   icon: okta.svg
   sourceType: api

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

@@ -6027,7 +6027,7 @@
         - - "client_secret"
         oauthFlowOutputParameters:
         - - "access_token"
-- dockerImage: "airbyte/source-okta:0.1.8"
+- dockerImage: "airbyte/source-okta:0.1.9"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/okta"
     connectionSpecification:

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.8
+LABEL io.airbyte.version=0.1.9
 LABEL io.airbyte.name=airbyte/source-okta

airbyte-integrations/connectors/source-okta/acceptance-test-config.yml

@@ -16,6 +16,8 @@ tests:
   basic_read:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
+      expect_records:
+        path: "integration_tests/expected_records.txt"
     - config_path: "secrets/config_api_token.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
   full_refresh:

airbyte-integrations/connectors/source-okta/integration_tests/expected_records.txt

@@ -0,0 +1 @@
+{"stream": "users", "data": {"id": "00u5vhynsalJZ2eMg5d7", "status": "DEPROVISIONED", "created":"2022-07-21T20:46:27.000Z", "activated":"2022-07-21T20:47:13.000Z", "statusChanged": "2022-07-21T20:49:13.000Z", "lastLogin": null, "lastUpdated": "2022-07-21T20:49:13.000Z", "passwordChanged": "2022-07-21T20:47:35.000Z", "type": {"id": "otymj7cw2AJFAHvOO5d6"}, "profile": {"firstName": "Test", "lastName": "DEPROVISIONED", "mobilePhone": null, "secondEmail": null, "login": "[email protected]", "email": "[email protected]"}, "credentials": {"emails": [{"value": "[email protected]", "status": "VERIFIED", "type": "PRIMARY"}], "provider": {"type": "OKTA", "name": "OKTA"}}, "_links": {"self": {"href": "https://dev-01177082.okta.com/api/v1/users/00u5vhynsalJZ2eMg5d7"}}}, "emitted_at": 1655800476224}

airbyte-integrations/connectors/source-okta/source_okta/schemas/groups.json

@@ -1,9 +1,6 @@
 {
   "properties": {
     "_links": {
-      "additionalProperties": {
-        "type": ["array", "object", "null"]
-      },
       "type": ["object", "null"]
     },
     "created": {

airbyte-integrations/connectors/source-okta/source_okta/schemas/shared/shared-user.json

@@ -2,9 +2,6 @@
   "$id": "shared-users.json",
   "properties": {
     "_links": {
-      "additionalProperties": {
-        "type": ["object", "null"]
-      },
       "type": ["object", "null"]
     },
     "activated": {

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

@@ -199,10 +199,28 @@ def request_params(
 class Users(IncrementalOktaStream):
     cursor_field = "lastUpdated"
     primary_key = "id"
+    # Should add all statuses to filter. Considering Okta documentation https://developer.okta.com/docs/reference/api/users/#list-all-users,
+    # users with "DEPROVISIONED" status are not returned by default.
+    statuses = ["ACTIVE", "DEPROVISIONED", "LOCKED_OUT", "PASSWORD_EXPIRED", "PROVISIONED", "RECOVERY", "STAGED", "SUSPENDED"]
 
     def path(self, **kwargs) -> str:
         return "users"
 
+    def request_params(
+        self,
+        stream_state: Mapping[str, Any],
+        stream_slice: Mapping[str, any] = None,
+        next_page_token: Mapping[str, Any] = None,
+    ) -> MutableMapping[str, Any]:
+        params = super().request_params(stream_state, stream_slice, next_page_token)
+        status_filters = " or ".join([f'status eq "{status}"' for status in self.statuses])
+        if "filter" in params:
+            # add status_filters to existing filters
+            params["filter"] = f'{params["filter"]} and ({status_filters})'
+        else:
+            params["filter"] = status_filters
+        return params
+
 
 class CustomRoles(OktaStream):
     primary_key = "id"

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

@@ -86,6 +86,12 @@ def auth_token_config():
     return {"credentials": {"auth_type": "api_token", "api_token": "test_token"}}
 
 
+@pytest.fixture()
+def user_status_filter():
+    statuses = ["ACTIVE", "DEPROVISIONED", "LOCKED_OUT", "PASSWORD_EXPIRED", "PROVISIONED", "RECOVERY", "STAGED", "SUSPENDED"]
+    return " or ".join([f'status eq "{status}"' for status in statuses])
+
+
 @pytest.fixture()
 def users_instance(api_url):
     """

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

@@ -19,7 +19,6 @@
 
 
 class TestAuthentication:
-
     def test_init_token_authentication_init(self, token_config, auth_token_config):
         source_okta = SourceOkta()
         token_authenticator_instance = source_okta.initialize_authenticator(config=token_config)

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

@@ -55,7 +55,6 @@ def test_should_retry(self, patch_base_class, http_status, should_retry, url_bas
 
 
 class TestOktaStream:
-
     def test_okta_stream_request_params(self, patch_base_class, url_base):
         stream = OktaStream(url_base=url_base)
         inputs = {"stream_slice": None, "stream_state": None, "next_page_token": None}
@@ -139,7 +138,6 @@ def test_okta_stream_http_method(self, patch_base_class, url_base):
 
 
 class TestNextPageToken:
-
     def test_next_page_token(self, patch_base_class, users_instance, url_base, api_url):
         stream = OktaStream(url_base=url_base)
         response = MagicMock(requests.Response)
@@ -169,29 +167,28 @@ def test_next_page_token_link_have_self_and_equal_next(self, patch_base_class, u
 
 
 class TestStreamUsers:
-
     def test_stream_users(self, requests_mock, patch_base_class, users_instance, url_base, api_url):
         stream = Users(url_base=url_base)
         requests_mock.get(f"{api_url}/users", json=[users_instance])
         inputs = {"sync_mode": SyncMode.incremental}
         assert list(stream.read_records(**inputs)) == [users_instance]
 
-    def test_users_request_params_out_of_next_page_token(self, patch_base_class, url_base):
+    def test_users_request_params_out_of_next_page_token(self, patch_base_class, url_base, user_status_filter):
         stream = Users(url_base=url_base)
         inputs = {"stream_slice": None, "stream_state": None, "next_page_token": None}
-        expected_params = {"limit": 200}
+        expected_params = {"limit": 200, "filter": user_status_filter}
         assert stream.request_params(**inputs) == expected_params
 
-    def test_users_source_request_params_have_next_cursor(self, patch_base_class, url_base):
+    def test_users_source_request_params_have_next_cursor(self, patch_base_class, url_base, user_status_filter):
         stream = Users(url_base=url_base)
         inputs = {"stream_slice": None, "stream_state": None, "next_page_token": {"next_cursor": "123"}}
-        expected_params = {"limit": 200, "next_cursor": "123"}
+        expected_params = {"limit": 200, "next_cursor": "123", "filter": user_status_filter}
         assert stream.request_params(**inputs) == expected_params
 
-    def test_users_source_request_params_have_latest_entry(self, patch_base_class, url_base):
+    def test_users_source_request_params_have_latest_entry(self, patch_base_class, url_base, user_status_filter):
         stream = Users(url_base=url_base)
         inputs = {"stream_slice": None, "stream_state": {"lastUpdated": "some_date"}, "next_page_token": {"next_cursor": "123"}}
-        expected_params = {"limit": 200, "next_cursor": "123", "filter": 'lastUpdated gt "some_date"'}
+        expected_params = {"limit": 200, "next_cursor": "123", "filter": f'lastUpdated gt "some_date" and ({user_status_filter})'}
         assert stream.request_params(**inputs) == expected_params
 
     def test_users_source_parse_response(self, requests_mock, patch_base_class, users_instance, url_base, api_url):
@@ -201,7 +198,6 @@ def test_users_source_parse_response(self, requests_mock, patch_base_class, user
 
 
 class TestStreamCustomRoles:
-
     def test_custom_roles(self, requests_mock, patch_base_class, custom_role_instance, url_base, api_url):
         stream = CustomRoles(url_base=url_base)
         record = {"roles": [custom_role_instance]}
@@ -217,7 +213,6 @@ def test_custom_roles_parse_response(self, requests_mock, patch_base_class, cust
 
 
 class TestStreamGroups:
-
     def test_groups(self, requests_mock, patch_base_class, groups_instance, url_base, api_url):
         stream = Groups(url_base=url_base)
         requests_mock.get(f"{api_url}/groups?limit=200", json=[groups_instance])
@@ -231,7 +226,6 @@ def test_groups_parse_response(self, requests_mock, patch_base_class, groups_ins
 
 
 class TestStreamGroupMembers:
-
     def test_group_members(self, requests_mock, patch_base_class, group_members_instance, url_base, api_url):
         stream = GroupMembers(url_base=url_base)
         group_id = "test_group_id"
@@ -267,7 +261,6 @@ def test_group_member_request_get_update_state(self, latest_record_instance, url
 
 
 class TestStreamGroupRoleAssignment:
-
     def test_group_role_assignments(self, requests_mock, patch_base_class, group_role_assignments_instance, url_base, api_url):
         stream = GroupRoleAssignments(url_base=url_base)
         group_id = "test_group_id"
@@ -292,7 +285,6 @@ def test_group_role_assignments_slice_stream(
 
 
 class TestStreamLogs:
-
     def test_logs(self, requests_mock, patch_base_class, logs_instance, url_base, api_url):
         stream = Logs(url_base=url_base)
         requests_mock.get(f"{api_url}/logs?limit=200", json=[logs_instance])
@@ -317,7 +309,6 @@ def test_logs_request_params_for_until(self, patch_base_class, logs_instance, ur
 
 
 class TestStreamUserRoleAssignment:
-
     def test_user_role_assignments(self, requests_mock, patch_base_class, user_role_assignments_instance, url_base, api_url):
         stream = UserRoleAssignments(url_base=url_base)
         user_id = "test_user_id"

docs/integrations/sources/okta.md

@@ -59,14 +59,15 @@ Different Okta APIs require different admin privilege levels. API tokens inherit
 
 ## Changelog
 
-| Version | Date       | Pull Request | Subject |
-|:--------|:-----------| :--- | :--- |
-| 0.1.8   | 2022-07-19 | [14710](https://github.com/airbytehq/airbyte/pull/14710)    | Implement OAuth2.0 authorization method                     |
+| Version | Date       | Pull Request                                             | Subject                                                                        |
+|:--------|:-----------|:---------------------------------------------------------|:-------------------------------------------------------------------------------|
+| 0.1.9   | 2022-07-25 | [15001](https://github.com/airbytehq/airbyte/pull/15001) | Return deprovisioned users                                                     |
+| 0.1.8   | 2022-07-19 | [14710](https://github.com/airbytehq/airbyte/pull/14710) | Implement OAuth2.0 authorization method                                        |
 | 0.1.7   | 2022-07-13 | [14556](https://github.com/airbytehq/airbyte/pull/14556) | add User_Role_Assignments and Group_Role_Assignments streams (full fetch only) |
-| 0.1.6   | 2022-07-11 | [14610](https://github.com/airbytehq/airbyte/pull/14610) | add custom roles stream |
-| 0.1.5   | 2022-07-04 | [14380](https://github.com/airbytehq/airbyte/pull/14380) | add Group_Members stream to okta source |
-| 0.1.4   | 2021-11-02 | [7584](https://github.com/airbytehq/airbyte/pull/7584) | Fix incremental params for log stream |
-| 0.1.3   | 2021-09-08 | [5905](https://github.com/airbytehq/airbyte/pull/5905) | Fix incremental stream defect |
-| 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 |
+| 0.1.6   | 2022-07-11 | [14610](https://github.com/airbytehq/airbyte/pull/14610) | add custom roles stream                                                        |
+| 0.1.5   | 2022-07-04 | [14380](https://github.com/airbytehq/airbyte/pull/14380) | add Group_Members stream to okta source                                        |
+| 0.1.4   | 2021-11-02 | [7584](https://github.com/airbytehq/airbyte/pull/7584)   | Fix incremental params for log stream                                          |
+| 0.1.3   | 2021-09-08 | [5905](https://github.com/airbytehq/airbyte/pull/5905)   | Fix incremental stream defect                                                  |
+| 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                                                                |