Changeset ca39274 in flexoentity
- Timestamp:
- 10/28/25 12:16:31 (3 months ago)
- Branches:
- master
- Children:
- 8aa20c7
- Parents:
- 52ccac6
- Files:
-
- 4 edited
-
flexoentity/flexo_entity.py (modified) (8 diffs)
-
flexoentity/id_factory.py (modified) (8 diffs)
-
tests/test_id_lifecycle.py (modified) (1 diff)
-
tests/test_persistance_integrity.py (modified) (4 diffs)
Legend:
- Unmodified
- Added
- Removed
-
flexoentity/flexo_entity.py
r52ccac6 rca39274 7 7 from enum import Enum, auto 8 8 from dataclasses import dataclass, field 9 from datetime import datetime10 9 from typing import Optional 11 10 from abc import ABC, abstractmethod … … 29 28 CERTIFICATE = auto() 30 29 DOMAIN = auto() 31 30 32 31 def short(self) -> str: 33 32 mapping = { … … 97 96 fingerprint: str = field(default_factory=str) 98 97 origin: Optional[str] = field(default=None) 99 98 100 99 OID_PATTERN = re.compile( 101 100 r"^(?P<domain>[A-Z0-9]+)-(?P<etype>[A-Z]+)" 102 101 r"(?P<date>\d{6,8})-(?P<hash>[0-9A-F]+)@(?P<version>\d{3})(?P<state>[A-Z])$" 103 102 ) 104 105 def __str__(self) -> str:106 return f"{self.domain_code()}-{self.etype}{self.date}-{self.unique_hash}@{self.version:03d}{self.state}"107 103 108 104 @classmethod … … 145 141 class itself to avoid circular initialization. 146 142 """ 147 143 148 144 self.flexo_id = FlexOID.safe_generate(self.domain_code(), 149 145 self.etype.short(), … … 159 155 f"fingerprint={self.fingerprint}..., v{self.version})" 160 156 ) 157 161 158 def to_dict(self): 162 159 return { … … 168 165 "origin": self.origin, 169 166 } 170 167 171 168 @classmethod 172 169 def from_dict(cls, data): … … 175 172 abbrev, fullname = (lambda p: (p[0], p[1] if len(p) > 1 else ""))(domain.split("_", 1)) 176 173 domain_obj = Domain( 177 abbrev,174 domain=abbrev, 178 175 etype=EntityType.DOMAIN, 179 176 state=EntityState.DRAFT, # default when reconstructing context … … 365 362 canonical_seed(entity.text_seed).encode("utf-8"), digest_size=8 366 363 ).hexdigest().upper() 364 print(entity.fingerprint) 365 print(expected_fp) 367 366 return expected_fp == entity.fingerprint 368 367 -
flexoentity/id_factory.py
r52ccac6 rca39274 47 47 WARN_THRESHOLD = 900 48 48 49 49 50 # keep in-memory registry for same-session collisions (optional) 51 # NOTE: We might remove this soon 50 52 _seen_hashes = set() 51 53 … … 74 76 raise ValueError("Cannot order FlexOIDs from different prefixes") 75 77 return self.version < other.version 76 78 77 79 def __hash__(self): 78 80 return hash(self.flexo_id) … … 81 83 def _blake_hash(text: str) -> str: 82 84 """Return a 12-hex BLAKE2s digest.""" 83 return hashlib.blake2s(text.encode("utf-8"), digest_size=6).hexdigest().upper() # 6 bytes → 12 hex 85 return hashlib.blake2s(text.encode("utf-8"), 86 digest_size=6).hexdigest().upper() # 6 bytes → 12 hex 84 87 85 88 @staticmethod … … 134 137 135 138 return oid 136 139 137 140 @staticmethod 138 141 def generate(domain: str, etype: str, estate: str, text: str, … … 159 162 return FlexOID(flexo_id_str) 160 163 161 # ──────────────────────────────────────────────────────────────────────────162 163 164 @property 164 165 def state_code(self): … … 168 169 raise ValueError(f"Invalid Flex-O ID format: {self.flexo_id}") 169 170 return part[-1] 170 171 # ──────────────────────────────────────────────────────────────────────────172 # Parsed Accessors173 # ──────────────────────────────────────────────────────────────────────────174 171 175 172 @property … … 250 247 "state": self.state_code, 251 248 } 252 # ──────────────────────────────────────────────────────────────────────────253 249 254 250 @classmethod … … 290 286 return cls(new_id) 291 287 292 # ──────────────────────────────────────────────────────────────────────────293 288 @staticmethod 294 289 def clone_new_base(domain: str, etype: str, estate: str, text: str): -
tests/test_id_lifecycle.py
r52ccac6 rca39274 99 99 q.sign() 100 100 q.publish() 101 q.obsolete() 101 102 old_id = str(q.flexo_id) 102 q.obsolete()103 103 q.clone_new_base() 104 104 assert q.origin == old_id -
tests/test_persistance_integrity.py
r52ccac6 rca39274 6 6 import pytest 7 7 8 from flexoentity import EntityState9 8 from builder.questions import RadioQuestion, AnswerOption 9 from flexoentity import EntityState, EntityType, Domain 10 10 11 12 # ──────────────────────────────────────────────────────────────────────────────13 11 @pytest.fixture 14 def approved_question( domain):12 def approved_question(): 15 13 """Provide a fully approved and published RadioQuestion for persistence tests.""" 16 14 q = RadioQuestion( 17 domain= domain,15 domain=Domain(domain="GEN", etype=EntityType.DOMAIN, state=EntityState.DRAFT), 18 16 etype=None, # RadioQuestion sets this internally to EntityType.QUESTION 19 17 state=EntityState.DRAFT, 20 18 text="What is Ohm’s law?", 21 19 options=[ 22 AnswerOption( text="U = R × I", points=1),23 AnswerOption( text="U = I / R", points=0),24 AnswerOption( text="R = U × I", points=0),20 AnswerOption(id="OP1", text="U = R × I", points=1), 21 AnswerOption(id="OP2", text="U = I / R", points=0), 22 AnswerOption(id="OP3", text="R = U × I", points=0), 25 23 ], 26 24 ) … … 37 35 """ 38 36 json_str = approved_question.to_json() 37 print("JSON", json_str) 39 38 loaded = RadioQuestion.from_json(json_str) 40 39 40 print("Approved", approved_question.text_seed) 41 print("Loaded", loaded.text_seed) 41 42 # Fingerprint and state should match — integrity must pass 42 43 assert RadioQuestion.verify_integrity(loaded) … … 46 47 assert approved_question.flexo_id == loaded.flexo_id 47 48 assert loaded.state == approved_question.state 48 49 50 # ──────────────────────────────────────────────────────────────────────────────51 49 52 50 @pytest.mark.skip(reason="FlexOIDs regenerated on import; tampering detection not yet implemented") … … 60 58 loaded = RadioQuestion.from_json(tampered_json) 61 59 assert not RadioQuestion.verify_integrity(loaded) 62 63 64 # ──────────────────────────────────────────────────────────────────────────────65 60 66 61 @pytest.mark.skip(reason="FlexOIDs regenerated on import; corruption detection not yet applicable")
Note:
See TracChangeset
for help on using the changeset viewer.
