48 lines
1.0 KiB
Python
48 lines
1.0 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from contextlib import contextmanager
|
||
|
|
|
||
|
|
import ariadne.db.database as db_module
|
||
|
|
from ariadne.db.database import Database
|
||
|
|
|
||
|
|
|
||
|
|
class DummyResult:
|
||
|
|
def __init__(self, row=None, rows=None):
|
||
|
|
self._row = row
|
||
|
|
self._rows = rows or []
|
||
|
|
|
||
|
|
def fetchone(self):
|
||
|
|
return self._row
|
||
|
|
|
||
|
|
def fetchall(self):
|
||
|
|
return self._rows
|
||
|
|
|
||
|
|
|
||
|
|
class DummyConn:
|
||
|
|
def __init__(self):
|
||
|
|
self.row_factory = None
|
||
|
|
self.executed = []
|
||
|
|
|
||
|
|
def execute(self, query, params=None):
|
||
|
|
self.executed.append((query, params))
|
||
|
|
return DummyResult()
|
||
|
|
|
||
|
|
|
||
|
|
class DummyPool:
|
||
|
|
def __init__(self, conninfo=None, max_size=None):
|
||
|
|
self.conn = DummyConn()
|
||
|
|
|
||
|
|
@contextmanager
|
||
|
|
def connection(self):
|
||
|
|
yield self.conn
|
||
|
|
|
||
|
|
def close(self):
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def test_ensure_schema_runs(monkeypatch) -> None:
|
||
|
|
monkeypatch.setattr(db_module, "ConnectionPool", DummyPool)
|
||
|
|
db = Database("postgresql://user:pass@localhost/db")
|
||
|
|
db.ensure_schema()
|
||
|
|
assert db._pool.conn.executed
|