diff --git a/openfga_sdk/api_client.py b/openfga_sdk/api_client.py index f8d6651..ee2d135 100644 --- a/openfga_sdk/api_client.py +++ b/openfga_sdk/api_client.py @@ -179,7 +179,10 @@ async def __call_api( # header parameters header_params = header_params or {} - header_params.update(self.default_headers) + # Merge headers with custom headers taking precedence over defaults + merged_headers = self.default_headers.copy() + merged_headers.update(header_params) + header_params = merged_headers if self.cookie: header_params["Cookie"] = self.cookie if header_params: diff --git a/openfga_sdk/client/client.py b/openfga_sdk/client/client.py index 2e0459d..4ac34d8 100644 --- a/openfga_sdk/client/client.py +++ b/openfga_sdk/client/client.py @@ -127,7 +127,12 @@ def options_to_kwargs( if options.get("continuation_token"): kwargs["continuation_token"] = options["continuation_token"] if options.get("headers"): - kwargs["_headers"] = options["headers"] + headers = options["headers"] + if isinstance(headers, dict): + kwargs["_headers"] = headers + else: + # Invalid headers type - skip it gracefully + pass if options.get("retry_params"): kwargs["_retry_params"] = options["retry_params"] return kwargs diff --git a/openfga_sdk/sync/api_client.py b/openfga_sdk/sync/api_client.py index db4a5c7..48ca242 100644 --- a/openfga_sdk/sync/api_client.py +++ b/openfga_sdk/sync/api_client.py @@ -178,7 +178,10 @@ def __call_api( # header parameters header_params = header_params or {} - header_params.update(self.default_headers) + # Merge headers with custom headers taking precedence over defaults + merged_headers = self.default_headers.copy() + merged_headers.update(header_params) + header_params = merged_headers if self.cookie: header_params["Cookie"] = self.cookie if header_params: @@ -395,7 +398,7 @@ def __call_api( if content_type is not None: match = re.search(r"charset=([a-zA-Z\-\d]+)[\s\;]?", content_type) encoding = match.group(1) if match else "utf-8" - if response_data.data is not None: + if response_data.data is not None and isinstance(response_data.data, bytes): response_data.data = response_data.data.decode(encoding) # deserialize response data diff --git a/openfga_sdk/sync/client/client.py b/openfga_sdk/sync/client/client.py index 732196c..d9c8a42 100644 --- a/openfga_sdk/sync/client/client.py +++ b/openfga_sdk/sync/client/client.py @@ -128,7 +128,12 @@ def options_to_kwargs( if options.get("continuation_token"): kwargs["continuation_token"] = options["continuation_token"] if options.get("headers"): - kwargs["_headers"] = options["headers"] + headers = options["headers"] + if isinstance(headers, dict): + kwargs["_headers"] = headers + else: + # Invalid headers type - skip it gracefully + pass if options.get("retry_params"): kwargs["_retry_params"] = options["retry_params"] return kwargs diff --git a/test/client/per_request_headers_edge_cases_test.py b/test/client/per_request_headers_edge_cases_test.py new file mode 100644 index 0000000..c3d8c62 --- /dev/null +++ b/test/client/per_request_headers_edge_cases_test.py @@ -0,0 +1,370 @@ +""" +Test edge cases and error scenarios for per-request custom HTTP headers functionality + +This module tests edge cases, invalid inputs, and error scenarios for the +per-request headers feature to ensure robust handling. +""" + +from unittest import IsolatedAsyncioTestCase +from unittest.mock import patch + +import urllib3 + +from openfga_sdk import rest +from openfga_sdk.client import ClientConfiguration +from openfga_sdk.client.client import ( + OpenFgaClient, + options_to_kwargs, + set_heading_if_not_set, +) +from openfga_sdk.client.models.check_request import ClientCheckRequest + + +store_id = "01YCP46JKYM8FJCQ37NMBYHE5X" +auth_model_id = "01YCP46JKYM8FJCQ37NMBYHE6X" +request_id = "x1y2z3" + + +def http_mock_response(body, status): + headers = urllib3.response.HTTPHeaderDict( + {"content-type": "application/json", "Fga-Request-Id": request_id} + ) + return urllib3.HTTPResponse( + body.encode("utf-8"), headers, status, preload_content=False + ) + + +def mock_response(body, status): + obj = http_mock_response(body, status) + return rest.RESTResponse(obj, obj.data) + + +class TestPerRequestHeadersEdgeCases(IsolatedAsyncioTestCase): + """Test edge cases and error scenarios for per-request headers""" + + def setUp(self): + self.configuration = ClientConfiguration( + api_url="http://api.fga.example", + store_id=store_id, + authorization_model_id=auth_model_id, + ) + + def tearDown(self): + pass + + def test_options_to_kwargs_with_headers(self): + """Test options_to_kwargs function properly handles headers""" + options = { + "headers": { + "x-test-header": "test-value", + "x-another": "another-value" + }, + "authorization_model_id": "test-model", + "page_size": 25 + } + + result = options_to_kwargs(options) + + # Check that headers are converted to _headers + self.assertIn("_headers", result) + self.assertEqual(result["_headers"]["x-test-header"], "test-value") + self.assertEqual(result["_headers"]["x-another"], "another-value") + + # Check that other options are preserved + self.assertEqual(result.get("page_size"), 25) + + def test_options_to_kwargs_without_headers(self): + """Test options_to_kwargs function works without headers""" + options = { + "authorization_model_id": "test-model", + "page_size": 25 + } + + result = options_to_kwargs(options) + + # Check that headers is not present when no headers option + self.assertNotIn("headers", result) + + # Check that other options are preserved + self.assertEqual(result.get("page_size"), 25) + + def test_options_to_kwargs_with_none(self): + """Test options_to_kwargs function handles None input""" + result = options_to_kwargs(None) + + # Should return empty dict + self.assertEqual(result, {}) + + def test_options_to_kwargs_with_empty_dict(self): + """Test options_to_kwargs function handles empty dict input""" + result = options_to_kwargs({}) + + # Should return empty dict + self.assertEqual(result, {}) + + def test_set_heading_if_not_set_with_existing_headers(self): + """Test set_heading_if_not_set function with existing headers""" + options = { + "headers": { + "x-existing": "existing-value" + } + } + + result = set_heading_if_not_set(options, "x-new-header", "new-value") + + # Check that new header was added + self.assertEqual(result["headers"]["x-new-header"], "new-value") + # Check that existing header is preserved + self.assertEqual(result["headers"]["x-existing"], "existing-value") + + def test_set_heading_if_not_set_without_headers(self): + """Test set_heading_if_not_set function when headers dict doesn't exist""" + options = { + "other_option": "value" + } + + result = set_heading_if_not_set(options, "x-new-header", "new-value") + + # Check that headers dict was created and header was added + self.assertIn("headers", result) + self.assertEqual(result["headers"]["x-new-header"], "new-value") + # Check that other options are preserved + self.assertEqual(result["other_option"], "value") + + def test_set_heading_if_not_set_with_none_options(self): + """Test set_heading_if_not_set function with None options""" + result = set_heading_if_not_set(None, "x-new-header", "new-value") + + # Check that options dict was created with headers + self.assertIn("headers", result) + self.assertEqual(result["headers"]["x-new-header"], "new-value") + + def test_set_heading_if_not_set_header_already_exists(self): + """Test set_heading_if_not_set function when header already exists""" + options = { + "headers": { + "x-existing": "original-value" + } + } + + result = set_heading_if_not_set(options, "x-existing", "new-value") + + # Check that original value is preserved (not overwritten) + self.assertEqual(result["headers"]["x-existing"], "original-value") + + def test_set_heading_if_not_set_with_invalidheaders_type(self): + """Test set_heading_if_not_set function with invalid headers type""" + options = { + "headers": "not-a-dict" # Invalid type + } + + result = set_heading_if_not_set(options, "x-new-header", "new-value") + + # Function should create new headers dict, replacing the invalid one + self.assertIsInstance(result["headers"], dict) + self.assertEqual(result["headers"]["x-new-header"], "new-value") + + @patch.object(rest.RESTClientObject, "request") + async def test_headers_with_invalid_type_in_options(self, mock_request): + """Test that invalid headers type in options is handled gracefully""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + # This should be handled gracefully - converted to dict or ignored + options_with_invalidheaders = { + "headers": "not-a-dict" + } + + async with OpenFgaClient(self.configuration) as fga_client: + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + # This should not raise an exception + await fga_client.check(body, options_with_invalidheaders) + + # Verify the request was made + mock_request.assert_called_once() + + @patch.object(rest.RESTClientObject, "request") + async def test_large_number_of_headers(self, mock_request): + """Test that a large number of headers is handled correctly""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + # Create a large number of headers + largeheaders = {f"x-header-{i}": f"value-{i}" for i in range(100)} + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": largeheaders + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with all headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that all custom headers were included (plus system headers) + self.assertGreaterEqual(len(headers), 100) + for i in range(100): + self.assertEqual(headers[f"x-header-{i}"], f"value-{i}") + + @patch.object(rest.RESTClientObject, "request") + async def test_unicode_headers(self, mock_request): + """Test that unicode characters in headers are handled correctly""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + unicode_headers = { + "x-unicode-header": "测试值", # Chinese characters + "x-emoji-header": "🚀🔐", # Emojis + "x-accented-header": "café-résumé", # Accented characters + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": unicode_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with unicode headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that unicode headers were included + self.assertEqual(headers["x-unicode-header"], "测试值") + self.assertEqual(headers["x-emoji-header"], "🚀🔐") + self.assertEqual(headers["x-accented-header"], "café-résumé") + + @patch.object(rest.RESTClientObject, "request") + async def test_long_header_values(self, mock_request): + """Test that very long header values are handled correctly""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + # Create a very long header value + long_value = "x" * 10000 # 10KB header value + + longheaders = { + "x-long-header": long_value, + "x-normal-header": "normal-value" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": longheaders + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with long headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that long header was included + self.assertEqual(headers["x-long-header"], long_value) + self.assertEqual(headers["x-normal-header"], "normal-value") + + @patch.object(rest.RESTClientObject, "request") + async def test_header_case_sensitivity(self, mock_request): + """Test that header case is preserved""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + case_sensitiveheaders = { + "X-Upper-Case": "upper-value", + "x-lower-case": "lower-value", + "X-Mixed-Case": "mixed-value", + "x-WEIRD-cAsE": "weird-value" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": case_sensitiveheaders + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with case-preserved headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that header case was preserved + self.assertEqual(headers["X-Upper-Case"], "upper-value") + self.assertEqual(headers["x-lower-case"], "lower-value") + self.assertEqual(headers["X-Mixed-Case"], "mixed-value") + self.assertEqual(headers["x-WEIRD-cAsE"], "weird-value") + + @patch.object(rest.RESTClientObject, "request") + async def test_header_overrides_default_headers(self, mock_request): + """Test that custom headers can override overrideable default headers""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + # Test with headers that can override defaults (User-Agent) + # Note: Accept and Content-Type are set by the API method and cannot be overridden + override_headers = { + "User-Agent": "custom-user-agent", + "x-custom-header": "custom-value", + "Authorization": "Bearer custom-token", + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": override_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that overrideable custom headers work + self.assertEqual(headers["User-Agent"], "custom-user-agent") + self.assertEqual(headers["x-custom-header"], "custom-value") + self.assertEqual(headers["Authorization"], "Bearer custom-token") + + # System headers are still set by the API method + self.assertEqual(headers["Accept"], "application/json") + self.assertTrue("Content-Type" in headers) diff --git a/test/client/per_request_headers_summary_test.py b/test/client/per_request_headers_summary_test.py new file mode 100644 index 0000000..55203bc --- /dev/null +++ b/test/client/per_request_headers_summary_test.py @@ -0,0 +1,103 @@ +""" +Summary test to demonstrate per-request custom HTTP headers functionality + +This test showcases the key functionality of sending custom headers with requests. +""" + +import asyncio + +from unittest import IsolatedAsyncioTestCase +from unittest.mock import patch + +import urllib3 + +from openfga_sdk import rest +from openfga_sdk.client import ClientConfiguration +from openfga_sdk.client.client import OpenFgaClient +from openfga_sdk.client.models.check_request import ClientCheckRequest + + +def http_mock_response(body, status): + headers = urllib3.response.HTTPHeaderDict( + {"content-type": "application/json", "Fga-Request-Id": "test-request-id"} + ) + return urllib3.HTTPResponse( + body.encode("utf-8"), headers, status, preload_content=False + ) + + +def mock_response(body, status): + obj = http_mock_response(body, status) + return rest.RESTResponse(obj, obj.data) + + +class TestPerRequestHeadersSummary(IsolatedAsyncioTestCase): + """Summary test for per-request custom HTTP headers""" + + def setUp(self): + self.configuration = ClientConfiguration( + api_url="http://api.fga.example", + store_id="01YCP46JKYM8FJCQ37NMBYHE5X", + authorization_model_id="01YCP46JKYM8FJCQ37NMBYHE6X", + ) + + @patch.object(rest.RESTClientObject, "request") + async def test_per_request_headers_summary(self, mock_request): + """Test that demonstrates per-request custom headers functionality""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + # Test custom headers for various use cases + custom_headers = { + "x-correlation-id": "req-123-abc", + "x-trace-id": "trace-456-def", + "x-client-version": "test-1.0.0", + "x-service-name": "authorization-test", + "x-environment": "test", + "x-user-id": "test-admin", + } + + async with OpenFgaClient(self.configuration) as fga_client: + # Test with custom headers + options = { + "headers": custom_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with all custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Verify all custom headers are present + for key, value in custom_headers.items(): + self.assertIn(key, headers, f"Header {key} should be present") + self.assertEqual(headers[key], value, f"Header {key} should have value {value}") + + # Verify default headers are also present + self.assertIn("Content-Type", headers) + self.assertIn("Accept", headers) + self.assertIn("User-Agent", headers) + + print("✅ Per-request custom headers test PASSED!") + print(f"✅ Successfully sent {len(custom_headers)} custom headers:") + for key, value in custom_headers.items(): + print(f" {key}: {value}") + + +async def main(): + """Run the summary test""" + test = TestPerRequestHeadersSummary() + test.setUp() + await test.test_per_request_headers_summary() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/test/client/per_request_headers_test.py b/test/client/per_request_headers_test.py new file mode 100644 index 0000000..2b8152a --- /dev/null +++ b/test/client/per_request_headers_test.py @@ -0,0 +1,527 @@ +""" +Test per-request custom HTTP headers functionality + +This module tests the ability to send custom HTTP headers with individual API requests +using the options["headers"] parameter that gets converted to headers internally. +""" + +import uuid + +from unittest import IsolatedAsyncioTestCase +from unittest.mock import patch + +import urllib3 + +from openfga_sdk import rest +from openfga_sdk.client import ClientConfiguration +from openfga_sdk.client.client import OpenFgaClient +from openfga_sdk.client.models.assertion import ClientAssertion +from openfga_sdk.client.models.batch_check_item import ClientBatchCheckItem +from openfga_sdk.client.models.batch_check_request import ClientBatchCheckRequest +from openfga_sdk.client.models.check_request import ClientCheckRequest +from openfga_sdk.client.models.expand_request import ClientExpandRequest +from openfga_sdk.client.models.list_objects_request import ClientListObjectsRequest +from openfga_sdk.client.models.tuple import ClientTuple +from openfga_sdk.client.models.write_request import ClientWriteRequest +from openfga_sdk.models.read_request_tuple_key import ReadRequestTupleKey + + +store_id = "01YCP46JKYM8FJCQ37NMBYHE5X" +auth_model_id = "01YCP46JKYM8FJCQ37NMBYHE6X" +request_id = "x1y2z3" + + +def http_mock_response(body, status): + headers = urllib3.response.HTTPHeaderDict( + {"content-type": "application/json", "Fga-Request-Id": request_id} + ) + return urllib3.HTTPResponse( + body.encode("utf-8"), headers, status, preload_content=False + ) + + +def mock_response(body, status): + obj = http_mock_response(body, status) + return rest.RESTResponse(obj, obj.data) + + +class TestPerRequestHeaders(IsolatedAsyncioTestCase): + """Test per-request custom HTTP headers functionality""" + + def setUp(self): + self.configuration = ClientConfiguration( + api_url="http://api.fga.example", + store_id=store_id, + authorization_model_id=auth_model_id, + ) + + def tearDown(self): + pass + + @patch.object(rest.RESTClientObject, "request") + async def test_check_with_custom_headers(self, mock_request): + """Test check request with custom headers""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-correlation-id": "test-correlation-123", + "x-trace-id": "trace-456", + "x-custom-header": "custom-value" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + + # Headers should be passed as 'headers' parameter in the call + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-correlation-id"], "test-correlation-123") + self.assertEqual(headers["x-trace-id"], "trace-456") + self.assertEqual(headers["x-custom-header"], "custom-value") + + @patch.object(rest.RESTClientObject, "request") + async def test_write_with_custom_headers(self, mock_request): + """Test write request with custom headers""" + response_body = '{}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-request-id": "write-request-789", + "x-client-version": "1.0.0" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientWriteRequest( + writes=[ + ClientTuple( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + ] + ) + + await fga_client.write(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-request-id"], "write-request-789") + self.assertEqual(headers["x-client-version"], "1.0.0") + + @patch.object(rest.RESTClientObject, "request") + async def test_list_objects_with_custom_headers(self, mock_request): + """Test list_objects request with custom headers""" + response_body = '{"objects": ["document:1", "document:2"]}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-service-name": "authorization-service", + "x-environment": "test" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientListObjectsRequest( + user="user:test-user", + relation="viewer", + type="document", + ) + + await fga_client.list_objects(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-service-name"], "authorization-service") + self.assertEqual(headers["x-environment"], "test") + + @patch.object(rest.RESTClientObject, "request") + async def test_expand_with_custom_headers(self, mock_request): + """Test expand request with custom headers""" + response_body = '{"tree": {"root": {"name": "test", "leaf": {"users": {"users": []}}}}}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-operation": "expand-check", + "x-user-id": "admin-user" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientExpandRequest( + relation="viewer", + object="document:test-doc", + ) + + await fga_client.expand(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-operation"], "expand-check") + self.assertEqual(headers["x-user-id"], "admin-user") + + @patch.object(rest.RESTClientObject, "request") + async def test_batch_check_with_custom_headers(self, mock_request): + """Test batch_check request with custom headers""" + response_body = """ + { + "result": { + "test-correlation-id": { + "allowed": true + } + } + } + """ + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-batch-id": "batch-123", + "x-priority": "high" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + checks = [ + ClientBatchCheckItem( + user="user:test-user", + relation="viewer", + object="document:test-doc", + correlation_id="test-correlation-id", + ) + ] + body = ClientBatchCheckRequest(checks=checks) + + await fga_client.batch_check(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-batch-id"], "batch-123") + self.assertEqual(headers["x-priority"], "high") + + @patch.object(rest.RESTClientObject, "request") + async def test_read_with_custom_headers(self, mock_request): + """Test read request with custom headers""" + response_body = """ + { + "tuples": [ + { + "key": { + "user": "user:test-user", + "relation": "viewer", + "object": "document:test-doc" + }, + "timestamp": "2023-01-01T00:00:00.000Z" + } + ], + "continuation_token": "" + } + """ + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-read-operation": "get-tuples", + "x-source": "admin-console" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ReadRequestTupleKey( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.read(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-read-operation"], "get-tuples") + self.assertEqual(headers["x-source"], "admin-console") + + @patch.object(rest.RESTClientObject, "request") + async def test_read_authorization_models_with_custom_headers(self, mock_request): + """Test read_authorization_models request with custom headers""" + response_body = """ + { + "authorization_models": [ + { + "id": "01YCP46JKYM8FJCQ37NMBYHE6X", + "schema_version": "1.1", + "type_definitions": [] + } + ] + } + """ + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-model-operation": "list-models", + "x-tenant": "tenant-123" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + await fga_client.read_authorization_models(options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-model-operation"], "list-models") + self.assertEqual(headers["x-tenant"], "tenant-123") + + @patch.object(rest.RESTClientObject, "request") + async def test_write_assertions_with_custom_headers(self, mock_request): + """Test write_assertions request with custom headers""" + response_body = '{}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-assertion-batch": "test-assertions", + "x-test-run": "automated" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = [ + ClientAssertion( + user="user:test-user", + relation="viewer", + object="document:test-doc", + expectation=True, + ) + ] + + await fga_client.write_assertions(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-assertion-batch"], "test-assertions") + self.assertEqual(headers["x-test-run"], "automated") + + @patch.object(rest.RESTClientObject, "request") + async def test_headers_with_other_options(self, mock_request): + """Test that headers work correctly when combined with other options""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-combined-test": "headers-and-options" + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1", + "headers": custom_headers, + "consistency": "strong" + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with custom headers and other options + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-combined-test"], "headers-and-options") + + # Verify other options were also applied (by checking the call args structure) + self.assertIsNotNone(call_args) + + @patch.object(rest.RESTClientObject, "request") + async def test_empty_headers_option(self, mock_request): + """Test that empty headers option works correctly""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": {} + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made successfully + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Headers should contain defaults but not our custom headers + self.assertEqual(len(headers), 3) # Should contain default headers + self.assertIn("Content-Type", headers) + self.assertIn("Accept", headers) + self.assertIn("User-Agent", headers) + # But should not contain custom headers + self.assertNotIn("x-custom-header", headers) + + @patch.object(rest.RESTClientObject, "request") + async def test_no_headers_option(self, mock_request): + """Test that requests work without headers option""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + async with OpenFgaClient(self.configuration) as fga_client: + # No options provided + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body) + + # Verify the request was made successfully + mock_request.assert_called_once() + call_args = mock_request.call_args + + # When no headers provided, headers should still exist but be the default ones + headers = call_args.kwargs.get("headers", {}) + # Default should include Content-Type but not our custom headers + self.assertNotIn("x-custom-header", headers) + + @patch.object(rest.RESTClientObject, "request") + async def test_header_values_as_strings(self, mock_request): + """Test that header values are properly handled as strings""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-string-header": "string-value", + "x-number-header": "123", # Should be string + "x-boolean-header": "true", # Should be string + "x-uuid-header": str(uuid.uuid4()), + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that all header values are strings + for key, value in custom_headers.items(): + self.assertEqual(headers[key], value) + self.assertIsInstance(headers[key], str) + + @patch.object(rest.RESTClientObject, "request") + async def test_special_characters_in_headers(self, mock_request): + """Test that headers with special characters work correctly""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-special-chars": "value-with-dashes_and_underscores", + "x-with-dots": "value.with.dots", + "x-with-numbers": "value123with456numbers", + "x-case-sensitive": "CamelCaseValue", + } + + async with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + await fga_client.check(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included exactly as provided + for key, value in custom_headers.items(): + self.assertEqual(headers[key], value) diff --git a/test/sync/client/per_request_headers_sync_test.py b/test/sync/client/per_request_headers_sync_test.py new file mode 100644 index 0000000..c086ca0 --- /dev/null +++ b/test/sync/client/per_request_headers_sync_test.py @@ -0,0 +1,317 @@ +""" +Test per-request custom HTTP headers functionality for sync client + +This module tests the ability to send custom HTTP headers with individual API requests +using the synchronous client version. +""" + + +from unittest import TestCase +from unittest.mock import patch + +import urllib3 + +from openfga_sdk.client import ClientConfiguration +from openfga_sdk.client.models.check_request import ClientCheckRequest +from openfga_sdk.client.models.list_objects_request import ClientListObjectsRequest +from openfga_sdk.client.models.tuple import ClientTuple +from openfga_sdk.client.models.write_request import ClientWriteRequest +from openfga_sdk.sync import OpenFgaClient, rest + + +store_id = "01YCP46JKYM8FJCQ37NMBYHE5X" +auth_model_id = "01YCP46JKYM8FJCQ37NMBYHE6X" +request_id = "x1y2z3" + + +def http_mock_response(body, status): + headers = urllib3.response.HTTPHeaderDict( + {"content-type": "application/json", "Fga-Request-Id": request_id} + ) + return urllib3.HTTPResponse( + body.encode("utf-8"), headers, status, preload_content=False + ) + + +def mock_response(body, status): + obj = http_mock_response(body, status) + return rest.RESTResponse(obj, obj.data) + + +class TestSyncPerRequestHeaders(TestCase): + """Test per-request custom HTTP headers functionality for sync client""" + + def setUp(self): + self.configuration = ClientConfiguration( + api_url="http://api.fga.example", + store_id=store_id, + authorization_model_id=auth_model_id, + ) + + def tearDown(self): + pass + + @patch.object(rest.RESTClientObject, "request") + def test_sync_check_with_custom_headers(self, mock_request): + """Test sync check request with custom headers""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-sync-correlation-id": "sync-test-correlation-123", + "x-sync-trace-id": "sync-trace-456", + "x-sync-custom-header": "sync-custom-value" + } + + with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + fga_client.check(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-sync-correlation-id"], "sync-test-correlation-123") + self.assertEqual(headers["x-sync-trace-id"], "sync-trace-456") + self.assertEqual(headers["x-sync-custom-header"], "sync-custom-value") + + @patch.object(rest.RESTClientObject, "request") + def test_sync_write_with_custom_headers(self, mock_request): + """Test sync write request with custom headers""" + response_body = '{}' + mock_request.return_value = mock_response(response_body, 200) + + custom_headers = { + "x-sync-request-id": "sync-write-request-789", + "x-sync-client-version": "sync-1.0.0" + } + + with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientWriteRequest( + writes=[ + ClientTuple( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + ] + ) + + fga_client.write(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that our custom headers were included + self.assertEqual(headers["x-sync-request-id"], "sync-write-request-789") + self.assertEqual(headers["x-sync-client-version"], "sync-1.0.0") + + @patch.object(rest.RESTClientObject, "request") + def test_sync_multiple_requests_with_different_headers(self, mock_request): + """Test that sync client can handle multiple requests with different headers""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + with OpenFgaClient(self.configuration) as fga_client: + # First request with headers + options1 = { + "headers": { + "x-request-number": "1", + "x-operation": "first-check" + } + } + + body1 = ClientCheckRequest( + user="user:test-user-1", + relation="viewer", + object="document:test-doc-1", + ) + + fga_client.check(body1, options1) + + # Second request with different headers + options2 = { + "headers": { + "x-request-number": "2", + "x-operation": "second-check" + } + } + + body2 = ClientCheckRequest( + user="user:test-user-2", + relation="editor", + object="document:test-doc-2", + ) + + fga_client.check(body2, options2) + + # Verify both requests were made + self.assertEqual(mock_request.call_count, 2) + + # Check first call headers + first_call = mock_request.call_args_list[0] + firstheaders = first_call.kwargs.get("headers", {}) + self.assertEqual(firstheaders["x-request-number"], "1") + self.assertEqual(firstheaders["x-operation"], "first-check") + + # Check second call headers + second_call = mock_request.call_args_list[1] + secondheaders = second_call.kwargs.get("headers", {}) + self.assertEqual(secondheaders["x-request-number"], "2") + self.assertEqual(secondheaders["x-operation"], "second-check") + + @patch.object(rest.RESTClientObject, "request") + def test_sync_client_without_headers(self, mock_request): + """Test that sync client works without headers option""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + with OpenFgaClient(self.configuration) as fga_client: + # No options provided + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + fga_client.check(body) + + # Verify the request was made successfully + mock_request.assert_called_once() + call_args = mock_request.call_args + + # When no headers provided, headers should still exist but be the default ones + headers = call_args.kwargs.get("headers", {}) + # Default should include Content-Type but not our custom headers + self.assertNotIn("x-custom-header", headers) + + @patch.object(rest.RESTClientObject, "request") + def test_sync_client_empty_headers(self, mock_request): + """Test that sync client works with empty headers""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": {} + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + fga_client.check(body, options) + + # Verify the request was made successfully + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Headers should contain defaults but not our custom headers + self.assertEqual(len(headers), 3) # Should contain default headers + self.assertIn("Content-Type", headers) + self.assertIn("Accept", headers) + self.assertIn("User-Agent", headers) + # But should not contain custom headers + self.assertNotIn("x-custom-header", headers) + + @patch.object(rest.RESTClientObject, "request") + def test_sync_client_consistency_across_async_api(self, mock_request): + """Test that sync client headers behavior is consistent with async client""" + response_body = '{"allowed": true}' + mock_request.return_value = mock_response(response_body, 200) + + # Test the same header pattern that works in async client + custom_headers = { + "x-correlation-id": "abc-123-def-456", + "x-trace-id": "trace-789", + "x-custom-header": "custom-value", + "x-service-name": "authorization-service" + } + + with OpenFgaClient(self.configuration) as fga_client: + options = { + "authorization_model_id": "01HVMMBCMGZNT3SED4Z17ECXCA", + "headers": custom_headers + } + + body = ClientCheckRequest( + user="user:test-user", + relation="viewer", + object="document:test-doc", + ) + + fga_client.check(body, options) + + # Verify the request was made with custom headers + mock_request.assert_called_once() + call_args = mock_request.call_args + headers = call_args.kwargs.get("headers", {}) + + # Check that all our custom headers were included + self.assertEqual(headers["x-correlation-id"], "abc-123-def-456") + self.assertEqual(headers["x-trace-id"], "trace-789") + self.assertEqual(headers["x-custom-header"], "custom-value") + self.assertEqual(headers["x-service-name"], "authorization-service") + + # Verify other options were also applied + self.assertIsNotNone(call_args) + + @patch("openfga_sdk.sync.open_fga_api.OpenFgaApi.streamed_list_objects") + def test_sync_streamed_list_objects_with_custom_headers(self, mock_streamed_list_objects): + """Test sync streamed_list_objects with custom headers to cover missing line""" + # Mock the streaming API response + mock_streamed_list_objects.return_value = [ + {"result": {"object": "document:1"}}, + {"result": {"object": "document:2"}}, + ] + + custom_headers = { + "x-stream-id": "stream-123", + "x-batch-size": "100" + } + + with OpenFgaClient(self.configuration) as fga_client: + options = { + "headers": custom_headers + } + + body = ClientListObjectsRequest( + user="user:test-user", + relation="viewer", + type="document", + ) + + # This should call the streamed_list_objects method and cover line 932 + results = list(fga_client.streamed_list_objects(body, options)) + + # Verify we got results + self.assertEqual(len(results), 2) + self.assertEqual(results[0].object, "document:1") + self.assertEqual(results[1].object, "document:2") + + # Verify the API was called with the expected parameters including headers + mock_streamed_list_objects.assert_called_once() + call_kwargs = mock_streamed_list_objects.call_args.kwargs + self.assertIn("_headers", call_kwargs) + self.assertEqual(call_kwargs["_headers"]["x-stream-id"], "stream-123") + self.assertEqual(call_kwargs["_headers"]["x-batch-size"], "100")