Advertisement

← refacto-feature-scout

tests/test_store_and_replies.py

13.2 KB Download file
from scout.decisions import apply_commands, ideas_by_status
from scout.store import Store


def seed(store: Store):
    a = store.add_idea("terminal", "Whale alerts", "Alert on big trades", evidence=[{"url": "https://x", "note": ""}])
    b = store.add_idea("terminal", "Brier score tracker", "Track calibration", evidence=[{"url": "https://y", "note": ""}])
    c = store.add_idea("terminal", "Cross-venue spread", "Poly vs Kalshi gap", evidence=[{"url": "https://z", "note": ""}])
    return a, b, c


def test_store_roundtrip(store):
    a, b, c = seed(store)
    idea = store.idea(a)
    assert idea["status"] == "proposed" and idea["evidence"][0]["url"] == "https://x"
    store.set_status(a, "approved")
    assert store.idea(a)["decided_at"] is not None
    assert [i["id"] for i in store.ideas(status="approved")] == [a]
    assert store.all_titles("terminal") == [("Whale alerts", "approved"), ("Brier score tracker", "proposed"),
                                            ("Cross-venue spread", "proposed")]


def test_apply_commands_respects_state(store):
    a, b, c = seed(store)
    lines = apply_commands(store, [("approve", [a]), ("reject", [b]), ("ship", [c]), ("approve", [99])])
    assert store.idea(a)["status"] == "approved"
    assert store.idea(b)["status"] == "rejected"
    assert store.idea(c)["status"] == "proposed"          # can't ship what isn't built
    assert any("can't ship" in l for l in lines) and any("99" in l for l in lines)
    # a built idea can be shipped; the request is recorded in its own column
    store.set_status(c, "built", build_branch="scout/3-x")
    apply_commands(store, [("ship", [c])])
    assert store.idea(c)["status"] == "built" and store.idea(c)["ship_requested_at"]
    from scout.build import pending_ships
    assert [i["id"] for i in pending_ships(store)] == [c]
    store.clear_ship_request(c)
    assert pending_ships(store) == []


def test_pause_flag_skips_builds(store, monkeypatch):
    from scout import cli
    monkeypatch.setenv("SCOUT_PAUSE_BUILDS", "1")
    a = store.add_idea("terminal", "X", "s", evidence=[{"url": "https://e", "note": ""}])
    store.set_status(a, "approved")
    cli.cmd_build(store)
    assert store.idea(a)["status"] == "approved"   # untouched while paused


def test_a_stopped_build_can_be_picked_up_from_the_command_line_too(store, monkeypatch):
    """The page and the command line must agree: one set of rules, not two."""
    from scout import build as buildmod
    from scout.decisions import apply_commands
    iid = store.add_idea("terminal", "Interrupted", "s", evidence=[{"url": "https://e", "note": ""}])
    store.start_run(f"build:{iid}")
    store.set_status(iid, "building", build_branch=f"scout/{iid}-x")

    # while it really is running, nothing may follow it
    monkeypatch.setattr(buildmod, "was_interrupted", lambda st, i, *a, **k: False)
    assert "can't approve" in apply_commands(store, [("approve", [iid])])[0]
    assert store.idea(iid)["status"] == "building"

    # once the machine has taken it out, it can be put back
    monkeypatch.setattr(buildmod, "was_interrupted", lambda st, i, *a, **k: True)
    said = apply_commands(store, [("approve", [iid])])[0]
    assert "back in the queue" in said and "carries on from where it stopped" in said
    assert store.idea(iid)["status"] == "approved"


def test_a_ship_already_waiting_survives_the_upgrade():
    """Someone who asked to ship just before this version arrived must not be forgotten."""
    import sqlite3
    from scout.store import SCHEMA, Store
    raw = sqlite3.connect(":memory:")
    raw.executescript(SCHEMA)   # the schema has no ship_requested_at; _migrate adds it
    raw.execute("INSERT INTO ideas (platform,title,summary,created_at,status,error) "
                "VALUES ('terminal','x','y',1,'built','SHIP_REQUESTED')")
    raw.commit()
    s = Store.__new__(Store); s.path = ":memory:"; s.conn = raw; s.conn.row_factory = sqlite3.Row
    s._migrate()
    assert s.conn.execute("select ship_requested_at from ideas").fetchone()[0]


def test_decide_command_applies_decisions(tmp_path, monkeypatch, capsys):
    from scout import cli, config
    monkeypatch.setattr(cli, "DB_PATH", tmp_path / "s.sqlite")
    from scout.store import Store
    s = Store(tmp_path / "s.sqlite")
    a = s.add_idea("terminal", "Thing", "x", evidence=[{"url": "https://e", "note": ""}])
    assert cli.main(["decide", "approve", str(a)]) == 0
    assert Store(tmp_path / "s.sqlite").idea(a)["status"] == "approved"
    assert "approved" in capsys.readouterr().out


