Changeset 02d288d in flexoentity for tests/test_id_stress.py

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

File:

• tests/test_id_stress.py (modified) (2 diffs)

tests/test_id_stress.py

@@ -3 +3 @@
 Focus: collision avoidance, version ceiling, reproducibility.
 """
+
 import pytest
 import random
+import logging
+from flexoentity import FlexOID, EntityType, EntityState
+from builder.questions import RadioQuestion, AnswerOption
 
-from flexoentity import FlexOID, EntityType, EntityState, Domain
+logger = logging.getLogger(__name__)
 
-from tests.conftest import DummyEntity
-
-# ──────────────────────────────────────────────────────────────────────────────
-def test_bulk_generation_uniqueness():
-    """Generate 10,000 IDs and assert uniqueness (statistical test)."""
-    domain = Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT,
-                    fullname="Signal Corps", classification="RESTRICTED", owner="MESE")
-
+def test_bulk_generation_uniqueness(domain):
+    """
+    Generate 10,000 IDs and ensure uniqueness using safe_generate().
+    If a collision occurs, safe_generate() must resolve it automatically
+    via salt + date adjustment.
+    """
     etype = EntityType.QUESTION
     estate = EntityState.DRAFT
-    seeds = [f"question {i}" for i in range(10_000)]
+    seeds = [f"question {i}" for i in range(4000000)]
 
-    ids = [FlexOID.generate(domain, etype, estate, seed) for seed in seeds]
+    # Simulate a simple in-memory repository for collision detection
+    repo = {}
 
-    assert len(ids) == len(set(ids)), "ID collisions detected in bulk generation"
+    def repo_get(oid_str):
+        return repo.get(str(oid_str))
 
+    # Generate IDs using safe_generate
+    ids = []
+    for seed in seeds:
+        oid = FlexOID.safe_generate(domain.domain, etype, estate, seed, repo=repo)
+        assert isinstance(oid, FlexOID)
+        ids.append(str(oid))
+        repo[str(oid)] = oid  # register for future collision detection
 
-def test_disambiguator_trigger():
+    unique_count = len(set(ids))
+    total_count = len(ids)
+    collisions = total_count - unique_count
+
+    logger.info(f"Generated {total_count} IDs ({collisions} collisions handled).")
+
+    # Assert that safe_generate avoided duplicates
+    assert total_count == unique_count, f"Unexpected duplicate IDs ({collisions} found)"
+
+    # Sanity check: IDs should look canonical
+    assert all(id_str.startswith("SIG-") for id_str in ids)
+    assert all("@" in id_str for id_str in ids)
+
+def test_id_generation_is_deterministic(domain):
     """
     Generating the same entity twice with same inputs yields identical ID.
     (No runtime disambiguation; IDs are deterministic by design.)
     """
-    domain = "AF"
     etype = EntityType.QUESTION
     estate = EntityState.DRAFT
     text = "identical question text"
-    id1 = FlexOID.generate(domain, etype, estate, text)
-    id2 = FlexOID.generate(domain, etype, estate, text)
-    # IDs must be identical, because we now enforce determinism, not randomization
+
+    id1 = FlexOID.generate(domain.domain, etype, estate, text)
+    id2 = FlexOID.generate(domain.domain, etype, estate, text)
+    # IDs must be identical because generation is deterministic
     assert id1 == id2
-    assert id1.signature == id2.signature
 
 
-def test_id_reproducibility_across_runs():
+def test_id_reproducibility_across_runs(domain):
     """
     The same seed on a new process (fresh _seen_hashes)
     should yield the same base ID (without suffix).
     """
-    domain = Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT,
-                    fullname="Signal Corps", classification="RESTRICTED")
     etype = EntityType.CATALOG
     estate = EntityState.DRAFT
     seed = "reproducibility test seed"
-    id1 = FlexOID.generate(domain, etype, estate, seed)
-    # Reset hash cache
+
+    id1 = FlexOID.generate(domain.domain, etype, estate, seed)
     FlexOID._seen_hashes.clear()
-    id2 = FlexOID.generate(domain, etype, estate, seed)
+    id2 = FlexOID.generate(domain.domain, etype, estate, seed)
+
     assert id1 == id2
-    assert id1.signature == id2.signature
 
 
-def test_version_ceiling_enforcement():
+def test_version_ceiling_enforcement(radio_question):
     """Simulate approaching @999 to trigger obsolescence guard."""
-    entity = DummyEntity(domain="AF", etype=EntityType.EXAM, state=EntityState.DRAFT, seed="Final Exam 2025")
-    entity.approve()
+    q = radio_question
+    q.approve()
+
     # artificially bump version number to near ceiling
-    entity.flexo_id = FlexOID.from_oid_and_version(entity.flexo_id, 998)
+    q.flexo_id = FlexOID.from_oid_and_version(q.flexo_id, 998)
 
     # 998 → 999 is allowed
-    entity.sign()
-    assert entity.flexo_id.version == 999
+    q.sign()
+    assert q.flexo_id.version == 999
 
     # 999 → 1000 should raise RuntimeError
     with pytest.raises(RuntimeError):
-        entity.sign()
+        q.sign()
 
 
-def test_massive_lifecycle_simulation():
+def test_massive_lifecycle_simulation(domain):
     """
-    Generate 100 random entities, simulate multiple edits and state transitions,
+    Generate 100 random RadioQuestions, simulate multiple edits and state transitions,
     ensure all final IDs and fingerprints are unique and valid.
     """
-    entities = [DummyEntity(domain="AF", etype=EntityType.QUESTION, state=EntityState.DRAFT, seed=f"random question {i}") for i in range(100)]
+    entities = [
+        RadioQuestion(
+            domain=domain,
+            etype=EntityType.QUESTION,
+            state=EntityState.DRAFT,
+            text=f"random question {i}",
+            options=[
+                AnswerOption(id="opt4", text="HF (3–30 MHz)", points=1),
+                AnswerOption(id="opt5", text="VHF (30–300 MHz)", points=0),
+            ],
+        )
+        for i in range(100)
+    ]
 
     for e in entities:
-        # random edit, approval, signing
-        e._seed += " updated"
+        # random edit
+        e.text += " updated"
         e._update_fingerprint()
+
+        # lifecycle transitions
         e.approve()
         if random.random() > 0.3:
@@ -92 +128 @@
             e.publish()
 
-    ids = [e.flexo_id for e in entities]
-    fps = [e.flexo_id.signature for e in entities]
-    assert len(ids) == len(set(ids)), "Duplicate IDs after random lifecycle"
-    assert len(fps) == len(set(fps)), "Duplicate fingerprints after random lifecycle"
+    flexoids = [e.flexo_id for e in entities]
+    assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs after lifecycle simulation"