Files
fengqun/tests/test_tpm_budget_acceptance.py
T
gongzhiyongandOmX a4d771ede5 Define numeric swarm acceptance gates
Add a concrete 0-100 swarmness/compliance score, local large-scale stress, and 3000 TPM budget acceptance so the repo can say when it is a swarm by measured criteria instead of prose alone.

Constraint: user required Chinese docs, explicit scenarios, parameters, formulas, pass/fail lines, and git upload.

Rejected: prose-only PASS reports | they did not answer whether the system is a swarm with a concrete score.

Confidence: high

Scope-risk: moderate

Directive: keep production runtime claims separate from local minimal swarm acceptance scores.

Tested: py_compile swarm_minimal examples tests; unittest discover -s tests 45 tests; run_swarm_compliance_score.py; run_tpm_budget_acceptance.py; run_academic_standard_evaluation.py; git diff --check; docs/script secret-pattern scan.

Not-tested: live S07 and production Kubernetes/NewAPI provider-rate-limit stress were not rerun in this upload step.

Co-authored-by: OmX <omx@oh-my-codex.dev>
2026-05-17 18:19:24 +08:00

47 lines
1.8 KiB
Python

import unittest
from examples.run_tpm_budget_acceptance import TpmConfig, run_acceptance
class TpmBudgetAcceptanceTest(unittest.TestCase):
def test_exact_3000_tpm_budget_passes_without_overrun(self) -> None:
report = run_acceptance(
TpmConfig(
target_tpm=3000,
agent_count=6,
task_count=60,
tokens_per_task=50,
)
)
self.assertEqual(report["status"], "PASS")
self.assertEqual(report["summary"]["total_reserved_tokens"], 3000)
self.assertEqual(report["summary"]["simulated_minutes"], 1)
self.assertEqual(report["summary"]["max_tokens_in_any_minute"], 3000)
self.assertEqual(report["summary"]["over_budget_windows"], {})
self.assertEqual(report["summary"]["completed_tasks"], 60)
self.assertEqual(report["summary"]["failed_tasks"], 0)
self.assertEqual(report["summary"]["duplicate_claims"], 0)
self.assertEqual(report["summary"]["participating_agents"], 6)
def test_budget_rolls_forward_without_exceeding_window(self) -> None:
report = run_acceptance(
TpmConfig(
target_tpm=3000,
agent_count=4,
task_count=61,
tokens_per_task=50,
)
)
self.assertEqual(report["status"], "PASS")
self.assertEqual(report["summary"]["total_reserved_tokens"], 3050)
self.assertEqual(report["summary"]["simulated_minutes"], 2)
self.assertEqual(report["summary"]["max_tokens_in_any_minute"], 3000)
self.assertEqual(report["summary"]["over_budget_windows"], {})
self.assertEqual(report["tokens_by_minute"], {"0": 3000, "1": 50})
if __name__ == "__main__":
unittest.main()