# -*- coding: utf-8 -*- """task-2686 regression — session propagation classifier 4 enum 검증. 회장 verbatim 6번: mismatch 시 ``AUTHORITATIVE_BUT_SESSION_DISCONTINUITY`` 분류 강제. 본 테스트는 ``classify_session_propagation`` 의 4 enum 분기 모두를 커버한다: 1. ``AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED`` — ANU + 3 SID 일치. 2. ``AUTHORITATIVE_BUT_SESSION_DISCONTINUITY`` — ANU + collector/delivery SID 가 chair_facing 과 불일치. 3. ``NON_AUTHORITATIVE_SELF_COLLECTOR`` — observed owner == executor (self). 4. ``SESSION_PROPAGATION_GAP`` — ANU owner 이나 chair_facing_session_id 누락/invalid. 또한 ``SESSION_PROPAGATION_ENUM`` 이 정확히 4개 enum 값을 포함하는지 확인. """ from __future__ import annotations import importlib.util import sys from pathlib import Path _ROOT = Path(__file__).resolve().parents[3] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) def _load_real(modname: str, relpath: str): existing = sys.modules.get(modname) if existing is not None and getattr(existing, "__file__", "").endswith(relpath): return existing spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod _load_real("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py") _load_real("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py") from dispatch.normal_fallback_callback_helper import ( # noqa: E402 SESSION_PROPAGATION_DISCONTINUITY, SESSION_PROPAGATION_ENUM, SESSION_PROPAGATION_GAP, SESSION_PROPAGATION_OK, SESSION_PROPAGATION_SELF_KEY, SessionPropagationVerdict, classify_session_propagation, ) ANU_KEY = "c119085addb0f8b7" EXECUTOR_KEY = "9c4f7c5e9c4f7c5e" CHAIR_SID = "53e89540-5bed-4692-a726-ed857820758a" OTHER_SID = "12345678-1234-1234-1234-123456789abc" def test_session_propagation_enum_has_exactly_four_values(): assert SESSION_PROPAGATION_ENUM == frozenset({ SESSION_PROPAGATION_OK, SESSION_PROPAGATION_DISCONTINUITY, SESSION_PROPAGATION_SELF_KEY, SESSION_PROPAGATION_GAP, }) assert len(SESSION_PROPAGATION_ENUM) == 4 def test_classify_authoritative_processed_when_all_sid_match(): v = classify_session_propagation( chair_facing_session_id=CHAIR_SID, collector_session_id=CHAIR_SID, delivery_session_id=CHAIR_SID, observed_owner_key=ANU_KEY, executor_key=EXECUTOR_KEY, ) assert isinstance(v, SessionPropagationVerdict) assert v.classification == SESSION_PROPAGATION_OK assert v.observed_owner_is_anu is True def test_classify_authoritative_processed_when_partial_sid_present(): """Tier 1 doctrine 호환: collector/delivery 미명시여도 chair_facing 만 일치 PASS.""" v = classify_session_propagation( chair_facing_session_id=CHAIR_SID, collector_session_id=None, delivery_session_id=None, observed_owner_key=ANU_KEY, executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_OK def test_classify_discontinuity_when_collector_sid_diverges(): v = classify_session_propagation( chair_facing_session_id=CHAIR_SID, collector_session_id=OTHER_SID, # ★ mismatch delivery_session_id=CHAIR_SID, observed_owner_key=ANU_KEY, executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_DISCONTINUITY assert v.observed_owner_is_anu is True assert any("mismatch" in r.lower() for r in v.reasons) def test_classify_discontinuity_when_delivery_sid_diverges(): v = classify_session_propagation( chair_facing_session_id=CHAIR_SID, collector_session_id=CHAIR_SID, delivery_session_id=OTHER_SID, # ★ mismatch observed_owner_key=ANU_KEY, executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_DISCONTINUITY def test_classify_self_collector_when_owner_is_executor(): v = classify_session_propagation( chair_facing_session_id=CHAIR_SID, collector_session_id=CHAIR_SID, delivery_session_id=CHAIR_SID, observed_owner_key=EXECUTOR_KEY, # ★ self-key executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_SELF_KEY assert v.observed_owner_is_anu is False def test_classify_self_collector_when_owner_not_anu(): """ANU 아닌 임의 key 도 동일하게 SELF_COLLECTOR 로 분류 (task-2680 doctrine).""" v = classify_session_propagation( chair_facing_session_id=CHAIR_SID, collector_session_id=CHAIR_SID, delivery_session_id=CHAIR_SID, observed_owner_key="ffffffffffffffff", # not ANU, not executor executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_SELF_KEY def test_classify_propagation_gap_when_chair_sid_missing(): v = classify_session_propagation( chair_facing_session_id=None, collector_session_id=None, delivery_session_id=None, observed_owner_key=ANU_KEY, executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_GAP assert v.observed_owner_is_anu is True def test_classify_propagation_gap_when_chair_sid_invalid(): v = classify_session_propagation( chair_facing_session_id="not-a-uuid", collector_session_id=CHAIR_SID, delivery_session_id=CHAIR_SID, observed_owner_key=ANU_KEY, executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_GAP def test_classify_self_key_takes_priority_over_session_gap(): """우선순위 1 (self-key) > 3 (gap): SID 누락이어도 self-key 가 먼저 분류.""" v = classify_session_propagation( chair_facing_session_id=None, collector_session_id=None, delivery_session_id=None, observed_owner_key=EXECUTOR_KEY, executor_key=EXECUTOR_KEY, ) assert v.classification == SESSION_PROPAGATION_SELF_KEY def test_verdict_to_json_roundtrip(): v = classify_session_propagation( chair_facing_session_id=CHAIR_SID, collector_session_id=OTHER_SID, delivery_session_id=CHAIR_SID, observed_owner_key=ANU_KEY, executor_key=EXECUTOR_KEY, ) payload = v.to_json() assert payload["classification"] == SESSION_PROPAGATION_DISCONTINUITY assert payload["chair_facing_session_id"] == CHAIR_SID assert payload["collector_session_id"] == OTHER_SID assert payload["delivery_session_id"] == CHAIR_SID assert payload["observed_owner_is_anu"] is True assert isinstance(payload["reasons"], list)