from unittest.mock import AsyncMock import pytest from fastapi import Request from monday_com_orca_backend.api.routers.decorators import cache_response, request_cache class TestCacheResponse: """Test suite for cache_response decorator.""" class Constants: DEFAULT_PATH = "/test" GET_METHOD = "GET" POST_METHOD = "POST" # Query strings QUERY_FOO_BAR = b"foo=bar" QUERY_ID_1 = b"id=1" QUERY_FOO_BAR_ID_1 = b"foo=bar&id=1" QUERY_ID_1_FOO_BAR = b"id=1&foo=bar" # JSON bodies JSON_ORDERED_1 = b'{"name":"test","id":123,"active":true}' JSON_ORDERED_2 = b'{"id":123,"active":true,"name":"test"}' JSON_SIMPLE = b'{"data":123}' # Non-JSON body RAW_BINARY_DATA = b"some raw binary data" # Expected responses FRESH_RESULT = b'{"result":"fresh"}' FRESH_POST_RESULT = b'{"result":"fresh_post"}' SHOULD_NOT_BE_CALLED = {"result": "should not be called"} @pytest.fixture(autouse=True) def clear_request_cache(self): """Clear cache before each test.""" request_cache.clear() yield @staticmethod def create_request( method=Constants.GET_METHOD, path=Constants.DEFAULT_PATH, query_string=b"", body_mock=None, ): """Helper to create test requests.""" scope = { "type": "http", "method": method, "path": path, "query_string": query_string, "headers": [], } request = Request(scope) if body_mock: request.body = body_mock return request def create_counter_view_func(self, prefix="fresh"): """Create a view function with call counter.""" call_count = {"value": 0} async def view_func(request: Request): call_count["value"] += 1 return {"result": f"{prefix}_{call_count['value']}"} return view_func, call_count def assert_single_cache_entry_used(self, call_count, expected_count=1): """Assert that function was called expected number of times and cache has one entry. """ assert ( call_count["value"] == expected_count ), f"Function should be called {expected_count} time(s)" assert len(request_cache) == 1, "Should have exactly one cache entry" class TestBasicCaching: """Test basic caching functionality.""" @pytest.mark.asyncio async def test_get_request_caching(self, outer_self=None): """Test basic GET request caching.""" if outer_self is None: outer_self = TestCacheResponse() async def view_func(request: Request): return {"result": "fresh"} request = outer_self.create_request( query_string=outer_self.Constants.QUERY_FOO_BAR ) decorated = cache_response(view_func) # First call response1 = await decorated(request=request) assert response1.body == outer_self.Constants.FRESH_RESULT # Second call with different function should be cached async def view_func2(request: Request): return outer_self.Constants.SHOULD_NOT_BE_CALLED decorated2 = cache_response(view_func2) response2 = await decorated2(request=request) assert response2.body == outer_self.Constants.FRESH_RESULT @pytest.mark.asyncio async def test_post_request_caching(self, outer_self=None): """Test basic POST request caching.""" if outer_self is None: outer_self = TestCacheResponse() async def view_func(request: Request): return {"result": "fresh_post"} mock_body = AsyncMock(return_value=outer_self.Constants.JSON_SIMPLE) request = outer_self.create_request( method=outer_self.Constants.POST_METHOD, query_string=outer_self.Constants.QUERY_ID_1, body_mock=mock_body, ) decorated = cache_response(view_func) # First call response1 = await decorated(request=request) assert response1.body == outer_self.Constants.FRESH_POST_RESULT # Second call with different function should be cached async def view_func2(request: Request): return outer_self.Constants.SHOULD_NOT_BE_CALLED decorated2 = cache_response(view_func2) response2 = await decorated2(request=request) assert response2.body == outer_self.Constants.FRESH_POST_RESULT class TestOrderIndependence: """Test that order of parameters doesn't affect caching.""" @pytest.mark.asyncio async def test_query_params_order_independence(self, outer_self=None): """Test that requests with same query params in different order use same cache entry. """ if outer_self is None: outer_self = TestCacheResponse() view_func, call_count = outer_self.create_counter_view_func() decorated = cache_response(view_func) # Same query params in different order request1 = outer_self.create_request( query_string=outer_self.Constants.QUERY_FOO_BAR_ID_1 ) request2 = outer_self.create_request( query_string=outer_self.Constants.QUERY_ID_1_FOO_BAR ) # First call executes function response1 = await decorated(request=request1) assert response1.body == b'{"result":"fresh_1"}' assert call_count["value"] == 1 # Second call uses cache response2 = await decorated(request=request2) assert response2.body == b'{"result":"fresh_1"}' outer_self.assert_single_cache_entry_used(call_count) @pytest.mark.asyncio async def test_post_body_order_independence(self, outer_self=None): """Test that POST requests with same JSON data in different key order use same cache entry. """ if outer_self is None: outer_self = TestCacheResponse() view_func, call_count = outer_self.create_counter_view_func("fresh_post") decorated = cache_response(view_func) # Same JSON data in different key order body1 = AsyncMock(return_value=outer_self.Constants.JSON_ORDERED_1) body2 = AsyncMock(return_value=outer_self.Constants.JSON_ORDERED_2) request1 = outer_self.create_request( method=outer_self.Constants.POST_METHOD, body_mock=body1 ) request2 = outer_self.create_request( method=outer_self.Constants.POST_METHOD, body_mock=body2 ) # First call executes function response1 = await decorated(request=request1) assert response1.body == b'{"result":"fresh_post_1"}' assert call_count["value"] == 1 # Second call with different JSON key order should use cache response2 = await decorated(request=request2) assert response2.body == b'{"result":"fresh_post_1"}' outer_self.assert_single_cache_entry_used(call_count) class TestNonJsonBodies: """Test handling of non-JSON request bodies.""" @pytest.mark.asyncio async def test_non_json_body_caching(self, outer_self=None): """Test that non-JSON POST bodies still work with raw body hashing.""" if outer_self is None: outer_self = TestCacheResponse() view_func, call_count = outer_self.create_counter_view_func("fresh_raw") decorated = cache_response(view_func) # Non-JSON body content raw_body = AsyncMock(return_value=outer_self.Constants.RAW_BINARY_DATA) request1 = outer_self.create_request( method=outer_self.Constants.POST_METHOD, body_mock=raw_body ) request2 = outer_self.create_request( method=outer_self.Constants.POST_METHOD, body_mock=raw_body ) # First call executes function response1 = await decorated(request=request1) assert response1.body == b'{"result":"fresh_raw_1"}' assert call_count["value"] == 1 # Second call with same raw body should use cache response2 = await decorated(request=request2) assert response2.body == b'{"result":"fresh_raw_1"}' outer_self.assert_single_cache_entry_used(call_count)