Changeset 5c72356 in flexoentity

Timestamp:

11/02/25 18:49:14 (2 months ago)

Author:

Enrico Schwass <ennoausberlin@…>

Branches:

master

Children:

bf30018

Parents:

8aa20c7

Message:

fix tests due to simplifying state and type

Files:

• flexoentity/domain.py (modified) (2 diffs)
• flexoentity/flexo_entity.py (modified) (12 diffs)
• tests/conftest.py (modified) (5 diffs)
• tests/test_id_lifecycle.py (modified) (4 diffs)
• tests/test_id_stress.py (modified) (1 diff)

flexoentity/domain.py

@@ -1 +1 @@
 from dataclasses import dataclass
-from flexoentity.flexo_entity import FlexoEntity, EntityType, EntityState
+from flexoentity.flexo_entity import FlexOID, FlexoEntity, EntityType, EntityState
 
 
@@ -12 +12 @@
     @classmethod
     def default(cls):
-        return cls(domain="GEN", entity_type=EntityType.DOMAIN, state=EntityState.DRAFT)
+        return cls(domain="GEN")
 
     def __post_init__(self):
         super().__post_init__()
 
+        # Generate FlexOID only if missing
+        if not getattr(self, "flexo_id", None):
+            domain_code = self.domain or "GEN"
+            self.flexo_id = FlexOID.safe_generate(
+                domain=domain_code,
+                entity_type=EntityType.DOMAIN.value,   # 'D'
+                estate=EntityState.DRAFT.value,      # 'D'
+                text=self.text_seed,
+                version=1,
+            )
+
+
+        
     @property
     def text_seed(self) -> str:

flexoentity/flexo_entity.py

@@ -1 +1 @@
 """
-versioned_entity.py — Shared base class for all Flex-O entities.
-Integrates with hardened id_factory.py and content fingerprint tracking.
+flexo_entity.py - I represent the mutable counterpart of the immutable FlexOID.
+
+I exist to connect an entity’s immutable identity with its mutable provenance
+and content.  My FlexOID encodes what an entity *is* - its domain, type,
+version, and lifecycle state - while I record *who* created it, *who* owns it
+now, and *where* it came from.
+
+Structure
+──────────
+- My `flexo_id` is the canonical source of truth.
+  It deterministically encodes:
+    - domain      (prefix)
+    - entity type (single letter, e.g. I, C, M)
+    - date + hash (prefix stability)
+    - version     (numeric counter)
+    - state       (single capital letter)
+
+- My derived properties (`state`, `entity_type`, `version`) read from that ID.
+  They are never stored separately, so I cannot become inconsistent.
+
+- My attributes `origin`, `originator_id`, and `owner_id` express authorship
+  and workflow context:
+    - `origin` links to the FlexOID of the entity I was derived from
+    - `originator_id` identifies the original creator (UUID)
+    - `owner_id` identifies the current responsible user or process (UUID)
+
+- My optional `domain` field is a semantic hint, useful for filtering or
+  multi-tenant setups, but the canonical domain remains encoded in my ID.
+
+Design principles
+──────────────────
+I separate immutable identity from mutable context:
+
+    ┌────────────────────────────┐
+    │ FlexOID                    │
+    │  – who I am                │
+    │  – lifecycle state         │
+    └────────────▲───────────────┘
+                 │
+    ┌────────────┴───────────────┐
+    │ FlexoEntity (this module)  │
+    │  – who created or owns me  │
+    │  – where I came from       │
+    │  – what content I hold     │
+    └────────────────────────────┘
+
+This means:
+  - My identity never changes; only my content or ownership does.
+  - My lifecycle transitions (approve, sign, publish) are represented by new
+    FlexOIDs, not by mutating state attributes.
+  - My audit trail and access control live in the provenance layer, not the ID.
+
+I am the living object behind a FlexOID — mutable, traceable, and accountable.
 """
+
 import json
-import re
 from uuid import UUID
 from enum import Enum
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Optional
 from abc import ABC, abstractmethod
@@ -67 +118 @@
 @dataclass(kw_only=True)
 class FlexoEntity(ABC):
