vignette/scripts/test_smoke_calibration_transfer_api.py

from __future__ import annotations

import importlib.util
import sys
import unittest
from pathlib import Path
from unittest.mock import patch


SCRIPT_PATH = Path(__file__).with_name("smoke-calibration-transfer-api.py")
SPEC = importlib.util.spec_from_file_location(
    "smoke_calibration_transfer_api", SCRIPT_PATH
)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = MODULE
SPEC.loader.exec_module(MODULE)


class FakeClient:
    def __init__(self, bodies: list[object]) -> None:
        self.bodies = list(bodies)
        self.paths: list[str] = []

    def request(self, method: str, path: str, *args: object, **kwargs: object):
        self.paths.append(f"{method} {path}")
        if not self.bodies:
            raise AssertionError("unexpected request")
        return MODULE.ApiResponse(200, self.bodies.pop(0))


class CalibrationTransferSmokeUnitTest(unittest.TestCase):
    def test_selects_distinct_source_and_practice_personas(self) -> None:
        client = FakeClient(
            [
                [
                    {"code": "P2", "source": "database", "degraded": False},
                    {"code": "P1", "source": "database", "degraded": False},
                    {"code": "P3", "source": "fallback", "degraded": False},
                ]
            ]
        )

        self.assertEqual(MODULE._choose_distinct_personas(client), ("P1", "P2"))

    def test_session_end_poll_is_bounded_and_fetches_ready_review(self) -> None:
        client = FakeClient(
            [
                {"review_ready": False},
                {"review_ready": True},
                {"reviewReady": True, "turns": []},
            ]
        )
        clock = iter([0.0, 0.0, 0.25])

        with (
            patch.object(MODULE.time, "monotonic", side_effect=lambda: next(clock)),
            patch.object(MODULE.time, "sleep") as sleep,
        ):
            result = MODULE._wait_for_session_review(
                client, "session-1", timeout=1.0, interval=0.25
            )

        self.assertEqual(result["poll_count"], 2)
        self.assertEqual(
            client.paths,
            [
                "GET /sessions/session-1",
                "GET /sessions/session-1",
                "GET /sessions/session-1/review",
            ],
        )
        sleep.assert_called_once_with(0.25)

    def test_actual_execution_request_contains_server_identifiers_only(self) -> None:
        payload = MODULE._actual_execution_request(
            original_transfer_trial_record_id="trial-1",
            practice_session_id="session-2",
        )

        self.assertEqual(
            payload,
            {
                "original_transfer_trial_record_id": "trial-1",
                "practice_session_id": "session-2",
            },
        )

    def test_minimal_suite_uses_novel_phrase_and_durable_uuid_refs(self) -> None:
        suite = MODULE._build_transfer_suite(
            fixture_suffix="abc123", evidence_turn_ids=["turn-1", "turn-2"]
        )
        trial = suite["trials"][0]

        self.assertEqual(trial["competency_id"], "competency.empathic_attunement")
        self.assertEqual(trial["evidence_refs"], ["turn-1", "turn-2"])
        self.assertNotIn(
            trial["variation"]["phrase_family_id"],
            suite["training_phrase_family_ids"],
        )

    def test_actual_read_proof_requires_durable_model_evidence_and_separation(
        self,
    ) -> None:
        proof = MODULE._actual_execution_read_proof(
            {
                "actual_executions": [
                    {
                        "execution_event_id": "event-1",
                        "original_transfer_trial_record_id": "trial-1",
                        "practice_session_id": "session-2",
                        "evidence_turn_ids": ["turn-1", "turn-2"],
                        "normalized_evaluator_labels": {
                            "technique_codes": ["reflection"],
                            "client_state_codes": ["affect_contact"],
                            "appropriateness": ["pos"],
                            "intent_deviation_dimensions": [],
                            "evaluator_error_count": 0,
                        },
                        "model_run_id": "model-1",
                        "source_kind": "model_inferred",
                        "perspective": "independent_observer",
                        "instrument_id": "unseen-transfer-g5",
                        "instrument_version": "1.0.0",
                        "observer_version": "calibration-actual-transfer-observer-v1",
                    }
                ],
                "actual_transfer_assessments": [
                    {
                        "evidence_source": "actual_practice_execution",
                        "actual_transfer_status": "insufficient_evidence",
                        "source_execution_event_ids": ["event-1"],
                    }
                ],
            },
            execution_event_id="event-1",
            original_transfer_trial_record_id="trial-1",
            practice_session_id="session-2",
            durable_turn_ids=["turn-1", "turn-2"],
        )

        self.assertEqual(proof["model_run_id"], "model-1")
        self.assertEqual(proof["source_kind"], "model_inferred")
        self.assertEqual(proof["perspective"], "independent_observer")
        self.assertEqual(proof["instrument_id"], "unseen-transfer-g5")
        self.assertEqual(proof["instrument_version"], "1.0.0")
        self.assertEqual(
            proof["observer_version"], "calibration-actual-transfer-observer-v1"
        )
        self.assertEqual(
            proof["evidence_source"], "actual_practice_execution"
        )

    def test_actual_read_proof_rejects_raw_text_keys(self) -> None:
        with self.assertRaisesRegex(MODULE.SmokeError, "raw-text"):
            MODULE._assert_no_raw_transcript(
                {"normalized_evaluator_labels": {"raw_transcript": "secret"}}
            )

    def test_smoke_covers_actual_rejections_without_media_or_voice(self) -> None:
        source = SCRIPT_PATH.read_text(encoding="utf-8").lower()
        for required in (
            "/internal/sessions/{session_id}/calibration/transfer-suites",
            "/calibration/transfer-executions",
            "same_source_session_rejected",
            "unended_practice_session_rejected",
            "other_learner_actual_execution_rejected",
            "same_actual_execution_retry_stable",
        ):
            self.assertIn(required, source)
        for forbidden in (
            "/voice",
            "getusermedia",
            "enumeratedevices",
            "mediadevices",
        ):
            self.assertNotIn(forbidden, source)


if __name__ == "__main__":
    unittest.main()