curo1305
27fa0d4631
test_storage.py (6 xfail tests, STORE-02):
- test_object_key_schema: regex {user_id}/{doc_id}/{uuid4}{ext}
- test_filename_not_in_object_key: human filename never in MinIO key
- test_storage_backend_abc_methods: incomplete subclass raises TypeError
- test_get_storage_backend_returns_minio: factory returns MinIOBackend
- test_put_object_uses_asyncio_to_thread: SDK call wrapped in to_thread
- test_minio_backend_health_check_returns_bool: True/False on ok/error
test_alembic.py (2 xfail tests, STORE-01 / D-02 / D-03):
- test_migration_creates_all_tables: all 11 v1 tables after upgrade head
- test_documents_user_id_nullable: user_id notnull=0 per D-03
117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
"""
|
|
Wave 0 integration tests for Plan 03 (Alembic migration).
|
|
|
|
Both tests are marked xfail(strict=False) because the Alembic migration and
|
|
db.models are implemented in Plan 03. The xfail markers will be removed once
|
|
Plan 03 lands and alembic upgrade head successfully creates all expected tables.
|
|
|
|
Requirements covered:
|
|
STORE-01 — Platform migrated to PostgreSQL; full v1 schema applied
|
|
D-02 — groups table stub included in the Phase 1 migration
|
|
D-03 — documents.user_id is nullable in Phase 1
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
|
|
EXPECTED_TABLES = {
|
|
"users",
|
|
"quotas",
|
|
"refresh_tokens",
|
|
"folders",
|
|
"documents",
|
|
"topics",
|
|
"document_topics",
|
|
"shares",
|
|
"audit_log",
|
|
"cloud_connections",
|
|
"groups",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 1: alembic upgrade head creates all 11 expected tables
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.xfail(strict=False, reason="implemented in plan 03")
|
|
def test_migration_creates_all_tables(tmp_path, monkeypatch):
|
|
"""After alembic upgrade head, the DB contains all 11 v1 tables (D-01, D-02)."""
|
|
try:
|
|
import alembic.command
|
|
from alembic.config import Config
|
|
except ImportError as exc:
|
|
pytest.skip(f"alembic not installed: {exc}")
|
|
|
|
db_path = tmp_path / "test_migrate.db"
|
|
db_url = f"sqlite+aiosqlite:///{db_path}"
|
|
|
|
monkeypatch.setenv("DATABASE_MIGRATE_URL", db_url)
|
|
|
|
# Run alembic upgrade head using the Python API
|
|
alembic_cfg = Config("alembic.ini")
|
|
alembic_cfg.set_main_option("sqlalchemy.url", db_url)
|
|
|
|
try:
|
|
alembic.command.upgrade(alembic_cfg, "head")
|
|
except Exception as exc:
|
|
pytest.skip(f"alembic upgrade failed — plan 03 not yet complete: {exc}")
|
|
|
|
# Connect synchronously (sqlite3 is available in stdlib) and check tables
|
|
import sqlite3
|
|
conn = sqlite3.connect(str(db_path))
|
|
try:
|
|
cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
|
actual_tables = {row[0] for row in cursor.fetchall()}
|
|
finally:
|
|
conn.close()
|
|
|
|
missing = EXPECTED_TABLES - actual_tables
|
|
assert not missing, (
|
|
f"Migration is missing tables: {sorted(missing)}. "
|
|
f"Tables found: {sorted(actual_tables)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 2: documents.user_id is nullable (D-03)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.xfail(strict=False, reason="implemented in plan 03")
|
|
def test_documents_user_id_nullable(tmp_path, monkeypatch):
|
|
"""documents.user_id must be nullable in Phase 1 (D-03 — no auth yet)."""
|
|
try:
|
|
import alembic.command
|
|
from alembic.config import Config
|
|
except ImportError as exc:
|
|
pytest.skip(f"alembic not installed: {exc}")
|
|
|
|
db_path = tmp_path / "test_nullable.db"
|
|
db_url = f"sqlite+aiosqlite:///{db_path}"
|
|
|
|
monkeypatch.setenv("DATABASE_MIGRATE_URL", db_url)
|
|
|
|
alembic_cfg = Config("alembic.ini")
|
|
alembic_cfg.set_main_option("sqlalchemy.url", db_url)
|
|
|
|
try:
|
|
alembic.command.upgrade(alembic_cfg, "head")
|
|
except Exception as exc:
|
|
pytest.skip(f"alembic upgrade failed — plan 03 not yet complete: {exc}")
|
|
|
|
# PRAGMA table_info returns: cid, name, type, notnull, dflt_value, pk
|
|
import sqlite3
|
|
conn = sqlite3.connect(str(db_path))
|
|
try:
|
|
cursor = conn.execute("PRAGMA table_info(documents)")
|
|
columns = {row[1]: {"notnull": row[3]} for row in cursor.fetchall()}
|
|
finally:
|
|
conn.close()
|
|
|
|
assert "user_id" in columns, (
|
|
"documents table has no user_id column — migration may not have applied"
|
|
)
|
|
assert columns["user_id"]["notnull"] == 0, (
|
|
"documents.user_id is NOT NULL but D-03 requires it to be nullable in Phase 1"
|
|
)
|