Make database changes outside of transaction a fatal error. Fix bootstrap and migrate to work with various latest changes.

paskia/db/jsonl.py

@@ -196,6 +196,9 @@ class JsonlStore:
                         )
                         _logger.info("Queued migration changes for persistence")
                     await self.flush()
+            else:
+                # No data loaded - _previous_builtins stays as empty dict
+                pass
         except ValueError:
             if self.db_path.exists():
                 raise
@@ -248,19 +251,18 @@ class JsonlStore:
         # Check for out-of-transaction modifications
         current_state = msgspec.to_builtins(self.db)
         if current_state != self._previous_builtins:
-            diff = compute_diff(self._previous_builtins, current_state)
-            diff_json = json.dumps(diff, default=str, indent=2)
-            _logger.error(
-                "Database state modified outside of transaction! "
-                "This indicates a bug where DB changes occurred without a transaction wrapper. "
-                "Resetting to last known state from JSONL file.\n"
-                f"Changes detected:\n{diff_json}"
-            )
-            # Hard reset to last known good state
-            decoder = msgspec.json.Decoder(DB)
-            self.db = decoder.decode(msgspec.json.encode(self._previous_builtins))
-            self.db._store = self
-            current_state = self._previous_builtins.copy()
+            # Allow bootstrap/migrate to create a new database from empty state
+            if action in _BOOTSTRAP_ACTIONS and not self._previous_builtins:
+                pass  # Expected: creating database from scratch
+            else:
+                diff = compute_diff(self._previous_builtins, current_state)
+                diff_json = json.dumps(diff, default=str, indent=2)
+                _logger.critical(
+                    "Database state modified outside of transaction! "
+                    "This indicates a bug where DB changes occurred without a transaction wrapper.\n"
+                    f"Changes detected:\n{diff_json}"
+                )
+                raise SystemExit(1)
 
         old_action = self._current_action
         old_user = self._current_user

paskia/migrate/__init__.py