+    """
+    I represent a mutable entity identified by an immutable FlexOID.
+
+    My `flexo_id` is the single source of truth for what I am:
+      - it encodes my domain, entity type, version, and lifecycle state.
+    I never store those values separately, but derive them on demand.
+
+    I extend the raw identity with provenance and authorship data:
+      - `origin`        → the FlexOID of the entity I was derived from
+      - `originator_id` → the UUID of the first creator
+      - `owner_id`      → the UUID of the current responsible person or system
+      - `domain`        → optional semantic domain hint (for filtering or indexing)
+
+    I am designed to be traceable and reproducible:
+      - My identity (`flexo_id`) never mutates.
+      - My lifecycle transitions are represented by new FlexOIDs.
+      - My provenance (origin, originator_id, owner_id) can change over time
+        as the entity moves through workflows or ownership transfers.
+
+    Behavior
+    ─────────
+    - `state` and `entity_type` are read-only views derived from my FlexOID.
+    - `to_dict()` and `from_dict()` serialize me with human-readable fields
+      while preserving the canonical FlexOID.
+    - Subclasses (such as questions, media items, or catalogs) extend me with
+      domain-specific content and validation, but not with new identity rules.
+
+    I am the living, editable layer that connects identity with accountability.
+    """
+
     domain: str
-    entity_type: EntityType
     subtype: str = "GENERIC"
-    state: EntityState
     flexo_id: Optional[FlexOID] = field(default=None)
     fingerprint: str = field(default_factory=str)
@@ -77 +156 @@
     origin: Optional[str] = field(default=None)
 
+    def with_new_owner(self, new_owner: UUID):
+        """Return a clone of this entity with a different owner."""
+        copy = replace(self)
+        copy.owner_id = new_owner
+        return copy
+
     @property
     @abstractmethod
@@ -83 +168 @@
         raise NotImplementedError("Subclasses must define text_seed property")
 
+    @property
+    def state(self) -> EntityState:
+        return EntityState(self.flexo_id.state_code)
+
+    @property
+    def entity_type(self) -> EntityType:
+        return EntityType(self.flexo_id.entity_type) 
+
     def canonical_seed(self) -> str:
         return canonical_seed(self.text_seed)
@@ -98 +191 @@
     def __post_init__(self):
         """
-        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.safe_generate(self.domain_code(),
-                                         self.entity_type.value,
-                                         self.state.value,
-                                         self.text_seed,
-                                         1)
-        seed = canonical_seed(self.text_seed)
-        self.fingerprint = hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper()
+        Optionally generate a new FlexOID and fingerprint if none exist.
+
+        I check for subclass attributes `ENTITY_TYPE` and `INITIAL_STATE`.
+        If they exist, I use them to generate a draft FlexOID.
+        Otherwise, I skip ID generation — leaving it to the subclass.
+        """
+
+        # If already has a FlexOID, do nothing
+        if getattr(self, "flexo_id", None):
+            return
+
+        # Skip if subclass doesn’t declare ENTITY_TYPE / INITIAL_STATE
+        etype = getattr(self.__class__, "ENTITY_TYPE", None)
+        if not etype:
+            # No info available, do nothing
+            return
+
+        # Generate a new draft FlexOID
+        self.flexo_id = FlexOID.safe_generate(
+            self.domain_code(),
+            etype.value,
+            EntityState.DRAFT.value,
+            self.text_seed,
+            1,
+        )
+
+        # Compute fingerprint
+        self.fingerprint = self._compute_fingerprint()
 
     def __str__(self):
@@ -135 +242 @@
         abbrev, fullname = (lambda p: (p[0], p[1] if len(p) > 1 else ""))(domain.split("_", 1))
         domain_obj = Domain(
-            domain=abbrev,
-            entity_type=EntityType.DOMAIN,
-            state=EntityState.DRAFT,  # default when reconstructing context
+            domain=abbrev
         )
         obj = cls(
             domain=domain_obj,
-            entity_type=EntityType[data["entity_type"]],
-            state=EntityState[data["state"]],
-        )
-        obj.flexo_id = FlexOID.from_string(data["flexo_id"])
+        )
+        obj.flexo_id = FlexOID(data["flexo_id"])
         obj.fingerprint = data.get("fingerprint", "")
         obj.origin = data.get("origin")
@@ -207 +310 @@
         # special case: marking obsolete
         if target_state == EntityState.OBSOLETE:
-            self.flexo_id = FlexOID.with_state(self.flexo_id, "O")
-            self.state = target_state
+            self.flexo_id = self.flexo_id.with_state(EntityState.OBSOLETE.value)
             return
 
@@ -258 +360 @@
             self.origin = self.flexo_id  # optional: keep audit trail
             self.flexo_id = new_fid
-            self.state = EntityState.APPROVED
             return self
         raise ValueError("Only drafts can be approved")
 
+
     def sign(self):
