File size: 7,079 Bytes
152be26 5806fe4 152be26 5806fe4 152be26 0553ea7 152be26 0553ea7 8de7c01 0553ea7 ef4748f 08e4d77 ef4748f 08e4d77 5806fe4 08e4d77 ef4748f 5806fe4 08e4d77 5806fe4 08e4d77 ef4748f 5806fe4 08e4d77 152be26 ef4748f 152be26 5806fe4 152be26 ef4748f 152be26 5806fe4 152be26 ef4748f 152be26 ef4748f 5806fe4 152be26 ef4748f 152be26 ef4748f 5806fe4 152be26 08e4d77 ef4748f 5806fe4 08e4d77 ef4748f 08e4d77 | 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | import os
import tempfile
from pathlib import Path
import gradio as gr
# Set test-only runtime configuration before importing app because app builds globals at import.
IMPORT_DB = Path(tempfile.gettempdir()) / "agent9-gradio-import-test.db"
if IMPORT_DB.exists():
IMPORT_DB.unlink()
os.environ["WORKFLOW_DB_PATH"] = str(IMPORT_DB)
os.environ["LIVE_AI_ENABLED"] = "false"
import app # noqa: E402
from src.demo_presentation import APP_CSS, PATTERN_HTML # noqa: E402
from src.persistence import SQLiteStateStore # noqa: E402
from src.schemas import WorkflowStatus # noqa: E402
def test_gradio_app_builds_as_blocks() -> None:
assert isinstance(app.demo, gr.Blocks)
def test_business_presentation_is_centered_and_vertical() -> None:
assert "max-width:1120px" in APP_CSS
assert "grid-template-columns:1fr" in APP_CSS
assert "flex-direction:column" in APP_CSS
assert "Humans retain authority over consequential exceptions" in PATTERN_HTML
def test_stream_story_yields_live_activity_and_full_completed_trace(
tmp_path: Path,
monkeypatch,
) -> None:
store = SQLiteStateStore(tmp_path / "stream-story.db")
monkeypatch.setattr(app, "STATE_STORE", store)
monkeypatch.setattr(app, "playback_delay", lambda _event_type: 0)
frames = list(app.stream_story_ui("Harbor Family — straightforward household"))
assert len(frames) > 3
assert "RUNNING" in frames[0][0]
assert "Workflow engine is running" in frames[0][0]
final = frames[-1]
run_id = final[2]
run = store.load(run_id)
assert run.status is WorkflowStatus.COMPLETED
assert "activity-scroll" in final[0]
assert "Workflow event trace complete" in final[0]
assert final[0].count("activity-event") >= len(run.events)
assert "COMPLETED" in final[1]
def test_one_click_standard_household_explains_ready_path(
tmp_path: Path,
monkeypatch,
) -> None:
store = SQLiteStateStore(tmp_path / "standard-story.db")
monkeypatch.setattr(app, "STATE_STORE", store)
bundle = app.run_story_ui("Harbor Family — straightforward household")
run = store.load(bundle[1])
assert run.status is WorkflowStatus.COMPLETED
assert "ready for advisor review" in bundle[6].lower()
assert "Live workflow path" in bundle[7]
assert "BOUNDED AI WORK PRODUCT" in bundle[8]
assert "DETERMINISTIC FALLBACK" in bundle[8]
assert "Standard path selected by code" in bundle[9]
assert "What happened, step by step" in bundle[10]
assert "AI Intake Organizer" in bundle[10]
assert "Standard package ready" in bundle[10]
def test_one_click_retry_story_explains_automatic_recovery(
tmp_path: Path,
monkeypatch,
) -> None:
store = SQLiteStateStore(tmp_path / "retry-story.db")
monkeypatch.setattr(app, "STATE_STORE", store)
bundle = app.run_story_ui("Harbor Family — temporary onboarding-service issue")
run = store.load(bundle[1])
assert run.status is WorkflowStatus.COMPLETED
assert run.node_runs["create_onboarding_package"].attempt == 2
assert "temporary onboarding-service problem" in bundle[6].lower()
assert "Recovered on attempt 2" in bundle[10]
def test_run_and_refresh_callbacks_use_persisted_sqlite_state(
tmp_path: Path,
monkeypatch,
) -> None:
store = SQLiteStateStore(tmp_path / "ui.db")
monkeypatch.setattr(app, "STATE_STORE", store)
bundle = app.run_request_ui(
"HH-CALLBACK-STANDARD",
"JOINT",
"Synthetic household seeking a standard advisory relationship.",
True,
"VERIFIED",
"STANDARD",
"NONE",
)
run_id = bundle[1]
assert run_id
assert store.load(run_id).status is WorkflowStatus.COMPLETED
assert "COMPLETED" in bundle[0]
refreshed = app.refresh_run_ui(run_id)
assert refreshed[1] == run_id
assert refreshed[2] == bundle[2]
assert refreshed[3] == bundle[3]
assert refreshed[6:] == bundle[6:]
def test_exception_callback_pauses_then_approves_without_replaying_upstream(
tmp_path: Path,
monkeypatch,
) -> None:
store = SQLiteStateStore(tmp_path / "human.db")
monkeypatch.setattr(app, "STATE_STORE", store)
paused_bundle = app.run_story_ui("Redwood Family Trust — human review required")
run_id = paused_bundle[1]
paused = store.load(run_id)
assert paused.status is WorkflowStatus.WAITING_FOR_HUMAN
assert paused.node_runs["validate_intake"].attempt == 1
assert paused.node_runs["review_gate"].attempt == 1
assert len(paused.human_reviews) == 1
assert "needs a person" in paused_bundle[6]
assert "Human review required by deterministic rules" in paused_bundle[9]
assert "Trust or entity structure" in paused_bundle[9]
assert "Relationship marked complex" in paused_bundle[9]
assert "Waiting for a person" in paused_bundle[10]
completed_bundle = app.submit_human_decision_ui(run_id, "APPROVE")
completed = store.load(run_id)
assert completed.status is WorkflowStatus.COMPLETED
assert completed.node_runs["validate_intake"].attempt == 1
assert completed.node_runs["review_gate"].attempt == 1
assert completed.node_runs["reviewed_onboarding"].attempt == 1
assert "COMPLETED" in completed_bundle[0]
assert "ready after human review" in completed_bundle[6].lower()
assert "Approved by a person" in completed_bundle[10]
def test_ai_work_product_is_used_in_simulated_package(tmp_path: Path, monkeypatch) -> None:
store = SQLiteStateStore(tmp_path / "ai-artifact.db")
monkeypatch.setattr(app, "STATE_STORE", store)
bundle = app.run_story_ui("Harbor Family — straightforward household")
run = store.load(bundle[1])
package = run.node_runs["create_onboarding_package"].output
assert package["ai_profile_category"] == "STANDARD_HOUSEHOLD"
assert package["ai_intake_summary"]
assert package["ai_source"] == "DETERMINISTIC_FALLBACK"
assert "included in the simulated onboarding package" in bundle[8]
def test_evaluation_callback_returns_ten_passing_cases() -> None:
summary, metrics, cases = app.run_evaluation_ui()
assert "10/10 passed" in summary
assert ["total_cases", "10"] in metrics
assert len(cases) == 10
assert all(row[2] == "PASS" for row in cases)
def test_missing_run_id_returns_safe_ui_message(tmp_path: Path, monkeypatch) -> None:
monkeypatch.setattr(app, "STATE_STORE", SQLiteStateStore(tmp_path / "missing.db"))
bundle = app.refresh_run_ui("does-not-exist")
assert bundle[1] == "does-not-exist"
assert "No persisted workflow run was found" in bundle[0]
assert bundle[2:6] == ([], [], [], {})
assert "Choose a fictional household story" in bundle[6]
assert "Waiting for a household" in bundle[8]
def test_unknown_story_returns_safe_explanation(tmp_path: Path, monkeypatch) -> None:
monkeypatch.setattr(app, "STATE_STORE", SQLiteStateStore(tmp_path / "unknown-story.db"))
bundle = app.run_story_ui("invented story")
assert "Choose one of the four onboarding stories" in bundle[0]
assert bundle[1] == ""
|