@@ -154,7 +154,7 @@ async def migrate_from_sql(
                 if perm_uuid:
                     new_permissions[perm_uuid] = True
             new_role = Role(
-                org=role.org_uuid,
+                org_uuid=role.org_uuid,
                 display_name=role.display_name,
                 permissions=new_permissions,
             )
@@ -172,7 +172,7 @@ async def migrate_from_sql(
             user_key: UUID = legacy_user.uuid
             new_user = User(
                 display_name=legacy_user.display_name,
-                role=legacy_user.role_uuid,
+                role_uuid=legacy_user.role_uuid,
                 created_at=legacy_user.created_at or datetime.now(UTC),
                 last_seen=legacy_user.last_seen,
                 visits=legacy_user.visits,

paskia/migrate/sql.py

@@ -25,11 +25,6 @@ from sqlalchemy.dialects.sqlite import BLOB
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
-from paskia.db import (
-    Org,
-    Role,
-)
-
 
 # Legacy User class for SQL schema (uses 'role_uuid' not 'role')
 @dataclass
@@ -71,6 +66,17 @@ class _LegacyRole:
     permissions: list[str] | None = None
 
 
+# Legacy Org class for SQL schema (has mutable permissions/roles lists)
+@dataclass
+class _LegacyOrg:
+    """Org as stored in the old SQL schema with mutable permissions/roles."""
+
+    uuid: UUID
+    display_name: str
+    permissions: list[str] | None = None
+    roles: list[_LegacyRole] | None = None
+
+
 # Legacy Session class for SQL schema (uses 'key' as field, 'user_uuid', 'credential_uuid')
 @dataclass
 class _LegacySession:
@@ -128,12 +134,13 @@ class OrgModel(Base):
 
     def as_dataclass(self):
         # Base Org without permissions/roles (filled by data accessors)
-        org = Org(display_name=self.display_name)
-        org.uuid = UUID(bytes=self.uuid)
-        return org
+        return _LegacyOrg(
+            uuid=UUID(bytes=self.uuid),
+            display_name=self.display_name,
+        )
 
     @staticmethod
-    def from_dataclass(org: Org):
+    def from_dataclass(org: _LegacyOrg):
         return OrgModel(uuid=org.uuid.bytes, display_name=org.display_name)
 
 
@@ -388,7 +395,7 @@ class DB:
             result = await session.execute(select(PermissionModel))
             return [p.as_dataclass() for p in result.scalars().all()]
 
-    async def list_organizations(self) -> list[Org]:
+    async def list_organizations(self) -> list[_LegacyOrg]:
         async with self.session() as session:
             # Load all orgs
             orgs_result = await session.execute(select(OrgModel))
@@ -415,13 +422,13 @@ class DB:
                 perms_by_role.setdefault(rp.role_uuid, []).append(rp.permission_id)
 
             # Build org dataclasses with roles and permission IDs
-            roles_by_org: dict[bytes, list[Role]] = {}
+            roles_by_org: dict[bytes, list[_LegacyRole]] = {}
             for rm in role_models:
                 r_dc = rm.as_dataclass()
                 r_dc.permissions = perms_by_role.get(rm.uuid, [])
                 roles_by_org.setdefault(rm.org_uuid, []).append(r_dc)
 
-            orgs: list[Org] = []
+            orgs: list[_LegacyOrg] = []
             for om in org_models:
                 o_dc = om.as_dataclass()
                 o_dc.permissions = perms_by_org.get(om.uuid, [])

tests/conftest.py

@@ -28,10 +28,7 @@ from paskia.db import (
     Permission,
     Role,
     User,
-    add_permission_to_org,
     create_credential,
-    create_org,
-    create_permission,
     create_reset_token,
     create_role,
     create_session,
@@ -55,7 +52,13 @@ def event_loop():
 
 @pytest_asyncio.fixture(scope="function")
 async def test_db() -> AsyncGenerator[DB, None]:
-    """Create an in-memory JSON database for testing."""
+    """Create an in-memory JSON database for testing.
+
+    Uses bootstrap() to properly initialize the database with:
+    - auth:admin and auth:org:admin permissions
+    - A default organization with Administration role
+    - An admin user with the Administration role
+    """
 
     with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=True) as f:
         db = DB()
@@ -64,6 +67,11 @@ async def test_db() -> AsyncGenerator[DB, None]:
         await store.load()
         ops_db._db = db
         ops_db._store = store
+        # Bootstrap creates the initial permissions, org, role, and admin user
+        ops_db.bootstrap(
+            org_name="Test Organization",
+            admin_name="Test Admin",
+        )
         yield db
         ops_db._db = None
         ops_db._store = None
@@ -82,49 +90,30 @@ async def passkey_instance() -> Passkey:
     paskia_globals.passkey._instance = None
 
 
-@pytest_asyncio.fixture(scope="function")
-async def test_org(test_db: DB, admin_permission: Permission) -> Org:
-    """Create a test organization with admin permission."""
-    org = Org.create(display_name="Test Organization")
-    create_org(org)
-    # Grant admin permission to this org
-    add_permission_to_org(org.uuid, admin_permission.uuid)
-    return org
-
-
 @pytest_asyncio.fixture(scope="function")
 async def admin_permission(test_db: DB) -> Permission:
-    """Create the auth:admin permission."""
-    perm = Permission.create(scope="auth:admin", display_name="Master Admin")
-    create_permission(perm)
-    return perm
+    """Get the auth:admin permission created by bootstrap."""
+    return next(p for p in test_db.permissions.values() if p.scope == "auth:admin")
 
 
 @pytest_asyncio.fixture(scope="function")
-async def org_admin_permission(test_db: DB, test_org: Org) -> Permission:
-    """Create the auth:org:admin permission."""
-    perm = Permission.create(scope="auth:org:admin", display_name="Organization Admin")
-    create_permission(perm)
-    # Make it grantable by the org
-    add_permission_to_org(test_org.uuid, perm.uuid)
-    return perm
+async def org_admin_permission(test_db: DB) -> Permission:
+    """Get the auth:org:admin permission created by bootstrap."""
+    return next(p for p in test_db.permissions.values() if p.scope == "auth:org:admin")
 
 
 @pytest_asyncio.fixture(scope="function")
-async def test_role(
-    test_db: DB,
-    test_org: Org,
-    admin_permission: Permission,
-    org_admin_permission: Permission,
-) -> Role:
-    """Create a test role with admin permission."""
-    role = Role.create(
-        org=test_org.uuid,
-        display_name="Test Admin Role",
-        permissions={admin_permission.uuid, org_admin_permission.uuid},
-    )
-    create_role(role)
-    return role
+async def test_org(test_db: DB) -> Org:
+    """Get the test organization created by bootstrap."""
+    # Bootstrap creates exactly one org
+    return next(iter(test_db.orgs.values()))
+
+
+@pytest_asyncio.fixture(scope="function")
+async def test_role(test_db: DB) -> Role:
+    """Get the Administration role created by bootstrap."""
+    # Bootstrap creates exactly one role (Administration)
+    return next(iter(test_db.roles.values()))
 
 
 @pytest_asyncio.fixture(scope="function")
@@ -139,14 +128,10 @@ async def user_role(test_db: DB, test_org: Org) -> Role:
 
 
 @pytest_asyncio.fixture(scope="function")
-async def test_user(test_db: DB, test_role: Role) -> User:
-    """Create a test user with admin role."""
-    user = User.create(
-        display_name="Test Admin",
-        role=test_role.uuid,
-    )
-    create_user(user)
-    return user
+async def test_user(test_db: DB) -> User:
+    """Get the admin user created by bootstrap."""
+    # Bootstrap creates exactly one user (admin)
+    return next(iter(test_db.users.values()))
 
 
 @pytest_asyncio.fixture(scope="function")