def test_builds_run_in_approval_order(store, monkeypatch, registered):
    from scout import cli
    ids = [store.add_idea("terminal", f"Idea {n}", "s", evidence=[{"url": "https://e", "note": ""}]) for n in range(3)]
    store.set_status(ids[2], "approved"); store.conn.execute("update ideas set decided_at=100 where id=?", (ids[2],))
    store.set_status(ids[0], "approved"); store.conn.execute("update ideas set decided_at=200 where id=?", (ids[0],))
    store.conn.commit()
    built = []
    monkeypatch.setattr(cli.buildmod, "build_idea", lambda p, idea, st: (built.append(idea["id"]) or {"status": "blocked", "blocker": "test"}))
    monkeypatch.setattr(cli.buildmod, "pending_ships", lambda st: [])
    cli.cmd_build(store, limit=1)
    assert built == [ids[2]]


def test_a_chain_ships_each_step_by_itself_and_approves_the_next(store, monkeypatch):
    """Ken (2026-09-07): "You can just do them all without asking me." Steps that build on each
    other ship as soon as they are built, and the next step is approved the moment the previous
    one has shipped — each step shows up under "Lately" on the page, but nobody has to press anything."""
    from scout import cli
    ev = [{"url": "https://e", "note": ""}]
    a = store.add_idea("terminal", "Step 1", "s", evidence=ev)
    b = store.add_idea("terminal", "Step 2", "s", evidence=ev)
    c = store.add_idea("terminal", "Step 3", "s", evidence=ev)
    store.chain([a, b, c])
    store.set_status(a, "approved")
    assert store.idea(b)["after_id"] == a and store.idea(c)["after_id"] == b and store.idea(a)["auto_ship"] == 1
    events = []

    def fake_build(p, idea, st):
        st.set_status(idea["id"], "built", build_branch=f"scout/{idea['id']}")
        events.append(("build", idea["id"]))
        return {"status": "done", "summary": "ok", "cost_usd": 1.0}

    def fake_ship(p, idea, st):
        st.set_status(idea["id"], "shipped")
        events.append(("ship", idea["id"]))
        return {"ok": True, "retry": False, "message": "live"}

    monkeypatch.setattr(cli.buildmod, "build_idea", fake_build)
    monkeypatch.setattr(cli.buildmod, "ship_idea", fake_ship)
    monkeypatch.setattr(cli, "project_for", lambda idea: cli.known_projects().get("terminal") or _StubProject())
    monkeypatch.delenv("SCOUT_PAUSE_BUILDS", raising=False)

    cli.cmd_build(store, limit=1)           # one pass: builds step 1, ships it, approves step 2
    assert events == [("build", a), ("ship", a)]
    assert store.idea(a)["status"] == "shipped"
    assert store.idea(b)["status"] == "approved" and store.idea(c)["status"] == "proposed"

    cli.cmd_build(store, limit=1)           # next pass: step 2, then step 3 is released
    assert events[2:] == [("build", b), ("ship", b)]
    assert store.idea(c)["status"] == "approved"


def test_a_chain_stops_when_a_step_does_not_build(store, monkeypatch):
    from scout import cli
    ev = [{"url": "https://e", "note": ""}]
    a = store.add_idea("terminal", "Step 1", "s", evidence=ev)
    b = store.add_idea("terminal", "Step 2", "s", evidence=ev)
    store.chain([a, b]); store.set_status(a, "approved")
    shipped = []
    monkeypatch.setattr(cli.buildmod, "build_idea", lambda p, idea, st: (st.set_status(idea["id"], "blocked") or {"status": "blocked", "blocker": "x"}))
    monkeypatch.setattr(cli.buildmod, "ship_idea", lambda p, idea, st: shipped.append(idea["id"]))
    monkeypatch.setattr(cli, "project_for", lambda idea: _StubProject())
    monkeypatch.delenv("SCOUT_PAUSE_BUILDS", raising=False)
    cli.cmd_build(store, limit=1)
    assert shipped == [] and store.idea(b)["status"] == "proposed", "nothing ships or advances on a blocked step"


class _StubProject:
    name = "Refacto Terminal"; key = "terminal"; deploy_command = None; spend_limit_usd = 0.0; folder = "/tmp/x"


def test_the_chain_command_sets_the_links(store, monkeypatch, capsys):
    from scout import cli
    monkeypatch.setattr(cli, "Store", lambda path: store)
    ev = [{"url": "https://e", "note": ""}]
    a = store.add_idea("terminal", "One", "s", evidence=ev); b = store.add_idea("terminal", "Two", "s", evidence=ev)
    assert cli.main(["chain", str(a), str(b)]) == 0
    assert "Hands-off chain set" in capsys.readouterr().out
    assert store.idea(b)["after_id"] == a and store.idea(a)["auto_ship"] == 1
    assert cli.main(["chain", "999"]) == 1


