tests/test_integration.py
| 1 | """Integration tests - end-to-end build/sign/prove/verify.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from quantumshield.identity.agent import AgentIdentity |
| 6 | |
| 7 | from pqc_training_data import ( |
| 8 | CommitmentBuilder, |
| 9 | CommitmentSigner, |
| 10 | CommitmentVerifier, |
| 11 | DataRecord, |
| 12 | TrainingCommitment, |
| 13 | ) |
| 14 | |
| 15 | |
| 16 | def test_full_pipeline_build_sign_prove_verify() -> None: |
| 17 | identity = AgentIdentity.create("big-creator") |
| 18 | signer = CommitmentSigner(identity) |
| 19 | |
| 20 | records = [ |
| 21 | DataRecord(content=f"record-{i}".encode(), metadata={"id": i, "kind": "doc"}) |
| 22 | for i in range(100) |
| 23 | ] |
| 24 | builder = CommitmentBuilder("ds-100", "1.2.3") |
| 25 | builder.add_records(records) |
| 26 | commitment = signer.sign(builder.build(description="100 records")) |
| 27 | |
| 28 | # Pick a few arbitrary indices and verify each |
| 29 | for idx in (0, 17, 42, 63, 99): |
| 30 | proof = builder.tree.inclusion_proof(idx) |
| 31 | result = CommitmentVerifier.verify(records[idx], proof, commitment) |
| 32 | assert result.fully_verified, ( |
| 33 | f"verification failed for index {idx}: {result.error}" |
| 34 | ) |
| 35 | |
| 36 | |
| 37 | def test_missing_record_rejected() -> None: |
| 38 | identity = AgentIdentity.create("honest") |
| 39 | signer = CommitmentSigner(identity) |
| 40 | |
| 41 | records = [ |
| 42 | DataRecord(content=f"legit-{i}".encode(), metadata={"id": i}) |
| 43 | for i in range(20) |
| 44 | ] |
| 45 | builder = CommitmentBuilder("legit-ds", "1.0.0") |
| 46 | builder.add_records(records) |
| 47 | commitment = signer.sign(builder.build()) |
| 48 | |
| 49 | # Attacker fabricates: tries to claim doc-999 is in the tree. |
| 50 | # Their best attack: reuse a real slot's proof with a forged record. |
| 51 | forged = DataRecord(content=b"doc-999-never-added", metadata={"id": 999}) |
| 52 | for attack_idx in (0, 5, 10, 19): |
| 53 | proof = builder.tree.inclusion_proof(attack_idx) |
| 54 | result = CommitmentVerifier.verify(forged, proof, commitment) |
| 55 | assert result.fully_verified is False |
| 56 | assert result.leaf_matches_record is False |
| 57 | |
| 58 | |
| 59 | def test_commitment_serialization_survives_network() -> None: |
| 60 | identity = AgentIdentity.create("serializer") |
| 61 | signer = CommitmentSigner(identity) |
| 62 | records = [ |
| 63 | DataRecord(content=f"s-{i}".encode(), metadata={"id": i}) for i in range(6) |
| 64 | ] |
| 65 | builder = CommitmentBuilder("serial-ds", "1.0.0") |
| 66 | builder.add_records(records) |
| 67 | commitment = signer.sign(builder.build()) |
| 68 | |
| 69 | # Round-trip commitment through JSON (simulating network / disk) |
| 70 | blob = commitment.to_json() |
| 71 | restored = TrainingCommitment.from_json(blob) |
| 72 | |
| 73 | # Signature still verifies |
| 74 | assert CommitmentSigner.verify(restored) is True |
| 75 | |
| 76 | # Still accepts valid inclusion proofs for the same tree |
| 77 | proof = builder.tree.inclusion_proof(3) |
| 78 | result = CommitmentVerifier.verify(records[3], proof, restored) |
| 79 | assert result.fully_verified is True |
| 80 | |