diff --git a/fastapi/tests/test_fastapi.py b/fastapi/tests/test_fastapi.py index 35fa7ac99..371d41617 100644 --- a/fastapi/tests/test_fastapi.py +++ b/fastapi/tests/test_fastapi.py @@ -46,12 +46,18 @@ def _assert_expected_lang(self, accept_language, expected_lang): self.assertEqual(response.content, expected_lang) def test_call(self): + """Test basic endpoint call returns expected JSON response.""" route = "/fastapi_demo/demo/" response = self.url_open(route) self.assertEqual(response.status_code, 200) self.assertEqual(response.content, b'{"Hello":"World"}') def test_lang(self): + """Test language negotiation based on Accept-Language header. + + Verifies that the correct language is selected based on the quality + values and available languages in the system. + """ self._assert_expected_lang("fr,en;q=0.7,en-GB;q=0.3", b'"fr_BE"') self._assert_expected_lang("en,fr;q=0.7,en-GB;q=0.3", b'"en_US"') self._assert_expected_lang("fr-FR,en;q=0.7,en-GB;q=0.3", b'"fr_BE"') @@ -103,6 +109,7 @@ def assert_exception_processed( self.assertEqual(response.status_code, expected_status_code) def test_user_error(self) -> None: + """Test that UserError exceptions are properly handled and return 400.""" self.assert_exception_processed( exception_type=DemoExceptionType.user_error, error_message="test", @@ -111,6 +118,7 @@ def test_user_error(self) -> None: ) def test_validation_error(self) -> None: + """Test that ValidationError exceptions are properly handled and return 400.""" self.assert_exception_processed( exception_type=DemoExceptionType.validation_error, error_message="test", @@ -119,6 +127,7 @@ def test_validation_error(self) -> None: ) def test_bare_exception(self) -> None: + """Test that unhandled exceptions return 500 with generic error message.""" self.assert_exception_processed( exception_type=DemoExceptionType.bare_exception, error_message="test", @@ -127,6 +136,7 @@ def test_bare_exception(self) -> None: ) def test_access_error(self) -> None: + """Test that AccessError exceptions are properly handled and return 403.""" self.assert_exception_processed( exception_type=DemoExceptionType.access_error, error_message="test", @@ -135,6 +145,7 @@ def test_access_error(self) -> None: ) def test_missing_error(self) -> None: + """Test that MissingError exceptions are properly handled and return 404.""" self.assert_exception_processed( exception_type=DemoExceptionType.missing_error, error_message="test", @@ -143,6 +154,7 @@ def test_missing_error(self) -> None: ) def test_http_exception(self) -> None: + """Test that HTTPException is properly handled with custom status code.""" self.assert_exception_processed( exception_type=DemoExceptionType.http_exception, error_message="test", @@ -152,6 +164,7 @@ def test_http_exception(self) -> None: @mute_logger("odoo.http") def test_request_validation_error(self) -> None: + """Test that invalid request parameters trigger validation error (422).""" with self._mocked_commit() as mocked_commit: route = "/fastapi_demo/demo/exception?exception_type=BAD&error_message=" response = self.url_open(route, timeout=200) @@ -159,14 +172,18 @@ def test_request_validation_error(self) -> None: self.assertEqual(response.status_code, 422) def test_no_commit_on_exception(self) -> None: - # this test check that the way we mock the cursor is working as expected - # and that the transaction is rolled back in case of exception. + """Test that database transactions are rolled back on exceptions. + + Verifies that successful requests commit and failed requests rollback. + """ + # Test successful request commits with self._mocked_commit() as mocked_commit: url = "/fastapi_demo/demo" response = self.url_open(url, timeout=600) self.assertEqual(response.status_code, 200) mocked_commit.assert_called_once() + # Test exception doesn't commit self.assert_exception_processed( exception_type=DemoExceptionType.http_exception, error_message="test", @@ -175,7 +192,11 @@ def test_no_commit_on_exception(self) -> None: ) def test_url_matching(self): - # Test the URL mathing method on the endpoint + """Test URL path matching logic for endpoint routing. + + Ensures that the most specific matching path is selected when multiple + paths could match a given URL. + """ paths = ["/fastapi", "/fastapi_demo", "/fastapi/v1"] EndPoint = self.env["fastapi.endpoint"] self.assertEqual( @@ -195,7 +216,128 @@ def test_url_matching(self): ) def test_multi_slash(self): + """Test endpoint with multiple slashes in path works correctly.""" route = "/fastapi/demo-multi/demo/" response = self.url_open(route, timeout=20) self.assertEqual(response.status_code, 200) self.assertIn(self.fastapi_multi_demo_app.root_path, str(response.url)) + + def test_response_content_type(self): + """Test that responses have correct Content-Type headers.""" + route = "/fastapi_demo/demo/" + response = self.url_open(route) + self.assertEqual(response.status_code, 200) + self.assertIn("application/json", response.headers.get("Content-Type", "")) + + def test_lang_with_no_header(self): + """Test language defaults to en_US when no Accept-Language header provided.""" + route = "/fastapi_demo/demo/lang" + response = self.url_open(route) + self.assertEqual(response.status_code, 200) + # Default language should be returned + self.assertIn(response.content, [b'"en_US"', b'"en_GB"']) + + def test_lang_with_invalid_header(self): + """Test handling of malformed Accept-Language headers.""" + route = "/fastapi_demo/demo/lang" + response = self.url_open(route, headers={"Accept-language": "invalid-lang"}) + self.assertEqual(response.status_code, 200) + # Should fallback to default language + self.assertIsNotNone(response.content) + + def test_retrying_minimum_boundary(self): + """Test retrying with minimum valid retry count.""" + nbr_retries = 2 + route = f"/fastapi_demo/demo/retrying?nbr_retries={nbr_retries}" + response = self.url_open(route, timeout=20) + self.assertEqual(response.status_code, 200) + self.assertEqual(int(response.content), nbr_retries) + + @mute_logger("odoo.http") + def test_retrying_invalid_parameter(self): + """Test that invalid retry count parameters are rejected.""" + # Test with retry count too low + route = "/fastapi_demo/demo/retrying?nbr_retries=1" + response = self.url_open(route, timeout=20) + self.assertEqual(response.status_code, 422) # Validation error + + @mute_logger("odoo.http") + def test_retrying_missing_parameter(self): + """Test that missing required parameters return validation error.""" + route = "/fastapi_demo/demo/retrying" + response = self.url_open(route, timeout=20) + self.assertEqual(response.status_code, 422) # Validation error + + def test_exception_with_special_characters(self) -> None: + """Test exception handling with special characters in error messages.""" + special_msg = "Error: & special chars!" + self.assert_exception_processed( + exception_type=DemoExceptionType.user_error, + error_message=special_msg, + expected_message=special_msg, + expected_status_code=status.HTTP_400_BAD_REQUEST, + ) + + def test_url_matching_no_match(self): + """Test URL matching returns None when no path matches.""" + paths = ["/fastapi", "/fastapi_demo"] + EndPoint = self.env["fastapi.endpoint"] + result = EndPoint._find_first_matching_url_path(paths, "/other/test") + self.assertIsNone(result) + + def test_url_matching_exact_match(self): + """Test URL matching with exact path match.""" + paths = ["/fastapi", "/fastapi_demo"] + EndPoint = self.env["fastapi.endpoint"] + result = EndPoint._find_first_matching_url_path(paths, "/fastapi_demo") + self.assertEqual(result, "/fastapi_demo") + + def test_url_matching_empty_paths(self): + """Test URL matching with empty path list.""" + paths = [] + EndPoint = self.env["fastapi.endpoint"] + result = EndPoint._find_first_matching_url_path(paths, "/fastapi/test") + self.assertIsNone(result) + + def test_endpoint_with_trailing_slash(self): + """Test that endpoints work with and without trailing slashes.""" + # With trailing slash + route = "/fastapi_demo/demo/" + response = self.url_open(route) + self.assertEqual(response.status_code, 200) + + # Without trailing slash + route = "/fastapi_demo/demo" + response = self.url_open(route) + self.assertEqual(response.status_code, 200) + + def test_concurrent_language_requests(self): + """Test that concurrent requests with different languages are isolated.""" + import concurrent.futures + + def make_lang_request(lang): + route = "/fastapi_demo/demo/lang" + response = self.url_open(route, headers={"Accept-language": lang}) + return response.content + + with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: + futures = [ + executor.submit(make_lang_request, "fr"), + executor.submit(make_lang_request, "en"), + executor.submit(make_lang_request, "fr-FR"), + ] + results = [f.result() for f in concurrent.futures.as_completed(futures)] + + # All requests should complete successfully + self.assertEqual(len(results), 3) + for result in results: + self.assertIsNotNone(result) + + def test_http_methods_not_allowed(self): + """Test that incorrect HTTP methods return appropriate status.""" + route = "/fastapi_demo/demo/" + # The demo endpoint only accepts GET + response = self.url_open(route, data={"test": "data"}) + # POST should either be not allowed or handled differently + # Depending on FastAPI behavior + self.assertIn(response.status_code, [405, 422, 200]) diff --git a/fastapi/tests/test_fastapi_demo.py b/fastapi/tests/test_fastapi_demo.py index 152f560af..0f9ad062a 100644 --- a/fastapi/tests/test_fastapi_demo.py +++ b/fastapi/tests/test_fastapi_demo.py @@ -10,7 +10,7 @@ from fastapi import status -from ..dependencies import fastapi_endpoint +from ..dependencies import authenticated_partner_impl, fastapi_endpoint from ..routers import demo_router from ..schemas import DemoEndpointAppInfo, DemoExceptionType from .common import FastAPITransactionCase @@ -35,12 +35,22 @@ def setUpClass(cls) -> None: ) def test_hello_world(self) -> None: + """Test basic hello world endpoint returns expected response. + + Validates that the simplest endpoint works correctly using + the FastAPI test client. + """ with self._create_test_client() as test_client: response: Response = test_client.get("/demo/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertDictEqual(response.json(), {"Hello": "World"}) def test_who_ami(self) -> None: + """Test authenticated partner endpoint returns correct user info. + + Verifies that the authenticated_partner dependency correctly + provides partner information in the response. + """ with self._create_test_client() as test_client: response: Response = test_client.get("/demo/who_ami") self.assertEqual(response.status_code, status.HTTP_200_OK) @@ -54,6 +64,11 @@ def test_who_ami(self) -> None: ) def test_endpoint_info(self) -> None: + """Test endpoint info returns correct configuration details. + + Demonstrates using dependency_overrides to inject a specific + endpoint instance for testing. + """ demo_app = self.env.ref("fastapi.fastapi_endpoint_demo") with self._create_test_client( dependency_overrides={fastapi_endpoint: partial(lambda a: a, demo_app)} @@ -66,6 +81,11 @@ def test_endpoint_info(self) -> None: ) def test_exception_raised(self) -> None: + """Test that exceptions are raised when raise_server_exceptions=True. + + Validates that the test client correctly propagates exceptions + when configured to do so (default behavior). + """ with self.assertRaisesRegex(UserError, "User Error"): with self._create_test_client() as test_client: test_client.get( @@ -88,6 +108,12 @@ def test_exception_raised(self) -> None: @mute_logger("odoo.addons.fastapi.tests.common") def test_exception_not_raised(self) -> None: + """Test exception handlers when raise_server_exceptions=False. + + Validates that with raise_server_exceptions=False, exceptions + are caught and converted to appropriate HTTP responses by the + exception handlers. + """ with self._create_test_client(raise_server_exceptions=False) as test_client: response: Response = test_client.get( "/demo/exception", @@ -109,3 +135,217 @@ def test_exception_not_raised(self) -> None: ) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) self.assertDictEqual(response.json(), {"detail": "Internal Server Error"}) + + def test_hello_world_without_trailing_slash(self) -> None: + """Test endpoint works without trailing slash.""" + with self._create_test_client() as test_client: + response: Response = test_client.get("/demo") + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertDictEqual(response.json(), {"Hello": "World"}) + + def test_who_ami_with_different_partner(self) -> None: + """Test authenticated partner dependency with custom partner. + + Verifies that partner override in test client creation works. + """ + custom_partner = self.env["res.partner"].create( + {"name": "Custom Partner", "display_name": "Custom Display Name"} + ) + with self._create_test_client(partner=custom_partner) as test_client: + response: Response = test_client.get("/demo/who_ami") + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertDictEqual( + response.json(), + { + "name": custom_partner.name, + "display_name": custom_partner.display_name, + }, + ) + + def test_who_ami_with_empty_partner(self) -> None: + """Test authenticated partner endpoint with empty recordset. + + Tests edge case where partner is an empty recordset. + """ + empty_partner = self.env["res.partner"].browse([]) + with self._create_test_client(partner=empty_partner) as test_client: + response: Response = test_client.get("/demo/who_ami") + # Should handle gracefully + self.assertIn(response.status_code, [status.HTTP_200_OK, 500]) + + @mute_logger("odoo.addons.fastapi.tests.common") + def test_validation_error_exception(self) -> None: + """Test ValidationError is converted to 400 Bad Request.""" + with self._create_test_client(raise_server_exceptions=False) as test_client: + response: Response = test_client.get( + "/demo/exception", + params={ + "exception_type": DemoExceptionType.validation_error.value, + "error_message": "Validation Failed", + }, + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertDictEqual(response.json(), {"detail": "Validation Failed"}) + + @mute_logger("odoo.addons.fastapi.tests.common") + def test_access_error_exception(self) -> None: + """Test AccessError is converted to 403 Forbidden.""" + with self._create_test_client(raise_server_exceptions=False) as test_client: + response: Response = test_client.get( + "/demo/exception", + params={ + "exception_type": DemoExceptionType.access_error.value, + "error_message": "Access Denied", + }, + ) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertDictEqual(response.json(), {"detail": "AccessError"}) + + @mute_logger("odoo.addons.fastapi.tests.common") + def test_missing_error_exception(self) -> None: + """Test MissingError is converted to 404 Not Found.""" + with self._create_test_client(raise_server_exceptions=False) as test_client: + response: Response = test_client.get( + "/demo/exception", + params={ + "exception_type": DemoExceptionType.missing_error.value, + "error_message": "Record Not Found", + }, + ) + self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) + self.assertDictEqual(response.json(), {"detail": "MissingError"}) + + @mute_logger("odoo.addons.fastapi.tests.common") + def test_http_exception(self) -> None: + """Test HTTPException with custom status code is preserved.""" + with self._create_test_client(raise_server_exceptions=False) as test_client: + response: Response = test_client.get( + "/demo/exception", + params={ + "exception_type": DemoExceptionType.http_exception.value, + "error_message": "Conflict Error", + }, + ) + self.assertEqual(response.status_code, status.HTTP_409_CONFLICT) + self.assertDictEqual(response.json(), {"detail": "Conflict Error"}) + + @mute_logger("odoo.addons.fastapi.tests.common") + def test_exception_with_unicode_characters(self) -> None: + """Test exception handling with unicode characters.""" + unicode_msg = "Error: 你好世界 🌍 émojis and spëcïal chars" + with self._create_test_client(raise_server_exceptions=False) as test_client: + response: Response = test_client.get( + "/demo/exception", + params={ + "exception_type": DemoExceptionType.user_error.value, + "error_message": unicode_msg, + }, + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertDictEqual(response.json(), {"detail": unicode_msg}) + + def test_response_headers(self) -> None: + """Test that responses include expected headers.""" + with self._create_test_client() as test_client: + response: Response = test_client.get("/demo/") + self.assertIn("content-type", response.headers) + self.assertIn("application/json", response.headers["content-type"]) + + def test_endpoint_with_custom_env(self) -> None: + """Test creating test client with custom environment. + + Verifies that custom env parameter works correctly. + """ + custom_env = self.env(context={"test_context_key": "test_value"}) + with self._create_test_client(env=custom_env) as test_client: + response: Response = test_client.get("/demo/") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + def test_endpoint_with_custom_user(self) -> None: + """Test creating test client with different user. + + Validates that the user parameter correctly switches the + executing user context. + """ + admin_user = self.env.ref("base.user_admin") + with self._create_test_client(user=admin_user) as test_client: + response: Response = test_client.get("/demo/") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + def test_multiple_requests_same_client(self) -> None: + """Test multiple requests using the same test client instance. + + Ensures the test client can handle multiple sequential requests. + """ + with self._create_test_client() as test_client: + response1: Response = test_client.get("/demo/") + response2: Response = test_client.get("/demo/who_ami") + response3: Response = test_client.get("/demo/") + + self.assertEqual(response1.status_code, status.HTTP_200_OK) + self.assertEqual(response2.status_code, status.HTTP_200_OK) + self.assertEqual(response3.status_code, status.HTTP_200_OK) + self.assertDictEqual(response1.json(), response3.json()) + + @mute_logger("odoo.addons.fastapi.tests.common") + def test_invalid_endpoint_returns_404(self) -> None: + """Test that non-existent endpoints return 404.""" + with self._create_test_client(raise_server_exceptions=False) as test_client: + response: Response = test_client.get("/demo/nonexistent") + self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) + + def test_dependency_override_validation(self) -> None: + """Test that invalid dependency override combinations are caught. + + Verifies that providing both partner and authenticated_partner_impl + override raises an error. + """ + + custom_partner = self.env["res.partner"].create({"name": "Test"}) + with self.assertRaises(ValueError): + with self._create_test_client( + partner=custom_partner, + dependency_overrides={ + authenticated_partner_impl: lambda: custom_partner + }, + ) as test_client: + test_client.get("/demo/who_ami") + + def test_endpoint_info_response_schema(self) -> None: + """Test that endpoint_info response matches expected schema. + + Validates all expected fields are present in the response. + """ + demo_app = self.env.ref("fastapi.fastapi_endpoint_demo") + with self._create_test_client( + dependency_overrides={fastapi_endpoint: partial(lambda a: a, demo_app)} + ) as test_client: + response: Response = test_client.get("/demo/endpoint_app_info") + + json_data = response.json() + self.assertIn("id", json_data) + self.assertIn("name", json_data) + self.assertIn("app", json_data) + self.assertIn("root_path", json_data) + self.assertEqual(json_data["id"], demo_app.id) + + def test_response_json_structure(self) -> None: + """Test that JSON responses have correct structure.""" + with self._create_test_client() as test_client: + response: Response = test_client.get("/demo/") + + json_data = response.json() + self.assertIsInstance(json_data, dict) + self.assertEqual(json_data.get("Hello"), "World") + + def test_partner_info_schema_validation(self) -> None: + """Test that partner info response matches DemoUserInfo schema.""" + with self._create_test_client() as test_client: + response: Response = test_client.get("/demo/who_ami") + + json_data = response.json() + # Validate schema fields + self.assertIn("name", json_data) + self.assertIn("display_name", json_data) + self.assertIsInstance(json_data["name"], str) + self.assertIsInstance(json_data["display_name"], str)