Changeset 02d288d in flexoentity for tests


Ignore:
Timestamp:
10/23/25 13:27:08 (3 months ago)
Author:
Enrico Schwass <ennoausberlin@…>
Branches:
master
Children:
4ceca57
Parents:
6a7dec1
Message:

improve hash generation and collision handler - move signature from FlexOID to FlexoEntity

Location:
tests
Files:
4 edited

Legend:

Unmodified
Added
Removed

  • tests/conftest.py

    r6a7dec1 r02d288d  
    55from flexoentity import FlexoEntity, EntityType, EntityState, Domain
    66
     7import pytest
     8import json
     9from flexoentity import EntityType, EntityState, Domain
     10from builder.questions import RadioQuestion, AnswerOption  # adjust path if different
     11from builder.media_items import NullMediaItem  # adjust import if needed
    712
    8 class DummyEntity(FlexoEntity):
    9     """Minimal concrete subclass for testing FlexoEntity logic."""
    1013
    11     def __init__(self, domain, etype, state, seed="DUMMY"):
    12         self._seed = seed
    13         super().__init__(domain, etype, state)
     14@pytest.fixture(scope="session")
     15def domain():
     16    """Provide a reusable domain for all entity tests."""
     17    return Domain(
     18        domain="SIG",
     19        etype=EntityType.DOMAIN,
     20        state=EntityState.DRAFT,
     21        fullname="Signal Corps",
     22        description="Questions related to communications and signaling systems.",
     23        classification="RESTRICTED",
     24        owner="test-suite"
     25    )
    1426
    15     @property
    16     def text_seed(self) -> str:
    17         return self._seed
    1827
    19     @classmethod
    20     def from_dict(cls, data):
    21         """Ensure enums and seed are reconstructed correctly."""
    22         domain = data["domain"]
    23         etype = EntityType[data["etype"]] if isinstance(data["etype"], str) else data["etype"]
    24         state = EntityState[data["state"]] if isinstance(data["state"], str) else data["state"]
    25         seed = data.get("text_seed", "DUMMY-CONTENT")
    26         return cls(domain=domain, etype=etype, state=state, seed=seed)
     28@pytest.fixture
     29def radio_question(domain):
     30    """Return a simple RadioQuestion entity for testing FlexoEntity logic."""
     31    q = RadioQuestion(
     32        domain=domain,
     33        etype=EntityType.QUESTION,
     34        state=EntityState.DRAFT,
     35        text="Which frequency band is used for shortwave communication?",
     36        options=[
     37            AnswerOption(id="opt1", text="HF (3–30 MHz)", points=1),
     38            AnswerOption(id="opt2", text="VHF (30–300 MHz)", points=0),
     39            AnswerOption(id="opt3", text="UHF (300–3000 MHz)", points=0),
     40        ]
     41    )
     42    return q
    2743
    28     @classmethod
    29     def from_json(cls, data_str: str):
    30         return cls.from_dict(json.loads(data_str))
    31    
     44
    3245@pytest.fixture
    33 def entity():
    34     """Generic FlexoEntity-like instance in draft state."""
    35     return DummyEntity(
    36         domain=Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT, fullname="Signal Corps", classification="RESTRICTED"),
    37         etype=EntityType.CATALOG,
    38         state=EntityState.DRAFT,
    39     )
     46def serialized_question(radio_question):
     47    """Provide the serialized JSON form for roundtrip tests."""
     48    return radio_question.to_json()
     49
     50
     51@pytest.fixture
     52def deserialized_question(serialized_question):
     53    """Recreate a question from JSON for consistency tests."""
     54    return RadioQuestion.from_json(serialized_question)
     55
    4056
    4157@pytest.fixture
    4258def null_media():
    43     """Provide a default NullMediaItem instance for tests."""
     59    """Provide a default NullMediaItem instance for media tests."""
    4460    return NullMediaItem(
    45         domain="GEN",
     61        domain=domain,
    4662        etype=EntityType.MEDIA,
    4763        state=EntityState.DRAFT

  • tests/test_id_lifecycle.py

    r6a7dec1 r02d288d  
    11import pytest
    2 
    3 from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState
    4 
    5 def test_initial_state(entity):
    6     assert entity.state == EntityState.DRAFT
    7     assert entity.flexo_id.version == 1
    8     assert len(entity.flexo_id.signature) == 16  # blake2s digest_size=8 → 16 hex
    9     assert FlexoEntity.verify_integrity(entity)
     2from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState
    103
    114
    12 def test_approval_bumps_version(entity):
    13     entity.approve()
    14     assert entity.state == EntityState.APPROVED
    15     assert entity.flexo_id.version == 2
     5# ──────────────────────────────────────────────────────────────────────────────
     6# Tests adapted to use real RadioQuestion fixture instead of DummyEntity
     7# ──────────────────────────────────────────────────────────────────────────────
     8
     9def test_initial_state(radio_question):
     10    q = radio_question
     11    assert q.state == EntityState.DRAFT
     12    assert q.flexo_id.version == 1
     13    assert FlexoEntity.verify_integrity(q)
    1614
    1715
    18 def test_signing_bumps_version(entity):
    19     entity.approve()
    20     v_before = entity.flexo_id
    21     entity.sign()
    22     assert entity.state == EntityState.APPROVED_AND_SIGNED
    23     assert entity.flexo_id != v_before
     16def test_approval_bumps_version(radio_question):
     17    q = radio_question
     18    q.approve()
     19    assert q.state == EntityState.APPROVED
     20    assert q.flexo_id.version == 2
    2421
    2522
    26 def test_publish_bumps_version(entity):
    27     entity.approve()
    28     entity.sign()
    29     v_before = entity.flexo_id.version
    30     entity.publish()
    31     assert entity.state == EntityState.PUBLISHED
    32     assert entity.flexo_id.version == v_before + 1
     23def test_signing_bumps_version(radio_question):
     24    q = radio_question
     25    q.approve()
     26    v_before = str(q.flexo_id)
     27    q.sign()
     28    assert q.state == EntityState.APPROVED_AND_SIGNED
     29    assert str(q.flexo_id) != v_before
    3330
    3431
    35 def test_modify_content_changes_fingerprint(entity):
    36     old_signature = entity.flexo_id.signature
    37     entity._seed = "Rephrased content"  # simulate text change
    38     entity._update_fingerprint()
    39     assert entity.flexo_id.signature != old_signature
     32def test_publish_bumps_version(radio_question):
     33    q = radio_question
     34    q.approve()
     35    q.sign()
     36    v_before = q.flexo_id.version
     37    q.publish()
     38    assert q.state == EntityState.PUBLISHED
     39    assert q.flexo_id.version == v_before + 1
    4040
    4141
    42 def test_no_version_bump_on_draft_edits(entity):
    43     entity._seed = "Draft edit only"
    44     entity._update_fingerprint()
    45     assert entity.flexo_id.version == 1
     42def test_modify_content_changes_fingerprint(radio_question):
     43    q = radio_question
     44    q.text = "Rephrased content"  # simulate text change
     45    changed = q._update_fingerprint()
     46    assert changed
    4647
    4748
    48 def test_version_bump_after_edit_and_sign(entity):
    49     entity.approve()
    50     v1 = entity.flexo_id
    51     entity._seed = "Changed content"
    52     entity.sign()
    53     assert entity.flexo_id != v1
     49def test_no_version_bump_on_draft_edits(radio_question):
     50    q = radio_question
     51    q.text = "Minor draft edit"
     52    q._update_fingerprint()
     53    assert q.flexo_id.version == 1
    5454
    5555
    56 def test_integrity_check_passes_and_fails(entity):
    57     entity.approve()
    58     assert FlexoEntity.verify_integrity(entity)
    59     # simulate tampering
    60     entity._seed = "Tampered text"
    61     assert not FlexoEntity.verify_integrity(entity)
     56def test_version_bump_after_edit_and_sign(radio_question):
     57    q = radio_question
     58    q.approve()
     59    v1 = str(q.flexo_id)
     60    q.text = "Changed content"
     61    q.sign()
     62    assert str(q.flexo_id) != v1
    6263
    6364
    64 def test_obsolete_state(entity):
    65     entity.approve()
    66     entity.sign()
    67     entity.publish()
    68     entity.obsolete()
    69     assert entity.state == EntityState.OBSOLETE
     65def test_integrity_check_passes_and_fails(radio_question):
     66    q = radio_question
     67    q.approve()
     68    assert FlexoEntity.verify_integrity(q)
     69
     70    # simulate tampering
     71    q.text = "Tampered text"
     72    assert not FlexoEntity.verify_integrity(q)
    7073
    7174
    72 def test_clone_new_base_resets_lineage(entity):
    73     entity.approve()
    74     entity.sign()
    75     entity.publish()
    76     entity.obsolete()
    77     old_id = entity.flexo_id
    78     entity.clone_new_base()
    79     assert entity.flexo_id != old_id
    80     assert entity.state == EntityState.DRAFT
    81     assert entity.flexo_id.version == 1
     75def test_obsolete_state(radio_question):
     76    q = radio_question
     77    q.approve()
     78    q.sign()
     79    q.publish()
     80    q.obsolete()
     81    assert q.state == EntityState.OBSOLETE
    8282
    8383
    84 def test_mass_version_increments_until_obsolete(entity):
    85     entity.approve()
     84def test_clone_new_base_resets_lineage(radio_question):
     85    q = radio_question
     86    q.approve()
     87    q.sign()
     88    q.publish()
     89    q.obsolete()
     90    old_id = str(q.flexo_id)
     91    q.clone_new_base()
     92    assert str(q.flexo_id) != old_id
     93    assert q.state == EntityState.DRAFT
     94    assert q.flexo_id.version == 1
     95
     96
     97def test_mass_version_increments_until_obsolete(radio_question):
     98    q = radio_question
     99    q.approve()
    86100    for _ in range(FlexOID.MAX_VERSION - 2):
    87         entity.sign()
     101        q.sign()
    88102    with pytest.raises(RuntimeError, match="mark obsolete"):
    89         entity.sign()
     103        q.sign()

  • tests/test_id_stress.py

    r6a7dec1 r02d288d  
    33Focus: collision avoidance, version ceiling, reproducibility.
    44"""
     5
    56import pytest
    67import random
     8import logging
     9from flexoentity import FlexOID, EntityType, EntityState
     10from builder.questions import RadioQuestion, AnswerOption
    711
    8 from flexoentity import FlexOID, EntityType, EntityState, Domain
     12logger = logging.getLogger(__name__)
    913
    10 from tests.conftest import DummyEntity
    11 
    12 # ──────────────────────────────────────────────────────────────────────────────
    13 def test_bulk_generation_uniqueness():
    14     """Generate 10,000 IDs and assert uniqueness (statistical test)."""
    15     domain = Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT,
    16                     fullname="Signal Corps", classification="RESTRICTED", owner="MESE")
    17 
     14def test_bulk_generation_uniqueness(domain):
     15    """
     16    Generate 10,000 IDs and ensure uniqueness using safe_generate().
     17    If a collision occurs, safe_generate() must resolve it automatically
     18    via salt + date adjustment.
     19    """
    1820    etype = EntityType.QUESTION
    1921    estate = EntityState.DRAFT
    20     seeds = [f"question {i}" for i in range(10_000)]
     22    seeds = [f"question {i}" for i in range(4000000)]
    2123
    22     ids = [FlexOID.generate(domain, etype, estate, seed) for seed in seeds]
     24    # Simulate a simple in-memory repository for collision detection
     25    repo = {}
    2326
    24     assert len(ids) == len(set(ids)), "ID collisions detected in bulk generation"
     27    def repo_get(oid_str):
     28        return repo.get(str(oid_str))
    2529
     30    # Generate IDs using safe_generate
     31    ids = []
     32    for seed in seeds:
     33        oid = FlexOID.safe_generate(domain.domain, etype, estate, seed, repo=repo)
     34        assert isinstance(oid, FlexOID)
     35        ids.append(str(oid))
     36        repo[str(oid)] = oid  # register for future collision detection
    2637
    27 def test_disambiguator_trigger():
     38    unique_count = len(set(ids))
     39    total_count = len(ids)
     40    collisions = total_count - unique_count
     41
     42    logger.info(f"Generated {total_count} IDs ({collisions} collisions handled).")
     43
     44    # Assert that safe_generate avoided duplicates
     45    assert total_count == unique_count, f"Unexpected duplicate IDs ({collisions} found)"
     46
     47    # Sanity check: IDs should look canonical
     48    assert all(id_str.startswith("SIG-") for id_str in ids)
     49    assert all("@" in id_str for id_str in ids)
     50
     51def test_id_generation_is_deterministic(domain):
    2852    """
    2953    Generating the same entity twice with same inputs yields identical ID.
    3054    (No runtime disambiguation; IDs are deterministic by design.)
    3155    """
    32     domain = "AF"
    3356    etype = EntityType.QUESTION
    3457    estate = EntityState.DRAFT
    3558    text = "identical question text"
    36     id1 = FlexOID.generate(domain, etype, estate, text)
    37     id2 = FlexOID.generate(domain, etype, estate, text)
    38     # IDs must be identical, because we now enforce determinism, not randomization
     59
     60    id1 = FlexOID.generate(domain.domain, etype, estate, text)
     61    id2 = FlexOID.generate(domain.domain, etype, estate, text)
     62    # IDs must be identical because generation is deterministic
    3963    assert id1 == id2
    40     assert id1.signature == id2.signature
    4164
    4265
    43 def test_id_reproducibility_across_runs():
     66def test_id_reproducibility_across_runs(domain):
    4467    """
    4568    The same seed on a new process (fresh _seen_hashes)
    4669    should yield the same base ID (without suffix).
    4770    """
    48     domain = Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT,
    49                     fullname="Signal Corps", classification="RESTRICTED")
    5071    etype = EntityType.CATALOG
    5172    estate = EntityState.DRAFT
    5273    seed = "reproducibility test seed"
    53     id1 = FlexOID.generate(domain, etype, estate, seed)
    54     # Reset hash cache
     74
     75    id1 = FlexOID.generate(domain.domain, etype, estate, seed)
    5576    FlexOID._seen_hashes.clear()
    56     id2 = FlexOID.generate(domain, etype, estate, seed)
     77    id2 = FlexOID.generate(domain.domain, etype, estate, seed)
     78
    5779    assert id1 == id2
    58     assert id1.signature == id2.signature
    5980
    6081
    61 def test_version_ceiling_enforcement():
     82def test_version_ceiling_enforcement(radio_question):
    6283    """Simulate approaching @999 to trigger obsolescence guard."""
    63     entity = DummyEntity(domain="AF", etype=EntityType.EXAM, state=EntityState.DRAFT, seed="Final Exam 2025")
    64     entity.approve()
     84    q = radio_question
     85    q.approve()
     86
    6587    # artificially bump version number to near ceiling
    66     entity.flexo_id = FlexOID.from_oid_and_version(entity.flexo_id, 998)
     88    q.flexo_id = FlexOID.from_oid_and_version(q.flexo_id, 998)
    6789
    6890    # 998 → 999 is allowed
    69     entity.sign()
    70     assert entity.flexo_id.version == 999
     91    q.sign()
     92    assert q.flexo_id.version == 999
    7193
    7294    # 999 → 1000 should raise RuntimeError
    7395    with pytest.raises(RuntimeError):
    74         entity.sign()
     96        q.sign()
    7597
    7698
    77 def test_massive_lifecycle_simulation():
     99def test_massive_lifecycle_simulation(domain):
    78100    """
    79     Generate 100 random entities, simulate multiple edits and state transitions,
     101    Generate 100 random RadioQuestions, simulate multiple edits and state transitions,
    80102    ensure all final IDs and fingerprints are unique and valid.
    81103    """
    82     entities = [DummyEntity(domain="AF", etype=EntityType.QUESTION, state=EntityState.DRAFT, seed=f"random question {i}") for i in range(100)]
     104    entities = [
     105        RadioQuestion(
     106            domain=domain,
     107            etype=EntityType.QUESTION,
     108            state=EntityState.DRAFT,
     109            text=f"random question {i}",
     110            options=[
     111                AnswerOption(id="opt4", text="HF (3–30 MHz)", points=1),
     112                AnswerOption(id="opt5", text="VHF (30–300 MHz)", points=0),
     113            ],
     114        )
     115        for i in range(100)
     116    ]
    83117
    84118    for e in entities:
    85         # random edit, approval, signing
    86         e._seed += " updated"
     119        # random edit
     120        e.text += " updated"
    87121        e._update_fingerprint()
     122
     123        # lifecycle transitions
    88124        e.approve()
    89125        if random.random() > 0.3:
    … …  
    92128            e.publish()
    93129
    94     ids = [e.flexo_id for e in entities]
    95     fps = [e.flexo_id.signature for e in entities]
    96     assert len(ids) == len(set(ids)), "Duplicate IDs after random lifecycle"
    97     assert len(fps) == len(set(fps)), "Duplicate fingerprints after random lifecycle"
     130    flexoids = [e.flexo_id for e in entities]
     131    assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs after lifecycle simulation"

  • tests/test_persistance_integrity.py

    r6a7dec1 r02d288d  
    66import pytest
    77
    8 from flexoentity import FlexOID, EntityType, EntityState
    9 from tests.conftest import DummyEntity
     8from flexoentity import EntityState
     9from builder.questions import RadioQuestion, AnswerOption
    1010
    1111
    1212# ──────────────────────────────────────────────────────────────────────────────
    1313@pytest.fixture
    14 def approved_entity():
    15     """A fully published dummy entity for persistence tests."""
    16     e = DummyEntity(
    17         domain="AF",
    18         etype=EntityType.QUESTION,
     14def 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
    1919        state=EntityState.DRAFT,
    20         seed="What is Ohm’s law?"
     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        ],
    2126    )
    22     e.approve()
    23     e.sign()
    24     e.publish()
    25     return e
     27    q.approve()
     28    q.sign()
     29    q.publish()
     30    return q
    2631
    27 @pytest.mark.skip(reason="FlexOIDs are regenerated on import; enable once JSON format is stable")
    28 def test_json_roundtrip_preserves_integrity(approved_entity):
     32
     33@pytest.mark.skip(reason="FlexOIDs regenerated on import; enable once JSON format is stable")
     34def test_json_roundtrip_preserves_integrity(approved_question):
    2935    """
    30     Export to JSON and reload — ensure fingerprints remain valid.
     36    Export to JSON and reload — ensure fingerprints and signatures remain valid.
    3137    """
    32     json_str = approved_entity.to_json()
    33     loaded = approved_entity.__class__.from_json(json_str)
     38    json_str = approved_question.to_json()
     39    loaded = RadioQuestion.from_json(json_str)
    3440
    3541    # Fingerprint and state should match — integrity must pass
    36     assert approved_entity.__class__.verify_integrity(loaded)
     42    assert RadioQuestion.verify_integrity(loaded)
    3743
    3844    # Metadata should be preserved exactly
    39     assert approved_entity.flexo_id.signature == loaded.flexo_id.signature
    40     assert approved_entity.flexo_id == loaded.flexo_id
    41     assert loaded.state == approved_entity.state
     45    assert approved_question.signature == loaded.signature
     46    assert approved_question.flexo_id == loaded.flexo_id
     47    assert loaded.state == approved_question.state
     48
    4249
    4350# ──────────────────────────────────────────────────────────────────────────────
    4451
    45 @pytest.mark.skip(reason="FlexOIDs regenerated on import; tampering detection not applicable yet")
    46 def test_json_tampering_detection(approved_entity):
     52@pytest.mark.skip(reason="FlexOIDs regenerated on import; tampering detection not yet implemented")
     53def test_json_tampering_detection(approved_question):
    4754    """Tampering with content should invalidate fingerprint verification."""
    48     json_str = approved_entity.to_json()
    49     tampered_data = json.loads(json_str)
    50     tampered_data["text_seed"] = "Tampered content injection"
    51     tampered_json = json.dumps(tampered_data)
     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)
    5259
    53     # We use DummyEntity.from_json to reconstruct (FlexoEntity is abstract)
    54     loaded = approved_entity.__class__.from_json(tampered_json)
    55     assert not approved_entity.__class__.verify_integrity(loaded)
     60    loaded = RadioQuestion.from_json(tampered_json)
     61    assert not RadioQuestion.verify_integrity(loaded)
    5662
    5763
    … …  
    5965
    6066@pytest.mark.skip(reason="FlexOIDs regenerated on import; corruption detection not yet applicable")
    61 def test_json_file_corruption(approved_entity, tmp_path):
     67def test_json_file_corruption(approved_question, tmp_path):
    6268    """Simulate file corruption — integrity check must fail."""
    63     file = tmp_path / "entity.json"
    64     json_str = approved_entity.to_json()
     69    file = tmp_path / "question.json"
     70    json_str = approved_question.to_json()
    6571    file.write_text(json_str)
    6672
    67     # Corrupt the file
     73    # Corrupt the file (simulate accidental byte modification)
    6874    corrupted = json_str.replace("Ohm’s", "Omm’s")
    6975    file.write_text(corrupted)
    7076
    71     loaded = approved_entity.__class__.from_json(file.read_text())
    72     assert not approved_entity.__class__.verify_integrity(loaded)
     77    loaded = RadioQuestion.from_json(file.read_text())
     78    assert not RadioQuestion.verify_integrity(loaded)
Note: See TracChangeset for help on using the changeset viewer.