# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
from __future__ import annotations
import ast
import inspect
import os
import re
import sys
import textwrap
from functools import cached_property
from pathlib import Path, PurePath
from string import ascii_letters, ascii_lowercase, digits
from typing import ClassVar
from urllib.parse import urlsplit
import pytest
from extra_platforms import ALL_PLATFORMS, Platform, is_windows
from meta_package_manager import (
cli_explore,
cli_maintenance,
managers as managers_module,
)
from meta_package_manager.capabilities import (
METHOD_OPERATIONS,
DelegatedMethod,
Operations,
)
from meta_package_manager.cli import XKCD_MANAGER_ORDER
from meta_package_manager.definitions import (
ALL_DEFINITION_OPERATIONS,
OPERATION_FACTORIES,
OVERRIDES_SECTION,
ConfigDrivenManager,
)
from meta_package_manager.execution import CLIExecutor
from meta_package_manager.manager import OPERATION_KEYS_BY_METHOD, PackageManager
from meta_package_manager.package import Package
from meta_package_manager.pool import pool
from meta_package_manager.version import TokenizedString
from .conftest import PROJECT_ROOT, all_managers, manager_classes_params, tomllib
from .fake_manager import FakeManager
""" Test the structure, data and types returned by all package managers.
This test suite try to automate most of the basic reviewing work for the addition of
new package manager definitions.
"""
[docs]
def test_xkcd_set():
assert len(set(XKCD_MANAGER_ORDER)) == len(XKCD_MANAGER_ORDER)
assert set(pool.all_manager_ids).issuperset(XKCD_MANAGER_ORDER)
[docs]
@all_managers
def test_unmaintained(manager):
"""An unmaintained manager documents itself with `unmaintained_message`.
The message must be non-empty, and any URL embedded in that markdown message must be
a well-formed `http(s)` link.
Conversely, an `unmaintained_message` is only meaningful on an unmaintained
manager.
"""
if manager.unmaintained:
assert manager.unmaintained_message, (
f"Unmaintained manager {manager.id!r} must set an unmaintained_message."
)
if manager.unmaintained_message is not None:
assert manager.unmaintained is True
assert manager.unmaintained_message.strip()
for url in re.findall(r"https?://[^\s)]+", manager.unmaintained_message):
location = urlsplit(url)
assert location
assert location.scheme.lower() in ("http", "https")
[docs]
@all_managers
def test_maintenance_note(manager):
"""A `maintenance_note` is MyST markdown with well-formed URLs.
Its embedded URLs must be well-formed `http(s)` links.
It is mutually exclusive with the `unmaintained` flag: a confirmed-dead manager
documents itself with an `unmaintained_message` instead of a watch note.
"""
if manager.maintenance_note is None:
return
assert manager.maintenance_note.strip()
assert manager.unmaintained is False, (
f"Manager {manager.id!r} cannot be both unmaintained and carry a "
"maintenance_note."
)
for url in re.findall(r"https?://[^\s)]+", manager.maintenance_note):
location = urlsplit(url)
assert location
assert location.scheme.lower() in ("http", "https")
[docs]
@all_managers
def test_operation_notes(manager):
"""Every declared operation note keys a real operation and stays one full sentence.
The *Notes* column of the manager pages renders each entry verbatim beside
its operation row, so a key outside the `Operations` vocabulary
would never reach a page, and a note spanning lines, or not reading as a
standalone sentence (no capital to open it, no period to close it, or an
opening code span a capital cannot sit on), would break the column's shape.
"""
for op_name, note in manager.operation_notes.items():
assert op_name in {op.name for op in Operations}
assert note.strip(), f"Empty operation note for {op_name!r}."
assert "\n" not in note, f"Multi-line operation note for {op_name!r}."
assert note[0].isupper(), (
f"Operation note {op_name!r} does not start with a capital."
)
assert not note.startswith("`"), (
f"Operation note {op_name!r} starts on a code span."
)
assert note.endswith("."), (
f"Operation note {op_name!r} does not end with a period."
)
def _privileged_methods(cls: type[PackageManager]) -> set[str]:
"""Names of the methods of `cls` whose body passes `sudo=True` to a call.
Each name resolves through the MRO, so a subclass answers for the methods it
inherits, and a delegated method answers with its source. The scan stops at
the base classes, which build no privileged command of their own.
"""
bases = (PackageManager, CLIExecutor, object)
names = {name for klass in cls.mro() if klass not in bases for name in vars(klass)}
marked = set()
for name in names:
owner = next(klass for klass in cls.mro() if name in vars(klass))
if owner in bases:
continue
member = vars(owner)[name]
if isinstance(member, DelegatedMethod):
member = member.method
elif isinstance(member, property):
member = member.fget
elif isinstance(member, cached_property):
member = member.func
elif isinstance(member, (staticmethod, classmethod)):
member = member.__func__
if not inspect.isfunction(member):
continue
tree = ast.parse(textwrap.dedent(inspect.getsource(member)))
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
for keyword in node.keywords:
if keyword.arg != "sudo":
continue
# A computed value would hide which way the call goes.
assert isinstance(keyword.value, ast.Constant), (
f"{owner.__name__}.{name} passes a computed `sudo` value."
)
if keyword.value.value:
marked.add(name)
return marked
[docs]
@all_managers
def test_privileged_operations(manager):
"""`privileged_operations` names exactly the `sudo=True` operations.
The operations whose methods carry a `sudo=True` marker.
`prime_sudo` reads the set to choose the managers of the up-front password
prompt. A missing operation lets an escalated command meet a cold cache
inside the fan-out, and an extra one asks for a password no command uses. A
marker on a method outside `METHOD_OPERATIONS` belongs to no operation, so it
fails the test too. A config-defined manager derives the set from its
definition, which is checked against the raw TOML.
"""
assert manager.privileged_operations <= {op.name for op in Operations}
if isinstance(manager, ConfigDrivenManager):
assert manager.definition_source
document = tomllib.loads(
(PROJECT_ROOT / manager.definition_source).read_text(encoding="UTF-8"),
)
definition = document["mpm"][OVERRIDES_SECTION][manager.id]
marked = {
OPERATION_FACTORIES[op_name][0]
for op_name, spec in definition.get("operations", {}).items()
if spec.get("sudo")
}
else:
marked = _privileged_methods(type(manager))
unmapped = marked - METHOD_OPERATIONS.keys()
assert not unmapped, (
f"{manager.id} marks {sorted(unmapped)} privileged, which no operation "
"claims in METHOD_OPERATIONS."
)
expected = {METHOD_OPERATIONS[name].name for name in marked}
assert manager.privileged_operations == expected, (
f"{manager.id} declares {sorted(manager.privileged_operations)}, but its "
f"markers make it {sorted(expected)}."
)
[docs]
@all_managers
def test_ascii_id(manager):
"""All package manager IDs should be short ASCII strings."""
assert manager.id
assert manager.id.isascii()
assert manager.id[0] in ascii_lowercase
assert manager.id[-1] in ascii_lowercase + digits
assert set(manager.id).issubset(ascii_lowercase + digits + "-")
# Make sure there is no potential conflix by prefixing a manager ID by
# "no", which is the convention for single manager flags in the CLI.
assert not manager.id.startswith("no")
[docs]
@all_managers
def test_name(manager):
"""Check all managers have a name."""
assert manager.name
assert set(manager.name).issubset(ascii_letters + digits + " '-")
[docs]
def test_unique_names():
assert len({m.name for m in pool.values()}) == len(pool.all_manager_ids)
[docs]
@all_managers
def test_homepage_url(manager):
"""A home page is optional where the repository stands for it."""
assert manager.homepage_url or manager.repository_url
if manager.homepage_url is None:
return
location = urlsplit(manager.homepage_url)
assert location
assert location.scheme.lower() in ("http", "https")
[docs]
@all_managers
def test_repository_url(manager):
"""A repository is optional, since proprietary tools have none."""
if manager.repository_url is None:
return
location = urlsplit(manager.repository_url)
assert location.scheme == "https"
assert location.hostname
assert "wikipedia.org" not in location.hostname
[docs]
@all_managers
def test_documentation_url(manager):
"""A documentation link is optional, and always a page of its own.
A tool documenting itself in its repository README alone declares none, the
repository link already reaching it, and neither does a project whose home
page is its manual. What is declared names the tool, so an article about it
stays in `wikipedia_url`, and every link is served over TLS.
"""
if manager.documentation_url is None:
return
location = urlsplit(manager.documentation_url)
assert location.scheme == "https"
assert location.hostname
assert "wikipedia.org" not in location.hostname
URL_ATTRIBUTES = tuple(
sorted(name for name in dir(PackageManager) if name.endswith("_url"))
)
"""Every link attribute a manager may declare, found by name rather than listed."""
def _address(url: str) -> str:
"""Reduce a URL to what tells two addresses of one page apart.
The scheme, a `www.` prefix, a trailing slash and letter case never do on the
hosts these links point at, where `ivan-hc/AM` and `ivan-hc/am` are the same
repository.
"""
location = urlsplit(url)
host = (location.hostname or "").casefold().removeprefix("www.")
return f"{host}{location.path.rstrip('/').casefold()}"
[docs]
@all_managers
def test_url_attributes_are_distinct(manager):
"""Check a manager's links each point somewhere the others do not.
A home page that is the repository is declared once, as `repository_url`:
stating it twice links the same page twice from the manager's card, under
two labels. Every link belongs to the class declaring it, never inherited,
since a subclass often wraps a project of its own.
"""
assert URL_ATTRIBUTES, "no `*_url` attribute found on PackageManager"
declared: dict[str, str] = {}
for name in URL_ATTRIBUTES:
assert name in vars(type(manager)), f"{manager.id} inherits `{name}`"
url = getattr(manager, name)
if url is None:
continue
duplicate = declared.get(_address(url))
assert duplicate is None, (
f"{manager.id} points `{name}` and `{duplicate}` at the same {url}: "
"keep only the more specific one"
)
declared[_address(url)] = name
DOCUMENTATION_LABEL = re.compile(r"^Documentation:", re.MULTILINE)
"""A `Documentation:` label opening a line of a manager docstring."""
MARKDOWN_LINK_URL = re.compile(r"\]\((https?://(?:[^()\s]|\([^()\s]*\))+)\)")
"""The address of a markdown link, balanced parentheses included."""
[docs]
@all_managers
def test_docstring_repeats_no_card_link(manager):
"""Check a docstring states no link its card already carries.
The manager's page opens on its card, which links every `*_url` attribute,
the documentation entry point among them. A `Documentation:` line therefore
has nothing left to name, and a prose link to an address the card holds
states it twice on one page. A further reference keeps its own sentence,
unlabelled, the way the command-equivalence tables do.
"""
own = {_address(url) for name in URL_ATTRIBUTES if (url := getattr(manager, name))}
docstring = inspect.getdoc(type(manager)) or ""
assert not DOCUMENTATION_LABEL.search(docstring), (
f"{manager.id} labels a reference `Documentation:` in its docstring; "
"declare `documentation_url` instead"
)
for url in MARKDOWN_LINK_URL.findall(docstring):
assert _address(url) not in own, (
f"{manager.id} links {url} in its docstring, which its card links"
)
[docs]
@all_managers
def test_requirement(manager):
"""Each manager is required to specify a version requirement or `None`."""
if manager.requirement is not None:
assert set(manager.requirement).issubset(digits + ".>=<!, ")
# Check each version component is lossless once passed via TokenizedString.
for part in manager.requirement.split(","):
part = part.strip().lstrip("><=!")
if part:
assert str(TokenizedString(part)) == part
[docs]
@all_managers
def test_cli_names_type(manager):
"""Check the pointed CLI name and path are file-system compatible."""
assert manager.cli_names
for name in manager.cli_names:
# Underscores appear in real binary names, like OpenBSD's pkg_info.
assert name.replace("-", "").replace(".", "").replace("_", "").isalnum()
assert PurePath(name).name == name
[docs]
@all_managers
def test_cli_search_path(manager):
assert len(set(manager.cli_search_path)) == len(manager.cli_search_path)
for search_path in manager.cli_search_path:
path_obj = Path(search_path).resolve()
assert path_obj.is_absolute()
# Reserved names only exist on Windows: POSIX's is_reserved() always returned
# False and was removed in Python 3.15, so the check is Windows-gated.
if is_windows():
if sys.version_info >= (3, 13):
assert not os.path.isreserved(path_obj)
else:
assert not path_obj.is_reserved()
if path_obj.exists():
assert path_obj.is_file() or path_obj.is_dir()
[docs]
@all_managers
def test_global_args_type(manager):
"""Check that definitions returns CLI args as a list of non-empty strings."""
for global_args in (manager.pre_cmds, manager.pre_args, manager.post_args):
assert all(global_args)
for arg in global_args:
assert set(arg).issubset(ascii_letters + digits + "-=+")
[docs]
@all_managers
def test_version_cli_options(manager):
"""Version CLI options must be a list of non empty strings."""
assert all(manager.version_cli_options)
[docs]
@all_managers
def test_version_regexes(manager):
"""Version regex is required.
Check it compiles and match has a `<version>` group.
"""
assert isinstance(manager.version_regexes, tuple)
for version_regex in manager.version_regexes:
assert isinstance(version_regex, str)
assert version_regex
regex = re.compile(version_regex)
assert "version" in regex.groupindex
[docs]
@pytest.mark.parametrize(
("manager_id", "sample_output", "expected_version"),
(
(
"guix",
"guix (GNU Guix) 1.4.0\nCopyright (C) 2026 the Guix authors\n",
"1.4.0",
),
(
"guix",
"guix (GNU Guix) 1.4.0-2563-gabc1234\nCopyright (C) 2026\n",
"1.4.0-2563-gabc1234",
),
("guix", "guix (GNU Guix) abc1234\nCopyright (C) 2026\n", "abc1234"),
# eopkg 5.0.0 renamed itself `eopkg.bin` in the banner; 4.x and below
# print the bare name, and both spellings must still parse.
("eopkg", "eopkg.bin 5.0.0\n", "5.0.0"),
("eopkg", "eopkg 4.4.0\n", "4.4.0"),
("mise", "2026.6.3 macos-arm64 (2026-06-13)\n", "2026.6.3"),
("nix", "nix-env (Nix) 2.18.1\n", "2.18.1"),
("stew", "stew version v0.7.0\n", "0.7.0"),
("steamcmd", "Valve Corporation - version 1234567890\n", "1234567890"),
),
)
def test_version_regex_matches_sample(manager_id, sample_output, expected_version):
"""Each manager's `version_regexes` parses the documented `--version` output.
Regression guard for the case where `re.VERBOSE` silently swallowed literal
spaces in patterns like `guix \\(GNU Guix\\) ...`, leaving the manager
unable to detect its own version.
"""
manager = pool[manager_id]
matched = None
for regex in manager.version_regexes:
match = re.search(regex, sample_output, re.MULTILINE)
if match and match.group("version"):
matched = match.group("version")
break
assert matched == expected_version
EOPKG_LISTING = (
"Package Name |St| Version| Rel.| Distro| Date\n"
"===========================================================================\n"
"aalib | i| 1.4.0_5| 9| Solus|07 Sep 2026 10:08\n"
"zram-generator-defaults | i| 1.2.1| 7| Solus|07 Sep 2026 10:09\n"
"zstd | i| 1.5.7| 33| Solus|07 Sep 2026 10:09\n"
"zxing-cpp | i| 2.3.0| 6| Solus|07 Sep 2026 10:09\n"
)
"""A four-package `eopkg` listing, header and `===` separator included.
`zram-generator-defaults` overflows the first column, pushing its pipe right,
and the last line is a package rather than a footer.
"""
[docs]
@pytest.mark.parametrize("member", ("installed", "outdated"))
def test_eopkg_listing_keeps_its_last_rows(member, stub_run_cli):
"""`eopkg`'s listings carry no footer, so no trailing line may be discarded.
Both listings used to drop the last two lines before matching, on the
assumption of a footer eopkg never prints: its final line is a package. The
two packages closing every listing were therefore lost, silently and with a
zero exit, so an inventory looked complete while missing its tail.
This guard is narrower than the bug class it comes from, and deliberately
so: what a parser may discard can only be read from the tool's own output,
not derived from the manager class. It pins the one listing shape measured
against eopkg `4.4.0` and `5.0.0`.
"""
manager = pool["eopkg"]
# `outdated` follows the listing with an `info` call; answering both with
# the listing leaves every candidate unresolved, which is the point here:
# this test is about which rows survive, not about the version join.
stub_run_cli(manager, EOPKG_LISTING)
ids = {package.id for package in getattr(manager, member)}
# The header and `===` separator carry no pipe-delimited version column, so
# they fall through the regex on their own and need no slicing.
assert ids == {"aalib", "zram-generator-defaults", "zstd", "zxing-cpp"}
[docs]
def test_eopkg_installed_reports_the_bare_version(stub_run_cli):
"""`installed` reports the `Version` column alone, release excluded.
Every snapshot carries this string, so the release stays out of it. Only
`outdated` pairs the two, where a release-only rebuild would otherwise read
as the same version on both sides.
"""
manager = pool["eopkg"]
stub_run_cli(manager, EOPKG_LISTING)
versions = {
package.id: str(package.installed_version) for package in manager.installed
}
assert versions == {
"aalib": "1.4.0_5",
"zram-generator-defaults": "1.2.1",
"zstd": "1.5.7",
"zxing-cpp": "2.3.0",
}
[docs]
def test_eopkg_outdated_joins_the_upgrade_candidate(monkeypatch):
"""`outdated` pairs the installed listing with the candidate `info` reports.
`list-upgrades --install-info` describes each package as installed, so the
upgrade target appears nowhere in it. The target comes from a second call,
`info` taking every name at once. Both sides carry the release, or the
`aalib` row below (a release-only rebuild) would read `1.4.0_5` on each side
and look like no upgrade at all.
"""
manager = pool["eopkg"]
info_output = (
"Installed package:\n"
"Name : aalib, version: 1.4.0_5, release: 9\n"
"Summary : An ASCII art library\n"
"\n"
"Package found in Solus repository:\n"
"Name : aalib, version: 1.4.0_5, release: 10\n"
"Summary : An ASCII art library\n"
"\n"
"Installed package:\n"
"Name : zstd, version: 1.5.7, release: 33\n"
"Summary : Zstandard compression\n"
"\n"
"Package found in Solus repository:\n"
"Name : zstd, version: 1.5.8, release: 34\n"
"Summary : Zstandard compression\n"
)
calls: list[tuple[str, ...]] = []
def fake_run_cli(*args, **kwargs) -> str:
calls.append(tuple(str(a) for a in args))
return info_output if args and args[0] == "info" else EOPKG_LISTING
monkeypatch.setattr(manager, "run_cli", fake_run_cli)
packages = {
package.id: (
str(package.installed_version),
str(package.latest_version) if package.latest_version else None,
)
for package in manager.outdated
}
# The candidate is read from the repository section alone: the identically
# shaped `Name :` line of the `Installed package:` section above it must not
# win, or every package would report its own installed version as the target.
assert packages["aalib"] == ("1.4.0_5-9", "1.4.0_5-10")
assert packages["zstd"] == ("1.5.7-33", "1.5.8-34")
# A package `info` said nothing about keeps its row, without a target.
assert packages["zxing-cpp"] == ("2.3.0-6", None)
# One `info` call for the whole listing, not one per package.
info_calls = [call for call in calls if call and call[0] == "info"]
assert len(info_calls) == 1
assert set(info_calls[0][1:]) == {
"aalib",
"zram-generator-defaults",
"zstd",
"zxing-cpp",
}
[docs]
@pytest.mark.parametrize(
("label", "info_output", "expected"),
(
(
"not installed: repository section only",
(
"htop package is not installed\n"
"Package found in Solus repository:\n"
"Name : htop, version: 3.5.3, release: 30\n"
"Summary : htop (interactive process viewer for Linux)\n"
"Description : htop is an interactive process viewer for Linux.\n" # noqa: E501
"Licenses : GPL-2.0-or-later\n"
"Component : system.utils\n"
"Dependencies : ncurses glibc libcap2 lm_sensors \n"
"Distribution : Solus, Dist. Release: 1\n"
"Architecture : x86_64, Installed Size: 397.03 KB, "
"Package Size: 159.05\n"
" KB\n"
"Reverse Dependencies: htop-dbginfo \n"
),
[("htop", "3.5.3", "30")],
),
(
"installed: both sections, only the repository one counts",
(
"Installed package:\n"
"Name : bash, version: 5.3.15, release: 91\n"
"Summary : bash (sh-compatible shell)\n"
"Licenses : GPL-3.0-or-later\n"
"Component : system.base\n"
"Dependencies : readline glibc \n"
"Distribution : Solus, Dist. Release: 1\n"
"Architecture : x86_64, Installed Size: 9.51 MB\n"
"Reverse Dependencies: \n"
"\n"
"Package found in Solus repository:\n"
"Name : bash, version: 5.3.15, release: 91\n"
"Summary : bash (sh-compatible shell)\n"
),
[("bash", "5.3.15", "91")],
),
),
)
def test_eopkg_candidate_regex_reads_real_info_output(label, info_output, expected):
"""The candidate pattern runs against `eopkg info` output captured verbatim.
Both blocks come off a Solus `4.9` host. The second is the one that matters:
an installed package prints an `Installed package:` section whose `Name :`
line is shaped exactly like the repository one, so a pattern keyed on the
field name alone would report the installed version as the upgrade target.
Exactly one match must come back, and it must be the repository's.
"""
matches = [
(m.group("package_id"), m.group("version"), m.group("release"))
for m in pool["eopkg"]._CANDIDATE_REGEXP.finditer(info_output)
]
assert matches == expected, label
[docs]
def test_eopkg_outdated_skips_the_info_call_when_nothing_is_outdated(monkeypatch):
"""An empty listing must not reach `info`, which prints help with no names."""
manager = pool["eopkg"]
header = (
"Package Name |St| Version| Rel.| Distro| Date\n"
"===========================================================================\n"
)
calls: list[tuple[str, ...]] = []
def fake_run_cli(*args, **kwargs) -> str:
calls.append(tuple(str(a) for a in args))
return header
monkeypatch.setattr(manager, "run_cli", fake_run_cli)
assert list(manager.outdated) == []
assert [call[0] for call in calls] == ["list-upgrades"]
[docs]
def test_eopkg_search_decodes_character_references(stub_run_cli):
"""`eopkg search` emits the index's raw summary, character references included.
`list-available` resolves the same text, so the two disagree on the very
same package: only `search` reaches a user through mpm, and it used to show
`ImageMagick® suite` where every other listing says `ImageMagick®
suite`. Measured against eopkg `4.4.0`.
"""
manager = pool["eopkg"]
output = (
"imagemagick - ImageMagick® suite to create, edit, compose\n"
"imagemagick-docs - Documentation for imagemagick\n"
)
stub_run_cli(manager, output)
descriptions = {
package.id: package.description
for package in manager.search("imagemagick", extended=False, exact=False)
}
assert descriptions["imagemagick"] == (
"ImageMagick® suite to create, edit, compose"
)
assert descriptions["imagemagick-docs"] == "Documentation for imagemagick"
[docs]
@pytest.mark.parametrize(
("manager_id", "member", "forbidden_arg"),
(("emerge", "outdated", "--quiet"), ("emerge", "search", "--quiet")),
)
def test_parsed_output_omits_reshaping_arg(
manager_id, member, forbidden_arg, monkeypatch
):
"""An operation whose output feeds a regex must not send a reshaping argument.
Some arguments do not only trim noise, they change the shape of what the
tool prints. `emerge --quiet` is one: `--search` drops to a bare
`* category/name` line without the `Latest version available:` and
`Description:` fields, and `--update --columns` drops the `[ebuild U ]`
state prefix and the brackets around the latest version. Every regex then
matches nothing, and the operation reports an empty set on a system that
has results, exiting zero with no diagnostic.
Verified against Portage `3.0.81.3`. This guard is deliberately narrow: it
names the arguments already known to reshape output rather than deriving
them, because whether an argument is safe can only be read from the tool's
output, not from the manager class.
It is not made redundant by
{func}`~tests.test_docstring_corpus.test_documented_query_command_matches_construction`,
which compares a docstring against the argv beside it: adding a reshaping
argument to the code *and* to the docstring leaves those two agreeing, and
only this test still fails. An agreement holds whichever way both sides
move.
"""
manager = pool[manager_id]
monkeypatch.setattr(
manager, "cli_path", Path("/usr/bin") / manager.cli_names[0], raising=False
)
monkeypatch.setattr(manager, "which", lambda cli_name: Path("/usr/bin") / cli_name)
built = []
def record_run_cli(*args, **kwargs) -> str:
build_kwargs = {
key: value
for key, value in kwargs.items()
if key in {"auto_pre_args", "override_cli_path", "override_pre_args"}
}
built.append([str(token) for token in manager.build_cli(*args, **build_kwargs)])
return ""
monkeypatch.setattr(manager, "run_cli", record_run_cli)
if member == "search":
tuple(manager.search("htop", extended=False, exact=False))
else:
tuple(getattr(manager, member))
assert built, f"{member} built no command to inspect"
for command in built:
assert forbidden_arg not in command, (
f"{manager_id}.{member} sends {forbidden_arg}, which reshapes the "
f"output its regex parses: {command}"
)
[docs]
@all_managers
def test_cli_path(manager):
if manager.cli_path is not None:
assert isinstance(manager.cli_path, Path)
assert manager.cli_path.is_absolute()
# Reserved names only exist on Windows: POSIX's is_reserved() always returned
# False and was removed in Python 3.15, so the check is Windows-gated.
if is_windows():
if sys.version_info >= (3, 13):
assert not os.path.isreserved(manager.cli_path)
else:
assert not manager.cli_path.is_reserved()
assert manager.cli_path.is_file()
[docs]
@pytest.mark.parametrize(
("query", "query_parts"),
(
("cli-l-cli", {"cli", "l"}),
("ab12--cd34", {"ab12", "cd34"}),
("123/extra.34", {"123", "extra", "34"}),
("AB ab", {"AB", "ab"}),
),
)
def test_query_parts(query, query_parts):
assert Package.query_parts(query) == query_parts
[docs]
@pytest.mark.parametrize(
("pkg_id", "pkg_name", "query", "extended", "exact", "expected"),
(
# Fuzzy (default): case-insensitive, tokenized substring on ID and name.
("gnu-sed", None, "sed", False, False, True),
("gnu-sed", None, "SED", False, False, True),
("gnu-sed", None, "GnU", False, False, True),
("gnu-sed", None, "gnu sed", False, False, True),
("gnu-sed", None, "perl", False, False, False),
# Fuzzy also matches on the display name.
("pkg-1", "GNU Sed", "sed", False, False, True),
# Description is ignored unless extended.
("pkg-1", None, "stream", False, False, False),
("pkg-1", None, "stream", True, False, True),
# Exact: whole-string, case-sensitive, ID or name only.
("gnu-sed", None, "gnu-sed", False, True, True),
("gnu-sed", None, "sed", False, True, False),
("gnu-sed", None, "GNU-SED", False, True, False),
("pkg-1", "GNU Sed", "GNU Sed", False, True, True),
# Empty or punctuation-only queries never match.
("gnu-sed", None, "", False, False, False),
("gnu-sed", None, "@@@", False, False, False),
),
)
def test_package_matches(pkg_id, pkg_name, query, extended, exact, expected):
package = Package(
id=pkg_id,
manager_id="fake",
name=pkg_name,
description="a stream editor",
)
assert package.matches(query, extended=extended, exact=exact) is expected
def _collect_class_members(klass: type, name: str) -> tuple[list[str], list[str]]:
"""Collect a base class's `(attributes, methods)` in declaration order.
Parses the AST so naked type annotations (`AnnAssign`) are collected
alongside plain assignments and function definitions.
"""
tree = ast.parse(Path(inspect.getfile(klass)).read_bytes())
class_node = next(
n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == name
)
attributes: list[str] = []
methods: list[str] = []
for node in class_node.body:
if isinstance(node, ast.AnnAssign):
attributes.append(node.target.id) # type: ignore[union-attr]
if isinstance(node, ast.Assign):
attributes.extend(t.id for t in node.targets) # type: ignore[attr-defined]
if isinstance(node, ast.FunctionDef):
methods.append(node.name)
return attributes, methods
CANONICAL_ATTRS = (
# Identity.
"scope",
"unmaintained",
"unmaintained_message",
"maintenance_note",
"operation_notes",
"id",
"name",
"homepage_url",
"documentation_url",
"repository_url",
"wikipedia_url",
"logo",
"keywords",
# Export mappings.
"brewfile_entry_type",
"brewfile_skip_warning",
"platforms",
"virtual",
# Escalation policy.
"sudo",
"default_sudo",
"internal_sudo",
"internal_sudo_operations",
"resets_sudo_timestamp",
"privileged_operations",
# Version gate and native cooldown marker.
"requirement",
"cooldown_env_var",
# CLI plumbing.
"cli_names",
"cli_search_path",
"extra_env",
"pre_cmds",
"pre_args",
"post_args",
"operation_extra_args",
# Version probe.
"version_cli",
"version_cli_options",
"version_regexes",
"version_from_stderr",
# Failure gate.
"stderr_noise",
# Behavior toggles and platform specifics.
"stop_on_error",
"dry_run",
"plan",
"timeout",
"_active_operation",
"_operation_key",
"_dormant_sudo",
"progress",
"cooldown",
"cooldown_policy",
"windows_creation_flags",
"windows_processes_to_cleanup",
"cli_errors",
"_last_run",
"run_cache",
# Parsing constants.
"_NAME_VERSION_REGEXP",
)
"""Canonical declaration order of the attributes a manager class may override.
The documented convention every manager module follows: identity first, then the
escalation policy, the version requirement, the CLI plumbing, the version probe,
and the rarely-overridden toggles. Manager-specific constants (the `_*_REGEXP`
parsers) are not governed and conventionally sit between the attributes and the
operations. Consumed by `test_content_order`, which checks each manager's
relative declaration order against this sequence.
"""
[docs]
def collect_props_ref():
"""Build the canonical member-order reference for manager classes.
Attributes follow {data}`CANONICAL_ATTRS`; methods follow the base classes'
own declaration order, `CLIExecutor`'s (binary discovery, version probe,
execution engine) before `PackageManager`'s (helpers, availability, then
the operations). The curated attribute list is cross-checked against the
base classes so a new base attribute cannot silently escape governance.
"""
pm_attrs, pm_methods = _collect_class_members(PackageManager, "PackageManager")
exec_attrs, exec_methods = _collect_class_members(CLIExecutor, "CLIExecutor")
declared = set(pm_attrs) | set(exec_attrs)
assert set(CANONICAL_ATTRS) == declared, (
"CANONICAL_ATTRS drifted from the base classes: "
f"missing {declared - set(CANONICAL_ATTRS)}, "
f"stale {set(CANONICAL_ATTRS) - declared}"
)
return (*CANONICAL_ATTRS, *exec_methods, *pm_methods)
props_ref = tuple(collect_props_ref())
[docs]
def test_operation_order():
"""Operation IDs are ordered and aligned to the base class and CLI.
Aligned to the base manager class and the CLI implementation.
The subcommands span two section modules: the read-only queries in
`cli_explore`, the state changers and diagnostics in `cli_maintenance`.
Walked in that order, their definitions must follow the enum order.
"""
direct_operation_ids = [op for op in Operations.__members__ if op != "upgrade_all"]
base_operations = [p for p in props_ref if p in direct_operation_ids]
assert list(direct_operation_ids) == list(base_operations)
implemented_operations: list[str] = []
for cli_module in (cli_explore, cli_maintenance):
cli_tree = ast.parse(Path(inspect.getfile(cli_module)).read_bytes())
implemented_operations.extend(
n.name
for n in cli_tree.body
if isinstance(n, ast.FunctionDef) and n.name in direct_operation_ids
)
assert list(direct_operation_ids) == list(implemented_operations)
# Check the code of each file registered in the pool.
[docs]
@manager_classes_params
def test_content_order(manager_class):
"""Lint each manager class to check its members follow the canonical order.
Attributes must follow {data}`CANONICAL_ATTRS` and methods the base classes'
own declaration order (see {func}`collect_props_ref`); ungoverned names
(manager-specific parsing constants and helpers) may sit anywhere.
"""
# Collect in order the IDs of all attributes (plain and annotated
# assignments alike) and functions of the class.
collected_props = []
klass_tree = ast.parse(inspect.getsource(manager_class))
class_node = klass_tree.body[0]
assert isinstance(class_node, ast.ClassDef)
for node in class_node.body:
if isinstance(node, ast.AnnAssign):
collected_props.append(node.target.id) # type: ignore[union-attr]
if isinstance(node, ast.Assign):
collected_props.extend([t.id for t in node.targets]) # type: ignore[attr-defined]
if isinstance(node, ast.FunctionDef):
collected_props.append(node.name)
enforced_props = tuple(p for p in collected_props if p in props_ref)
expected_order = tuple(p for p in props_ref if p in collected_props)
assert enforced_props == expected_order
def _is_line_anchored(pattern: str) -> bool:
"""Does the pattern bind to the start or the end of a line?
Blank lines and `#` comments are stripped first, so a `re.VERBOSE` pattern
opening on its own line is read the same as a compact one.
"""
stripped = re.sub(r"(?m)^\s*(#.*)?$", "", pattern).strip()
return stripped.startswith("^") or bool(re.search(r"(?<!\\)\$$", stripped))
[docs]
def test_anchored_regexes_scanning_whole_output_are_multiline():
"""A line-anchored regex walked over a whole output must set `re.MULTILINE`.
Without the flag `^` binds to the start of the *string*, so `findall` and
`finditer` return at most one match and every result after the first is
dropped: silently, since a short listing is indistinguishable from a real
one. The managers feeding a regex to
{meth}`~meta_package_manager.manager.PackageManager.parse_regex_lines` are
not concerned, that helper splitting the output before matching, which is
why the invariant is keyed on the call site rather than on the pattern.
`flatpak` shipped exactly this bug: its search reported only the first
package because the sample in its docstring carries a single result line.
"""
offenders = []
for path in sorted(Path("meta_package_manager").rglob("*.py")):
tree = ast.parse(path.read_text(encoding="UTF-8"))
compiled: dict[str, tuple[str, bool]] = {}
for node in ast.walk(tree):
if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call):
continue
function = node.value.func
if not (isinstance(function, ast.Attribute) and function.attr == "compile"):
continue
if not node.value.args:
continue
# A pattern built at runtime rather than spelled as a literal is not
# readable here, so it is skipped instead of guessed at.
literal = node.value.args[0]
if not isinstance(literal, ast.Constant) or not isinstance(
literal.value, str
):
continue
flags = ast.unparse(node.value.args[1]) if len(node.value.args) > 1 else ""
for target in node.targets:
if isinstance(target, ast.Name):
compiled[target.id] = (literal.value, "MULTILINE" in flags)
for node in ast.walk(tree):
if not (
isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
):
continue
if node.func.attr not in {"findall", "finditer"}:
continue
owner = node.func.value
name = (
owner.attr
if isinstance(owner, ast.Attribute)
else getattr(owner, "id", None)
)
if name not in compiled:
continue
pattern, multiline = compiled[name]
if _is_line_anchored(pattern) and not multiline:
offenders.append(f"{path.as_posix()}:{node.lineno}: {name}")
assert not offenders, (
"line-anchored regexes scanned over a whole output without "
f"re.MULTILINE: {offenders}"
)
[docs]
def test_argv_literals_carry_no_shell_quoting():
"""No argv literal may wrap itself in the quotes a shell would have eaten.
Managers build an argument list, never a shell line, so a literal like
`'"%n %v %o %c"'` reaches the tool with its quotes intact and is read as
part of the value. `ports` shipped exactly that for years: `pkg` echoed the
quotes back around every record, so each package id began with a `"` and
every description ended with one, and nothing failed loudly enough to
notice. The docstring above it showed the shell form, which is correct for
a human to type and wrong to paste into an argv.
The check is cheap because the population is small and static: every string
constant handed to `run_cli`, `build_cli` or `run` across the package.
"""
offenders = []
for path in sorted(Path(inspect.getfile(managers_module)).parent.rglob("*.py")):
tree = ast.parse(path.read_text(encoding="UTF-8"))
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
name = getattr(node.func, "attr", getattr(node.func, "id", ""))
if name not in {"run_cli", "build_cli", "run"}:
continue
for arg in node.args:
value = getattr(arg, "value", None)
if (
isinstance(value, str)
and len(value) > 1
and value.startswith('"')
and value.endswith('"')
):
offenders.append(f"{path.name}:{arg.lineno}: {value!r}")
assert not offenders, (
"these argv literals carry shell quoting the tool will read as part of "
"the value: " + ", ".join(offenders)
)
[docs]
def test_reads_inside_mutating_methods_survive_plan_mode():
"""A read a mutating method needs must be exempted from plan capture.
`mpm --plan` decides what to record from the operation in flight, not from
the command: any CLI call made while `install`, `remove` or `upgrade` is
stamped gets written to the plan recorder and hands its caller an empty
string. A method that queries before it acts therefore reads back nothing
and concludes the package does not exist. `ports` did exactly that, failing
every `mpm --plan install` with `Could not resolve port origin`, and
`sdkman` reported every installed candidate as missing.
Two exemptions are accepted, matching the two shapes in the codebase: the
helper passes `force_exec=True` (which lifts plan for that one call), or the
caller re-stamps the block with `acting_as` naming the read it performs. The
mutating call itself must stay outside such a stamp, or plan mode would run
the change it exists to only print; that half is not checkable here and is
left to review.
The check is deliberately narrow. It knows two ways a method triggers a
read, the `installed`/`outdated` properties and a `_resolve*` helper, so a
read reached by any other route passes unnoticed. It covers the population
that has produced the bug twice, not every way of writing it.
"""
mutating = {
"cleanup_cache",
"cleanup_orphan",
"cleanup_repair",
"install",
"mark_explicit",
"remove",
"remove_orphan",
"restore",
"sync",
"upgrade",
"upgrade_all_cli",
"upgrade_one_cli",
}
read_properties = {"installed", "outdated"}
def forces_exec(class_node: ast.ClassDef, method_name: str) -> bool:
"""Does the named sibling method pass `force_exec=True` to any call?"""
for node in class_node.body:
if (
isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name == method_name
):
for call in (n for n in ast.walk(node) if isinstance(n, ast.Call)):
for keyword in call.keywords:
if keyword.arg == "force_exec" and (
getattr(keyword.value, "value", None) is True
):
return True
return False
offenders = []
for path in sorted(Path("meta_package_manager").rglob("*.py")):
tree = ast.parse(path.read_text(encoding="UTF-8"))
for class_node in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)):
methods = [
n
for n in class_node.body
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))
]
for method in methods:
if method.name not in mutating:
continue
stamped = any(
isinstance(n, ast.Attribute) and n.attr == "acting_as"
for n in ast.walk(method)
)
if stamped:
continue
triggers = set()
for node in ast.walk(method):
if (
isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id == "self"
and node.attr in read_properties
):
triggers.add(f"self.{node.attr}")
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "self"
and node.func.attr.startswith("_resolve")
and not forces_exec(class_node, node.func.attr)
):
triggers.add(f"self.{node.func.attr}()")
if triggers:
offenders.append(
f"{class_node.name}.{method.name} reads "
f"{sorted(triggers)} in {path}"
)
assert not offenders, (
"reads issued inside a mutating method are captured by --plan and return "
f"an empty string; exempt them with force_exec=True or acting_as: {offenders}"
)
[docs]
def test_operation_vocabularies_agree():
"""The operation, capability and schema maps name the same methods.
The method-to-operation map, the capability map and the definition schema name the
same methods and operations.
`OPERATION_KEYS_BY_METHOD` decides which `operations` entry a command takes its
`extra_args` from. A method it misses builds commands that no entry reaches, and
a name that differs from `OPERATION_FACTORIES` makes an override and a
definition disagree on what the operation is called.
"""
assert set(OPERATION_KEYS_BY_METHOD) == set(METHOD_OPERATIONS)
for op_name, (method_name, _) in OPERATION_FACTORIES.items():
assert OPERATION_KEYS_BY_METHOD[method_name] == op_name
assert set(OPERATION_KEYS_BY_METHOD.values()) == ALL_DEFINITION_OPERATIONS
[docs]
def test_operation_methods_stamp_their_own_entry(monkeypatch):
"""A command gets the `extra_args` of the method that builds it.
Three shapes would fool a stamp keyed on the routed operation: a query that
runs lazily, after its property returned; a query nested in another, like an
`outdated` that reads `installed`; and the one-by-one fallback of a full
upgrade, whose commands are `upgrade_one` commands.
"""
class Recorder(FakeManager):
id = "recorder"
recorded: ClassVar[list[tuple[str, str | None]]] = []
@property
def installed(self):
self.recorded.append(("installed", self._operation_key))
yield from ()
@property
def outdated(self):
self.recorded.append(("outdated", self._operation_key))
list(self.installed)
self.recorded.append(("outdated", self._operation_key))
yield self.package(id="lemon", installed_version="1.0")
def upgrade_all_cli(self):
raise NotImplementedError
def upgrade_one_cli(self, package_id, version=None):
self.recorded.append(("upgrade_one_cli", self._operation_key))
return self.build_cli("upgrade", package_id)
manager = Recorder()
monkeypatch.setattr(manager, "run", lambda *args, **kwargs: "")
outdated = manager.outdated
# The generator body has not run yet: no stamp recorded before iteration.
assert manager.recorded == []
assert [package.id for package in outdated] == ["lemon"]
manager.upgrade()
assert manager.recorded == [
("outdated", "outdated"),
("installed", "installed"),
("outdated", "outdated"),
("outdated", "outdated"),
("installed", "installed"),
("outdated", "outdated"),
("upgrade_one_cli", "upgrade_one"),
]
assert manager._operation_key is None
[docs]
def declared_keywords(manager) -> tuple[str, ...]:
"""Keywords the manager's own class body sets, ignoring anything inherited.
A subclass reuses its parent's implementation, which is not a second claim on
the parent's names: `apt-mint` inherits `debian` from `apt`, and every AUR
helper inherits `arch` from `pacman`. Generation reads the *inherited* value,
so a label still collects its members' terms through a set union, but
ownership is only what a class declares for itself.
"""
# Annotated local, as in `_load_benchmark_toml`: a class `__dict__` is typed
# as `Any`, and this laundering keeps the function clean of `no-any-return`
# without an ignore that a different mypy then calls unused.
keywords: tuple[str, ...] = type(manager).__dict__.get("keywords", ())
return keywords
[docs]
def test_manager_keywords_have_one_owner():
"""No two managers declare the same keyword alias.
A keyword names exactly one manager, so a term spanning several of them (an
ecosystem name like `zsh`, answering for five plugin managers) belongs in the
globally curated `KEYWORDS_EXTRAS` instead, which owes nothing to a single
class. Without this, two managers could disagree about who a name refers to
and both would ship it.
"""
owners: dict[str, list[str]] = {}
for manager in pool.values():
for keyword in declared_keywords(manager):
owners.setdefault(keyword.casefold(), []).append(manager.id)
shared = {keyword: sorted(ids) for keyword, ids in owners.items() if len(ids) > 1}
assert not shared, f"Keywords claimed by more than one manager: {shared}"
[docs]
def test_manager_keywords_are_not_manager_ids():
"""A keyword never restates a manager ID.
Every ID already reaches the PyPI keywords straight from the pool, so
repeating one is redundant. Repeating *another* manager's ID is worse, since
it would file that manager's own name under this class.
"""
ids = {manager_id.casefold() for manager_id in pool.all_manager_ids}
offenders = {}
for manager in pool.values():
clashing = sorted(k for k in declared_keywords(manager) if k.casefold() in ids)
if clashing:
offenders[manager.id] = clashing
assert not offenders, f"Keywords restating a manager ID: {offenders}"
[docs]
def test_manager_keywords_are_normalized():
"""Keywords are stripped and lowercase.
They are merged into one case-insensitively sorted set, so a stray capital or
a trailing space would produce a near-duplicate entry that the uniqueness
check above cannot see.
"""
offenders = {}
for manager in pool.values():
malformed = sorted(
k for k in declared_keywords(manager) if k != k.strip().casefold() or not k
)
if malformed:
offenders[manager.id] = malformed
assert not offenders, f"Keywords that are not stripped and lowercase: {offenders}"
def _parsed_json_subscripts(source: str) -> tuple[str, ...]:
"""Constant-key subscripts applied to the result of a `parse_json()` call.
Walks each function for the names bound to a `self.parse_json(...)` result,
then reports every `name["key"]` reading one of them. Scoped to that one
helper: a parser decoding JSON some other way is out of reach here, and
nothing in the pool currently does.
"""
findings: list[str] = []
for func in ast.walk(ast.parse(source)):
if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
parsed: set[str] = set()
for node in ast.walk(func):
if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call):
called = node.value.func
if isinstance(called, ast.Attribute) and called.attr == "parse_json":
parsed.update(
target.id
for target in node.targets
if isinstance(target, ast.Name)
)
findings.extend(
f"{func.name}() line {node.lineno}: {node.value.id}[{node.slice.value!r}]"
for node in ast.walk(func)
if isinstance(node, ast.Subscript)
and isinstance(node.value, ast.Name)
and node.value.id in parsed
and isinstance(node.slice, ast.Constant)
)
return tuple(findings)
[docs]
def test_parsed_json_is_never_subscripted():
"""No parser indexes a top-level key its tool is free to omit.
A CLI with nothing to report often answers a different shape rather than an
empty one. `fwupdmgr --json` replaces its whole payload with an `Error`
object and exits `2`, so `data["Devices"]` aborted every query reaching
`fwupd` on any host with no updatable device: a VM, a container, plenty of
real machines. Two such crashes came out of that one parser within a month
([#2073](https://github.com/kdeldycke/meta-package-manager/pull/2073) was
the first), each found by a user rather than by CI.
They surface only on the empty result, which is the case nobody has the
hardware to stage, so the guard is structural rather than another fixture:
read a parsed payload with `.get()` and a default.
"""
offenders = {}
for module_path in sorted(Path(managers_module.__file__).parent.glob("*.py")):
found = _parsed_json_subscripts(module_path.read_text(encoding="UTF-8"))
if found:
offenders[module_path.name] = found
assert not offenders, f"Parsed JSON subscripted directly: {offenders}"