Changeset 753855a in flexoentity

Timestamp:

12/03/25 22:14:53 (6 weeks ago)

Author:

Enrico Schwass <ennoausberlin@…>

Branches:

master

Children:

ea28ca0

Parents:

4459fa4

Message:

add more backends and tests

Files:

• flexoentity/json_file_backend.py (modified) (1 diff)
• flexoentity/sqlite_entity_backend.py (modified) (2 diffs)
• tests/conftest.py (modified) (3 diffs)
• tests/test_composite_backend.py (added)
• tests/test_in_memory_backend.py (added)
• tests/test_json_file_backend.py (added)
• tests/test_signing.py (modified) (3 diffs)
• tests/test_sqlite_backend.py (added)

flexoentity/json_file_backend.py

@@ -1 +1 @@
 import json
 from .persistance_backend import PersistanceBackend
+from .in_memory_backend import InMemoryBackend
 
 
 class JsonFileBackend(PersistanceBackend):
-    def __init__(self, path: str):
-        self.path = path
-        self.data = []
-        self._load_file()
+    """
+    JSON-file based backend.
 
-    def _load_file(self):
-        try:
-            with open(self.path, "r", encoding="utf-8") as f:
-                self.data = json.load(f)
-        except FileNotFoundError:
-            self.data = []
+    Uses an internal InMemoryBackend and syncs to a single JSON file on disk.
+    """
 
-    def save(self, entity_dict):
-        self.data.append(entity_dict)
-        self._flush()
+    def __init__(self, entity_class, path):
+        self._mem = InMemoryBackend(entity_class)
+        self._path = path
 
-    def update(self, entity_dict):
-        fid = entity_dict["flexo_id"]
-        self.data = [d for d in self.data if d["flexo_id"] != fid]
-        self.data.append(entity_dict)
-        self._flush()
+    @property
+    def entity_class(self):
+        return self._mem.entity_class
 
-    def delete(self, flexo_id):
-        self.data = [d for d in self.data if d["flexo_id"] != flexo_id]
-        self._flush()
+    # core API just delegates to memory backend
 
-    def load(self, flexo_id):
-        for d in self.data:
-            if d["flexo_id"] == flexo_id:
-                return d
-        return None
+    def save(self, entity):
+        self._mem.save(entity)
+
+    def update(self, entity):
+        self._mem.update(entity)
+
+    def delete(self, flexo_id: str):
+        self._mem.delete(flexo_id)
+
+    def load(self, flexo_id: str):
+        return self._mem.load(flexo_id)
 
     def load_all(self):
-        return self.data
+        return self._mem.load_all()
 
