642 lines
24 KiB
Python
642 lines
24 KiB
Python
"""계정별 학습자 AI 피드백 정책 회귀 테스트."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, patch
|
|
from uuid import UUID
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from .deps import Principal, Role
|
|
from . import session_persistence
|
|
from .routes import admin as admin_routes
|
|
from .routes import (
|
|
calibration_transfer,
|
|
deliberate_practices,
|
|
measurements,
|
|
multimodal_alliance,
|
|
outcome_trajectories,
|
|
rupture_repairs,
|
|
sessions,
|
|
)
|
|
from .services import deliberate_practice_store, feedback_policy, state_machine
|
|
from .services.persona import P1
|
|
from .session_read_model import SessionReviewReadInput, build_session_review
|
|
from .store import InProcSession, TurnRecord
|
|
|
|
|
|
def _session(*, learner_feedback_enabled: bool) -> InProcSession:
|
|
state = state_machine.init_state(params=P1.openness_params())
|
|
return InProcSession(
|
|
session_id="00000000-0000-4000-8000-000000000401",
|
|
case_id="feedback-policy-case",
|
|
learner_id="00000000-0000-0000-0000-000000000401",
|
|
persona_code=P1.code,
|
|
theory_mode="humanistic",
|
|
persona=P1,
|
|
state=state,
|
|
created_at=1_000.0,
|
|
ended_at=1_120.0,
|
|
ended=True,
|
|
learner_feedback_enabled=learner_feedback_enabled,
|
|
turns=[
|
|
TurnRecord(
|
|
turn_seq=1,
|
|
speaker="counselor",
|
|
stage=state.stage.value,
|
|
text="상담자 발화",
|
|
text_masked="상담자 발화",
|
|
evaluation={
|
|
"appropriateness": "pos",
|
|
"appropriateness_note": "AI 학습 피드백",
|
|
"techniques": [{"label": "반영"}],
|
|
},
|
|
),
|
|
TurnRecord(
|
|
turn_seq=2,
|
|
speaker="client",
|
|
stage=state.stage.value,
|
|
text="가상 내담자 응답",
|
|
text_masked="가상 내담자 응답",
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
class FeedbackPolicyTest(unittest.TestCase):
|
|
def test_admin_contract_defaults_existing_and_new_accounts_to_enabled(self) -> None:
|
|
created = admin_routes.AdminUserCreate(
|
|
email="feedback-policy@hs.ac.kr",
|
|
display_name="피드백 정책 대상",
|
|
)
|
|
patched = admin_routes.AdminUserPatch(learner_feedback_enabled=False)
|
|
|
|
self.assertTrue(created.learner_feedback_enabled)
|
|
self.assertFalse(patched.learner_feedback_enabled)
|
|
|
|
def test_only_learner_viewer_is_blocked(self) -> None:
|
|
sess = _session(learner_feedback_enabled=False)
|
|
|
|
self.assertFalse(feedback_policy.learner_feedback_enabled(sess))
|
|
self.assertFalse(
|
|
feedback_policy.can_expose_learner_feedback(
|
|
sess,
|
|
viewer_role="learner",
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
feedback_policy.can_expose_learner_feedback(
|
|
sess,
|
|
viewer_role="teacher",
|
|
)
|
|
)
|
|
|
|
def test_current_account_off_overrides_past_enabled_session_snapshot(self) -> None:
|
|
sess = _session(learner_feedback_enabled=True)
|
|
principal = Principal(
|
|
user_id=sess.learner_id,
|
|
role=Role.LEARNER,
|
|
learner_feedback_enabled=False,
|
|
)
|
|
|
|
self.assertFalse(
|
|
feedback_policy.effective_learner_feedback_enabled(sess, principal)
|
|
)
|
|
self.assertFalse(
|
|
feedback_policy.can_expose_principal_learner_feedback(sess, principal)
|
|
)
|
|
self.assertTrue(
|
|
feedback_policy.can_expose_learner_feedback(
|
|
sess,
|
|
viewer_role="admin",
|
|
)
|
|
)
|
|
|
|
def test_disabled_review_preserves_transcript_but_removes_feedback(self) -> None:
|
|
sess = _session(learner_feedback_enabled=False)
|
|
|
|
review = build_session_review(
|
|
SessionReviewReadInput(
|
|
session=sess,
|
|
evaluation_record={"status": "ready", "payload": {"summary": "AI 요약"}},
|
|
evaluation_durable=True,
|
|
learner_feedback_enabled=False,
|
|
expose_learner_feedback=False,
|
|
now_ts=1_120.0,
|
|
)
|
|
)
|
|
|
|
self.assertFalse(review.learnerFeedbackEnabled)
|
|
self.assertFalse(review.reviewReady)
|
|
self.assertEqual(review.supervisorState, "피드백 비활성")
|
|
self.assertIn("AI 학습 피드백을 비활성화", review.summary)
|
|
self.assertEqual([turn.text for turn in review.turns], ["상담자 발화", "가상 내담자 응답"])
|
|
self.assertTrue(all(turn.note is None for turn in review.turns))
|
|
self.assertTrue(all(not turn.techniques for turn in review.turns))
|
|
self.assertTrue(all(not turn.nonverbal for turn in review.turns))
|
|
self.assertEqual(review.rubric, [])
|
|
self.assertEqual(review.goodMoments, [])
|
|
self.assertEqual(review.growthPoints, [])
|
|
self.assertIsNone(review.nextLine)
|
|
self.assertIsNone(review.clientFeedback)
|
|
self.assertEqual(review.caseWorksheet.status, "empty")
|
|
|
|
def test_disabled_review_preserves_saved_learner_worksheet_verbatim(self) -> None:
|
|
sess = _session(learner_feedback_enabled=False)
|
|
saved = {
|
|
"status": "saved_by_learner",
|
|
"generatedBy": "learner",
|
|
"sections": [
|
|
{
|
|
"key": "learner-notes",
|
|
"title": "내 기록",
|
|
"items": [
|
|
{
|
|
"key": "presenting-complaint",
|
|
"label": "주호소",
|
|
"value": "학습자가 직접 저장한 원문",
|
|
"evidence": [],
|
|
"confidence": "none",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
"limitations": ["학습자 작성"],
|
|
"savedAt": "2026-08-27T01:02:03Z",
|
|
}
|
|
|
|
review = build_session_review(
|
|
SessionReviewReadInput(
|
|
session=sess,
|
|
saved_worksheet_payload=saved,
|
|
learner_feedback_enabled=False,
|
|
expose_learner_feedback=False,
|
|
now_ts=1_120.0,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(review.caseWorksheet.status, "saved_by_learner")
|
|
self.assertEqual(
|
|
review.caseWorksheet.sections[0].items[0].value,
|
|
"학습자가 직접 저장한 원문",
|
|
)
|
|
self.assertEqual(review.caseWorksheet.limitations, ["학습자 작성"])
|
|
|
|
def test_calibration_redaction_keeps_only_learner_authored_prediction(self) -> None:
|
|
payload = {
|
|
"prediction_histories": [
|
|
{
|
|
"history_id": "history-1",
|
|
"revisions": [{"revision_reason": "학습자 입력"}],
|
|
"lock": {"lock_id": "lock-1"},
|
|
"external_observation": {"status": "passed"},
|
|
}
|
|
],
|
|
"calibration_assessments": [{"assessment": "ai"}],
|
|
"transfer_suites": [{"suite": "ai"}],
|
|
"teacher_reviews": [{"review": "teacher"}],
|
|
"actual_executions": [{"execution": "ai"}],
|
|
"actual_transfer_assessments": [{"assessment": "ai"}],
|
|
}
|
|
|
|
redacted = calibration_transfer._learner_input_only_payload(payload)
|
|
|
|
self.assertEqual(
|
|
redacted["prediction_histories"][0]["revisions"],
|
|
[{"revision_reason": "학습자 입력"}],
|
|
)
|
|
self.assertEqual(
|
|
redacted["prediction_histories"][0]["lock"],
|
|
{"lock_id": "lock-1"},
|
|
)
|
|
self.assertIsNone(
|
|
redacted["prediction_histories"][0]["external_observation"]
|
|
)
|
|
for key in (
|
|
"calibration_assessments",
|
|
"transfer_suites",
|
|
"teacher_reviews",
|
|
"actual_executions",
|
|
"actual_transfer_assessments",
|
|
):
|
|
self.assertEqual(redacted[key], [])
|
|
|
|
|
|
class FeedbackPolicyRouteTest(unittest.IsolatedAsyncioTestCase):
|
|
@staticmethod
|
|
def _principal(*, enabled: bool, role: Role = Role.LEARNER) -> Principal:
|
|
return Principal(
|
|
user_id="00000000-0000-0000-0000-000000000401",
|
|
role=role,
|
|
learner_feedback_enabled=enabled,
|
|
)
|
|
|
|
async def test_session_derived_routes_return_403_before_store_read(self) -> None:
|
|
principal = self._principal(enabled=True)
|
|
sess = _session(learner_feedback_enabled=False)
|
|
session_id = UUID(sess.session_id)
|
|
|
|
with (
|
|
patch.object(
|
|
session_persistence,
|
|
"load_session",
|
|
AsyncMock(return_value=sess),
|
|
),
|
|
patch.object(
|
|
rupture_repairs.rupture_repair_store,
|
|
"read_rupture_repairs",
|
|
AsyncMock(),
|
|
) as rupture_read,
|
|
patch.object(
|
|
outcome_trajectories.outcome_trajectory_store,
|
|
"read_outcome_trajectory",
|
|
AsyncMock(),
|
|
) as outcome_read,
|
|
):
|
|
for call in (
|
|
lambda: rupture_repairs.get_rupture_repairs(session_id, principal),
|
|
lambda: outcome_trajectories.get_outcome_trajectory(
|
|
session_id,
|
|
principal,
|
|
),
|
|
lambda: outcome_trajectories.recompute_outcome_trajectory(
|
|
session_id,
|
|
outcome_trajectories.OutcomeTrajectoryRecomputeRequest(),
|
|
principal,
|
|
),
|
|
):
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await call()
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
self.assertEqual(
|
|
caught.exception.detail,
|
|
feedback_policy.LEARNER_FEEDBACK_DISABLED_DETAIL,
|
|
)
|
|
|
|
rupture_read.assert_not_awaited()
|
|
outcome_read.assert_not_awaited()
|
|
|
|
async def test_current_off_redacts_past_enabled_review_without_eval_read(self) -> None:
|
|
principal = self._principal(enabled=False)
|
|
sess = _session(learner_feedback_enabled=True)
|
|
with (
|
|
patch.object(
|
|
sessions,
|
|
"_load_review_session_or_404",
|
|
AsyncMock(return_value=(sess, principal)),
|
|
),
|
|
patch.object(
|
|
sessions.session_evaluation_repository,
|
|
"load_session_evaluation",
|
|
AsyncMock(),
|
|
) as evaluation_read,
|
|
patch.object(
|
|
sessions.session_persistence,
|
|
"load_case_worksheet",
|
|
AsyncMock(return_value=(None, True)),
|
|
),
|
|
):
|
|
review = await sessions.get_session_review(sess.session_id, principal)
|
|
|
|
evaluation_read.assert_not_awaited()
|
|
self.assertFalse(review.learnerFeedbackEnabled)
|
|
self.assertFalse(review.reviewReady)
|
|
self.assertEqual([turn.text for turn in review.turns], ["상담자 발화", "가상 내담자 응답"])
|
|
self.assertTrue(all(turn.note is None for turn in review.turns))
|
|
|
|
async def test_current_off_marks_past_enabled_session_detail_input_only(self) -> None:
|
|
principal = self._principal(enabled=False)
|
|
sess = _session(learner_feedback_enabled=True)
|
|
with (
|
|
patch.object(
|
|
sessions,
|
|
"_load_session_or_404",
|
|
AsyncMock(return_value=sess),
|
|
),
|
|
patch.object(
|
|
sessions.session_evaluation_repository,
|
|
"load_session_evaluation",
|
|
AsyncMock(),
|
|
) as evaluation_read,
|
|
):
|
|
detail = await sessions.get_session_detail(sess.session_id, principal)
|
|
|
|
evaluation_read.assert_not_awaited()
|
|
self.assertFalse(detail.learner_feedback_enabled)
|
|
self.assertFalse(detail.review_ready)
|
|
self.assertEqual([turn.text for turn in detail.turns], ["상담자 발화", "가상 내담자 응답"])
|
|
|
|
async def test_deliberate_practice_read_returns_403_before_store_read(self) -> None:
|
|
principal = self._principal(enabled=False)
|
|
with patch.object(
|
|
deliberate_practices.deliberate_practice_store,
|
|
"read_deliberate_practice",
|
|
AsyncMock(),
|
|
) as read:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await deliberate_practices.get_my_deliberate_practice(principal)
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
self.assertEqual(
|
|
caught.exception.detail,
|
|
feedback_policy.LEARNER_FEEDBACK_DISABLED_DETAIL,
|
|
)
|
|
read.assert_not_awaited()
|
|
|
|
async def test_deliberate_practice_read_rejects_disabled_source_snapshot(self) -> None:
|
|
principal = self._principal(enabled=True)
|
|
with patch.object(
|
|
deliberate_practices.deliberate_practice_store,
|
|
"read_deliberate_practice",
|
|
AsyncMock(
|
|
side_effect=deliberate_practice_store.DeliberatePracticeFeedbackDisabledError(
|
|
"source session feedback disabled"
|
|
)
|
|
),
|
|
) as read:
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await deliberate_practices.get_my_deliberate_practice(principal)
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
self.assertEqual(
|
|
caught.exception.detail,
|
|
feedback_policy.LEARNER_FEEDBACK_DISABLED_DETAIL,
|
|
)
|
|
read.assert_awaited_once_with(principal=principal)
|
|
|
|
async def test_calibration_read_redacts_derived_fields_at_route_boundary(self) -> None:
|
|
principal = self._principal(enabled=False)
|
|
payload = {
|
|
"learner_id": UUID(principal.user_id),
|
|
"requested_view": "learner",
|
|
"clinical_claim_allowed": False,
|
|
"prediction_histories": [],
|
|
"calibration_assessments": [{"private": "ai"}],
|
|
"transfer_suites": [{"private": "ai"}],
|
|
"teacher_reviews": [{"private": "teacher"}],
|
|
"actual_executions": [{"private": "ai"}],
|
|
"actual_transfer_assessments": [{"private": "ai"}],
|
|
}
|
|
with patch.object(
|
|
calibration_transfer.calibration_transfer_store,
|
|
"read_calibration_transfer",
|
|
AsyncMock(return_value=payload),
|
|
) as read:
|
|
response = await calibration_transfer.get_my_calibration_transfer(
|
|
principal,
|
|
)
|
|
|
|
read.assert_awaited_once()
|
|
self.assertEqual(response.prediction_histories, [])
|
|
self.assertEqual(response.calibration_assessments, [])
|
|
self.assertEqual(response.transfer_suites, [])
|
|
self.assertEqual(response.teacher_reviews, [])
|
|
self.assertEqual(response.actual_executions, [])
|
|
self.assertEqual(response.actual_transfer_assessments, [])
|
|
|
|
async def test_calibration_aggregate_current_on_redacts_past_disabled_snapshot(
|
|
self,
|
|
) -> None:
|
|
principal = self._principal(enabled=True)
|
|
payload = {
|
|
"learner_id": UUID(principal.user_id),
|
|
"requested_view": "learner",
|
|
"clinical_claim_allowed": False,
|
|
"_learner_feedback_snapshot_enabled": False,
|
|
"prediction_histories": [],
|
|
"calibration_assessments": [{"private": "past-session-ai"}],
|
|
"transfer_suites": [{"private": "past-session-ai"}],
|
|
"teacher_reviews": [{"private": "teacher"}],
|
|
"actual_executions": [{"private": "past-session-ai"}],
|
|
"actual_transfer_assessments": [{"private": "past-session-ai"}],
|
|
}
|
|
with patch.object(
|
|
calibration_transfer.calibration_transfer_store,
|
|
"read_calibration_transfer",
|
|
AsyncMock(return_value=payload),
|
|
) as read:
|
|
response = await calibration_transfer.get_my_calibration_transfer(
|
|
principal,
|
|
)
|
|
|
|
read.assert_awaited_once_with(principal=principal)
|
|
self.assertEqual(response.prediction_histories, [])
|
|
self.assertEqual(response.calibration_assessments, [])
|
|
self.assertEqual(response.transfer_suites, [])
|
|
self.assertEqual(response.teacher_reviews, [])
|
|
self.assertEqual(response.actual_executions, [])
|
|
self.assertEqual(response.actual_transfer_assessments, [])
|
|
|
|
async def test_actual_execution_input_is_saved_but_ai_result_is_redacted(self) -> None:
|
|
principal = self._principal(enabled=False)
|
|
body = calibration_transfer.ActualTransferExecutionRequest(
|
|
original_transfer_trial_record_id=UUID(
|
|
"00000000-0000-4000-8000-000000000411"
|
|
),
|
|
practice_session_id=UUID("00000000-0000-4000-8000-000000000412"),
|
|
)
|
|
with patch.object(
|
|
calibration_transfer.calibration_transfer_store,
|
|
"append_actual_transfer_execution",
|
|
AsyncMock(
|
|
return_value={
|
|
"execution": {"private": "ai"},
|
|
"assessment": {"private": "ai"},
|
|
"idempotent_replay": False,
|
|
}
|
|
),
|
|
) as append:
|
|
response = await calibration_transfer.create_actual_transfer_execution(
|
|
body,
|
|
principal,
|
|
)
|
|
|
|
append.assert_awaited_once_with(
|
|
principal=principal,
|
|
**body.model_dump(),
|
|
)
|
|
self.assertIsNone(response.execution)
|
|
self.assertIsNone(response.assessment)
|
|
self.assertFalse(response.idempotent_replay)
|
|
|
|
async def test_multimodal_derived_read_is_403_but_privacy_read_is_redacted(self) -> None:
|
|
principal = self._principal(enabled=True)
|
|
sess = _session(learner_feedback_enabled=False)
|
|
session_id = UUID(sess.session_id)
|
|
payload = {
|
|
"session_id": session_id,
|
|
"learner_id": UUID(principal.user_id),
|
|
"clinical_claim_allowed": False,
|
|
"consent_snapshots": [{"consent_status": "granted"}],
|
|
"timelines": [{"timeline_id": "derived"}],
|
|
"word_timestamps": [{"word_index": 1}],
|
|
"voice_events": [{"event_id": "derived"}],
|
|
"measurements": [{"axis": "bond"}],
|
|
"fusion_decisions": [{"axis": "bond"}],
|
|
"deletion_requests": [{"status": "requested"}],
|
|
}
|
|
|
|
with (
|
|
patch.object(
|
|
session_persistence,
|
|
"load_session",
|
|
AsyncMock(return_value=sess),
|
|
),
|
|
patch.object(
|
|
multimodal_alliance.multimodal_alliance_store,
|
|
"read_session_metadata",
|
|
AsyncMock(return_value=payload),
|
|
) as read,
|
|
):
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await multimodal_alliance.get_multimodal_session_metadata(
|
|
session_id,
|
|
principal,
|
|
True,
|
|
)
|
|
privacy = await multimodal_alliance.get_multimodal_session_metadata(
|
|
session_id,
|
|
principal,
|
|
False,
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
self.assertEqual(caught.exception.detail, "learner_feedback_disabled")
|
|
read.assert_awaited_once()
|
|
self.assertEqual(privacy.consent_snapshots, payload["consent_snapshots"])
|
|
self.assertEqual(privacy.deletion_requests, payload["deletion_requests"])
|
|
self.assertEqual(privacy.timelines, [])
|
|
self.assertEqual(privacy.word_timestamps, [])
|
|
self.assertEqual(privacy.voice_events, [])
|
|
self.assertEqual(privacy.measurements, [])
|
|
self.assertEqual(privacy.fusion_decisions, [])
|
|
|
|
async def test_alliance_read_keeps_self_scores_but_redacts_measurements(self) -> None:
|
|
principal = self._principal(enabled=False)
|
|
session_id = UUID("00000000-0000-4000-8000-000000000421")
|
|
payload = {
|
|
"pulse_id": UUID("00000000-0000-4000-8000-000000000422"),
|
|
"checkpoint": "post",
|
|
"status": "ready",
|
|
"learner_locked_at": datetime.now(timezone.utc),
|
|
"revealed_at": datetime.now(timezone.utc),
|
|
"error_code": None,
|
|
"self_scores": {"goal": 0.8, "task": 0.6, "bond": 1.0},
|
|
"measurements": [{"private": "ai rationale and evidence"}],
|
|
}
|
|
with patch.object(
|
|
measurements.alliance_measurement,
|
|
"list_alliance_pulses",
|
|
AsyncMock(return_value=[payload]),
|
|
) as read:
|
|
response = await measurements.get_alliance_pulses(
|
|
session_id,
|
|
principal,
|
|
)
|
|
|
|
read.assert_awaited_once_with(principal=principal, session_id=session_id)
|
|
self.assertEqual(response.items[0].self_scores.model_dump(), payload["self_scores"])
|
|
self.assertEqual(response.items[0].measurements, [])
|
|
|
|
async def test_teacher_multimodal_view_keeps_derived_output(self) -> None:
|
|
principal = self._principal(enabled=False, role=Role.TEACHER)
|
|
session_id = UUID("00000000-0000-4000-8000-000000000401")
|
|
payload = {
|
|
"session_id": session_id,
|
|
"learner_id": UUID("00000000-0000-0000-0000-000000000401"),
|
|
"clinical_claim_allowed": False,
|
|
"consent_snapshots": [],
|
|
"timelines": [{"timeline_id": "teacher-visible"}],
|
|
"word_timestamps": [],
|
|
"voice_events": [],
|
|
"measurements": [],
|
|
"fusion_decisions": [],
|
|
"deletion_requests": [],
|
|
}
|
|
with patch.object(
|
|
multimodal_alliance.multimodal_alliance_store,
|
|
"read_session_metadata",
|
|
AsyncMock(return_value=payload),
|
|
):
|
|
response = await multimodal_alliance.get_multimodal_session_metadata(
|
|
session_id,
|
|
principal,
|
|
True,
|
|
)
|
|
|
|
self.assertEqual(response.timelines, payload["timelines"])
|
|
|
|
async def test_dashboard_keeps_session_counts_but_hides_feedback_metrics(self) -> None:
|
|
principal = Principal(
|
|
user_id="00000000-0000-0000-0000-000000000401",
|
|
role=Role.LEARNER,
|
|
learner_feedback_enabled=False,
|
|
)
|
|
sess = _session(learner_feedback_enabled=True)
|
|
|
|
with (
|
|
patch.object(
|
|
sessions.session_persistence,
|
|
"list_recent_sessions",
|
|
AsyncMock(return_value=([sess], True)),
|
|
),
|
|
patch.object(
|
|
sessions.session_evaluation_repository,
|
|
"load_session_evaluation",
|
|
AsyncMock(return_value=({"status": "ready"}, True)),
|
|
),
|
|
patch.object(
|
|
sessions.session_persistence,
|
|
"list_session_archives",
|
|
AsyncMock(return_value=({}, True)),
|
|
),
|
|
):
|
|
dashboard = await sessions.learner_dashboard(principal)
|
|
|
|
self.assertEqual(dashboard.overview.total_sessions, 1)
|
|
self.assertEqual(dashboard.overview.completed_sessions, 1)
|
|
self.assertEqual(dashboard.overview.review_ready_sessions, 0)
|
|
self.assertEqual(dashboard.growth.evaluated_sessions, 0)
|
|
self.assertEqual(dashboard.recent_feedback, [])
|
|
self.assertEqual(dashboard.persona_progress[0].sessions, 1)
|
|
self.assertIsNone(dashboard.persona_progress[0].latest_score)
|
|
|
|
async def test_live_coach_is_blocked_before_generation(self) -> None:
|
|
principal = Principal(
|
|
user_id="00000000-0000-0000-0000-000000000401",
|
|
role=Role.LEARNER,
|
|
learner_feedback_enabled=False,
|
|
)
|
|
sess = _session(learner_feedback_enabled=True)
|
|
|
|
with (
|
|
patch.object(
|
|
sessions,
|
|
"_load_session_or_404",
|
|
AsyncMock(return_value=sess),
|
|
),
|
|
patch.object(
|
|
sessions.live_coach,
|
|
"generate_live_coaching",
|
|
AsyncMock(),
|
|
) as generate,
|
|
):
|
|
with self.assertRaises(HTTPException) as caught:
|
|
await sessions.live_coach_turn(
|
|
sess.session_id,
|
|
sessions.LiveCoachRequest(learner_text="도와줘"),
|
|
principal,
|
|
)
|
|
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
self.assertEqual(
|
|
caught.exception.detail,
|
|
feedback_policy.LEARNER_FEEDBACK_DISABLED_DETAIL,
|
|
)
|
|
generate.assert_not_awaited()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|