-        self._transition(EntityState.APPROVED_AND_SIGNED)
+        """
+        Mark entity as approved and signed without bumping version.
+        Only changes the state letter in the FlexOID.
+        """
+        if self.state != EntityState.APPROVED:
+            raise ValueError("Only approved entities can be signed.")
+
+        # Change only the trailing state letter (A → S)
+        self.flexo_id = self.flexo_id.with_state(EntityState.APPROVED_AND_SIGNED.value)
+        return self
 
     def publish(self):
@@ -279 +390 @@
             )
 
-        new_version = self.flexo_id.version + 1
         new_fid = FlexOID.safe_generate(
             self.domain_code(),
@@ -285 +395 @@
             EntityState.PUBLISHED.value,
             self.text_seed,
-            version=new_version
+            version=self.version
         )
 
         self.origin = self.flexo_id
         self.flexo_id = new_fid
-        self.state = EntityState.PUBLISHED
         return self
 
@@ -302 +411 @@
         if self.state != EntityState.OBSOLETE:
             self._transition(EntityState.OBSOLETE)
+        return self
 
     def clone_new_base(self):
@@ -309 +419 @@
             self.domain_code(),
             self.entity_type.value,
-            self.state.value,
+            EntityState.DRAFT.value,
             self.text_seed,
         )
-        self.state = EntityState.DRAFT
+        return self
 
     # ───────────────────────────────────────────────────────────────
     # Integrity verification
     # ───────────────────────────────────────────────────────────────
+    @staticmethod
+    def debug_integrity(entity):
+        """
+        Return a dict with all values used by verify_integrity so we can see
+        exactly why a check passes/fails.
+        """
+        info = {}
+        try:
+            fid = entity.flexo_id
+            info["fid"] = str(fid)
+            info["fid_domain"] = fid.domain
+            info["fid_type"] = fid.entity_type
+            info["fid_state"] = fid.state_code
+            info["fid_hash_part"] = fid.hash_part
+            info["version"] = fid.version
+
+            # Derived views
+            info["derived_domain_code"] = getattr(entity, "domain_code")() if hasattr(entity, "domain_code") else None
+            info["derived_entity_type_value"] = entity.entity_type.value
+            info["derived_state_value"] = entity.state.value
+
+            # Seeds & fingerprints
+            info["text_seed"] = entity.text_seed
+            exp_fp = entity._compute_fingerprint()
+            info["expected_fingerprint"] = exp_fp
+            info["stored_fingerprint"] = getattr(entity, "fingerprint", None)
+
+            # Expected ID hash from current seed (should match fid.hash_part if content didn't change)
+            exp_hash = FlexOID._blake_hash(canonical_seed(f"{fid.domain}:{fid.entity_type}:{entity.text_seed}"))
+            info["expected_id_hash_from_seed"] = exp_hash
+
+            # Quick mismatches
+            info["mismatch_fp"] = (info["stored_fingerprint"] != exp_fp)
+            info["mismatch_hash"] = (fid.hash_part != exp_hash)
+            info["mismatch_type"] = (entity.entity_type.value != fid.entity_type)
+            info["mismatch_state"] = (entity.state.value != fid.state_code)
+            info["mismatch_domain"] = (info["derived_domain_code"] is not None and info["derived_domain_code"] != fid.domain)
+        except Exception as e:
+            info["error"] = repr(e)
+        return info
 
     @staticmethod
     def verify_integrity(entity) -> bool:
-        """Verify that an entity’s content fingerprint matches its actual content."""
-        expected_fp = hashlib.blake2s(
-            canonical_seed(entity.text_seed).encode("utf-8"), digest_size=8
-        ).hexdigest().upper()
-        return expected_fp == entity.fingerprint
+        """
+        Verify that an entity's FlexOID, state, and fingerprint are consistent.
+
+        Returns False if:
+          - flexo_id format is invalid
+          - derived state/type mismatch the ID
+          - fingerprint doesn't match the canonical text seed
+        """
+        try:
+            if not isinstance(entity.flexo_id, FlexOID):
+                return False
+
+            # Validate domain and ID coherence
+            if entity.domain and entity.domain_code() != entity.flexo_id.domain:
+                return False
+            if entity.entity_type.value != entity.flexo_id.entity_type:
+                return False
+            if entity.state.value != entity.flexo_id.state_code:
+                return False
+
+            # --- Fingerprint validation ---
+            if hasattr(entity, "fingerprint") and entity.fingerprint:
+                seed = canonical_seed(entity.text_seed)
+                expected_fp = entity._compute_fingerprint()
+                if expected_fp != entity.fingerprint:
+                    return False
+
+            return True
+
+        except Exception as e:
+            return False
 
     def allowed_transitions(self) -> list[str]:

