import pytest from pytest_mock import MockerFixture from starlette.applications import Starlette from starlette.middleware import Middleware from starlette.requests import Request from starlette.responses import PlainTextResponse from starlette.routing import Route from starlette.testclient import TestClient from fansifter_common import context from fansifter_common.context.asgi.middleware import ( CorrelationIdMiddleware, RequestContextMiddleware, ) @pytest.fixture(scope="session") def app() -> Starlette: def endpoint(request: Request) -> PlainTextResponse: return PlainTextResponse("ok") app = Starlette( middleware=[ Middleware(CorrelationIdMiddleware), Middleware(RequestContextMiddleware), ], routes=[Route("/test", endpoint)], ) return app @pytest.fixture(scope="session") def client(app: Starlette) -> TestClient: return TestClient(app) def test_set_correlation_default(mocker: MockerFixture, client: TestClient) -> None: correlation_id = "test" mocker.patch( "fansifter_common.context.asgi.middleware.get_correlation_id", return_value=correlation_id, ) response = client.get("/test") assert response.headers["Correlation-Id"] == correlation_id def test_set_correlation_in_header(client: TestClient) -> None: correlation_id = "test" response = client.get("/test", headers={"Correlation-Id": correlation_id}) assert response.headers["Correlation-Id"] == correlation_id def test_set_request_context(mocker: MockerFixture, client: TestClient) -> None: profile_type = "AudienceProfile" profile_id = "1000" set_request_context_mock = mocker.patch( "fansifter_common.context.asgi.middleware.set_request_context" ) reset_request_context_mock = mocker.patch( "fansifter_common.context.asgi.middleware.reset_request_context" ) client.get( "/test", headers={ "Orchard-Profile-Type": profile_type, "Orchard-Profile-Id": profile_id, }, ) set_request_context_mock.assert_called_once_with( context.RequestContext( profile_type=profile_type, profile_id=int(profile_id), ) ) reset_request_context_mock.assert_called_once()