test/unit/python/jw/pkg/lib/Package/Makefile

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

test/unit/python/jw/pkg/lib/Package/test.py

@@ -0,0 +1,80 @@
+from jw.pkg.lib.Package import Package, meta_tags
+
+# -- parse_spec_str --
+
+p = Package.parse_spec_str(
+    'jw-core|jw|Jan|https://example.com/jw-core|Jan <jan@example.com>'
+)
+assert p.name == 'jw-core'
+assert p.vendor == 'jw'
+assert p.packager == 'Jan'
+assert p.url == 'https://example.com/jw-core'
+assert p.maintainer == 'Jan <jan@example.com>'
+
+# An empty field is preserved as an empty string
+p = Package.parse_spec_str('name||||')
+assert p.name == 'name'
+assert p.vendor == ''
+assert p.packager == ''
+assert p.url == ''
+assert p.maintainer == ''
+
+# A custom delimiter works
+p = Package.parse_spec_str('a,b,c,d,e', delimiter = ',')
+assert p.name == 'a'
+assert p.maintainer == 'e'
+
+# Wrong field counts raise
+for spec in ('a', 'a|b', 'a|b|c|d', 'a|b|c|d|e|f'):
+    try:
+        Package.parse_spec_str(spec)
+        assert False, f'Should have raised for "{spec}"'
+    except ValueError:
+        pass
+
+# -- parse_specs_str --
+
+specs = (
+    'jw-core|jw|Jan|https://example.com/jw-core|Jan\n'
+    'jw-base|jw|Jan|https://example.com/jw-base|Jan'
+)
+packages = Package.parse_specs_str(specs)
+assert [p.name for p in packages] == ['jw-core', 'jw-base']
+# A trailing newline does not create a phantom package
+assert len(packages) == 2
+# A trailing newline is not needed
+packages = Package.parse_specs_str(specs + '\n')
+assert [p.name for p in packages] == ['jw-core', 'jw-base']
+
+# An empty string parses to no packages
+assert Package.parse_specs_str('') == []
+
+# -- order_tags --
+
+mapping = {
+    'maintainer': 'Jan',
+    'name': 'jw-core',
+    'url': 'https://example.com/jw-core',
+}
+ordered = Package.order_tags(mapping)
+assert list(ordered) == meta_tags
+assert ordered == {
+    'name': 'jw-core',
+    'vendor': '',
+    'packager': '',
+    'url': 'https://example.com/jw-core',
+    'maintainer': 'Jan',
+}
+
+# -- __repr__ --
+
+p = Package.parse_spec_str('jw-core|jw|Jan|url|Jan')
+assert repr(p) == (
+    'name           : jw-core\n'
+    'vendor         : jw\n'
+    'packager       : Jan\n'
+    'url            : url\n'
+    'maintainer     : Jan'
+)
+
+print('All Package tests passed')

test/unit/python/jw/pkg/lib/PackageFilter/Makefile

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

test/unit/python/jw/pkg/lib/PackageFilter/test.py

@@ -0,0 +1,34 @@
+from jw.pkg.lib.Package import Package
+from jw.pkg.lib.PackageFilter import PackageFilterString
+
+pkg = Package(name = 'jw-core', url = 'https://example.com/jw-core')
+pkg_other = Package(name = 'jw-base', url = 'https://other.org/jw-base')
+pkg_nourl = Package(name = 'no-url')
+
+# -- url=~ filters --
+
+f = PackageFilterString('url=~example\\.com')
+assert f.match(pkg)
+assert not f.match(pkg_other)
+# A package without a url never matches
+assert not f.match(pkg_nourl)
+
+# Whitespace around the operator is tolerated
+f = PackageFilterString('  url  =~  example')
+assert f.match(pkg)
+
+# An unanchored regex matches anywhere in the url
+f = PackageFilterString('url=~jw-core$')
+assert f.match(pkg)
+assert not f.match(pkg_other)
+
+# -- Unsupported definitions raise --
+
+for definition in ('', 'url=example.com', 'name=jw-core', 'url ~ example'):
+    try:
+        PackageFilterString(definition)
+        assert False, f'Should have raised for "{definition}"'
+    except Exception:
+        pass
+
+print('All PackageFilter tests passed')

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

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')

test/unit/python/jw/pkg/lib/TarIo/Makefile

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

test/unit/python/jw/pkg/lib/TarIo/test.py

@@ -0,0 +1,53 @@
+import io
+import tarfile
+
+from jw.pkg.lib.TarIo import TarIoTarFile
+
+def make_tar(files: dict[str, bytes]) -> bytes:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj = buf, mode = 'w') as tf:
+        for name, content in files.items():
+            info = tarfile.TarInfo(name)
+            info.size = len(content)
+            tf.addfile(info, io.BytesIO(content))
+    return buf.getvalue()
+
+def members(blob: bytes) -> dict[str, bytes | None]:
+    ret = {}
+    with tarfile.open(fileobj = io.BytesIO(blob)) as tf:
+        for info in tf.getmembers():
+            f = tf.extractfile(info)
+            ret[info.name] = f.read() if f is not None else None
+    return ret
+
+# -- Bypass the constructor: _filter_tar_file() and _match() do not use
+# the copy contexts, so a bare instance is sufficient
+tar = object.__new__(TarIoTarFile)
+
+files = {
+    'a.txt': b'AAA',
+    'dir/b.txt': b'BBB',
+    'c.txt': b'CCC',
+}
+blob = make_tar(files)
+
+# -- _match is exact matching
+assert tar._match('a.txt', ['a.txt', 'c.txt'])
+assert not tar._match('a.txt', ['x.txt'])
+assert not tar._match('a.txt', [])
+
+# -- _filter_tar_file without a filter is a round trip
+ret = tar._filter_tar_file(blob)
+assert members(ret) == files
+
+# -- With a filter, only the exact paths survive
+matched: list[str] = []
+ret = tar._filter_tar_file(blob, ['a.txt', 'c.txt'], matched = matched)
+assert members(ret) == {'a.txt': b'AAA', 'c.txt': b'CCC'}
+assert sorted(matched) == ['a.txt', 'c.txt']
+
+# -- A filter without hits yields an empty tar
+ret = tar._filter_tar_file(blob, ['nope.txt'])
+assert members(ret) == {}
+
+print('All TarIo tests passed')