def test_claudes_usage_limit_pauses_the_queue_instead_of_failing_ideas(store, monkeypatch):
    """2026-09-08 04:05 UTC: the Max plan's session limit hit; #38, #10 and #11 were each marked
    failed at $0 without running, one per pass, until the limit reset. A limit is a pause."""
    from scout import build as b, cli
    from scout.claude_runner import ClaudeResult
    ev = [{"url": "https://e", "note": ""}]
    a = store.add_idea("terminal", "First", "s", evidence=ev); store.set_status(a, "approved")
    c = store.add_idea("terminal", "Second", "s", evidence=ev); store.set_status(c, "approved")
    place = store.idea(a)["decided_at"]
    attempted = []
    real_build = b.build_idea

    def fake_build(p, idea, st):
        attempted.append(idea["id"])
        return real_build(p, idea, st, runner=lambda *x, **k: ClaudeResult(ok=False, error="You've hit your session limit · resets 5:30am (UTC)"))

    monkeypatch.setattr(cli.buildmod, "build_idea", fake_build)
    monkeypatch.setattr(cli.buildmod, "prepare_worktree", lambda p, i: (__import__("pathlib").Path("/tmp"), "scout/x"))
    monkeypatch.setattr(b, "prepare_worktree", lambda p, i: (__import__("pathlib").Path("/tmp"), "scout/x"))
    monkeypatch.setattr(b, "CLOUD", False)
    from scout.projects import Project
    monkeypatch.setattr(cli, "project_for", lambda idea: Project(key="terminal", name="T", folder="/tmp"))
    monkeypatch.delenv("SCOUT_PAUSE_BUILDS", raising=False)
    cli.cmd_build(store, limit=5)
    assert attempted == [a], "the pass stops at the first limit hit; the second idea is not even tried"
    assert store.idea(a)["status"] == "approved" and store.idea(a)["decided_at"] == place
    assert "usage limit" in store.idea(a)["error"]
    assert store.idea(c)["status"] == "approved"
    paused = [n for n in store.notices() if n["kind"] == "paused"]
    assert len(paused) == 1 and "usage limit" in paused[0]["text"]
    cli.cmd_build(store, limit=5)
    assert len([n for n in store.notices() if n["kind"] == "paused"]) == 1, "said once, not every half hour"
    assert b.is_usage_limit("Error: 429 rate_limit_error") and not b.is_usage_limit("the builder gave no final report")


def test_a_build_and_a_ship_each_leave_a_line_on_the_board(store, monkeypatch):
    """Ken (2026-09-11): no more emails, the page is the way to be told. Everything the scout
    used to write to him now lands under "Lately" on the page."""
    from scout import cli
    ev = [{"url": "https://e", "note": ""}]
    a = store.add_idea("terminal", "Alerts", "s", evidence=ev); store.set_status(a, "approved")
    monkeypatch.setattr(cli.buildmod, "build_idea", lambda p, idea, st: (
        st.set_status(idea["id"], "built", build_branch="scout/1") or {"status": "done", "summary": "ok"}))
    monkeypatch.setattr(cli.buildmod, "ship_idea", lambda p, idea, st: (
        st.set_status(idea["id"], "shipped") or {"ok": True, "retry": False, "message": f"#{idea['id']} is in."}))
    monkeypatch.setattr(cli, "project_for", lambda idea: _StubProject())
    monkeypatch.delenv("SCOUT_PAUSE_BUILDS", raising=False)
    cli.cmd_build(store, limit=1)
    kinds = [n["kind"] for n in store.notices()]
    assert kinds == ["done"] and "Built #1 Alerts" in store.notices()[0]["text"]
    store.request_ship(a)
    cli.cmd_build(store, limit=1)
    assert [n["kind"] for n in store.notices()] == ["shipped", "done"]
    # a failed build says why, and a problem is a problem
    b = store.add_idea("terminal", "Broken", "s", evidence=ev); store.set_status(b, "approved")
    monkeypatch.setattr(cli.buildmod, "build_idea", lambda p, idea, st: (
        st.set_status(idea["id"], "failed") or {"status": "failed", "error": "tests fail"}))
    cli.cmd_build(store, limit=1)
    assert store.notices()[0]["kind"] == "failed" and "tests fail" in store.notices()[0]["text"]
    cli.alert(store, "poll crashed", "Traceback ... boom")
    assert store.notices()[0]["kind"] == "problem" and "poll crashed" in store.notices()[0]["text"]


def test_the_mail_commands_are_gone(store, monkeypatch):
    import pytest
    from scout import cli
    for argv in (["digest"], ["replies"], ["test-mail"], ["poll", "--no-mail"]):
        with pytest.raises(SystemExit):
            cli.main(argv)