16e791e044
8월 7일까지 워킹트리에만 남아 있던 미커밋 작업을 커밋한다. 여러 사본 폴더(worktree·clone)에 흩어져 있던 중간 스냅샷을 정리하기 전에 원본을 git 이력으로 고정하는 것이 목적이다. - contracts/routes/services: measurement, outcome_trajectory, rupture_repair, deliberate_practice, calibration_transfer, supervision_research, multimodal_alliance, continuous_improvement 계열 신규 모듈과 테스트 - infra/db/init: 07~16 마이그레이션(측정 기반~calibration transfer 실행) - apps/web: 세션 리뷰 카드·관리 화면·E2E 스펙 추가 - docs/ops: G0~G8 라이브 통합·배포·롤백 증거 문서와 evidence JSON/PNG - scripts: smoke·ledger·릴리스 에이전트·NAS 프리뷰 운영 스크립트 engine.public 로그 .bak과 apps/web/test-results 산출물은 커밋에서 제외했다.
126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
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-deliberate-practice-api.py")
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"smoke_deliberate_practice_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[dict[str, 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 DeliberatePracticeSmokeUnitTest(unittest.TestCase):
|
|
def test_selects_two_distinct_database_personas_with_p1_as_source(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_review_poll_is_bounded_and_fetches_review_only_when_ready(self) -> None:
|
|
client = FakeClient(
|
|
[
|
|
{"review_ready": False},
|
|
{"review_ready": True},
|
|
{"reviewReady": True, "turns": [{"turn_id": "a"}, {"turn_id": "b"}]},
|
|
]
|
|
)
|
|
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(result["review"]["reviewReady"], True)
|
|
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_runtime_read_model_requires_unseen_independent_model_evidence(self) -> None:
|
|
proof = MODULE._runtime_attempt_read_proof(
|
|
{
|
|
"episodes": [
|
|
{
|
|
"episode_submission_id": "submission-1",
|
|
"session_id": "session-2",
|
|
"attempts": [
|
|
{
|
|
"attempt_record_id": "attempt-1",
|
|
"scenario_novelty": "unseen_transfer",
|
|
"learner_claimed_success": False,
|
|
"evidence_turn_ids": ["turn-1", "turn-2"],
|
|
"attempt_payload": {
|
|
"observation": {
|
|
"criterion": {
|
|
"source_kind": "model_inferred",
|
|
"perspective": "independent_observer",
|
|
"model_run_id": "model-run-1",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
},
|
|
session_id="session-2",
|
|
durable_turn_ids=["turn-1", "turn-2"],
|
|
)
|
|
|
|
self.assertEqual(proof["episode_submission_id"], "submission-1")
|
|
self.assertEqual(proof["model_run_ids"], ["model-run-1"])
|
|
self.assertEqual(proof["durable_evidence_turn_ids"], ["turn-1", "turn-2"])
|
|
|
|
def test_smoke_never_calls_browser_media_or_voice_api(self) -> None:
|
|
source = SCRIPT_PATH.read_text(encoding="utf-8").lower()
|
|
for forbidden in (
|
|
"/voice",
|
|
"getusermedia",
|
|
"enumeratedevices",
|
|
"mediadevices",
|
|
):
|
|
self.assertNotIn(forbidden, source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|