Cheng Zhou
d5279b124f
Storage Persistence Guard / storage-persistence-audit (push) Has been cancelled
Client Quality Gates / Shared client and Web (push) Has been cancelled
Client Quality Gates / macOS Desktop (push) Has been cancelled
Client Quality Gates / Shared client and Web (pull_request) Has been cancelled
Client Quality Gates / macOS Desktop (pull_request) Has been cancelled
Storage Persistence Guard / storage-persistence-audit (pull_request) Has been cancelled
123 lines
2.9 KiB
Python
123 lines
2.9 KiB
Python
from contextlib import asynccontextmanager
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
|
|
from app import main
|
|
|
|
|
|
def test_db_base_registers_permission_models_for_create_all():
|
|
backend_dir = Path(__file__).resolve().parents[1]
|
|
code = """
|
|
from app.db.base_class import Base
|
|
import app.db.base
|
|
|
|
required_tables = {
|
|
"api_endpoint_permissions",
|
|
"api_endpoint_registries",
|
|
"permission_templates",
|
|
"permission_template_versions",
|
|
}
|
|
missing = sorted(required_tables - set(Base.metadata.tables))
|
|
if missing:
|
|
print(",".join(missing))
|
|
raise SystemExit(1)
|
|
"""
|
|
env = {
|
|
**os.environ,
|
|
"PYTHONDONTWRITEBYTECODE": "1",
|
|
"PYTHONPATH": str(backend_dir),
|
|
}
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", code],
|
|
cwd=backend_dir,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
|
|
|
|
class _DummyTask:
|
|
def __init__(self):
|
|
self.awaited = False
|
|
self.cancelled = False
|
|
|
|
def cancel(self):
|
|
self.cancelled = True
|
|
|
|
def done(self):
|
|
return False
|
|
|
|
def __await__(self):
|
|
async def _wait():
|
|
self.awaited = True
|
|
|
|
return _wait().__await__()
|
|
|
|
|
|
class _DummySession:
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
class _DummyBegin:
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def run_sync(self, fn):
|
|
return None
|
|
|
|
|
|
class _DummyEngine:
|
|
def begin(self):
|
|
return _DummyBegin()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_development_startup_does_not_seed_business_data(monkeypatch):
|
|
ensure_admin_exists = AsyncMock()
|
|
create_all = AsyncMock()
|
|
legacy_pk = AsyncMock()
|
|
scheduler_task = _DummyTask()
|
|
|
|
async def fake_scheduler(stop_event):
|
|
return None
|
|
|
|
def fake_create_task(coroutine):
|
|
coroutine.close()
|
|
return scheduler_task
|
|
|
|
@asynccontextmanager
|
|
async def fake_session_local():
|
|
yield _DummySession()
|
|
|
|
monkeypatch.setattr(main.settings, "ENV", "development")
|
|
monkeypatch.setattr(main, "ensure_admin_exists", ensure_admin_exists)
|
|
monkeypatch.setattr(main, "run_daily_lost_visit_job", fake_scheduler)
|
|
monkeypatch.setattr(main, "resolve_monitoring_server_location", AsyncMock(return_value=None))
|
|
monkeypatch.setattr(main.asyncio, "create_task", fake_create_task)
|
|
monkeypatch.setattr(main, "engine", _DummyEngine())
|
|
monkeypatch.setattr(main, "SessionLocal", fake_session_local)
|
|
monkeypatch.setattr(main, "_ensure_legacy_primary_keys", legacy_pk)
|
|
monkeypatch.setattr(main.Base.metadata, "create_all", create_all)
|
|
|
|
async with main.lifespan(FastAPI()):
|
|
pass
|
|
|
|
ensure_admin_exists.assert_awaited_once()
|