test_cost_awareness_e2e.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """End-to-end tests for Track P / P-cost (§PC.9).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | from io import StringIO |
| 7 | from pathlib import Path |
| 8 | from unittest.mock import patch |
| 9 | |
| 10 | import yaml |
| 11 | |
| 12 | from cli.context import CliContext |
| 13 | from cli.kit_root import kit_root |
| 14 | from cli.main import main |
| 15 | from cli.output import OutputContext |
| 16 | from tests.fixtures.cost_awareness import seed_cost_e2e_repo |
| 17 | from tests.fixtures.model_routing import write_routing_policy |
| 18 | from tests.support import git_status_runner, make_runner, ok, run_cli |
| 19 | |
| 20 | |
| 21 | def test_e2e_thinking_freeze_slice_paid_high(tmp_path: Path) -> None: |
| 22 | seed_cost_e2e_repo(tmp_path) |
| 23 | out = StringIO() |
| 24 | with patch("sys.stdout", out): |
| 25 | code = main( |
| 26 | ["status", "--json"], |
| 27 | ctx=CliContext.create( |
| 28 | cwd=tmp_path, |
| 29 | runner=git_status_runner(), |
| 30 | output=OutputContext(json_mode=True), |
| 31 | kit=kit_root(), |
| 32 | ), |
| 33 | ) |
| 34 | assert code == 0 |
| 35 | payload = json.loads(out.getvalue()) |
| 36 | slices = payload["cost_awareness"]["slices"] |
| 37 | thinking = next(item for item in slices if item["phase_id"] == "Demo Thinking freeze") |
| 38 | assert thinking["model_tier"] == "deep-reasoning" |
| 39 | assert thinking["cost_class"] == "high" |
| 40 | assert thinking["paid_step_before_spend"] is True |
| 41 | |
| 42 | |
| 43 | def test_e2e_auto_slice_moderate(tmp_path: Path) -> None: |
| 44 | seed_cost_e2e_repo(tmp_path) |
| 45 | out = StringIO() |
| 46 | with patch("sys.stdout", out): |
| 47 | code = main( |
| 48 | ["status", "--json"], |
| 49 | ctx=CliContext.create( |
| 50 | cwd=tmp_path, |
| 51 | runner=git_status_runner(), |
| 52 | output=OutputContext(json_mode=True), |
| 53 | kit=kit_root(), |
| 54 | ), |
| 55 | ) |
| 56 | assert code == 0 |
| 57 | payload = json.loads(out.getvalue()) |
| 58 | slices = payload["cost_awareness"]["slices"] |
| 59 | auto = next(item for item in slices if item["phase_id"] == "Demo Auto build") |
| 60 | assert auto["model_tier"] == "standard" |
| 61 | assert auto["cost_class"] == "moderate" |
| 62 | assert auto["paid_step_before_spend"] is True |
| 63 | |
| 64 | |
| 65 | def test_e2e_local_offline_unpaid(tmp_path: Path) -> None: |
| 66 | seed_cost_e2e_repo(tmp_path) |
| 67 | write_routing_policy( |
| 68 | tmp_path, |
| 69 | """ |
| 70 | version: 1 |
| 71 | defaults: |
| 72 | model_tier: local-offline |
| 73 | fallback: [local-offline, standard, human] |
| 74 | routes: |
| 75 | - id: freeze-thinking |
| 76 | when: { gate: freeze_review, phase_tier: thinking } |
| 77 | model_tier: deep-reasoning |
| 78 | fallback: [deep-reasoning, human] |
| 79 | - id: auto-build |
| 80 | when: { phase_tier: auto } |
| 81 | model_tier: standard |
| 82 | fallback: [standard, human] |
| 83 | """, |
| 84 | ) |
| 85 | out = StringIO() |
| 86 | with patch("sys.stdout", out): |
| 87 | code = main( |
| 88 | ["status", "--json"], |
| 89 | ctx=CliContext.create( |
| 90 | cwd=tmp_path, |
| 91 | runner=git_status_runner(), |
| 92 | output=OutputContext(json_mode=True), |
| 93 | kit=kit_root(), |
| 94 | ), |
| 95 | ) |
| 96 | assert code == 0 |
| 97 | payload = json.loads(out.getvalue()) |
| 98 | offline = next(item for item in payload["cost_awareness"]["slices"] if item["phase_id"] == "Demo Offline worker") |
| 99 | assert offline["model_tier"] == "local-offline" |
| 100 | assert offline["cost_class"] == "free" |
| 101 | assert offline["paid_step_before_spend"] is False |
| 102 | |
| 103 | |
| 104 | def test_e2e_governance_sync_footer(tmp_path: Path) -> None: |
| 105 | seed_cost_e2e_repo(tmp_path) |
| 106 | runner = make_runner( |
| 107 | { |
| 108 | "git rev-parse --abbrev-ref HEAD": ok("main"), |
| 109 | "git status --porcelain": ok(""), |
| 110 | "git rev-parse origin/main": ok("cafebabe"), |
| 111 | "gh pr list --state merged --limit 5 --json number,title,mergeCommit,mergedAt": ok("[]"), |
| 112 | "git remote get-url origin": ok("[email protected]:owner/repo.git"), |
| 113 | } |
| 114 | ) |
| 115 | out = StringIO() |
| 116 | with patch("sys.stdout", out): |
| 117 | code = main( |
| 118 | ["governance-sync"], |
| 119 | ctx=CliContext.create(cwd=tmp_path, runner=runner, kit=kit_root()), |
| 120 | ) |
| 121 | assert code == 0 |
| 122 | assert "cost_awareness:" in out.getvalue() |
| 123 | |
| 124 | |
| 125 | def test_e2e_git_only_and_muse_regime_identical(tmp_path: Path) -> None: |
| 126 | seed_cost_e2e_repo(tmp_path) |
| 127 | from adapters.config import load_config |
| 128 | from tools.cost_awareness.surface import build_cost_awareness_report |
| 129 | |
| 130 | config_git = load_config(tmp_path / ".overseer" / "config.yaml") |
| 131 | handover = (tmp_path / "docs" / "OVERSEER-HANDOVER.md").read_text(encoding="utf-8") |
| 132 | roadmap = (tmp_path / "docs" / "ROADMAP.md").read_text(encoding="utf-8") |
| 133 | git_report = build_cost_awareness_report( |
| 134 | config_git, tmp_path, kit_root=kit_root(), handover_text=handover, roadmap_text=roadmap |
| 135 | ) |
| 136 | |
| 137 | cfg = tmp_path / ".overseer" / "config.yaml" |
| 138 | data = yaml.safe_load(cfg.read_text(encoding="utf-8")) |
| 139 | data["vcs"]["regime"] = "muse+git-mirror" |
| 140 | data["vcs"]["canonical"] = "muse" |
| 141 | data["vcs"]["git"]["mirror_branch"] = "muse-mirror" |
| 142 | data["vcs"]["muse"]["staging_remote"] = "origin" |
| 143 | data["vcs"]["muse"]["main_branch"] = "main" |
| 144 | cfg.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") |
| 145 | config_muse = load_config(cfg) |
| 146 | muse_report = build_cost_awareness_report( |
| 147 | config_muse, tmp_path, kit_root=kit_root(), handover_text=handover, roadmap_text=roadmap |
| 148 | ) |
| 149 | assert git_report.slices == muse_report.slices |