mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 06:40:45 +08:00
oliver
527472976a
Ensure startup safety-net adds a_ifname/z_ifname on brownfield DBs, and normalize ADD COLUMN IF NOT EXISTS for SQLite tests. Co-authored-by: Cursor <cursoragent@cursor.com>
87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
"""Schema patch / Alembic wiring smoke tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from netx_api.db import Base
|
|
import netx_api.models # noqa: F401
|
|
from netx_api.schema_patches import (
|
|
apply_auth_schema_patches,
|
|
apply_domain_schema_patches,
|
|
apply_topology_schema_safety_net,
|
|
)
|
|
|
|
|
|
class SchemaPatchesTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(bind=self.engine)
|
|
|
|
def tearDown(self) -> None:
|
|
self.engine.dispose()
|
|
|
|
def test_auth_patches_idempotent(self) -> None:
|
|
with self.engine.begin() as conn:
|
|
apply_auth_schema_patches(conn)
|
|
apply_auth_schema_patches(conn)
|
|
insp = inspect(self.engine)
|
|
user_cols = {c["name"] for c in insp.get_columns("app_user")}
|
|
token_cols = {c["name"] for c in insp.get_columns("api_token")}
|
|
self.assertIn("scopes", user_cols)
|
|
self.assertIn("must_change_password", user_cols)
|
|
self.assertIn("scopes", token_cols)
|
|
self.assertIn("expires_at", token_cols)
|
|
self.assertIn("auth_session", insp.get_table_names())
|
|
|
|
def test_topology_safety_net_adds_ume_link_ifnames(self) -> None:
|
|
"""Brownfield: table without ifnames; safety-net must add them (sync depends on it)."""
|
|
with self.engine.begin() as conn:
|
|
conn.execute(text("DROP TABLE IF EXISTS ume_topo_link"))
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
CREATE TABLE ume_topo_link (
|
|
link_id VARCHAR(128) PRIMARY KEY,
|
|
name VARCHAR(1024) DEFAULT '',
|
|
a_ptp VARCHAR(256) DEFAULT '',
|
|
z_ptp VARCHAR(256) DEFAULT ''
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
apply_topology_schema_safety_net(conn)
|
|
apply_topology_schema_safety_net(conn)
|
|
cols = {c["name"] for c in inspect(self.engine).get_columns("ume_topo_link")}
|
|
self.assertIn("a_ifname", cols)
|
|
self.assertIn("z_ifname", cols)
|
|
|
|
def test_domain_patches_do_not_raise(self) -> None:
|
|
with self.engine.begin() as conn:
|
|
apply_domain_schema_patches(conn)
|
|
apply_domain_schema_patches(conn)
|
|
|
|
def test_worker_default_off_inline(self) -> None:
|
|
from netx_api.config import Settings
|
|
|
|
s = Settings(_env_file=None)
|
|
self.assertTrue(s.run_inline_schedulers)
|
|
self.assertTrue(s.alembic_upgrade_on_start)
|
|
|
|
versions = Path(__file__).resolve().parents[1] / "alembic" / "versions"
|
|
files = sorted(p.name for p in versions.glob("*.py") if p.name != "__init__.py")
|
|
self.assertIn("20260802_scopes.py", files)
|
|
self.assertIn("20260802_legacy_schema.py", files)
|
|
self.assertIn("20260806_auth_session.py", files)
|
|
self.assertIn("20260806_auth_refresh.py", files)
|
|
text_legacy = (versions / "20260802_legacy_schema.py").read_text(encoding="utf-8")
|
|
self.assertIn('down_revision', text_legacy)
|
|
self.assertIn("20260802_scopes", text_legacy)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|