Spaces:
Sleeping
Sleeping
File size: 5,301 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | """Shared fixtures and reusable dummy steps for pipeline engine tests.
No ACE imports — every step here is a generic dummy that only uses the
pipeline primitives (StepContext, StepProtocol).
"""
from __future__ import annotations
import asyncio
import threading
import time
from types import MappingProxyType
import pytest
from pipeline import StepContext
# ---------------------------------------------------------------------------
# Reusable dummy step classes (no ACE knowledge)
# ---------------------------------------------------------------------------
class Noop:
"""Pass-through step — does not change context."""
requires = frozenset()
provides = frozenset()
def __call__(self, ctx: StepContext) -> StepContext:
return ctx
class SetA:
"""Writes metadata['a'] = 1. No requirements."""
requires = frozenset()
provides = frozenset({"a"})
def __call__(self, ctx: StepContext) -> StepContext:
return ctx.replace(metadata=MappingProxyType({**ctx.metadata, "a": 1}))
class SetB:
"""Reads 'a', writes metadata['b'] = metadata['a'] + 1."""
requires = frozenset({"a"})
provides = frozenset({"b"})
def __call__(self, ctx: StepContext) -> StepContext:
return ctx.replace(
metadata=MappingProxyType({**ctx.metadata, "b": ctx.metadata["a"] + 1})
)
class SetC:
"""Reads 'b', writes metadata['c'] = metadata['b'] * 2."""
requires = frozenset({"b"})
provides = frozenset({"c"})
def __call__(self, ctx: StepContext) -> StepContext:
return ctx.replace(
metadata=MappingProxyType({**ctx.metadata, "c": ctx.metadata["b"] * 2})
)
class Boom:
"""Always raises RuntimeError."""
requires = frozenset()
provides = frozenset()
def __call__(self, ctx: StepContext) -> StepContext:
raise RuntimeError("boom")
class Slow:
"""Sleeps for *delay* seconds then sets metadata['done'] = True."""
requires = frozenset()
provides = frozenset({"done"})
def __init__(self, delay: float = 0.05):
self.delay = delay
def __call__(self, ctx: StepContext) -> StepContext:
time.sleep(self.delay)
return ctx.replace(metadata=MappingProxyType({**ctx.metadata, "done": True}))
class AsyncStep:
"""Async step — sets metadata['async'] = True."""
requires = frozenset()
provides = frozenset({"async_done"})
async def __call__(self, ctx: StepContext) -> StepContext:
await asyncio.sleep(0) # yield to event loop
return ctx.replace(
metadata=MappingProxyType({**ctx.metadata, "async_done": True})
)
class Recorder:
"""Records every ctx it receives via call_log (thread-safe)."""
requires = frozenset()
provides = frozenset()
def __init__(self):
self.call_log: list[StepContext] = []
self._lock = threading.Lock()
def __call__(self, ctx: StepContext) -> StepContext:
with self._lock:
self.call_log.append(ctx)
return ctx
class BoundaryStep:
"""Foreground step that marks the async_boundary handoff."""
requires = frozenset()
provides = frozenset({"bg_result"})
async_boundary = True
max_workers = 2
def __call__(self, ctx: StepContext) -> StepContext:
time.sleep(0.01) # simulate background work
return ctx.replace(
metadata=MappingProxyType({**ctx.metadata, "bg_result": True})
)
class SlowBoundaryStep:
"""Slow boundary step for timeout testing."""
requires = frozenset()
provides = frozenset({"slow_bg"})
async_boundary = True
max_workers = 1
def __call__(self, ctx: StepContext) -> StepContext:
time.sleep(2.0) # intentionally slow
return ctx.replace(metadata=MappingProxyType({**ctx.metadata, "slow_bg": True}))
class SerialStep:
"""Background step that must serialize (max_workers=1). Appends to a shared log."""
requires = frozenset()
provides = frozenset({"serial_done"})
max_workers = 1
_log: list[str] = []
_log_lock = threading.Lock()
def __call__(self, ctx: StepContext) -> StepContext:
with self._log_lock:
SerialStep._log.append(f"start-{ctx.sample}")
time.sleep(0.02) # ensure ordering is visible if concurrent
SerialStep._log.append(f"end-{ctx.sample}")
return ctx.replace(
metadata=MappingProxyType({**ctx.metadata, "serial_done": True})
)
# ---------------------------------------------------------------------------
# Pytest fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def noop():
return Noop()
@pytest.fixture
def set_a():
return SetA()
@pytest.fixture
def set_b():
return SetB()
@pytest.fixture
def set_c():
return SetC()
@pytest.fixture
def boom():
return Boom()
@pytest.fixture
def recorder():
return Recorder()
@pytest.fixture
def base_ctx():
"""A minimal StepContext with sample='test'."""
return StepContext(sample="test")
|