judgevet.testing._conformance_async_media
Status: draft.
judgevet.testing._conformance_async_media
Observers and checks for the asynchronous media rules.
AsyncMediaProbe wraps an async media port, counts media dispatches and
requires system_one_media to return an awaitable. refused_problem,
undeclared_problem and declared_problem drive async_judge_with_images
for the three media rules. Each returns None when the provider keeps the rule,
or a failure message when it breaks it. RuleSkipped tells the rule to skip.
RuleFailed carries a failure whose cause the rule keeps.
None of these import pytest.
Source: https://github.com/Alberto-Codes/judgevet/issues/252#issuecomment-5924224316.
Examples:
from contextlib import nullcontext
import anyio
from judgevet.testing import AsyncFakeSystemOnePort
from judgevet.testing._conformance_async_media import refused_problem
factory = lambda: nullcontext(AsyncFakeSystemOnePort())
assert anyio.run(refused_problem, factory, "conformance-model") is None
See Also
- judgevet.testing._conformance_async: The rules that use these checks
- judgevet.testing._conformance_probes: The synchronous
MediaProbe - judgevet.media:
async_judge_with_imagesand its capability checks
MediaCheck = Callable[[AsyncMediaSystemOnePort, str], Awaitable[str | None]]
module-attribute
A check that runs against one selected async media port and model.
AsyncMediaProbe
Delegate every async media method and count media dispatches.
Attributes:
| Name | Type | Description |
|---|---|---|
port |
AsyncMediaSystemOnePort
|
The wrapped media port. |
media_calls |
int
|
How many media judgments reached the port. |
Examples:
from judgevet.testing import AsyncFakeSystemOnePort
probe = AsyncMediaProbe(AsyncFakeSystemOnePort())
assert probe.media_calls == 0
__init__(port)
Wrap an async media port.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
port
|
AsyncMediaSystemOnePort
|
The media port to observe. |
required |
capabilities(model)
Return the wrapped port's declaration unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
The selected model. |
required |
Returns:
| Type | Description |
|---|---|
MediaCapabilities
|
The wrapped port's capabilities. |
system_one(state, questions, model)
async
Await the wrapped port's text judgment.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state
|
str | dict[str, Any] | list[Any]
|
The content to judge. |
required |
questions
|
Mapping[str, Question | Mapping[str, Any]]
|
Question names mapped to question definitions. |
required |
model
|
str
|
The selected model. |
required |
Returns:
| Type | Description |
|---|---|
SystemOneResponse
|
The wrapped port's response. |
system_one_media(state, questions, model, *, evidence)
async
Count the media dispatch and await the wrapped port's result.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state
|
str | dict[str, Any] | list[Any]
|
The content to judge. |
required |
questions
|
Mapping[str, Question | Mapping[str, Any]]
|
Question names mapped to question definitions. |
required |
model
|
str
|
The selected model. |
required |
evidence
|
ImageEvidence
|
The ordered image evidence. |
required |
Returns:
| Type | Description |
|---|---|
SystemOneResponse
|
The wrapped port's response. |
Raises:
| Type | Description |
|---|---|
NotAwaitableError
|
If the wrapped method returns a value that is not awaitable. |
NotAwaitableError
Bases: Exception
Signal that system_one_media returned a value that is not awaitable.
Examples:
error = NotAwaitableError("system_one_media returned dict")
assert "dict" in str(error)
RuleFailed
Bases: Exception
Fail a media rule and keep the exception that broke it as the cause.
Examples:
error = RuleFailed("map backend failures to a ProviderError subclass")
assert "ProviderError" in str(error)
RuleSkipped
Bases: Exception
Signal that a media rule does not apply to the provider.
Examples:
try:
raise RuleSkipped("The provider supplies no media port.")
except RuleSkipped as skip:
assert "media port" in str(skip)
declared_problem(factory, media_port, model)
async
Check that declared media reaches the port and returns typed answers.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
factory
|
AsyncProviderFactory
|
The provider's factory. |
required |
media_port
|
AsyncMediaSystemOnePort | None
|
The dedicated media port, or None. |
required |
model
|
str
|
The selected model. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The failure message, or None when the port keeps the rule. |
refused_problem(factory, model)
async
Check that a text-only port exposes no partial media surface.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
factory
|
AsyncProviderFactory
|
The provider's factory. |
required |
model
|
str
|
The selected model. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The failure message, or None when the port keeps the rule. |
Raises:
| Type | Description |
|---|---|
RuleSkipped
|
If the factory's port supports media. |
undeclared_problem(factory, media_port, model)
async
Check that media outside the declared capabilities is refused first.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
factory
|
AsyncProviderFactory
|
The provider's factory. |
required |
media_port
|
AsyncMediaSystemOnePort | None
|
The dedicated media port, or None. |
required |
model
|
str
|
The selected model. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The failure message, or None when the port keeps the rule. |