| 1 | """
|
|---|
| 2 | Persistence and integrity verification tests for Flex-O entities.
|
|---|
| 3 | Ensures fingerprints survive JSON export/import and detect tampering.
|
|---|
| 4 | """
|
|---|
| 5 | import json
|
|---|
| 6 | import pytest
|
|---|
| 7 |
|
|---|
| 8 | from flexoentity import EntityState
|
|---|
| 9 | from builder.questions import RadioQuestion, AnswerOption
|
|---|
| 10 |
|
|---|
| 11 |
|
|---|
| 12 | # ──────────────────────────────────────────────────────────────────────────────
|
|---|
| 13 | @pytest.fixture
|
|---|
| 14 | def approved_question(domain):
|
|---|
| 15 | """Provide a fully approved and published RadioQuestion for persistence tests."""
|
|---|
| 16 | q = RadioQuestion(
|
|---|
| 17 | domain=domain,
|
|---|
| 18 | etype=None, # RadioQuestion sets this internally to EntityType.QUESTION
|
|---|
| 19 | state=EntityState.DRAFT,
|
|---|
| 20 | text="What is Ohm’s law?",
|
|---|
| 21 | options=[
|
|---|
| 22 | AnswerOption(text="U = R × I", points=1),
|
|---|
| 23 | AnswerOption(text="U = I / R", points=0),
|
|---|
| 24 | AnswerOption(text="R = U × I", points=0),
|
|---|
| 25 | ],
|
|---|
| 26 | )
|
|---|
| 27 | q.approve()
|
|---|
| 28 | q.sign()
|
|---|
| 29 | q.publish()
|
|---|
| 30 | return q
|
|---|
| 31 |
|
|---|
| 32 |
|
|---|
| 33 | @pytest.mark.skip(reason="FlexOIDs regenerated on import; enable once JSON format is stable")
|
|---|
| 34 | def test_json_roundtrip_preserves_integrity(approved_question):
|
|---|
| 35 | """
|
|---|
| 36 | Export to JSON and reload — ensure fingerprints and signatures remain valid.
|
|---|
| 37 | """
|
|---|
| 38 | json_str = approved_question.to_json()
|
|---|
| 39 | loaded = RadioQuestion.from_json(json_str)
|
|---|
| 40 |
|
|---|
| 41 | # Fingerprint and state should match — integrity must pass
|
|---|
| 42 | assert RadioQuestion.verify_integrity(loaded)
|
|---|
| 43 |
|
|---|
| 44 | # Metadata should be preserved exactly
|
|---|
| 45 | assert approved_question.signature == loaded.signature
|
|---|
| 46 | assert approved_question.flexo_id == loaded.flexo_id
|
|---|
| 47 | assert loaded.state == approved_question.state
|
|---|
| 48 |
|
|---|
| 49 |
|
|---|
| 50 | # ──────────────────────────────────────────────────────────────────────────────
|
|---|
| 51 |
|
|---|
| 52 | @pytest.mark.skip(reason="FlexOIDs regenerated on import; tampering detection not yet implemented")
|
|---|
| 53 | def test_json_tampering_detection(approved_question):
|
|---|
| 54 | """Tampering with content should invalidate fingerprint verification."""
|
|---|
| 55 | json_str = approved_question.to_json()
|
|---|
| 56 | tampered = json.loads(json_str)
|
|---|
| 57 | tampered["text"] = "Tampered content injection"
|
|---|
| 58 | tampered_json = json.dumps(tampered)
|
|---|
| 59 |
|
|---|
| 60 | loaded = RadioQuestion.from_json(tampered_json)
|
|---|
| 61 | assert not RadioQuestion.verify_integrity(loaded)
|
|---|
| 62 |
|
|---|
| 63 |
|
|---|
| 64 | # ──────────────────────────────────────────────────────────────────────────────
|
|---|
| 65 |
|
|---|
| 66 | @pytest.mark.skip(reason="FlexOIDs regenerated on import; corruption detection not yet applicable")
|
|---|
| 67 | def test_json_file_corruption(approved_question, tmp_path):
|
|---|
| 68 | """Simulate file corruption — integrity check must fail."""
|
|---|
| 69 | file = tmp_path / "question.json"
|
|---|
| 70 | json_str = approved_question.to_json()
|
|---|
| 71 | file.write_text(json_str)
|
|---|
| 72 |
|
|---|
| 73 | # Corrupt the file (simulate accidental byte modification)
|
|---|
| 74 | corrupted = json_str.replace("Ohm’s", "Omm’s")
|
|---|
| 75 | file.write_text(corrupted)
|
|---|
| 76 |
|
|---|
| 77 | loaded = RadioQuestion.from_json(file.read_text())
|
|---|
| 78 | assert not RadioQuestion.verify_integrity(loaded)
|
|---|