host
0db51f8145
- cloud/attempt_ledger.py: separate task_attempts table (not tasks columns) with per-attempt rows, MAX+1 attempt_no, append-only intent; interrupted rows kept but never counted; is_defect is explicit, never inferred from failure; first defect category wins; no-ledger tasks stay unknown (None), never backfilled. SQLite DDL is CREATE TABLE IF NOT EXISTS (idempotent); PG_DDL is the verbatim manual step for production Postgres, kept out of POSTGRES_SCHEMA on purpose (migrations coverage untouched). - cloud/server.py GET /metrics-report: aggregates the ledger into compute_metrics records; missing ledger table degrades to unknown rates, not an error. Cost dual-column code untouched. - cloud/metrics.py: docstring only (ledger is now the source); compute_metrics logic and all #238 semantics unchanged. - Tests: 16 ledger tests (store/aggregate/metrics/PG adapter evidence) + 3 endpoint tests (real rates, old task beside ledger task, interrupted status non-terminal).
202 lines
8.0 KiB
Python
202 lines
8.0 KiB
Python
"""Endpoint test for GET /metrics-report.
|
|
|
|
Only the standard library http.client; a temporary SQLite DB and a
|
|
temporary port; no real model, no network, no secrets.
|
|
|
|
Covers: unauthenticated -> 401, empty project -> report shape with
|
|
unknown rates (None, not 0) and pending golden set, a completed
|
|
task with a priced cost row -> terminal count and dual-column cost,
|
|
and (audit gap 3) ledger-backed real rates, old tasks staying
|
|
unknown beside ledger tasks, and interrupted status staying
|
|
non-terminal even with ledger rows.
|
|
"""
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.auth import AuthStore
|
|
from cloud.cost import CostStore
|
|
from cloud.server import create_server
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.auth import AuthStore
|
|
from cloud.cost import CostStore
|
|
from cloud.server import create_server
|
|
|
|
|
|
class MetricsReportEndpointTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
self.auth = AuthStore(self.db_path)
|
|
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
|
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
|
self.port = self.server.server_address[1]
|
|
self.thread = threading.Thread(
|
|
target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
status, self.project = self._req(
|
|
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
|
self.assertEqual(status, 201)
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=5)
|
|
try:
|
|
self.auth.conn.close()
|
|
except Exception:
|
|
pass
|
|
self.tmp.cleanup()
|
|
|
|
def _req(self, method, path, body=None, key=None):
|
|
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
|
hdrs = {}
|
|
payload = None
|
|
if body is not None:
|
|
payload = json.dumps(body).encode("utf-8")
|
|
hdrs["Content-Type"] = "application/json"
|
|
if key is not None:
|
|
hdrs["X-API-Key"] = key
|
|
try:
|
|
conn.request(method, path, body=payload, headers=hdrs)
|
|
resp = conn.getresponse()
|
|
data = resp.read()
|
|
finally:
|
|
conn.close()
|
|
try:
|
|
parsed = json.loads(data.decode("utf-8")) if data else None
|
|
except ValueError:
|
|
parsed = None
|
|
return resp.status, parsed
|
|
|
|
def test_unauthenticated_is_401(self):
|
|
status, _ = self._req("GET", "/metrics-report")
|
|
self.assertEqual(status, 401)
|
|
|
|
def test_empty_project_report_is_pending_and_unknown(self):
|
|
status, report = self._req(
|
|
"GET", "/metrics-report", key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertFalse(report["decision_ready"])
|
|
self.assertEqual(report["golden_set"]["status"],
|
|
"pending_human_approval")
|
|
self.assertIsNone(report["metrics"]["first_pass_rate"])
|
|
self.assertIsNone(report["metrics"]["defect_rate"])
|
|
self.assertIsNone(report["metrics"]["total_usd_micros"])
|
|
self.assertEqual(report["thresholds_source"],
|
|
"default_unapproved")
|
|
|
|
def test_completed_task_with_priced_cost_row(self):
|
|
from cloud.db import connect as db_connect
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
|
|
("t-1", self.project["id"], "completed"))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
store = CostStore(self.db_path)
|
|
store.record("t-1", self.project["id"], "gpt-6-sol",
|
|
input_tokens=1000, output_tokens=100,
|
|
usd_micros=5000, source="actual")
|
|
status, report = self._req(
|
|
"GET", "/metrics-report", key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(report["metrics"]["terminal_tasks"], 1)
|
|
# No ledger rows for this task (and no ledger table yet):
|
|
# first-pass stays unknown, it must not be fabricated as
|
|
# 0 or 1 from status alone.
|
|
self.assertIsNone(report["metrics"]["first_pass_rate"])
|
|
self.assertEqual(report["metrics"]["total_usd_micros"], 5000)
|
|
self.assertTrue(report["metrics"]["cost_complete"])
|
|
|
|
def test_ledger_task_gives_real_rates(self):
|
|
from cloud.attempt_ledger import record_attempt
|
|
from cloud.db import connect as db_connect
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
|
|
("t-led", self.project["id"], "completed"))
|
|
conn.commit()
|
|
record_attempt(conn, "t-led", "success")
|
|
finally:
|
|
conn.close()
|
|
status, report = self._req(
|
|
"GET", "/metrics-report", key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
metrics = report["metrics"]
|
|
self.assertEqual(metrics["terminal_tasks"], 1)
|
|
# A ledger row exists, so 0 defects here is a real 0, not
|
|
# a fabricated one: defect_known is True alongside 0.0.
|
|
self.assertAlmostEqual(metrics["first_pass_rate"], 1.0)
|
|
self.assertTrue(metrics["first_pass_known"])
|
|
self.assertAlmostEqual(metrics["defect_rate"], 0.0)
|
|
self.assertTrue(metrics["defect_known"])
|
|
self.assertAlmostEqual(metrics["rework_rate"], 0.0)
|
|
|
|
def test_old_task_without_ledger_does_not_dilute_real_rate(self):
|
|
from cloud.attempt_ledger import record_attempt
|
|
from cloud.db import connect as db_connect
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
|
|
("t-old", self.project["id"], "completed"))
|
|
conn.execute(
|
|
"INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
|
|
("t-new", self.project["id"], "merged"))
|
|
conn.commit()
|
|
record_attempt(conn, "t-new", "failure", is_defect=True,
|
|
defect_category="contract")
|
|
record_attempt(conn, "t-new", "success")
|
|
finally:
|
|
conn.close()
|
|
status, report = self._req(
|
|
"GET", "/metrics-report", key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
metrics = report["metrics"]
|
|
self.assertEqual(metrics["terminal_tasks"], 2)
|
|
# The old task is excluded from the denominator (known-only
|
|
# denominator, as in compute_metrics); it is neither counted
|
|
# as first-pass nor as defect-free.
|
|
self.assertAlmostEqual(metrics["first_pass_rate"], 0.0)
|
|
self.assertAlmostEqual(metrics["rework_rate"], 1.0)
|
|
self.assertAlmostEqual(metrics["defect_rate"], 1.0)
|
|
self.assertEqual(metrics["defect_by_category"],
|
|
{"contract": 1})
|
|
|
|
def test_interrupted_status_excluded_even_with_ledger_rows(self):
|
|
from cloud.attempt_ledger import record_attempt
|
|
from cloud.db import connect as db_connect
|
|
conn = db_connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
|
|
("t-int", self.project["id"], "interrupted"))
|
|
conn.commit()
|
|
record_attempt(conn, "t-int", "failure", is_defect=True,
|
|
defect_category="contract")
|
|
finally:
|
|
conn.close()
|
|
status, report = self._req(
|
|
"GET", "/metrics-report", key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
metrics = report["metrics"]
|
|
self.assertEqual(metrics["terminal_tasks"], 0)
|
|
self.assertIsNone(metrics["first_pass_rate"])
|
|
self.assertFalse(metrics["first_pass_known"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|