diff --git a/test/unit/python/jw/pkg/lib/ProcFilter/Makefile b/test/unit/python/jw/pkg/lib/ProcFilter/Makefile new file mode 100644 index 00000000..de693650 --- /dev/null +++ b/test/unit/python/jw/pkg/lib/ProcFilter/Makefile @@ -0,0 +1,7 @@ +TOPDIR = ../../../../../../.. + +include $(TOPDIR)/make/proj.mk +include $(JWBDIR)/make/py-run.mk + +all: +test: run diff --git a/test/unit/python/jw/pkg/lib/ProcFilter/test.py b/test/unit/python/jw/pkg/lib/ProcFilter/test.py new file mode 100644 index 00000000..250c9b8f --- /dev/null +++ b/test/unit/python/jw/pkg/lib/ProcFilter/test.py @@ -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')