from __future__ import annotations import asyncio from typing import TYPE_CHECKING, Iterable, cast, override from jw.pkg.lib.Distro import Distro from jw.pkg.lib.distros.suse.Distro import Distro as SuseDistro from jw.pkg.lib.ec.Local import Local from jw.pkg.lib.Package import Package from jw.pkg.lib.PackageFilter import PackageFilterString from jw.pkg.lib.Result import Result if TYPE_CHECKING: from jw.pkg.lib.ExecContext import ExecContext from jw.pkg.lib.FlaggedFile import FlaggedFile OS_RELEASE_UBUNTU = """\ NAME="Ubuntu" VERSION="24.04.2 LTS (Noble Numbat)" ID=ubuntu ID_LIKE=debian PRETTY_NAME="Ubuntu 24.04.2 LTS" VERSION_ID="24.04" VERSION_CODENAME=noble HOME_URL="https://www.ubuntu.com/" """ OS_RELEASE_SUSE = """\ NAME="openSUSE Tumbleweed" # VERSION="20260901" ID=opensuse-tumbleweed ID_LIKE="opensuse suse" PRETTY_NAME="openSUSE Tumbleweed" VERSION_ID="20260901" HOME_URL="https://www.opensuse.org/" """ class DummyDistro(Distro): @override async def _ref(self) -> None: pass @override async def _dup(self, download_only: bool) -> None: pass @override async def _reboot_required(self, verbose: bool) -> bool: return False @override async def _select_by_name(self, names: Iterable[str]) -> Iterable[Package]: return [] @override async def _install(self, names: Iterable[str], only_update: bool) -> None: pass @override async def _delete(self, names: Iterable[str]) -> None: pass @override async def _pkg_files(self, names: list[str] | None) -> Iterable[str]: return [] @override async def _verify(self, names: Iterable[str]) -> list[FlaggedFile]: return [] # -- verify() -- class VerifyDistro(DummyDistro): def __init__( self, ec: ExecContext, id: str, os_release_str: str | None = None ) -> None: super().__init__(ec, id, os_release_str) self.verified: list[str] | None = None @override async def _verify(self, names: Iterable[str]) -> list[FlaggedFile]: self.verified = list(names) return [] vd = VerifyDistro(ec = cast('ExecContext', object()), id = 'suse') # -- Given names are passed to the distro backend untouched assert asyncio.run(vd.verify(names = ['a', 'b'])) == [] assert vd.verified == ['a', 'b'] # -- Without names, the (empty) selection is verified, so the # backend is not invoked vd.verified = None assert asyncio.run(vd.verify()) == [] assert vd.verified is None # -- pkg_files -- # -- Without a filter, the selection is passed through as the # whole system; the paths come back unfiltered class TmplDistro(DummyDistro): def __init__(self) -> None: super().__init__(ec = cast('ExecContext', object()), id = 'suse') self.files: dict[str, list[str]] = {} @override async def _select_by_name(self, names: Iterable[str]) -> list[Package]: return [Package(n) for n in self.files] @override async def _pkg_files(self, names: list[str] | None) -> list[str]: if names is None: names = list(self.files) ret: list[str] = [] for n in names: ret.extend(self.files[n]) return ret t = TmplDistro() t.files = { 'a': ['/usr/share/a/a.conf.jw-tmpl', '/usr/bin/a'], 'b': ['/etc/b/b.conf.jw-tmpl'], } assert sorted(asyncio.run(t.pkg_files())) == [ '/etc/b/b.conf.jw-tmpl', '/usr/bin/a', '/usr/share/a/a.conf.jw-tmpl' ] # -- A single package name as a str is accepted for compatibility assert sorted(asyncio.run(t.pkg_files('a'))) == [ '/usr/bin/a', '/usr/share/a/a.conf.jw-tmpl', ] def distro(os_release_str: str) -> Distro: # -- The execution context is stored but not used by the properties # under test, so a dummy object is sufficient return DummyDistro( ec = cast('ExecContext', object()), id = Distro.parse_os_release_field_id(os_release_str), os_release_str = os_release_str ) # -- parse_os_release_field -- assert Distro.parse_os_release_field('ID', OS_RELEASE_UBUNTU) == 'ubuntu' assert Distro.parse_os_release_field('VERSION_CODENAME', OS_RELEASE_UBUNTU) == \ 'noble' # Quoted values are de-quoted assert Distro.parse_os_release_field('NAME', OS_RELEASE_UBUNTU) == 'Ubuntu' # ID must not match ID_LIKE assert Distro.parse_os_release_field('ID_LIKE', OS_RELEASE_UBUNTU) == 'debian' try: Distro.parse_os_release_field('NO_SUCH_KEY', OS_RELEASE_UBUNTU) assert False, 'Should have raised' except Exception: pass # -- parse_os_release_field_id -- assert Distro.parse_os_release_field_id(OS_RELEASE_UBUNTU) == 'ubuntu' assert Distro.parse_os_release_field_id(OS_RELEASE_SUSE) == 'suse' # -- Properties -- d = distro(OS_RELEASE_UBUNTU) assert d.id == 'ubuntu' assert d.name == 'Ubuntu' assert d.codename == 'noble' assert d.os == 'ubuntu-noble' assert d.pkg_ext == 'debian' assert d.os_cascade == ['os', 'linux', 'pkg-debian', 'pm-apt', 'ubuntu', 'ubuntu-noble'] assert d.cascade == 'os linux pkg-debian pm-apt ubuntu ubuntu-noble' s = distro(OS_RELEASE_SUSE) assert s.id == 'suse' assert s.codename == 'tumbleweed' assert s.os == 'suse-tumbleweed' assert s.pkg_ext == 'rpm' assert s.os_cascade == [ 'os', 'linux', 'pkg-rpm', 'pm-zypper', 'suse', 'suse-tumbleweed' ] # A codename with a version number cascades down to its prefixes v = distro('ID=fake\nVERSION_CODENAME=1.2.3\n') assert v.os_cascade == ['os', 'fake-1.2', 'fake-1', 'fake', 'fake-1.2.3'] # -- expand_macros -- assert d.expand_macros('%{id}') == 'ubuntu' assert d.expand_macros('%{name}') == 'Ubuntu' assert d.expand_macros('%{os}') == 'ubuntu-noble' assert d.expand_macros('%{codename}') == 'noble' assert d.expand_macros('%{pkg-ext}') == 'debian' assert d.expand_macros('%{cascade}') == d.cascade # %{os-release} expands to the whole /etc/os-release content assert d.expand_macros('%{os-release}') == d.os_release_str assert d.expand_macros('%{os-release}') == OS_RELEASE_UBUNTU assert d.os_release == d.os_release_str # A mix of macros and literal text assert d.expand_macros('running on %{id} (%{codename}) now') == \ 'running on ubuntu (noble) now' # List input expands entry by entry assert d.expand_macros(['%{id}', '%{name}']) == ['ubuntu', 'Ubuntu'] # An unknown macro raises try: d.expand_macros('%{no-such-macro}') assert False, 'Should have raised' except Exception: pass # -- macros -- # -- The doubled percent escapes to a single one when the macros are # embedded in an argparse help string assert Distro.macros() == ['%%{' + name + '}' for name in Distro.macro_names] for macro in Distro.macros(): # -- Every advertised macro must expand without error once the # argparse help escaping has been undone d.expand_macros(macro.replace('%%', '%')) # -- The rpm distro's batched enumeration -- # -- Without a package filter, the selection is the whole system, # which is listed without package names; with one, the selected # names are listed in one command, and a selection without # matches issues no listing at all class CannedRpmLocal(Local): def __init__(self) -> None: super().__init__('local') self.cmds: list[list[str]] = [] @override async def _run( self, cmd: list[str], wd: str | None, verbose: bool, cmd_input: bytes | None, mod_env: dict[str, str] | None, interactive: bool, log_prefix: str, ) -> Result: self.cmds.append(cmd) if '-qal' in cmd: out = '/usr/bin/x\n/etc/a/a.conf.jw-tmpl\n' elif '-ql' in cmd: out = '/etc/a/a.conf.jw-tmpl\n/etc/b/b.conf.jw-tmpl\n' else: out = 'a|v|p|https://jw.dev/a|m\nb|v|p|https://jw.dev/b|m\n' return Result(stdout = out.encode(), status = 0) cl = CannedRpmLocal() sd = SuseDistro(cl, id = 'suse', os_release_str = OS_RELEASE_SUSE) assert asyncio.run(sd.pkg_files()) == [ '/usr/bin/x', '/etc/a/a.conf.jw-tmpl', ] assert cl.cmds[-1] == ['/usr/bin/rpm', '-qal'] sdf = SuseDistro( cl, id = 'suse', os_release_str = OS_RELEASE_SUSE, default_pkg_filter = PackageFilterString('url=~jw.dev') ) assert asyncio.run(sdf.pkg_files()) == \ ['/etc/a/a.conf.jw-tmpl', '/etc/b/b.conf.jw-tmpl'] assert cl.cmds[-1] == ['/usr/bin/rpm', '-ql', 'a', 'b'] sde = SuseDistro( cl, id = 'suse', os_release_str = OS_RELEASE_SUSE, default_pkg_filter = PackageFilterString('url=~zzz-nomatch') ) assert asyncio.run(sde.pkg_files()) == [] assert '--queryformat' in cl.cmds[-1] print('All Distro tests passed')