ElebobaandClaude Opus 5.5
8d12aa3d08
CI / ruff · pytest (pull_request) Successful in 1m37s
- personas degrade: /demo/personas → doctor's patients (hero first) → env token - scenario file guessed from the transliterated first name (andrey.jsonl), like the backend - «Подключить демо-пациента»: connects the synthetic hero via /patients/connect when no seed - connect code prefilled from ?code= of the console link Refs #17 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
420 lines
18 KiB
Python
420 lines
18 KiB
Python
"""Demo console (#17): scenario parser, press resolver, clock math, playback runner, API client."""
|
||
|
||
import json
|
||
import sys
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
# Streamlit runs pages with `dashboard/` on sys.path and they import `console`; tests mirror that.
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "dashboard"))
|
||
|
||
from console.api import ApiError, ConsoleApi, personas_or_fallback # noqa: E402
|
||
from console.chatlog import ChatLog, normalize_reply # noqa: E402
|
||
from console.playback import ( # noqa: E402
|
||
PlaybackRunner,
|
||
PlaybackState,
|
||
hours_until,
|
||
parse_clock_now,
|
||
planned_date,
|
||
resolve_press,
|
||
target_datetime,
|
||
)
|
||
from console.scenario import ( # noqa: E402
|
||
Action,
|
||
default_scenario_path,
|
||
load_scenario,
|
||
parse_action,
|
||
parse_scenario,
|
||
persona_slug,
|
||
)
|
||
|
||
MSK = timezone(timedelta(hours=3))
|
||
|
||
|
||
def _jsonl(*rows: Any) -> str:
|
||
return "\n".join(r if isinstance(r, str) else json.dumps(r, ensure_ascii=False) for r in rows)
|
||
|
||
|
||
def _reply(text: str, *labels_data: tuple[str, str], kind: str = "normal") -> dict[str, Any]:
|
||
return {"text": text, "kind": kind, "buttons": [[{"label": lab, "data": d} for lab, d in labels_data]]}
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ scenario
|
||
def test_parse_scenario_valid_sorted_and_normalized():
|
||
sc = parse_scenario(
|
||
_jsonl(
|
||
{"day": 2, "note": "тошнота", "actions": [{"advance_to": "8:00"}, {"press": "Принял"}]},
|
||
"",
|
||
"# comment",
|
||
{"day": 0, "actions": [{"start": None}, {"text": "привет"}, {"expect_color": "GREEN"}]},
|
||
)
|
||
)
|
||
assert sc.errors == []
|
||
assert sc.day_numbers == [0, 2]
|
||
assert sc.day(2).actions == (Action("advance_to", "08:00"), Action("press", "Принял"))
|
||
assert sc.day(0).actions[0] == Action("start", "")
|
||
assert sc.day(0).actions[2] == Action("expect_color", "green")
|
||
assert sc.day(2).note == "тошнота"
|
||
|
||
|
||
def test_parse_scenario_reports_and_skips_bad_parts():
|
||
sc = parse_scenario(
|
||
_jsonl(
|
||
"{not json",
|
||
[1, 2],
|
||
{"day": -1, "actions": []},
|
||
{"day": 1, "actions": [{"advance_to": "25:00"}, {"press": "Да"}, {"fly": "x"}, {"a": 1, "b": 2}]},
|
||
{"day": 1, "actions": [{"text": "дубль"}]},
|
||
{"day": 3, "actions": "oops"},
|
||
{"day": 4, "actions": [{"expect_color": "purple"}, {"text": " "}]},
|
||
)
|
||
)
|
||
assert sc.day_numbers == [1, 4]
|
||
assert sc.day(1).actions == (Action("press", "Да"),)
|
||
assert sc.day(4).actions == ()
|
||
assert len(sc.errors) == 10
|
||
assert any("строка 1" in e and "не JSON" in e for e in sc.errors)
|
||
assert any("повторяется" in e for e in sc.errors)
|
||
assert any("ЧЧ:ММ" in e for e in sc.errors)
|
||
|
||
|
||
@pytest.mark.parametrize("raw", [{"press": 5}, "press", {"advance_to": "8"}, {}])
|
||
def test_parse_action_rejects(raw):
|
||
with pytest.raises(ValueError):
|
||
parse_action(raw)
|
||
|
||
|
||
def test_load_scenario_missing_and_present(tmp_path):
|
||
missing = load_scenario("nope.jsonl", root=tmp_path)
|
||
assert missing.days == [] and "не найден" in missing.errors[0]
|
||
(tmp_path / "s.jsonl").write_text(_jsonl({"day": 5, "actions": [{"text": "брошу"}]}), encoding="utf-8")
|
||
assert load_scenario("s.jsonl", root=tmp_path).day_numbers == [5]
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ press resolver
|
||
def test_resolve_press_newest_reply_first_and_exact_beats_substring():
|
||
replies = [
|
||
_reply("вчера", ("Принял", "med:taken:1")),
|
||
_reply("сегодня", ("Принял позже", "med:later:2"), ("Принял", "med:taken:2")),
|
||
]
|
||
assert resolve_press(replies, "принял") == ("med:taken:2", "Принял")
|
||
assert resolve_press(replies, "позже") == ("med:later:2", "Принял позже")
|
||
|
||
|
||
def test_resolve_press_window_normalization_and_miss():
|
||
old = _reply("старое", ("Ещё вопрос", "q:1"))
|
||
filler = [_reply(f"r{i}") for i in range(10)]
|
||
assert resolve_press([old, *filler], "еще") is None # outside the 10-reply window
|
||
assert resolve_press([old, *filler[:3]], " ЕЩЕ вопрос ") == ("q:1", "Ещё вопрос")
|
||
assert resolve_press([old], "нет такой") is None
|
||
assert resolve_press([old], "") is None
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ clock math
|
||
def test_target_datetime_next_occurrence_without_planned_date():
|
||
now = datetime(2026, 10, 5, 7, 0, tzinfo=MSK)
|
||
assert target_datetime(now, "08:00", None) == datetime(2026, 10, 5, 8, 0, tzinfo=MSK)
|
||
late = datetime(2026, 10, 5, 21, 0, tzinfo=MSK)
|
||
assert target_datetime(late, "08:00", None) == datetime(2026, 10, 6, 8, 0, tzinfo=MSK)
|
||
# Just reached (virtual time ticked a bit past): stays today, no rollover.
|
||
just = datetime(2026, 10, 5, 8, 2, tzinfo=MSK)
|
||
assert target_datetime(just, "08:00", None) == datetime(2026, 10, 5, 8, 0, tzinfo=MSK)
|
||
|
||
|
||
def test_target_datetime_with_planned_date_and_hours_until():
|
||
now = datetime(2026, 10, 5, 20, 30, tzinfo=MSK)
|
||
target = target_datetime(now, "08:00", date(2026, 10, 6))
|
||
assert hours_until(now, target) == pytest.approx(11.5)
|
||
assert hours_until(now, now - timedelta(minutes=3)) == 0.0
|
||
assert hours_until(now, now - timedelta(hours=2)) is None
|
||
|
||
|
||
def test_planned_date_chains_days_and_ignores_stale_dates():
|
||
dates = {0: date(2026, 10, 5), 2: date(2026, 10, 7)}
|
||
assert planned_date(dates, 5, today=date(2026, 10, 7)) == date(2026, 10, 10)
|
||
assert planned_date(dates, 2, today=date(2026, 10, 7)) == date(2026, 10, 7)
|
||
assert planned_date({}, 3, today=date(2026, 10, 7)) is None
|
||
# Presenter moved the clock manually past the chained date → fall back to "next occurrence".
|
||
assert planned_date(dates, 3, today=date(2026, 10, 12)) is None
|
||
|
||
|
||
def test_parse_clock_now():
|
||
assert parse_clock_now({"now": "2026-10-05T08:00:00+03:00"}) == datetime(2026, 10, 5, 8, 0, tzinfo=MSK)
|
||
with pytest.raises(ValueError):
|
||
parse_clock_now({})
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ chat log
|
||
def test_chatlog_normalizes_and_tracks_active_keyboard():
|
||
assert normalize_reply({"text": None, "buttons": [[{"label": "x"}], "bad"]}) == {
|
||
"dir": "out",
|
||
"text": "",
|
||
"kind": "normal",
|
||
"buttons": [],
|
||
}
|
||
chat = ChatLog()
|
||
chat.add_replies([_reply("a", ("Да", "y")), {"text": "b"}])
|
||
chat.add_in("Да", pressed=True)
|
||
assert chat.last_buttons_index() == 0
|
||
assert [r["text"] for r in chat.recent_replies(1)] == ["b"]
|
||
chat.add_history([{"direction": "in", "text": "hi"}, {"direction": "out", "text": "yo", "kind": "info"}])
|
||
assert chat.entries[-1]["kind"] == "info" and chat.entries[-2]["dir"] == "in"
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ runner
|
||
class FakeApi:
|
||
"""Demo backend double: a virtual clock with scheduled pushes and scripted dialog answers."""
|
||
|
||
def __init__(self, now: datetime):
|
||
self.now = now
|
||
self.calls: list[tuple] = []
|
||
self.scheduled: list[tuple[datetime, dict[str, Any]]] = []
|
||
self.box: list[dict[str, Any]] = []
|
||
self.color = "green"
|
||
self.fail_on: str | None = None
|
||
self.answers: dict[str, list[dict[str, Any]]] = {}
|
||
|
||
def _maybe_fail(self, what: str) -> None:
|
||
if self.fail_on == what:
|
||
raise ApiError(f"{what}: HTTP 500")
|
||
|
||
def push_at(self, hhmm: str, day: date, reply: dict[str, Any]) -> None:
|
||
h, m = map(int, hhmm.split(":"))
|
||
self.scheduled.append((datetime(day.year, day.month, day.day, h, m, tzinfo=MSK), reply))
|
||
|
||
def clock(self) -> dict[str, Any]:
|
||
self._maybe_fail("clock")
|
||
return {"now": self.now.isoformat(), "virtual": True, "offset_days": 0}
|
||
|
||
def advance(self, days: float = 0, hours: float = 0) -> dict[str, Any]:
|
||
self._maybe_fail("advance")
|
||
self.calls.append(("advance", round(hours, 3)))
|
||
self.now += timedelta(days=days, hours=hours)
|
||
due = [(t, r) for t, r in self.scheduled if t <= self.now]
|
||
self.scheduled = [(t, r) for t, r in self.scheduled if t > self.now]
|
||
self.box.extend(r for _, r in due)
|
||
return self.clock()
|
||
|
||
def send_event(self, token, type_, *, text=None, button=None, params=None):
|
||
self._maybe_fail(type_)
|
||
self.calls.append((type_, text or button or params))
|
||
return self.answers.get(text or button or type_, [{"text": "ок"}])
|
||
|
||
def outbox(self, token):
|
||
out, self.box = self.box, []
|
||
return out
|
||
|
||
def patients(self, doctor):
|
||
self._maybe_fail("patients")
|
||
return [{"token": "tok-other", "status_color": "red"}, {"token": "tok-andrey", "status_color": self.color}]
|
||
|
||
|
||
def _runner(api: FakeApi, state: PlaybackState | None = None, chat: ChatLog | None = None) -> PlaybackRunner:
|
||
sleeps: list[float] = []
|
||
r = PlaybackRunner(api, "tok-andrey", chat or ChatLog(), state or PlaybackState(), delay=0.5, sleep=sleeps.append)
|
||
r.sleeps = sleeps # type: ignore[attr-defined]
|
||
return r
|
||
|
||
|
||
SCENARIO = parse_scenario(
|
||
_jsonl(
|
||
{"day": 1, "note": "приём", "actions": [{"advance_to": "08:00"}, {"press": "Принял"}]},
|
||
{
|
||
"day": 2,
|
||
"actions": [
|
||
{"advance_to": "08:00"},
|
||
{"press": "принял"},
|
||
{"advance_to": "20:00"},
|
||
{"text": "подташнивает с утра"},
|
||
{"expect_color": "yellow"},
|
||
],
|
||
},
|
||
{"day": 5, "actions": [{"advance_to": "20:00"}, {"text": "брошу"}]},
|
||
)
|
||
)
|
||
|
||
|
||
def test_runner_plays_day_in_order_with_dynamic_button_data():
|
||
api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK))
|
||
api.push_at("08:00", date(2026, 10, 6), _reply("Пора принять сертралин", ("Принял", "med:taken:42")))
|
||
runner = _runner(api)
|
||
logs = list(runner.play_day(SCENARIO, 1))
|
||
assert api.calls == [("advance", 11.0), ("button", "med:taken:42")]
|
||
assert [log.status for log in logs] == ["info", "ok", "ok", "ok"]
|
||
assert runner.state.done == [1] and runner.state.dates[1] == date(2026, 10, 6)
|
||
assert runner.chat.entries[1] == {"dir": "in", "text": "Принял", "pressed": True}
|
||
assert runner.sleeps == [0.5] # one pause between two steps
|
||
|
||
|
||
def test_runner_chains_next_day_date_and_soft_color_check():
|
||
api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK))
|
||
api.push_at("08:00", date(2026, 10, 6), _reply("доза 1", ("Принял", "med:taken:1")))
|
||
api.push_at("08:00", date(2026, 10, 7), _reply("доза 2", ("Принял", "med:taken:2")))
|
||
runner = _runner(api)
|
||
list(runner.play_day(SCENARIO, 1))
|
||
logs = list(runner.play_day(SCENARIO, 2))
|
||
assert api.calls[2:] == [
|
||
("advance", 24.0),
|
||
("button", "med:taken:2"),
|
||
("advance", 12.0),
|
||
("text", "подташнивает с утра"),
|
||
]
|
||
color_log = logs[-2]
|
||
assert color_log.status == "warn" and "green" in color_log.message
|
||
assert runner.state.done == [1, 2] and runner.state.failed is None
|
||
|
||
|
||
def test_scroll_to_plays_all_unplayed_days_fast():
|
||
api = FakeApi(datetime(2026, 10, 5, 7, 0, tzinfo=MSK))
|
||
api.color = "yellow"
|
||
for d in (5, 6):
|
||
api.push_at("08:00", date(2026, 10, d), _reply("доза", ("Принял", f"med:taken:{d}")))
|
||
runner = _runner(api)
|
||
runner.delay = 0
|
||
logs = list(runner.scroll_to(SCENARIO, 5))
|
||
assert runner.state.done == [1, 2, 5]
|
||
# Day 1 lands on today 08:00 (nothing played yet), day 5 is chained: 5 − 2 = 3 days after day 2.
|
||
assert runner.state.dates == {1: date(2026, 10, 5), 2: date(2026, 10, 6), 5: date(2026, 10, 9)}
|
||
assert all(log.status != "fail" for log in logs)
|
||
assert runner.sleeps == []
|
||
assert list(runner.scroll_to(SCENARIO, 5))[0].status == "info" # nothing left to scroll
|
||
|
||
|
||
def test_missing_button_pauses_and_resume_can_skip_or_retry():
|
||
api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK)) # no reminder scheduled → button never shows up
|
||
runner = _runner(api)
|
||
logs = list(runner.play_day(SCENARIO, 1))
|
||
assert logs[-1].status == "fail" and "не найдена" in logs[-1].message
|
||
assert runner.state.failed is not None and (runner.state.failed.day, runner.state.failed.step) == (1, 1)
|
||
assert runner.state.done == []
|
||
assert runner.sleeps == [0.5] + [runner.retry_wait] * runner.press_retries # step pause, then retries
|
||
|
||
api.box.append(_reply("доза", ("Принял", "med:taken:7")))
|
||
retry = list(runner.resume(SCENARIO, skip=False))
|
||
assert retry[-1].status == "ok" and ("button", "med:taken:7") in api.calls
|
||
assert runner.state.failed is None and runner.state.done == [1]
|
||
|
||
|
||
def test_resume_skip_continues_until_target_day():
|
||
api = FakeApi(datetime(2026, 10, 5, 21, 0, tzinfo=MSK))
|
||
runner = _runner(api)
|
||
runner.delay = 0
|
||
list(runner.scroll_to(SCENARIO, 2)) # fails on day 1 «Принял»
|
||
assert runner.state.failed.until == 2
|
||
list(runner.resume(SCENARIO, skip=True)) # skip the press, play day 1 rest and day 2 → fails on day 2 press
|
||
assert runner.state.done == [1]
|
||
assert (runner.state.failed.day, runner.state.failed.step) == (2, 1)
|
||
|
||
|
||
def test_api_errors_never_escape_the_runner():
|
||
api = FakeApi(datetime(2026, 10, 5, 19, 0, tzinfo=MSK))
|
||
api.fail_on = "text"
|
||
runner = _runner(api)
|
||
logs = list(runner.play_day(SCENARIO, 5))
|
||
assert logs[-1].status == "fail" and "HTTP 500" in logs[-1].message
|
||
assert runner.state.failed is not None
|
||
|
||
api.fail_on = "clock"
|
||
logs = list(_runner(api).play_day(SCENARIO, 5))
|
||
assert logs[-1].status == "fail" and "часы" in logs[-1].message
|
||
|
||
api.fail_on = "patients"
|
||
color_only = parse_scenario(_jsonl({"day": 0, "actions": [{"expect_color": "green"}]}))
|
||
logs = list(_runner(api).play_day(color_only, 0))
|
||
assert logs[1].status == "warn" and logs[-1].status == "ok"
|
||
|
||
|
||
def test_past_target_is_skipped_with_warning():
|
||
api = FakeApi(datetime(2026, 10, 6, 21, 0, tzinfo=MSK))
|
||
state = PlaybackState(dates={5: date(2026, 10, 6)})
|
||
logs = list(_runner(api, state).play_day(SCENARIO, 5)) # replaying day 5 after 20:00 the same evening
|
||
assert logs[1].status == "warn" and "уже прошло" in logs[1].message
|
||
assert ("text", "брошу") in api.calls and not any(c[0] == "advance" for c in api.calls)
|
||
|
||
|
||
def test_play_unknown_day_is_info():
|
||
logs = list(_runner(FakeApi(datetime(2026, 10, 5, 8, 0, tzinfo=MSK))).play_day(SCENARIO, 42))
|
||
assert [log.status for log in logs] == ["info"]
|
||
|
||
|
||
def test_start_action_sends_code():
|
||
api = FakeApi(datetime(2026, 10, 5, 8, 0, tzinfo=MSK))
|
||
sc = parse_scenario(_jsonl({"day": 0, "actions": [{"start": "DEMO-0001"}, {"start": ""}]}))
|
||
list(_runner(api).play_day(sc, 0))
|
||
assert api.calls == [("start", {"code": "DEMO-0001"}), ("start", {})]
|
||
|
||
|
||
# ------------------------------------------------------------------------------------------------ http client
|
||
def _client(handler) -> ConsoleApi:
|
||
return ConsoleApi("http://api", client=httpx.Client(transport=httpx.MockTransport(handler)))
|
||
|
||
|
||
def test_console_api_paths_and_bodies():
|
||
seen: list[tuple[str, str, Any]] = []
|
||
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
body = json.loads(req.content) if req.content else None
|
||
seen.append((req.method, req.url.path, body))
|
||
if req.url.path.endswith("/events"):
|
||
return httpx.Response(200, json={"replies": [{"text": "hi"}]})
|
||
if req.url.path.endswith("/outbox"):
|
||
return httpx.Response(200, json={"replies": []})
|
||
return httpx.Response(200, json={"now": "2026-10-05T08:00:00+03:00"})
|
||
|
||
api = _client(handler)
|
||
assert api.send_event("tok-andrey", "button", button="med:taken:1") == [{"text": "hi"}]
|
||
assert api.outbox("tok-andrey") == []
|
||
api.advance(hours=1.5)
|
||
assert seen == [
|
||
("POST", "/api/v1/dialog/tok-andrey/events", {"type": "button", "params": {}, "button": "med:taken:1"}),
|
||
("GET", "/api/v1/dialog/tok-andrey/outbox", None),
|
||
("POST", "/api/v1/demo/clock/advance", {"days": 0, "hours": 1.5}),
|
||
]
|
||
|
||
|
||
def test_console_api_errors_become_api_error_and_personas_fall_back(monkeypatch):
|
||
def down(req: httpx.Request) -> httpx.Response:
|
||
raise httpx.ConnectError("refused")
|
||
|
||
api = _client(down)
|
||
with pytest.raises(ApiError, match="недоступен"):
|
||
api.clock()
|
||
assert api.health() is False
|
||
monkeypatch.setenv("DEMO_PERSONA_TOKEN", "tok-andrey")
|
||
personas, error = personas_or_fallback(api)
|
||
assert personas[0]["token"] == "tok-andrey" and error
|
||
|
||
api = _client(lambda req: httpx.Response(500))
|
||
with pytest.raises(ApiError, match="HTTP 500"):
|
||
api.outbox("t")
|
||
|
||
ok = _client(lambda req: httpx.Response(200, json=[{"token": "tok-andrey", "display_name": "Андрей С."}]))
|
||
personas, error = personas_or_fallback(ok)
|
||
assert error is None and personas[0]["display_name"] == "Андрей С."
|
||
|
||
|
||
def test_personas_fall_back_to_doctor_patients_with_hero_first():
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
if req.url.path.endswith("/demo/personas"):
|
||
return httpx.Response(200, json=[])
|
||
return httpx.Response(
|
||
200, json=[{"token": "h1", "display_name": "Сергей П."}, {"token": "h2", "display_name": "Андрей С."}]
|
||
)
|
||
|
||
personas, error = personas_or_fallback(_client(handler))
|
||
assert [p["token"] for p in personas] == ["h2", "h1"] and "пациенты врача" in error
|
||
|
||
|
||
def test_scenario_path_guessed_from_name_or_token(tmp_path):
|
||
(tmp_path / "data" / "synthetic").mkdir(parents=True)
|
||
(tmp_path / "data" / "synthetic" / "andrey.jsonl").write_text("", encoding="utf-8")
|
||
assert persona_slug("Андрей С.") == "andrey" and persona_slug("Наталья Ф.") == "natalya"
|
||
assert default_scenario_path({"token": "h", "display_name": "Андрей С."}, tmp_path) == "data/synthetic/andrey.jsonl"
|
||
assert default_scenario_path({"token": "tok-andrey"}, tmp_path) == "data/synthetic/andrey.jsonl"
|
||
assert default_scenario_path({"token": "x", "display_name": "Ольга Н."}, tmp_path) is None
|
||
assert default_scenario_path({"token": "x", "scenario": "s.jsonl"}, tmp_path) == "s.jsonl"
|