886 lines
34 KiB
Python
886 lines
34 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import ValidationError
|
|
|
|
from .contracts.deliberate_practice import PracticeEpisodeAssessment
|
|
from .deps import Principal, Role
|
|
from .routes import deliberate_practices
|
|
from .services import deliberate_practice_store
|
|
from .services.deliberate_practice import (
|
|
assess_practice_episode,
|
|
load_practice_benchmark,
|
|
prescribe_from_coaching_cards,
|
|
)
|
|
|
|
|
|
BENCHMARK_PATH = (
|
|
Path(__file__).resolve().parent
|
|
/ "data"
|
|
/ "deliberate_practice_benchmark_g4.v1.json"
|
|
)
|
|
|
|
|
|
def _principal(role: Role = Role.LEARNER) -> Principal:
|
|
return Principal(
|
|
user_id=str(uuid4()),
|
|
role=role,
|
|
cohort_ids=["g4-cohort"],
|
|
)
|
|
|
|
|
|
class DeliberatePracticeStoreContractTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.pack = load_practice_benchmark(BENCHMARK_PATH)
|
|
|
|
def test_canonical_hash_is_order_independent(self) -> None:
|
|
self.assertEqual(
|
|
deliberate_practice_store._canonical_hash({"b": 2, "a": 1}),
|
|
deliberate_practice_store._canonical_hash({"a": 1, "b": 2}),
|
|
)
|
|
|
|
def test_evidence_turn_ids_must_be_nonempty_and_unique(self) -> None:
|
|
with self.assertRaises(deliberate_practice_store.DeliberatePracticeStateError):
|
|
deliberate_practice_store._ensure_unique_evidence(())
|
|
identifier = uuid4()
|
|
with self.assertRaises(deliberate_practice_store.DeliberatePracticeStateError):
|
|
deliberate_practice_store._ensure_unique_evidence((identifier, identifier))
|
|
|
|
def test_persisted_evidence_refs_must_be_turn_uuids(self) -> None:
|
|
refs = self.pack.cases[0].coaching_cards[0].evidence_refs
|
|
with self.assertRaisesRegex(
|
|
deliberate_practice_store.DeliberatePracticeStateError,
|
|
"turn UUID",
|
|
):
|
|
deliberate_practice_store._uuid_evidence_refs(refs)
|
|
|
|
def test_persisted_mastery_requires_nonblank_novel_template(self) -> None:
|
|
case = self.pack.cases[3]
|
|
prescription = prescribe_from_coaching_cards(case.coaching_cards)[0]
|
|
assessment = assess_practice_episode(prescription, case.episodes[0])
|
|
payload = assessment.model_dump(mode="json")
|
|
payload["attempts"][-1]["utterance_template_id"] = None
|
|
altered = PracticeEpisodeAssessment.model_validate(payload)
|
|
|
|
with self.assertRaisesRegex(
|
|
deliberate_practice_store.DeliberatePracticeStateError,
|
|
"memorized phrase",
|
|
):
|
|
deliberate_practice_store._ensure_persistable_transfer(altered)
|
|
|
|
def test_route_models_reject_duplicate_evidence(self) -> None:
|
|
evidence_id = uuid4()
|
|
with self.assertRaises(ValidationError):
|
|
deliberate_practices.PracticeTeacherCorrectionRequest(
|
|
submission_id=uuid4(),
|
|
corrected_outcome="needs_retry",
|
|
correction_reason="근거를 다시 확인했다.",
|
|
evidence_turn_ids=[evidence_id, evidence_id],
|
|
)
|
|
|
|
def test_attempt_response_cannot_claim_mastery_without_mastery_allowed(
|
|
self,
|
|
) -> None:
|
|
with self.assertRaisesRegex(ValidationError, "only mastered"):
|
|
deliberate_practices.PracticeAttemptSubmissionResponse(
|
|
submission_id=uuid4(),
|
|
progress="mastered",
|
|
mastery_allowed=False,
|
|
snapshot_id=uuid4(),
|
|
decision_id=uuid4(),
|
|
next_prescription_id="oas-g4-practice-next",
|
|
idempotent_replay=False,
|
|
)
|
|
|
|
def test_http_conflict_maps_to_409(self) -> None:
|
|
error = deliberate_practice_store.DeliberatePracticeConflictError("conflict")
|
|
response = deliberate_practices._http_error(error)
|
|
self.assertEqual(response.status_code, 409)
|
|
|
|
|
|
class DeliberatePracticeStoreAsyncTests(unittest.IsolatedAsyncioTestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.pack = load_practice_benchmark(BENCHMARK_PATH)
|
|
|
|
async def test_existing_submission_replays_same_content(self) -> None:
|
|
submission_id = uuid4()
|
|
conn = AsyncMock()
|
|
conn.fetchrow.return_value = {
|
|
"submission_id": submission_id,
|
|
"content_hash": "a" * 64,
|
|
}
|
|
|
|
row = await deliberate_practice_store._existing_submission(
|
|
conn,
|
|
table="app.practice_prescription_submission",
|
|
id_column="submission_id",
|
|
submission_id=submission_id,
|
|
content_hash="a" * 64,
|
|
)
|
|
|
|
self.assertEqual(row["submission_id"], submission_id)
|
|
|
|
async def test_existing_submission_with_different_content_is_conflict(self) -> None:
|
|
conn = AsyncMock()
|
|
conn.fetchrow.return_value = {
|
|
"episode_submission_id": uuid4(),
|
|
"content_hash": "a" * 64,
|
|
}
|
|
|
|
with self.assertRaises(
|
|
deliberate_practice_store.DeliberatePracticeConflictError
|
|
):
|
|
await deliberate_practice_store._existing_submission(
|
|
conn,
|
|
table="app.practice_episode_submission",
|
|
id_column="episode_submission_id",
|
|
submission_id=uuid4(),
|
|
content_hash="b" * 64,
|
|
)
|
|
|
|
async def test_internal_submission_appends_card_prescription_snapshot_and_decision(
|
|
self,
|
|
) -> None:
|
|
case = self.pack.cases[0]
|
|
session_id = uuid4()
|
|
learner_id = uuid4()
|
|
card_record_id = uuid4()
|
|
prescription_record_id = uuid4()
|
|
snapshot_id = uuid4()
|
|
decision_id = uuid4()
|
|
evidence_id = uuid4()
|
|
prescription = prescribe_from_coaching_cards(case.coaching_cards)[0]
|
|
conn = AsyncMock()
|
|
conn.fetchrow.side_effect = [
|
|
{"id": session_id, "learner_id": learner_id},
|
|
None,
|
|
None,
|
|
{"coaching_card_record_id": card_record_id},
|
|
{"prescription_record_id": prescription_record_id},
|
|
None,
|
|
{"snapshot_id": snapshot_id, "snapshot_no": 1},
|
|
{"decision_id": decision_id},
|
|
]
|
|
conn.fetch.return_value = [
|
|
{
|
|
"prescription_record_id": prescription_record_id,
|
|
"prescription_key": prescription.prescription_id,
|
|
"prescription_payload": prescription.model_dump(mode="json"),
|
|
}
|
|
]
|
|
|
|
result = await deliberate_practice_store.append_prescription_submission(
|
|
conn=conn,
|
|
session_id=session_id,
|
|
submission_id=uuid4(),
|
|
coaching_cards=case.coaching_cards,
|
|
graph=case.graph,
|
|
evidence_turn_ids=(evidence_id,),
|
|
)
|
|
|
|
self.assertEqual(result["snapshot_id"], snapshot_id)
|
|
self.assertEqual(result["decision_id"], decision_id)
|
|
self.assertEqual(result["next_prescription_id"], prescription.prescription_id)
|
|
self.assertFalse(result["idempotent_replay"])
|
|
inserts = "\n".join(
|
|
str(call.args[0]) for call in conn.fetchrow.await_args_list if call.args
|
|
)
|
|
self.assertIn("app.practice_coaching_card", inserts)
|
|
self.assertIn("app.practice_prescription", inserts)
|
|
self.assertIn("app.competency_graph_snapshot", inserts)
|
|
self.assertIn("app.practice_curriculum_decision_event", inserts)
|
|
|
|
async def test_nonlearner_cannot_submit_attempt_before_db(self) -> None:
|
|
with self.assertRaisesRegex(
|
|
deliberate_practice_store.DeliberatePracticeStateError,
|
|
"learner role",
|
|
):
|
|
await deliberate_practice_store.append_learner_attempt_submission(
|
|
principal=_principal(Role.TEACHER),
|
|
submission_id=uuid4(),
|
|
prescription_id="oas-g4-practice-any",
|
|
episode=self.pack.cases[0].episodes[0],
|
|
)
|
|
|
|
async def test_attempt_rejects_disabled_prescription_source_before_replay(
|
|
self,
|
|
) -> None:
|
|
principal = _principal(Role.LEARNER)
|
|
case = self.pack.cases[0]
|
|
episode_payload = case.episodes[0].model_dump(mode="python")
|
|
for attempt in episode_payload["attempts"]:
|
|
for ref in attempt["evidence_refs"]:
|
|
ref["ref_id"] = str(uuid4())
|
|
for ref in attempt["criterion"]["evidence_refs"]:
|
|
ref["ref_id"] = str(uuid4())
|
|
episode = type(case.episodes[0]).model_validate(episode_payload)
|
|
source_session_id = uuid4()
|
|
conn = AsyncMock()
|
|
conn.fetchrow.return_value = {
|
|
"prescription_record_id": uuid4(),
|
|
"session_id": source_session_id,
|
|
"prescription_payload": {},
|
|
"created_at": datetime.now(timezone.utc),
|
|
"source_learner_feedback_enabled": False,
|
|
}
|
|
|
|
@asynccontextmanager
|
|
async def fake_acquire(**_kwargs):
|
|
yield conn
|
|
|
|
with patch.object(deliberate_practice_store.db, "acquire", fake_acquire):
|
|
with self.assertRaises(
|
|
deliberate_practice_store.DeliberatePracticeFeedbackDisabledError
|
|
):
|
|
await deliberate_practice_store.append_learner_attempt_submission(
|
|
principal=principal,
|
|
submission_id=uuid4(),
|
|
prescription_id=episode.prescription_id,
|
|
episode=episode,
|
|
)
|
|
|
|
self.assertEqual(conn.fetchrow.await_count, 1)
|
|
source_query = conn.fetchrow.await_args.args[0]
|
|
self.assertIn("source_session.learner_feedback_enabled", source_query)
|
|
self.assertNotIn("app.practice_episode_submission", source_query)
|
|
|
|
async def test_runtime_observer_derives_attempt_from_later_ready_session(
|
|
self,
|
|
) -> None:
|
|
principal = _principal(Role.LEARNER)
|
|
learner_id = UUID(principal.user_id)
|
|
source_session_id = uuid4()
|
|
practice_session_id = uuid4()
|
|
source_case_id = uuid4()
|
|
persona_id = uuid4()
|
|
counselor_turn_id = uuid4()
|
|
client_turn_id = uuid4()
|
|
prescription = prescribe_from_coaching_cards(
|
|
self.pack.cases[0].coaching_cards
|
|
)[0]
|
|
created_at = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
conn = AsyncMock()
|
|
conn.fetchrow.side_effect = [
|
|
{
|
|
"prescription_payload": prescription.model_dump(mode="json"),
|
|
"source_session_id": source_session_id,
|
|
"prescription_created_at": created_at,
|
|
"source_case_id": source_case_id,
|
|
"source_persona_id": persona_id,
|
|
},
|
|
{
|
|
"id": practice_session_id,
|
|
"case_id": source_case_id,
|
|
"persona_id": persona_id,
|
|
"started_at": created_at + timedelta(minutes=5),
|
|
"ended_at": created_at + timedelta(minutes=30),
|
|
"evaluation_status": "ready",
|
|
"evaluation_scope": "session_end",
|
|
},
|
|
]
|
|
conn.fetch.return_value = [
|
|
{
|
|
"counselor_turn_id": counselor_turn_id,
|
|
"counselor_turn_seq": 1,
|
|
"client_turn_id": client_turn_id,
|
|
"client_turn_seq": 2,
|
|
"technique_codes": ["reflection"],
|
|
"client_state_codes": ["affect_contact"],
|
|
"appropriateness": "pos",
|
|
"intent_deviation_dimensions": [],
|
|
"evaluator_error": None,
|
|
"utterance_fingerprint": "sha256:runtime",
|
|
"has_voice_feature": False,
|
|
}
|
|
]
|
|
|
|
@asynccontextmanager
|
|
async def fake_acquire(**kwargs):
|
|
self.assertEqual(kwargs["user_id"], str(learner_id))
|
|
yield conn
|
|
|
|
persisted = {
|
|
"submission_id": uuid4(),
|
|
"progress": "transfer_pending",
|
|
"mastery_allowed": False,
|
|
"snapshot_id": uuid4(),
|
|
"decision_id": uuid4(),
|
|
"next_prescription_id": prescription.prescription_id,
|
|
"idempotent_replay": False,
|
|
}
|
|
with (
|
|
patch.object(deliberate_practice_store.db, "acquire", fake_acquire),
|
|
patch.object(
|
|
deliberate_practice_store,
|
|
"_ensure_runtime_observer_model_runs",
|
|
AsyncMock(),
|
|
) as ensure_runs,
|
|
patch.object(
|
|
deliberate_practice_store,
|
|
"append_learner_attempt_submission",
|
|
AsyncMock(return_value=persisted),
|
|
) as append_attempt,
|
|
):
|
|
result = await deliberate_practice_store.append_runtime_practice_session(
|
|
principal=principal,
|
|
prescription_id=prescription.prescription_id,
|
|
practice_session_id=practice_session_id,
|
|
)
|
|
|
|
self.assertEqual(result["progress"], "transfer_pending")
|
|
ensure_runs.assert_awaited_once()
|
|
runtime_turn_query = conn.fetch.await_args_list[0].args[0]
|
|
self.assertIn("app.digest(", runtime_turn_query)
|
|
self.assertEqual(
|
|
append_attempt.await_args.kwargs["practice_session_id"],
|
|
practice_session_id,
|
|
)
|
|
episode = append_attempt.await_args.kwargs["episode"]
|
|
self.assertEqual(episode.attempts[0].criterion.source_kind, "model_inferred")
|
|
self.assertEqual(episode.attempts[0].scenario_novelty, "familiar")
|
|
|
|
async def test_learner_cannot_append_teacher_correction(self) -> None:
|
|
with self.assertRaisesRegex(
|
|
deliberate_practice_store.DeliberatePracticeStateError,
|
|
"teacher or admin",
|
|
):
|
|
await deliberate_practice_store.append_teacher_correction(
|
|
principal=_principal(Role.LEARNER),
|
|
attempt_record_id=uuid4(),
|
|
submission_id=uuid4(),
|
|
corrected_outcome="needs_retry",
|
|
correction_reason="근거 재확인",
|
|
evidence_turn_ids=(uuid4(),),
|
|
counterevidence=(),
|
|
)
|
|
|
|
async def test_teacher_correction_appends_superseding_event_in_cohort_context(
|
|
self,
|
|
) -> None:
|
|
principal = _principal(Role.TEACHER)
|
|
attempt_id = uuid4()
|
|
episode_id = uuid4()
|
|
session_id = uuid4()
|
|
learner_id = uuid4()
|
|
correction_id = uuid4()
|
|
conn = AsyncMock()
|
|
conn.fetchrow.side_effect = [
|
|
None,
|
|
{
|
|
"attempt_record_id": attempt_id,
|
|
"episode_submission_id": episode_id,
|
|
"session_id": session_id,
|
|
"learner_id": learner_id,
|
|
},
|
|
None,
|
|
{"correction_id": correction_id, "correction_no": 1},
|
|
]
|
|
|
|
@asynccontextmanager
|
|
async def fake_acquire(**kwargs):
|
|
self.assertEqual(kwargs["role"], "teacher")
|
|
self.assertEqual(kwargs["cohort_ids"], ["g4-cohort"])
|
|
yield conn
|
|
|
|
with patch.object(deliberate_practice_store.db, "acquire", fake_acquire):
|
|
result = await deliberate_practice_store.append_teacher_correction(
|
|
principal=principal,
|
|
attempt_record_id=attempt_id,
|
|
submission_id=uuid4(),
|
|
corrected_outcome="needs_retry",
|
|
correction_reason="내담자 반응 근거를 다시 확인했다.",
|
|
evidence_turn_ids=(uuid4(),),
|
|
counterevidence=("client_response_not_engaged",),
|
|
)
|
|
|
|
self.assertEqual(result["correction_id"], correction_id)
|
|
self.assertEqual(result["correction_no"], 1)
|
|
insert_query = conn.fetchrow.await_args_list[-1].args[0]
|
|
self.assertIn("app.practice_teacher_correction", insert_query)
|
|
|
|
async def test_empty_learner_read_uses_self_rls_context(self) -> None:
|
|
principal = _principal(Role.LEARNER)
|
|
conn = AsyncMock()
|
|
conn.fetch.side_effect = [[], []]
|
|
conn.fetchrow.return_value = None
|
|
|
|
@asynccontextmanager
|
|
async def fake_acquire(**kwargs):
|
|
self.assertEqual(kwargs["role"], "learner")
|
|
self.assertEqual(kwargs["user_id"], principal.user_id)
|
|
yield conn
|
|
|
|
with patch.object(deliberate_practice_store.db, "acquire", fake_acquire):
|
|
result = await deliberate_practice_store.read_deliberate_practice(
|
|
principal=principal
|
|
)
|
|
|
|
self.assertEqual(result["learner_id"], UUID(principal.user_id))
|
|
self.assertEqual(result["prescriptions"], [])
|
|
self.assertEqual(result["episodes"], [])
|
|
self.assertIsNone(result["competency_graph"])
|
|
self.assertFalse(result["clinical_claim_allowed"])
|
|
|
|
async def test_learner_read_rejects_any_disabled_source_snapshot(self) -> None:
|
|
principal = _principal(Role.LEARNER)
|
|
conn = AsyncMock()
|
|
conn.fetch.side_effect = [
|
|
[
|
|
{
|
|
"prescription_record_id": uuid4(),
|
|
"session_id": uuid4(),
|
|
"source_learner_feedback_enabled": False,
|
|
}
|
|
],
|
|
[],
|
|
]
|
|
conn.fetchrow.return_value = None
|
|
|
|
@asynccontextmanager
|
|
async def fake_acquire(**_kwargs):
|
|
yield conn
|
|
|
|
with patch.object(deliberate_practice_store.db, "acquire", fake_acquire):
|
|
with self.assertRaises(
|
|
deliberate_practice_store.DeliberatePracticeFeedbackDisabledError
|
|
):
|
|
await deliberate_practice_store.read_deliberate_practice(
|
|
principal=principal
|
|
)
|
|
|
|
prescription_query = conn.fetch.await_args_list[0].args[0]
|
|
episode_query = conn.fetch.await_args_list[1].args[0]
|
|
self.assertIn("source_session.learner_feedback_enabled", prescription_query)
|
|
self.assertIn("source_session.learner_feedback_enabled", episode_query)
|
|
|
|
async def test_teacher_read_keeps_disabled_source_snapshot_visible(self) -> None:
|
|
principal = _principal(Role.TEACHER)
|
|
learner_id = uuid4()
|
|
conn = AsyncMock()
|
|
conn.fetchval.return_value = True
|
|
conn.fetch.side_effect = [
|
|
[
|
|
{
|
|
"prescription_record_id": uuid4(),
|
|
"session_id": uuid4(),
|
|
"source_learner_feedback_enabled": False,
|
|
}
|
|
],
|
|
[],
|
|
]
|
|
conn.fetchrow.return_value = None
|
|
|
|
@asynccontextmanager
|
|
async def fake_acquire(**_kwargs):
|
|
yield conn
|
|
|
|
with patch.object(deliberate_practice_store.db, "acquire", fake_acquire):
|
|
result = await deliberate_practice_store.read_deliberate_practice(
|
|
principal=principal,
|
|
learner_id=learner_id,
|
|
)
|
|
|
|
self.assertEqual(len(result["prescriptions"]), 1)
|
|
self.assertNotIn(
|
|
"source_learner_feedback_enabled",
|
|
result["prescriptions"][0],
|
|
)
|
|
|
|
async def test_learner_read_cannot_target_another_learner(self) -> None:
|
|
with self.assertRaises(
|
|
deliberate_practice_store.DeliberatePracticeNotFoundError
|
|
):
|
|
await deliberate_practice_store.read_deliberate_practice(
|
|
principal=_principal(Role.LEARNER), learner_id=uuid4()
|
|
)
|
|
|
|
async def test_teacher_read_fails_closed_outside_cohort_scope(self) -> None:
|
|
principal = _principal(Role.TEACHER)
|
|
conn = AsyncMock()
|
|
conn.fetchval.return_value = False
|
|
|
|
@asynccontextmanager
|
|
async def fake_acquire(**kwargs):
|
|
self.assertEqual(kwargs["cohort_ids"], ["g4-cohort"])
|
|
yield conn
|
|
|
|
with patch.object(deliberate_practice_store.db, "acquire", fake_acquire):
|
|
with self.assertRaises(
|
|
deliberate_practice_store.DeliberatePracticeNotFoundError
|
|
):
|
|
await deliberate_practice_store.read_deliberate_practice(
|
|
principal=principal,
|
|
learner_id=uuid4(),
|
|
)
|
|
conn.fetch.assert_not_awaited()
|
|
|
|
async def test_attempt_route_maps_store_conflict_to_409(self) -> None:
|
|
case = self.pack.cases[0]
|
|
body = deliberate_practices.PracticeAttemptSubmissionRequest(
|
|
submission_id=uuid4(),
|
|
episode=case.episodes[0],
|
|
)
|
|
with patch.object(
|
|
deliberate_practices.deliberate_practice_store,
|
|
"append_learner_attempt_submission",
|
|
AsyncMock(
|
|
side_effect=deliberate_practice_store.DeliberatePracticeConflictError(
|
|
"idempotency conflict"
|
|
)
|
|
),
|
|
):
|
|
with self.assertRaises(HTTPException) as captured:
|
|
await deliberate_practices.create_practice_attempt(
|
|
prescription_id=case.episodes[0].prescription_id,
|
|
body=body,
|
|
principal=_principal(Role.LEARNER),
|
|
)
|
|
self.assertEqual(captured.exception.status_code, 409)
|
|
|
|
async def test_attempt_route_maps_disabled_source_snapshot_to_403(self) -> None:
|
|
case = self.pack.cases[0]
|
|
body = deliberate_practices.PracticeAttemptSubmissionRequest(
|
|
submission_id=uuid4(),
|
|
episode=case.episodes[0],
|
|
)
|
|
with patch.object(
|
|
deliberate_practices.deliberate_practice_store,
|
|
"append_learner_attempt_submission",
|
|
AsyncMock(
|
|
side_effect=deliberate_practice_store.DeliberatePracticeFeedbackDisabledError(
|
|
"source session feedback disabled"
|
|
)
|
|
),
|
|
) as append:
|
|
with self.assertRaises(HTTPException) as captured:
|
|
await deliberate_practices.create_practice_attempt(
|
|
prescription_id=case.episodes[0].prescription_id,
|
|
body=body,
|
|
principal=_principal(Role.LEARNER),
|
|
)
|
|
|
|
self.assertEqual(captured.exception.status_code, 403)
|
|
self.assertEqual(captured.exception.detail, "learner_feedback_disabled")
|
|
append.assert_awaited_once()
|
|
|
|
async def test_runtime_practice_route_forwards_completed_session_identity(
|
|
self,
|
|
) -> None:
|
|
practice_session_id = uuid4()
|
|
prescription_id = "oas-g4-practice-runtime-route"
|
|
mocked = AsyncMock(
|
|
return_value={
|
|
"submission_id": uuid4(),
|
|
"progress": "transfer_pending",
|
|
"mastery_allowed": False,
|
|
"snapshot_id": uuid4(),
|
|
"decision_id": uuid4(),
|
|
"next_prescription_id": prescription_id,
|
|
"idempotent_replay": False,
|
|
}
|
|
)
|
|
with patch.object(
|
|
deliberate_practices.deliberate_practice_store,
|
|
"append_runtime_practice_session",
|
|
mocked,
|
|
):
|
|
response = await deliberate_practices.observe_completed_practice_session(
|
|
prescription_id=prescription_id,
|
|
practice_session_id=practice_session_id,
|
|
principal=_principal(Role.LEARNER),
|
|
)
|
|
|
|
self.assertEqual(response.progress, "transfer_pending")
|
|
self.assertEqual(mocked.await_args.kwargs["practice_session_id"], practice_session_id)
|
|
self.assertEqual(mocked.await_args.kwargs["prescription_id"], prescription_id)
|
|
|
|
async def test_teacher_route_forwards_append_only_correction(self) -> None:
|
|
attempt_id = uuid4()
|
|
correction_id = uuid4()
|
|
body = deliberate_practices.PracticeTeacherCorrectionRequest(
|
|
submission_id=uuid4(),
|
|
corrected_outcome="needs_retry",
|
|
correction_reason="근거를 다시 판정했다.",
|
|
evidence_turn_ids=[uuid4()],
|
|
counterevidence=["client_response_not_engaged"],
|
|
)
|
|
mocked = AsyncMock(
|
|
return_value={
|
|
"submission_id": body.submission_id,
|
|
"correction_id": correction_id,
|
|
"correction_no": 1,
|
|
"idempotent_replay": False,
|
|
}
|
|
)
|
|
with patch.object(
|
|
deliberate_practices.deliberate_practice_store,
|
|
"append_teacher_correction",
|
|
mocked,
|
|
):
|
|
response = await deliberate_practices.correct_practice_attempt(
|
|
attempt_record_id=attempt_id,
|
|
body=body,
|
|
principal=_principal(Role.TEACHER),
|
|
)
|
|
|
|
self.assertEqual(response.correction_id, correction_id)
|
|
self.assertEqual(mocked.await_args.kwargs["attempt_record_id"], attempt_id)
|
|
|
|
|
|
class DeliberatePracticeInternalAuthenticationTests(unittest.IsolatedAsyncioTestCase):
|
|
TOKEN = "g4-test-token-with-at-least-32-characters-0001"
|
|
|
|
@staticmethod
|
|
def _settings(token: str):
|
|
return deliberate_practices.Settings(
|
|
_env_file=None,
|
|
practice_internal_token=token,
|
|
)
|
|
|
|
async def test_unconfigured_token_disables_endpoint_before_db_acquire(self) -> None:
|
|
reached = False
|
|
|
|
async def fake_provider():
|
|
nonlocal reached
|
|
reached = True
|
|
yield AsyncMock()
|
|
|
|
dependency = deliberate_practices.practice_internal_evaluator_db(
|
|
settings=self._settings(""),
|
|
presented_token=None,
|
|
)
|
|
with patch.object(
|
|
deliberate_practices, "_evaluator_db_provider", fake_provider
|
|
):
|
|
with self.assertRaises(HTTPException) as captured:
|
|
await anext(dependency)
|
|
|
|
self.assertEqual(captured.exception.status_code, 503)
|
|
self.assertEqual(
|
|
captured.exception.detail,
|
|
"internal practice ingestion is unavailable",
|
|
)
|
|
self.assertFalse(reached)
|
|
|
|
async def test_missing_header_is_401_before_db_acquire(self) -> None:
|
|
reached = False
|
|
|
|
async def fake_provider():
|
|
nonlocal reached
|
|
reached = True
|
|
yield AsyncMock()
|
|
|
|
dependency = deliberate_practices.practice_internal_evaluator_db(
|
|
settings=self._settings(self.TOKEN),
|
|
presented_token=None,
|
|
)
|
|
with patch.object(
|
|
deliberate_practices, "_evaluator_db_provider", fake_provider
|
|
):
|
|
with self.assertRaises(HTTPException) as captured:
|
|
await anext(dependency)
|
|
|
|
self.assertEqual(captured.exception.status_code, 401)
|
|
self.assertFalse(reached)
|
|
|
|
async def test_mismatched_header_is_403_and_uses_compare_digest(self) -> None:
|
|
reached = False
|
|
|
|
async def fake_provider():
|
|
nonlocal reached
|
|
reached = True
|
|
yield AsyncMock()
|
|
|
|
presented = "wrong-token-that-must-not-be-reflected"
|
|
dependency = deliberate_practices.practice_internal_evaluator_db(
|
|
settings=self._settings(self.TOKEN),
|
|
presented_token=presented,
|
|
)
|
|
with (
|
|
patch.object(deliberate_practices, "_evaluator_db_provider", fake_provider),
|
|
patch.object(
|
|
deliberate_practices.secrets,
|
|
"compare_digest",
|
|
wraps=deliberate_practices.secrets.compare_digest,
|
|
) as compared,
|
|
):
|
|
with self.assertRaises(HTTPException) as captured:
|
|
await anext(dependency)
|
|
|
|
self.assertEqual(captured.exception.status_code, 403)
|
|
self.assertEqual(captured.exception.detail, "internal authentication failed")
|
|
compared.assert_called_once_with(presented, self.TOKEN)
|
|
self.assertNotIn(presented, str(captured.exception.detail))
|
|
self.assertNotIn(self.TOKEN, str(captured.exception.detail))
|
|
self.assertFalse(reached)
|
|
|
|
async def test_valid_token_reaches_evaluator_db_provider(self) -> None:
|
|
connection = AsyncMock()
|
|
reached = 0
|
|
|
|
async def fake_provider():
|
|
nonlocal reached
|
|
reached += 1
|
|
yield connection
|
|
|
|
dependency = deliberate_practices.practice_internal_evaluator_db(
|
|
settings=self._settings(self.TOKEN),
|
|
presented_token=self.TOKEN,
|
|
)
|
|
with patch.object(
|
|
deliberate_practices, "_evaluator_db_provider", fake_provider
|
|
):
|
|
result = await anext(dependency)
|
|
await dependency.aclose()
|
|
|
|
self.assertIs(result, connection)
|
|
self.assertEqual(reached, 1)
|
|
|
|
async def test_short_configured_token_is_fail_closed_as_unavailable(self) -> None:
|
|
dependency = deliberate_practices.practice_internal_evaluator_db(
|
|
settings=self._settings("too-short"),
|
|
presented_token="too-short",
|
|
)
|
|
with self.assertRaises(HTTPException) as captured:
|
|
await anext(dependency)
|
|
|
|
self.assertEqual(captured.exception.status_code, 503)
|
|
|
|
async def test_secret_setting_repr_and_json_do_not_expose_token(self) -> None:
|
|
settings = self._settings(self.TOKEN)
|
|
|
|
self.assertNotIn(self.TOKEN, repr(settings))
|
|
self.assertNotIn(self.TOKEN, settings.model_dump_json())
|
|
self.assertEqual(
|
|
settings.practice_internal_token.get_secret_value(),
|
|
self.TOKEN,
|
|
)
|
|
|
|
async def test_openapi_documents_typed_header_only_on_internal_write(
|
|
self,
|
|
) -> None:
|
|
app = FastAPI()
|
|
app.include_router(deliberate_practices.router)
|
|
paths = app.openapi()["paths"]
|
|
|
|
internal_parameters = paths[
|
|
"/internal/sessions/{session_id}/practice/prescriptions"
|
|
]["post"].get("parameters", [])
|
|
internal_headers = {
|
|
item["name"]: item["schema"]
|
|
for item in internal_parameters
|
|
if item["in"] == "header"
|
|
}
|
|
self.assertEqual(
|
|
set(internal_headers),
|
|
{deliberate_practices.INTERNAL_TOKEN_HEADER},
|
|
)
|
|
self.assertIn(
|
|
{"type": "string"},
|
|
internal_headers[deliberate_practices.INTERNAL_TOKEN_HEADER]["anyOf"],
|
|
)
|
|
|
|
for path, method in (
|
|
("/practice/{prescription_id}/attempts", "post"),
|
|
(
|
|
"/practice/{prescription_id}/attempts/from-session/{practice_session_id}",
|
|
"post",
|
|
),
|
|
("/practice/attempts/{attempt_record_id}/correction", "patch"),
|
|
("/practice/learners/me", "get"),
|
|
("/practice/learners/{learner_id}", "get"),
|
|
):
|
|
header_names = {
|
|
item["name"]
|
|
for item in paths[path][method].get("parameters", [])
|
|
if item["in"] == "header"
|
|
}
|
|
self.assertNotIn(
|
|
deliberate_practices.INTERNAL_TOKEN_HEADER,
|
|
header_names,
|
|
)
|
|
|
|
|
|
class DeliberatePracticeSchemaStaticTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.sql = (
|
|
Path(__file__).resolve().parents[3]
|
|
/ "infra"
|
|
/ "db"
|
|
/ "init"
|
|
/ "10_deliberate_practice.sql"
|
|
).read_text(encoding="utf-8")
|
|
|
|
def test_schema_has_eight_append_only_rls_ledgers(self) -> None:
|
|
tables = (
|
|
"practice_prescription_submission",
|
|
"practice_coaching_card",
|
|
"practice_prescription",
|
|
"practice_episode_submission",
|
|
"practice_attempt_evidence",
|
|
"competency_graph_snapshot",
|
|
"practice_curriculum_decision_event",
|
|
"practice_teacher_correction",
|
|
)
|
|
for table in tables:
|
|
self.assertIn(f"CREATE TABLE IF NOT EXISTS app.{table}", self.sql)
|
|
self.assertIn("ALTER TABLE app.%I ENABLE ROW LEVEL SECURITY", self.sql)
|
|
self.assertIn("audit.reject_measurement_mutation()", self.sql)
|
|
|
|
def test_schema_enforces_submission_hash_and_turn_ownership(self) -> None:
|
|
self.assertGreaterEqual(self.sql.count("content_hash TEXT NOT NULL"), 5)
|
|
self.assertIn("practice evidence turns must belong to its session", self.sql)
|
|
self.assertIn("submission_id UUID PRIMARY KEY", self.sql)
|
|
self.assertIn("episode_submission_id UUID PRIMARY KEY", self.sql)
|
|
|
|
def test_schema_preserves_all_three_hacking_guards(self) -> None:
|
|
self.assertIn("cannot reward repeated easy familiar practice", self.sql)
|
|
self.assertIn("memorized phrase", self.sql)
|
|
self.assertIn("novel unseen transfer evidence", self.sql)
|
|
self.assertIn("new transfer_verified competency requires", self.sql)
|
|
|
|
def test_cross_session_runtime_migration_preserves_learner_and_transfer_gates(
|
|
self,
|
|
) -> None:
|
|
migration = (
|
|
Path(__file__).resolve().parents[3]
|
|
/ "infra"
|
|
/ "db"
|
|
/ "init"
|
|
/ "15_self_directed_practice_runtime.sql"
|
|
).read_text(encoding="utf-8")
|
|
self.assertIn(
|
|
"FOREIGN KEY (prescription_record_id, learner_id)", migration
|
|
)
|
|
self.assertIn("cross-session self-directed practice runtime", migration.lower())
|
|
self.assertIn("durable_prior_familiar", migration)
|
|
self.assertIn("memorized phrase", migration)
|
|
|
|
def test_schema_has_learner_self_and_teacher_cohort_rls(self) -> None:
|
|
self.assertIn("learner_id = app.current_uid()", self.sql)
|
|
self.assertIn(
|
|
"u.cohort = current_setting('app.current_cohort', true)", self.sql
|
|
)
|
|
self.assertIn("created_by_role = 'instructor'", self.sql)
|
|
|
|
def test_teacher_correction_is_superseding_event_not_update(self) -> None:
|
|
correction_sql = self.sql.split(
|
|
"CREATE TABLE IF NOT EXISTS app.practice_teacher_correction", 1
|
|
)[1].split("CREATE INDEX", 1)[0]
|
|
self.assertIn("supersedes_correction_id", correction_sql)
|
|
self.assertIn("correction_no", correction_sql)
|
|
self.assertIn("content_hash", correction_sql)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|