"""Exercise the complete G2 five-session outcome-trajectory HTTP flow. The smoke intentionally retains its unique dev/E2E fixture. It proves case continuity, three independent outcome axes, retry idempotency, cross-role and cross-cohort access boundaries, role-safe relationship memory, and the non-clinical synthetic-arc notice without printing credentials, cookies, e-mail addresses, or transcript text. """ from __future__ import annotations import argparse import json import secrets import time import urllib.error import urllib.request from dataclasses import dataclass from http.cookiejar import CookieJar from pathlib import Path from typing import Any from uuid import uuid4 AXES = ("distress_load", "daily_functioning", "learning_engagement") EXPECTED_SCORES = ( {"distress_load": 0.70, "daily_functioning": 0.30, "learning_engagement": 0.40}, {"distress_load": 0.62, "daily_functioning": 0.40, "learning_engagement": 0.48}, {"distress_load": 0.54, "daily_functioning": 0.50, "learning_engagement": 0.56}, {"distress_load": 0.46, "daily_functioning": 0.60, "learning_engagement": 0.64}, {"distress_load": 0.38, "daily_functioning": 0.68, "learning_engagement": 0.72}, ) COHORT_ID = "e2e-hanshin" class SmokeError(RuntimeError): pass @dataclass(frozen=True) class ApiResponse: status: int body: Any class ApiClient: def __init__(self, base_url: str, timeout: float) -> None: self.base_url = base_url.rstrip("/") self.timeout = timeout self._opener = urllib.request.build_opener( urllib.request.HTTPCookieProcessor(CookieJar()) ) def request( self, method: str, path: str, payload: dict[str, Any] | None = None, *, expected: set[int] | None = None, ) -> ApiResponse: data = None headers = {"Accept": "application/json"} if payload is not None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json" request = urllib.request.Request( f"{self.base_url}{path}", data=data, headers=headers, method=method ) try: with self._opener.open(request, timeout=self.timeout) as response: raw = response.read().decode("utf-8") result = ApiResponse( status=response.status, body=json.loads(raw) if raw else {}, ) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") try: body = json.loads(raw) if raw else {} except json.JSONDecodeError: body = {"detail": raw[:500]} result = ApiResponse(status=exc.code, body=body) except urllib.error.URLError as exc: raise SmokeError( f"{method} {path} transport failed: {type(exc.reason).__name__}" ) from exc allowed = expected or {200} if result.status not in allowed: detail = ( result.body.get("detail") if isinstance(result.body, dict) else None ) raise SmokeError( f"{method} {path} returned HTTP {result.status}; detail={detail!r}" ) return result def _onboarding_payload(display_name: str) -> dict[str, Any]: return { "legal_name": display_name, "affiliation": "한신대학교", "department": "상담심리학과", "grade_level": "통합검증", "phone": "010-0000-0000", "contact_address": "경기도 오산시 한신대학교", "nickname": display_name, "self_introduction": "G2 5회기 성과 궤적 API를 검증하는 개발 fixture입니다.", "avatar_url": "", "terms_accepted": True, "privacy_accepted": True, } def _sign_in( client: ApiClient, *, suffix: str, identity: str, role: str, cohort_ids: list[str], ) -> None: login = client.request( "POST", "/auth/dev-login", { "email": f"dev.e2e.outcome.{identity}.{suffix}@hs.ac.kr", "role": role, "display_name": f"Outcome {identity.title()}", "cohort_ids": cohort_ids, }, ) if login.body.get("role") != role: raise SmokeError(f"dev-login role mismatch for {identity}") client.request( "POST", "/users/me/onboarding", _onboarding_payload(f"Outcome {identity.title()}"), ) def _choose_persona(client: ApiClient) -> str: response = client.request("GET", "/personas") if not isinstance(response.body, list): raise SmokeError("persona catalog did not return a list") usable = [ item for item in response.body if isinstance(item, dict) and item.get("source") == "database" and not item.get("degraded") and item.get("code") ] if not usable: raise SmokeError("persona catalog has no non-degraded database persona") preferred = next((item for item in usable if item.get("code") == "P1"), usable[0]) return str(preferred["code"]) def _assert_no_total_score(value: Any, path: str = "response") -> None: if isinstance(value, dict): forbidden = {"total", "total_score", "overall_score"} & set(value) if forbidden: raise SmokeError(f"{path} exposed forbidden total score keys: {forbidden}") for key, child in value.items(): _assert_no_total_score(child, f"{path}.{key}") elif isinstance(value, list): for index, child in enumerate(value): _assert_no_total_score(child, f"{path}[{index}]") def _assert_trajectory(payload: dict[str, Any], *, session_count: int) -> None: expected_arc = payload.get("expected_arc") or {} assessment = payload.get("assessment") or {} sessions = assessment.get("sessions") or [] observations = payload.get("observations") or [] if expected_arc.get("clinical_claim_allowed") is not False: raise SmokeError("synthetic expected arc did not prohibit clinical claims") if expected_arc.get("data_classification") != "synthetic_educational": raise SmokeError("expected arc omitted synthetic educational classification") if len(expected_arc.get("distributions") or []) != 15: raise SmokeError("expected arc is not 5 sessions x 3 independent axes") if assessment.get("clinical_claim_allowed") is not False: raise SmokeError("assessment did not prohibit clinical claims") if len(sessions) != session_count: raise SmokeError( f"trajectory contains {len(sessions)} sessions, expected {session_count}" ) if len(observations) != session_count * 3: raise SmokeError( f"trajectory contains {len(observations)} observations, " f"expected {session_count * 3}" ) for expected_no, item in enumerate(sessions, start=1): if item.get("session_no") != expected_no or item.get("status") != "on_track": raise SmokeError(f"session trajectory differs at S{expected_no}: {item}") axes = item.get("axes") or [] if len(axes) != 3 or {axis.get("axis") for axis in axes} != set(AXES): raise SmokeError(f"S{expected_no} did not preserve three independent axes") _assert_no_total_score(payload) def _first_durable_turn_id(review: dict[str, Any]) -> str: turns = review.get("turns") or [] durable_ids = [str(item.get("turn_id")) for item in turns if item.get("turn_id")] if not durable_ids: raise SmokeError("session review omitted durable turn UUID evidence") return durable_ids[0] def run(args: argparse.Namespace) -> dict[str, Any]: health_client = ApiClient(args.api_base_url, args.request_timeout) health = health_client.request("GET", "/health") if not health.body.get("db") or not health.body.get("engine"): raise SmokeError("API health is not DB+engine ready") suffix = f"{int(time.time())}.{secrets.token_hex(3)}" learner = ApiClient(args.api_base_url, args.request_timeout) teacher = ApiClient(args.api_base_url, args.request_timeout) other_learner = ApiClient(args.api_base_url, args.request_timeout) other_teacher = ApiClient(args.api_base_url, args.request_timeout) _sign_in( learner, suffix=suffix, identity="learner", role="learner", cohort_ids=[COHORT_ID], ) _sign_in( teacher, suffix=suffix, identity="teacher", role="teacher", cohort_ids=[COHORT_ID], ) _sign_in( other_learner, suffix=suffix, identity="other-learner", role="learner", cohort_ids=[COHORT_ID], ) _sign_in( other_teacher, suffix=suffix, identity="other-teacher", role="teacher", cohort_ids=["e2e-other-cohort"], ) persona_code = _choose_persona(learner) case_id = "" session_ids: list[str] = [] submission_ids: list[str] = [] measurement_ids: list[list[str]] = [] first_turn_id = "" final_payload: dict[str, Any] = {} first_retry_stable = False conflict_rejected = False for session_no, scores in enumerate(EXPECTED_SCORES, start=1): started = learner.request( "POST", "/sessions", { "persona_code": persona_code, "theory_mode": "humanistic", "goal_stages": ["라포", "탐색"], }, expected={201}, ) if started.body.get("degraded"): raise SmokeError(f"S{session_no} start was degraded") if started.body.get("session_no") != session_no: raise SmokeError( f"session continuity differs: {started.body.get('session_no')} != {session_no}" ) current_case_id = str(started.body.get("case_id") or "") if not current_case_id: raise SmokeError("session start omitted case_id") if case_id and current_case_id != case_id: raise SmokeError("five sessions did not remain in one case") case_id = current_case_id session_id = str(started.body.get("session_id") or "") session_ids.append(session_id) if session_no == 1: turn = learner.request( "POST", f"/sessions/{session_id}/turn", {"text": "오늘 확인할 목표를 함께 정해도 괜찮을까요?"}, ) if not turn.body.get("client_reply"): raise SmokeError("first-session turn omitted client reply") learner.request("POST", f"/sessions/{session_id}/end") if session_no == 1: review = learner.request("GET", f"/sessions/{session_id}/review") first_turn_id = _first_durable_turn_id(review.body) submission_id = str(uuid4()) submission_ids.append(submission_id) submission = { "submission_id": submission_id, "scores": scores, "confidences": {axis: 0.9 for axis in AXES}, "evidence_turn_ids": [first_turn_id] if session_no == 1 else [], } created = learner.request( "POST", f"/sessions/{session_id}/outcome-observations", submission, expected={201}, ) _assert_trajectory(created.body, session_count=session_no) ids = [ str(item) for item in created.body.get("submitted_measurement_ids") or [] ] if len(ids) != 3: raise SmokeError(f"S{session_no} did not return three measurement IDs") measurement_ids.append(ids) final_payload = created.body if session_no == 1: retried = learner.request( "POST", f"/sessions/{session_id}/outcome-observations", submission, expected={201}, ) first_retry_stable = retried.body.get("submitted_measurement_ids") == ids if not first_retry_stable: raise SmokeError("same submission retry changed measurement IDs") changed = dict(submission) changed["scores"] = dict(scores, distress_load=0.99) conflict = learner.request( "POST", f"/sessions/{session_id}/outcome-observations", changed, expected={409}, ) conflict_rejected = conflict.status == 409 final_session_id = session_ids[-1] relationship = teacher.request( "POST", f"/sessions/{session_ids[0]}/relationship-memory-events", { "event_type": "goal_agreement", "summaries": { "counselor": "첫 회기의 학습 목표를 내담자와 명시적으로 합의했다.", "supervisor": "목표 합의 발화가 실제 축어록 근거로 확인됐다.", }, "evidence_turn_ids": [first_turn_id], }, expected={201}, ) if relationship.body.get("status") != "recorded": raise SmokeError("relationship memory was not recorded") teacher_read = teacher.request( "GET", f"/sessions/{final_session_id}/outcome-trajectory" ) _assert_trajectory(teacher_read.body, session_count=5) relationship_memory = teacher_read.body.get("relationship_memory") or [] if len(relationship_memory) != 1: raise SmokeError("teacher trajectory omitted role-safe relationship memory") if set(relationship_memory[0]) & {"summaries", "client", "evaluator"}: raise SmokeError( "teacher relationship projection leaked another role's summary" ) learner_read = learner.request( "GET", f"/sessions/{final_session_id}/outcome-trajectory" ) _assert_trajectory(learner_read.body, session_count=5) if len(learner_read.body.get("relationship_memory") or []) != 1: raise SmokeError("learner trajectory omitted counselor relationship memory") other_learner.request( "GET", f"/sessions/{final_session_id}/outcome-trajectory", expected={404} ) other_teacher.request( "GET", f"/sessions/{final_session_id}/outcome-trajectory", expected={404} ) return { "ok": True, "api_base_url": args.api_base_url, "health": { "status": health.body.get("status"), "db": health.body.get("db"), "engine": health.body.get("engine"), "engine_mode": health.body.get("engine_mode"), }, "fixture_policy": "retained unique dev:e2e identities; no fixture deletion", "cohort_id": COHORT_ID, "persona_code": persona_code, "case_id": case_id, "session_ids": session_ids, "submission_ids": submission_ids, "proof": { "session_numbers": [1, 2, 3, 4, 5], "same_case_across_sessions": True, "durable_review_turn_uuid_exposed": bool(first_turn_id), "observed_axis_count": len(final_payload.get("observations") or []), "session_statuses": [ item.get("status") for item in (final_payload.get("assessment") or {}).get("sessions", []) ], "three_axes_without_total_score": True, "synthetic_non_clinical_notice": True, "same_submission_measurement_ids_stable": first_retry_stable, "changed_payload_same_submission_rejected": conflict_rejected, "measurement_id_count": sum(len(items) for items in measurement_ids), "teacher_cohort_read": True, "other_learner_rejected": True, "cross_cohort_teacher_rejected": True, "role_safe_relationship_memory_count": len(relationship_memory), }, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--api-base-url", default="http://127.0.0.1:8005") parser.add_argument("--request-timeout", type=float, default=180.0) parser.add_argument("--out", default="") args = parser.parse_args() result = run(args) text = json.dumps(result, ensure_ascii=False, indent=2) if args.out: out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(text + "\n", encoding="utf-8") print(text) if __name__ == "__main__": main()