Spaces:
Sleeping
Sleeping
File size: 4,271 Bytes
ba71b93 3fb99da ba71b93 3fb99da ba71b93 6b8bf73 ba71b93 e28d52e ba71b93 6b8bf73 ba71b93 6b8bf73 ba71b93 6b8bf73 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """Lightweight smoke tests for app.py.
Building the Gradio Blocks doesn't require the system to be initialised
(no API keys), so we can verify the UI compiles cleanly and the per-call
helpers work without ever launching a server.
"""
from __future__ import annotations
import json
import pytest
def test_app_imports() -> None:
import app # noqa: F401
def test_build_demo_compiles() -> None:
"""Calling build_demo() must not raise — catches Gradio API drift."""
pytest.importorskip("gradio")
from app import build_demo
demo = build_demo()
assert demo is not None
def test_build_user_profile_round_trip() -> None:
from app import build_user_profile
payload = build_user_profile(
name="Test",
age=30,
sex="male",
height_cm=175,
weight_kg=72,
activity="moderately active",
goal="maintain weight",
allergies="peanut, shrimp",
dislikes="okra",
country="Egypt",
conditions="hypertension",
medications="lisinopril",
lab_results="HbA1c 6.1%, LDL 145 mg/dL",
)
# Round-trip via JSON to mirror what the hidden Textbox carries.
serialised = json.dumps(payload)
parsed = json.loads(serialised)
assert parsed["user_profile"]["name"] == "Test"
assert parsed["user_profile"]["allergies"] == ["peanut", "shrimp"]
assert parsed["medical_history"]["conditions"] == ["hypertension"]
assert parsed["medical_history"]["lab_results"] == "HbA1c 6.1%, LDL 145 mg/dL"
def test_render_metrics_is_markdown() -> None:
from app import _render_metrics
snap = {
"agents": {"Coach": {"calls": 1, "total_seconds": 0.5, "errors": 0, "last_seconds": 0.5}},
"tools": {"QuantitiesFinder": {"calls": 2, "total_seconds": 0.1, "errors": 0, "last_seconds": 0.05}},
"parsing": {"native": 5, "fallback": 0, "failure": 0, "by_model": {}},
}
md = _render_metrics(snap)
assert "Coach" in md
assert "QuantitiesFinder" in md
assert "native=5" in md
def test_session_state_default_shape() -> None:
from app import SessionState
s = SessionState()
assert s.initialised is False
assert s.memory == {
"user_profile": {},
"medical_history": {},
"flags_and_assessments": {},
"plans": {},
}
assert s.conversation_history == []
def test_chat_handles_uninitialised_system() -> None:
"""Calling chat() before init must not crash; returns a friendly error.
``chat`` is a generator that streams (chat, trace, metrics, session,
progress) tuples — drain to the last yield and inspect the final
chatbot history.
"""
pytest.importorskip("gradio")
from app import SessionState, chat
# Make sure mealgraph.APP is None so we hit the guard.
import mealgraph
mealgraph.APP = None
last = None
for last in chat(
user_message="hi", history=[], session=SessionState(), profile_json=""
):
pass
assert last is not None
history, _log, _metrics, _session, progress = last
# messages-format chatbot: list of {role, content} dicts
assert history[-1]["role"] == "assistant"
assert history[-1]["content"].startswith("❌ System not initialised")
# Progress panel should explain the early exit, not be empty.
assert "system not initialised" in progress.lower()
def test_request_stop_with_no_active_run() -> None:
"""Pressing Stop with no in-flight run is a friendly no-op."""
pytest.importorskip("gradio")
from app import request_stop, _set_current_bus
# Make sure no leftover bus from another test is hanging around.
_set_current_bus(None)
msg = request_stop()
assert "No run is in progress" in msg
def test_progress_bus_check_stop_raises_after_set() -> None:
"""ProgressBus.check_stop is the gate every agent / tool wrapper hits
on entry. Once the stop flag is flipped, the next check raises so the
in-flight workflow unwinds at the next agent / tool boundary."""
pytest.importorskip("gradio")
from app import ProgressBus, StopRequested
bus = ProgressBus()
# Fresh bus: no raise.
bus.check_stop()
bus.stop_event.set()
with pytest.raises(StopRequested):
bus.check_stop()
|