feat: Projektgerüst, Datenmodell und initiale Migration

Schritt 1: src-Layout mit FastAPI-Stack, SQLAlchemy-2.x-Modell inkl.
Buchführung für Kopiermodus und Quarantäne, Alembic-Migration 0001,
strukturierte Logs, CLI (migrate/version), Changelog und Devlog.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,14 @@
+.venv/
+__pycache__/
+*.py[cod]
+*.egg-info/
+build/
+dist/
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+*.db
+*.db-journal
+*.db-wal
+*.db-shm
+/data/

CHANGELOG.md

@@ -0,0 +1,22 @@
+# Changelog
+
+Alle nennenswerten Änderungen an NetMover stehen hier.
+
+Format nach [Keep a Changelog](https://keepachangelog.com/de/1.1.0/),
+Versionierung nach [Semantic Versioning](https://semver.org/lang/de/).
+Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
+
+## [Unreleased]
+
+### Hinzugefügt
+- Projektgerüst (Python ≥ 3.12, FastAPI-Stack, src-Layout, ruff, mypy strict, pytest).
+- Datenmodell mit SQLAlchemy 2.x: `credentials`, `mounts`, `rules`, `runs`,
+  `run_items`, `copied_files`, `file_failures`, `settings`.
+- Alembic-Migration `0001` (initiales Schema); NetMover migriert beim Start selbst.
+- Zeitzonenbewusste Zeitstempel (UTC) auch auf SQLite.
+- Strukturierte Logs (logfmt oder JSON, ein Ereignis pro Zeile).
+- CLI `netmover migrate` / `netmover version`.
+
+### Behoben
+- Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
+  Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.

DEVLOG.md

@@ -0,0 +1,42 @@
+# Devlog
+
+Arbeitsprotokoll mit Zeiterfassung. Ein Eintrag pro Arbeitsblock, neueste oben.
+Zeiten in Ortszeit (Europe/Berlin).
+
+## Zeitübersicht
+
+| Datum      | Von   | Bis   | Dauer | Schritt | Thema                                  |
+|------------|-------|-------|-------|---------|----------------------------------------|
+| 2026-10-10 | 17:21 | 17:27 | 0:06  | 1       | Gerüst, Datenmodell, Migrationen       |
+| **Summe**  |       |       | **0:06** |      |                                        |
+
+---
+
+## 2026-10-10 · 17:21–17:27 · Schritt 1: Gerüst, Datenmodell, Migrationen
+
+**Erledigt**
+- Repo angelegt, venv (lokal Python 3.14; Code zielt auf ≥ 3.12, Debian 13 liefert 3.13).
+  `python3-venv` fehlt auf der Entwicklungsmaschine → venv mit `--without-pip` + `get-pip`.
+- `pyproject.toml` (hatchling, src-Layout), ruff + mypy strict.
+- Datenmodell nach Spezifikation plus Ergänzungen (siehe Docstring `models.py`):
+  `credentials.domain`, `mounts.state_since/write_ok/root_squash/last_test_*`,
+  `rules.op_timeout_seconds/quarantine_after`, `runs.status/trigger/dry_run`,
+  Tabelle `file_failures`, `copied_files.mtime_ns` statt Float-`mtime`.
+- Alembic-Migration 0001; Migrationen liegen im Paket, damit sie mit installiert werden.
+- 10 Tests: Schema = Modell (kein Drift), Down-/Upgrade, Defaults, UTC-Roundtrip,
+  RESTRICT bei benutzten Mounts/Zugangsdaten, CASCADE bei Regeln, Eindeutigkeit
+  in `copied_files`, CHECK-Constraints der Enums.
+
+**Entscheidungen**
+- Enums als `VARCHAR` + `CHECK` (kein natives Enum; SQLite kann das ohnehin nicht).
+- `UTCDateTime` lehnt naive `datetime` ab, statt sie still als UTC zu interpretieren.
+- SQLite mit WAL, `foreign_keys=ON`, `busy_timeout=5000` (Scheduler + Web + Worker-Threads).
+
+**Stolpersteine**
+- Alembic-Autogenerate erzeugt CHECK-Constraints bei `Enum(create_constraint=True)`
+  dreifach → explizite `CheckConstraint`-Zeilen aus der Migration entfernt.
+- Test deckte auf: ORM setzt ohne `passive_deletes` `mounts.cred_id` auf NULL,
+  statt das `RESTRICT` greifen zu lassen. Behoben.
+
+**Offen / nächster Schritt**
+- Schritt 2: Mount-Schicht, Validierung, `FakeMountBackend`, Fernet-Verschlüsselung.

README.md

@@ -0,0 +1,31 @@
+# NetMover
+
+Selbst gehosteter Dienst mit Web-Oberfläche, der SMB/CIFS- und NFS-Freigaben
+verwaltet und Dateien zeitgesteuert zwischen ihnen verschiebt.
+
+> Status: in Entwicklung (Schritt 1 von 6). Vollständige Doku folgt mit Schritt 6.
+
+## Entwicklung
+
+```bash
+python3 -m venv .venv            # oder: --without-pip + get-pip.py
+.venv/bin/pip install -e '.[dev]'
+.venv/bin/pytest
+.venv/bin/ruff check . && .venv/bin/ruff format --check . && .venv/bin/mypy
+NETMOVER_DATA_DIR=$PWD/data .venv/bin/netmover migrate
+```
+
+Neue Migration nach Modelländerung:
+
+```bash
+NETMOVER_DATA_DIR=$PWD/data .venv/bin/alembic revision --autogenerate -m "…"
+```
+
+## Arbeitsweise
+
+- Commits nach [Conventional Commits](https://www.conventionalcommits.org/de/)
+  (`feat:`, `fix:`, `docs:`, `test:`, `chore:` …).
+- Jede nutzerrelevante Änderung bekommt einen Eintrag unter `[Unreleased]` in
+  [CHANGELOG.md](CHANGELOG.md). Beim Release wird der Abschnitt versioniert,
+  die Version in `src/netmover/__init__.py` angehoben und ein Tag `vX.Y.Z` gesetzt.
+- Arbeitszeit und Entscheidungen: [DEVLOG.md](DEVLOG.md).

alembic.ini

@@ -0,0 +1,37 @@
+# Nur für die Entwicklung (alembic revision --autogenerate).
+# Im Betrieb migriert NetMover beim Start selbst (netmover.db.upgrade).
+[alembic]
+script_location = src/netmover/migrations
+# sqlalchemy.url bleibt leer → env.py nimmt NETMOVER_DATABASE_URL / NETMOVER_DATA_DIR
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARNING
+handlers = console
+
+[logger_sqlalchemy]
+level = WARNING
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s

pyproject.toml

@@ -0,0 +1,74 @@
+[build-system]
+requires = ["hatchling>=1.25"]
+build-backend = "hatchling.build"
+
+[project]
+name = "netmover"
+dynamic = ["version"]
+description = "Web-GUI für NFS/SMB-Mounts und zeitgesteuertes Verschieben von Dateien"
+readme = "README.md"
+requires-python = ">=3.12"
+license = { text = "MIT" }
+authors = [{ name = "Florian Wilde", email = "fw@wildeflorian.de" }]
+dependencies = [
+    "fastapi>=0.115",
+    "uvicorn[standard]>=0.30",
+    "sqlalchemy>=2.0.30",
+    "alembic>=1.13",
+    "jinja2>=3.1",
+    "python-multipart>=0.0.9",
+    "apscheduler>=3.10,<4",
+    "cryptography>=43",
+    "argon2-cffi>=23.1",
+    "pydantic>=2.7",
+    "pydantic-settings>=2.3",
+    "itsdangerous>=2.2",
+    "httpx>=0.27",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.2",
+    "ruff>=0.6",
+    "mypy>=1.11",
+]
+
+[project.scripts]
+netmover = "netmover.cli:main"
+
+[tool.hatch.version]
+path = "src/netmover/__init__.py"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/netmover"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+addopts = "-ra"
+
+[tool.ruff]
+line-length = 100
+target-version = "py312"
+src = ["src", "tests"]
+
+[tool.ruff.lint]
+select = ["E", "F", "W", "I", "B", "UP", "S", "SIM", "RUF", "PTH"]
+ignore = [
+    "S603", # subprocess ohne shell=True ist genau das gewünschte Muster
+    "RUF001", "RUF002", "RUF003", # deutsche Texte mit Typografie
+]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/**" = ["S101", "S105", "S106", "S108"]
+"src/netmover/migrations/**" = ["E501"]
+
+[tool.mypy]
+python_version = "3.12"
+strict = true
+mypy_path = "src"
+packages = ["netmover"]
+plugins = ["pydantic.mypy"]
+
+[[tool.mypy.overrides]]
+module = ["apscheduler.*"]
+ignore_missing_imports = true

src/netmover/__init__.py

@@ -0,0 +1,3 @@
+"""NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
+
+__version__ = "0.1.0.dev0"

src/netmover/cli.py

@@ -0,0 +1,42 @@
+"""Kommandozeile: ``netmover migrate`` / ``netmover version``.
+
+``netmover serve`` folgt mit dem Web-Teil (Schritt 5).
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from collections.abc import Sequence
+
+from netmover import __version__
+from netmover.config import get_settings
+from netmover.db import upgrade
+from netmover.logging_setup import event, get_logger, setup_logging
+
+log = get_logger("netmover.cli")
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="netmover")
+    sub = parser.add_subparsers(dest="cmd", required=True)
+    sub.add_parser("version", help="Version ausgeben")
+    sub.add_parser("migrate", help="Datenbank auf den neuesten Stand bringen")
+    args = parser.parse_args(argv)
+
+    settings = get_settings()
+    setup_logging(settings.log_level, settings.log_format)
+
+    if args.cmd == "version":
+        print(__version__)
+        return 0
+    if args.cmd == "migrate":
+        settings.data_dir.mkdir(parents=True, exist_ok=True)
+        upgrade(settings.effective_database_url)
+        event(log, "db.migrated", url=settings.effective_database_url)
+        return 0
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/netmover/config.py

@@ -0,0 +1,51 @@
+"""Start-Konfiguration aus Umgebungsvariablen (Präfix ``NETMOVER_``).
+
+Hier steht nur, was vor dem Öffnen der Datenbank feststehen muss. Alles, was der
+Nutzer zur Laufzeit ändert (Aufbewahrung, Webhook, …), liegt in der Tabelle
+``settings``.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Literal
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="NETMOVER_", env_file=None, extra="ignore")
+
+    data_dir: Path = Path("/var/lib/netmover")
+    run_dir: Path = Path("/run/netmover")
+    mount_base: Path = Path("/mnt/netmover")
+
+    host: str = "127.0.0.1"
+    port: int = Field(default=8080, ge=1, le=65535)
+
+    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
+    log_format: Literal["logfmt", "json"] = "logfmt"
+
+    # "fake" nur für Tests und Entwicklung ohne root
+    mount_backend: Literal["linux", "fake"] = "linux"
+
+    database_url: str | None = None
+
+    @property
+    def db_path(self) -> Path:
+        return self.data_dir / "netmover.db"
+
+    @property
+    def secret_key_path(self) -> Path:
+        return self.data_dir / "secret.key"
+
+    @property
+    def effective_database_url(self) -> str:
+        return self.database_url or f"sqlite:///{self.db_path}"
+
+
+@lru_cache(maxsize=1)
+def get_settings() -> Settings:
+    return Settings()

src/netmover/db.py

@@ -0,0 +1,61 @@
+"""Datenbank-Engine, Sessions und Migrationen."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+from importlib.resources import files
+from typing import Any
+
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import Engine, create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+
+def _sqlite_pragmas(dbapi_conn: Any, _record: Any) -> None:
+    cur = dbapi_conn.cursor()
+    cur.execute("PRAGMA foreign_keys=ON")
+    cur.execute("PRAGMA journal_mode=WAL")
+    cur.execute("PRAGMA synchronous=NORMAL")
+    cur.execute("PRAGMA busy_timeout=5000")
+    cur.close()
+
+
+def make_engine(url: str) -> Engine:
+    connect_args: dict[str, Any] = {}
+    if url.startswith("sqlite"):
+        # Worker-Threads der Regelläufe greifen ebenfalls zu.
+        connect_args["check_same_thread"] = False
+    engine = create_engine(url, connect_args=connect_args)
+    if url.startswith("sqlite"):
+        event.listen(engine, "connect", _sqlite_pragmas)
+    return engine
+
+
+def make_session_factory(engine: Engine) -> sessionmaker[Session]:
+    return sessionmaker(bind=engine, expire_on_commit=False)
+
+
+@contextmanager
+def session_scope(factory: sessionmaker[Session]) -> Iterator[Session]:
+    session = factory()
+    try:
+        yield session
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+
+def alembic_config(url: str) -> Config:
+    cfg = Config()
+    cfg.set_main_option("script_location", str(files("netmover") / "migrations"))
+    cfg.set_main_option("sqlalchemy.url", url)
+    return cfg
+
+
+def upgrade(url: str, revision: str = "head") -> None:
+    command.upgrade(alembic_config(url), revision)

src/netmover/logging_setup.py

@@ -0,0 +1,72 @@
+"""Strukturierte Logs: ein Ereignis pro Zeile, wahlweise logfmt oder JSON.
+
+Benutzung::
+
+    log = get_logger(__name__)
+    log.info("mount.ok", extra={"fields": {"mount": "scans", "took_ms": 120}})
+
+oder kürzer über :func:`event`::
+
+    event(log, "mount.ok", mount="scans", took_ms=120)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import sys
+from datetime import UTC, datetime
+from typing import Any
+
+_RESERVED = {"ts", "level", "logger", "event"}
+
+
+def _logfmt_value(value: Any) -> str:
+    text = str(value)
+    if text == "" or any(c in text for c in ' ="\n\t'):
+        text = '"' + text.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") + '"'
+    return text
+
+
+class StructuredFormatter(logging.Formatter):
+    def __init__(self, fmt: str = "logfmt") -> None:
+        super().__init__()
+        self.fmt = fmt
+
+    def format(self, record: logging.LogRecord) -> str:
+        data: dict[str, Any] = {
+            "ts": datetime.fromtimestamp(record.created, tz=UTC).isoformat(timespec="milliseconds"),
+            "level": record.levelname.lower(),
+            "logger": record.name,
+            "event": record.getMessage(),
+        }
+        fields = getattr(record, "fields", None)
+        if isinstance(fields, dict):
+            for key, value in fields.items():
+                data[f"f_{key}" if key in _RESERVED else key] = value
+        if record.exc_info:
+            data["exc"] = self.formatException(record.exc_info)
+        if self.fmt == "json":
+            return json.dumps(data, default=str, ensure_ascii=False)
+        return " ".join(f"{k}={_logfmt_value(v)}" for k, v in data.items())
+
+
+def setup_logging(level: str = "INFO", fmt: str = "logfmt") -> None:
+    handler = logging.StreamHandler(sys.stderr)
+    handler.setFormatter(StructuredFormatter(fmt))
+    root = logging.getLogger()
+    root.handlers[:] = [handler]
+    root.setLevel(level)
+    # Uvicorn bringt eigene Handler mit; wir wollen ein einheitliches Format.
+    for name in ("uvicorn", "uvicorn.error", "uvicorn.access", "apscheduler"):
+        lg = logging.getLogger(name)
+        lg.handlers[:] = []
+        lg.propagate = True
+
+
+def get_logger(name: str) -> logging.Logger:
+    return logging.getLogger(name)
+
+
+def event(logger: logging.Logger, name: str, level: int = logging.INFO, **fields: Any) -> None:
+    logger.log(level, name, extra={"fields": fields})

src/netmover/migrations/env.py

@@ -0,0 +1,52 @@
+"""Alembic-Umgebung für NetMover."""
+
+from __future__ import annotations
+
+from logging.config import fileConfig
+
+from alembic import context
+
+from netmover.config import get_settings
+from netmover.db import make_engine
+from netmover.models import Base
+
+config = context.config
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name, disable_existing_loggers=False)
+
+target_metadata = Base.metadata
+
+
+def _url() -> str:
+    return config.get_main_option("sqlalchemy.url") or get_settings().effective_database_url
+
+
+def run_migrations_offline() -> None:
+    context.configure(
+        url=_url(),
+        target_metadata=target_metadata,
+        literal_binds=True,
+        render_as_batch=True,
+    )
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    engine = make_engine(_url())
+    with engine.connect() as connection:
+        context.configure(
+            connection=connection,
+            target_metadata=target_metadata,
+            render_as_batch=True,
+            compare_type=True,
+        )
+        with context.begin_transaction():
+            context.run_migrations()
+    engine.dispose()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

src/netmover/migrations/script.py.mako

@@ -0,0 +1,27 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+import netmover.models
+${imports if imports else ""}
+
+revision: str = ${repr(up_revision)}
+down_revision: str | None = ${repr(down_revision)}
+branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
+depends_on: str | Sequence[str] | None = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

src/netmover/migrations/versions/0001_initial_schema.py

@@ -0,0 +1,318 @@
+"""initial schema
+
+Revision ID: 0001
+Revises:
+Create Date: 2026-10-10
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+import netmover.models
+
+revision: str = "0001"
+down_revision: str | None = None
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "credentials",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(length=100), nullable=False),
+        sa.Column("username", sa.String(length=255), nullable=False),
+        sa.Column("domain", sa.String(length=255), nullable=True),
+        sa.Column("secret_enc", sa.Text(), nullable=False),
+        sa.Column("created_at", netmover.models.UTCDateTime(), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_credentials")),
+        sa.UniqueConstraint("name", name=op.f("uq_credentials_name")),
+    )
+    op.create_table(
+        "settings",
+        sa.Column("key", sa.String(length=100), nullable=False),
+        sa.Column("value", sa.Text(), nullable=False),
+        sa.PrimaryKeyConstraint("key", name=op.f("pk_settings")),
+    )
+    op.create_table(
+        "mounts",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(length=100), nullable=False),
+        sa.Column("slug", sa.String(length=64), nullable=False),
+        sa.Column(
+            "type",
+            sa.Enum(
+                "cifs",
+                "nfs4",
+                "nfs3",
+                name="mount_type",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column("server", sa.String(length=255), nullable=False),
+        sa.Column("share", sa.String(length=1024), nullable=False),
+        sa.Column("mountpoint", sa.String(length=1024), nullable=False),
+        sa.Column("cred_id", sa.Integer(), nullable=True),
+        sa.Column("uid", sa.Integer(), nullable=True),
+        sa.Column("gid", sa.Integer(), nullable=True),
+        sa.Column("file_mode", sa.String(length=5), nullable=True),
+        sa.Column("dir_mode", sa.String(length=5), nullable=True),
+        sa.Column("extra_options", sa.Text(), nullable=False),
+        sa.Column("auto_mount", sa.Boolean(), nullable=False),
+        sa.Column("enabled", sa.Boolean(), nullable=False),
+        sa.Column(
+            "state",
+            sa.Enum(
+                "unmounted",
+                "mounted",
+                "degraded",
+                "error",
+                name="mount_state",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column("state_since", netmover.models.UTCDateTime(), nullable=True),
+        sa.Column("last_error", sa.Text(), nullable=True),
+        sa.Column("last_mounted_at", netmover.models.UTCDateTime(), nullable=True),
+        sa.Column("last_test_at", netmover.models.UTCDateTime(), nullable=True),
+        sa.Column("last_test_result", sa.Text(), nullable=True),
+        sa.Column("write_ok", sa.Boolean(), nullable=True),
+        sa.Column("root_squash", sa.Boolean(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["cred_id"],
+            ["credentials.id"],
+            name=op.f("fk_mounts_cred_id_credentials"),
+            ondelete="RESTRICT",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_mounts")),
+        sa.UniqueConstraint("mountpoint", name=op.f("uq_mounts_mountpoint")),
+        sa.UniqueConstraint("name", name=op.f("uq_mounts_name")),
+        sa.UniqueConstraint("slug", name=op.f("uq_mounts_slug")),
+    )
+    op.create_table(
+        "rules",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(length=100), nullable=False),
+        sa.Column("enabled", sa.Boolean(), nullable=False),
+        sa.Column("src_mount_id", sa.Integer(), nullable=False),
+        sa.Column("src_subpath", sa.String(length=1024), nullable=False),
+        sa.Column("dst_mount_id", sa.Integer(), nullable=True),
+        sa.Column("dst_path", sa.String(length=1024), nullable=False),
+        sa.Column("pattern", sa.String(length=512), nullable=False),
+        sa.Column(
+            "pattern_type",
+            sa.Enum(
+                "glob",
+                "regex",
+                name="pattern_type",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column("recursive", sa.Boolean(), nullable=False),
+        sa.Column("interval_seconds", sa.Integer(), nullable=False),
+        sa.Column("stability_seconds", sa.Integer(), nullable=False),
+        sa.Column("min_age_seconds", sa.Integer(), nullable=False),
+        sa.Column("max_size_mb", sa.Integer(), nullable=True),
+        sa.Column("op_timeout_seconds", sa.Integer(), nullable=False),
+        sa.Column(
+            "action",
+            sa.Enum(
+                "move",
+                "copy",
+                name="rule_action",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column(
+            "on_collision",
+            sa.Enum(
+                "rename",
+                "skip",
+                "overwrite",
+                name="collision_policy",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column("set_owner", sa.String(length=64), nullable=True),
+        sa.Column("set_mode", sa.String(length=5), nullable=True),
+        sa.Column("dry_run", sa.Boolean(), nullable=False),
+        sa.Column("quarantine_path", sa.String(length=1024), nullable=True),
+        sa.Column("quarantine_after", sa.Integer(), nullable=False),
+        sa.Column("last_run_at", netmover.models.UTCDateTime(), nullable=True),
+        sa.Column("last_success_at", netmover.models.UTCDateTime(), nullable=True),
+        sa.Column("consecutive_failures", sa.Integer(), nullable=False),
+        sa.Column("created_at", netmover.models.UTCDateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["dst_mount_id"],
+            ["mounts.id"],
+            name=op.f("fk_rules_dst_mount_id_mounts"),
+            ondelete="RESTRICT",
+        ),
+        sa.ForeignKeyConstraint(
+            ["src_mount_id"],
+            ["mounts.id"],
+            name=op.f("fk_rules_src_mount_id_mounts"),
+            ondelete="RESTRICT",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_rules")),
+        sa.UniqueConstraint("name", name=op.f("uq_rules_name")),
+    )
+    op.create_table(
+        "copied_files",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("rule_id", sa.Integer(), nullable=False),
+        sa.Column("src_relpath", sa.String(length=2048), nullable=False),
+        sa.Column("size", sa.BigInteger(), nullable=False),
+        sa.Column("mtime_ns", sa.BigInteger(), nullable=False),
+        sa.Column("dst_path", sa.String(length=2048), nullable=False),
+        sa.Column("copied_at", netmover.models.UTCDateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["rule_id"],
+            ["rules.id"],
+            name=op.f("fk_copied_files_rule_id_rules"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_copied_files")),
+        sa.UniqueConstraint(
+            "rule_id",
+            "src_relpath",
+            "size",
+            "mtime_ns",
+            name=op.f("uq_copied_files_rule_id_src_relpath_size_mtime_ns"),
+        ),
+    )
+    op.create_table(
+        "file_failures",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("rule_id", sa.Integer(), nullable=False),
+        sa.Column("src_relpath", sa.String(length=2048), nullable=False),
+        sa.Column("size", sa.BigInteger(), nullable=False),
+        sa.Column("mtime_ns", sa.BigInteger(), nullable=False),
+        sa.Column("attempts", sa.Integer(), nullable=False),
+        sa.Column("last_error", sa.Text(), nullable=True),
+        sa.Column("last_attempt_at", netmover.models.UTCDateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["rule_id"],
+            ["rules.id"],
+            name=op.f("fk_file_failures_rule_id_rules"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_file_failures")),
+        sa.UniqueConstraint(
+            "rule_id", "src_relpath", name=op.f("uq_file_failures_rule_id_src_relpath")
+        ),
+    )
+    op.create_table(
+        "runs",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("rule_id", sa.Integer(), nullable=False),
+        sa.Column("started_at", netmover.models.UTCDateTime(), nullable=False),
+        sa.Column("finished_at", netmover.models.UTCDateTime(), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum(
+                "running",
+                "ok",
+                "partial",
+                "error",
+                "skipped",
+                name="run_status",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column(
+            "trigger",
+            sa.Enum(
+                "schedule",
+                "manual",
+                name="run_trigger",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column("dry_run", sa.Boolean(), nullable=False),
+        sa.Column("seen", sa.Integer(), nullable=False),
+        sa.Column("moved", sa.Integer(), nullable=False),
+        sa.Column("skipped", sa.Integer(), nullable=False),
+        sa.Column("failed", sa.Integer(), nullable=False),
+        sa.Column("message", sa.Text(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["rule_id"], ["rules.id"], name=op.f("fk_runs_rule_id_rules"), ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_runs")),
+    )
+    with op.batch_alter_table("runs", schema=None) as batch_op:
+        batch_op.create_index("ix_runs_rule_started", ["rule_id", "started_at"], unique=False)
+        batch_op.create_index(batch_op.f("ix_runs_started_at"), ["started_at"], unique=False)
+
+    op.create_table(
+        "run_items",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("run_id", sa.Integer(), nullable=False),
+        sa.Column("src", sa.String(length=2048), nullable=False),
+        sa.Column("dst", sa.String(length=2048), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum(
+                "moved",
+                "copied",
+                "skipped",
+                "failed",
+                "quarantined",
+                "planned",
+                name="item_status",
+                native_enum=False,
+                create_constraint=True,
+                length=24,
+            ),
+            nullable=False,
+        ),
+        sa.Column("size", sa.BigInteger(), nullable=True),
+        sa.Column("message", sa.Text(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["run_id"], ["runs.id"], name=op.f("fk_run_items_run_id_runs"), ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_run_items")),
+    )
+    with op.batch_alter_table("run_items", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_run_items_run_id"), ["run_id"], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("run_items", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_run_items_run_id"))
+
+    op.drop_table("run_items")
+    with op.batch_alter_table("runs", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_runs_started_at"))
+        batch_op.drop_index("ix_runs_rule_started")
+
+    op.drop_table("runs")
+    op.drop_table("file_failures")
+    op.drop_table("copied_files")
+    op.drop_table("rules")
+    op.drop_table("mounts")
+    op.drop_table("settings")
+    op.drop_table("credentials")

src/netmover/models.py

@@ -0,0 +1,318 @@
+"""SQLAlchemy-Datenmodell.
+
+Abweichungen bzw. Ergänzungen gegenüber dem Mindestumfang aus der Spezifikation:
+
+* ``credentials.domain`` – Windows-/AD-Freigaben brauchen oft eine Domäne.
+* ``mounts.state_since`` – für „hängt seit …“ / „gestört seit …“ auf der Statusseite.
+* ``mounts.write_ok`` / ``mounts.root_squash`` / ``mounts.last_test_*`` – Ergebnis
+  des letzten Mount-Tests; ``root_squash`` entscheidet, ob ``chown`` auf NFS-Zielen
+  versucht wird (Fallstrick 12).
+* ``rules.op_timeout_seconds`` – Zeitlimit des Worker-Laufs (Fallstrick 13).
+* ``rules.quarantine_after`` + Tabelle ``file_failures`` – Zählung wiederholt
+  scheiternder Dateien für den Quarantäneordner.
+* ``runs.status`` / ``runs.trigger`` / ``runs.dry_run`` – Filter im Protokoll.
+* ``copied_files.mtime_ns`` – ``mtime`` als ganzzahlige Nanosekunden statt Float,
+  damit der Gleichheitsvergleich exakt ist (Fallstrick 11).
+"""
+
+from __future__ import annotations
+
+import enum
+from datetime import UTC, datetime
+
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    DateTime,
+    Enum,
+    ForeignKey,
+    Index,
+    Integer,
+    MetaData,
+    String,
+    Text,
+    TypeDecorator,
+    UniqueConstraint,
+)
+from sqlalchemy.engine import Dialect
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+
+
+def utcnow() -> datetime:
+    return datetime.now(UTC)
+
+
+class UTCDateTime(TypeDecorator[datetime]):
+    """Speichert naive UTC-Zeitstempel (SQLite kennt keine Zeitzonen) und liefert
+    immer zeitzonenbewusste ``datetime``-Objekte zurück."""
+
+    impl = DateTime
+    cache_ok = True
+
+    def process_bind_param(self, value: datetime | None, dialect: Dialect) -> datetime | None:
+        if value is None:
+            return None
+        if value.tzinfo is None:
+            raise ValueError("naive datetime – bitte zeitzonenbewusst (UTC) übergeben")
+        return value.astimezone(UTC).replace(tzinfo=None)
+
+    def process_result_value(self, value: datetime | None, dialect: Dialect) -> datetime | None:
+        if value is None:
+            return None
+        return value.replace(tzinfo=UTC)
+
+
+# Einheitliche Constraint-Namen, damit Alembic-Batch-Migrationen auf SQLite
+# Constraints wiederfinden.
+NAMING_CONVENTION = {
+    "ix": "ix_%(column_0_label)s",
+    "uq": "uq_%(table_name)s_%(column_0_N_name)s",
+    "ck": "ck_%(table_name)s_%(constraint_name)s",
+    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+    "pk": "pk_%(table_name)s",
+}
+
+
+class Base(DeclarativeBase):
+    metadata = MetaData(naming_convention=NAMING_CONVENTION)
+
+
+def _enum(cls: type[enum.Enum], name: str) -> Enum:
+    return Enum(
+        cls,
+        name=name,
+        native_enum=False,
+        create_constraint=True,
+        validate_strings=True,
+        length=24,
+        values_callable=lambda e: [m.value for m in e],
+    )
+
+
+class MountType(enum.StrEnum):
+    CIFS = "cifs"
+    NFS4 = "nfs4"
+    NFS3 = "nfs3"
+
+
+class MountState(enum.StrEnum):
+    UNMOUNTED = "unmounted"
+    MOUNTED = "mounted"
+    DEGRADED = "degraded"  # gestört: Wiederherstellung läuft
+    ERROR = "error"  # Wiederherstellung gescheitert
+
+
+class PatternType(enum.StrEnum):
+    GLOB = "glob"
+    REGEX = "regex"
+
+
+class RuleAction(enum.StrEnum):
+    MOVE = "move"
+    COPY = "copy"
+
+
+class CollisionPolicy(enum.StrEnum):
+    RENAME = "rename"
+    SKIP = "skip"
+    OVERWRITE = "overwrite"
+
+
+class RunStatus(enum.StrEnum):
+    RUNNING = "running"
+    OK = "ok"  # keine Dateifehler
+    PARTIAL = "partial"  # einzelne Dateien gescheitert
+    ERROR = "error"  # Lauf abgebrochen (Mount fehlt, Zeitlimit, …)
+    SKIPPED = "skipped"  # Termin übersprungen, voriger Lauf noch aktiv
+
+
+class RunTrigger(enum.StrEnum):
+    SCHEDULE = "schedule"
+    MANUAL = "manual"
+
+
+class ItemStatus(enum.StrEnum):
+    MOVED = "moved"
+    COPIED = "copied"
+    SKIPPED = "skipped"
+    FAILED = "failed"
+    QUARANTINED = "quarantined"
+    PLANNED = "planned"  # Probelauf
+
+
+class Credential(Base):
+    __tablename__ = "credentials"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    name: Mapped[str] = mapped_column(String(100), unique=True)
+    username: Mapped[str] = mapped_column(String(255))
+    domain: Mapped[str | None] = mapped_column(String(255))
+    secret_enc: Mapped[str] = mapped_column(Text)
+    created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+
+    # passive_deletes="all": das ORM soll cred_id nicht still auf NULL setzen,
+    # sondern das RESTRICT der Datenbank greifen lassen.
+    mounts: Mapped[list[Mount]] = relationship(back_populates="credential", passive_deletes="all")
+
+
+class Mount(Base):
+    __tablename__ = "mounts"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    name: Mapped[str] = mapped_column(String(100), unique=True)
+    slug: Mapped[str] = mapped_column(String(64), unique=True)
+    type: Mapped[MountType] = mapped_column(_enum(MountType, "mount_type"))
+    server: Mapped[str] = mapped_column(String(255))
+    share: Mapped[str] = mapped_column(String(1024))
+    mountpoint: Mapped[str] = mapped_column(String(1024), unique=True)
+    cred_id: Mapped[int | None] = mapped_column(ForeignKey("credentials.id", ondelete="RESTRICT"))
+    uid: Mapped[int | None] = mapped_column(Integer)
+    gid: Mapped[int | None] = mapped_column(Integer)
+    file_mode: Mapped[str | None] = mapped_column(String(5))
+    dir_mode: Mapped[str | None] = mapped_column(String(5))
+    extra_options: Mapped[str] = mapped_column(Text, default="")
+    auto_mount: Mapped[bool] = mapped_column(Boolean, default=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, default=True)
+
+    state: Mapped[MountState] = mapped_column(
+        _enum(MountState, "mount_state"), default=MountState.UNMOUNTED
+    )
+    state_since: Mapped[datetime | None] = mapped_column(UTCDateTime)
+    last_error: Mapped[str | None] = mapped_column(Text)
+    last_mounted_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
+
+    last_test_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
+    last_test_result: Mapped[str | None] = mapped_column(Text)  # JSON
+    write_ok: Mapped[bool | None] = mapped_column(Boolean)
+    root_squash: Mapped[bool | None] = mapped_column(Boolean)
+
+    credential: Mapped[Credential | None] = relationship(back_populates="mounts")
+
+
+class Rule(Base):
+    __tablename__ = "rules"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    name: Mapped[str] = mapped_column(String(100), unique=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, default=True)
+
+    src_mount_id: Mapped[int] = mapped_column(ForeignKey("mounts.id", ondelete="RESTRICT"))
+    src_subpath: Mapped[str] = mapped_column(String(1024), default="")
+    # dst_mount_id NULL → dst_path ist ein absoluter lokaler Pfad
+    dst_mount_id: Mapped[int | None] = mapped_column(ForeignKey("mounts.id", ondelete="RESTRICT"))
+    dst_path: Mapped[str] = mapped_column(String(1024), default="")
+
+    pattern: Mapped[str] = mapped_column(String(512), default="*")
+    pattern_type: Mapped[PatternType] = mapped_column(
+        _enum(PatternType, "pattern_type"), default=PatternType.GLOB
+    )
+    recursive: Mapped[bool] = mapped_column(Boolean, default=False)
+
+    interval_seconds: Mapped[int] = mapped_column(Integer, default=30)
+    stability_seconds: Mapped[int] = mapped_column(Integer, default=5)
+    min_age_seconds: Mapped[int] = mapped_column(Integer, default=0)
+    max_size_mb: Mapped[int | None] = mapped_column(Integer)
+    op_timeout_seconds: Mapped[int] = mapped_column(Integer, default=120)
+
+    action: Mapped[RuleAction] = mapped_column(
+        _enum(RuleAction, "rule_action"), default=RuleAction.MOVE
+    )
+    on_collision: Mapped[CollisionPolicy] = mapped_column(
+        _enum(CollisionPolicy, "collision_policy"), default=CollisionPolicy.RENAME
+    )
+    set_owner: Mapped[str | None] = mapped_column(String(64))  # "uid:gid", NULL = unverändert
+    set_mode: Mapped[str | None] = mapped_column(String(5))  # "0644", NULL = unverändert
+    dry_run: Mapped[bool] = mapped_column(Boolean, default=False)
+
+    quarantine_path: Mapped[str | None] = mapped_column(String(1024))
+    quarantine_after: Mapped[int] = mapped_column(Integer, default=3)
+
+    last_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
+    last_success_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
+    consecutive_failures: Mapped[int] = mapped_column(Integer, default=0)
+    created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+
+    src_mount: Mapped[Mount] = relationship(foreign_keys=[src_mount_id])
+    dst_mount: Mapped[Mount | None] = relationship(foreign_keys=[dst_mount_id])
+    runs: Mapped[list[Run]] = relationship(
+        back_populates="rule", cascade="all, delete-orphan", passive_deletes=True
+    )
+
+
+class Run(Base):
+    __tablename__ = "runs"
+    __table_args__ = (Index("ix_runs_rule_started", "rule_id", "started_at"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE"))
+    started_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, index=True)
+    finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
+    status: Mapped[RunStatus] = mapped_column(
+        _enum(RunStatus, "run_status"), default=RunStatus.RUNNING
+    )
+    trigger: Mapped[RunTrigger] = mapped_column(
+        _enum(RunTrigger, "run_trigger"), default=RunTrigger.SCHEDULE
+    )
+    dry_run: Mapped[bool] = mapped_column(Boolean, default=False)
+    seen: Mapped[int] = mapped_column(Integer, default=0)
+    moved: Mapped[int] = mapped_column(Integer, default=0)
+    skipped: Mapped[int] = mapped_column(Integer, default=0)
+    failed: Mapped[int] = mapped_column(Integer, default=0)
+    message: Mapped[str | None] = mapped_column(Text)
+
+    rule: Mapped[Rule] = relationship(back_populates="runs")
+    items: Mapped[list[RunItem]] = relationship(
+        back_populates="run", cascade="all, delete-orphan", passive_deletes=True
+    )
+
+
+class RunItem(Base):
+    __tablename__ = "run_items"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    run_id: Mapped[int] = mapped_column(ForeignKey("runs.id", ondelete="CASCADE"), index=True)
+    src: Mapped[str] = mapped_column(String(2048))
+    dst: Mapped[str | None] = mapped_column(String(2048))
+    status: Mapped[ItemStatus] = mapped_column(_enum(ItemStatus, "item_status"))
+    size: Mapped[int | None] = mapped_column(BigInteger)
+    message: Mapped[str | None] = mapped_column(Text)
+
+    run: Mapped[Run] = relationship(back_populates="items")
+
+
+class CopiedFile(Base):
+    """Buchführung für Aktion ``copy`` (Fallstrick 11)."""
+
+    __tablename__ = "copied_files"
+    __table_args__ = (UniqueConstraint("rule_id", "src_relpath", "size", "mtime_ns"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE"))
+    src_relpath: Mapped[str] = mapped_column(String(2048))
+    size: Mapped[int] = mapped_column(BigInteger)
+    mtime_ns: Mapped[int] = mapped_column(BigInteger)
+    dst_path: Mapped[str] = mapped_column(String(2048))
+    copied_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+
+
+class FileFailure(Base):
+    """Fehlversuche pro Datei, Grundlage für die Quarantäne."""
+
+    __tablename__ = "file_failures"
+    __table_args__ = (UniqueConstraint("rule_id", "src_relpath"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE"))
+    src_relpath: Mapped[str] = mapped_column(String(2048))
+    size: Mapped[int] = mapped_column(BigInteger)
+    mtime_ns: Mapped[int] = mapped_column(BigInteger)
+    attempts: Mapped[int] = mapped_column(Integer, default=0)
+    last_error: Mapped[str | None] = mapped_column(Text)
+    last_attempt_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
+
+
+class Setting(Base):
+    __tablename__ = "settings"
+
+    key: Mapped[str] = mapped_column(String(100), primary_key=True)
+    value: Mapped[str] = mapped_column(Text)

tests/conftest.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from sqlalchemy.orm import Session
+
+from netmover.db import make_engine, make_session_factory, upgrade
+
+
+@pytest.fixture
+def db_url(tmp_path: Path) -> str:
+    url = f"sqlite:///{tmp_path / 'test.db'}"
+    upgrade(url)
+    return url
+
+
+@pytest.fixture
+def session(db_url: str) -> Iterator[Session]:
+    engine = make_engine(db_url)
+    factory = make_session_factory(engine)
+    with factory() as s:
+        yield s
+    engine.dispose()

tests/test_models.py

@@ -0,0 +1,167 @@
+"""Schritt 1: Schema, Migrationen und Integritätsregeln."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta, timezone
+
+import pytest
+from alembic.autogenerate import compare_metadata
+from alembic.migration import MigrationContext
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError, StatementError
+from sqlalchemy.orm import Session
+
+from netmover.db import make_engine
+from netmover.models import (
+    Base,
+    CopiedFile,
+    Credential,
+    ItemStatus,
+    Mount,
+    MountState,
+    MountType,
+    Rule,
+    Run,
+    RunItem,
+)
+
+
+def _mount(name: str = "scans", **kw: object) -> Mount:
+    defaults: dict[str, object] = {
+        "name": name,
+        "slug": name,
+        "type": MountType.CIFS,
+        "server": "192.168.1.10",
+        "share": "scans",
+        "mountpoint": f"/mnt/netmover/{name}",
+    }
+    defaults.update(kw)
+    return Mount(**defaults)
+
+
+def test_migration_matches_models(db_url: str) -> None:
+    engine = make_engine(db_url)
+    with engine.connect() as conn:
+        ctx = MigrationContext.configure(conn, opts={"compare_type": True})
+        diff = compare_metadata(ctx, Base.metadata)
+    engine.dispose()
+    assert diff == []
+
+
+def test_downgrade_and_upgrade_roundtrip(db_url: str) -> None:
+    from alembic import command
+
+    from netmover.db import alembic_config, upgrade
+
+    command.downgrade(alembic_config(db_url), "base")
+    upgrade(db_url)
+
+
+def test_defaults(session: Session) -> None:
+    m = _mount()
+    session.add(m)
+    session.flush()
+    r = Rule(name="r", src_mount_id=m.id, dst_path="/opt/consume")
+    session.add(r)
+    session.commit()
+    assert m.state is MountState.UNMOUNTED
+    assert m.auto_mount is True
+    assert r.interval_seconds == 30
+    assert r.stability_seconds == 5
+    assert r.op_timeout_seconds == 120
+    assert r.dst_mount_id is None
+
+
+def test_datetime_roundtrip_is_utc_aware(session: Session) -> None:
+    berlin = timezone(timedelta(hours=2))
+    t = datetime(2026, 10, 10, 17, 0, tzinfo=berlin)
+    m = _mount(last_mounted_at=t)
+    session.add(m)
+    session.commit()
+    session.expire_all()
+    loaded = session.scalars(select(Mount)).one()
+    assert loaded.last_mounted_at == t
+    assert loaded.last_mounted_at is not None
+    assert loaded.last_mounted_at.tzinfo is UTC
+
+
+def test_naive_datetime_rejected(session: Session) -> None:
+    session.add(_mount(last_mounted_at=datetime(2026, 1, 1)))
+    with pytest.raises(StatementError):
+        session.commit()
+
+
+def test_mount_in_use_cannot_be_deleted(session: Session) -> None:
+    m = _mount()
+    session.add(m)
+    session.flush()
+    session.add(Rule(name="r", src_mount_id=m.id, dst_path="/x"))
+    session.commit()
+    session.delete(m)
+    with pytest.raises(IntegrityError):
+        session.commit()
+
+
+def test_credential_in_use_cannot_be_deleted(session: Session) -> None:
+    c = Credential(name="nas", username="scan", secret_enc="x")
+    session.add(c)
+    session.flush()
+    session.add(_mount(cred_id=c.id))
+    session.commit()
+    session.delete(c)
+    with pytest.raises(IntegrityError):
+        session.commit()
+
+
+def test_rule_delete_cascades_runs_items_and_bookkeeping(session: Session) -> None:
+    m = _mount()
+    session.add(m)
+    session.flush()
+    r = Rule(name="r", src_mount_id=m.id, dst_path="/x")
+    session.add(r)
+    session.flush()
+    run = Run(rule_id=r.id)
+    session.add(run)
+    session.flush()
+    session.add(RunItem(run_id=run.id, src="a.pdf", status=ItemStatus.MOVED))
+    session.add(
+        CopiedFile(rule_id=r.id, src_relpath="a.pdf", size=1, mtime_ns=1, dst_path="/x/a.pdf")
+    )
+    session.commit()
+
+    session.delete(r)
+    session.commit()
+    assert session.scalars(select(Run)).all() == []
+    assert session.scalars(select(RunItem)).all() == []
+    assert session.scalars(select(CopiedFile)).all() == []
+
+
+def test_copied_files_unique_per_version(session: Session) -> None:
+    m = _mount()
+    session.add(m)
+    session.flush()
+    r = Rule(name="r", src_mount_id=m.id, dst_path="/x")
+    session.add(r)
+    session.flush()
+    session.add(
+        CopiedFile(rule_id=r.id, src_relpath="a.pdf", size=1, mtime_ns=1, dst_path="/x/a.pdf")
+    )
+    # neue Version derselben Datei ist erlaubt
+    session.add(
+        CopiedFile(rule_id=r.id, src_relpath="a.pdf", size=2, mtime_ns=2, dst_path="/x/a-1.pdf")
+    )
+    session.commit()
+    session.add(
+        CopiedFile(rule_id=r.id, src_relpath="a.pdf", size=1, mtime_ns=1, dst_path="/x/a-2.pdf")
+    )
+    with pytest.raises(IntegrityError):
+        session.commit()
+
+
+def test_enum_check_constraint_enforced_in_db(session: Session) -> None:
+    from sqlalchemy import text
+
+    session.add(_mount())
+    session.commit()
+    with pytest.raises(IntegrityError):
+        session.execute(text("UPDATE mounts SET type = 'ftp'"))