Spaces:
Running
Running
| """ | |
| BeamBridge v2 Comprehensive Verification & Security Audit Suite. | |
| Verifies: | |
| 1. Official Beam-1.0-501B-A23B BPE Tokenizer (vocab=151,681) | |
| 2. Zero forbidden legacy imports (`moneyed`, `price_parser`, `schwifty`, `stdnum`) in synthesized MCP modules | |
| 3. 160/160 (100.0%) live differential parity against untouched cloned GitHub libraries | |
| 4. Real OpenRouter `gen-...` execution telemetry & 160-row `legacy2mcp_bench.jsonl` dataset | |
| 5. Live `/mcp` JSON-RPC 2.0, `/api/mcp/invoke`, `/api/bootstrap`, and `/api/github/ingest` | |
| 6. Strict Zero-Secret Leak Audit across the entire workspace | |
| """ | |
| from __future__ import annotations | |
| import ast | |
| import json | |
| import sys | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parent.parent | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| from fastapi.testclient import TestClient | |
| from backend.ast_analyzer import analyze_repository | |
| from backend.main import app | |
| from backend.parity_verifier import get_empirical_ledger, run_all_verifications | |
| from backend.real_oracles import get_beam_tokenizer | |
| client = TestClient(app) | |
| def test_official_beam_bpe_tokenizer(): | |
| tok = get_beam_tokenizer() | |
| assert tok.get_vocab_size() == 151681 | |
| ids = tok.encode("def modernize_legacy_to_mcp(x: int) -> dict:").ids | |
| assert len(ids) > 5 | |
| def test_synthesized_mcp_modules_have_zero_legacy_imports(): | |
| forbidden = {"moneyed", "price_parser", "schwifty", "stdnum"} | |
| for rk in ("py_moneyed", "price_parser", "schwifty", "python_stdnum"): | |
| p = ROOT / "modernized_mcp" / f"{rk}_mcp.py" | |
| assert p.exists(), f"Missing {p}" | |
| tree = ast.parse(p.read_text(encoding="utf-8")) | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Import): | |
| for al in node.names: | |
| assert al.name.split(".")[0] not in forbidden, f"{rk} illegally imported {al.name}" | |
| elif isinstance(node, ast.ImportFrom) and node.module: | |
| assert node.module.split(".")[0] not in forbidden, f"{rk} illegally imported {node.module}" | |
| def test_all_160_real_github_differential_assertions_pass(): | |
| summary = run_all_verifications() | |
| assert summary["total_repositories"] == 4 | |
| assert summary["total_cases"] == 160 | |
| assert summary["beam_1m_passed"] == 160 | |
| assert summary["beam_1m_parity_pct"] == 100.0 | |
| assert summary["chunked_rag_passed"] == 114 | |
| assert summary["chunked_rag_parity_pct"] == 71.25 | |
| def test_empirical_ledger_and_dataset_have_real_openrouter_request_ids(): | |
| ledger = get_empirical_ledger() | |
| assert len(ledger["head_to_head_benchmark"]) >= 4 | |
| for rk in ("py_moneyed", "price_parser", "schwifty", "python_stdnum"): | |
| wr = ledger["repositories"][rk]["whole_repo_1m"] | |
| assert wr["final_passed"] == 40 | |
| for t in wr["turns"]: | |
| assert str(t["request_id"]).startswith("gen-") | |
| ds_lines = (ROOT / "dataset" / "legacy2mcp_bench.jsonl").read_text(encoding="utf-8").strip().splitlines() | |
| assert len(ds_lines) == 160 | |
| sample = json.loads(ds_lines[0]) | |
| assert sample["whole_repo_openrouter_request_ids"][0].startswith("gen-") | |
| def test_fastapi_bootstrap_and_mcp_jsonrpc_endpoints(): | |
| r_boot = client.get("/api/bootstrap") | |
| assert r_boot.status_code == 200 | |
| boot = r_boot.json() | |
| assert boot["portfolio_totals"]["total_verification_assertions"] == 160 | |
| assert boot["portfolio_totals"]["mcp_tools_exposed"] == 12 | |
| assert boot["portfolio_totals"]["beam_1m_parity_pct"] == 100.0 | |
| r_list = client.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"}) | |
| assert r_list.status_code == 200 | |
| assert len(r_list.json()["result"]["tools"]) == 12 | |
| r_call = client.post( | |
| "/mcp", | |
| json={ | |
| "jsonrpc": "2.0", | |
| "id": 2, | |
| "method": "tools/call", | |
| "params": {"name": "validate_eu_vat_number", "arguments": {"raw_vat": "DE 136 695 976"}}, | |
| }, | |
| ) | |
| assert r_call.status_code == 200 | |
| sc = r_call.json()["result"]["structuredContent"] | |
| assert sc["live_oracle_verified"] is True | |
| assert sc["output"]["is_valid"] is True | |
| assert sc["output"]["compact"] == "DE136695976" | |
| r_inv = client.post( | |
| "/api/mcp/invoke", | |
| json={"repo_id": "price_parser", "tool_name": "parse_procurement_price", "arguments": {"raw_text": "1.299,95 €"}}, | |
| ) | |
| assert r_inv.status_code == 200 | |
| assert r_inv.json()["parity_verified"] is True | |
| assert r_inv.json()["modern_output"]["amount_str"] == "1299.95" | |
| assert r_inv.json()["modern_output"]["currency"] == "€" | |
| def test_live_github_ingest_endpoint(): | |
| r_gh = client.post("/api/github/ingest", json={"github_url": "https://github.com/limist/py-moneyed", "max_files": 20}) | |
| assert r_gh.status_code == 200 | |
| body = r_gh.json() | |
| assert body["repo"] == "limist/py-moneyed" | |
| assert body["exact_beam_bpe_tokens"] > 5000 | |
| assert body["fits_single_1m_context"] is True | |
| def test_zero_api_keys_persisted_anywhere_on_disk(): | |
| import re | |
| secret_re = re.compile(r"sk-or-v1-[0-9a-fA-F]{16,}") | |
| skip_dirs = {".git", "__pycache__", ".pytest_cache", "node_modules"} | |
| for p in ROOT.rglob("*"): | |
| if any(part in skip_dirs for part in p.parts) or not p.is_file(): | |
| continue | |
| if p.suffix in {".png", ".jpg", ".zip", ".pyc"} or p.name == "beam_tokenizer.json": | |
| continue | |
| txt = p.read_text(encoding="utf-8", errors="ignore") | |
| assert not secret_re.search(txt), f"Live OpenRouter key pattern found in {p}" | |