Index: README.md =================================================================== --- README.md (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ README.md (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -78,7 +78,7 @@ *Highlights* -- =generate(domain, etype, estate, text, version=1)= → create a new ID +- =generate(domain, entity_type, estate, text, version=1)= → create a new ID - =next_version(oid)= → increment version safely -- =clone_new_base(domain, etype, estate, text)= → start a new lineage +- =clone_new_base(domain, entity_type, estate, text)= → start a new lineage - Deterministic prefix, state-dependent signature @@ -105,5 +105,5 @@ #+BEGIN_SRC python -entity = Question("AF", EntityType.QUESTION, "What is Ohm’s law?") +entity = Question("AF", EntityType.ITEM, "What is Ohm’s law?") json_str = entity.to_json() reloaded = Question.from_json(json_str) @@ -125,5 +125,5 @@ def __init__(self, domain, text): self._text = text - super().__init__(domain, EntityType.QUESTION) + super().__init__(domain, EntityType.ITEM) @property @@ -145,5 +145,5 @@ { "domain": "AF", - "etype": "QUESTION", + "entity_type": "QUESTION", "text_seed": "What is Ohm’s law?", "state": "APPROVED_AND_SIGNED", Index: flexoentity/domain.py =================================================================== --- flexoentity/domain.py (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ flexoentity/domain.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -1,4 +1,5 @@ from dataclasses import dataclass from flexoentity.flexo_entity import FlexoEntity, EntityType, EntityState + @dataclass @@ -11,5 +12,5 @@ @classmethod def default(cls): - return cls(domain="GEN", etype=EntityType.DATABASE, state=EntityState.DRAFT) + return cls(domain="GEN", entity_type=EntityType.DOMAIN, state=EntityState.DRAFT) def __post_init__(self): Index: flexoentity/flexo_entity.py =================================================================== --- flexoentity/flexo_entity.py (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ flexoentity/flexo_entity.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -5,5 +5,6 @@ import json import re -from enum import Enum, auto +from uuid import UUID +from enum import Enum from dataclasses import dataclass, field from typing import Optional @@ -21,21 +22,28 @@ class EntityType(Enum): - QUESTION = auto() - MEDIA = auto() - CATALOG = auto() - EXAM = auto() - DATABASE = auto() - CERTIFICATE = auto() - DOMAIN = auto() - + GENERIC = "G" + DOMAIN = "D" + MEDIA = "M" + ITEM = "I" + CATALOG = "C" + TEXT = "T" + HANDOUT = "H" + OUTPUT = "O" + RECORD = "R" + SESSION = "S" + USER = "U" + CONFIG = "F" + EVENT = "E" + + @classmethod + def from_letter(cls, a_letter): + return cls(a_letter) + + # FIXME: Add more mappings def short(self) -> str: mapping = { - EntityType.QUESTION: "Q", EntityType.MEDIA: "M", EntityType.DOMAIN: "DOM", EntityType.CATALOG: "CAT", - EntityType.EXAM: "EX", - EntityType.DATABASE: "DB", - EntityType.CERTIFICATE: "CERT" } return mapping[self] @@ -47,40 +55,9 @@ class EntityState(Enum): - DRAFT = auto() - APPROVED = auto() - APPROVED_AND_SIGNED = auto() - PUBLISHED = auto() - OBSOLETE = auto() - - def short(self) -> str: - """ - Return a one-letter abbreviation for the state, used in Flex-O IDs. - """ - mapping = { - EntityState.DRAFT: "D", - EntityState.APPROVED: "A", - EntityState.APPROVED_AND_SIGNED: "S", - EntityState.PUBLISHED: "P", - EntityState.OBSOLETE: "O", - - } - return mapping[self] - - @classmethod - def from_short(cls, char: str): - """ - Inverse of .short(): restore the EntityState from its one-letter code. - """ - reverse = { - "D": cls.DRAFT, - "A": cls.APPROVED, - "S": cls.APPROVED_AND_SIGNED, - "P": cls.PUBLISHED, - "O": cls.OBSOLETE, - } - try: - return reverse[char.upper()] - except KeyError: - raise ValueError(f"Unknown state abbreviation: {char}") + DRAFT = "D" + APPROVED = "A" + APPROVED_AND_SIGNED = "S" + PUBLISHED = "P" + OBSOLETE = "O" def __str__(self): @@ -91,27 +68,13 @@ class FlexoEntity(ABC): domain: str - etype: EntityType + entity_type: EntityType + subtype: str = "GENERIC" state: EntityState flexo_id: Optional[FlexOID] = field(default=None) fingerprint: str = field(default_factory=str) + originator_id: UUID = field(default=UUID(int=0)) + owner_id: UUID = field(default=UUID(int=0)) origin: Optional[str] = field(default=None) - OID_PATTERN = re.compile( - r"^(?P[A-Z0-9]+)-(?P[A-Z]+)" - r"(?P\d{6,8})-(?P[0-9A-F]+)@(?P\d{3})(?P[A-Z])$" - ) - - @classmethod - def from_string(cls, s: str) -> "FlexOID": - """Rehydrate a FlexOID from its canonical string form.""" - m = cls.OID_PATTERN.match(s.strip()) - if not m: - raise ValueError(f"Invalid FlexOID string: {s}") - gd = m.groupdict() - return cls( - domain=gd["domain"], - etype=gd["etype"], - state=gd["state"], - ) @property @abstractmethod @@ -143,6 +106,6 @@ self.flexo_id = FlexOID.safe_generate(self.domain_code(), - self.etype.short(), - self.state.short(), + self.entity_type.value, + self.state.value, self.text_seed, 1) @@ -152,5 +115,5 @@ def __str__(self): return ( - f"{self.etype.name}({self.flexo_id}, {self.state.name}, " + f"{self.entity_type.name}({self.flexo_id}, {self.state.name}, " f"fingerprint={self.fingerprint}..., v{self.version})" ) @@ -159,5 +122,5 @@ return { "domain": self.domain_code(), - "etype": self.etype.name, + "entity_type": self.entity_type.name, "state": self.state.name, "flexo_id": str(self.flexo_id), @@ -173,10 +136,10 @@ domain_obj = Domain( domain=abbrev, - etype=EntityType.DOMAIN, + entity_type=EntityType.DOMAIN, state=EntityState.DRAFT, # default when reconstructing context ) obj = cls( domain=domain_obj, - etype=EntityType[data["etype"]], + entity_type=EntityType[data["entity_type"]], state=EntityState[data["state"]], ) @@ -221,6 +184,6 @@ self.fingerprint = new_fp self.flexo_id = FlexOID.safe_generate(self.domain_code(), - self.etype.short(), - self.state.short(), + self.entity_type.value, + self.state.value, self.text_seed, self.flexo_id.version) @@ -244,5 +207,5 @@ # special case: marking obsolete if target_state == EntityState.OBSOLETE: - self.flexo_id = FlexOID(self.flexo_id.flexo_id[:-1] + "O") + self.flexo_id = FlexOID.with_state(self.flexo_id, "O") self.state = target_state return @@ -287,12 +250,11 @@ if self.state == EntityState.DRAFT: - new_version = self.flexo_id.version + 1 new_fid = FlexOID.safe_generate(self.domain_code(), - self.etype.short(), - EntityState.APPROVED.short(), + self.entity_type.value, + EntityState.APPROVED.value, self.text_seed, - version=new_version + version=self.version ) - self.previous_id = self.flexo_id # optional: keep audit trail + self.origin = self.flexo_id # optional: keep audit trail self.flexo_id = new_fid self.state = EntityState.APPROVED @@ -320,11 +282,11 @@ new_fid = FlexOID.safe_generate( self.domain_code(), - self.etype.short(), - EntityState.PUBLISHED.short(), + self.entity_type.value, + EntityState.PUBLISHED.value, self.text_seed, version=new_version ) - self.previous_id = self.flexo_id + self.origin = self.flexo_id self.flexo_id = new_fid self.state = EntityState.PUBLISHED @@ -346,6 +308,6 @@ self.flexo_id = FlexOID.clone_new_base( self.domain_code(), - self.etype.short(), - self.state.short(), + self.entity_type.value, + self.state.value, self.text_seed, ) @@ -362,6 +324,4 @@ canonical_seed(entity.text_seed).encode("utf-8"), digest_size=8 ).hexdigest().upper() - print(entity.fingerprint) - print(expected_fp) return expected_fp == entity.fingerprint Index: flexoentity/id_factory.py =================================================================== --- flexoentity/id_factory.py (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ flexoentity/id_factory.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -1,32 +1,97 @@ """ -id_factory.py — Central Flex-O ID generator and versioning control (hardened). - -Improvements: -- BLAKE2s hashing (modern, fast, stdlib) -- 6 hex-digit hash (≈16.7M combinations) -- UTC-based dates for consistency -- Collision disambiguator (-A, -B, ...) -- Canonical seed and content fingerprint helpers +id_factory.py — Central Flex-O ID generation and lifecycle logic. + +I represent the identity backbone of the Flex-O ecosystem. +I define how entities receive, preserve, and evolve their unique identifiers +throughout their lifecycle. + +Each Flex-O ID (FlexOID) is a self-contained string that encodes: + + --@ + +Example: + GEN-ITEM251101-3139ACFAE38B@002A + +where + DOMAIN — a short prefix identifying the origin domain (e.g., GEN) + ETYPE — a compact entity type code (e.g., ITEM) + YYMMDD — the UTC creation date + HASH — a 12-hex BLAKE2s digest derived from canonical content + VERSION — a three-digit lineage counter (001-999) + STATE — a single capital letter indicating lifecycle state + +─────────────────────────────────────────────────────────────────────────────── +Lifecycle semantics +─────────────────────────────────────────────────────────────────────────────── + +I distinguish clearly between *content evolution* and *workflow state*: + +• Draft → Approved: + I generate a fully new FlexOID whose hash and version reflect + the new, stable content. This step marks the transition from + provisional to permanent identity. No extra version bump is required. + +• Approved → Signed → Published → Obsolete: + These transitions do not change the content hash. + They represent confidence or visibility, not content. + The same FlexOID (same version) remains valid across them. + +In short: + Version changes record evolution of content. + State changes record evolution of trust. + +─────────────────────────────────────────────────────────────────────────────── +Implementation notes +─────────────────────────────────────────────────────────────────────────────── + +• Hashing: I use modern BLAKE2s (6-byte digest) for deterministic, + collision-resistant identifiers. + +• Canonicalization: I flatten entity data into a reproducible text form + so equivalent content always yields the same hash. + +• Generation: I provide safe and deterministic factories + (FlexOID.generate, FlexOID.safe_generate) + that handle rare hash collisions by minimal salting. + +• Validation: Every FlexOID validates itself through a strict regex pattern + and exposes lightweight accessors (domain, type, date, hash, + version, state). + +─────────────────────────────────────────────────────────────────────────────── +Philosophy +─────────────────────────────────────────────────────────────────────────────── + +I strive to be deterministic, minimal, and auditable. +My design favors immutability and clarity over flexibility. +A FlexOID never lies about its lineage: once created, it is final. + +Version marks evolution of content. +State marks evolution of confidence. """ -import logging -from datetime import datetime, timezone +import re import hashlib import secrets -import itertools import json - -logger = logging.getLogger(__name__) - -# ────────────────────────────────────────────────────────────────────────────── -# Canonicalization helpers -# ────────────────────────────────────────────────────────────────────────────── - +from datetime import datetime, timezone +from logging import Logger + +logger = Logger(__file__) def canonical_seed(obj) -> str: """ - Deterministically flatten an entity's core data into a string - for hashing and deduplication. + I transform *obj* into a deterministic, comparable text form. + + I remove irrelevant formatting differences so that two equal + pieces of data always yield the same hash seed. + + Rules: + - If *obj* is a string, I normalize whitespace. + - If *obj* is a dict, I JSON-encode it with sorted keys. + - If *obj* is an object, I recurse on its __dict__. + - Otherwise, I coerce *obj* to str(). """ + if isinstance(obj, str): text = " ".join(obj.split()) @@ -38,208 +103,206 @@ return str(obj) -# ────────────────────────────────────────────────────────────────────────────── -# ID Factory -# ────────────────────────────────────────────────────────────────────────────── - - -class FlexOID: + +class FlexOID(str): + """ + I represent a canonical textual identifier within the Flex-O ecosystem. +i I am immutable and behave like a normal string, but I understand + my own internal structure. I can reveal my parts, create new + versions, and switch between lifecycle states. + """ + MAX_VERSION = 999 WARN_THRESHOLD = 900 - - # keep in-memory registry for same-session collisions (optional) - # NOTE: We might remove this soon - _seen_hashes = set() + OID_PATTERN = re.compile( + r"^(?P[A-Z0-9]+)-" + r"(?P[A-Z0-9]+)" + r"(?P\d{6})-" + r"(?P[A-F0-9]+)@" + r"(?P\d{3})" + r"(?P[A-Z])$" + ) + + def __new__(cls, value: str): + """ + I create a new validated Flex-O ID from *value*. + I verify that the given string matches the required pattern + and remember the parsed match for later access. + """ + m = cls.OID_PATTERN.match(value) + if not m: + raise ValueError(f"Invalid FlexOID format: {value}") + obj = super().__new__(cls, value) + obj._m = m + return obj + + # ─────────────────────────────────────────── + # Parsed accessors + # ─────────────────────────────────────────── + @property + def domain(self) -> str: + """I answer the domain prefix (e.g., 'GEN').""" + return self._m.group("domain") + + @property + def entity_type(self) -> str: + """I answer the short entity-type code (e.g., 'ITEM').""" + return self._m.group("etype") + + @property + def date_str(self) -> str: + """I answer the YYMMDD creation date as a string.""" + return self._m.group("date") + + @property + def date(self) -> date: + """I answer the creation date as a `datetime.date` instance.""" + return datetime.strptime(self.date_str, "%y%m%d").date() + + @property + def hash_part(self) -> str: + """I answer the twelve-hex BLAKE2s digest that stabilizes my prefix.""" + return self._m.group("hash") + + @property + def version(self) -> int: + """I answer my numeric version (1–999).""" + return int(self._m.group("version")) + + @property + def state_code(self) -> str: + """I answer my one-letter lifecycle state code (e.g., 'D', 'A', 'S').""" + return self._m.group("state") + + @property + def prefix(self) -> str: + """I answer everything before the '@' symbol — my immutable lineage prefix.""" + return self.split('@', 1)[0] + + # ─────────────────────────────────────────── + # Transformations + # ─────────────────────────────────────────── + def with_state(self, new_state: str) -> "FlexOID": + """ + I create a copy of myself with the same version but a new state letter. + The given *new_state* must be a single capital letter. + """ + if not (isinstance(new_state, str) and len(new_state) == 1 and new_state.isalpha()): + raise ValueError("State must be a single capital letter.") + return FlexOID(f"{self.prefix}@{self.version:03d}{new_state}") + + def __repr__(self): + """I display myself in a developer-friendly representation.""" + return f"FlexOID({str(self)})" + + # ─────────────────────────────────────────── + # Generation helpers + # ─────────────────────────────────────────── + @staticmethod + def _blake_hash(text: str) -> str: + """ + I compute a 12-hex-digit BLAKE2s digest for *text*. + + I am modern, fast, and collision-resistant. + My output is deterministic and stable across platforms. + """ + + return hashlib.blake2s(text.encode("utf-8"), digest_size=6).hexdigest().upper() + + @staticmethod + def generate(domain: str, entity_type: str, estate: str, text: str, version: int = 1): + """ + I create a new deterministic Flex-O ID. + + I combine the domain, entity type, and canonicalized *text* + into a stable BLAKE2s hash. My prefix therefore remains + unchanged when only the state or version changes. + """ + if not (1 <= version <= FlexOID.MAX_VERSION): + raise ValueError(f"Version {version} exceeds limit; mark obsolete.") + + if not (isinstance(estate, str) and len(estate) == 1 and estate.isalpha()): + raise ValueError("estate must be a single capital letter.") + + date_part = datetime.now(timezone.utc).strftime("%y%m%d") + hash_seed = canonical_seed(f"{domain}:{entity_type}:{canonical_seed(text)}") + base_hash = FlexOID._blake_hash(hash_seed) + return FlexOID(f"{domain}-{entity_type}{date_part}-{base_hash}@{version:03d}{estate}") + + @staticmethod + def safe_generate(domain, entity_type, estate, text, version=1, repo=None): + """ + I create a new deterministic ID like `generate`, + but I also consult an optional *repo* to avoid hash collisions. + + If a different seed has already produced the same prefix, + I deterministically salt my seed and regenerate a unique ID. + """ + domain_code = getattr(domain, "domain", domain) + oid = FlexOID.generate(domain_code, entity_type, estate, text, version=version) + + if repo is None: + return oid + + existing = repo.get(str(oid)) if hasattr(repo, "get") else repo.get(oid) + if not existing: + return oid + + try: + same_seed = ( + getattr(existing, "text_seed", None) == text + or getattr(existing, "canonical_seed", lambda: None)() == canonical_seed(text) + ) + except Exception: + same_seed = False + + if same_seed: + return oid + + logger.warning(f"FlexOID collision detected for {oid}") + salt = secrets.token_hex(1) + salted_text = f"{text}|salt:{salt}" + return FlexOID.generate(domain_code, entity_type, estate, salted_text, version=version) @classmethod - def from_string(cls, id_str: str): - return cls(id_str) + def next_version(cls, oid: "FlexOID") -> "FlexOID": + """ + I create the next version within the same lineage. + I increment the numeric version but keep the same prefix and state. + """ + new_ver = oid.version + 1 + if new_ver > cls.WARN_THRESHOLD and new_ver < cls.MAX_VERSION: + logger.warning(f"{oid} approaching obsolescence ({new_ver}/999).") + if new_ver > cls.MAX_VERSION: + raise RuntimeError(f"{oid} exceeded {cls.MAX_VERSION}; mark obsolete.") + return FlexOID(f"{oid.prefix}@{new_ver:03d}{oid.state_code}") @classmethod - def from_oid_and_version(cls, oid, version: int): + def from_oid_and_version(cls, oid, version): + """ + I recreate *oid* with an explicitly given version number. + I keep its prefix and state, but replace the numeric counter. + """ 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}") - def __init__(self, flexo_id: str): - self.flexo_id = flexo_id - - def __eq__(self, other): - if not isinstance(other, FlexOID): - return NotImplemented - 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) - @staticmethod - def _blake_hash(text: str) -> str: - """Return a 12-hex BLAKE2s digest.""" - return hashlib.blake2s(text.encode("utf-8"), - digest_size=6).hexdigest().upper() # 6 bytes → 12 hex - - @staticmethod - 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): - """ - Generate a deterministic Flex-O ID. - - - The hash (and therefore prefix) depends only on domain, etype, and text. - → Prefix stays stable across state changes. - """ - - if not (1 <= version <= FlexOID.MAX_VERSION): - raise ValueError(f"Version {version} exceeds limit; mark obsolete.") - - date_part = datetime.now(timezone.utc).strftime("%y%m%d") - - # state-independent hash seed → prefix stability - hash_seed = canonical_seed(f"{domain}:{etype}:{text}") - base_hash = FlexOID._blake_hash(hash_seed) - - ver_part = f"{version:03d}{estate}" - flexo_id_str = f"{domain}-{etype}{date_part}-{base_hash}@{ver_part}" - - return FlexOID(flexo_id_str) - - @property - def state_code(self): - part = self.flexo_id.rsplit("@", 1)[-1] - - if not (part and part[-1].isalpha()): - raise ValueError(f"Invalid Flex-O ID format: {self.flexo_id}") - return part[-1] - - @property - def domain(self) -> str: - """Return the domain prefix (e.g., 'AF').""" - try: - return self.flexo_id.split('-', 1)[0] - except IndexError: - raise ValueError(f"Malformed Flex-O ID: {self.flexo_id}") - - @property - def etype(self) -> str: - """Return the entity type code (e.g., 'Q', 'CAT', etc.).""" - try: - part = self.flexo_id.split('-', 1)[1] - return ''.join(filter(str.isalpha, part.split('-')[0])) # up to first dash - except IndexError: - raise ValueError(f"Malformed Flex-O ID: {self.flexo_id}") - - @property - def date_str(self) -> str: - """Return the YYMMDD creation date as string.""" - try: - part = self.flexo_id.split('-', 1)[1] - # e.g. "Q251019" → skip type prefix, take next 6 digits - digits = ''.join(ch for ch in part if ch.isdigit()) - return digits[:6] - except IndexError: - raise ValueError(f"Malformed Flex-O ID: {self.flexo_id}") - - @property - def date(self) -> datetime: - """Return the creation date as datetime.date object (UTC, naive).""" - try: - ds = self.date_str - return datetime.strptime(ds, "%y%m%d").date() - except Exception as e: - raise ValueError(f"Invalid date in Flex-O ID: {self.flexo_id}") from e - - @property - def hash_part(self) -> str: - """Return the 6-hex BLAKE hash portion (e.g., '9B3E2').""" - try: - after_dash = self.flexo_id.split('-', 2)[2] - return after_dash.split('@')[0] - except IndexError: - raise ValueError(f"Malformed Flex-O ID: {self.flexo_id}") - - @property - def suffix(self) -> str: - """Return the full suffix after '@' (e.g., '001A').""" - try: - return self.flexo_id.split('@', 1)[1] - except IndexError: - raise ValueError(f"Malformed Flex-O ID: {self.flexo_id}") - - @property - def version(self) -> int: - try: - return int(self.suffix[:-1]) # drop state suffix - except (ValueError, IndexError): - return 1 - - @property - def prefix(self) -> str: - # nur bis einschließlich Hash-Teil - return self.flexo_id.split('@', 1)[0] - # return self.flexo_id.split('@')[0].rsplit('-', 1)[0] + def clone_new_base(domain: str, entity_type: str, estate: str, text: str) -> "FlexOID": + """ + I start a completely new lineage (version 1) for a derived entity. + I am used when copying or forking an existing object that should + not share version history with its origin. + """ + return FlexOID.safe_generate(domain, entity_type, estate, text, version=1) def parsed(self) -> dict: - """Return a structured breakdown of the Flex-O ID.""" + """ + I answer a dictionary that describes all my components — + useful for debugging, serialization tests, or human inspection. + """ return { "domain": self.domain, - "etype": self.etype, + "entity_type": self.entity_type, "date": self.date, "hash": self.hash_part, @@ -247,68 +310,2 @@ "state": self.state_code, } - - @classmethod - def next_version(cls, oid) -> str: - """ - Create the next version in the same ID lineage. - - Increments the version counter of an existing FlexOID while preserving - its prefix. Used when an entity transitions to a - new revision within the same lifecycle (e.g., minor updates or approvals). - - Parameters - ---------- - oid : FlexOID - The existing ID whose version is to be incremented. - - Returns - ------- - FlexOID - A new Flex-O ID with the same prefix, but version +1. - - Raises - ------ - RuntimeError - If the maximum allowed version (`MAX_VERSION`) is exceeded. - - Notes - ----- - - Warnings are logged when the version approaches obsolescence. - """ - new_ver = oid.version + 1 - - if new_ver > cls.WARN_THRESHOLD and new_ver < cls.MAX_VERSION: - logger.warning(f"{oid} approaching obsolescence ({new_ver}/999).") - if new_ver > cls.MAX_VERSION: - raise RuntimeError(f"{oid} exceeded {cls.MAX_VERSION}; mark obsolete.") - - new_id = f"{oid.prefix}@{new_ver:03d}{oid.state_code}" - return cls(new_id) - - @staticmethod - def clone_new_base(domain: str, etype: str, estate: str, text: str): - """ - Start a new Flex-O ID lineage for a derived or duplicated entity. - - This helper creates a completely new base ID (version 1) using the given - parameters, instead of incrementing an existing version chain. It is used - when an entity is copied, forked, or conceptually replaced by a new one. - - Returns - ------- - FlexOID - A new base ID starting at version 1, unrelated to the original lineage. - - Notes - ----- - - Equivalent to calling `generate(..., version=1)` explicitly. - - Used when creating "clones" or "variants" that should not share version history. - """ - return FlexOID.safe_generate(domain, etype, estate, text, version=1) - - def __str__(self): - return self.flexo_id - - def __repr__(self): - return f"" - Index: tests/conftest.py =================================================================== --- tests/conftest.py (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ tests/conftest.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -1,64 +1,93 @@ -# tests/conftest.py - +# tests/stubs/single_choice_question.py import pytest -import json +from datetime import datetime +from dataclasses import dataclass, field +from typing import List 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 +@pytest.fixture +def fixed_datetime(monkeypatch): + class FixedDate(datetime): + @classmethod + def now(cls, tz=None): + return datetime(2025, 11, 1, tzinfo=tz) + monkeypatch.setattr("flexoentity.id_factory.datetime", FixedDate) + return FixedDate -@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" - ) +@dataclass +class AnswerOption: + id: str + text: str + points: float = 0.0 + + def to_dict(self): + return {"id": self.id, "text": self.text, "points": self.points} + + @classmethod + def from_dict(cls, data): + return cls( + id=data.get("id", ""), + text=data.get("text", ""), + points=data.get("points", 0.0) + ) -@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 +@dataclass +class SingleChoiceQuestion(FlexoEntity): + """A minimal stub to test FlexoEntity integration.""" + text: str = "" + options: List[AnswerOption] = field(default_factory=list) -@pytest.fixture -def serialized_question(radio_question): - """Provide the serialized JSON form for roundtrip tests.""" - return radio_question.to_json() + @classmethod + def default(cls): + return cls(domain=Domain(domain="GEN", + entity_type=EntityType.DOMAIN, + state=EntityState.DRAFT), + state=EntityState.DRAFT, entity_type=EntityType.ITEM) + def to_dict(self): + base = super().to_dict() + base.update({ + "text": self.text, + "options": [opt.to_dict() for opt in self.options], + }) + return base + + @property + def text_seed(self) -> str: + """Include answer options (and points) for deterministic ID generation.""" + + joined = "|".join( + f"{opt.text.strip()}:{opt.points}" + for opt in sorted(self.options, key=lambda o: o.text.strip().lower()) + ) + return f"{self.text}{joined}" + + @classmethod + def from_dict(cls, data): + obj = cls( + text=data.get("text", ""), + options=[AnswerOption.from_dict(o) for o in data.get("options", [])], + ) + # restore FlexoEntity core fields + obj.domain = data.get("domain") + obj.entity_type = EntityType[data.get("etype")] if "etype" in data else EntityType.ITEM + obj.state = EntityState[data.get("state")] if "state" in data else EntityState.DRAFT + if "flexo_id" in data: + from flexoentity import FlexOID + obj.flexo_id = FlexOID.parsed(data["flexo_id"]) + return obj @pytest.fixture -def deserialized_question(serialized_question): - """Recreate a question from JSON for consistency tests.""" - return RadioQuestion.from_json(serialized_question) - +def domain(): + return Domain.default() @pytest.fixture -def null_media(): - """Provide a default NullMediaItem instance for media tests.""" - return NullMediaItem( - domain=domain, - etype=EntityType.MEDIA, - state=EntityState.DRAFT - ) +def sample_question(): + return SingleChoiceQuestion(domain=Domain.default(), + text="What is 2 + 2?", + options=[], + entity_type=EntityType.ITEM, + state=EntityState.DRAFT) Index: tests/test_flexoid.py =================================================================== --- tests/test_flexoid.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) +++ tests/test_flexoid.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 + +# tests/test_flexoid.py +""" +Test suite for id_factory.FlexOID. + +I verify that FlexOID behaves deterministically, validates itself strictly, +and evolves correctly through version and state transitions. +""" + +import re +from datetime import datetime, date +import pytest +from logging import Logger +from flexoentity import FlexOID, canonical_seed + + +logger = Logger(__file__) + +# ────────────────────────────────────────────── +# Basic construction and validation +# ────────────────────────────────────────────── + +def test_valid_flexoid_parsing(): + fid = FlexOID("GEN-I251101-ABCDEF123456@001D") + assert isinstance(fid, str) + assert fid.domain == "GEN" + assert fid.entity_type == "I" + assert fid.date_str == "251101" + assert fid.version == 1 + assert fid.state_code == "D" + assert fid.prefix.endswith("ABCDEF123456") + assert str(fid).endswith("@001D") + + +def test_invalid_flexoid_raises(): + with pytest.raises(ValueError): + FlexOID("INVALIDFORMAT") + with pytest.raises(ValueError): + FlexOID("GEN-I251101-ABCDEF@1D") # bad version width + + +def test_regex_is_strict(): + pat = FlexOID.OID_PATTERN + assert re.match(pat, "GEN-I251101-123456ABCDEF@001A") + assert not re.match(pat, "gen-item251101-123456@001A") # lowercase not allowed + assert not re.match(pat, "GEN-I251101-123456@01A") # wrong digit count + + +# ────────────────────────────────────────────── +# Generation and deterministic hashing +# ────────────────────────────────────────────── + +def test_generate_and_hash_stability(fixed_datetime): + # Fix the date so test is stable + fid1 = FlexOID.generate("GEN", "I", "D", "test content") + fid2 = FlexOID.generate("GEN", "I", "D", "test content") + assert fid1 == fid2 # deterministic + assert fid1.hash_part == fid2.hash_part + assert fid1.domain == "GEN" + assert fid1.entity_type == "I" + + +def test_safe_generate_collision(monkeypatch): + # Fake repo that always returns a conflicting item with different seed + class DummyRepo(dict): + def get(self, key): return True + + repo = DummyRepo() + fid = FlexOID.safe_generate("GEN", "I", "D", "abc", repo=repo) + assert isinstance(fid, FlexOID) + assert fid.state_code == "D" + + +# ────────────────────────────────────────────── +# State and version transitions +# ────────────────────────────────────────────── + +def test_with_state_creates_new_instance(): + fid = FlexOID("GEN-I251101-ABCDEF123456@001D") + fid2 = fid.with_state("A") + assert fid != fid2 + assert fid.version == fid2.version + assert fid2.state_code == "A" + assert str(fid2).endswith("@001A") + + +def test_next_version_increments_properly(): + fid = FlexOID("GEN-I251101-ABCDEF123456@001A") + fid2 = FlexOID.next_version(fid) + assert fid2.version == 2 + assert fid2.state_code == fid.state_code + assert fid.prefix == fid2.prefix + + +def test_from_oid_and_version_sets_exact_version(): + fid = FlexOID("GEN-I251101-ABCDEF123456@001A") + fid2 = FlexOID.from_oid_and_version(fid, 10) + assert fid2.version == 10 + assert fid2.state_code == "A" + + +def test_clone_new_base_starts_at_one(fixed_datetime): + fid = FlexOID.clone_new_base("GEN", "I", "D", "clone text") + assert fid.version == 1 + assert fid.state_code == "D" + + +# ────────────────────────────────────────────── +# Canonical seed behavior +# ────────────────────────────────────────────── + +def test_canonical_seed_for_strings_and_dicts(): + s1 = " Hello world " + s2 = "Hello world" + assert canonical_seed(s1) == canonical_seed(s2) + + d1 = {"b": 1, "a": 2} + d2 = {"a": 2, "b": 1} + assert canonical_seed(d1) == canonical_seed(d2) + + class Dummy: + def __init__(self): + self.x = 1 + self.y = 2 + obj = Dummy() + assert isinstance(canonical_seed(obj), str) + + +# ────────────────────────────────────────────── +# Parsed structure and representations +# ────────────────────────────────────────────── + +def test_parsed_returns_expected_dict(): + fid = FlexOID("GEN-I251101-ABCDEF123456@007S") + data = fid.parsed() + assert data["domain"] == "GEN" + assert data["entity_type"] == "I" + assert data["version"] == 7 + assert data["state"] == "S" + assert isinstance(data["date"], date) + + +def test_repr_is_readable(): + fid = FlexOID("GEN-I251101-ABCDEF123456@001D") + s = repr(fid) + assert "FlexOID" in s + assert "GEN-I" in s Index: tests/test_id_lifecycle.py =================================================================== --- tests/test_id_lifecycle.py (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ tests/test_id_lifecycle.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -1,12 +1,12 @@ import pytest -from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState +from flexoentity import FlexOID, FlexoEntity, EntityState # ────────────────────────────────────────────────────────────────────────────── -# Tests adapted to use real RadioQuestion fixture instead of DummyEntity +# Tests adapted to use real SingleChoiceQuestion fixture instead of DummyEntity # ────────────────────────────────────────────────────────────────────────────── -def test_initial_state(radio_question): - q = radio_question +def test_initial_state(sample_question): + q = sample_question assert q.state == EntityState.DRAFT assert q.flexo_id.version == 1 @@ -14,13 +14,13 @@ -def test_approval_bumps_version(radio_question): - q = radio_question +def test_approval_does_not_bump_version(sample_question): + q = sample_question q.approve() assert q.state == EntityState.APPROVED - assert q.flexo_id.version == 2 + assert q.flexo_id.version == 1 -def test_signing_bumps_version(radio_question): - q = radio_question +def test_signing_bumps_version(sample_question): + q = sample_question q.approve() v_before = str(q.flexo_id) @@ -30,6 +30,6 @@ -def test_publish_bumps_version(radio_question): - q = radio_question +def test_publish_bumps_version(sample_question): + q = sample_question q.approve() q.sign() @@ -40,13 +40,13 @@ -def test_modify_content_changes_fingerprint(radio_question): - q = radio_question - q.text = "Rephrased content" # simulate text change +def test_modify_content_changes_fingerprint(sample_question): + q = sample_question + q.text += "Rephrased content" # simulate text change changed = q._update_fingerprint() assert changed -def test_no_version_bump_on_draft_edits(radio_question): - q = radio_question +def test_no_version_bump_on_draft_edits(sample_question): + q = sample_question q.text = "Minor draft edit" q._update_fingerprint() @@ -54,6 +54,6 @@ -def test_version_bump_after_edit_and_sign(radio_question): - q = radio_question +def test_version_bump_after_edit_and_sign(sample_question): + q = sample_question q.approve() v1 = str(q.flexo_id) @@ -63,6 +63,6 @@ -def test_integrity_check_passes_and_fails(radio_question): - q = radio_question +def test_integrity_check_passes_and_fails(sample_question): + q = sample_question q.approve() assert FlexoEntity.verify_integrity(q) @@ -73,6 +73,6 @@ -def test_obsolete_state(radio_question): - q = radio_question +def test_obsolete_state(sample_question): + q = sample_question q.approve() q.sign() @@ -82,6 +82,6 @@ -def test_clone_new_base_resets_lineage(radio_question): - q = radio_question +def test_clone_new_base_resets_lineage(sample_question): + q = sample_question q.approve() q.sign() @@ -94,6 +94,6 @@ assert q.flexo_id.version == 1 -def test_clone_new_base_sets_origin(radio_question): - q = radio_question +def test_clone_new_base_sets_origin(sample_question): + q = sample_question q.approve() q.sign() @@ -107,8 +107,8 @@ assert q.flexo_id != old_id -def test_mass_version_increments_until_obsolete(radio_question): - q = radio_question +def test_mass_version_increments_until_obsolete(sample_question): + q = sample_question q.approve() - for _ in range(FlexOID.MAX_VERSION - 2): + for _ in range(FlexOID.MAX_VERSION - 1): q.bump_version() with pytest.raises(RuntimeError, match="mark obsolete"): Index: tests/test_id_stress.py =================================================================== --- tests/test_id_stress.py (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ tests/test_id_stress.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -4,9 +4,11 @@ """ +import copy +import logging +import random + import pytest -import random -import logging + from flexoentity import FlexOID, EntityType, EntityState -from builder.questions import RadioQuestion, AnswerOption logger = logging.getLogger(__name__) @@ -18,5 +20,5 @@ via salt + date adjustment. """ - etype = EntityType.QUESTION + entity_type = EntityType.ITEM estate = EntityState.DRAFT seeds = [f"question {i}" for i in range(100000)] @@ -31,5 +33,5 @@ ids = [] for seed in seeds: - oid = FlexOID.safe_generate(domain.domain, etype, estate, seed, repo=repo) + oid = FlexOID.safe_generate(domain.domain, entity_type.value, estate.value, seed, repo=repo) assert isinstance(oid, FlexOID) ids.append(str(oid)) @@ -46,5 +48,5 @@ # Sanity check: IDs should look canonical - assert all(id_str.startswith("SIG-") for id_str in ids) + assert all(id_str.startswith("GEN") for id_str in ids) assert all("@" in id_str for id_str in ids) @@ -54,33 +56,33 @@ (No runtime disambiguation; IDs are deterministic by design.) """ - etype = EntityType.QUESTION + entity_type = EntityType.ITEM estate = EntityState.DRAFT text = "identical question text" - id1 = FlexOID.generate(domain.domain, etype, estate, text) - id2 = FlexOID.generate(domain.domain, etype, estate, text) + id1 = FlexOID.generate(domain.domain, entity_type.value, estate.value, text) + id2 = FlexOID.generate(domain.domain, entity_type.value, estate.value, text) # IDs must be identical because generation is deterministic assert id1 == id2 -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). - """ - etype = EntityType.CATALOG - estate = EntityState.DRAFT - seed = "reproducibility test seed" +# 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). +# """ +# entity_type = EntityType.CATALOG +# estate = EntityState.DRAFT +# seed = "reproducibility test seed" - id1 = FlexOID.generate(domain.domain, etype, estate, seed) - FlexOID._seen_hashes.clear() - id2 = FlexOID.generate(domain.domain, etype, estate, seed) +# id1 = FlexOID.generate(domain.domain, entity_type.value, estate.value, seed) +# FlexOID._seen_hashes.clear() +# id2 = FlexOID.generate(domain.domain, entity_type.value, estate.value, seed) - assert id1 == id2 +# assert id1 == id2 -def test_version_ceiling_enforcement(radio_question): +def test_version_ceiling_enforcement(sample_question): """Simulate approaching @999 to trigger obsolescence guard.""" - q = radio_question + q = sample_question q.approve() @@ -97,26 +99,16 @@ -def test_massive_lifecycle_simulation(domain): +def test_massive_lifecycle_simulation(sample_question): """ - Generate 100 random RadioQuestions, simulate multiple edits and state transitions, + Generate 100 random SingleChoiceQuestions, simulate multiple edits and state transitions, ensure all final IDs and fingerprints are unique and valid. """ 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) + copy.deepcopy(sample_question) for _ in range(100) ] - for e in entities: + for i, e in enumerate(entities): # random edit - e.text += " updated" + e.text += f" updated #{i}" e._update_fingerprint() Index: tests/test_persistance_integrity.py =================================================================== --- tests/test_persistance_integrity.py (revision ca3927462b83d9ae55fd59d487c07e1a3f0f994c) +++ tests/test_persistance_integrity.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) @@ -6,13 +6,12 @@ import pytest -from builder.questions import RadioQuestion, AnswerOption from flexoentity import EntityState, EntityType, Domain @pytest.fixture def approved_question(): - """Provide a fully approved and published RadioQuestion for persistence tests.""" - q = RadioQuestion( - domain=Domain(domain="GEN", etype=EntityType.DOMAIN, state=EntityState.DRAFT), - etype=None, # RadioQuestion sets this internally to EntityType.QUESTION + """Provide a fully approved and published SingleChoiceQuestion for persistence tests.""" + q = SingleChoiceQuestion( + domain=Domain(domain="GEN", entity_type=EntityType.DOMAIN, state=EntityState.DRAFT), + entity_type=None, # SingleChoiceQuestion sets this internally to EntityType.ITEM state=EntityState.DRAFT, text="What is Ohm’s law?", @@ -36,10 +35,10 @@ json_str = approved_question.to_json() print("JSON", json_str) - loaded = RadioQuestion.from_json(json_str) + loaded = SingleChoiceQuestion.from_json(json_str) print("Approved", approved_question.text_seed) print("Loaded", loaded.text_seed) # Fingerprint and state should match — integrity must pass - assert RadioQuestion.verify_integrity(loaded) + assert SingleChoiceQuestion.verify_integrity(loaded) # Metadata should be preserved exactly @@ -56,6 +55,6 @@ tampered_json = json.dumps(tampered) - loaded = RadioQuestion.from_json(tampered_json) - assert not RadioQuestion.verify_integrity(loaded) + loaded = SingleChoiceQuestion.from_json(tampered_json) + assert not SingleChoiceQuestion.verify_integrity(loaded) @pytest.mark.skip(reason="FlexOIDs regenerated on import; corruption detection not yet applicable") @@ -70,4 +69,4 @@ file.write_text(corrupted) - loaded = RadioQuestion.from_json(file.read_text()) - assert not RadioQuestion.verify_integrity(loaded) + loaded = SingleChoiceQuestion.from_json(file.read_text()) + assert not SingleChoiceQuestion.verify_integrity(loaded)