Changeset 4ceca57 in flexoentity
- Timestamp:
- 10/23/25 14:50:45 (3 months ago)
- Branches:
- master
- Children:
- 33be5a0
- Parents:
- 02d288d
- Files:
-
- 4 edited
-
flexoentity/flexo_entity.py (modified) (5 diffs)
-
flexoentity/id_factory.py (modified) (2 diffs)
-
tests/test_id_lifecycle.py (modified) (1 diff)
-
tests/test_id_stress.py (modified) (2 diffs)
Legend:
- Unmodified
- Added
- Removed
-
flexoentity/flexo_entity.py
r02d288d r4ceca57 146 146 """ 147 147 148 self.flexo_id = FlexOID. generate(self.domain_code(),148 self.flexo_id = FlexOID.safe_generate(self.domain_code(), 149 149 self.etype.short(), 150 150 self.state.short(), … … 221 221 if new_sig != self.signature: 222 222 self.signature = new_sig 223 self.flexo_id = FlexOID. generate(self.domain_code(),223 self.flexo_id = FlexOID.safe_generate(self.domain_code(), 224 224 self.etype.short(), 225 225 self.state.short(), … … 261 261 self.updated_at = datetime.now(UTC) 262 262 263 def lineage(self, repo): 264 """Return full ancestry chain [origin → ... → self].""" 265 chain = [self] 266 current = self 267 while current.origin and current.origin in repo: 268 current = repo[current.origin] 269 chain.insert(0, current) 270 return chain 271 263 272 def approve(self): 264 273 """ … … 268 277 if self.state == EntityState.DRAFT: 269 278 new_version = self.flexo_id.version + 1 270 new_fid = FlexOID. generate(self.domain_code(),279 new_fid = FlexOID.safe_generate(self.domain_code(), 271 280 self.etype.short(), 272 281 EntityState.APPROVED.short(), … … 294 303 def clone_new_base(self): 295 304 """Start new lineage when obsolete.""" 305 self.origin = str(self.flexo_id) 296 306 self.flexo_id = FlexOID.clone_new_base( 297 307 self.domain_code(), -
flexoentity/id_factory.py
r02d288d r4ceca57 80 80 @staticmethod 81 81 def _blake_hash(text: str) -> str: 82 """Return a 6-hex BLAKE2s digest."""82 """Return a 12-hex BLAKE2s digest.""" 83 83 return hashlib.blake2s(text.encode("utf-8"), digest_size=6).hexdigest().upper() # 6 bytes → 12 hex 84 84 … … 310 310 - Used when creating "clones" or "variants" that should not share version history. 311 311 """ 312 return FlexOID. generate(domain, etype, estate, text, version=1)312 return FlexOID.safe_generate(domain, etype, estate, text, version=1) 313 313 314 314 def __str__(self): -
tests/test_id_lifecycle.py
r02d288d r4ceca57 94 94 assert q.flexo_id.version == 1 95 95 96 def test_clone_new_base_sets_origin(radio_question): 97 q = radio_question 98 q.approve() 99 q.sign() 100 q.publish() 101 old_id = str(q.flexo_id) 102 q.obsolete() 103 q.clone_new_base() 104 assert q.origin == old_id 105 assert q.state == EntityState.DRAFT 106 assert q.flexo_id.version == 1 107 assert q.flexo_id != old_id 96 108 97 109 def test_mass_version_increments_until_obsolete(radio_question): -
tests/test_id_stress.py
r02d288d r4ceca57 14 14 def test_bulk_generation_uniqueness(domain): 15 15 """ 16 Generate 10 ,000 IDs and ensure uniqueness using safe_generate().16 Generate 100,000 IDs and ensure uniqueness using safe_generate(). 17 17 If a collision occurs, safe_generate() must resolve it automatically 18 18 via salt + date adjustment. … … 20 20 etype = EntityType.QUESTION 21 21 estate = EntityState.DRAFT 22 seeds = [f"question {i}" for i in range( 4000000)]22 seeds = [f"question {i}" for i in range(100000)] 23 23 24 24 # Simulate a simple in-memory repository for collision detection
Note:
See TracChangeset
for help on using the changeset viewer.
