| """Hosted demo (Hugging Face Space): live-run guards, key safety, shipped files.
|
|
|
| See docs/hosting-plan.md, sections 2-5.
|
| """
|
|
|
| import csv
|
| import getpass
|
| import json
|
| import re
|
| import runpy
|
| import subprocess
|
| import sys
|
| from datetime import date
|
| from pathlib import Path
|
| from types import SimpleNamespace
|
|
|
| import httpx2
|
| import pytest
|
| from test_demo import _lens_script_client, _scripted_client
|
|
|
| from beampilot import app as demo_app
|
| from beampilot.agent import AgentConfig
|
|
|
| REPO = Path(__file__).resolve().parents[1]
|
| SENTINEL = "dummy-key-SENTINEL-7f3a9c"
|
|
|
|
|
| @pytest.fixture(autouse=True)
|
| def clean_env(monkeypatch):
|
| for name in ("BEAMPILOT_LIVE_RUNS", "BEAMPILOT_SPEND_CAP_USD"):
|
| monkeypatch.delenv(name, raising=False)
|
| monkeypatch.setattr(demo_app, "_GLOBAL", demo_app.GlobalBudget())
|
|
|
|
|
| def _request(name: str) -> SimpleNamespace:
|
| demo_app._BROWSERS.pop(name, None)
|
| return SimpleNamespace(session_hash=name)
|
|
|
|
|
|
|
|
|
|
|
| def test_spend_cap_is_read_from_the_environment(monkeypatch) -> None:
|
| monkeypatch.setenv("BEAMPILOT_SPEND_CAP_USD", "0.3")
|
| assert demo_app.spend_cap_from_env() == 0.3
|
| assert demo_app.GlobalBudget().spend_cap_usd == 0.3
|
| for bad in ("", "lots", "-1", "0", "inf", "nan"):
|
| monkeypatch.setenv("BEAMPILOT_SPEND_CAP_USD", bad)
|
| assert demo_app.spend_cap_from_env() == demo_app.DEFAULT_SPEND_CAP_USD == 5.0
|
|
|
|
|
| def test_spend_cap_admits_a_run_only_if_a_worst_case_run_still_fits() -> None:
|
| budget = demo_app.GlobalBudget(spend_cap_usd=1.0)
|
| assert budget.reserve() is None
|
| assert budget.reserved_usd == pytest.approx(0.55)
|
| budget.add_cost(0.40)
|
| budget.release()
|
| assert budget.reserved_usd == 0.0
|
| assert budget.reserve() is None
|
| budget.add_cost(0.10)
|
| budget.release()
|
|
|
| assert budget.out_of_budget() == demo_app.BUDGET_TEXT
|
| assert budget.reserve() == demo_app.BUDGET_TEXT
|
| assert budget.runs == 2 and budget.active == 0
|
|
|
|
|
| def test_a_run_that_never_started_frees_its_reserve() -> None:
|
| budget = demo_app.GlobalBudget(spend_cap_usd=0.6)
|
| assert budget.reserve() is None
|
| budget.release(started=False)
|
| assert budget.runs == 0 and budget.reserved_usd == 0.0
|
| assert budget.out_of_budget() is None
|
|
|
|
|
| def test_live_runs_add_their_cost_and_stop_at_the_cap(monkeypatch, tmp_path) -> None:
|
| monkeypatch.setattr(demo_app, "_client", _scripted_client)
|
| monkeypatch.setattr(demo_app, "LIVE_RUNS_DIR", tmp_path)
|
| monkeypatch.setattr(demo_app, "_GLOBAL", demo_app.GlobalBudget(spend_cap_usd=0.6))
|
| list(demo_app.live_run("clean", 0, "", _request("spend-1")))
|
| (saved,) = tmp_path.glob("*.json")
|
| cost = json.loads(saved.read_text(encoding="utf-8"))["cost_usd"]
|
| assert cost > 0
|
| assert demo_app._GLOBAL.spent_usd == pytest.approx(cost, abs=1e-5)
|
| assert demo_app._GLOBAL.reserved_usd == 0.0
|
| demo_app._GLOBAL.add_cost(0.6 - 0.55 - demo_app._GLOBAL.spent_usd + 0.01)
|
| update = next(demo_app.live_run("clean", 0, "", _request("spend-2")))
|
| assert update[2] == demo_app.BUDGET_TEXT
|
|
|
|
|
|
|
|
|
|
|
| def test_live_switch_values(monkeypatch) -> None:
|
| for off in ("off", "OFF", " false ", "0", "no"):
|
| monkeypatch.setenv("BEAMPILOT_LIVE_RUNS", off)
|
| assert not demo_app.live_switch_on()
|
| for on in ("on", "", "yes", "1"):
|
| monkeypatch.setenv("BEAMPILOT_LIVE_RUNS", on)
|
| assert demo_app.live_switch_on()
|
|
|
|
|
| def test_switched_off_live_runs_never_start(monkeypatch) -> None:
|
| def no_client():
|
| raise AssertionError("no client may be made when live runs are off")
|
|
|
| monkeypatch.setattr(demo_app, "_client", no_client)
|
| monkeypatch.setenv("BEAMPILOT_LIVE_RUNS", "off")
|
| request = _request("switch-off")
|
| update = next(demo_app.live_run("clean", 0, "", request))
|
| assert update[2] == demo_app.LIVE_OFF_TEXT
|
| assert demo_app._GLOBAL.runs == 0 and demo_app._browser(request).runs == 0
|
|
|
|
|
| def test_banner_and_start_button_at_page_load(monkeypatch, tmp_path) -> None:
|
| monkeypatch.chdir(tmp_path)
|
| monkeypatch.delenv("MODEL_API_KEY", raising=False)
|
| banner, enabled = demo_app.live_banner()
|
| assert demo_app.NO_KEY_TEXT in banner and not enabled
|
| assert "MODEL_API_KEY" in banner
|
|
|
| monkeypatch.setenv("BEAMPILOT_LIVE_RUNS", "off")
|
| banner, enabled = demo_app.live_banner()
|
| assert demo_app.LIVE_OFF_TEXT in banner and not enabled
|
|
|
| monkeypatch.delenv("BEAMPILOT_LIVE_RUNS")
|
| monkeypatch.setenv("MODEL_API_KEY", SENTINEL)
|
| assert demo_app.live_banner() == ("", True)
|
| assert SENTINEL not in json.dumps(demo_app.live_banner())
|
|
|
| monkeypatch.setenv("BEAMPILOT_SPEND_CAP_USD", "0.3")
|
| monkeypatch.setattr(demo_app, "_GLOBAL", demo_app.GlobalBudget())
|
| banner, enabled = demo_app.live_banner()
|
| assert demo_app.BUDGET_TEXT in banner and not enabled
|
|
|
|
|
| def test_live_runs_have_a_queue_of_one_and_replays_do_not_wait() -> None:
|
| pytest.importorskip("gradio")
|
| app = demo_app.build_app()
|
| fns = {getattr(f.fn, "__name__", ""): f for f in app.fns.values()}
|
| assert demo_app.MAX_CONCURRENT_RUNS == 1
|
| assert fns["start_run"].concurrency_limit == 1
|
| assert fns["start_run"].concurrency_id == "live"
|
| assert fns["replay"].concurrency_id != "live"
|
| assert fns["replay"].concurrency_limit == "default"
|
|
|
|
|
|
|
|
|
|
|
| def _saved_config(tmp_path: Path) -> dict:
|
| (saved,) = tmp_path.glob("*.json")
|
| return json.loads(saved.read_text(encoding="utf-8"))["config"]
|
|
|
|
|
| def test_live_lens_runs_use_the_v3_limits(monkeypatch, tmp_path) -> None:
|
| monkeypatch.setattr(demo_app, "_client", _lens_script_client)
|
| monkeypatch.setattr(demo_app, "LIVE_RUNS_DIR", tmp_path)
|
| request = _request("lens-limits")
|
| list(demo_app.live_run("waist_outside_range", 0, "", request, lens_kit=True))
|
| config = _saved_config(tmp_path)
|
| assert config["max_model_calls"] == 80
|
| assert config["max_total_tokens"] == 3_000_000
|
| assert config["max_cost_usd"] == 0.50
|
| assert config["change_lens"] is True
|
| assert demo_app._browser(request).session.scenario.max_positions == 28
|
|
|
|
|
| def test_normal_live_runs_keep_their_limits(monkeypatch, tmp_path) -> None:
|
| monkeypatch.setattr(demo_app, "_client", _scripted_client)
|
| monkeypatch.setattr(demo_app, "LIVE_RUNS_DIR", tmp_path)
|
| list(demo_app.live_run("clean", 0, "", _request("normal-limits")))
|
| config, default = _saved_config(tmp_path), AgentConfig()
|
| assert config["max_model_calls"] == default.max_model_calls
|
| assert config["max_total_tokens"] == default.max_total_tokens
|
| assert config["max_cost_usd"] == default.max_cost_usd
|
| assert not config.get("change_lens")
|
|
|
|
|
|
|
|
|
|
|
| def test_dummy_key_never_reaches_ui_run_file_or_console(
|
| monkeypatch, tmp_path, capfd
|
| ) -> None:
|
| """The real client with a fake server that echoes the key in its error."""
|
| import openai
|
|
|
| seen = []
|
|
|
| def server(request: httpx2.Request) -> httpx2.Response:
|
| auth = request.headers.get("authorization", "")
|
| seen.append(auth)
|
| message = f"Incorrect API key provided: {auth}"
|
| return httpx2.Response(401, json={"error": {"message": message}})
|
|
|
| real = openai.OpenAI
|
|
|
| def offline_openai(*args, **kwargs):
|
| transport = httpx2.MockTransport(server)
|
| return real(*args, http_client=httpx2.Client(transport=transport), **kwargs)
|
|
|
| monkeypatch.setattr(openai, "OpenAI", offline_openai)
|
| monkeypatch.setattr(demo_app, "_CLIENT", None)
|
| monkeypatch.setattr(demo_app, "LIVE_RUNS_DIR", tmp_path)
|
| monkeypatch.setenv("MODEL_API_KEY", SENTINEL)
|
| updates = list(demo_app.live_run("clean", 0, "", _request("sentinel")))
|
|
|
| assert seen and SENTINEL in seen[0]
|
| shown = json.dumps(updates, default=str)
|
| assert SENTINEL not in shown
|
| assert "AuthenticationError (HTTP 401)" in shown
|
| files = list(tmp_path.glob("*.json"))
|
| assert files
|
| for path in files:
|
| assert SENTINEL not in path.read_text(encoding="utf-8")
|
| out, err = capfd.readouterr()
|
| assert SENTINEL not in out and SENTINEL not in err
|
|
|
|
|
|
|
|
|
|
|
| def _personal_strings() -> set[str]:
|
| """Local user name, home folder and git e-mail of whoever runs the tests."""
|
| values = {getpass.getuser(), Path.home().name}
|
| try:
|
| email = subprocess.run(
|
| ["git", "config", "user.email"],
|
| capture_output=True,
|
| text=True,
|
| cwd=REPO,
|
| timeout=10,
|
| ).stdout.strip()
|
| except (OSError, subprocess.SubprocessError):
|
| email = ""
|
| if "@" in email:
|
| values |= {email, email.split("@", 1)[1]}
|
| generic = {"runner", "root", "user", "admin", "home"}
|
| return {v.lower() for v in values if len(v) >= 4 and v.lower() not in generic}
|
|
|
|
|
| FORBIDDEN = {
|
| "Windows path": re.compile(r"[A-Za-z]:(\\\\|/)"),
|
| "home path": re.compile(r"/(home|Users)/\w"),
|
| "e-mail address": re.compile(r"[\w.+-]+@[\w-]+\.[A-Za-z]{2,}"),
|
| "key-like string": re.compile(r"\bsk-[A-Za-z0-9_-]{8,}|Bearer\s", re.IGNORECASE),
|
| "key name": re.compile(r"api[_-]?key|MODEL_API_KEY|BASETEN", re.IGNORECASE),
|
| }
|
|
|
|
|
| def test_shipped_run_files_hold_no_paths_names_or_keys() -> None:
|
| personal = _personal_strings()
|
| runs = demo_app.demo_runs()
|
| assert len(runs) == 5
|
| for path in runs:
|
| text = path.read_text(encoding="utf-8")
|
| for kind, pattern in FORBIDDEN.items():
|
| match = pattern.search(text)
|
| assert match is None, f"{path.name}: {kind} {match.group(0)!r}"
|
| lowered = text.lower()
|
| assert not [v for v in personal if v in lowered], path.name
|
|
|
|
|
| def test_every_replay_has_a_recording_date() -> None:
|
| with demo_app.RECORDED_MANIFEST.open(encoding="utf-8", newline="") as handle:
|
| rows = list(csv.DictReader(handle))
|
| assert {row["file"] for row in rows} == {p.name for p in demo_app.demo_runs()}
|
| for row in rows:
|
| assert date.fromisoformat(row["recorded"]) >= date(2026, 10, 26)
|
| assert row["source"]
|
|
|
|
|
| def test_replays_are_labelled_as_recorded(tmp_path) -> None:
|
| for label, value in demo_app.replay_choices():
|
| assert re.match(r"Recorded run, 2026-\d\d-\d\d, muse-spark-1\.2: ", label)
|
| assert (REPO / value).exists()
|
| shipped = (demo_app.DEMO_RUNS_DIR / "lens_v3_L5.json").relative_to(REPO)
|
| title = list(demo_app.replay(shipped.as_posix(), None, 0.0))[-1][0]
|
| assert "Recorded run, 2026-10-31, muse-spark-1.2" in title
|
| assert "a replay, not a live run" in title and "L5 (seed 904)" in title
|
| upload = tmp_path / "lens_v3_L5.json"
|
| upload.write_bytes((REPO / shipped).read_bytes())
|
| title = list(demo_app.replay(None, str(upload), 0.0))[-1][0]
|
| assert "Recorded run, date unknown, muse-spark-1.2" in title
|
|
|
|
|
|
|
|
|
|
|
| def _requirement_name(spec: str) -> str:
|
| name = re.split(r"[<>=!~;\[ ]", spec.strip(), maxsplit=1)[0]
|
| return name.lower().replace("_", "-")
|
|
|
|
|
| def test_requirements_pin_exactly_the_project_dependencies() -> None:
|
| tomllib = pytest.importorskip("tomllib")
|
| project = tomllib.loads((REPO / "pyproject.toml").read_text(encoding="utf-8"))
|
| wanted = {_requirement_name(d) for d in project["project"]["dependencies"]}
|
| lines = (REPO / "requirements.txt").read_text(encoding="utf-8").splitlines()
|
| specs = [line.strip() for line in lines if line.strip() and line[0] != "#"]
|
| assert all(re.fullmatch(r"[A-Za-z0-9_.-]+==[0-9][\w.]*", s) for s in specs)
|
| assert {_requirement_name(s) for s in specs} == wanted
|
| assert "gradio" not in wanted
|
|
|
|
|
| def _front_matter(path: Path) -> dict[str, str]:
|
| text = path.read_text(encoding="utf-8")
|
| assert text.startswith("---\n")
|
| block = text[4 : text.index("\n---\n", 4)]
|
| pairs = (line.split(":", 1) for line in block.splitlines() if line.strip())
|
| return {key.strip(): value.strip().strip('"') for key, value in pairs}
|
|
|
|
|
| def test_space_readme_front_matter_matches_the_project() -> None:
|
| meta = _front_matter(REPO / "space" / "README.md")
|
| pyproject = (REPO / "pyproject.toml").read_text(encoding="utf-8")
|
| gradio_pin = re.search(r'"gradio==([\d.]+)"', pyproject).group(1)
|
| assert meta["sdk"] == "gradio"
|
| assert meta["sdk_version"] == gradio_pin
|
| assert meta["python_version"] == "3.12"
|
| assert (REPO / meta["app_file"]).is_file()
|
| assert meta["license"] == "mit"
|
| assert meta["suggested_hardware"] == "cpu-basic"
|
| assert 0 < len(meta["short_description"]) <= 60
|
| text = (REPO / "space" / "README.md").read_text(encoding="utf-8")
|
| assert "simulated" in text and "https://github.com/Farshadmhd/beampilot" in text
|
|
|
| assert not (REPO / "README.md").read_text(encoding="utf-8").startswith("---")
|
|
|
|
|
| def test_space_entry_point_runs_the_app_from_the_src_layout(monkeypatch) -> None:
|
| monkeypatch.setattr(sys, "path", list(sys.path))
|
| namespace = runpy.run_path(str(REPO / "scripts" / "space_app.py"), run_name="x")
|
| assert namespace["main"] is demo_app.main
|
| assert str(REPO / "src") == sys.path[0]
|
| assert demo_app.DEMO_RUNS_DIR == REPO / "docs" / "runs"
|
|
|
