judgevet.testing._conformance_async_scope

Status: draft.

judgevet.testing._conformance_async_scope

Observers and checks for the asynchronous provider scope rules.

AsyncScopeRecorder wraps an AsyncProviderFactory and records every async context it constructs. The kit observes entry and exit itself instead of trusting the provider. exit_problem and body_problem drive async_provider_scope and describe a broken lifecycle. Each returns None when the provider keeps the rule. The synchronous kit's ScopeRecorder does the same for a ProviderFactory. None of these import pytest. Source: https://github.com/Alberto-Codes/judgevet/issues/251.

Examples:

from contextlib import nullcontext

import anyio

from judgevet.providers import async_provider_scope
from judgevet.testing import AsyncFakeSystemOnePort
from judgevet.testing._conformance_async_scope import (
    AsyncScopeRecorder,
    body_problem,
    exit_problem,
)

recorder = AsyncScopeRecorder(lambda: nullcontext(AsyncFakeSystemOnePort()))


async def enter_once():
    async with async_provider_scope(factory=recorder) as port:
        assert callable(port.system_one)


anyio.run(enter_once)
assert recorder.contexts[0].exits == [None]

factory = lambda: nullcontext(AsyncFakeSystemOnePort())
assert anyio.run(exit_problem, factory) is None
assert anyio.run(body_problem, factory) is None
See Also

AsyncRecordedContext

Bases: AbstractAsyncContextManager[AsyncSystemOnePort]

Delegate to one async provider context and record its entry and exit.

Attributes:

Name Type Description
inner AbstractAsyncContextManager[AsyncSystemOnePort]

The provider's async context.
enters int

How many times entry ran.
exits list[BaseException | None]

The exception each exit received, or None for a clean exit.

Examples:

from contextlib import nullcontext

import anyio

from judgevet.testing import AsyncFakeSystemOnePort

recorded = AsyncRecordedContext(nullcontext(AsyncFakeSystemOnePort()))


async def enter():
    async with recorded:
        pass


anyio.run(enter)
assert (recorded.enters, recorded.exits) == (1, [None])

__aenter__() async

Count the entry and enter the provider context.

Returns:

Type Description
AsyncSystemOnePort

The port the provider context yields.

__aexit__(exc_type, exc, traceback) async

Record the exit and pass the provider's suppression answer through.

Parameters:

Name Type Description Default
exc_type type[BaseException] | None

The escaping exception type, or None.

 required
exc BaseException | None

The escaping exception, or None.

 required
traceback TracebackType | None

The escaping traceback, or None.

 required

Returns:

Type Description
bool | None

The provider context's own return value.

__init__(inner)

Wrap one async provider context.

Parameters:

Name Type Description Default
inner AbstractAsyncContextManager[AsyncSystemOnePort]

The async context the provider factory returned.

 required

AsyncScopeRecorder

Wrap an async provider factory and record every context it constructs.

Attributes:

Name Type Description
factory AsyncProviderFactory

The provider's own factory.
contexts list[AsyncRecordedContext]

One recorder per factory call.

Examples:

from contextlib import nullcontext

from judgevet.testing import AsyncFakeSystemOnePort

recorder = AsyncScopeRecorder(lambda: nullcontext(AsyncFakeSystemOnePort()))
recorder()
assert len(recorder.contexts) == 1

__call__()

Construct one provider context and start recording it.

Returns:

Type Description
AsyncRecordedContext

A recorder that delegates to the new provider context.

__init__(factory)

Store the provider factory.

Parameters:

Name Type Description Default
factory AsyncProviderFactory

The provider's own factory.

 required

body_problem(factory) async

Raise inside a scope and describe a lost, replaced or suppressed error.

The check also enters one context directly, as AsyncProviderFactory callers may, and requires that its exit does not suppress the error.

Parameters:

Name Type Description Default
factory AsyncProviderFactory

The provider's factory.

 required

Returns:

Type Description
str | None

A failure message, or None when the body exception reaches the caller
str | None

unchanged and the scope exits once with it.

exit_problem(factory) async

Enter two scopes and describe a shared context or a wrong exit count.

Parameters:

Name Type Description Default
factory AsyncProviderFactory

The provider's factory.

 required

Returns:

Type Description
str | None

A failure message, or None when each scope has its own context that
str | None

entered once and exited once cleanly.