source: flexoentity/tests/test_id_stress.py@ 02d288d

Last change on this file since 02d288d was 02d288d, checked in by Enrico Schwass <ennoausberlin@…>, 3 months ago

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

  • Property mode set to 100644
File size: 4.0 KB
Line 
1"""
2Stress tests for the Flex-O ID lifecycle.
3Focus: collision avoidance, version ceiling, reproducibility.
4"""
5
6import pytest
7import random
8import logging
9from flexoentity import FlexOID, EntityType, EntityState
10from builder.questions import RadioQuestion, AnswerOption
11
12logger = logging.getLogger(__name__)
13
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 """
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
51def 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
66def 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
82def 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
99def 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"
Note: See TracBrowser for help on using the repository browser.