lib.ProcFilter: Add unit tests

ProcPipeline and the run() helper are pure logic without test coverage.

Add unit tests for the identity filter, chained filter execution order, the
append() layouts, and the chain handling of run().

Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v0.85.1
This commit is contained in:
Jan Lindemann 2026-09-08 22:33:50 +02:00
commit b8d4402fe4
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
2 changed files with 71 additions and 0 deletions

View file

@ -0,0 +1,7 @@
TOPDIR = ../../../../../../..
include $(TOPDIR)/make/proj.mk
include $(JWBDIR)/make/py-run.mk
all:
test: run

View file

@ -0,0 +1,64 @@
import asyncio
from typing import override
from jw.pkg.lib.ProcFilter import ProcFilter, ProcFilterIdentity, ProcPipeline, run
from jw.pkg.lib.Result import Result
class Upper(ProcFilter):
@override
async def _run(self, data: bytes | None) -> Result:
return Result(data.upper() if data else None, None, 0)
class Exclaim(ProcFilter):
@override
async def _run(self, data: bytes | None) -> Result:
return Result(data + b'!' if data else None, None, 0)
# -- ProcFilterIdentity --
assert asyncio.run(ProcFilterIdentity().run(b'abc')).stdout == b'abc'
# A None input comes back as a result with no output; stdout would
# report b'' for a status-0 result, so use stdout_or_none here
assert asyncio.run(ProcFilterIdentity().run(None)).stdout_or_none is None
assert asyncio.run(ProcFilterIdentity().run(b'abc')).status == 0
# -- ProcPipeline --
# A single filter
pl = ProcPipeline(Upper())
assert asyncio.run(pl.run(b'abc')).stdout == b'ABC'
# Filters are applied in order
pl = ProcPipeline([Upper(), Exclaim()])
assert asyncio.run(pl.run(b'abc')).stdout == b'ABC!'
# append() accepts filters and iterables of filters, nested
pl = ProcPipeline()
pl.append(Upper())
pl.append([Exclaim()])
assert asyncio.run(pl.run(b'abc')).stdout == b'ABC!'
# Input may be bytes or a Result
assert asyncio.run(pl.run(Result(b'abc', None, 0))).stdout == b'ABC!'
# An empty pipeline is the identity
pl = ProcPipeline()
assert asyncio.run(pl.run(b'abc')).stdout == b'abc'
# -- run() --
# Without a chain, bytes are wrapped and Results pass through
ret = asyncio.run(run(b'abc'))
assert isinstance(ret, Result) and ret.stdout == b'abc'
r = Result(b'xyz', None, 0)
assert asyncio.run(run(r)) is r
# A single filter and a list of filters are both accepted
assert asyncio.run(run(b'abc', Upper())).stdout == b'ABC'
assert asyncio.run(run(b'abc', [Upper(), Exclaim()])).stdout == b'ABC!'
assert asyncio.run(run(b'abc', ProcPipeline([Upper()]))).stdout == b'ABC'
print('All ProcFilter tests passed')