v2.0

tests/test_db_session.py

@@ -0,0 +1,77 @@
+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 == []