Spaces:
Running
Running
| import unittest | |
| from fastapi import FastAPI | |
| from fastapi.testclient import TestClient | |
| import a11oy_ayllu as ayllu | |
| import a11oy_nemo_core as nemo | |
| class CouncilContractTests(unittest.TestCase): | |
| def _result(self): | |
| return { | |
| "participants": ["Amaru", "Yupaq"], | |
| "mode": "single-round", | |
| "rounds": [ | |
| {"persona": "Amaru", "round": 1, "model": "test-model", | |
| "stub": False, "answer": "Architecture proposal."}, | |
| {"persona": "Yupaq", "round": 1, "model": "test-model", | |
| "stub": False, "answer": "Evidence gap."}, | |
| ], | |
| } | |
| def test_generic_nemo_route_is_governance_first(self): | |
| route = nemo.govern_route("Review a high-consequence agent action", top_k=2) | |
| self.assertEqual(route["experts_selected"], ["governance", "code"]) | |
| self.assertEqual(route["routing_evidence_state"], "HEURISTIC") | |
| self.assertEqual(route["experts"][0]["selection_basis"], | |
| "explicit-governance-first-fallback") | |
| def test_contract_is_proposal_only_and_replay_is_deterministic(self): | |
| routing = {"state": "HEURISTIC", "experts_selected": ["governance"]} | |
| one = ayllu._build_council_contract("review this", self._result(), routing) | |
| two = ayllu._build_council_contract("review this", self._result(), routing) | |
| self.assertEqual(one["schema"], ayllu.COUNCIL_SCHEMA) | |
| self.assertEqual(one["decision_state"], "PROPOSAL_ONLY") | |
| self.assertEqual(one["semantic_consensus"]["state"], "NOT_MEASURED") | |
| self.assertTrue(one["human_checkpoint"]["required"]) | |
| self.assertEqual(one["evidence_state"], "LIVE") | |
| self.assertEqual(one["replay"]["key"], two["replay"]["key"]) | |
| def test_runtime_signer_is_preferred(self): | |
| def signer(payload): | |
| return {"signed": True, "payload": "encoded", "signatures": [{"sig": "x"}]} | |
| receipt = ayllu._make_receipt({"decision": "proposal"}, sign_fn=signer) | |
| self.assertTrue(receipt["signed"]) | |
| self.assertEqual(receipt["payload"], "encoded") | |
| def test_manifest_discloses_no_trained_weights(self): | |
| manifest = ayllu.council_manifest() | |
| self.assertFalse(manifest["nemo"]["weights_present"]) | |
| self.assertEqual(manifest["nemo"]["training_state"], "NOT_PERFORMED") | |
| self.assertEqual(manifest["limits"]["decision_state"], "PROPOSAL_ONLY") | |
| def test_http_council_emits_signed_contract(self): | |
| app = FastAPI() | |
| app.state.szl_sign_receipt = lambda payload: { | |
| "signed": True, "payload": "encoded", "signatures": [{"sig": "test"}]} | |
| ayllu.register(app) | |
| async def fake_complete(system, prompt, **kwargs): | |
| return {"text": "bounded answer", "model": "test-model", "stub": False} | |
| original = ayllu._backend.model_complete | |
| ayllu._backend.model_complete = fake_complete | |
| try: | |
| response = TestClient(app).post( | |
| "/api/a11oy/v1/ayllu/council", | |
| json={"prompt": "Review this proposal", "personas": ["Amaru"]}) | |
| finally: | |
| ayllu._backend.model_complete = original | |
| self.assertEqual(response.status_code, 200) | |
| body = response.json() | |
| self.assertEqual(body["contract"]["decision_state"], "PROPOSAL_ONLY") | |
| self.assertTrue(body["receipt"]["signed"]) | |
| self.assertEqual(body["contract"]["routing"]["state"], "HEURISTIC") | |
| if __name__ == "__main__": | |
| unittest.main() | |