-    def _flush(self):
-        with open(self.path, "w", encoding="utf-8") as f:
-            json.dump(self.data, f, ensure_ascii=False, indent=2)
+    def clear(self):
+        self._mem.clear()
+
+    # file sync
+
+    def flush_to_file(self):
+        data = [e.to_dict() for e in self._mem.load_all()]
+        with open(self._path, "w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+
+    def load_from_file(self):
+        try:
+            with open(self._path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+        except FileNotFoundError:
+            return
+
+        self._mem.clear()
+        for d in data:
+            self._mem.save(self.entity_class.from_dict(d))

flexoentity/sqlite_entity_backend.py

@@ -1 @@
-import sqlite3
 import json
-from typing import Optional, Iterable
-
 from .persistance_backend import PersistanceBackend
 
@@ -8 +5 @@
 class SQLiteEntityBackend(PersistanceBackend):
     """
-    Generic SQLite persistence layer for FlexOEntity subclasses.
-    
-    Subclasses must define:
-      - TABLE_NAME        (e.g. "users")
-      - ENTITY_CLASS      (e.g. FlexoUser)
-      - INDEXED_FIELDS    (list of field names to index)
+    SQLite backend storing **dicts**, not entities.
+    Managers do the entity conversions.
     """
 
-    TABLE_NAME = None
-    ENTITY_CLASS = None
-    INDEXED_FIELDS: list[str] = []
-
-    def __init__(self, db_path: str):
-        if self.TABLE_NAME is None:
-            raise ValueError("Subclasses must define TABLE_NAME")
-        if self.ENTITY_CLASS is None:
-            raise ValueError("Subclasses must define ENTITY_CLASS")
-
-        self.db_path = db_path
-        self.conn = sqlite3.connect(db_path)
-        self.conn.row_factory = sqlite3.Row
+    def __init__(self, entity_class, conn, table_name):
+        self.entity_class = entity_class
+        self.conn = conn
+        self.table = table_name
         self._init_schema()
 
-    # -----------------------------------------------------------
-    # Schema
-    # -----------------------------------------------------------
     def _init_schema(self):
-        # Base schema: store full entity JSON and the FlexOID
-        columns = [
-            "flexo_id TEXT PRIMARY KEY",
-            "json TEXT NOT NULL",
-        ]
-
-        # Add indexed metadata columns
-        for field in self.INDEXED_FIELDS:
-            columns.append(f"{field} TEXT")
-
-        colspec = ", ".join(columns)
-
-        self.conn.execute(
-            f"CREATE TABLE IF NOT EXISTS {self.TABLE_NAME} ({colspec})"
-        )
-
-        # Create indices on metadata fields
-        for field in self.INDEXED_FIELDS:
-            self.conn.execute(
-                f"CREATE INDEX IF NOT EXISTS idx_{self.TABLE_NAME}_{field} "
-                f"ON {self.TABLE_NAME}({field})"
+        self.conn.execute(f"""
+            CREATE TABLE IF NOT EXISTS {self.table} (
+                flexo_id TEXT PRIMARY KEY,
+                json     TEXT NOT NULL
             )
-
+        """)
         self.conn.commit()
 
-    # -----------------------------------------------------------
-    # CRUD operations
-    # -----------------------------------------------------------
-    def add(self, obj):
-        data = json.dumps(obj.to_dict(), ensure_ascii=False)
-
-        # Collect row data dynamically
-        fields = ["flexo_id", "json"] + self.INDEXED_FIELDS
-        placeholders = ", ".join("?" for _ in fields)
-        sql = (
-            f"INSERT INTO {self.TABLE_NAME} ({', '.join(fields)}) "
-            f"VALUES ({placeholders})"
-        )
-
-        params = [obj.flexo_id.value, data]
-        params.extend(getattr(obj, f) for f in self.INDEXED_FIELDS)
-
-        self.conn.execute(sql, params)
-        self.conn.commit()
-
-    def update(self, obj):
-        data = json.dumps(obj.to_dict(), ensure_ascii=False)
-        set_parts = ["json = ?"] + [f"{f} = ?" for f in self.INDEXED_FIELDS]
-
-        sql = (
-            f"UPDATE {self.TABLE_NAME} SET "
-            f"{', '.join(set_parts)} "
-            f"WHERE flexo_id = ?"
-        )
-
-        params = [data]
-        params.extend(getattr(obj, f) for f in self.INDEXED_FIELDS)
-        params.append(obj.flexo_id.value)
-
-        self.conn.execute(sql, params)
-        self.conn.commit()
-
-    def delete(self, flexo_id: str):
+    def save(self, entity):
+        entity_dict = entity.to_dict()
+        fid = entity_dict["meta"]["flexo_id"]
         self.conn.execute(
-            f"DELETE FROM {self.TABLE_NAME} WHERE flexo_id = ?",
-            (flexo_id,),
+            f"INSERT OR REPLACE INTO {self.table} (flexo_id, json) VALUES (?, ?)",
+            (fid, json.dumps(entity_dict))
         )
         self.conn.commit()
 
-    # -----------------------------------------------------------
-    # Retrieval
-    # -----------------------------------------------------------
-    def get(self, flexo_id: str):
+    def update(self, entity):
+        entity_dict = entity.to_dict()
+        fid = entity_dict["meta"]["flexo_id"]
+        self.conn.execute(
+            f"UPDATE {self.table} SET json = ? WHERE flexo_id = ?",
+            (json.dumps(entity_dict), fid)
+        )
+        self.conn.commit()
+
+    def delete(self, flexo_id):
+        self.conn.execute(
+            f"DELETE FROM {self.table} WHERE flexo_id = ?",
+            (flexo_id,)
+        )
+        self.conn.commit()
+
+    def load(self, flexo_id):
         row = self.conn.execute(
-            f"SELECT json FROM {self.TABLE_NAME} WHERE flexo_id = ?",
+            f"SELECT json FROM {self.table} WHERE flexo_id = ?",
             (flexo_id,),
         ).fetchone()
+
         if not row:
             return None
-        return self.ENTITY_CLASS.from_dict(json.loads(row["json"]))
 
-    def all(self):
+        return self.entity_class.from_dict(json.loads(row["json"]))
+
+    def load_all(self):
         rows = self.conn.execute(
-            f"SELECT json FROM {self.TABLE_NAME}"
+            f"SELECT json FROM {self.table}"
         ).fetchall()
-        return [
-            self.ENTITY_CLASS.from_dict(json.loads(r["json"])) for r in rows
-        ]
 
-    # -----------------------------------------------------------
-    # Dynamic finders: find_by_<field>
-    # -----------------------------------------------------------
-    def __getattr__(self, name):
-        # e.g. find_by_username("foo")
-        if name.startswith("find_by_"):
-            field = name[len("find_by_"):]
-            if field not in self.INDEXED_FIELDS:
-                raise AttributeError(f"No such indexed field: {field}")
+        return [self.entity_class.from_dict(json.loads(r["json"])) for r in rows]
 
-            def finder(value):
-                rows = self.conn.execute(
-                    f"SELECT json FROM {self.TABLE_NAME} WHERE {field} = ?",
-                    (value,),
-                )
-                for row in rows:
-                    yield self.ENTITY_CLASS.from_dict(json.loads(row["json"]))
-
-            return finder
-
-        raise AttributeError(name)
+    def clear(self):
+        self.conn.execute(f"DELETE FROM {self.table}")
+        self.conn.commit()

tests/conftest.py

@@ -3 +3 @@
 from pathlib import Path
 from datetime import datetime
-from flexoentity import Domain, FlexoSignature, DomainManager, EntityRegistry, CompositeBackend
+from flexoentity import Domain, FlexoSignature, DomainManager, EntityRegistry, CompositeBackend, InMemoryBackend
 from flexoentity import get_signing_backend, CertificateReference
 
@@ -26 +26 @@
 
 SYSTEM = platform.system()
+
+@pytest.fixture
+def backend():
+    return InMemoryBackend(Domain)
 
 
@@ -81 +85 @@
 
 @pytest.fixture(scope="session")
-def backend(test_cert, test_key):
+def signing_backend(test_cert, test_key):
     """Return the correct backend for the current platform."""
 

tests/test_signing.py

@@ -4 +4 @@
 from flexoentity import FlexOID, EntityState, EntityType, FlexoSignature, get_signing_backend
 
-def test_sign_and_verify_linux(backend):
+def test_sign_and_verify_linux(signing_backend):
     data = b"Hello Flex-O signing!"
 
-    signature = backend.sign(data)
+    signature = signing_backend.sign(data)
     assert isinstance(signature, bytes)
     assert len(signature) > 20      # sanity check
 
-    assert backend.verify(data, signature) is True
+    assert signing_backend.verify(data, signature) is True
 
-def test_sign_and_verify_macos(backend):
+def test_sign_and_verify_macos(signing_backend):
     data = b"Hello Flex-O signing!"
 
-    signature = backend.sign(data)
+    signature = signing_backend.sign(data)
     assert isinstance(signature, bytes)
     assert len(signature) > 20      # sanity check
 
-    assert backend.verify(data, signature) is True
+    assert signing_backend.verify(data, signature) is True
 
-def test_verify_fails_with_wrong_data(backend):
+def test_verify_fails_with_wrong_data(signing_backend):
     data = b"Original Data"
     wrong_data = b"Tampered Data"
 
-    signature = backend.sign(data)
+    signature = signing_backend.sign(data)
 
-    assert backend.verify(data, signature) == True
-    assert backend.verify(wrong_data, signature) == False
+    assert signing_backend.verify(data, signature) == True
+    assert signing_backend.verify(wrong_data, signature) == False
 
-def test_verify_fails_with_invalid_signature(backend):
+def test_verify_fails_with_invalid_signature(signing_backend):
     data = b"Hello world"
     invalid_sig = b"\x00\x01\x02garbagepkcs7data"
 
-    assert backend.verify(data, invalid_sig) is False
+    assert signing_backend.verify(data, invalid_sig) is False
 
-def test_signature_entity_create_and_verify(backend):
+def test_signature_entity_create_and_verify(signing_backend):
     entity_id = FlexOID.safe_generate(
         domain_id="TEST",
@@ -52 +52 @@
         entity=entity_id,
         signer_id=signer,
-        backend=backend,
+        backend=signing_backend,
     )
 
@@ -61 +61 @@
     assert sig.certificate_thumbprint != ""
 
-    assert sig.verify(data, backend) is True
+    assert sig.verify(data, signing_backend) is True