Phase 1: Kern mit Plugin-System, Auth/Rollen, Migrations, Plugin-Stubs
- redaktionskern (src/): schlanker Kern — App-Fabrik, Plugin-Loader (Entry-Points + plugins/-Verzeichnis), Migrations-Laufzeit (schema_migrations pro Plugin, SQLite-/Postgres-portabel), Auth mit Argon2id + Session-Cookies, Rollen admin/redakteur/rezensent, Benutzerverwaltung für Admins - plugins/: 8 ladbare Stubs (neuheiten, dedup, planung, archiv, erinnerung, benachrichtigung, audit-log, export) nach Plugin-Vertrag - Frontend: Jinja2 + Tailwind (CDN) + HTMX + Alpine.js, UI deutsch - Tests: 28 pytest-Fälle (Loader, Lifecycle, Entry-Points, Migrations- Idempotenz, Login/Logout, Rollen-Zugriff, Benutzerverwaltung) - Docker/Podman: Compose (Traefik-Labels) + Dockerfile (uv) - README.md mit Setup-Anleitung
This commit is contained in:
78
tests/test_migrations.py
Normal file
78
tests/test_migrations.py
Normal file
@@ -0,0 +1,78 @@
|
||||
"""Tests: Migrations-Laufzeit (Kern-Tabelle, Idempotenz, Plugin-Migration)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from redaktionskern.config import Settings
|
||||
|
||||
|
||||
def test_kern_tabellen_existieren(app):
|
||||
with app.state.engine.connect() as conn:
|
||||
zeilen = conn.execute(text("SELECT name FROM sqlite_master WHERE type='table'"))
|
||||
namen = {zeile[0] for zeile in zeilen}
|
||||
assert "users" in namen
|
||||
assert "schema_migrations" in namen
|
||||
|
||||
|
||||
def test_migration_wird_protokolliert(app):
|
||||
with app.state.engine.connect() as conn:
|
||||
zeilen = conn.execute(
|
||||
text("SELECT plugin, version FROM schema_migrations")
|
||||
).fetchall()
|
||||
protokoll = {(p, v) for p, v in zeilen}
|
||||
assert ("core", "0001_benutzer") in protokoll
|
||||
|
||||
|
||||
def test_plugin_migration_idempotent(tmp_path):
|
||||
"""Ein Plugin mit eigener Migration; zweiter Start darf nichts doppelt tun."""
|
||||
plugin_ordner = tmp_path / "plugins" / "testmig"
|
||||
plugin_ordner.mkdir(parents=True)
|
||||
(plugin_ordner / "__init__.py").write_text(
|
||||
"""
|
||||
from sqlalchemy import text
|
||||
from redaktionskern.contracts import BasePlugin, Migration
|
||||
|
||||
|
||||
def _tabelle(conn):
|
||||
conn.execute(text(
|
||||
"CREATE TABLE IF NOT EXISTS testplugin_dings (id INTEGER PRIMARY KEY)"
|
||||
))
|
||||
|
||||
|
||||
class TestMigPlugin(BasePlugin):
|
||||
name = "testmig"
|
||||
title = "Testmigration"
|
||||
|
||||
def migrations(self):
|
||||
return [Migration(version="0001_tabelle", up=_tabelle)]
|
||||
|
||||
|
||||
plugin = TestMigPlugin()
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
settings = Settings(
|
||||
database_url=f"sqlite:///{tmp_path / 'mig.db'}",
|
||||
session_secret="s",
|
||||
initial_admin_password="x",
|
||||
plugins_dir=tmp_path / "plugins",
|
||||
)
|
||||
|
||||
from redaktionskern.app import create_app
|
||||
|
||||
create_app(settings)
|
||||
app2 = create_app(settings) # zweiter Lauf: darf nicht fehlschlagen
|
||||
|
||||
with app2.state.engine.connect() as conn:
|
||||
tabellen = {
|
||||
zeile[0]
|
||||
for zeile in conn.execute(text("SELECT name FROM sqlite_master WHERE type='table'"))
|
||||
}
|
||||
protokoll = conn.execute(
|
||||
text("SELECT version FROM schema_migrations WHERE plugin='testmig'")
|
||||
).fetchall()
|
||||
assert "testplugin_dings" in tabellen
|
||||
assert [z[0] for z in protokoll] == ["0001_tabelle"]
|
||||
Reference in New Issue
Block a user