Add "impersonator" field to WorkOSAuthenticationResponse

tests/conftest.py

@@ -26,7 +26,7 @@ def __init__(self, content, status_code, headers=None):
 
 @pytest.fixture
 def set_api_key(monkeypatch):
-    monkeypatch.setattr(workos, "api_key", "sk_abdsomecharactersm284")
+    monkeypatch.setattr(workos, "api_key", "sk_test")
 
 
 @pytest.fixture

tests/test_directory_sync.py

@@ -13,7 +13,6 @@ def setup(self, set_api_key, set_client_id):
 
     @pytest.fixture
     def mock_users(self):
-
         user_list = [MockDirectoryUser(id=str(i)).to_dict() for i in range(100)]
 
         return {
@@ -36,7 +35,6 @@ def mock_users(self):
 
     @pytest.fixture
     def mock_default_limit_users(self):
-
         user_list = [MockDirectoryUser(id=str(i)).to_dict() for i in range(10)]
 
         return {
@@ -59,7 +57,6 @@ def mock_default_limit_users(self):
 
     @pytest.fixture
     def mock_default_limit_users_v2(self):
-
         user_list = [MockDirectoryUser(id=str(i)).to_dict() for i in range(10)]
 
         dict_response = {
@@ -84,7 +81,6 @@ def mock_default_limit_users_v2(self):
 
     @pytest.fixture
     def mock_users_pagination_response(self):
-
         user_list = [MockDirectoryUser(id=str(i)).to_dict() for i in range(90)]
 
         return {
@@ -107,7 +103,6 @@ def mock_users_pagination_response(self):
 
     @pytest.fixture
     def mock_groups(self):
-
         group_list = [MockDirectoryGroup(id=str(i)).to_dict() for i in range(5000)]
 
         return {
@@ -130,7 +125,6 @@ def mock_groups(self):
 
     @pytest.fixture
     def mock_default_limit_groups(self):
-
         group_list = [MockDirectoryGroup(id=str(i)).to_dict() for i in range(10)]
 
         return {
@@ -153,7 +147,6 @@ def mock_default_limit_groups(self):
 
     @pytest.fixture
     def mock_default_limit_groups_v2(self):
-
         group_list = [MockDirectoryGroup(id=str(i)).to_dict() for i in range(10)]
 
         dict_response = {
@@ -178,7 +171,6 @@ def mock_default_limit_groups_v2(self):
 
     @pytest.fixture
     def mock_groups_pagination_reponse(self):
-
         group_list = [MockDirectoryGroup(id=str(i)).to_dict() for i in range(4990)]
 
         return {
@@ -243,7 +235,6 @@ def mock_group(self):
 
     @pytest.fixture
     def mock_directories(self):
-
         directory_list = [MockDirectory(id=str(i)).to_dict() for i in range(5000)]
 
         return {
@@ -266,7 +257,6 @@ def mock_directories(self):
 
     @pytest.fixture
     def mock_directories_with_limit(self):
-
         directory_list = [MockDirectory(id=str(i)).to_dict() for i in range(4)]
 
         return {
@@ -288,7 +278,6 @@ def mock_directories_with_limit(self):
 
     @pytest.fixture
     def mock_directories_with_limit_v2(self):
-
         directory_list = [MockDirectory(id=str(i)).to_dict() for i in range(4)]
 
         dict_response = {
@@ -312,7 +301,6 @@ def mock_directories_with_limit_v2(self):
 
     @pytest.fixture
     def mock_default_limit_directories(self):
-
         directory_list = [MockDirectory(id=str(i)).to_dict() for i in range(10)]
 
         return {
@@ -335,7 +323,6 @@ def mock_default_limit_directories(self):
 
     @pytest.fixture
     def mock_default_limit_directories_v2(self):
-
         directory_list = [MockDirectory(id=str(i)).to_dict() for i in range(10)]
 
         dict_response = {
@@ -360,7 +347,6 @@ def mock_default_limit_directories_v2(self):
 
     @pytest.fixture
     def mock_directories_pagination_response(self):
-
         directory_list = [MockDirectory(id=str(i)).to_dict() for i in range(4990)]
 
         return {
@@ -502,7 +488,6 @@ def test_list_directories_auto_pagination_v2(
         mock_directories,
         mock_request_method,
     ):
-
         directories = mock_default_limit_directories_v2
         mock_request_method("get", mock_directories_pagination_response, 200)
         all_directories = directories.auto_paging_iter()

tests/test_sso.py

@@ -63,7 +63,6 @@ def mock_connection(self):
 
     @pytest.fixture
     def mock_connections(self):
-
         connection_list = [MockConnection(id=str(i)).to_dict() for i in range(5000)]
 
         return {
@@ -84,7 +83,6 @@ def mock_connections(self):
 
     @pytest.fixture
     def mock_connections_with_limit(self):
-
         connection_list = [MockConnection(id=str(i)).to_dict() for i in range(4)]
 
         return {
@@ -106,7 +104,6 @@ def mock_connections_with_limit(self):
 
     @pytest.fixture
     def mock_connections_with_limit_v2(self, set_api_key_and_client_id):
-
         connection_list = [MockConnection(id=str(i)).to_dict() for i in range(4)]
 
         dict_response = {
@@ -129,7 +126,6 @@ def mock_connections_with_limit_v2(self, set_api_key_and_client_id):
 
     @pytest.fixture
     def mock_connections_with_default_limit(self):
-
         connection_list = [MockConnection(id=str(i)).to_dict() for i in range(10)]
 
         return {
@@ -152,7 +148,6 @@ def mock_connections_with_default_limit(self):
 
     @pytest.fixture
     def mock_connections_with_default_limit_v2(self, setup_with_client_id):
-
         connection_list = [MockConnection(id=str(i)).to_dict() for i in range(10)]
 
         dict_response = {
@@ -176,7 +171,6 @@ def mock_connections_with_default_limit_v2(self, setup_with_client_id):
 
     @pytest.fixture
     def mock_connections_pagination_response(self):
-
         connection_list = [MockConnection(id=str(i)).to_dict() for i in range(4990)]
 
         return {
@@ -565,7 +559,6 @@ def test_list_connections_honors_limit(
         mock_request_method,
         setup_with_client_id,
     ):
-
         connections = mock_connections_with_limit
         mock_request_method("get", mock_connections_pagination_response, 200)
         all_connections = SSO.construct_from_response(connections).auto_paging_iter()
@@ -579,7 +572,6 @@ def test_list_connections_honors_limit_v2(
         mock_request_method,
         setup_with_client_id,
     ):
-
         connections = mock_connections_with_limit_v2
         mock_request_method("get", mock_connections_pagination_response, 200)
         all_connections = connections.auto_paging_iter()

tests/test_user_management.py

@@ -140,6 +140,19 @@ def mock_auth_response(self):
 
         return {"user": user, "organization_id": "org_12345"}
 
+    @pytest.fixture
+    def mock_auth_response_with_impersonator(self):
+        user = MockUser("user_01H7ZGXFP5C6BBQY6Z7277ZCT0").to_dict()
+
+        return {
+            "user": user,
+            "organization_id": "org_12345",
+            "impersonator": {
+                "email": "[email protected]",
+                "reason": "Debugging an account issue.",
+            },
+        }
+
     @pytest.fixture
     def mock_magic_auth_challenge_response(self):
         return {
@@ -546,7 +559,7 @@ def test_authenticate_with_password(
         assert request["json"]["user_agent"] == user_agent
         assert request["json"]["ip_address"] == ip_address
         assert request["json"]["client_id"] == "client_b27needthisforssotemxo"
-        assert request["json"]["client_secret"] == "sk_abdsomecharactersm284"
+        assert request["json"]["client_secret"] == "sk_test"
         assert request["json"]["grant_type"] == "password"
 
     def test_authenticate_with_code(self, capture_and_mock_request, mock_auth_response):
@@ -569,9 +582,28 @@ def test_authenticate_with_code(self, capture_and_mock_request, mock_auth_respon
         assert request["json"]["user_agent"] == user_agent
         assert request["json"]["ip_address"] == ip_address
         assert request["json"]["client_id"] == "client_b27needthisforssotemxo"
-        assert request["json"]["client_secret"] == "sk_abdsomecharactersm284"
+        assert request["json"]["client_secret"] == "sk_test"
         assert request["json"]["grant_type"] == "authorization_code"
 
+    def test_authenticate_impersonator_with_code(
+        self, capture_and_mock_request, mock_auth_response_with_impersonator
+    ):
+        code = "test_code"
+
+        url, request = capture_and_mock_request(
+            "post", mock_auth_response_with_impersonator, 200
+        )
+
+        response = self.user_management.authenticate_with_code(
+            code=code,
+        )
+
+        print(response)
+        assert url[0].endswith("user_management/authenticate")
+        assert response["user"]["id"] == "user_01H7ZGXFP5C6BBQY6Z7277ZCT0"
+        assert response["impersonator"]["email"] == "[email protected]"
+        assert response["impersonator"]["reason"] == "Debugging an account issue."
+
     def test_authenticate_with_magic_auth(
         self, capture_and_mock_request, mock_auth_response
     ):
@@ -597,7 +629,7 @@ def test_authenticate_with_magic_auth(
         assert request["json"]["email"] == email
         assert request["json"]["ip_address"] == ip_address
         assert request["json"]["client_id"] == "client_b27needthisforssotemxo"
-        assert request["json"]["client_secret"] == "sk_abdsomecharactersm284"
+        assert request["json"]["client_secret"] == "sk_test"
         assert (
             request["json"]["grant_type"]
             == "urn:workos:oauth:grant-type:magic-auth:code"
@@ -631,7 +663,7 @@ def test_authenticate_with_email_verification(
         )
         assert request["json"]["ip_address"] == ip_address
         assert request["json"]["client_id"] == "client_b27needthisforssotemxo"
-        assert request["json"]["client_secret"] == "sk_abdsomecharactersm284"
+        assert request["json"]["client_secret"] == "sk_test"
         assert (
             request["json"]["grant_type"]
             == "urn:workos:oauth:grant-type:email-verification:code"
@@ -669,7 +701,7 @@ def test_authenticate_with_totp(self, capture_and_mock_request, mock_auth_respon
         )
         assert request["json"]["ip_address"] == ip_address
         assert request["json"]["client_id"] == "client_b27needthisforssotemxo"
-        assert request["json"]["client_secret"] == "sk_abdsomecharactersm284"
+        assert request["json"]["client_secret"] == "sk_test"
         assert request["json"]["grant_type"] == "urn:workos:oauth:grant-type:mfa-totp"
 
     def test_authenticate_with_organization_selection(
@@ -700,7 +732,7 @@ def test_authenticate_with_organization_selection(
         )
         assert request["json"]["ip_address"] == ip_address
         assert request["json"]["client_id"] == "client_b27needthisforssotemxo"
-        assert request["json"]["client_secret"] == "sk_abdsomecharactersm284"
+        assert request["json"]["client_secret"] == "sk_test"
         assert (
             request["json"]["grant_type"]
             == "urn:workos:oauth:grant-type:organization-selection"

tests/test_webhooks.py

@@ -81,15 +81,13 @@ def test_timestamp_outside_threshold(
         assert "Timestamp outside the tolerance zone" in str(err.value)
 
     def test_sig_hash_does_not_match_expected_sig_length(self, mock_sig_hash):
-
         result = self.webhooks.constant_time_compare(
             mock_sig_hash,
             "df25b6efdd39d82e7b30e75ea19655b306860ad5cde3eeaeb6f1dfea029ea25",
         )
         assert result == False
 
     def test_sig_hash_does_not_match_expected_sig_value(self, mock_sig_hash):
-
         result = self.webhooks.constant_time_compare(
             mock_sig_hash,
             "df25b6efdd39d82e7b30e75ea19655b306860ad5cde3eeaeb6f1dfea029ea252",

workos/resources/user_management.py

@@ -21,6 +21,14 @@ def construct_from_response(cls, response):
         user = WorkOSUser.construct_from_response(response["user"])
         authentication_response.user = user
 
+        if "impersonator" in response:
+            impersonator = WorkOSImpersonator.construct_from_response(
+                response["impersonator"]
+            )
+            authentication_response.impersonator = impersonator
+        else:
+            authentication_response.impersonator = None
+
         return authentication_response
 
     def to_dict(self):
@@ -31,6 +39,9 @@ def to_dict(self):
         user_dict = self.user.to_dict()
         authentication_response_dict["user"] = user_dict
 
+        if self.impersonator:
+            authentication_response_dict["impersonator"] = self.impersonator.to_dict()
+
         return authentication_response_dict
 
 
@@ -120,3 +131,16 @@ class WorkOSUser(WorkOSBaseResource):
         "created_at",
         "updated_at",
     ]
+
+
+class WorkOSImpersonator(WorkOSBaseResource):
+    """Representation of a WorkOS Dashboard member impersonating a user
+
+    Attributes:
+        OBJECT_FIELDS (list): List of fields a WorkOSImpersonator comprises.
+    """
+
+    OBJECT_FIELDS = [
+        "email",
+        "reason",
+    ]

workos/webhooks.py

@@ -23,7 +23,6 @@ def request_helper(self):
     DEFAULT_TOLERANCE = 180
 
     def verify_event(self, payload, sig_header, secret, tolerance=DEFAULT_TOLERANCE):
-
         if payload is None:
             raise ValueError("Payload body is missing and is a required parameter")
         if sig_header is None: