316 lines
13 KiB
Python
316 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
from uuid import UUID, uuid4
|
|
|
|
import asyncpg
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from .deps import Principal, Role
|
|
from .routes import calibration_transfer
|
|
from .services import calibration_transfer_store
|
|
from .services.calibration_transfer_store import (
|
|
CalibrationTransferConflictError,
|
|
CalibrationTransferStateError,
|
|
)
|
|
|
|
|
|
def _principal(role: Role = Role.LEARNER, user_id: str | None = None) -> Principal:
|
|
return Principal(
|
|
user_id=user_id or str(uuid4()),
|
|
role=role,
|
|
cohort_ids=["g5-cohort"],
|
|
)
|
|
|
|
|
|
class CalibrationValidationTDDTests(unittest.IsolatedAsyncioTestCase):
|
|
"""Multi-angle TDD tests for OAS prediction revisions and lock boundaries."""
|
|
|
|
async def test_append_prediction_revision_rejects_out_of_bounds_probability(self) -> None:
|
|
"""predicted_success_probability must be strictly bounded within [0.0, 1.0]."""
|
|
principal = _principal()
|
|
submission_id = uuid4()
|
|
history_id = uuid4()
|
|
session_id = uuid4()
|
|
|
|
# > 1.0 should fail
|
|
with self.assertRaises(CalibrationTransferStateError) as ctx:
|
|
await calibration_transfer_store.append_prediction_revision(
|
|
principal=principal,
|
|
submission_id=submission_id,
|
|
prediction_revision_id=uuid4(),
|
|
history_id=history_id,
|
|
session_id=session_id,
|
|
competency_id="competency.empathic_reflection",
|
|
practice_block_id="oas-g5-block-1",
|
|
scenario_variant_id="variant-1",
|
|
phrase_family_id="family-1",
|
|
revision_no=1,
|
|
supersedes_prediction_revision_id=None,
|
|
predicted_success_probability=1.25,
|
|
confidence=0.8,
|
|
recorded_sequence=1,
|
|
revision_reason="Valid reason",
|
|
instrument_id="calibration-mirror-g5",
|
|
instrument_version="1.0.0",
|
|
)
|
|
self.assertIn("predicted_success_probability", str(ctx.exception))
|
|
|
|
# < 0.0 should fail
|
|
with self.assertRaises(CalibrationTransferStateError) as ctx:
|
|
await calibration_transfer_store.append_prediction_revision(
|
|
principal=principal,
|
|
submission_id=submission_id,
|
|
prediction_revision_id=uuid4(),
|
|
history_id=history_id,
|
|
session_id=session_id,
|
|
competency_id="competency.empathic_reflection",
|
|
practice_block_id="oas-g5-block-1",
|
|
scenario_variant_id="variant-1",
|
|
phrase_family_id="family-1",
|
|
revision_no=1,
|
|
supersedes_prediction_revision_id=None,
|
|
predicted_success_probability=-0.05,
|
|
confidence=0.8,
|
|
recorded_sequence=1,
|
|
revision_reason="Valid reason",
|
|
instrument_id="calibration-mirror-g5",
|
|
instrument_version="1.0.0",
|
|
)
|
|
self.assertIn("predicted_success_probability", str(ctx.exception))
|
|
|
|
async def test_append_prediction_revision_rejects_out_of_bounds_confidence(self) -> None:
|
|
"""confidence must be strictly bounded within [0.0, 1.0]."""
|
|
principal = _principal()
|
|
submission_id = uuid4()
|
|
history_id = uuid4()
|
|
session_id = uuid4()
|
|
|
|
with self.assertRaises(CalibrationTransferStateError) as ctx:
|
|
await calibration_transfer_store.append_prediction_revision(
|
|
principal=principal,
|
|
submission_id=submission_id,
|
|
prediction_revision_id=uuid4(),
|
|
history_id=history_id,
|
|
session_id=session_id,
|
|
competency_id="competency.empathic_reflection",
|
|
practice_block_id="oas-g5-block-1",
|
|
scenario_variant_id="variant-1",
|
|
phrase_family_id="family-1",
|
|
revision_no=1,
|
|
supersedes_prediction_revision_id=None,
|
|
predicted_success_probability=0.7,
|
|
confidence=1.5,
|
|
recorded_sequence=1,
|
|
revision_reason="Valid reason",
|
|
instrument_id="calibration-mirror-g5",
|
|
instrument_version="1.0.0",
|
|
)
|
|
self.assertIn("confidence", str(ctx.exception))
|
|
|
|
async def test_append_prediction_revision_rejects_empty_or_whitespace_reason(self) -> None:
|
|
"""Blank or whitespace-only revision_reason must be rejected."""
|
|
principal = _principal()
|
|
|
|
for blank in ("", " ", "\t\n"):
|
|
with self.assertRaises(CalibrationTransferStateError) as ctx:
|
|
await calibration_transfer_store.append_prediction_revision(
|
|
principal=principal,
|
|
submission_id=uuid4(),
|
|
prediction_revision_id=uuid4(),
|
|
history_id=uuid4(),
|
|
session_id=uuid4(),
|
|
competency_id="competency.empathic_reflection",
|
|
practice_block_id="oas-g5-block-1",
|
|
scenario_variant_id="variant-1",
|
|
phrase_family_id="family-1",
|
|
revision_no=1,
|
|
supersedes_prediction_revision_id=None,
|
|
predicted_success_probability=0.7,
|
|
confidence=0.8,
|
|
recorded_sequence=1,
|
|
revision_reason=blank,
|
|
instrument_id="calibration-mirror-g5",
|
|
instrument_version="1.0.0",
|
|
)
|
|
self.assertIn("revision_reason", str(ctx.exception))
|
|
|
|
async def test_append_prediction_revision_blocks_locked_history(self) -> None:
|
|
"""A locked prediction history must reject subsequent revision appends."""
|
|
principal = _principal()
|
|
history_id = uuid4()
|
|
session_id = uuid4()
|
|
|
|
# Mock db connection to simulate trigger-level lock rejection on history
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetchrow.side_effect = [
|
|
# 1. _visible_session
|
|
{"id": session_id, "session_id": session_id, "learner_id": principal.user_id},
|
|
# 2. _existing_by_submission
|
|
None,
|
|
# 3. SELECT * FROM app.calibration_prediction_history WHERE history_id = $1
|
|
{
|
|
"history_id": history_id,
|
|
"session_id": session_id,
|
|
"learner_id": principal.user_id,
|
|
"competency_id": "competency.empathic_reflection",
|
|
"practice_block_id": "oas-g5-block-1",
|
|
"scenario_variant_id": "variant-1",
|
|
"phrase_family_id": "family-1",
|
|
},
|
|
# 4. INSERT INTO app.calibration_prediction_revision (trigger raises lock error)
|
|
asyncpg.ObjectNotInPrerequisiteStateError(
|
|
"self-prediction cannot be revised after lock or external reveal"
|
|
),
|
|
]
|
|
|
|
with patch("app.services.calibration_transfer_store.db.acquire") as mock_acquire:
|
|
mock_acquire.return_value.__aenter__.return_value = mock_conn
|
|
with self.assertRaises(CalibrationTransferStateError) as ctx:
|
|
await calibration_transfer_store.append_prediction_revision(
|
|
principal=principal,
|
|
submission_id=uuid4(),
|
|
prediction_revision_id=uuid4(),
|
|
history_id=history_id,
|
|
session_id=session_id,
|
|
competency_id="competency.empathic_reflection",
|
|
practice_block_id="oas-g5-block-1",
|
|
scenario_variant_id="variant-1",
|
|
phrase_family_id="family-1",
|
|
revision_no=2,
|
|
supersedes_prediction_revision_id=uuid4(),
|
|
predicted_success_probability=0.7,
|
|
confidence=0.8,
|
|
recorded_sequence=2,
|
|
revision_reason="Trying to revise locked prediction",
|
|
instrument_id="calibration-mirror-g5",
|
|
instrument_version="1.0.0",
|
|
)
|
|
self.assertIn("locked", str(ctx.exception).lower())
|
|
|
|
async def test_append_prediction_lock_requires_learner_role(self) -> None:
|
|
"""Only learners can append prediction locks."""
|
|
principal = _principal(role=Role.TEACHER)
|
|
with self.assertRaises(CalibrationTransferStateError) as ctx:
|
|
await calibration_transfer_store.append_prediction_lock(
|
|
principal=principal,
|
|
submission_id=uuid4(),
|
|
lock_id=uuid4(),
|
|
history_id=uuid4(),
|
|
prediction_revision_id=uuid4(),
|
|
locked_sequence=1,
|
|
)
|
|
self.assertIn("learner role", str(ctx.exception).lower())
|
|
|
|
async def test_append_prediction_lock_idempotent_replay(self) -> None:
|
|
"""Replaying identical lock submission returns idempotent_replay=True."""
|
|
principal = _principal(role=Role.LEARNER)
|
|
submission_id = uuid4()
|
|
history_id = uuid4()
|
|
lock_id = uuid4()
|
|
revision_id = uuid4()
|
|
|
|
payload = {
|
|
"lock_id": str(lock_id),
|
|
"history_id": str(history_id),
|
|
"prediction_revision_id": str(revision_id),
|
|
"locked_sequence": 1,
|
|
"learner_id": principal.user_id,
|
|
}
|
|
content_hash = calibration_transfer_store._canonical_hash(payload)
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetchrow.side_effect = [
|
|
# 1. history lookup
|
|
{
|
|
"history_id": history_id,
|
|
"session_id": uuid4(),
|
|
"learner_id": principal.user_id,
|
|
},
|
|
# 2. _existing_by_submission finds existing lock with identical content_hash
|
|
{"lock_id": lock_id, "content_hash": content_hash},
|
|
]
|
|
|
|
with patch("app.services.calibration_transfer_store.db.acquire") as mock_acquire:
|
|
mock_acquire.return_value.__aenter__.return_value = mock_conn
|
|
result = await calibration_transfer_store.append_prediction_lock(
|
|
principal=principal,
|
|
submission_id=submission_id,
|
|
lock_id=lock_id,
|
|
history_id=history_id,
|
|
prediction_revision_id=revision_id,
|
|
locked_sequence=1,
|
|
)
|
|
self.assertTrue(result.get("idempotent_replay"))
|
|
self.assertEqual(result.get("lock_id"), lock_id)
|
|
|
|
async def test_append_prediction_lock_replayed_with_different_content_raises_conflict(self) -> None:
|
|
"""Replaying lock submission with changed payload raises conflict error."""
|
|
principal = _principal(role=Role.LEARNER)
|
|
submission_id = uuid4()
|
|
history_id = uuid4()
|
|
lock_id = uuid4()
|
|
revision_id = uuid4()
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetchrow.side_effect = [
|
|
# 1. history lookup
|
|
{
|
|
"history_id": history_id,
|
|
"session_id": uuid4(),
|
|
"learner_id": principal.user_id,
|
|
},
|
|
# 2. _existing_by_submission finds existing lock but different content_hash
|
|
{"lock_id": lock_id, "content_hash": "different-hash-value"},
|
|
]
|
|
|
|
with patch("app.services.calibration_transfer_store.db.acquire") as mock_acquire:
|
|
mock_acquire.return_value.__aenter__.return_value = mock_conn
|
|
with self.assertRaises(CalibrationTransferConflictError) as ctx:
|
|
await calibration_transfer_store.append_prediction_lock(
|
|
principal=principal,
|
|
submission_id=submission_id,
|
|
lock_id=lock_id,
|
|
history_id=history_id,
|
|
prediction_revision_id=revision_id,
|
|
locked_sequence=1,
|
|
)
|
|
self.assertIn("different content", str(ctx.exception))
|
|
|
|
async def test_append_prediction_lock_conflict_raises_conflict_error(self) -> None:
|
|
"""A conflicting concurrent lock raises CalibrationTransferConflictError."""
|
|
principal = _principal(role=Role.LEARNER)
|
|
submission_id = uuid4()
|
|
history_id = uuid4()
|
|
lock_id = uuid4()
|
|
revision_id = uuid4()
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetchrow.side_effect = [
|
|
# 1. history lookup
|
|
{
|
|
"history_id": history_id,
|
|
"session_id": uuid4(),
|
|
"learner_id": principal.user_id,
|
|
},
|
|
# 2. _existing_by_submission returns None (new submission)
|
|
None,
|
|
# 3. INSERT raises UniqueViolationError
|
|
asyncpg.UniqueViolationError("duplicate key value violates unique constraint"),
|
|
]
|
|
|
|
with patch("app.services.calibration_transfer_store.db.acquire") as mock_acquire:
|
|
mock_acquire.return_value.__aenter__.return_value = mock_conn
|
|
with self.assertRaises(CalibrationTransferConflictError):
|
|
await calibration_transfer_store.append_prediction_lock(
|
|
principal=principal,
|
|
submission_id=submission_id,
|
|
lock_id=lock_id,
|
|
history_id=history_id,
|
|
prediction_revision_id=revision_id,
|
|
locked_sequence=1,
|
|
)
|
|
|