- Timestamp:
- 11/01/25 15:51:10 (2 months ago)
- Branches:
- master
- Children:
- 5c72356
- Parents:
- ca39274
- Location:
- tests
- Files:
-
- 1 added
- 4 edited
-
conftest.py (modified) (1 diff)
-
test_flexoid.py (added)
-
test_id_lifecycle.py (modified) (10 diffs)
-
test_id_stress.py (modified) (6 diffs)
-
test_persistance_integrity.py (modified) (4 diffs)
Legend:
- Unmodified
- Added
- Removed
-
tests/conftest.py
rca39274 r8aa20c7 1 # tests/conftest.py 2 1 # tests/stubs/single_choice_question.py 3 2 import pytest 4 import json 3 from datetime import datetime 4 from dataclasses import dataclass, field 5 from typing import List 5 6 from flexoentity import FlexoEntity, EntityType, EntityState, Domain 6 7 7 import pytest 8 import json 9 from flexoentity import EntityType, EntityState, Domain 10 from builder.questions import RadioQuestion, AnswerOption # adjust path if different 11 from builder.media_items import NullMediaItem # adjust import if needed 8 @pytest.fixture 9 def fixed_datetime(monkeypatch): 10 class FixedDate(datetime): 11 @classmethod 12 def now(cls, tz=None): 13 return datetime(2025, 11, 1, tzinfo=tz) 14 monkeypatch.setattr("flexoentity.id_factory.datetime", FixedDate) 15 return FixedDate 12 16 13 17 14 @pytest.fixture(scope="session") 15 def 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 ) 18 @dataclass 19 class AnswerOption: 20 id: str 21 text: str 22 points: float = 0.0 23 24 def to_dict(self): 25 return {"id": self.id, "text": self.text, "points": self.points} 26 27 @classmethod 28 def from_dict(cls, data): 29 return cls( 30 id=data.get("id", ""), 31 text=data.get("text", ""), 32 points=data.get("points", 0.0) 33 ) 26 34 27 35 28 @pytest.fixture 29 def 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 36 @dataclass 37 class SingleChoiceQuestion(FlexoEntity): 38 """A minimal stub to test FlexoEntity integration.""" 39 text: str = "" 40 options: List[AnswerOption] = field(default_factory=list) 43 41 44 42 45 @pytest.fixture 46 def serialized_question(radio_question): 47 """Provide the serialized JSON form for roundtrip tests.""" 48 return radio_question.to_json() 43 @classmethod 44 def default(cls): 45 return cls(domain=Domain(domain="GEN", 46 entity_type=EntityType.DOMAIN, 47 state=EntityState.DRAFT), 48 state=EntityState.DRAFT, entity_type=EntityType.ITEM) 49 49 50 def to_dict(self): 51 base = super().to_dict() 52 base.update({ 53 "text": self.text, 54 "options": [opt.to_dict() for opt in self.options], 55 }) 56 return base 57 58 @property 59 def text_seed(self) -> str: 60 """Include answer options (and points) for deterministic ID generation.""" 61 62 joined = "|".join( 63 f"{opt.text.strip()}:{opt.points}" 64 for opt in sorted(self.options, key=lambda o: o.text.strip().lower()) 65 ) 66 return f"{self.text}{joined}" 67 68 @classmethod 69 def from_dict(cls, data): 70 obj = cls( 71 text=data.get("text", ""), 72 options=[AnswerOption.from_dict(o) for o in data.get("options", [])], 73 ) 74 # restore FlexoEntity core fields 75 obj.domain = data.get("domain") 76 obj.entity_type = EntityType[data.get("etype")] if "etype" in data else EntityType.ITEM 77 obj.state = EntityState[data.get("state")] if "state" in data else EntityState.DRAFT 78 if "flexo_id" in data: 79 from flexoentity import FlexOID 80 obj.flexo_id = FlexOID.parsed(data["flexo_id"]) 81 return obj 50 82 51 83 @pytest.fixture 52 def deserialized_question(serialized_question): 53 """Recreate a question from JSON for consistency tests.""" 54 return RadioQuestion.from_json(serialized_question) 55 84 def domain(): 85 return Domain.default() 56 86 57 87 @pytest.fixture 58 def null_media(): 59 """Provide a default NullMediaItem instance for media tests.""" 60 return NullMediaItem( 61 domain=domain, 62 etype=EntityType.MEDIA, 63 state=EntityState.DRAFT 64 ) 88 def sample_question(): 89 return SingleChoiceQuestion(domain=Domain.default(), 90 text="What is 2 + 2?", 91 options=[], 92 entity_type=EntityType.ITEM, 93 state=EntityState.DRAFT) -
tests/test_id_lifecycle.py
rca39274 r8aa20c7 1 1 import pytest 2 from flexoentity import FlexOID, FlexoEntity, Entity Type, EntityState2 from flexoentity import FlexOID, FlexoEntity, EntityState 3 3 4 4 5 5 # ────────────────────────────────────────────────────────────────────────────── 6 # Tests adapted to use real RadioQuestion fixture instead of DummyEntity6 # Tests adapted to use real SingleChoiceQuestion fixture instead of DummyEntity 7 7 # ────────────────────────────────────────────────────────────────────────────── 8 8 9 def test_initial_state( radio_question):10 q = radio_question9 def test_initial_state(sample_question): 10 q = sample_question 11 11 assert q.state == EntityState.DRAFT 12 12 assert q.flexo_id.version == 1 … … 14 14 15 15 16 def test_approval_ bumps_version(radio_question):17 q = radio_question16 def test_approval_does_not_bump_version(sample_question): 17 q = sample_question 18 18 q.approve() 19 19 assert q.state == EntityState.APPROVED 20 assert q.flexo_id.version == 220 assert q.flexo_id.version == 1 21 21 22 22 23 def test_signing_bumps_version( radio_question):24 q = radio_question23 def test_signing_bumps_version(sample_question): 24 q = sample_question 25 25 q.approve() 26 26 v_before = str(q.flexo_id) … … 30 30 31 31 32 def test_publish_bumps_version( radio_question):33 q = radio_question32 def test_publish_bumps_version(sample_question): 33 q = sample_question 34 34 q.approve() 35 35 q.sign() … … 40 40 41 41 42 def test_modify_content_changes_fingerprint( radio_question):43 q = radio_question44 q.text = "Rephrased content" # simulate text change42 def test_modify_content_changes_fingerprint(sample_question): 43 q = sample_question 44 q.text += "Rephrased content" # simulate text change 45 45 changed = q._update_fingerprint() 46 46 assert changed 47 47 48 48 49 def test_no_version_bump_on_draft_edits( radio_question):50 q = radio_question49 def test_no_version_bump_on_draft_edits(sample_question): 50 q = sample_question 51 51 q.text = "Minor draft edit" 52 52 q._update_fingerprint() … … 54 54 55 55 56 def test_version_bump_after_edit_and_sign( radio_question):57 q = radio_question56 def test_version_bump_after_edit_and_sign(sample_question): 57 q = sample_question 58 58 q.approve() 59 59 v1 = str(q.flexo_id) … … 63 63 64 64 65 def test_integrity_check_passes_and_fails( radio_question):66 q = radio_question65 def test_integrity_check_passes_and_fails(sample_question): 66 q = sample_question 67 67 q.approve() 68 68 assert FlexoEntity.verify_integrity(q) … … 73 73 74 74 75 def test_obsolete_state( radio_question):76 q = radio_question75 def test_obsolete_state(sample_question): 76 q = sample_question 77 77 q.approve() 78 78 q.sign() … … 82 82 83 83 84 def test_clone_new_base_resets_lineage( radio_question):85 q = radio_question84 def test_clone_new_base_resets_lineage(sample_question): 85 q = sample_question 86 86 q.approve() 87 87 q.sign() … … 94 94 assert q.flexo_id.version == 1 95 95 96 def test_clone_new_base_sets_origin( radio_question):97 q = radio_question96 def test_clone_new_base_sets_origin(sample_question): 97 q = sample_question 98 98 q.approve() 99 99 q.sign() … … 107 107 assert q.flexo_id != old_id 108 108 109 def test_mass_version_increments_until_obsolete( radio_question):110 q = radio_question109 def test_mass_version_increments_until_obsolete(sample_question): 110 q = sample_question 111 111 q.approve() 112 for _ in range(FlexOID.MAX_VERSION - 2):112 for _ in range(FlexOID.MAX_VERSION - 1): 113 113 q.bump_version() 114 114 with pytest.raises(RuntimeError, match="mark obsolete"): -
tests/test_id_stress.py
rca39274 r8aa20c7 4 4 """ 5 5 6 import copy 7 import logging 8 import random 9 6 10 import pytest 7 import random 8 import logging 11 9 12 from flexoentity import FlexOID, EntityType, EntityState 10 from builder.questions import RadioQuestion, AnswerOption11 13 12 14 logger = logging.getLogger(__name__) … … 18 20 via salt + date adjustment. 19 21 """ 20 e type = EntityType.QUESTION22 entity_type = EntityType.ITEM 21 23 estate = EntityState.DRAFT 22 24 seeds = [f"question {i}" for i in range(100000)] … … 31 33 ids = [] 32 34 for seed in seeds: 33 oid = FlexOID.safe_generate(domain.domain, e type, estate, seed, repo=repo)35 oid = FlexOID.safe_generate(domain.domain, entity_type.value, estate.value, seed, repo=repo) 34 36 assert isinstance(oid, FlexOID) 35 37 ids.append(str(oid)) … … 46 48 47 49 # Sanity check: IDs should look canonical 48 assert all(id_str.startswith(" SIG-") for id_str in ids)50 assert all(id_str.startswith("GEN") for id_str in ids) 49 51 assert all("@" in id_str for id_str in ids) 50 52 … … 54 56 (No runtime disambiguation; IDs are deterministic by design.) 55 57 """ 56 e type = EntityType.QUESTION58 entity_type = EntityType.ITEM 57 59 estate = EntityState.DRAFT 58 60 text = "identical question text" 59 61 60 id1 = FlexOID.generate(domain.domain, e type, estate, text)61 id2 = FlexOID.generate(domain.domain, e type, estate, text)62 id1 = FlexOID.generate(domain.domain, entity_type.value, estate.value, text) 63 id2 = FlexOID.generate(domain.domain, entity_type.value, estate.value, text) 62 64 # IDs must be identical because generation is deterministic 63 65 assert id1 == id2 64 66 65 67 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.CATALOG72 estate = EntityState.DRAFT73 seed = "reproducibility test seed"68 # def test_id_reproducibility_across_runs(domain): 69 # """ 70 # The same seed on a new process (fresh _seen_hashes) 71 # should yield the same base ID (without suffix). 72 # """ 73 # entity_type = EntityType.CATALOG 74 # estate = EntityState.DRAFT 75 # seed = "reproducibility test seed" 74 76 75 id1 = FlexOID.generate(domain.domain, etype, estate, seed)76 FlexOID._seen_hashes.clear()77 id2 = FlexOID.generate(domain.domain, etype, estate, seed)77 # id1 = FlexOID.generate(domain.domain, entity_type.value, estate.value, seed) 78 # FlexOID._seen_hashes.clear() 79 # id2 = FlexOID.generate(domain.domain, entity_type.value, estate.value, seed) 78 80 79 assert id1 == id281 # assert id1 == id2 80 82 81 83 82 def test_version_ceiling_enforcement( radio_question):84 def test_version_ceiling_enforcement(sample_question): 83 85 """Simulate approaching @999 to trigger obsolescence guard.""" 84 q = radio_question86 q = sample_question 85 87 q.approve() 86 88 … … 97 99 98 100 99 def test_massive_lifecycle_simulation( domain):101 def test_massive_lifecycle_simulation(sample_question): 100 102 """ 101 Generate 100 random RadioQuestions, simulate multiple edits and state transitions,103 Generate 100 random SingleChoiceQuestions, simulate multiple edits and state transitions, 102 104 ensure all final IDs and fingerprints are unique and valid. 103 105 """ 104 106 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) 107 copy.deepcopy(sample_question) for _ in range(100) 116 108 ] 117 109 118 for e in entities:110 for i, e in enumerate(entities): 119 111 # random edit 120 e.text += " updated"112 e.text += f" updated #{i}" 121 113 e._update_fingerprint() 122 114 -
tests/test_persistance_integrity.py
rca39274 r8aa20c7 6 6 import pytest 7 7 8 from builder.questions import RadioQuestion, AnswerOption9 8 from flexoentity import EntityState, EntityType, Domain 10 9 11 10 @pytest.fixture 12 11 def approved_question(): 13 """Provide a fully approved and published RadioQuestion for persistence tests."""14 q = RadioQuestion(15 domain=Domain(domain="GEN", e type=EntityType.DOMAIN, state=EntityState.DRAFT),16 e type=None, # RadioQuestion sets this internally to EntityType.QUESTION12 """Provide a fully approved and published SingleChoiceQuestion for persistence tests.""" 13 q = SingleChoiceQuestion( 14 domain=Domain(domain="GEN", entity_type=EntityType.DOMAIN, state=EntityState.DRAFT), 15 entity_type=None, # SingleChoiceQuestion sets this internally to EntityType.ITEM 17 16 state=EntityState.DRAFT, 18 17 text="What is Ohm’s law?", … … 36 35 json_str = approved_question.to_json() 37 36 print("JSON", json_str) 38 loaded = RadioQuestion.from_json(json_str)37 loaded = SingleChoiceQuestion.from_json(json_str) 39 38 40 39 print("Approved", approved_question.text_seed) 41 40 print("Loaded", loaded.text_seed) 42 41 # Fingerprint and state should match — integrity must pass 43 assert RadioQuestion.verify_integrity(loaded)42 assert SingleChoiceQuestion.verify_integrity(loaded) 44 43 45 44 # Metadata should be preserved exactly … … 56 55 tampered_json = json.dumps(tampered) 57 56 58 loaded = RadioQuestion.from_json(tampered_json)59 assert not RadioQuestion.verify_integrity(loaded)57 loaded = SingleChoiceQuestion.from_json(tampered_json) 58 assert not SingleChoiceQuestion.verify_integrity(loaded) 60 59 61 60 @pytest.mark.skip(reason="FlexOIDs regenerated on import; corruption detection not yet applicable") … … 70 69 file.write_text(corrupted) 71 70 72 loaded = RadioQuestion.from_json(file.read_text())73 assert not RadioQuestion.verify_integrity(loaded)71 loaded = SingleChoiceQuestion.from_json(file.read_text()) 72 assert not SingleChoiceQuestion.verify_integrity(loaded)
Note:
See TracChangeset
for help on using the changeset viewer.
