Index: flexoentity/__init__.py =================================================================== --- flexoentity/__init__.py (revision 4a79b76446b26c92cb27fc0805ac45b099f34b9d) +++ flexoentity/__init__.py (revision df459f718c057333d059679915ea06d65dead882) @@ -15,4 +15,5 @@ from .flexo_collection import FlexoCollection from .domain import Domain +from .domain_manager import DomainManager, DuplicateDomainError from .flexo_signature import FlexoSignature, CertificateReference from .signing_backends import get_signing_backend @@ -24,4 +25,6 @@ "EntityType", "Domain", + "DomainManager", + "DuplicateDomainError", "EntityState", "FlexoCollection", Index: flexoentity/domain_manager.py =================================================================== --- flexoentity/domain_manager.py (revision df459f718c057333d059679915ea06d65dead882) +++ flexoentity/domain_manager.py (revision df459f718c057333d059679915ea06d65dead882) @@ -0,0 +1,113 @@ +# flexoentity/domain_manager.py + +from __future__ import annotations + +from typing import Iterable + +from .flexo_collection import FlexoCollection +from .domain import Domain +from .flexo_entity import EntityState + + +class DomainNotFoundError(KeyError): + pass + + +class InvalidDomainError(ValueError): + pass + + +class DuplicateDomainError(ValueError): + pass + + +class DomainManager: + """ + Manager for Domain entities, implemented using composition. + + - Uses FlexoCollection internally. + - Domains are keyed by domain.domain_id (explicit override). + - No inheritance from FlexoCollection. + - Responsible ONLY for domain-specific rules. + """ + + def __init__(self, domains: Iterable[Domain] | None = None): + # store is a general-purpose collection + self._store = FlexoCollection() + + if domains: + for dom in domains: + self.add(dom) + + # --------------------------------------------------------------- + # Core API + # --------------------------------------------------------------- + + def add(self, domain: Domain) -> None: + """ + Add a domain using domain.domain_id as key. + """ + if not isinstance(domain, Domain): + raise TypeError( + f"DomainManager accepts only Domain instances, got {type(domain)}" + ) + if domain.domain_id in self._store: + raise DuplicateDomainError(f"Domain '{domain.domain_id} already exists.") + self._store.add(domain, key=domain.domain_id) + + def get(self, domain_id: str) -> Domain: + """Retrieve a domain or raise DomainNotFoundError.""" + if isinstance(domain_id, Domain): + raise TypeError("DomainManager.get() expects a domain_id string, not a Domain object.") + result = self._store.get(domain_id) + if result is None: + raise DomainNotFoundError(f"Domain '{domain_id}' not found.") + return result + + def has(self, domain_id: str) -> bool: + """Check if a domain exists.""" + return self._store.has(domain_id) + + def remove(self, domain_id: str) -> None: + """Remove a domain (rarely used, mostly for tests).""" + if not self._store.has(domain_id): + raise DomainNotFoundError(domain_id) + self._store.remove(domain_id) + + def all(self) -> list[Domain]: + """List of all domains.""" + return list(self._store.values()) + + def clear(self): + self._store.clear() + + # --------------------------------------------------------------- + # Lifecycle / Approval Helpers + # --------------------------------------------------------------- + + def ensure_approved(self, domain_id: str) -> None: + """ + Used during entity.transition_to(). + Raises if the domain does not exist or is not APPROVED. + """ + dom = self.get(domain_id) + + if dom.state != EntityState.APPROVED: + raise InvalidDomainError( + f"Domain '{domain_id}' must be APPROVED before " + f"entities in this domain may be promoted. " + f"(Current state: {dom.state})" + ) + + # --------------------------------------------------------------- + # Representations + # --------------------------------------------------------------- + + def __len__(self) -> int: + return len(self._store) + + def __contains__(self, domain_id: str) -> bool: + return self.has(domain_id) + + def __repr__(self) -> str: + return f"" Index: flexoentity/flexo_collection.py =================================================================== --- flexoentity/flexo_collection.py (revision 4a79b76446b26c92cb27fc0805ac45b099f34b9d) +++ flexoentity/flexo_collection.py (revision df459f718c057333d059679915ea06d65dead882) @@ -5,13 +5,10 @@ A minimal collection for FlexOEntities. -- No domain rules -- No uniqueness validation (last write wins) -- No state or lifecycle logic -- No typing constraints - -The goal is to provide a Smalltalk-like Collection object with a simple -protocol for storing, retrieving, iterating, and serializing FlexOEntities. +- Default key = entity.flexo_id +- Optional override key=... for special cases (e.g., domains) +- Smalltalk-like protocol preserved +- Dict-based: fast lookup +- Backwards compatible API """ - from .flexo_entity import FlexoEntity @@ -21,10 +18,11 @@ class FlexoCollection: """ - A minimal collection of FlexOEntities, keyed by FlexOID. + A minimal collection of FlexOEntities, keyed by FlexOID by default. Examples: coll = FlexoCollection() - coll.add(entity) - entity = coll.get(some_id) + coll.add(entity) # uses entity.flexo_id as key + coll.add(domain, key=domain.domain_id) # override key if needed + ent = coll.get(some_id) for e in coll: ... @@ -32,4 +30,5 @@ def __init__(self, items=None): + # internal dict-based storage self._items = {} @@ -38,11 +37,31 @@ self.add(it) - def add(self, entity: FlexoEntity): - """Add or replace an entity by its FlexOID. Overwrites on duplicate IDs.""" - self._items[entity.flexo_id] = entity + # ------------------------------------------------------------------ + # Core API (unchanged externally, enhanced internally) + # ------------------------------------------------------------------ + + def add(self, entity: FlexoEntity, key=None): + """ + Add or replace an entity. + + - If key is None → use entity.flexo_id + - If key is given → use caller-provided key + """ + if key is None: + try: + key = entity.flexo_id + except AttributeError: + raise TypeError( + "Items must have .flexo_id or explicitly use add(entity, key=...)." + ) + + self._items[key] = entity # overwrite is intentional & preserved def remove(self, oid: FlexOID): - """Remove an entity by ID, ignoring if missing.""" + """Remove an entity by ID, ignoring if missing (API preserved).""" self._items.pop(oid, None) + + def clear(self): + self._items = {} def get(self, oid: FlexOID): @@ -60,14 +79,18 @@ return iter(self._items.values()) + # ------------------------------------------------------------------ + # Additional public helpers (unchanged) + # ------------------------------------------------------------------ + def entities(self): - """Return all entities.""" + """Return all entities as a list.""" return list(self._items.values()) def ids(self): - """Return all IDs.""" + """Return all stored keys.""" return list(self._items.keys()) # ------------------------------------------------------------------ - # Smalltalk-inspired interface + # Smalltalk-inspired interface (unchanged) # ------------------------------------------------------------------ @@ -77,5 +100,5 @@ def at_put(self, oid, entity): - """Smalltalk-style setter.""" + """Smalltalk-style setter (kept).""" self._items[oid] = entity @@ -90,9 +113,9 @@ # ------------------------------------------------------------------ - # Serialization helpers + # Serialization helpers (unchanged) # ------------------------------------------------------------------ def to_dict_list(self): - """Serialize entities to a list of dicts.""" + """Serialize all entities to list of dicts.""" return [e.to_dict() for e in self._items.values()] @@ -102,5 +125,5 @@ Deserialize a list of dicts into a FlexoCollection. - Uses FlexoEntity.from_dict to dispatch to the correct subclass. + Uses FlexoEntity.from_dict() to dispatch to the correct subclass. """ c = cls() @@ -110,4 +133,6 @@ return c + # ------------------------------------------------------------------ + def __repr__(self) -> str: return f"" Index: tests/conftest.py =================================================================== --- tests/conftest.py (revision 4a79b76446b26c92cb27fc0805ac45b099f34b9d) +++ tests/conftest.py (revision df459f718c057333d059679915ea06d65dead882) @@ -3,5 +3,5 @@ from pathlib import Path from datetime import datetime -from flexoentity import Domain, FlexoSignature +from flexoentity import Domain, FlexoSignature, DomainManager from flexoentity import get_signing_backend, CertificateReference @@ -28,7 +28,12 @@ +@pytest.fixture +def sample_domain_manager(): + return DomainManager() + # ───────────────────────────────────────────────────────────── # Basic test data directory + PEM test files # ───────────────────────────────────────────────────────────── + @pytest.fixture(scope="session") Index: tests/test_domain.py =================================================================== --- tests/test_domain.py (revision df459f718c057333d059679915ea06d65dead882) +++ tests/test_domain.py (revision df459f718c057333d059679915ea06d65dead882) @@ -0,0 +1,56 @@ +import pytest +from flexoentity import Domain, DomainManager, DuplicateDomainError + + +# --------------------------------------------------------------- +# Setup/Teardown (clear DomainManager between tests) +# --------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def clear_domain_manager(sample_domain_manager): + sample_domain_manager.clear() + + +# --------------------------------------------------------------- +# Domain creation + registration +# --------------------------------------------------------------- +def test_domain_registration(sample_domain_manager, sample_domain): + + sample_domain_manager.add(sample_domain) + # Manager must know it now + assert sample_domain_manager.get("PY_ARITHM") is sample_domain + + +# --------------------------------------------------------------- +# Uniqueness: registering the same code twice must fail +# --------------------------------------------------------------- +def test_domain_uniqueness_enforced(sample_domain, sample_domain_manager): + + sample_domain_manager.add(sample_domain) + with pytest.raises(DuplicateDomainError): + sample_domain_manager.add(sample_domain) + + +# --------------------------------------------------------------- +# Lookup by FlexOID +# --------------------------------------------------------------- +def test_lookup_by_oid(sample_domain, sample_domain_manager): + sample_domain_manager.add(sample_domain) + found = sample_domain_manager.get(sample_domain.domain_id) + assert found is sample_domain + + +# --------------------------------------------------------------- +# JSON roundtrip should preserve identity and not regenerate FlexOID +# --------------------------------------------------------------- +def test_domain_json_roundtrip(sample_domain): + sample_data = sample_domain.to_dict() + loaded = Domain.from_dict(sample_data) + + # Same exact instance + assert loaded == sample_domain + + # Same FlexOID + assert str(loaded.flexo_id) == str(sample_domain.flexo_id) +