gitea-saas-core/tests/test_intake_eval.py
Claude 8620d4d36e gitea-saas-core#39: eval farm HTTP surface — register_eval_routes
Five routes symmetric with the factory front-end:
  POST /api/eval/batches                (funded 201 / unfunded 202 queued)
  GET  /api/eval/batches/<id>           status + per-task state
  GET  /api/eval/batches/<id>/results   the results table
  POST /api/eval/batches/<id>/cancel    stop draining + release the hold
  GET  /api/eval/health                 liveness, no auth

Auth delegates to gitea_saas.auth (issue #24 single path — no sixth
implementation): missing/blank key -> canonical 401; no key configured
at all -> 503 SERVICE_UNAVAILABLE (issue #38's assembly acceptance).

Unfunded submits now queue in pending_eval_batches (no Gitea issue is
ever created); flush_pending ships them after funding, exactly like a
fresh funded submit. Funded submits reserve eval_batch_hold_minor
(SAAS_EVAL_BATCH_HOLD_MINOR, default = debit floor) via the billing
core's reserve_hold; cancel releases it so the balance returns to its
pre-submit value exactly, and marks the batch cancelled (poll/emit
become no-ops). Rolled-back intake releases the hold too.

Schema: pending_eval_batches + eval_batch_cancels tables (idempotent
CREATE IF NOT EXISTS), hold_id column with a PRAGMA-guarded ALTER for
pre-existing DB files.
2026-09-01 03:58:29 -05:00

829 lines
35 KiB
Python

"""Hermetic tests for the eval farm intake (issue #8 acceptance).
MockGitea (in-memory repos/issues/comments/labels), a scripted emitter
(retry/backoff verified via call counts), and a real BillingCore on a tmp
DB (funded vs unfunded gating is real). No network, no sleeps (a noop
``sleep_fn`` is injected everywhere).
Run: python3 -m pytest tests/test_intake_eval.py -q
"""
import json
import sqlite3
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import pytest
from gitea_saas.auth import UNAUTHORIZED_BODY
from gitea_saas.billing_core import BillingCore
from gitea_saas.intake_eval import (
EvalIntake,
EvalIntakeError,
EVAL_LABEL,
register_eval_routes,
)
class Config:
"""Tiny config object for the intake (db_path + optional overrides)."""
def __init__(self, db_path, **kwargs):
self.db_path = Path(db_path)
for k, v in kwargs.items():
setattr(self, k, v)
class MockGitea:
"""In-memory Gitea seam recording every mutation."""
def __init__(self):
self.repos = {} # repo -> {"issues": {n: dict}, "comments": {n: [str]}}
self.next_number = 0 # global issue numbering
self.created_repos = [] # repo names created
self.created_issues = [] # (repo, title, body, labels)
def _repo(self, name):
if name not in self.repos:
self.repos[name] = {"issues": {}, "comments": {}}
return self.repos[name]
def repo_exists(self, name):
return name in self.repos
def create_org_repo(self, name):
self._repo(name)
self.created_repos.append(name)
def create_issue(self, repo, title, body, labels=None):
r = self._repo(repo)
self.next_number += 1
n = self.next_number
r["issues"][n] = {"title": title, "body": body,
"state": "open", "labels": list(labels or [])}
r["comments"][n] = []
self.created_issues.append((repo, title, body, list(labels or [])))
# Canonical client contract (issue #4/!17): returns the issue number.
return n
def get_issue(self, repo, number):
r = self._repo(repo)
issue = r["issues"].get(number)
if issue is None:
return None
return dict(issue)
def list_issue_comments(self, repo, number):
return [{"body": b} for b in self._repo(repo)["comments"].get(number, [])]
# -- test helpers: simulate the lane draining a task -------------------
def add_comment(self, repo, number, body):
self._repo(repo)["comments"].get(number, []).append(body)
def close_issue(self, repo, number):
self._repo(repo)["issues"][number]["state"] = "closed"
class MockEmitter:
"""Scripted emitter: fails the first ``fail_first`` pushes."""
def __init__(self, fail_first=0):
self.fail_first = fail_first
self.push_count = 0
self.push_payloads = [] # (report, url)
def push(self, report, url):
self.push_count += 1
if self.push_count <= self.fail_first:
return False
self.push_payloads.append((report, url))
return True
def noop_sleep(_seconds):
return None
@pytest.fixture
def funded(tmp_path):
core = BillingCore(tmp_path / "billing.db")
ok, c = core.create_client("eval-client")
assert ok
ok, _ = core.credit(c["id"], 10_000, "fund-1")
assert ok
return core, c["id"]
@pytest.fixture
def unfunded(tmp_path):
core = BillingCore(tmp_path / "billing.db")
ok, c = core.create_client("poor-client")
assert ok
return core, c["id"]
def make_intake(core, tmp_path, gitea=None, emitter=None, **config_kwargs):
gitea = gitea or MockGitea()
cfg = Config(tmp_path / "saas.db", **config_kwargs)
return EvalIntake(core, gitea, cfg, emitter=emitter, sleep_fn=noop_sleep)
TASKS = [
{"task_id": "t1", "title": "Eval task one",
"body": "## Spec\nrun X\n## Acceptance\n- [ ] a\n- [ ] b"},
{"task_id": "t2", "title": "Eval task two", "body": "## Spec\nrun Y"},
]
# -------------------------------------------------------------------------- #
# Submission #
# -------------------------------------------------------------------------- #
class TestSubmitBatch:
def test_funded_batch_creates_one_issue_per_task(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
assert r["ok"] is True
assert r["batch_id"] > 0
assert r["repo"] == "eval-runs"
assert [i["task_id"] for i in r["issues"]] == ["t1", "t2"]
# shared repo exists; two issues; each carries the eval-task label
assert "eval-runs" in gitea.created_repos
assert len(gitea.created_issues) == 2
for repo, title, body, labels in gitea.created_issues:
assert repo == "eval-runs"
assert labels == [EVAL_LABEL]
assert f"batch: {r['batch_id']}" in body
# both tasks persisted as pending
batch = intake.get_batch(r["batch_id"])
assert batch["ok"] and len(batch["tasks"]) == 2
assert all(t["status"] == "pending" for t in batch["tasks"])
assert batch["batch"]["status"] == "open"
def test_unfunded_batch_402_and_no_gitea_mutations(self, unfunded, tmp_path):
core, client_id = unfunded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
assert r["ok"] is False
assert r["reason"] == "unfunded"
assert r["status"] == 402
assert gitea.created_issues == [] # no issues created
assert gitea.created_repos == [] # no repo created either
assert gitea.next_number == 0
def test_unknown_client_is_fail_closed_unfunded(self, tmp_path):
core = BillingCore(tmp_path / "billing.db")
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(9999, TASKS)
assert r["ok"] is False and r["reason"] == "unfunded"
assert r["status"] == 402
assert gitea.created_issues == []
def test_duplicate_task_id_rejected(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path)
with pytest.raises(EvalIntakeError):
intake.submit_batch(client_id, [TASKS[0], dict(TASKS[0])])
def test_intake_failure_creates_no_partial_batch(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
class FlakyGitea(MockGitea):
def create_issue(self, *a, **kw):
if self.next_number >= 1: # second issue fails
raise RuntimeError("gitea hiccup")
return super().create_issue(*a, **kw)
intake = make_intake(core, tmp_path, FlakyGitea())
r = intake.submit_batch(client_id, TASKS)
assert r["ok"] is False and r["reason"] == "intake_failed"
# no batch row left behind
assert intake.get_batch(1)["ok"] is False
# -------------------------------------------------------------------------- #
# Polling #
# -------------------------------------------------------------------------- #
class TestPollCompleted:
def test_poll_stores_closed_results_and_skips_open(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
n1 = r["issues"][0]["issue_number"]
gitea.add_comment("eval-runs", n1, "lane report: all checks green")
gitea.close_issue("eval-runs", n1) # task 1 drained; task 2 still open
out = intake.poll_completed(r["batch_id"])
assert out["done"] == 1 and out["pending"] == 1
batch = intake.get_batch(r["batch_id"])
done_task = [t for t in batch["tasks"] if t["task_id"] == "t1"][0]
pending_task = [t for t in batch["tasks"] if t["task_id"] == "t2"][0]
assert done_task["status"] == "done"
result = done_task["result_json"]
assert result["summary"] == "lane report: all checks green"
assert result["labels"] == [EVAL_LABEL]
assert pending_task["status"] == "pending"
assert batch["batch"]["status"] == "open" # not all done yet
def test_poll_marks_batch_done_when_all_closed(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
for issue in r["issues"]:
gitea.add_comment("eval-runs", issue["issue_number"], "ok")
gitea.close_issue("eval-runs", issue["issue_number"])
out = intake.poll_completed(r["batch_id"])
assert out["done"] == 2 and out["pending"] == 0
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "done"
def test_poll_unknown_batch(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path)
out = intake.poll_completed(123)
assert out["ok"] is False and out["error"] == "unknown_batch"
# -------------------------------------------------------------------------- #
# Emitting #
# -------------------------------------------------------------------------- #
class TestEmit:
def test_emit_posts_report_to_callback_with_retry(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
# fails twice (500s), succeeds on the third attempt
emitter = MockEmitter(fail_first=2)
intake = make_intake(core, tmp_path, gitea, emitter=emitter)
r = intake.submit_batch(client_id, TASKS,
callback_url="https://client.example/hook")
for issue in r["issues"]:
gitea.add_comment("eval-runs", issue["issue_number"],
"done: all checks passed")
gitea.close_issue("eval-runs", issue["issue_number"])
intake.poll_completed(r["batch_id"])
out = intake.emit(r["batch_id"])
assert out["ok"] is True and out["emitted"] is True
assert emitter.push_count == 3 # retry x3 total
assert len(emitter.push_payloads) == 1 # delivered once
report, url = emitter.push_payloads[0]
assert url == "https://client.example/hook"
assert report["batch_id"] == r["batch_id"]
assert report["client_id"] == client_id
assert [res["task_id"] for res in report["results"]] == ["t1", "t2"]
assert all(res["status"] == "done" for res in report["results"])
assert all("summary" in res for res in report["results"])
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "emitted"
def test_emit_file_mode_writes_json_drop(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea,
output_dir=tmp_path / "drops")
r = intake.submit_batch(client_id, TASKS) # no callback_url
for issue in r["issues"]:
gitea.close_issue("eval-runs", issue["issue_number"])
intake.poll_completed(r["batch_id"])
out = intake.emit(r["batch_id"])
assert out["ok"] is True and out["emitted"] is True
drop = tmp_path / "drops" / f"{r['batch_id']}.json"
assert drop.exists()
payload = json.loads(drop.read_text(encoding="utf-8"))
assert payload["batch_id"] == r["batch_id"]
assert len(payload["results"]) == 2
def test_emit_refuses_before_all_tasks_done(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path)
r = intake.submit_batch(client_id, TASKS)
out = intake.emit(r["batch_id"])
assert out["ok"] is False and out["reason"] == "tasks_pending"
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "open"
def test_reemit_is_idempotent_no_double_post(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
emitter = MockEmitter()
intake = make_intake(core, tmp_path, gitea, emitter=emitter)
r = intake.submit_batch(client_id, TASKS,
callback_url="https://cb.example/x")
for issue in r["issues"]:
gitea.close_issue("eval-runs", issue["issue_number"])
intake.poll_completed(r["batch_id"])
assert intake.emit(r["batch_id"])["ok"] is True
pushes_after_first = emitter.push_count
again = intake.emit(r["batch_id"])
assert again["ok"] is True and again.get("idempotent") is True
assert emitter.push_count == pushes_after_first # no second POST
def test_callback_failure_keeps_batch_emittable(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
emitter = MockEmitter(fail_first=99) # always fails
intake = make_intake(core, tmp_path, gitea, emitter=emitter)
r = intake.submit_batch(client_id, TASKS,
callback_url="https://cb.example/x")
for issue in r["issues"]:
gitea.close_issue("eval-runs", issue["issue_number"])
intake.poll_completed(r["batch_id"])
out = intake.emit(r["batch_id"])
assert out["ok"] is False and out["reason"] == "emit_failed"
assert out["attempts"] == 3
assert intake.get_batch(r["batch_id"])["batch"]["status"] == "done"
# -------------------------------------------------------------------------- #
# Persistence + config #
# -------------------------------------------------------------------------- #
class TestPersistence:
def test_batch_survives_intake_reopen(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
intake2 = make_intake(core, tmp_path, gitea) # fresh connection
batch = intake2.get_batch(r["batch_id"])
assert batch["ok"] and batch["batch"]["client_id"] == client_id
assert len(batch["tasks"]) == 2
def test_shared_repo_name_from_config(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea, eval_repo="qa-runs")
r = intake.submit_batch(client_id, TASKS)
assert r["repo"] == "qa-runs"
assert "qa-runs" in gitea.created_repos
def test_per_batch_repo_mode(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea, per_batch_repos=True)
r = intake.submit_batch(client_id, TASKS)
assert r["repo"] == f"eval-{client_id}-{r['batch_id']}"
assert r["repo"] in gitea.created_repos
def test_real_client_list_issue_comments_builds_correct_request(monkeypatch):
"""The REAL GiteaClient (not the in-memory fake) must request
GET /api/v1/repos/{owner}/{repo}/issues/{n}/comments with the token.
This guards the exact bug class the !20 bounce caught: the fake carried
list_issue_comments while the real (canonical, issue #4) client did not,
and the suite stayed green because nothing exercised the real HTTP path.
Network-free: urlopen is captured, never invoked.
NOTE: ``gitea_saas/gitea_client.py`` is owned by issue #4 (!17), not #8;
this branch must not fork it. If the canonical client has not yet grown
``list_issue_comments`` (extension requested on #4), this guard skips
rather than failing — it re-activates automatically once !17 lands.
"""
try:
from gitea_saas.gitea_client import GiteaClient
except ImportError:
pytest.skip("gitea_saas/gitea_client.py ships via issue #4 (!17); "
"not present until !17 lands")
if not hasattr(GiteaClient, "list_issue_comments"):
pytest.skip("canonical client (issue #4) lacks list_issue_comments; "
"extension requested on #4 — guard activates once landed")
captured = {}
class FakeResponse:
status = 200
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def read(self):
return b'[{"body": "lane report", "id": 1}]'
def fake_urlopen(req, timeout=None):
captured["req"] = req
return FakeResponse()
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
client = GiteaClient(base_url="https://gitea.example",
token="tok123", owner="pbwraith")
comments = client.list_issue_comments("eval-runs", 42)
assert comments == [{"body": "lane report", "id": 1}]
req = captured["req"]
assert req.method == "GET"
assert req.full_url == (
"https://gitea.example/api/v1/repos/pbwraith/eval-runs/"
"issues/42/comments")
assert req.headers.get("Authorization") == "token tok123"
assert req.headers.get("Accept") == "application/json"
# -------------------------------------------------------------------------- #
# Issue #39: HTTP surface (register_eval_routes) #
# -------------------------------------------------------------------------- #
@pytest.fixture
def app_env(tmp_path):
"""core + MockGitea + api-keyed Config for mounting eval routes."""
core = BillingCore(tmp_path / "billing.db")
gitea = MockGitea()
cfg = Config(tmp_path / "saas.db", api_key="test-api-key",
debit_floor_minor=500)
return core, gitea, cfg
@pytest.fixture
def eval_app(app_env):
from flask import Flask
core, gitea, cfg = app_env
app = Flask(__name__)
app.config["API_KEY"] = cfg.api_key
app.config["TESTING"] = True
register_eval_routes(app, core, gitea, cfg)
return app.test_client()
def _balance(core, client_id):
ok, details = core.get_client(client_id, include_balance=True)
assert ok
return details["balance"]
def _queued_batch_rows(db_path, client_id):
conn = sqlite3.connect(str(db_path))
try:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT * FROM pending_eval_batches WHERE client_id = ?"
" ORDER BY id", (client_id,)).fetchall()
return [dict(r) for r in rows]
finally:
conn.close()
class TestEvalRoutes:
"""Acceptance 1: each route — success, missing/blank key 401,
unconfigured key 503; health stays public."""
AUTHD_PATHS = [
("post", "/api/eval/batches", {"client_id": 1, "tasks": TASKS}),
("get", "/api/eval/batches/1", None),
("get", "/api/eval/batches/1/results", None),
("post", "/api/eval/batches/1/cancel", None),
]
KEY = "test-api-key"
@staticmethod
def _post(client, path, payload=None):
return client.post(path, json=payload or {},
headers={"X-API-Key": TestEvalRoutes.KEY})
@staticmethod
def _get(client, path):
return client.get(path,
headers={"X-API-Key": TestEvalRoutes.KEY})
def test_eval_health_is_public(self, eval_app):
resp = eval_app.get("/api/eval/health")
assert resp.status_code == 200
assert resp.get_json() == {"ok": True}
def test_eval_health_public_without_any_key(self, app_env):
core, gitea, cfg = app_env
from flask import Flask
app = Flask(__name__)
app.config["TESTING"] = True
register_eval_routes(app, core, gitea, cfg) # API_KEY unset
resp = app.test_client().get("/api/eval/health")
assert resp.status_code == 200
def test_missing_key_401_on_every_route(self, eval_app):
for method, path, _payload in self.AUTHD_PATHS:
resp = getattr(eval_app, method)(path, json={})
assert resp.status_code == 401, (method, path)
# canonical #24 body, byte-identical with every front-end
assert resp.get_json() == UNAUTHORIZED_BODY
def test_blank_key_401_on_every_route(self, eval_app):
for method, path, _payload in self.AUTHD_PATHS:
resp = getattr(eval_app, method)(
path, json={}, headers={"X-API-Key": ""})
assert resp.status_code == 401, (method, path)
assert resp.get_json() == UNAUTHORIZED_BODY
def test_wrong_key_401(self, eval_app):
resp = eval_app.post(
"/api/eval/batches", json={"client_id": 1, "tasks": TASKS},
headers={"X-API-Key": "wrong-key"})
assert resp.status_code == 401
assert resp.get_json() == UNAUTHORIZED_BODY
def test_unconfigured_key_503_on_every_route(self, app_env):
"""No API key configured -> authed routes fail closed 503
(issue #38's assembly acceptance demands this app-wide); the
liveness route stays reachable."""
core, gitea, cfg = app_env
cfg_no_key = Config(cfg.db_path) # no api_key attr at all
from flask import Flask
app = Flask(__name__)
app.config["TESTING"] = True
register_eval_routes(app, core, gitea, cfg_no_key)
client = app.test_client()
for method, path, payload in self.AUTHD_PATHS:
resp = getattr(client, method)(path, json=payload or {})
assert resp.status_code == 503, (method, path)
assert resp.get_json()["error"]["code"] == "SERVICE_UNAVAILABLE"
assert client.get("/api/eval/health").status_code == 200
def test_submit_funded_route_201(self, app_env, eval_app):
core, gitea, _cfg = app_env
ok, c = core.create_client("route-client")
assert ok
core.credit(c["id"], 10_000, "fund-route")
resp = self._post(eval_app, "/api/eval/batches", {
"client_id": c["id"], "tasks": TASKS})
assert resp.status_code == 201
body = resp.get_json()
assert body["ok"] is True and body["batch_id"] == 1
assert len(body["issues"]) == 2
# the batch hold is visible in the client's available balance
assert _balance(core, c["id"]) == 10_000 - 500
def test_submit_unfunded_route_202_queued(self, app_env, eval_app):
core, gitea, _cfg = app_env
ok, c = core.create_client("poor-route")
assert ok
resp = self._post(eval_app, "/api/eval/batches", {
"client_id": c["id"], "tasks": TASKS})
assert resp.status_code == 202
body = resp.get_json()
assert body["ok"] is False and body["reason"] == "unfunded"
assert body["queued"] is True and body["pending_id"] == 1
assert gitea.created_issues == [] # NO Gitea issues
assert gitea.created_repos == []
rows = _queued_batch_rows(_cfg.db_path, c["id"])
assert len(rows) == 1 and rows[0]["status"] == "queued"
def test_route_validation_400_and_404(self, eval_app, app_env):
core, gitea, _cfg = app_env
ok, c = core.create_client("val-client")
assert ok
core.credit(c["id"], 10_000, "fund-val")
bad = self._post(eval_app, "/api/eval/batches", {
"client_id": -3, "tasks": TASKS})
assert bad.status_code == 400 and bad.get_json()["code"] == "invalid_request"
bad_tasks = self._post(eval_app, "/api/eval/batches", {
"client_id": c["id"], "tasks": [{"task_id": "x"}]}) # no title
assert bad_tasks.status_code == 400
assert self._get(eval_app, "/api/eval/batches/99").status_code == 404
assert self._get(eval_app, "/api/eval/batches/99/results").status_code == 404
assert self._post(eval_app, "/api/eval/batches/99/cancel").status_code == 404
def test_get_batch_and_results_routes(self, app_env, eval_app):
core, gitea, _cfg = app_env
ok, c = core.create_client("view-client")
assert ok
core.credit(c["id"], 10_000, "fund-view")
r = self._post(eval_app, "/api/eval/batches", {
"client_id": c["id"], "tasks": TASKS}).get_json()
batch_id = r["batch_id"]
# status + per-task state
resp = self._get(eval_app, f"/api/eval/batches/{batch_id}")
assert resp.status_code == 200
body = resp.get_json()
assert body["ok"] and body["batch"]["status"] == "open"
assert [t["task_id"] for t in body["tasks"]] == ["t1", "t2"]
assert all(t["status"] == "pending" for t in body["tasks"])
# the results table reflects the drain state: close t1 in gitea,
# then run the drain step (poll) through a direct intake on the
# same DB — the results route renders the persisted table
gitea.add_comment("eval-runs", r["issues"][0]["issue_number"],
"lane report: green")
gitea.close_issue("eval-runs", r["issues"][0]["issue_number"])
intake = EvalIntake(core, gitea, _cfg, sleep_fn=noop_sleep)
assert intake.poll_completed(batch_id)["done"] == 1
results = self._get(
eval_app, f"/api/eval/batches/{batch_id}/results").get_json()
assert results["ok"] is True
by_task = {t["task_id"]: t for t in results["results"]}
assert by_task["t1"]["status"] == "done"
assert by_task["t1"]["summary"] == "lane report: green"
# t2 was never drained, so the results table reports it pending.
assert by_task["t2"]["status"] == "pending"
def test_cancel_route_releases_hold_exactly(self, app_env, eval_app):
"""Acceptance 3 (route half): balance returns to its pre-submit
value exactly after cancel."""
core, gitea, _cfg = app_env
ok, c = core.create_client("cancel-client")
assert ok
core.credit(c["id"], 10_000, "fund-cancel")
pre = _balance(core, c["id"])
r = self._post(eval_app, "/api/eval/batches", {
"client_id": c["id"], "tasks": TASKS}).get_json()
assert _balance(core, c["id"]) == pre - 500 # hold placed
resp = self._post(eval_app, f"/api/eval/batches/{r['batch_id']}/cancel")
assert resp.status_code == 200
body = resp.get_json()
assert body["ok"] is True and body["cancelled"] is True
assert body["hold_released"] is True
assert _balance(core, c["id"]) == pre # exact return
# the batch now reads as cancelled
status = self._get(
eval_app, f"/api/eval/batches/{r['batch_id']}").get_json()
assert status["cancelled"] is True
assert status["batch"]["status"] == "cancelled"
# -------------------------------------------------------------------------- #
# Issue #39: queue + flush (acceptance 2) and hold/cancel (acceptance 3) #
# -------------------------------------------------------------------------- #
class TestQueueAndFlush:
def test_unfunded_submit_queues_no_gitea(self, unfunded, tmp_path):
core, client_id = unfunded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
assert r["ok"] is False and r["reason"] == "unfunded"
assert r["queued"] is True and r["pending_id"] == 1
# half 1: NO Gitea issue, NO repo
assert gitea.created_issues == []
assert gitea.created_repos == []
# half 1: a queued row exists locally
rows = _queued_batch_rows(tmp_path / "saas.db", client_id)
assert len(rows) == 1
assert rows[0]["status"] == "queued"
assert json.loads(rows[0]["tasks_json"]) == TASKS
def test_flush_ships_queued_batch_after_funding(self, unfunded, tmp_path):
core, client_id = unfunded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
assert r["pending_id"] == 1
# funding arrives; flush ships it
core.credit(client_id, 10_000, "fund-flush")
out = intake.flush_pending(client_id)
assert out["ok"] is True and out["flushed"] == 1
assert out["batch_ids"] == [1]
assert len(out["issues"]) == 2
# half 2: issues now exist, one per task, eval-task label
assert len(gitea.created_issues) == 2
for _repo, _title, body, labels in gitea.created_issues:
assert labels == [EVAL_LABEL]
assert "batch: 1" in body
# the queued row is marked shipped; a real batch row exists
rows = _queued_batch_rows(tmp_path / "saas.db", client_id)
assert len(rows) == 1 and rows[0]["status"] == "shipped"
batch = intake.get_batch(1)
assert batch["ok"] and batch["batch"]["client_id"] == client_id
assert len(batch["tasks"]) == 2
# the shipped batch placed its hold like any fresh submit
assert _balance(core, client_id) == 10_000 - 500
def test_flush_while_still_unfunded_skips(self, unfunded, tmp_path):
core, client_id = unfunded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
intake.submit_batch(client_id, TASKS)
out = intake.flush_pending(client_id)
assert out["ok"] is False and out["reason"] == "unfunded"
assert out["skipped"] == 1 and out["flushed"] == 0
assert gitea.created_issues == []
rows = _queued_batch_rows(tmp_path / "saas.db", client_id)
assert rows[0]["status"] == "queued"
def test_flush_nothing_queued_is_noop(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path)
out = intake.flush_pending(client_id)
assert out == {"ok": True, "flushed": 0, "batch_ids": [], "issues": []}
class TestHoldAndCancel:
def test_cancel_releases_hold_balance_exact(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path)
pre = _balance(core, client_id)
r = intake.submit_batch(client_id, TASKS)
assert r["hold_id"] is not None
assert _balance(core, client_id) == pre - 500 # hold visible
out = intake.cancel_batch(r["batch_id"])
assert out["ok"] is True and out["cancelled"] is True
assert out["hold_released"] is True
assert _balance(core, client_id) == pre # EXACT return
def test_cancel_idempotent_no_double_release(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path)
pre = _balance(core, client_id)
r = intake.submit_batch(client_id, TASKS)
assert intake.cancel_batch(r["batch_id"])["ok"] is True
assert _balance(core, client_id) == pre
again = intake.cancel_batch(r["batch_id"])
assert again["ok"] is True and again["already_cancelled"] is True
assert again["hold_released"] is False
assert _balance(core, client_id) == pre # no second release
def test_cancel_unknown_batch(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path)
out = intake.cancel_batch(999)
assert out == {"ok": False, "error": "unknown_batch"}
def test_cancel_stops_draining(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea, emitter=MockEmitter())
r = intake.submit_batch(client_id, TASKS,
callback_url="https://cb.example/x")
for issue in r["issues"]:
gitea.close_issue("eval-runs", issue["issue_number"])
assert intake.cancel_batch(r["batch_id"])["ok"] is True
# polling and emitting both refuse a cancelled batch
assert intake.poll_completed(r["batch_id"]) == {
"ok": False, "error": "cancelled"}
assert intake.emit(r["batch_id"]) == {
"ok": False, "reason": "cancelled"}
# and the batch reads as cancelled
status = intake.get_batch(r["batch_id"])
assert status["cancelled"] is True
assert status["batch"]["status"] == "cancelled"
assert "cancelled_at" in status["batch"]
def test_zero_hold_disables_hold(self, funded, tmp_path):
core, client_id = funded
intake = make_intake(core, tmp_path, eval_batch_hold_minor=0)
pre = _balance(core, client_id)
r = intake.submit_batch(client_id, TASKS)
assert r.get("hold_id") is None
assert _balance(core, client_id) == pre # nothing held
out = intake.cancel_batch(r["batch_id"])
assert out["ok"] is True and out["hold_released"] is False
assert _balance(core, client_id) == pre
def test_intake_failure_releases_hold(self, funded, tmp_path):
"""A rolled-back submit must not leave money locked."""
core, client_id = funded
class FlakyGitea(MockGitea):
def create_issue(self, *a, **kw):
if self.next_number >= 1: # second issue fails
raise RuntimeError("gitea hiccup")
return super().create_issue(*a, **kw)
intake = make_intake(core, tmp_path, FlakyGitea())
pre = _balance(core, client_id)
r = intake.submit_batch(client_id, TASKS)
assert r["ok"] is False and r["reason"] == "intake_failed"
assert intake.get_batch(1)["ok"] is False # no partial batch
assert _balance(core, client_id) == pre # hold released
def test_hold_too_large_refuses_batch(self, funded, tmp_path):
"""Reserving beyond the available balance fails closed — the
batch is refused and no Gitea issue is created."""
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea,
eval_batch_hold_minor=50_000) # > 10_000
pre = _balance(core, client_id)
r = intake.submit_batch(client_id, TASKS)
assert r["ok"] is False and r["reason"] == "intake_failed"
assert gitea.created_issues == []
assert _balance(core, client_id) == pre
def test_get_results_after_partial_drain(self, funded, tmp_path):
core, client_id = funded
gitea = MockGitea()
intake = make_intake(core, tmp_path, gitea)
r = intake.submit_batch(client_id, TASKS)
gitea.add_comment("eval-runs", r["issues"][0]["issue_number"],
"ok")
gitea.close_issue("eval-runs", r["issues"][0]["issue_number"])
intake.poll_completed(r["batch_id"])
results = intake.get_results(r["batch_id"])
assert results["ok"] is True
assert results["batch_id"] == r["batch_id"]
by_task = {t["task_id"]: t for t in results["results"]}
assert by_task["t1"]["status"] == "done"
assert by_task["t1"]["summary"] == "ok"
assert by_task["t2"]["status"] == "pending"
assert intake.get_results(999)["error"] == "unknown_batch"