8545282536

Committed 03 Apr 2024 08:44PM UTC coverage: 46.25%. Remained the same

Build # 8545282536

Build Type

push

github

Committed by

sivakov512

Commit Message

Make linter happy

Run Details

0 of 3 new or added lines in 1 file covered. (0.0%)

37 of 80 relevant lines covered (46.25%)

1.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

53.85

/asyncpg_engine/pytest_plugin.py

import typing

import pytest
import pytest_asyncio
from asyncpg import Connection

from asyncpg_engine import Engine


def pytest_configure(config: typing.Any) -> None:
    config.addinivalue_line("markers", "asyncpg_engine: configure asyncpg-engine plugin behaviour.")


@pytest.fixture
def asyncpg_engine_cls() -> typing.Type[Engine]:
    return Engine


@pytest_asyncio.fixture
async def db(
    request: pytest.FixtureRequest, asyncpg_engine_cls: typing.Type[Engine], postgres_url: str
) -> typing.AsyncGenerator[Engine, None]:
    plugin_config = request.node.get_closest_marker("asyncpg_engine")

    transactional = getattr(plugin_config, "kwargs", {}).get("transactional", True)

    _db = await asyncpg_engine_cls.create(url=postgres_url, use_single_connection=transactional)

    con: Connection = await _db.acquire()  # type: ignore
    tr = con.transaction()
    await tr.start()

    yield _db

    await tr.rollback()
    await _db.release(con, force=transactional)
    await _db.close()


@pytest_asyncio.fixture
async def con(db: Engine) -> typing.AsyncGenerator[Connection, None]:
    async with db.acquire() as _con:
        yield _con

1	import typing	×
2
3	import pytest	×
4	import pytest_asyncio	×
5	from asyncpg import Connection	×
6
7	from asyncpg_engine import Engine	×
8
9
10	def pytest_configure(config: typing.Any) -> None:	×
11	config.addinivalue_line("markers", "asyncpg_engine: configure asyncpg-engine plugin behaviour.")	3✔
12
13
NEW 14	@pytest.fixture	×
15	def asyncpg_engine_cls() -> typing.Type[Engine]:	×
16	return Engine	3✔
17
18
NEW 19	@pytest_asyncio.fixture	×
20	async def db(	×
21	request: pytest.FixtureRequest, asyncpg_engine_cls: typing.Type[Engine], postgres_url: str
22	) -> typing.AsyncGenerator[Engine, None]:
23	plugin_config = request.node.get_closest_marker("asyncpg_engine")	3✔
24
25	transactional = getattr(plugin_config, "kwargs", {}).get("transactional", True)	3✔
26
27	_db = await asyncpg_engine_cls.create(url=postgres_url, use_single_connection=transactional)	3✔
28
29	con: Connection = await _db.acquire() # type: ignore	3✔
30	tr = con.transaction()	3✔
31	await tr.start()	3✔
32
33	yield _db	3✔
34
35	await tr.rollback()	3✔
36	await _db.release(con, force=transactional)	3✔
37	await _db.close()	3✔
38
39
NEW 40	@pytest_asyncio.fixture	×
41	async def con(db: Engine) -> typing.AsyncGenerator[Connection, None]:	×
42	async with db.acquire() as _con:	3✔
43	yield _con	3✔

sivakov512 / asyncpg-engine / 8545282536

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous