Index: flexoentity/domain.py =================================================================== --- flexoentity/domain.py (revision 6a7dec1b7c47ac7a8618ebec0981ce40dbc3224d) +++ flexoentity/domain.py (revision 02d288d33183db0124582a151433ccd6cf55cac7) @@ -14,5 +14,4 @@ def __post_init__(self): - self.etype = EntityType.DOMAIN super().__post_init__() @@ -20,5 +19,5 @@ def text_seed(self) -> str: """Deterministic text seed for ID generation.""" - return f"{self.domain}|{self.fullname}|{self.classification}|{self.owner}" + return f"{self.fullname}|{self.classification}|{self.owner}" def to_dict(self): Index: flexoentity/flexo_entity.py =================================================================== --- flexoentity/flexo_entity.py (revision 6a7dec1b7c47ac7a8618ebec0981ce40dbc3224d) +++ flexoentity/flexo_entity.py (revision 02d288d33183db0124582a151433ccd6cf55cac7) @@ -89,10 +89,13 @@ -@dataclass +@dataclass(kw_only=True) class FlexoEntity(ABC): domain: str etype: EntityType state: EntityState - + flexo_id: Optional[FlexOID] = field(default=None) + signature: str = field(default_factory=str) + origin: Optional[str] = field(default=None) + OID_PATTERN = re.compile( r"^(?P[A-Z0-9]+)-(?P[A-Z]+)" @@ -101,5 +104,5 @@ def __str__(self) -> str: - return f"{self.domain}-{self.etype}{self.date}-{self.unique_hash}@{self.version:03d}{self.state}" + return f"{self.domain_code()}-{self.etype}{self.date}-{self.unique_hash}@{self.version:03d}{self.state}" @classmethod @@ -121,4 +124,7 @@ raise NotImplementedError("Subclasses must define text_seed property") + def canonical_seed(self) -> str: + return canonical_seed(self.text_seed) + @classmethod @abstractmethod @@ -127,37 +133,58 @@ raise NotImplementedError("Subclasses must implement default()") + def domain_code(self) -> str: + """Return canonical domain code for serialization and ID generation.""" + return self.domain.domain if hasattr(self.domain, "domain") else self.domain + def __post_init__(self): - """Generate ID and content fingerprint.""" - self.flexo_id = FlexOID.generate(self.domain, + """ + Generate ID and content fingerprint. + + All entities must carry a `.domain` attribute exposing a domain code string. + This may be a `Domain` instance or a temporary wrapper used by the `Domain` + class itself to avoid circular initialization. + """ + + self.flexo_id = FlexOID.generate(self.domain_code(), self.etype.short(), self.state.short(), self.text_seed, 1) + seed = canonical_seed(self.text_seed) + self.signature = hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper() def __str__(self): return ( f"{self.etype.name}({self.flexo_id}, {self.state.name}, " - f"sig={self.flexo_id.signature}..., v{self.version})" - ) - + f"sig={self.signature}..., v{self.version})" + ) def to_dict(self): return { - "domain": self.domain, + "domain": self.domain_code(), "etype": self.etype.name, "state": self.state.name, "flexo_id": str(self.flexo_id), + "signature": self.signature, + "origin": self.origin, } - + @classmethod def from_dict(cls, data): + from flexoentity.domain import Domain # avoid circular import + domain_obj = Domain( + domain=data["domain"], + etype=EntityType.DOMAIN, + state=EntityState.DRAFT, # default when reconstructing context + ) obj = cls( - data["domain"], - EntityType[data["etype"]], - data["text_seed"], - EntityState[data["state"]], - ) - obj.flexo_id = FlexOID(data["flexo_id"], data.get("signature", "")) + domain=domain_obj, + etype=EntityType[data["etype"]], + state=EntityState[data["state"]], + ) + obj.flexo_id = FlexOID.from_string(data["flexo_id"]) + obj.signature = data.get("signature", "") + obj.origin = data.get("origin") return obj - + def to_json(self, *, indent: int | None = None) -> str: """Serialize entity (and its FlexOID) into JSON.""" @@ -184,15 +211,22 @@ EntityState.PUBLISHED, ) - - # ─────────────────────────────────────────────────────────────── + def _update_signature(self) -> str: + """Always recompute the entity's content signature.""" + seed = self.canonical_seed() + return hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper() + def _update_fingerprint(self) -> bool: - """Recalculate fingerprint and return True if content changed.""" - # extract version from current flexo_id - new_oid = FlexOID.generate(self.domain, self.etype.short(), self.state.short(), self.text_seed, self.flexo_id.version) - if new_oid.signature != self.flexo_id.signature: - self.flexo_id = new_oid + """Update FlexOID if the content signature changed.""" + new_sig = self._update_signature() + if new_sig != self.signature: + self.signature = new_sig + self.flexo_id = FlexOID.generate(self.domain_code(), + self.etype.short(), + self.state.short(), + self.text_seed, + self.flexo_id.version) return True return False - + # ─────────────────────────────────────────────────────────────── def _transition(self, target_state: EntityState): @@ -234,6 +268,5 @@ if self.state == EntityState.DRAFT: new_version = self.flexo_id.version + 1 - new_fid = FlexOID.generate( - self.domain, + new_fid = FlexOID.generate(self.domain_code(), self.etype.short(), EntityState.APPROVED.short(), @@ -262,5 +295,5 @@ """Start new lineage when obsolete.""" self.flexo_id = FlexOID.clone_new_base( - self.domain, + self.domain_code(), self.etype.short(), self.state.short(), @@ -276,11 +309,9 @@ @staticmethod def verify_integrity(entity) -> bool: - # --- inhaltlicher (kryptographischer) Check --- - # Hash ohne State, Signatur mit State - hash_seed = canonical_seed(f"{entity.domain}:{entity.etype.short()}:{entity.text_seed}") - sig_seed = f"{hash_seed}:{entity.state.short()}" - - expected_sig = hashlib.blake2s(sig_seed.encode("utf-8"), digest_size=8).hexdigest().upper() - return expected_sig == entity.flexo_id.signature + """Verify that an entity’s content signature matches its actual content.""" + expected_sig = hashlib.blake2s( + canonical_seed(entity.text_seed).encode("utf-8"), digest_size=8 + ).hexdigest().upper() + return expected_sig == entity.signature def allowed_transitions(self) -> list[str]: Index: flexoentity/id_factory.py =================================================================== --- flexoentity/id_factory.py (revision 6a7dec1b7c47ac7a8618ebec0981ce40dbc3224d) +++ flexoentity/id_factory.py (revision 02d288d33183db0124582a151433ccd6cf55cac7) @@ -13,8 +13,10 @@ from datetime import datetime, timezone import hashlib +import secrets import itertools import json logger = logging.getLogger(__name__) + # ────────────────────────────────────────────────────────────────────────────── # Canonicalization helpers @@ -50,6 +52,5 @@ @classmethod def from_string(cls, id_str: str): - # reconstruct without a known signature - return cls(id_str, signature="") + return cls(id_str) @classmethod @@ -57,44 +58,84 @@ if not (1 <= version <= cls.MAX_VERSION): raise ValueError(f"Version {version} out of bounds (1..{cls.MAX_VERSION}).") - return FlexOID(f"{oid.prefix}@{version:03d}{oid.state_code}", oid.signature) - - def __init__(self, flexo_id: str, signature: str): + return FlexOID(f"{oid.prefix}@{version:03d}{oid.state_code}") + + def __init__(self, flexo_id: str): self.flexo_id = flexo_id - self.signature = signature def __eq__(self, other): if not isinstance(other, FlexOID): return NotImplemented - return self.flexo_id == other.flexo_id and self.signature == other.signature + return self.flexo_id == other.flexo_id def __lt__(self, other): + if not isinstance(other, FlexOID): + return NotImplemented + if self.prefix != other.prefix: + raise ValueError("Cannot order FlexOIDs from different prefixes") return self.version < other.version - + def __hash__(self): - return hash((self.flexo_id, self.signature)) + return hash(self.flexo_id) @staticmethod def _blake_hash(text: str) -> str: """Return a 6-hex BLAKE2s digest.""" - return hashlib.blake2s(text.encode("utf-8"), digest_size=3).hexdigest().upper() # 3 bytes → 6 hex + return hashlib.blake2s(text.encode("utf-8"), digest_size=6).hexdigest().upper() # 6 bytes → 12 hex @staticmethod - def _ensure_unique(hash_part: str) -> str: - """Append disambiguator only if the hash was already seen this session.""" - if hash_part not in FlexOID._seen_hashes: - FlexOID._seen_hashes.add(hash_part) - return hash_part - # fallback only if truly same hash (rare) - for suffix in range(1, 100): - alt = f"{hash_part}-{suffix}" - if alt not in FlexOID._seen_hashes: - FlexOID._seen_hashes.add(alt) - return alt - raise RuntimeError("Too many collisions; adjust hash length or logic.") - - + def safe_generate(domain, etype, estate, text, version=1, repo=None): + """ + Generate a new FlexOID with deterministic hashing, handling rare collisions. + """ + + # Normalize domain (Domain object or string) + domain_code = getattr(domain, "domain", domain) + + # Generate the deterministic candidate OID + oid = FlexOID.generate(domain_code, etype, estate, text, version=version) + + # Collision detection — only if a repository is available + if repo is not None: + existing = repo.get(str(oid)) if hasattr(repo, "get") else repo.get(oid) + else: + existing = None + + if existing: + try: + same_seed = existing.text_seed == text or \ + existing.canonical_seed() == canonical_seed(text) + except Exception: + same_seed = False + + if not same_seed: + # Collision detected — regenerate deterministically + print("Collision detected", len(repo)) + logger.warning(f"FlexOID collision detected for {oid}") + + # (A) refresh date + date_part = datetime.now(timezone.utc).strftime("%y%m%d") + + # (B) add minimal deterministic salt (2 hex chars) + salt = secrets.token_hex(1) + salted_text = f"{text}|salt:{salt}" + + # (C) generate new OID with new date and salted seed + oid = FlexOID.generate( + domain_code, + etype, + estate, + salted_text, + version=version, + ) + + # (D) record lineage if the caller has `origin` tracking + if hasattr(existing, "flexo_id"): + logger.info(f"New lineage created from {existing.flexo_id}") + + return oid + @staticmethod def generate(domain: str, etype: str, estate: str, text: str, - version: int = 1, enforce_unique=True): + version: int = 1): """ Generate a deterministic Flex-O ID. @@ -102,5 +143,4 @@ - The hash (and therefore prefix) depends only on domain, etype, and text. → Prefix stays stable across state changes. - - The signature still includes the state for audit integrity. """ @@ -117,9 +157,5 @@ flexo_id_str = f"{domain}-{etype}{date_part}-{base_hash}@{ver_part}" - # state-dependent signature → per-state integrity - sig_seed = f"{hash_seed}:{estate}" - signature = hashlib.blake2s(sig_seed.encode("utf-8"), digest_size=8).hexdigest().upper() - - return FlexOID(flexo_id_str, signature) + return FlexOID(flexo_id_str) # ────────────────────────────────────────────────────────────────────────── @@ -213,5 +249,4 @@ "version": self.version, "state": self.state_code, - "signature": self.signature, } # ────────────────────────────────────────────────────────────────────────── @@ -223,5 +258,5 @@ Increments the version counter of an existing FlexOID while preserving - its prefix and digital signature. Used when an entity transitions to a + its prefix. Used when an entity transitions to a new revision within the same lifecycle (e.g., minor updates or approvals). @@ -234,5 +269,5 @@ ------- FlexOID - A new Flex-O ID with the same prefix and signature, but version +1. + A new Flex-O ID with the same prefix, but version +1. Raises @@ -243,5 +278,4 @@ Notes ----- - - The signature remains unchanged since the entity lineage is continuous. - Warnings are logged when the version approaches obsolescence. """ @@ -254,5 +288,5 @@ new_id = f"{oid.prefix}@{new_ver:03d}{oid.state_code}" - return cls(new_id, oid.signature) + return cls(new_id) # ────────────────────────────────────────────────────────────────────────── @@ -282,4 +316,4 @@ def __repr__(self): - return f"" - + return f"" + Index: tests/conftest.py =================================================================== --- tests/conftest.py (revision 6a7dec1b7c47ac7a8618ebec0981ce40dbc3224d) +++ tests/conftest.py (revision 02d288d33183db0124582a151433ccd6cf55cac7) @@ -5,43 +5,59 @@ from flexoentity import FlexoEntity, EntityType, EntityState, Domain +import pytest +import json +from flexoentity import EntityType, EntityState, Domain +from builder.questions import RadioQuestion, AnswerOption # adjust path if different +from builder.media_items import NullMediaItem # adjust import if needed -class DummyEntity(FlexoEntity): - """Minimal concrete subclass for testing FlexoEntity logic.""" - def __init__(self, domain, etype, state, seed="DUMMY"): - self._seed = seed - super().__init__(domain, etype, state) +@pytest.fixture(scope="session") +def domain(): + """Provide a reusable domain for all entity tests.""" + return Domain( + domain="SIG", + etype=EntityType.DOMAIN, + state=EntityState.DRAFT, + fullname="Signal Corps", + description="Questions related to communications and signaling systems.", + classification="RESTRICTED", + owner="test-suite" + ) - @property - def text_seed(self) -> str: - return self._seed - @classmethod - def from_dict(cls, data): - """Ensure enums and seed are reconstructed correctly.""" - domain = data["domain"] - etype = EntityType[data["etype"]] if isinstance(data["etype"], str) else data["etype"] - state = EntityState[data["state"]] if isinstance(data["state"], str) else data["state"] - seed = data.get("text_seed", "DUMMY-CONTENT") - return cls(domain=domain, etype=etype, state=state, seed=seed) +@pytest.fixture +def radio_question(domain): + """Return a simple RadioQuestion entity for testing FlexoEntity logic.""" + q = RadioQuestion( + domain=domain, + etype=EntityType.QUESTION, + state=EntityState.DRAFT, + text="Which frequency band is used for shortwave communication?", + options=[ + AnswerOption(id="opt1", text="HF (3–30 MHz)", points=1), + AnswerOption(id="opt2", text="VHF (30–300 MHz)", points=0), + AnswerOption(id="opt3", text="UHF (300–3000 MHz)", points=0), + ] + ) + return q - @classmethod - def from_json(cls, data_str: str): - return cls.from_dict(json.loads(data_str)) - + @pytest.fixture -def entity(): - """Generic FlexoEntity-like instance in draft state.""" - return DummyEntity( - domain=Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT, fullname="Signal Corps", classification="RESTRICTED"), - etype=EntityType.CATALOG, - state=EntityState.DRAFT, - ) +def serialized_question(radio_question): + """Provide the serialized JSON form for roundtrip tests.""" + return radio_question.to_json() + + +@pytest.fixture +def deserialized_question(serialized_question): + """Recreate a question from JSON for consistency tests.""" + return RadioQuestion.from_json(serialized_question) + @pytest.fixture def null_media(): - """Provide a default NullMediaItem instance for tests.""" + """Provide a default NullMediaItem instance for media tests.""" return NullMediaItem( - domain="GEN", + domain=domain, etype=EntityType.MEDIA, state=EntityState.DRAFT Index: tests/test_id_lifecycle.py =================================================================== --- tests/test_id_lifecycle.py (revision 6a7dec1b7c47ac7a8618ebec0981ce40dbc3224d) +++ tests/test_id_lifecycle.py (revision 02d288d33183db0124582a151433ccd6cf55cac7) @@ -1,89 +1,103 @@ import pytest - -from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState - -def test_initial_state(entity): - assert entity.state == EntityState.DRAFT - assert entity.flexo_id.version == 1 - assert len(entity.flexo_id.signature) == 16 # blake2s digest_size=8 → 16 hex - assert FlexoEntity.verify_integrity(entity) +from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState -def test_approval_bumps_version(entity): - entity.approve() - assert entity.state == EntityState.APPROVED - assert entity.flexo_id.version == 2 +# ────────────────────────────────────────────────────────────────────────────── +# Tests adapted to use real RadioQuestion fixture instead of DummyEntity +# ────────────────────────────────────────────────────────────────────────────── + +def test_initial_state(radio_question): + q = radio_question + assert q.state == EntityState.DRAFT + assert q.flexo_id.version == 1 + assert FlexoEntity.verify_integrity(q) -def test_signing_bumps_version(entity): - entity.approve() - v_before = entity.flexo_id - entity.sign() - assert entity.state == EntityState.APPROVED_AND_SIGNED - assert entity.flexo_id != v_before +def test_approval_bumps_version(radio_question): + q = radio_question + q.approve() + assert q.state == EntityState.APPROVED + assert q.flexo_id.version == 2 -def test_publish_bumps_version(entity): - entity.approve() - entity.sign() - v_before = entity.flexo_id.version - entity.publish() - assert entity.state == EntityState.PUBLISHED - assert entity.flexo_id.version == v_before + 1 +def test_signing_bumps_version(radio_question): + q = radio_question + q.approve() + v_before = str(q.flexo_id) + q.sign() + assert q.state == EntityState.APPROVED_AND_SIGNED + assert str(q.flexo_id) != v_before -def test_modify_content_changes_fingerprint(entity): - old_signature = entity.flexo_id.signature - entity._seed = "Rephrased content" # simulate text change - entity._update_fingerprint() - assert entity.flexo_id.signature != old_signature +def test_publish_bumps_version(radio_question): + q = radio_question + q.approve() + q.sign() + v_before = q.flexo_id.version + q.publish() + assert q.state == EntityState.PUBLISHED + assert q.flexo_id.version == v_before + 1 -def test_no_version_bump_on_draft_edits(entity): - entity._seed = "Draft edit only" - entity._update_fingerprint() - assert entity.flexo_id.version == 1 +def test_modify_content_changes_fingerprint(radio_question): + q = radio_question + q.text = "Rephrased content" # simulate text change + changed = q._update_fingerprint() + assert changed -def test_version_bump_after_edit_and_sign(entity): - entity.approve() - v1 = entity.flexo_id - entity._seed = "Changed content" - entity.sign() - assert entity.flexo_id != v1 +def test_no_version_bump_on_draft_edits(radio_question): + q = radio_question + q.text = "Minor draft edit" + q._update_fingerprint() + assert q.flexo_id.version == 1 -def test_integrity_check_passes_and_fails(entity): - entity.approve() - assert FlexoEntity.verify_integrity(entity) - # simulate tampering - entity._seed = "Tampered text" - assert not FlexoEntity.verify_integrity(entity) +def test_version_bump_after_edit_and_sign(radio_question): + q = radio_question + q.approve() + v1 = str(q.flexo_id) + q.text = "Changed content" + q.sign() + assert str(q.flexo_id) != v1 -def test_obsolete_state(entity): - entity.approve() - entity.sign() - entity.publish() - entity.obsolete() - assert entity.state == EntityState.OBSOLETE +def test_integrity_check_passes_and_fails(radio_question): + q = radio_question + q.approve() + assert FlexoEntity.verify_integrity(q) + + # simulate tampering + q.text = "Tampered text" + assert not FlexoEntity.verify_integrity(q) -def test_clone_new_base_resets_lineage(entity): - entity.approve() - entity.sign() - entity.publish() - entity.obsolete() - old_id = entity.flexo_id - entity.clone_new_base() - assert entity.flexo_id != old_id - assert entity.state == EntityState.DRAFT - assert entity.flexo_id.version == 1 +def test_obsolete_state(radio_question): + q = radio_question + q.approve() + q.sign() + q.publish() + q.obsolete() + assert q.state == EntityState.OBSOLETE -def test_mass_version_increments_until_obsolete(entity): - entity.approve() +def test_clone_new_base_resets_lineage(radio_question): + q = radio_question + q.approve() + q.sign() + q.publish() + q.obsolete() + old_id = str(q.flexo_id) + q.clone_new_base() + assert str(q.flexo_id) != old_id + assert q.state == EntityState.DRAFT + assert q.flexo_id.version == 1 + + +def test_mass_version_increments_until_obsolete(radio_question): + q = radio_question + q.approve() for _ in range(FlexOID.MAX_VERSION - 2): - entity.sign() + q.sign() with pytest.raises(RuntimeError, match="mark obsolete"): - entity.sign() + q.sign() Index: tests/test_id_stress.py =================================================================== --- tests/test_id_stress.py (revision 6a7dec1b7c47ac7a8618ebec0981ce40dbc3224d) +++ tests/test_id_stress.py (revision 02d288d33183db0124582a151433ccd6cf55cac7) @@ -3,87 +3,123 @@ 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,6 +128,4 @@ 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" Index: tests/test_persistance_integrity.py =================================================================== --- tests/test_persistance_integrity.py (revision 6a7dec1b7c47ac7a8618ebec0981ce40dbc3224d) +++ tests/test_persistance_integrity.py (revision 02d288d33183db0124582a151433ccd6cf55cac7) @@ -6,52 +6,58 @@ import pytest -from flexoentity import FlexOID, EntityType, EntityState -from tests.conftest import DummyEntity +from flexoentity import EntityState +from builder.questions import RadioQuestion, AnswerOption # ────────────────────────────────────────────────────────────────────────────── @pytest.fixture -def approved_entity(): - """A fully published dummy entity for persistence tests.""" - e = DummyEntity( - domain="AF", - etype=EntityType.QUESTION, +def approved_question(domain): + """Provide a fully approved and published RadioQuestion for persistence tests.""" + q = RadioQuestion( + domain=domain, + etype=None, # RadioQuestion sets this internally to EntityType.QUESTION state=EntityState.DRAFT, - seed="What is Ohm’s law?" + text="What is Ohm’s law?", + options=[ + AnswerOption(text="U = R × I", points=1), + AnswerOption(text="U = I / R", points=0), + AnswerOption(text="R = U × I", points=0), + ], ) - e.approve() - e.sign() - e.publish() - return e + q.approve() + q.sign() + q.publish() + return q -@pytest.mark.skip(reason="FlexOIDs are regenerated on import; enable once JSON format is stable") -def test_json_roundtrip_preserves_integrity(approved_entity): + +@pytest.mark.skip(reason="FlexOIDs regenerated on import; enable once JSON format is stable") +def test_json_roundtrip_preserves_integrity(approved_question): """ - Export to JSON and reload — ensure fingerprints remain valid. + Export to JSON and reload — ensure fingerprints and signatures remain valid. """ - json_str = approved_entity.to_json() - loaded = approved_entity.__class__.from_json(json_str) + json_str = approved_question.to_json() + loaded = RadioQuestion.from_json(json_str) # Fingerprint and state should match — integrity must pass - assert approved_entity.__class__.verify_integrity(loaded) + assert RadioQuestion.verify_integrity(loaded) # Metadata should be preserved exactly - assert approved_entity.flexo_id.signature == loaded.flexo_id.signature - assert approved_entity.flexo_id == loaded.flexo_id - assert loaded.state == approved_entity.state + assert approved_question.signature == loaded.signature + assert approved_question.flexo_id == loaded.flexo_id + assert loaded.state == approved_question.state + # ────────────────────────────────────────────────────────────────────────────── -@pytest.mark.skip(reason="FlexOIDs regenerated on import; tampering detection not applicable yet") -def test_json_tampering_detection(approved_entity): +@pytest.mark.skip(reason="FlexOIDs regenerated on import; tampering detection not yet implemented") +def test_json_tampering_detection(approved_question): """Tampering with content should invalidate fingerprint verification.""" - json_str = approved_entity.to_json() - tampered_data = json.loads(json_str) - tampered_data["text_seed"] = "Tampered content injection" - tampered_json = json.dumps(tampered_data) + json_str = approved_question.to_json() + tampered = json.loads(json_str) + tampered["text"] = "Tampered content injection" + tampered_json = json.dumps(tampered) - # We use DummyEntity.from_json to reconstruct (FlexoEntity is abstract) - loaded = approved_entity.__class__.from_json(tampered_json) - assert not approved_entity.__class__.verify_integrity(loaded) + loaded = RadioQuestion.from_json(tampered_json) + assert not RadioQuestion.verify_integrity(loaded) @@ -59,14 +65,14 @@ @pytest.mark.skip(reason="FlexOIDs regenerated on import; corruption detection not yet applicable") -def test_json_file_corruption(approved_entity, tmp_path): +def test_json_file_corruption(approved_question, tmp_path): """Simulate file corruption — integrity check must fail.""" - file = tmp_path / "entity.json" - json_str = approved_entity.to_json() + file = tmp_path / "question.json" + json_str = approved_question.to_json() file.write_text(json_str) - # Corrupt the file + # Corrupt the file (simulate accidental byte modification) corrupted = json_str.replace("Ohm’s", "Omm’s") file.write_text(corrupted) - loaded = approved_entity.__class__.from_json(file.read_text()) - assert not approved_entity.__class__.verify_integrity(loaded) + loaded = RadioQuestion.from_json(file.read_text()) + assert not RadioQuestion.verify_integrity(loaded)