"""Tests for request context.""" import uuid from collections.abc import Mapping import pytest from owscontext.context import RequestContext, request_context_from_headers from owscontext.context.base import ( get_correlation_id, get_request_context, reset_correlation_id, reset_request_context, set_correlation_id, set_request_context, ) @pytest.mark.parametrize( "headers, expected", [ pytest.param( { "Orchard-Profile-Type": "AudienceProfile", "Orchard-Profile-Id": "1000", "Orchard-Profile-UUID": "d130f3ef-ae51-4d1d-8117-5485d81681c1", }, RequestContext( profile_type="AudienceProfile", profile_id=1000, profile_uuid=uuid.UUID("d130f3ef-ae51-4d1d-8117-5485d81681c1"), ), id="profile-context", ), pytest.param( { "Orchard-Identity-Id": "999", }, RequestContext( identity_id="999", ), id="context-profile", ), ], ) def test_request_context_from_headers( headers: Mapping[str, str], expected: RequestContext ) -> None: """Test that request context is correctly parsed from headers.""" request_context = request_context_from_headers(headers) assert request_context == expected def test_get_correlation_id_default_is_none() -> None: """get_correlation_id returns None when not set.""" assert get_correlation_id() is None def test_set_and_get_correlation_id() -> None: """set_correlation_id stores the value retrievable by get_correlation_id.""" token = set_correlation_id("test-id-123") try: assert get_correlation_id() == "test-id-123" finally: reset_correlation_id(token) def test_reset_correlation_id_restores_previous_value() -> None: """reset_correlation_id restores the previous value.""" token = set_correlation_id("first") try: inner_token = set_correlation_id("second") assert get_correlation_id() == "second" reset_correlation_id(inner_token) assert get_correlation_id() == "first" finally: reset_correlation_id(token) def test_set_correlation_id_none() -> None: """set_correlation_id accepts None.""" token = set_correlation_id(None) try: assert get_correlation_id() is None finally: reset_correlation_id(token) def test_get_request_context_default_is_none() -> None: """get_request_context returns None when not set.""" assert get_request_context() is None def test_set_and_get_request_context() -> None: """set_request_context stores the value retrievable by get_request_context.""" context = RequestContext(profile_type="AudienceProfile", profile_id=1) token = set_request_context(context) try: assert get_request_context() == context finally: reset_request_context(token) def test_reset_request_context_restores_previous_value() -> None: """reset_request_context restores the previous value.""" context = RequestContext(profile_type="AudienceProfile") token = set_request_context(context) try: assert get_request_context() == context reset_request_context(token) assert get_request_context() is None finally: pass def test_request_context_dict() -> None: """RequestContext.dict() returns all fields.""" context = RequestContext(profile_type="AudienceProfile", profile_id=1) result = context.dict() assert result["profile_type"] == "AudienceProfile" assert result["profile_id"] == 1 def test_request_context_dict_exclude_empty() -> None: """RequestContext.dict(exclude_empty=True) omits None fields.""" context = RequestContext(profile_type="AudienceProfile") result = context.dict(exclude_empty=True) assert "profile_type" in result assert "profile_id" not in result def test_request_context_dict_exclude_keys() -> None: """RequestContext.dict(exclude_keys=...) omits specified fields.""" context = RequestContext(profile_type="AudienceProfile", profile_id=1) result = context.dict(exclude_keys=["profile_id"]) assert "profile_type" in result assert "profile_id" not in result