222 lines
8.5 KiB
Python
222 lines
8.5 KiB
Python
"""probe-jev-dialogue.py의 외부 연결 없는 측정 계약 단위 검증."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import io
|
|
import unittest
|
|
from contextlib import redirect_stderr
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
|
|
SCRIPT_PATH = Path(__file__).with_name("probe-jev-dialogue.py")
|
|
SPEC = importlib.util.spec_from_file_location("probe_jev_dialogue", SCRIPT_PATH)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
probe = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(probe)
|
|
|
|
|
|
class ProbeJevDialogueTest(unittest.TestCase):
|
|
def _cleanup_runtime(self, response: object, *, live_provider: str = "openai", engine_mode: str = "remote"):
|
|
requests: list[str] = []
|
|
|
|
class Client:
|
|
async def delete(self, path: str) -> object:
|
|
requests.append(path)
|
|
return response
|
|
|
|
return {
|
|
"settings": SimpleNamespace(
|
|
live_client_provider=live_provider,
|
|
engine_mode=engine_mode,
|
|
),
|
|
"engine_client": SimpleNamespace(client=Client()),
|
|
}, requests
|
|
|
|
@staticmethod
|
|
def _phase(provider: str) -> dict[str, object]:
|
|
return {
|
|
"status": "done",
|
|
"turns": [
|
|
{
|
|
"status": "done",
|
|
"generation": {"provider": provider},
|
|
}
|
|
],
|
|
"session_closed": False,
|
|
}
|
|
|
|
def test_parser_validates_phase_and_repeat_contract(self) -> None:
|
|
args = probe.build_parser().parse_args(
|
|
["--output", "report.json", "--phases", "jev,legacy", "--repeats", "3", "--label", "trial-a"]
|
|
)
|
|
|
|
self.assertEqual(args.phases, ("jev", "legacy"))
|
|
self.assertEqual(args.repeats, 3)
|
|
self.assertEqual(args.label, "trial-a")
|
|
with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
probe.build_parser().parse_args(["--output", "report.json", "--phases", "legacy,invalid"])
|
|
with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
probe.build_parser().parse_args(["--output", "report.json", "--repeats", "6"])
|
|
|
|
def test_repeat_order_alternates_and_fixture_preserves_recall_expectations(self) -> None:
|
|
phases = ("legacy", "jev")
|
|
|
|
self.assertEqual(probe._phase_order(phases, 1), ("legacy", "jev"))
|
|
self.assertEqual(probe._phase_order(phases, 2), ("jev", "legacy"))
|
|
self.assertEqual(probe.SYNTHETIC_UTTERANCES[0]["id"], "empathy")
|
|
self.assertEqual(probe.SYNTHETIC_UTTERANCES[1]["id"], "contradictory_recall")
|
|
self.assertEqual(probe.SYNTHETIC_UTTERANCES[1]["label"], "unsupported_recall")
|
|
self.assertEqual(probe.SYNTHETIC_UTTERANCES[2]["id"], "ambivalence")
|
|
self.assertEqual(probe.SYNTHETIC_MEMORY["pinned_facts"], ["나는 외동이다."])
|
|
self.assertIn("과거 사실로 확정하지 않음", probe.SYNTHETIC_MEMORY["expectations"]["unsupported_recall"])
|
|
self.assertIn("형제자매 존재에 동의하지 않음", probe.SYNTHETIC_MEMORY["expectations"]["contradictory_recall"])
|
|
|
|
def test_cold_and_warm_summaries_use_turn_order_not_repeat_order(self) -> None:
|
|
phases = [
|
|
{
|
|
"provider_mode": "legacy",
|
|
"status": "done",
|
|
"repeat": 2,
|
|
"order_index": 1,
|
|
"turns": [
|
|
{"status": "done", "turn_temperature": "cold", "ttft_ms": 10.0, "total_ms": 30.0},
|
|
{"status": "done", "turn_temperature": "warm", "ttft_ms": 8.0, "total_ms": 24.0},
|
|
],
|
|
},
|
|
]
|
|
|
|
summaries = probe._phase_summaries(phases)
|
|
|
|
self.assertEqual(
|
|
summaries,
|
|
[
|
|
{
|
|
"provider_mode": "legacy",
|
|
"turn_temperature": "cold",
|
|
"phase_run_count": 1,
|
|
"completed_phase_run_count": 1,
|
|
"attempted_turn_count": 1,
|
|
"completed_turn_count": 1,
|
|
"ttft_ms": {"sample_count": 1, "p50": 10.0, "p95": 10.0},
|
|
"total_ms": {"sample_count": 1, "p50": 30.0, "p95": 30.0},
|
|
},
|
|
{
|
|
"provider_mode": "legacy",
|
|
"turn_temperature": "warm",
|
|
"phase_run_count": 1,
|
|
"completed_phase_run_count": 1,
|
|
"attempted_turn_count": 1,
|
|
"completed_turn_count": 1,
|
|
"ttft_ms": {"sample_count": 1, "p50": 8.0, "p95": 8.0},
|
|
"total_ms": {"sample_count": 1, "p50": 24.0, "p95": 24.0},
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_openai_stateless_cleanup_allows_http_200_closed_false(self) -> None:
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict[str, bool]:
|
|
return {"closed": False}
|
|
|
|
response = Response()
|
|
runtime, requests = self._cleanup_runtime(response)
|
|
phase = self._phase("openai")
|
|
|
|
asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-1"))
|
|
|
|
self.assertEqual(requests, ["/session/session-1"])
|
|
self.assertEqual(phase["status"], "done")
|
|
self.assertFalse(phase["session_closed"])
|
|
self.assertEqual(phase["cleanup_http_status"], 200)
|
|
self.assertFalse(phase["cleanup_closed"])
|
|
self.assertEqual(phase["cleanup_status"], "not_required")
|
|
self.assertNotIn("cleanup_error", phase)
|
|
|
|
def test_non_openai_cleanup_closed_false_is_failure(self) -> None:
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict[str, bool]:
|
|
return {"closed": False}
|
|
|
|
response = Response()
|
|
runtime, requests = self._cleanup_runtime(response)
|
|
phase = self._phase("claude")
|
|
|
|
asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-2"))
|
|
|
|
self.assertEqual(requests, ["/session/session-2"])
|
|
self.assertEqual(phase["status"], "error")
|
|
self.assertFalse(phase["session_closed"])
|
|
self.assertEqual(phase["cleanup_http_status"], 200)
|
|
self.assertFalse(phase["cleanup_closed"])
|
|
self.assertEqual(phase["cleanup_status"], "failed")
|
|
self.assertEqual(phase["cleanup_error"], "engine_close_failed")
|
|
|
|
def test_openai_cleanup_http_failure_is_failure(self) -> None:
|
|
class Response:
|
|
status_code = 503
|
|
|
|
def __init__(self) -> None:
|
|
self.json_called = False
|
|
|
|
def raise_for_status(self) -> None:
|
|
raise RuntimeError("http failure")
|
|
|
|
def json(self) -> dict[str, bool]:
|
|
self.json_called = True
|
|
return {"closed": False}
|
|
|
|
response = Response()
|
|
runtime, requests = self._cleanup_runtime(response)
|
|
phase = self._phase("openai")
|
|
|
|
asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-3"))
|
|
|
|
self.assertEqual(requests, ["/session/session-3"])
|
|
self.assertEqual(phase["status"], "error")
|
|
self.assertFalse(phase["session_closed"])
|
|
self.assertEqual(phase["cleanup_http_status"], 503)
|
|
self.assertFalse(response.json_called)
|
|
self.assertNotIn("cleanup_closed", phase)
|
|
self.assertEqual(phase["cleanup_status"], "failed")
|
|
self.assertEqual(phase["cleanup_error"], "engine_close_failed")
|
|
|
|
def test_openai_cleanup_requires_boolean_closed_json(self) -> None:
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict[str, int]:
|
|
return {"closed": 0}
|
|
|
|
response = Response()
|
|
runtime, requests = self._cleanup_runtime(response)
|
|
phase = self._phase("openai")
|
|
|
|
asyncio.run(probe._cleanup_synthetic_session(runtime=runtime, phase=phase, session_id="session-4"))
|
|
|
|
self.assertEqual(requests, ["/session/session-4"])
|
|
self.assertEqual(phase["status"], "error")
|
|
self.assertFalse(phase["session_closed"])
|
|
self.assertEqual(phase["cleanup_http_status"], 200)
|
|
self.assertNotIn("cleanup_closed", phase)
|
|
self.assertEqual(phase["cleanup_status"], "failed")
|
|
self.assertEqual(phase["cleanup_error"], "engine_close_failed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|