pyfect ships a pytest plugin, registered when you install the package. Decorate a generator
test with @testing.effect and write it like @fx.gen.
The test clock#
@testing.effect runs the test with a TestClock: time only moves when the test calls
testing.adjust or set_time, and every fiber woken by a move runs until it waits again. Tests
of retries, timeouts and schedules are exact and take no real time.
A test that would wait forever fails with StalledError instead of hanging, and each test has a
real-time timeout (@testing.effect(timeout=10), 5 seconds by default) that interrupts it and
runs its finalizers.
from dataclasses import dataclass
import pyfect as fx
from pyfect import schedule, testing
@dataclass
class Flaky(fx.TaggedError):
pass
calls: list[float] = []
def flaky_call() -> fx.Effect[str, Flaky]:
"""Fails three times, then succeeds; records the virtual time of each call."""
def call(now: float) -> fx.Effect[str, Flaky]:
calls.append(now)
return fx.fail(Flaky()) if len(calls) <= 3 else fx.succeed("ok")
return fx.clock.monotonic().flat_map(call)
@testing.effect
def test_retry_backs_off():
calls.clear()
fiber = yield from flaky_call().retry(schedule.exponential(1).up_to(times=3)).fork()
yield from testing.adjust(7) # virtual time: the test takes no real time
assert (yield from fiber.join()) == "ok"
assert [t - calls[0] for t in calls] == [0, 1, 3, 7] # waits of 1, 2 and 4 seconds
@testing.effect
def test_timeout():
fiber = yield from fx.sleep(10).timeout(5).exit().fork()
yield from testing.adjust(5)
assert isinstance((yield from fiber.join()), fx.Failure)Services in tests#
testing.layer_fixture(layer, scope=...) builds a layer once per pytest scope. A test takes the
fixture as its first parameter, annotated testing.Services[...] with what the layer builds;
pyright rejects a test that needs anything else. The test console records what was logged.
More: @testing.live (real clock and console), @testing.effect(asyncio=True),
testing.effect_fixture, testing.flaky, testing.with_live, and assert_exit_success and
friends, which print the full cause on failure. Run only fast tests with
pytest -m "not pyfect_live".
from typing import Protocol
import pyfect as fx
from pyfect import layer, testing
class UserStore(Protocol):
def find(self, user_id: str) -> fx.Effect[str | None]: ...
class Users(fx.Tag[UserStore]): ...
class InMemoryUsers:
def find(self, user_id: str) -> fx.Effect[str | None]:
return fx.succeed({"1": "Ada"}.get(user_id))
# Built once for this module and released when its tests are done.
users = testing.layer_fixture(layer.succeed(Users, InMemoryUsers()), scope="module")
@testing.effect
def test_finds_ada(users: testing.Services[Users]):
store = yield from Users()
assert (yield from store.find("1")) == "Ada"
@testing.effect
def test_logs_go_to_the_test_console():
yield from fx.console.log("hello")
assert (yield from testing.log_lines) == ["hello"]Reference: pyfect.testing.