81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import passlib.context
|
|
|
|
|
|
def _load_migration_module():
|
|
path = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "alembic"
|
|
/ "versions"
|
|
/ "add_users_table_rbac.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location("add_users_table_rbac_migration", path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_add_users_table_migration_reuses_precreated_postgres_enums(monkeypatch) -> None:
|
|
migration = _load_migration_module()
|
|
created_types: list[tuple[str, bool]] = []
|
|
|
|
class _FakeDialect:
|
|
name = "postgresql"
|
|
|
|
class _FakeBind:
|
|
dialect = _FakeDialect()
|
|
|
|
class _FakeOp:
|
|
def __init__(self) -> None:
|
|
self.created_table: tuple[str, tuple[object, ...]] | None = None
|
|
self.executed_sql: list[str] = []
|
|
|
|
def get_bind(self):
|
|
return _FakeBind()
|
|
|
|
def create_table(self, table_name: str, *columns, **kwargs):
|
|
_ = kwargs
|
|
self.created_table = (table_name, columns)
|
|
|
|
def execute(self, sql) -> None: # noqa: ANN001
|
|
self.executed_sql.append(str(sql))
|
|
|
|
class _FakeCryptContext:
|
|
def __init__(self, **kwargs) -> None:
|
|
_ = kwargs
|
|
|
|
def hash(self, raw: str) -> str:
|
|
return f"hashed::{raw}"
|
|
|
|
def _fake_enum_create(self, bind, checkfirst: bool = False) -> None: # noqa: ANN001
|
|
_ = bind
|
|
created_types.append((str(getattr(self, "name", "")), bool(checkfirst)))
|
|
|
|
fake_op = _FakeOp()
|
|
monkeypatch.setattr(migration, "op", fake_op)
|
|
monkeypatch.setattr(migration.ENUM, "create", _fake_enum_create)
|
|
monkeypatch.setattr(passlib.context, "CryptContext", _FakeCryptContext)
|
|
|
|
migration.upgrade()
|
|
|
|
assert created_types == [("userrole", True), ("userstatus", True)]
|
|
assert fake_op.created_table is not None
|
|
|
|
table_name, columns = fake_op.created_table
|
|
assert table_name == "users"
|
|
column_map = {column.name: column for column in columns if getattr(column, "name", None)}
|
|
|
|
role_type = column_map["role"].type
|
|
status_type = column_map["status"].type
|
|
|
|
assert isinstance(role_type, migration.ENUM)
|
|
assert isinstance(status_type, migration.ENUM)
|
|
assert role_type.create_type is False
|
|
assert status_type.create_type is False
|