78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
import pytest
|
|
|
|
from app.db.session import _normalize_payment_order_provision_username_index
|
|
|
|
|
|
class _FakeResult:
|
|
def __init__(self, rows):
|
|
self._rows = rows
|
|
|
|
def __iter__(self):
|
|
return iter(self._rows)
|
|
|
|
|
|
class _FakeConnection:
|
|
def __init__(self, *, dialect_name: str, responses: list[list[tuple[str]]]) -> None:
|
|
self.dialect = type("Dialect", (), {"name": dialect_name})()
|
|
self._responses = list(responses)
|
|
self.executed: list[tuple[str, dict[str, object] | None]] = []
|
|
|
|
async def execute(self, statement, params=None):
|
|
self.executed.append((str(statement), params))
|
|
rows = self._responses.pop(0) if self._responses else []
|
|
return _FakeResult(rows)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normalize_payment_order_index_drops_stale_unique_and_recreates_plain_index() -> None:
|
|
connection = _FakeConnection(
|
|
dialect_name="mysql",
|
|
responses=[
|
|
[("ix_payment_orders_provision_username",)],
|
|
[],
|
|
[],
|
|
[],
|
|
],
|
|
)
|
|
|
|
await _normalize_payment_order_provision_username_index(connection)
|
|
|
|
executed_sql = [sql for sql, _ in connection.executed]
|
|
|
|
assert any(
|
|
"DROP INDEX `ix_payment_orders_provision_username` ON `payment_orders`" in sql
|
|
for sql in executed_sql
|
|
)
|
|
assert any(
|
|
"CREATE INDEX `ix_payment_orders_provision_username` ON `payment_orders` (`provision_username`)"
|
|
in sql
|
|
for sql in executed_sql
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normalize_payment_order_index_keeps_existing_non_unique_index() -> None:
|
|
connection = _FakeConnection(
|
|
dialect_name="mysql",
|
|
responses=[
|
|
[],
|
|
[("custom_payment_orders_provision_username",)],
|
|
],
|
|
)
|
|
|
|
await _normalize_payment_order_provision_username_index(connection)
|
|
|
|
executed_sql = [sql for sql, _ in connection.executed]
|
|
|
|
assert not any("DROP INDEX" in sql for sql in executed_sql)
|
|
assert not any("CREATE INDEX" in sql for sql in executed_sql)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normalize_payment_order_index_skips_non_mysql_backends() -> None:
|
|
connection = _FakeConnection(dialect_name="sqlite", responses=[])
|
|
|
|
await _normalize_payment_order_provision_username_index(connection)
|
|
|
|
assert connection.executed == []
|