Files
2026-07-13 13:12:33 +08:00

70 lines
2.4 KiB
Python

"""Migration-directory resolution: unusable env overrides must be loud.
``OPENSQUILLA_MIGRATIONS_DIR`` pointing at a missing directory (or one with
no ``V*.py`` files) used to fall through silently to a different migration
set — the operator believes a pinned set is in effect while another one
runs. The fallback behavior stays (boot should not die on a typo), but it
must be logged.
"""
from __future__ import annotations
from pathlib import Path
import pytest
import structlog
from opensquilla.gateway.boot import _resolve_migrations_dir
def _env_ignored_events(captured: list[dict]) -> list[dict]:
return [
entry
for entry in captured
if entry.get("event") == "resolve_migrations_dir.env_override_ignored"
]
def test_missing_env_dir_warns_and_falls_through(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
missing = tmp_path / "does-not-exist"
monkeypatch.setenv("OPENSQUILLA_MIGRATIONS_DIR", str(missing))
with structlog.testing.capture_logs() as captured:
resolved = _resolve_migrations_dir()
assert resolved != missing
assert any(resolved.glob("V*.py")) # fell through to a usable set
events = _env_ignored_events(captured)
assert len(events) == 1
assert events[0]["path"] == str(missing)
assert events[0]["reason"] == "directory does not exist"
assert events[0]["log_level"] == "warning"
def test_empty_env_dir_warns_with_no_migrations_reason(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
empty = tmp_path / "empty-migrations"
empty.mkdir()
monkeypatch.setenv("OPENSQUILLA_MIGRATIONS_DIR", str(empty))
with structlog.testing.capture_logs() as captured:
resolved = _resolve_migrations_dir()
assert resolved != empty
events = _env_ignored_events(captured)
assert len(events) == 1
assert events[0]["path"] == str(empty)
assert events[0]["reason"] == "no V*.py migration files found"
def test_usable_env_dir_wins_without_warning(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
pinned = tmp_path / "pinned-migrations"
pinned.mkdir()
(pinned / "V001__synthetic.py").write_text("steps = []\n")
monkeypatch.setenv("OPENSQUILLA_MIGRATIONS_DIR", str(pinned))
with structlog.testing.capture_logs() as captured:
resolved = _resolve_migrations_dir()
assert resolved == pinned
assert _env_ignored_events(captured) == []