gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into copilot/implement-log-off-everywhere-functionality

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-03-17 12:12:50 +01:00
committed by GitHub
parent 581adf0e26 b8075b6821
commit 8905031d16
31 changed files with 2271 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_check_alembic_migrations.py
+199
View File
@@ -0,0 +1,199 @@
"""Tests for scripts/check_alembic_migrations.py."""
# The script lives outside of the ``app`` package, so we import it by path.
import importlib.util
import textwrap
from pathlib import Path
import pytest
_SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "check_alembic_migrations.py"
_spec = importlib.util.spec_from_file_location("check_alembic_migrations", _SCRIPT)
assert _spec and _spec.loader
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod) # type: ignore[union-attr]
check_migrations = _mod.check_migrations
main = _mod.main
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _write_migration(
directory: Path, filename: str, revision: str, down_revision: str | tuple[str, ...] | None
) -> Path:
"""Helper to create a minimal migration file."""
if down_revision is None:
down_rev_str = "None"
elif isinstance(down_revision, tuple):
down_rev_str = repr(down_revision)
else:
down_rev_str = f'"{down_revision}"'
content = textwrap.dedent(f'''\
"""Test migration."""
from typing import Union
revision: str = "{revision}"
down_revision: Union[str, None] = {down_rev_str}
depends_on: Union[str, None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
''')
path = directory / filename
path.write_text(content)
return path
@pytest.fixture
def versions_dir(tmp_path: Path) -> Path:
"""Return a temporary versions directory."""
d = tmp_path / "versions"
d.mkdir()
return d
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestCheckMigrations:
"""Tests for the check_migrations function."""
def test_valid_linear_chain(self, versions_dir: Path) -> None:
"""A simple linear chain should pass with no errors."""
_write_migration(versions_dir, "001_initial.py", "001_initial", None)
_write_migration(versions_dir, "002_add_col.py", "002_add_col", "001_initial")
_write_migration(versions_dir, "003_add_table.py", "003_add_table", "002_add_col")
errors = check_migrations(versions_dir)
assert errors == []
def test_valid_merge_migration(self, versions_dir: Path) -> None:
"""A chain with a merge point should pass."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
_write_migration(versions_dir, "002_a.py", "002_a", "001_base")
_write_migration(versions_dir, "002_b.py", "002_b", "001_base")
# Merge file with tuple down_revision
content = textwrap.dedent('''\
"""Merge."""
from typing import Union
revision: str = "003_merge"
down_revision: Union[str, tuple] = ("002_a", "002_b")
depends_on: Union[str, None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
''')
(versions_dir / "003_merge.py").write_text(content)
errors = check_migrations(versions_dir)
assert errors == []
def test_multiple_heads_detected(self, versions_dir: Path) -> None:
"""Two unmerged branches should report multiple heads."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
_write_migration(versions_dir, "002_a.py", "002_a", "001_base")
_write_migration(versions_dir, "002_b.py", "002_b", "001_base")
errors = check_migrations(versions_dir)
assert len(errors) == 1
assert "Multiple migration heads" in errors[0]
assert "002_a" in errors[0]
assert "002_b" in errors[0]
def test_broken_down_revision(self, versions_dir: Path) -> None:
"""A migration pointing to a non-existent parent should be flagged."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
_write_migration(versions_dir, "002_orphan.py", "002_orphan", "NONEXISTENT")
errors = check_migrations(versions_dir)
assert any("Broken chain" in e for e in errors)
assert any("NONEXISTENT" in e for e in errors)
def test_duplicate_revision(self, versions_dir: Path) -> None:
"""Two files declaring the same revision should be flagged."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
_write_migration(versions_dir, "002_first.py", "002_dup", "001_base")
_write_migration(versions_dir, "002_second.py", "002_dup", "001_base")
errors = check_migrations(versions_dir)
assert any("Duplicate revision" in e for e in errors)
def test_filename_mismatch(self, versions_dir: Path) -> None:
"""A file whose revision doesn't match its filename should be flagged."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
# filename stem is "002_wrong_name" but revision says "002_correct_name"
_write_migration(versions_dir, "002_wrong_name.py", "002_correct_name", "001_base")
errors = check_migrations(versions_dir)
assert any("Filename mismatch" in e for e in errors)
def test_empty_directory(self, versions_dir: Path) -> None:
"""An empty versions directory should report an error."""
errors = check_migrations(versions_dir)
assert len(errors) == 1
assert "No migration files found" in errors[0]
def test_init_py_is_skipped(self, versions_dir: Path) -> None:
"""__init__.py files should be ignored."""
(versions_dir / "__init__.py").write_text("")
_write_migration(versions_dir, "001_base.py", "001_base", None)
errors = check_migrations(versions_dir)
assert errors == []
def test_non_migration_file_skipped(self, versions_dir: Path) -> None:
"""A .py file without a revision variable should be silently skipped."""
(versions_dir / "helper.py").write_text("# just a helper\nx = 1\n")
_write_migration(versions_dir, "001_base.py", "001_base", None)
errors = check_migrations(versions_dir)
assert errors == []
@pytest.mark.unit
class TestMainCLI:
"""Tests for the CLI entry-point."""
def test_success_returns_zero(self, versions_dir: Path) -> None:
"""Valid chain should exit 0."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
rc = main(["--versions-dir", str(versions_dir)])
assert rc == 0
def test_failure_returns_one(self, versions_dir: Path) -> None:
"""Invalid chain should exit 1."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
_write_migration(versions_dir, "002_a.py", "002_a", "001_base")
_write_migration(versions_dir, "002_b.py", "002_b", "001_base")
rc = main(["--versions-dir", str(versions_dir)])
assert rc == 1
def test_missing_directory_returns_two(self, tmp_path: Path) -> None:
"""Non-existent versions directory should exit 2."""
rc = main(["--versions-dir", str(tmp_path / "does_not_exist")])
assert rc == 2
def test_verbose_flag(self, versions_dir: Path) -> None:
"""The --verbose flag should not crash."""
_write_migration(versions_dir, "001_base.py", "001_base", None)
rc = main(["--versions-dir", str(versions_dir), "--verbose"])
assert rc == 0
def test_real_migrations(self) -> None:
"""Smoke test against the actual project migrations."""
real_dir = Path(__file__).resolve().parent.parent / "migrations" / "versions"
if not real_dir.is_dir():
pytest.skip("migrations/versions directory not found in working tree")
rc = main(["--versions-dir", str(real_dir)])
assert rc == 0
tests/test_system_reset.py
+393
View File
@@ -0,0 +1,393 @@
"""Tests for the system reset feature (app/api/system_reset.py, app/utils/system_reset.py, app/views/system_reset.py)."""
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.database import Base
from app.models import (
DocumentMetadata,
FileProcessingStep,
FileRecord,
ProcessingLog,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def reset_workdir():
"""Create a temporary workdir populated with sample user data."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create data subdirectories with dummy files
for subdir in ("original", "processed", "tmp", "pdfa", "backups"):
d = Path(tmpdir) / subdir
d.mkdir()
(d / "sample.pdf").write_bytes(b"%PDF-1.4 fake")
# Create cache files
for cache in ("watch_folder_processed.json", "ftp_ingest_processed.json"):
(Path(tmpdir) / cache).write_text("{}")
# Create a per-user watch folder cache
(Path(tmpdir) / "user_wf_42.json").write_text("{}")
# Create a loose PDF in workdir root
(Path(tmpdir) / "abc123.pdf").write_bytes(b"%PDF-1.4 loose")
yield tmpdir
@pytest.fixture
def reset_db_session():
"""Fresh in-memory database with sample user data rows."""
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(bind=engine)
Session = sessionmaker(bind=engine)
session = Session()
# Seed with sample data
fr = FileRecord(
filehash="abc123",
original_filename="test.pdf",
local_filename="uuid.pdf",
file_size=1024,
mime_type="application/pdf",
)
session.add(fr)
session.flush()
session.add(ProcessingLog(file_id=fr.id, task_id="t1", step_name="hash_file", status="success"))
session.add(FileProcessingStep(file_id=fr.id, step_name="hash_file", status="success"))
session.add(DocumentMetadata(filename="test.pdf", sender="Alice", recipient="Bob"))
session.commit()
yield session
session.close()
Base.metadata.drop_all(bind=engine)
# ---------------------------------------------------------------------------
# Unit tests for app/utils/system_reset.py
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestWipeWorkdirData:
"""Tests for _wipe_workdir_data()."""
def test_removes_data_subdirs(self, reset_workdir):
from app.utils.system_reset import _wipe_workdir_data
result = _wipe_workdir_data(reset_workdir)
# All data subdirectories should be gone
for subdir in ("original", "processed", "tmp", "pdfa", "backups"):
assert not (Path(reset_workdir) / subdir).exists()
assert result["deleted_dirs"] == 5
def test_removes_cache_files(self, reset_workdir):
from app.utils.system_reset import _wipe_workdir_data
result = _wipe_workdir_data(reset_workdir)
assert not (Path(reset_workdir) / "watch_folder_processed.json").exists()
assert not (Path(reset_workdir) / "ftp_ingest_processed.json").exists()
assert not (Path(reset_workdir) / "user_wf_42.json").exists()
assert result["deleted_files"] >= 3
def test_removes_loose_document_files(self, reset_workdir):
from app.utils.system_reset import _wipe_workdir_data
_wipe_workdir_data(reset_workdir)
assert not (Path(reset_workdir) / "abc123.pdf").exists()
def test_preserves_workdir_directory(self, reset_workdir):
from app.utils.system_reset import _wipe_workdir_data
_wipe_workdir_data(reset_workdir)
assert Path(reset_workdir).is_dir()
def test_handles_empty_workdir(self):
"""No errors when workdir has no data dirs or caches."""
from app.utils.system_reset import _wipe_workdir_data
with tempfile.TemporaryDirectory() as empty_dir:
result = _wipe_workdir_data(empty_dir)
assert result["deleted_dirs"] == 0
assert result["deleted_files"] == 0
@pytest.mark.unit
class TestWipeDatabase:
"""Tests for _wipe_database()."""
def test_deletes_all_user_data(self, reset_db_session):
from app.utils.system_reset import _wipe_database
result = _wipe_database(reset_db_session)
assert result.get("files", 0) >= 1
assert result.get("processing_logs", 0) >= 1
assert result.get("file_processing_steps", 0) >= 1
assert result.get("document_metadata", 0) >= 1
def test_tables_are_empty_after_wipe(self, reset_db_session):
from app.utils.system_reset import _wipe_database
_wipe_database(reset_db_session)
assert reset_db_session.query(FileRecord).count() == 0
assert reset_db_session.query(ProcessingLog).count() == 0
assert reset_db_session.query(FileProcessingStep).count() == 0
assert reset_db_session.query(DocumentMetadata).count() == 0
@pytest.mark.unit
class TestPerformFullReset:
"""Tests for perform_full_reset()."""
def test_wipes_db_and_filesystem(self, reset_db_session, reset_workdir):
from app.utils.system_reset import perform_full_reset
with patch("app.utils.system_reset.settings") as mock_settings:
mock_settings.workdir = reset_workdir
result = perform_full_reset(reset_db_session)
assert "database" in result
assert "filesystem" in result
assert reset_db_session.query(FileRecord).count() == 0
assert not (Path(reset_workdir) / "original").exists()
@pytest.mark.unit
class TestPerformResetAndReimport:
"""Tests for perform_reset_and_reimport()."""
def test_copies_originals_to_reimport_then_wipes(self, reset_db_session, reset_workdir):
from app.utils.system_reset import perform_reset_and_reimport
with patch("app.utils.system_reset.settings") as mock_settings:
mock_settings.workdir = reset_workdir
mock_settings.watch_folders = ""
mock_settings.watch_folder_delete_after_process = False
result = perform_reset_and_reimport(reset_db_session)
reimport_dir = Path(reset_workdir) / "reimport"
assert reimport_dir.is_dir()
assert result["reimport"]["files_moved"] >= 1
# DB should be wiped
assert reset_db_session.query(FileRecord).count() == 0
# Reimport folder should contain the original file
reimport_files = list(reimport_dir.iterdir())
assert len(reimport_files) >= 1
def test_configures_watch_folder(self, reset_db_session, reset_workdir):
from app.utils.system_reset import perform_reset_and_reimport
with patch("app.utils.system_reset.settings") as mock_settings:
mock_settings.workdir = reset_workdir
mock_settings.watch_folders = "/some/other/folder"
mock_settings.watch_folder_delete_after_process = False
perform_reset_and_reimport(reset_db_session)
reimport_path = str(Path(reset_workdir) / "reimport")
# watch_folders should now include the reimport path
assert reimport_path in mock_settings.watch_folders
@pytest.mark.unit
class TestStartupReset:
"""Tests for perform_startup_reset()."""
def test_startup_reset_calls_full_reset(self):
from app.utils.system_reset import perform_startup_reset
with patch("app.utils.system_reset.perform_full_reset") as mock_reset:
with patch("app.database.SessionLocal") as mock_sl:
mock_db = mock_sl.return_value
perform_startup_reset()
mock_reset.assert_called_once_with(mock_db)
mock_db.close.assert_called_once()
def test_startup_reset_handles_errors(self):
from app.utils.system_reset import perform_startup_reset
with patch("app.utils.system_reset.perform_full_reset", side_effect=RuntimeError("boom")):
with patch("app.database.SessionLocal") as mock_sl:
mock_db = mock_sl.return_value
# Should not raise
perform_startup_reset()
mock_db.rollback.assert_called_once()
mock_db.close.assert_called_once()
# ---------------------------------------------------------------------------
# Integration tests for API endpoints
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestSystemResetApi:
"""Tests for the /api/admin/system-reset/ endpoints."""
def test_full_reset_requires_admin(self, client):
"""Non-admin users get 403."""
response = client.post(
"/api/admin/system-reset/full",
json={"confirmation": "DELETE"},
)
assert response.status_code == 403
def test_full_reset_requires_feature_flag(self, client):
"""Returns 404 when ENABLE_FACTORY_RESET is false."""
from app.api.system_reset import _require_admin
client.app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
try:
with patch("app.api.system_reset.settings") as mock_s:
mock_s.enable_factory_reset = False
response = client.post(
"/api/admin/system-reset/full",
json={"confirmation": "DELETE"},
)
finally:
client.app.dependency_overrides.pop(_require_admin, None)
assert response.status_code == 404
def test_full_reset_requires_confirmation(self, client):
"""Wrong confirmation string gets 400."""
from app.api.system_reset import _require_admin
client.app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
try:
with patch("app.api.system_reset.settings") as mock_s:
mock_s.enable_factory_reset = True
response = client.post(
"/api/admin/system-reset/full",
json={"confirmation": "WRONG"},
)
finally:
client.app.dependency_overrides.pop(_require_admin, None)
assert response.status_code == 400
def test_reimport_requires_confirmation(self, client):
"""Wrong confirmation string gets 400."""
from app.api.system_reset import _require_admin
client.app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
try:
with patch("app.api.system_reset.settings") as mock_s:
mock_s.enable_factory_reset = True
response = client.post(
"/api/admin/system-reset/reimport",
json={"confirmation": "WRONG"},
)
finally:
client.app.dependency_overrides.pop(_require_admin, None)
assert response.status_code == 400
def test_status_endpoint(self, client):
"""The status endpoint returns feature-flag state."""
from app.api.system_reset import _require_admin
client.app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
try:
response = client.get("/api/admin/system-reset/status")
finally:
client.app.dependency_overrides.pop(_require_admin, None)
assert response.status_code == 200
data = response.json()
assert "enabled" in data
assert "factory_reset_on_startup" in data
def test_full_reset_success(self, client):
"""Full reset succeeds with correct confirmation and feature flag."""
from app.api.system_reset import _require_admin
client.app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
try:
with patch("app.api.system_reset.settings") as mock_s:
mock_s.enable_factory_reset = True
with patch(
"app.utils.system_reset.perform_full_reset", return_value={"database": {}, "filesystem": {}}
):
response = client.post(
"/api/admin/system-reset/full",
json={"confirmation": "DELETE"},
)
finally:
client.app.dependency_overrides.pop(_require_admin, None)
assert response.status_code == 200
assert response.json()["status"] == "ok"
def test_reimport_success(self, client):
"""Reimport succeeds with correct confirmation."""
from app.api.system_reset import _require_admin
client.app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
try:
with patch("app.api.system_reset.settings") as mock_s:
mock_s.enable_factory_reset = True
with patch(
"app.utils.system_reset.perform_reset_and_reimport",
return_value={"database": {}, "filesystem": {}, "reimport": {"files_moved": 3}},
):
response = client.post(
"/api/admin/system-reset/reimport",
json={"confirmation": "REIMPORT"},
)
finally:
client.app.dependency_overrides.pop(_require_admin, None)
assert response.status_code == 200
assert response.json()["status"] == "ok"
# ---------------------------------------------------------------------------
# Integration tests for the view
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestSystemResetView:
"""Tests for the /admin/system-reset view."""
def test_view_redirects_when_disabled(self, client):
"""When ENABLE_FACTORY_RESET=False, accessing the page redirects away."""
with client:
client.cookies.set("session", "test")
with patch("app.views.system_reset.settings") as mock_s:
mock_s.enable_factory_reset = False
response = client.get("/admin/system-reset", follow_redirects=False)
# Redirect to /settings (302) when disabled, or to login (302/307) when unauthenticated
assert response.status_code in (302, 307)
def test_view_requires_auth(self, client):
"""Unauthenticated users are redirected away from the page."""
with patch("app.views.system_reset.settings") as mock_s:
mock_s.enable_factory_reset = True
mock_s.factory_reset_on_startup = False
response = client.get("/admin/system-reset", follow_redirects=False)
# Should redirect to login since there's no active session
assert response.status_code in (302, 307)