| 1 | """
|
|---|
| 2 | Stress tests for the Flex-O ID lifecycle.
|
|---|
| 3 | Focus: collision avoidance, version ceiling, reproducibility.
|
|---|
| 4 | """
|
|---|
| 5 |
|
|---|
| 6 | import pytest
|
|---|
| 7 | import random
|
|---|
| 8 | import logging
|
|---|
| 9 | from flexoentity import FlexOID, EntityType, EntityState
|
|---|
| 10 | from builder.questions import RadioQuestion, AnswerOption
|
|---|
| 11 |
|
|---|
| 12 | logger = logging.getLogger(__name__)
|
|---|
| 13 |
|
|---|
| 14 | def 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 | """
|
|---|
| 20 | etype = EntityType.QUESTION
|
|---|
| 21 | estate = EntityState.DRAFT
|
|---|
| 22 | seeds = [f"question {i}" for i in range(4000000)]
|
|---|
| 23 |
|
|---|
| 24 | # Simulate a simple in-memory repository for collision detection
|
|---|
| 25 | repo = {}
|
|---|
| 26 |
|
|---|
| 27 | def repo_get(oid_str):
|
|---|
| 28 | return repo.get(str(oid_str))
|
|---|
| 29 |
|
|---|
| 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
|
|---|
| 37 |
|
|---|
| 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 |
|
|---|
| 51 | def test_id_generation_is_deterministic(domain):
|
|---|
| 52 | """
|
|---|
| 53 | Generating the same entity twice with same inputs yields identical ID.
|
|---|
| 54 | (No runtime disambiguation; IDs are deterministic by design.)
|
|---|
| 55 | """
|
|---|
| 56 | etype = EntityType.QUESTION
|
|---|
| 57 | estate = EntityState.DRAFT
|
|---|
| 58 | text = "identical question text"
|
|---|
| 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
|
|---|
| 63 | assert id1 == id2
|
|---|
| 64 |
|
|---|
| 65 |
|
|---|
| 66 | def test_id_reproducibility_across_runs(domain):
|
|---|
| 67 | """
|
|---|
| 68 | The same seed on a new process (fresh _seen_hashes)
|
|---|
| 69 | should yield the same base ID (without suffix).
|
|---|
| 70 | """
|
|---|
| 71 | etype = EntityType.CATALOG
|
|---|
| 72 | estate = EntityState.DRAFT
|
|---|
| 73 | seed = "reproducibility test seed"
|
|---|
| 74 |
|
|---|
| 75 | id1 = FlexOID.generate(domain.domain, etype, estate, seed)
|
|---|
| 76 | FlexOID._seen_hashes.clear()
|
|---|
| 77 | id2 = FlexOID.generate(domain.domain, etype, estate, seed)
|
|---|
| 78 |
|
|---|
| 79 | assert id1 == id2
|
|---|
| 80 |
|
|---|
| 81 |
|
|---|
| 82 | def test_version_ceiling_enforcement(radio_question):
|
|---|
| 83 | """Simulate approaching @999 to trigger obsolescence guard."""
|
|---|
| 84 | q = radio_question
|
|---|
| 85 | q.approve()
|
|---|
| 86 |
|
|---|
| 87 | # artificially bump version number to near ceiling
|
|---|
| 88 | q.flexo_id = FlexOID.from_oid_and_version(q.flexo_id, 998)
|
|---|
| 89 |
|
|---|
| 90 | # 998 → 999 is allowed
|
|---|
| 91 | q.sign()
|
|---|
| 92 | assert q.flexo_id.version == 999
|
|---|
| 93 |
|
|---|
| 94 | # 999 → 1000 should raise RuntimeError
|
|---|
| 95 | with pytest.raises(RuntimeError):
|
|---|
| 96 | q.sign()
|
|---|
| 97 |
|
|---|
| 98 |
|
|---|
| 99 | def test_massive_lifecycle_simulation(domain):
|
|---|
| 100 | """
|
|---|
| 101 | Generate 100 random RadioQuestions, simulate multiple edits and state transitions,
|
|---|
| 102 | ensure all final IDs and fingerprints are unique and valid.
|
|---|
| 103 | """
|
|---|
| 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 | ]
|
|---|
| 117 |
|
|---|
| 118 | for e in entities:
|
|---|
| 119 | # random edit
|
|---|
| 120 | e.text += " updated"
|
|---|
| 121 | e._update_fingerprint()
|
|---|
| 122 |
|
|---|
| 123 | # lifecycle transitions
|
|---|
| 124 | e.approve()
|
|---|
| 125 | if random.random() > 0.3:
|
|---|
| 126 | e.sign()
|
|---|
| 127 | if random.random() > 0.6:
|
|---|
| 128 | e.publish()
|
|---|
| 129 |
|
|---|
| 130 | flexoids = [e.flexo_id for e in entities]
|
|---|
| 131 | assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs after lifecycle simulation"
|
|---|