codewhale-hq/Codewhale
1
0 Fork 0
mirror of https://github.com/codewhale-hq/Codewhale.git synced 2026-11-11 01:51:34 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

test: ratchet source ownership structure

Browse Source 
Add an offline source-only budget for workspace packages, binary targets, thousand-line owned module identities, the largest owned Rust file, and aggregate owned Rust lines. Permit deletion and line-neutral consolidation while requiring explicit reviewed updates for new structural surface, then run the checker in the existing heavy CI gate.
This commit is contained in:
Hmbown committed 2026-07-30 10:46:06 -07:00
1 parent 1ff6e7872f
commit 38268346f0
4 files changed
+706
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.github/workflows/ci.yml
+6
View File
@@ -252,6 +252,12 @@ jobs:
- name: Check runtime-contract budget
if: needs.changes.outputs.heavy == 'true'
run: python3 scripts/check-runtime-contract-budget.py
# Source-only ownership ratchet. Deletion and line-neutral consolidation
# pass; new packages/binaries/thousand-line module paths, a larger maximum
# module, or aggregate owned Rust growth require reviewed updates.
- name: Check source-structure budget
if: needs.changes.outputs.heavy == 'true'
run: python3 scripts/check-source-structure-budget.py
- name: Check README translations stay in sync
if: github.event_name != 'schedule'
run: python3 scripts/check-readme-translations.py
scripts/check-source-structure-budget.py
Executable
+328
View File
@@ -0,0 +1,328 @@
#!/usr/bin/env python3
"""Enforce one-way ceilings on Codewhale's source ownership structure.
The budget permits deletion and line-neutral ownership moves freely. Adding a
workspace package or binary, creating a new thousand-line production Rust
module, increasing the largest owned file, or growing aggregate owned Rust
source requires an explicit budget update and review.
"""
from __future__ import annotations
import argparse
import json
import os
import stat
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Sequence
REPO_ROOT = Path(__file__).resolve().parent.parent
BUDGET_PATH = REPO_ROOT / "scripts" / "source-structure-budget.json"
DOCUMENT_KIND = "codewhale.source_structure_budget"
SCHEMA_VERSION = 1
LARGE_MODULE_THRESHOLD = 1_000
class StructureBudgetError(ValueError):
"""The budget or measured source structure is invalid."""
@dataclass(frozen=True)
class StructureSnapshot:
workspace_packages: tuple[str, ...]
binary_targets: tuple[str, ...]
module_lines: dict[str, int]
@dataclass(frozen=True)
class StructureBudget:
workspace_packages: tuple[str, ...]
binary_targets: tuple[str, ...]
allowed_large_modules: tuple[str, ...]
max_large_module_count: int
max_module_lines: int
max_total_owned_rust_lines: int
def production_rust_files(root: Path) -> list[Path]:
files = []
for path in (root / "crates").glob("*/src/**/*.rs"):
relative = path.relative_to(root)
if path.name == "tests.rs" or "tests" in relative.parts:
continue
files.append(path)
return sorted(files)
def source_line_count(path: Path) -> int:
return len(path.read_bytes().splitlines())
def cargo_metadata(root: Path) -> dict[str, Any]:
environment = os.environ.copy()
environment["CARGO_NET_OFFLINE"] = "true"
process = subprocess.run(
[
"cargo",
"metadata",
"--offline",
"--locked",
"--no-deps",
"--format-version",
"1",
],
cwd=root,
env=environment,
capture_output=True,
text=True,
check=False,
)
if process.returncode != 0:
sys.stderr.write(process.stderr)
raise StructureBudgetError(
f"cargo metadata failed with exit code {process.returncode}"
)
try:
document = json.loads(process.stdout)
except json.JSONDecodeError as error:
raise StructureBudgetError(f"cargo metadata emitted invalid JSON: {error}") from error
if not isinstance(document, dict) or not isinstance(document.get("packages"), list):
raise StructureBudgetError("cargo metadata did not contain a package list")
return document
def measure(root: Path = REPO_ROOT) -> StructureSnapshot:
metadata = cargo_metadata(root)
packages = tuple(sorted(package["name"] for package in metadata["packages"]))
binaries = tuple(
sorted(
f"{package['name']}:{target['name']}"
for package in metadata["packages"]
for target in package.get("targets", [])
if "bin" in target.get("kind", [])
)
)
module_lines = {
path.relative_to(root).as_posix(): source_line_count(path)
for path in production_rust_files(root)
}
return StructureSnapshot(packages, binaries, module_lines)
def require_string_list(document: dict[str, Any], field: str) -> tuple[str, ...]:
value = document.get(field)
if (
not isinstance(value, list)
or any(not isinstance(item, str) or not item for item in value)
or value != sorted(set(value))
):
raise StructureBudgetError(f"`{field}` must be sorted unique non-empty strings")
return tuple(value)
def require_non_negative_int(document: dict[str, Any], field: str) -> int:
value = document.get(field)
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise StructureBudgetError(f"`{field}` must be a non-negative integer")
return value
def validate_budget(document: dict[str, Any]) -> StructureBudget:
if document.get("document_kind") != DOCUMENT_KIND:
raise StructureBudgetError(f"document_kind must be `{DOCUMENT_KIND}`")
schema_version = document.get("schema_version")
if isinstance(schema_version, bool) or schema_version != SCHEMA_VERSION:
raise StructureBudgetError(f"schema_version must be {SCHEMA_VERSION}")
threshold = document.get("large_module_threshold_lines")
if isinstance(threshold, bool) or threshold != LARGE_MODULE_THRESHOLD:
raise StructureBudgetError(
f"large_module_threshold_lines must be {LARGE_MODULE_THRESHOLD}"
)
allowed_large_modules = require_string_list(document, "allowed_large_modules")
for path in allowed_large_modules:
if (
not path.startswith("crates/")
or Path(path).is_absolute()
or ".." in Path(path).parts
):
raise StructureBudgetError(f"invalid large-module path: {path!r}")
return StructureBudget(
require_string_list(document, "workspace_packages"),
require_string_list(document, "binary_targets"),
allowed_large_modules,
require_non_negative_int(document, "max_large_module_count"),
require_non_negative_int(document, "max_module_lines"),
require_non_negative_int(document, "max_total_owned_rust_lines"),
)
def load_budget(path: Path) -> tuple[dict[str, Any], StructureBudget]:
try:
document = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError as error:
raise StructureBudgetError(f"missing source-structure budget: {path}") from error
except (OSError, json.JSONDecodeError) as error:
raise StructureBudgetError(f"invalid source-structure budget {path}: {error}") from error
if not isinstance(document, dict):
raise StructureBudgetError("source-structure budget must be a JSON object")
return document, validate_budget(document)
def compare(current: StructureSnapshot, budget: StructureBudget) -> tuple[list[str], list[str]]:
failures: list[str] = []
improvements: list[str] = []
added_packages = sorted(set(current.workspace_packages) - set(budget.workspace_packages))
if added_packages:
failures.append(f"workspace packages added: {', '.join(added_packages)}")
removed_packages = sorted(set(budget.workspace_packages) - set(current.workspace_packages))
if removed_packages:
improvements.append(f"workspace packages removed: {', '.join(removed_packages)}")
added_binaries = sorted(set(current.binary_targets) - set(budget.binary_targets))
if added_binaries:
failures.append(f"binary targets added: {', '.join(added_binaries)}")
removed_binaries = sorted(set(budget.binary_targets) - set(current.binary_targets))
if removed_binaries:
improvements.append(f"binary targets removed: {', '.join(removed_binaries)}")
current_large = {
path for path, lines in current.module_lines.items() if lines >= LARGE_MODULE_THRESHOLD
}
allowed_large = set(budget.allowed_large_modules)
for path in sorted(current_large - allowed_large):
failures.append(
f"new thousand-line production module: {path} has "
f"{current.module_lines[path]} lines"
)
removed_large = sorted(allowed_large - current_large)
if removed_large:
improvements.append(
"large modules removed or split below "
f"{LARGE_MODULE_THRESHOLD}: {', '.join(removed_large)}"
)
large_count = len(current_large)
if large_count > budget.max_large_module_count:
failures.append(
f"large-module count grew: {large_count} > {budget.max_large_module_count}"
)
elif large_count < budget.max_large_module_count:
improvements.append(
f"large-module count shrank: {large_count} < {budget.max_large_module_count}"
)
largest = max(current.module_lines.values(), default=0)
if largest > budget.max_module_lines:
failures.append(f"largest module grew: {largest} > {budget.max_module_lines} lines")
elif largest < budget.max_module_lines:
improvements.append(f"largest module shrank: {largest} < {budget.max_module_lines} lines")
total = sum(current.module_lines.values())
if total > budget.max_total_owned_rust_lines:
failures.append(
"aggregate owned Rust source grew: "
f"{total} > {budget.max_total_owned_rust_lines} lines"
)
elif total < budget.max_total_owned_rust_lines:
improvements.append(
"aggregate owned Rust source shrank: "
f"{total} < {budget.max_total_owned_rust_lines} lines"
)
return failures, improvements
def budget_document(snapshot: StructureSnapshot) -> dict[str, Any]:
large_modules = sorted(
path for path, lines in snapshot.module_lines.items() if lines >= LARGE_MODULE_THRESHOLD
)
return {
"_comment": (
"One-way source ownership ceilings. Deletion and line-neutral ownership moves "
"pass; new packages, binaries, 1000-line module paths, a larger maximum module, "
"or aggregate owned Rust growth require an explicit reviewed update."
),
"allowed_large_modules": large_modules,
"binary_targets": list(snapshot.binary_targets),
"document_kind": DOCUMENT_KIND,
"large_module_threshold_lines": LARGE_MODULE_THRESHOLD,
"max_large_module_count": len(large_modules),
"max_module_lines": max(snapshot.module_lines.values(), default=0),
"max_total_owned_rust_lines": sum(snapshot.module_lines.values()),
"schema_version": SCHEMA_VERSION,
"workspace_packages": list(snapshot.workspace_packages),
}
def write_json_atomic(path: Path, document: dict[str, Any]) -> None:
mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else 0o644
path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(
prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
json.dump(document, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.chmod(temporary, mode)
os.replace(temporary, path)
except BaseException:
temporary.unlink(missing_ok=True)
raise
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--budget", type=Path, default=BUDGET_PATH, help=argparse.SUPPRESS)
parser.add_argument("--update", action="store_true", help="lock in current improvements")
args = parser.parse_args(argv)
try:
current = measure()
if args.update and not args.budget.exists():
write_json_atomic(args.budget, budget_document(current))
print(f"[source-structure-budget] initialized {args.budget}")
return 0
_document, budget = load_budget(args.budget)
failures, improvements = compare(current, budget)
except (OSError, StructureBudgetError) as error:
print(f"[source-structure-budget] ERROR: {error}", file=sys.stderr)
return 2
if failures:
print("[source-structure-budget] FAIL:", file=sys.stderr)
for failure in failures:
print(f" {failure}", file=sys.stderr)
return 1
if args.update:
try:
write_json_atomic(args.budget, budget_document(current))
except OSError as error:
print(f"[source-structure-budget] ERROR: {error}", file=sys.stderr)
return 2
print(
f"[source-structure-budget] tightened {args.budget}: "
f"{len(improvements)} improvement(s) locked in"
)
return 0
print(
"[source-structure-budget] PASS: "
f"{len(current.workspace_packages)} packages, "
f"{len(current.binary_targets)} binaries, "
f"{sum(lines >= LARGE_MODULE_THRESHOLD for lines in current.module_lines.values())} "
"large owned modules, "
f"{sum(current.module_lines.values())} owned Rust lines"
)
for improvement in improvements:
print(f" can tighten: {improvement}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/source-structure-budget.json
+209
View File
@@ -0,0 +1,209 @@
{
"_comment": "One-way source ownership ceilings. Deletion and line-neutral ownership moves pass; new packages, binaries, 1000-line module paths, a larger maximum module, or aggregate owned Rust growth require an explicit reviewed update.",
"allowed_large_modules": [
"crates/agent/src/lib.rs",
"crates/app-server/src/chat_completions.rs",
"crates/app-server/src/lib.rs",
"crates/cli/src/lib.rs",
"crates/cli/src/metrics.rs",
"crates/cli/src/update.rs",
"crates/config/src/lib.rs",
"crates/config/src/provider.rs",
"crates/config/src/user_constitution.rs",
"crates/config/src/xai_credentials.rs",
"crates/core/src/lib.rs",
"crates/execpolicy/src/lib.rs",
"crates/lane/src/control.rs",
"crates/lane/src/runtime.rs",
"crates/mcp/src/lib.rs",
"crates/protocol/src/fleet.rs",
"crates/secrets/src/lib.rs",
"crates/state/src/lib.rs",
"crates/tui/src/acp_server.rs",
"crates/tui/src/automation_manager.rs",
"crates/tui/src/child_env.rs",
"crates/tui/src/client.rs",
"crates/tui/src/client/anthropic.rs",
"crates/tui/src/client/chat.rs",
"crates/tui/src/client/prepared.rs",
"crates/tui/src/client/responses.rs",
"crates/tui/src/command_safety.rs",
"crates/tui/src/commands/groups/config/config.rs",
"crates/tui/src/commands/groups/core/constitution.rs",
"crates/tui/src/commands/groups/core/core.rs",
"crates/tui/src/commands/groups/project/init.rs",
"crates/tui/src/commands/groups/session/export.rs",
"crates/tui/src/commands/groups/session/session.rs",
"crates/tui/src/commands/groups/session/structcopy.rs",
"crates/tui/src/commands/groups/skills/skills.rs",
"crates/tui/src/commands/mod.rs",
"crates/tui/src/commands/user_registry.rs",
"crates/tui/src/compaction.rs",
"crates/tui/src/config.rs",
"crates/tui/src/config_persistence.rs",
"crates/tui/src/config_ui.rs",
"crates/tui/src/context_report.rs",
"crates/tui/src/core/engine.rs",
"crates/tui/src/core/engine/preview.rs",
"crates/tui/src/core/engine/tool_catalog.rs",
"crates/tui/src/core/engine/turn_loop.rs",
"crates/tui/src/cost_status.rs",
"crates/tui/src/fleet/exact.rs",
"crates/tui/src/fleet/executor.rs",
"crates/tui/src/fleet/host.rs",
"crates/tui/src/fleet/ledger.rs",
"crates/tui/src/fleet/manager.rs",
"crates/tui/src/fleet/profile.rs",
"crates/tui/src/fleet/scheduler.rs",
"crates/tui/src/fleet/task_spec.rs",
"crates/tui/src/fleet/worker_runtime.rs",
"crates/tui/src/hooks/config.rs",
"crates/tui/src/hooks/executor.rs",
"crates/tui/src/llm_client/mod.rs",
"crates/tui/src/localization.rs",
"crates/tui/src/main.rs",
"crates/tui/src/mcp.rs",
"crates/tui/src/mcp/oauth.rs",
"crates/tui/src/model_routing.rs",
"crates/tui/src/models.rs",
"crates/tui/src/palette/themes.rs",
"crates/tui/src/plugins/manifest.rs",
"crates/tui/src/plugins/registry.rs",
"crates/tui/src/pricing.rs",
"crates/tui/src/project_context.rs",
"crates/tui/src/prompts.rs",
"crates/tui/src/provider_lake.rs",
"crates/tui/src/provider_readiness.rs",
"crates/tui/src/remote_control.rs",
"crates/tui/src/repl/runtime.rs",
"crates/tui/src/request_manifest.rs",
"crates/tui/src/route_billing.rs",
"crates/tui/src/route_runtime.rs",
"crates/tui/src/runtime_api.rs",
"crates/tui/src/runtime_api/sessions.rs",
"crates/tui/src/runtime_threads.rs",
"crates/tui/src/sandbox/mod.rs",
"crates/tui/src/scorecard.rs",
"crates/tui/src/session_control_acceptance.rs",
"crates/tui/src/session_manager.rs",
"crates/tui/src/settings.rs",
"crates/tui/src/skills/audit.rs",
"crates/tui/src/skills/install.rs",
"crates/tui/src/skills/mod.rs",
"crates/tui/src/skills/mutation.rs",
"crates/tui/src/slop_ledger.rs",
"crates/tui/src/snapshot/repo.rs",
"crates/tui/src/task_manager.rs",
"crates/tui/src/tool_inspection.rs",
"crates/tui/src/tools/apply_patch.rs",
"crates/tui/src/tools/file.rs",
"crates/tui/src/tools/goal.rs",
"crates/tui/src/tools/registry.rs",
"crates/tui/src/tools/review.rs",
"crates/tui/src/tools/rlm.rs",
"crates/tui/src/tools/schema_sanitize.rs",
"crates/tui/src/tools/search.rs",
"crates/tui/src/tools/shell.rs",
"crates/tui/src/tools/spec.rs",
"crates/tui/src/tools/subagent/coord.rs",
"crates/tui/src/tools/subagent/mod.rs",
"crates/tui/src/tools/tasks.rs",
"crates/tui/src/tools/terminal_session.rs",
"crates/tui/src/tools/todo.rs",
"crates/tui/src/tools/tool_result_retrieval.rs",
"crates/tui/src/tools/truncate.rs",
"crates/tui/src/tools/verifier.rs",
"crates/tui/src/tools/verify.rs",
"crates/tui/src/tools/web/extract.rs",
"crates/tui/src/tools/web_run.rs",
"crates/tui/src/tools/web_search.rs",
"crates/tui/src/tools/workflow.rs",
"crates/tui/src/tools/workflow_plan_approval.rs",
"crates/tui/src/tui/ambient_life.rs",
"crates/tui/src/tui/app.rs",
"crates/tui/src/tui/app/composer.rs",
"crates/tui/src/tui/app/types.rs",
"crates/tui/src/tui/approval.rs",
"crates/tui/src/tui/auto_review.rs",
"crates/tui/src/tui/clipboard.rs",
"crates/tui/src/tui/color_compat.rs",
"crates/tui/src/tui/command_palette.rs",
"crates/tui/src/tui/context_inspector.rs",
"crates/tui/src/tui/file_mention.rs",
"crates/tui/src/tui/file_picker.rs",
"crates/tui/src/tui/footer_ui.rs",
"crates/tui/src/tui/history.rs",
"crates/tui/src/tui/hotbar/actions.rs",
"crates/tui/src/tui/hotbar/setup.rs",
"crates/tui/src/tui/live_transcript.rs",
"crates/tui/src/tui/markdown_render.rs",
"crates/tui/src/tui/model_picker.rs",
"crates/tui/src/tui/mouse_ui.rs",
"crates/tui/src/tui/notifications.rs",
"crates/tui/src/tui/pager.rs",
"crates/tui/src/tui/prompt_suggestion.rs",
"crates/tui/src/tui/provider_picker.rs",
"crates/tui/src/tui/session_picker.rs",
"crates/tui/src/tui/setup/mod.rs",
"crates/tui/src/tui/setup/tools_mcp.rs",
"crates/tui/src/tui/sidebar.rs",
"crates/tui/src/tui/startup_defaults.rs",
"crates/tui/src/tui/subagent_routing.rs",
"crates/tui/src/tui/tool_routing.rs",
"crates/tui/src/tui/transcript.rs",
"crates/tui/src/tui/ui.rs",
"crates/tui/src/tui/ui/activity_detail.rs",
"crates/tui/src/tui/underwater.rs",
"crates/tui/src/tui/views/fleet_setup.rs",
"crates/tui/src/tui/views/help.rs",
"crates/tui/src/tui/views/mod.rs",
"crates/tui/src/tui/views/skills_manager.rs",
"crates/tui/src/tui/widgets/agent_card.rs",
"crates/tui/src/tui/widgets/footer.rs",
"crates/tui/src/tui/widgets/header.rs",
"crates/tui/src/tui/widgets/mod.rs",
"crates/tui/src/tui/widgets/workflow_panel.rs",
"crates/tui/src/tui/work_surface/mod.rs",
"crates/tui/src/tui/work_surface/model.rs",
"crates/tui/src/utils.rs",
"crates/tui/src/work_graph/runtime.rs",
"crates/tui/src/working_set.rs",
"crates/tui/src/xai_oauth.rs",
"crates/workflow-js/src/vm.rs",
"crates/workflow/src/fleet_exact.rs",
"crates/workflow/src/fleet_reasoning.rs",
"crates/workflow/src/lib.rs"
],
"binary_targets": [
"codewhale-cli:codew",
"codewhale-cli:codewhale",
"codewhale-tui:codewhale-tui"
],
"document_kind": "codewhale.source_structure_budget",
"large_module_threshold_lines": 1000,
"max_large_module_count": 172,
"max_module_lines": 19072,
"max_total_owned_rust_lines": 646345,
"schema_version": 1,
"workspace_packages": [
"codewhale-agent",
"codewhale-app-server",
"codewhale-build-support",
"codewhale-cli",
"codewhale-config",
"codewhale-core",
"codewhale-execpolicy",
"codewhale-hooks",
"codewhale-lane",
"codewhale-mcp",
"codewhale-paths",
"codewhale-protocol",
"codewhale-release",
"codewhale-secrets",
"codewhale-state",
"codewhale-tools",
"codewhale-tui",
"codewhale-workflow",
"codewhale-workflow-js"
]
}
scripts/test_check_source_structure_budget.py
+163
View File
@@ -0,0 +1,163 @@
#!/usr/bin/env python3
"""Hermetic tests for check-source-structure-budget.py."""
from __future__ import annotations
import importlib.util
import json
import os
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "check-source-structure-budget.py"
SPEC = importlib.util.spec_from_file_location("check_source_structure_budget", SCRIPT)
assert SPEC and SPEC.loader
mod = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = mod
SPEC.loader.exec_module(mod)
def snapshot() -> mod.StructureSnapshot:
return mod.StructureSnapshot(
("codewhale-a", "codewhale-b"),
("codewhale-a:codew", "codewhale-a:codewhale"),
{
"crates/a/src/lib.rs": 1_200,
"crates/b/src/main.rs": 1_050,
"crates/b/src/small.rs": 200,
},
)
class StructureBudgetTests(unittest.TestCase):
def test_equal_snapshot_passes(self) -> None:
current = snapshot()
budget = mod.validate_budget(mod.budget_document(current))
self.assertEqual(mod.compare(current, budget), ([], []))
def test_deletion_and_decomposition_pass_freely(self) -> None:
budget = mod.validate_budget(mod.budget_document(snapshot()))
current = mod.StructureSnapshot(
("codewhale-a",),
("codewhale-a:codew",),
{
"crates/a/src/lib.rs": 900,
"crates/b/src/small.rs": 150,
},
)
failures, improvements = mod.compare(current, budget)
self.assertEqual(failures, [])
self.assertGreaterEqual(len(improvements), 4)
def test_added_package_and_binary_fail(self) -> None:
baseline = snapshot()
budget = mod.validate_budget(mod.budget_document(baseline))
current = mod.StructureSnapshot(
(*baseline.workspace_packages, "codewhale-c"),
(*baseline.binary_targets, "codewhale-c:whale-helper"),
dict(baseline.module_lines),
)
failures, _ = mod.compare(current, budget)
self.assertTrue(any("packages added" in failure for failure in failures))
self.assertTrue(any("binary targets added" in failure for failure in failures))
def test_new_large_file_and_aggregate_growth_fail(self) -> None:
baseline = snapshot()
budget = mod.validate_budget(mod.budget_document(baseline))
modules = dict(baseline.module_lines)
modules["crates/c/src/lib.rs"] = mod.LARGE_MODULE_THRESHOLD
failures, _ = mod.compare(
mod.StructureSnapshot(
baseline.workspace_packages,
baseline.binary_targets,
modules,
),
budget,
)
self.assertTrue(any("new thousand-line" in failure for failure in failures))
self.assertTrue(any("aggregate owned" in failure for failure in failures))
def test_line_neutral_ownership_move_passes_but_larger_maximum_fails(self) -> None:
baseline = snapshot()
budget = mod.validate_budget(mod.budget_document(baseline))
moved = dict(baseline.module_lines)
moved["crates/a/src/lib.rs"] -= 50
moved["crates/b/src/main.rs"] += 50
self.assertEqual(
mod.compare(
mod.StructureSnapshot(
baseline.workspace_packages, baseline.binary_targets, moved
),
budget,
)[0],
[],
)
moved["crates/a/src/lib.rs"] += 51
failures, _ = mod.compare(
mod.StructureSnapshot(
baseline.workspace_packages, baseline.binary_targets, moved
),
budget,
)
self.assertTrue(any("largest module grew" in failure for failure in failures))
def test_budget_document_round_trips_with_exact_types(self) -> None:
document = mod.budget_document(snapshot())
budget = mod.validate_budget(document)
self.assertEqual(budget.max_module_lines, 1_200)
self.assertEqual(budget.max_total_owned_rust_lines, 2_450)
document["max_module_lines"] = True
with self.assertRaisesRegex(mod.StructureBudgetError, "non-negative integer"):
mod.validate_budget(document)
def test_source_discovery_excludes_owned_test_modules(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "crates" / "a" / "src"
(source / "tests").mkdir(parents=True)
(source / "lib.rs").write_text("one\ntwo\n", encoding="utf-8")
(source / "tests.rs").write_text("test\n", encoding="utf-8")
(source / "tests" / "fixture.rs").write_text("fixture\n", encoding="utf-8")
discovered = [path.relative_to(root).as_posix() for path in mod.production_rust_files(root)]
self.assertEqual(discovered, ["crates/a/src/lib.rs"])
def test_metadata_command_is_locked_no_deps_and_offline(self) -> None:
completed = subprocess.CompletedProcess(
args=[], returncode=0, stdout=json.dumps({"packages": []}), stderr=""
)
with mock.patch.object(mod.subprocess, "run", return_value=completed) as run:
self.assertEqual(mod.cargo_metadata(ROOT), {"packages": []})
command = run.call_args.args[0]
environment = run.call_args.kwargs["env"]
self.assertEqual(command[0:2], ["cargo", "metadata"])
self.assertIn("--offline", command)
self.assertIn("--locked", command)
self.assertIn("--no-deps", command)
self.assertEqual(environment["CARGO_NET_OFFLINE"], "true")
def test_atomic_update_preserves_permissions_and_cleans_failure(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "budget.json"
path.write_text("{}", encoding="utf-8")
os.chmod(path, 0o640)
mod.write_json_atomic(path, mod.budget_document(snapshot()))
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o640)
original = path.read_text(encoding="utf-8")
with (
mock.patch.object(mod.os, "replace", side_effect=OSError("stop")),
self.assertRaisesRegex(OSError, "stop"),
):
mod.write_json_atomic(path, {"replacement": True})
self.assertEqual(path.read_text(encoding="utf-8"), original)
self.assertEqual(list(Path(tmp).glob(".budget.json.*.tmp")), [])
if __name__ == "__main__":
raise SystemExit(unittest.main())