tests/conftest.py

@@ -4 +4 @@
 from dataclasses import dataclass, field
 from typing import List
-from flexoentity import FlexoEntity, EntityType, EntityState, Domain
+from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState, Domain
 
 @pytest.fixture
@@ -37 +37 @@
 class SingleChoiceQuestion(FlexoEntity):
     """A minimal stub to test FlexoEntity integration."""
+    ENTITY_TYPE = EntityType.ITEM
+
     text: str = ""
     options: List[AnswerOption] = field(default_factory=list)
 
+    def __post_init__(self):
+        # If no FlexOID yet, generate a draft ID now.
+        if not getattr(self, "flexo_id", None):
+            domain_code = (
+                self.domain.domain if isinstance(self.domain, Domain) else "GEN"
+            )
+            self.flexo_id = FlexOID.safe_generate(
+                domain=domain_code,
+                entity_type=EntityType.ITEM.value,     # 'I'
+                estate=EntityState.DRAFT.value,        # 'D'
+                text=self.text_seed or self.text,
+                version=1,
+            )
 
     @classmethod
     def default(cls):
-        return cls(domain=Domain(domain="GEN",
-                                 entity_type=EntityType.DOMAIN,
-                                 state=EntityState.DRAFT),
-                   state=EntityState.DRAFT, entity_type=EntityType.ITEM)
+        return cls(domain=Domain(domain="GEN"))
 
     def to_dict(self):
@@ -68 +80 @@
     @classmethod
     def from_dict(cls, data):
-        obj = cls(
+        obj = cls(domain=Domain(domain="GEN"),
             text=data.get("text", ""),
             options=[AnswerOption.from_dict(o) for o in data.get("options", [])],
@@ -74 +86 @@
         # 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
@@ -87 +96 @@
 @pytest.fixture
 def sample_question():
-    return SingleChoiceQuestion(domain=Domain.default(),
-                               text="What is 2 + 2?",
-                               options=[],
-                               entity_type=EntityType.ITEM,
-                               state=EntityState.DRAFT)
+    q = SingleChoiceQuestion(domain=Domain.default(),
+                             text="What is 2 + 2?",
+                             options=[])
+    q._update_fingerprint()
+    return q

tests/test_id_lifecycle.py

@@ -20 +20 @@
     assert q.flexo_id.version == 1
 
-
-def test_signing_bumps_version(sample_question):
+def test_signing_does_not_bump_version(sample_question):
     q = sample_question
     q.approve()
-    v_before = str(q.flexo_id)
+    before = q.flexo_id
     q.sign()
+    after = q.flexo_id
+
+    # state changed
     assert q.state == EntityState.APPROVED_AND_SIGNED
-    assert str(q.flexo_id) != v_before
+
+    # version unchanged
+    assert before.version == after.version
+
+    # only suffix letter differs
+    assert before.prefix == after.prefix
+    assert before.state_code == "A"
+    assert after.state_code == "S"
 
 
-def test_publish_bumps_version(sample_question):
+def test_publish_does_not_bump_version(sample_question):
     q = sample_question
     q.approve()
@@ -37 +46 @@
     q.publish()
     assert q.state == EntityState.PUBLISHED
-    assert q.flexo_id.version == v_before + 1
+    assert q.flexo_id.version == v_before
 
 
@@ -70 +79 @@
     # simulate tampering
     q.text = "Tampered text"
+    print(FlexoEntity.debug_integrity(q))
     assert not FlexoEntity.verify_integrity(q)
 
@@ -112 +122 @@
     for _ in range(FlexOID.MAX_VERSION - 1):
         q.bump_version()
+
+    # Next one must raise
     with pytest.raises(RuntimeError, match="mark obsolete"):
-        q.sign()
+        q.bump_version()

tests/test_id_stress.py

@@ -82 +82 @@
 
 
-def test_version_ceiling_enforcement(sample_question):
-    """Simulate approaching @999 to trigger obsolescence guard."""
-    q = sample_question
-    q.approve()
-
-    # artificially bump version number to near ceiling
-    q.flexo_id = FlexOID.from_oid_and_version(q.flexo_id, 998)
-
-    # 998 → 999 is allowed
-    q.sign()
-    assert q.flexo_id.version == 999
-
-    # 999 → 1000 should raise RuntimeError
-    with pytest.raises(ValueError):
-        q.publish()
-
-
 def test_massive_lifecycle_simulation(sample_question):
     """