tests/test_integration.py
| 1 | """End-to-end integration tests for pqc-ai-governance.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from pqc_ai_governance import ( |
| 8 | AuthorizationChain, |
| 9 | AuthorizationGrant, |
| 10 | ByzantineDetectedError, |
| 11 | ConsensusRound, |
| 12 | GovernanceAuditLog, |
| 13 | GovernanceNode, |
| 14 | GovernanceProposal, |
| 15 | NodeRegistry, |
| 16 | ProposalKind, |
| 17 | ProposalStatus, |
| 18 | QuorumPolicy, |
| 19 | VoteDecision, |
| 20 | VoteTally, |
| 21 | ) |
| 22 | |
| 23 | |
| 24 | def test_happy_path_authorize_model( |
| 25 | nodes: list[GovernanceNode], |
| 26 | registry: NodeRegistry, |
| 27 | sample_proposal: GovernanceProposal, |
| 28 | ) -> None: |
| 29 | alice, bob, carol, dave, eve = nodes |
| 30 | |
| 31 | rnd = ConsensusRound(proposal=sample_proposal, registry=registry) |
| 32 | # 4 approvals, 1 reject => 5 approve-weight (alice+bob+carol+dave=1+1+1+2), 1 reject |
| 33 | for voter in (alice, bob, carol, dave): |
| 34 | rnd.cast(voter.cast_vote(sample_proposal, VoteDecision.APPROVE)) |
| 35 | rnd.cast(eve.cast_vote(sample_proposal, VoteDecision.REJECT)) |
| 36 | |
| 37 | result = rnd.finalize(alice) |
| 38 | assert result.decision == "passed" |
| 39 | assert sample_proposal.status == ProposalStatus.PASSED |
| 40 | assert ConsensusRound.verify_result(result) is True |
| 41 | |
| 42 | # Now bind the passing proposal into an authorization chain. |
| 43 | chain = AuthorizationChain(subject_id=sample_proposal.subject_id) |
| 44 | chain.add( |
| 45 | AuthorizationGrant( |
| 46 | subject_id=sample_proposal.subject_id, |
| 47 | kind=sample_proposal.kind, |
| 48 | result=result, |
| 49 | ) |
| 50 | ) |
| 51 | assert chain.is_authorized(ProposalKind.AUTHORIZE_MODEL) is True |
| 52 | |
| 53 | |
| 54 | def test_byzantine_detection( |
| 55 | nodes: list[GovernanceNode], |
| 56 | registry: NodeRegistry, |
| 57 | sample_proposal: GovernanceProposal, |
| 58 | ) -> None: |
| 59 | alice, _, _, _, eve = nodes |
| 60 | tally = VoteTally(proposal=sample_proposal, registry=registry) |
| 61 | tally.add(eve.cast_vote(sample_proposal, VoteDecision.APPROVE)) |
| 62 | with pytest.raises(ByzantineDetectedError): |
| 63 | tally.add(eve.cast_vote(sample_proposal, VoteDecision.REJECT)) |
| 64 | |
| 65 | |
| 66 | def test_no_quorum_path( |
| 67 | nodes: list[GovernanceNode], |
| 68 | registry: NodeRegistry, |
| 69 | sample_proposal: GovernanceProposal, |
| 70 | ) -> None: |
| 71 | alice, bob, *_ = nodes |
| 72 | rnd = ConsensusRound( |
| 73 | proposal=sample_proposal, |
| 74 | registry=registry, |
| 75 | policy=QuorumPolicy(), |
| 76 | ) |
| 77 | rnd.cast(alice.cast_vote(sample_proposal, VoteDecision.APPROVE)) |
| 78 | rnd.cast(bob.cast_vote(sample_proposal, VoteDecision.APPROVE)) |
| 79 | |
| 80 | # Audit: log every vote cast, plus the eventual outcome |
| 81 | audit = GovernanceAuditLog() |
| 82 | audit.log_proposal_created(sample_proposal) |
| 83 | for sv in rnd.tally.valid_votes: |
| 84 | audit.log_vote_cast(sv) |
| 85 | |
| 86 | result = rnd.finalize(alice) |
| 87 | audit.log_consensus_reached(result) |
| 88 | |
| 89 | assert result.decision == "rejected" |
| 90 | assert "participation" in result.reason |
| 91 | assert sample_proposal.status == ProposalStatus.REJECTED |
| 92 | assert any(e.operation == "consensus_reached" for e in audit.entries()) |
| 93 | |