fix: tests

Author: SoulPancake

test/oauth2_test.py

@@ -12,13 +12,7 @@
 
 from datetime import datetime, timedelta
 from unittest import IsolatedAsyncioTestCase
-
-import unittest
-import pytest
-from unittest.mock import patch, MagicMock
-from openfga_sdk.credentials import CredentialConfiguration
-from openfga_sdk.oauth2 import OAuth2Client
-
+from unittest.mock import patch
 
 import urllib3
 
@@ -501,126 +495,116 @@ async def test_get_authentication_add_scheme_and_path(self, mock_request):
         )
         await rest_client.close()
 
-class TestOAuth2ClientScopeHandling(unittest.TestCase):
-    """Test OAuth2Client scope serialization in synchronous context"""
+    @patch.object(rest.RESTClientObject, "request")
+    async def test_get_authentication_obtain_client_credentials_with_scopes_list(self, mock_request):
+        """
+        Test getting authentication header when method is client credentials with scopes as list
+        """
+        response_body = """
+{
+  "expires_in": 120,
+  "access_token": "AABBCCDD"
+}
+        """
+        mock_request.return_value = mock_response(response_body, 200)
 
-    def setUp(self):
-        self.config = CredentialConfiguration(
-            client_id="test_client_id",
-            client_secret="test_client_secret",
-            api_issuer="https://example.com",
-            api_audience="test_audience"
-        )
-        self.oauth_client = OAuth2Client(self.config)
-
-    @patch('requests.post')
-    def test_list_scopes_serialization(self, mock_post):
-        """Test that list scopes are serialized as space-delimited string"""
-        # Setup mock response
-        mock_response = MagicMock()
-        mock_response.json.return_value = {"access_token": "test_token", "expires_in": 3600}
-        mock_response.status_code = 200
-        mock_post.return_value = mock_response
-
-        # Set list scopes
-        self.config.scopes = ["read", "write", "admin"]
-
-        # Make token request
-        self.oauth_client.request_token()
-
-        # Verify scope parameter was correctly formatted
-        args, kwargs = mock_post.call_args
-        self.assertIn('data', kwargs)
-        self.assertIn('scope', kwargs['data'])
-        self.assertEqual(kwargs['data']['scope'], "read write admin")
-
-    @patch('requests.post')
-    def test_string_scope_serialization(self, mock_post):
-        """Test that string scope is used unchanged"""
-        # Setup mock response
-        mock_response = MagicMock()
-        mock_response.json.return_value = {"access_token": "test_token", "expires_in": 3600}
-        mock_response.status_code = 200
-        mock_post.return_value = mock_response
-
-        # Set string scope
-        self.config.scopes = "read write admin"
-
-        # Make token request
-        self.oauth_client.request_token()
-
-        # Verify scope parameter was passed unchanged
-        args, kwargs = mock_post.call_args
-        self.assertIn('data', kwargs)
-        self.assertIn('scope', kwargs['data'])
-        self.assertEqual(kwargs['data']['scope'], "read write admin")
-
-
-class TestOAuth2ClientScopeHandlingAsync:
-    """Test OAuth2Client scope serialization in asynchronous context"""
-
-    @pytest.fixture
-    def oauth_client(self):
-        config = CredentialConfiguration(
-            client_id="test_client_id",
-            client_secret="test_client_secret",
-            api_issuer="https://example.com",
-            api_audience="test_audience"
-        )
-        return OAuth2Client(config)
-
-    @pytest.mark.asyncio
-    async def test_list_scopes_serialization_async(self, oauth_client, monkeypatch):
-        """Test that list scopes are serialized as space-delimited string in async context"""
-        # Mock the async HTTP response
-        mock_response = MagicMock()
-        mock_response.json = MagicMock(return_value={"access_token": "test_token", "expires_in": 3600})
-        mock_response.status_code = 200
-
-        # Set up mock for the async HTTP client
-        mock_client_session = MagicMock()
-        mock_client_session.post = MagicMock(return_value=mock_response)
-        mock_client_session.__aenter__ = MagicMock(return_value=mock_client_session)
-        mock_client_session.__aexit__ = MagicMock(return_value=None)
-
-        monkeypatch.setattr('aiohttp.ClientSession', MagicMock(return_value=mock_client_session))
-
-        # Set list scopes
-        oauth_client.configuration.scopes = ["read", "write", "admin"]
-
-        # Make async token request
-        await oauth_client.request_token_async()
-
-        # Verify scope parameter was correctly formatted
-        args, kwargs = mock_client_session.post.call_args
-        self.assertIn('data', kwargs)
-        self.assertIn('scope', kwargs['data'])
-        self.assertEqual(kwargs['data']['scope'], "read write admin")
-
-    @pytest.mark.asyncio
-    async def test_string_scope_serialization_async(self, oauth_client, monkeypatch):
-        """Test that string scope is used unchanged in async context"""
-        # Mock the async HTTP response
-        mock_response = MagicMock()
-        mock_response.json = MagicMock(return_value={"access_token": "test_token", "expires_in": 3600})
-        mock_response.status_code = 200
-
-        # Set up mock for the async HTTP client
-        mock_client_session = MagicMock()
-        mock_client_session.post = MagicMock(return_value=mock_response)
-        mock_client_session.__aenter__ = MagicMock(return_value=mock_client_session)
-        mock_client_session.__aexit__ = MagicMock(return_value=None)
-
-        monkeypatch.setattr('aiohttp.ClientSession', MagicMock(return_value=mock_client_session))
-
-        # Set string scope
-        oauth_client.configuration.scopes = "read write admin"
-
-        # Make async token request
-        await oauth_client.request_token_async()
-
-        # Verify scope parameter was passed unchanged
-        args, kwargs = mock_client_session.post.call_args
-        self.assertIn('data', kwargs)
-        self.assertIn('scope', kwargs['data'])
-        self.assertEqual(kwargs['data']['scope'], "read write admin")
+        credentials = Credentials(
+            method="client_credentials",
+            configuration=CredentialConfiguration(
+                client_id="myclientid",
+                client_secret="mysecret",
+                api_issuer="issuer.fga.example",
+                api_audience="myaudience",
+                scopes=["read", "write", "admin"],
+            ),
+        )
+        rest_client = rest.RESTClientObject(Configuration())
+        current_time = datetime.now()
+        client = OAuth2Client(credentials)
+        auth_header = await client.get_authentication_header(rest_client)
+        self.assertEqual(auth_header, {"Authorization": "Bearer AABBCCDD"})
+        self.assertEqual(client._access_token, "AABBCCDD")
+        self.assertGreaterEqual(
+            client._access_expiry_time, current_time + timedelta(seconds=120)
+        )
+        expected_header = urllib3.response.HTTPHeaderDict(
+            {
+                "Accept": "application/json",
+                "Content-Type": "application/x-www-form-urlencoded",
+                "User-Agent": "openfga-sdk (python) 0.9.5",
+            }
+        )
+        mock_request.assert_called_once_with(
+            method="POST",
+            url="https://issuer.fga.example/oauth/token",
+            headers=expected_header,
+            query_params=None,
+            body=None,
+            _preload_content=True,
+            _request_timeout=None,
+            post_params={
+                "client_id": "myclientid",
+                "client_secret": "mysecret",
+                "audience": "myaudience",
+                "grant_type": "client_credentials",
+                "scope": "read write admin",
+            },
+        )
+        await rest_client.close()
+
+    @patch.object(rest.RESTClientObject, "request")
+    async def test_get_authentication_obtain_client_credentials_with_scopes_string(self, mock_request):
+        """
+        Test getting authentication header when method is client credentials with scopes as string
+        """
+        response_body = """
+{
+  "expires_in": 120,
+  "access_token": "AABBCCDD"
+}
+        """
+        mock_request.return_value = mock_response(response_body, 200)
+
+        credentials = Credentials(
+            method="client_credentials",
+            configuration=CredentialConfiguration(
+                client_id="myclientid",
+                client_secret="mysecret",
+                api_issuer="issuer.fga.example",
+                api_audience="myaudience",
+                scopes="read write admin",
+            ),
+        )
+        rest_client = rest.RESTClientObject(Configuration())
+        current_time = datetime.now()
+        client = OAuth2Client(credentials)
+        auth_header = await client.get_authentication_header(rest_client)
+        self.assertEqual(auth_header, {"Authorization": "Bearer AABBCCDD"})
+        self.assertEqual(client._access_token, "AABBCCDD")
+        self.assertGreaterEqual(
+            client._access_expiry_time, current_time + timedelta(seconds=120)
+        )
+        expected_header = urllib3.response.HTTPHeaderDict(
+            {
+                "Accept": "application/json",
+                "Content-Type": "application/x-www-form-urlencoded",
+                "User-Agent": "openfga-sdk (python) 0.9.5",
+            }
+        )
+        mock_request.assert_called_once_with(
+            method="POST",
+            url="https://issuer.fga.example/oauth/token",
+            headers=expected_header,
+            query_params=None,
+            body=None,
+            _preload_content=True,
+            _request_timeout=None,
+            post_params={
+                "client_id": "myclientid",
+                "client_secret": "mysecret",
+                "audience": "myaudience",
+                "grant_type": "client_credentials",
+                "scope": "read write admin",
+            },
+        )
+        await rest_client.close()