# 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.
"""Tests for new package managers defined from the configuration.
From the `[mpm.overrides.<id>]` configuration section.
"""
from __future__ import annotations
import inspect
import json
import os
import re
import sys
from operator import attrgetter
from pathlib import Path
from types import SimpleNamespace
import click
import pytest
from click_extra import ValidationError
from click_extra.context import CONF_SOURCE
import meta_package_manager
from meta_package_manager.__main__ import main
from meta_package_manager.capabilities import (
Operations,
cleanup_orphan_is_synthesized,
implements,
implements_method,
)
from meta_package_manager.config import (
_candidate_config_path,
_collect_definitions,
_config_source,
_is_remote_url,
config_file_is_trusted,
discover_config_definitions,
register_config_managers,
register_eager_config_managers,
validate_manager_overrides_section,
)
from meta_package_manager.definitions import (
OVERRIDES_SECTION,
ConfigDrivenManager,
ManagerDefinition,
OperationSpec,
build_manager_class,
load_bundled_definitions,
parse_manager_definition,
)
from meta_package_manager.pool import pool
from .conftest import tomllib
skip_windows = pytest.mark.skipif(
not hasattr(os, "getuid"),
reason="POSIX-only: needs a shell fake CLI and POSIX file ownership.",
)
DEFINITION_TOML = """\
[mpm.overrides.mytool]
platforms = ["all_platforms"]
cli_names = ["mytool"]
[mpm.overrides.mytool.operations.sync]
args = ["update"]
"""
"""A configuration file defining one valid manager."""
def _clean_definitions() -> None:
"""Remove every config-defined manager from the singleton pool."""
for manager_id in list(pool.config_defined_ids):
pool.register.pop(manager_id, None)
pool.config_defined_ids.clear()
pool._evict_id_caches()
[docs]
@pytest.fixture
def reset_definitions():
"""Drop config-defined managers from the singleton pool before and after a test.
The pool is a module-level singleton, so a manager registered by one test would
otherwise leak into the next. Mirrors `reset_overrides` in
{mod}`tests.test_manager_overrides`.
"""
_clean_definitions()
yield
_clean_definitions()
# Parsing and validation.
[docs]
def test_parse_definition_returns_dataclass():
definition = parse_manager_definition(
"deno",
{
"name": "Deno",
"platforms": ["linux", "macos"],
"homepage_url": "https://deno.land",
"cli_names": ["deno"],
"requirement": ">=1.40",
"operations": {
"installed": {
"args": ["list"],
"regex": r"^(?P<package_id>\S+)@(?P<installed_version>\S+)$",
},
"install": {"args": ["install", "{package_id}"]},
},
},
)
assert isinstance(definition, ManagerDefinition)
assert definition.manager_id == "deno"
assert definition.name == "Deno"
assert definition.platforms == ("linux", "macos")
assert definition.cli_fields["cli_names"] == ("deno",)
assert set(definition.operations) == {"installed", "install"}
assert definition.operations["installed"].parse_mode == "regex"
assert definition.operations["install"].parse_mode == "none"
[docs]
def test_parse_definition_json_list_path():
"""A JSON query keeps the dotted `list_path` that leads to its package array."""
definition = parse_manager_definition(
"mytool",
{
"platforms": ["linux"],
"operations": {
"outdated": {
"args": ["outdated", "--json"],
"format": "json",
"list_path": "packages",
"fields": {
"package_id": "name",
"installed_version": "current",
"latest_version": "latest",
},
},
},
},
)
spec = definition.operations["outdated"]
assert spec.parse_mode == "json"
assert spec.list_path == "packages"
assert spec.fields is not None
assert spec.fields["latest_version"] == "latest"
[docs]
def test_parse_definition_accepts_operation_notes():
"""An `operation_notes` table lands on the class with its keys validated.
A key naming no `Operations` member is rejected by the converter, the
parse-time twin of `test_operation_notes` over the pool: a definition
declaring one would otherwise render no note at all, with no error naming
the key.
"""
definition = parse_manager_definition(
"mytool",
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["update"]}},
"operation_notes": {
"upgrade": "no upgrade command at all",
"outdated": "no command lists available upgrades",
},
},
)
assert definition.cli_fields["operation_notes"] == {
"upgrade": "no upgrade command at all",
"outdated": "no command lists available upgrades",
}
[docs]
def test_parse_definition_accepts_orphan_operations():
"""The schema recognizes the `remove_orphan` and `cleanup_orphan` refinements.
`remove_orphan` requires its ``{package_id}`` placeholder like `remove`.
"""
definition = parse_manager_definition(
"mytool",
{
"platforms": ["all_platforms"],
"cli_names": ["mytool"],
"operations": {
"remove_orphan": {"args": ["remove", "--cascade", "{package_id}"]},
"cleanup_orphan": {"args": ["autoremove"]},
},
},
)
assert set(definition.operations) == {"remove_orphan", "cleanup_orphan"}
assert definition.operations["remove_orphan"].parse_mode == "none"
assert definition.operations["cleanup_orphan"].parse_mode == "none"
[docs]
def test_parse_definition_rejects_legacy_cleanup():
"""The monolithic `cleanup` operation is gone from the schema.
Declaring it names the category keys that replaced it.
"""
with pytest.raises(ValidationError, match="cleanup_orphan, cleanup_cache"):
parse_manager_definition(
"mytool",
{
"platforms": ["all_platforms"],
"cli_names": ["mytool"],
"operations": {"cleanup": {"args": ["clean"]}},
},
)
[docs]
@pytest.mark.parametrize("operation", ("mark_explicit", "remove_orphan"))
def test_parse_definition_package_operation_requires_package_id(operation):
"""A per-package operation without ``{package_id}`` is rejected.
It targets nothing.
"""
with pytest.raises(ValidationError, match="package_id"):
parse_manager_definition(
"mytool",
{
"platforms": ["all_platforms"],
"cli_names": ["mytool"],
"operations": {operation: {"args": ["mark", "--cascade"]}},
},
)
[docs]
def test_parse_definition_versionless_catalog_manager():
"""A tool with no per-package versions, no search command and a Brewfile mapping.
Locks three schema affordances at once: an `installed` regex may capture only
the package ID (Clear Linux bundles and Cygwin listings carry no version), a
`search` may omit ``{query}`` to list the whole catalog and rely on
client-side refiltering, and the Brewfile export fields land on the built class.
"""
definition = parse_manager_definition(
"bundletool",
{
"platforms": ["linux"],
"brewfile_entry_type": "brew",
"brewfile_skip_warning": "Skipping {count} bundle(s).",
"operations": {
"installed": {
"args": ["list"],
"regex": r"^(?P<package_id>\S+)$",
},
"search": {
"args": ["list", "--all"],
"regex": r"^(?P<package_id>\S+)$",
},
},
},
)
klass = build_manager_class(definition)
assert klass.brewfile_entry_type == "brew"
assert klass.brewfile_skip_warning == "Skipping {count} bundle(s)."
[docs]
@pytest.mark.parametrize(
("section", "expected"),
(
pytest.param("update", "mytool: expected a table, got str", id="not-a-table"),
pytest.param(
{"platforms": ["narnia"], "operations": {"sync": {"args": ["s"]}}},
"unknown platform",
id="unknown-platform",
),
pytest.param(
{"platforms": [], "operations": {"sync": {"args": ["s"]}}},
"must list at least one platform or group",
id="empty-platforms",
),
pytest.param(
{"platforms": ["linux"], "operations": {"frobnicate": {"args": ["s"]}}},
"unknown operation",
id="unknown-operation",
),
pytest.param(
{"platforms": ["linux"], "operations": {"sync": "update"}},
"mytool.operations.sync: expected a table, got str",
id="operation-not-a-table",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"sync": {"args": ["s"], "regex": r"^(?P<package_id>\S+)$"},
},
},
"mytool.operations.sync.regex: unknown key. Allowed: args, cli, sudo.",
id="parser-on-command-operation",
),
pytest.param(
{"platforms": ["linux"], "operations": {"sync": {"sudo": True}}},
"mytool.operations.sync.args: required.",
id="missing-args",
),
pytest.param(
{"platforms": ["linux"], "operations": {"sync": {"args": []}}},
"mytool.operations.sync.args: must be a non-empty list.",
id="empty-args",
),
pytest.param(
{"platforms": ["linux"], "operations": {"install": {"args": ["install"]}}},
"must reference the {package_id} placeholder",
id="missing-package-id-placeholder",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"search": {
"args": ["s", "{qeury}"],
"regex": r"^(?P<package_id>\S+)$",
},
},
},
"unknown placeholder(s): {qeury}",
id="typoed-placeholder",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s", "{package_id}"]}},
},
"sync args take no placeholder",
id="placeholder-on-placeholderless-operation",
),
pytest.param(
{"platforms": ["linux"], "operations": {"installed": {"args": ["list"]}}},
"needs a 'regex' or a JSON 'fields' parser",
id="query-without-parser",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"outdated": {"args": ["o"], "regex": r"^(?P<package_id>\S+)$"},
},
},
"missing required",
id="regex-missing-latest-version",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"installed": {"args": ["list"], "regex": r"^(?P<package_id>\S+"},
},
},
"invalid regular expression",
id="unbalanced-regex",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"installed": {
"args": ["list"],
"regex": r"^(?P<package_id>\S+) (?P<version>\S+)$",
},
},
},
"unrecognized field(s): version.",
id="regex-group-names-no-field",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"outdated": {"args": ["o"], "format": "json"}},
},
"JSON parsing requires a 'fields' mapping",
id="json-without-fields",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"installed": {
"args": ["list"],
"regex": r"(?P<package_id>\S+)(?P<installed_version>\S+)",
"fields": {"package_id": "n"},
},
},
},
"not both",
id="regex-and-json",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"]}},
"bogus": 1,
},
"unknown field",
id="unknown-top-level-field",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"search": {
"args": ["search", "{query}", "{exact_args}"],
"regex": r"^(?P<package_id>\S+)$",
},
},
},
"declares no 'exact_args' list",
id="search-marker-without-refinement-list",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"search": {
"args": ["search", "{query}"],
"exact_args": ["--exact"],
"regex": r"^(?P<package_id>\S+)$",
},
},
},
"no {exact_args} marker",
id="search-refinement-list-without-marker",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"search": {
"args": ["search", "{query}", "--flag={exact_args}"],
"exact_args": ["--exact"],
"regex": r"^(?P<package_id>\S+)$",
},
},
},
"must be a standalone argument",
id="search-embedded-refinement-marker",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"search": {
"args": ["search", "{query}", "{exact_args}"],
"exact_args": ["--match={query}"],
"regex": r"^(?P<package_id>\S+)$",
},
},
},
"may not carry a placeholder",
id="search-refinement-with-placeholder",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"installed": {
"args": ["list", "--json"],
"format": "json",
"fields": {"package_id": "name", "installed_version": "v[x]"},
},
},
},
"invalid selector",
id="json-bad-field-selector",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {
"installed": {
"args": ["list", "{exact_args}"],
"regex": r"^(?P<package_id>\S+)$",
},
},
},
"unknown placeholder",
id="refinement-marker-outside-search",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"], "cli": ""}},
},
"must be a non-empty string",
id="empty-operation-cli",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"], "sudo": "yes"}},
},
"expected a boolean",
id="non-boolean-sudo",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"]}},
"version_cli": 1,
},
"expected a string",
id="non-string-version-cli",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"]}},
"default_sudo": "yes",
},
"expected a boolean",
id="non-boolean-default-sudo",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"]}},
"internal_sudo": "yes",
},
"expected a boolean",
id="non-boolean-internal-sudo",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"]}},
"operation_notes": {"upgrade": "no upgrade command", "upgrade-all": ""},
},
"unknown operation 'upgrade-all' in operation_notes",
id="unknown-operation-note-key",
),
pytest.param(
{
"platforms": ["linux"],
"operations": {"sync": {"args": ["s"]}},
"internal_sudo": True,
"internal_sudo_operations": ["sync", "purge"],
},
"unknown operation 'purge'",
id="unknown-internal-sudo-operation",
),
pytest.param(
{"platforms": ["linux"]},
"must declare 'operations'",
id="missing-operations",
),
pytest.param(
{"operations": {"sync": {"args": ["s"]}}},
"must declare 'platforms'",
id="missing-platforms",
),
pytest.param(
{"platforms": ["linux"], "operations": {}},
"non-empty table",
id="empty-operations",
),
),
)
def test_parse_definition_rejects(section, expected):
with pytest.raises(ValidationError) as raised:
parse_manager_definition("mytool", section)
assert expected in str(raised.value)
[docs]
def test_parse_definition_rejects_bad_id():
with pytest.raises(ValidationError, match="invalid manager ID"):
parse_manager_definition(
"MyTool", {"platforms": ["linux"], "operations": {"sync": {"args": ["s"]}}}
)
[docs]
def test_parse_definition_multi_binary_and_sudo():
"""A definition can span binaries, mark privileged operations, and more.
It can span sibling binaries, mark privileged operations, and declare an alternate
version probe.
"""
definition = parse_manager_definition(
"urpmi-like",
{
"platforms": ["linux"],
"cli_names": ["urpmi"],
"default_sudo": True,
"version_cli": "uname",
"operations": {
"search": {
"args": ["--fuzzy", "{query}"],
"cli": "urpmq",
"regex": r"^(?P<package_id>\S+)$",
},
"install": {"args": ["--auto", "{package_id}"], "sudo": True},
"remove": {
"args": ["--auto", "{package_id}"],
"cli": "urpme",
"sudo": True,
},
},
},
)
assert definition.cli_fields["default_sudo"] is True
assert definition.cli_fields["version_cli"] == "uname"
assert definition.operations["search"].cli == "urpmq"
assert definition.operations["search"].sudo is False
assert definition.operations["install"].cli is None
assert definition.operations["install"].sudo is True
assert definition.operations["remove"].cli == "urpme"
assert definition.operations["remove"].sudo is True
[docs]
def test_parse_definition_internal_sudo():
"""A definition can mark its tool as escalating internally, like fink.
The flag lands on the built class without touching the wrapping policy.
"""
definition = parse_manager_definition(
"finklike",
{
"platforms": ["macos"],
"internal_sudo": True,
"operations": {
"install": {"args": ["install", "{package_id}"]},
},
},
)
assert definition.cli_fields["internal_sudo"] is True
klass = build_manager_class(definition)
assert klass.internal_sudo is True
assert klass.default_sudo is False
assert klass.resets_sudo_timestamp is False
# Every mutating operation may run the tool's own `sudo` by default.
assert {"sync", "cleanup", "install"} <= klass.internal_sudo_operations
[docs]
def test_parse_definition_resets_sudo_timestamp():
"""A definition can mark its tool as dropping cached `sudo` credentials.
Like Homebrew, beside the internal escalation it qualifies.
"""
definition = parse_manager_definition(
"brewlike",
{
"platforms": ["macos"],
"internal_sudo": True,
"internal_sudo_operations": ["install", "remove"],
"resets_sudo_timestamp": True,
"operations": {
"install": {"args": ["install", "{package_id}"]},
},
},
)
assert definition.cli_fields["resets_sudo_timestamp"] is True
klass = build_manager_class(definition)
assert klass.resets_sudo_timestamp is True
# The definition narrows the operations its tool escalates in.
assert klass.internal_sudo_operations == frozenset({"install", "remove"})
# Section routing: built-in override vs new-manager definition.
[docs]
def test_routing_builtin_is_override():
# A bad override field on a built-in is reported as an unknown override field,
# not treated as a definition.
with pytest.raises(ValidationError, match="unknown field"):
validate_manager_overrides_section({"pip": {"nope": 1}}, pool=pool)
[docs]
def test_routing_typo_of_builtin_is_unknown_manager():
with pytest.raises(ValidationError, match="unknown manager ID"):
validate_manager_overrides_section({"brw": {"cli_names": ["brew"]}}, pool=pool)
[docs]
def test_routing_definition_is_validated():
# A non-built-in ID carrying operations is validated as a definition.
validate_manager_overrides_section(
{
"mytool": {
"platforms": ["linux"],
"operations": {"sync": {"args": ["update"]}},
},
},
pool=pool,
)
# Factory.
def _definition(
manager_id: str = "mytool",
name: str = "My Tool",
cli_fields: dict[str, object] | None = None,
**operations,
):
"""Build a definition from its operations, with stand-in identity fields."""
if cli_fields is None:
cli_fields = {"cli_names": (manager_id,)}
return ManagerDefinition(
manager_id=manager_id,
name=name,
platforms=("all_platforms",),
homepage_url=None,
documentation_url=None,
repository_url=None,
wikipedia_url=None,
logo=None,
cli_fields=cli_fields,
operations=operations,
)
[docs]
def test_build_class_identity():
klass = build_manager_class(_definition(sync=OperationSpec(args=("update",))))
manager = klass()
assert manager.id == "mytool"
assert manager.name == "My Tool"
assert isinstance(manager, ConfigDrivenManager)
assert manager.virtual is False
[docs]
@pytest.mark.parametrize(
("operations", "implemented", "not_implemented"),
(
pytest.param(
{
"installed": OperationSpec(
args=("l",),
parse_mode="regex",
regex=r"(?P<package_id>\S+)\s+(?P<installed_version>\S+)",
)
},
(Operations.installed,),
(Operations.outdated, Operations.install, Operations.remove),
id="installed-only",
),
pytest.param(
{
"installed": OperationSpec(
args=("l",),
parse_mode="regex",
regex=r"(?P<package_id>\S+)\s+(?P<installed_version>\S+)",
),
"upgrade_one": OperationSpec(args=("up", "{package_id}")),
},
(Operations.installed, Operations.upgrade),
(Operations.upgrade_all, Operations.outdated),
id="upgrade-needs-installed",
),
pytest.param(
{"upgrade_all": OperationSpec(args=("up",))},
(Operations.upgrade_all,),
(Operations.upgrade, Operations.installed),
id="upgrade-all-only",
),
),
)
def test_build_class_implements(operations, implemented, not_implemented):
manager = build_manager_class(_definition(**operations))()
for operation in implemented:
assert implements(manager, operation) is True
for operation in not_implemented:
assert implements(manager, operation) is False
[docs]
def test_factory_regex_parsing(stub_run_cli):
manager = build_manager_class(
_definition(
installed=OperationSpec(
args=("list",),
parse_mode="regex",
regex=r"^(?P<package_id>\S+)@(?P<installed_version>\S+)$",
),
),
)()
stub_run_cli(manager, "[email protected]\n[email protected]")
assert [(p.id, str(p.installed_version)) for p in manager.installed] == [
("ruff", "0.1.2"),
("black", "24.1.0"),
]
[docs]
def test_factory_json_parsing(stub_run_cli):
manager = build_manager_class(
_definition(
outdated=OperationSpec(
args=("outdated", "--json"),
parse_mode="json",
list_path="packages",
fields={
"package_id": "name",
"installed_version": "current",
"latest_version": "latest",
},
),
),
)()
stub_run_cli(
manager,
json.dumps(
{"packages": [{"name": "ruff", "current": "0.1.2", "latest": "0.2.0"}]},
),
)
outdated = list(manager.outdated)
assert len(outdated) == 1
assert outdated[0].id == "ruff"
assert str(outdated[0].installed_version) == "0.1.2"
assert str(outdated[0].latest_version) == "0.2.0"
[docs]
def test_factory_command_substitution(monkeypatch):
manager = build_manager_class(
_definition(
install=OperationSpec(args=("install", "{package_id}")),
upgrade_one=OperationSpec(args=("install", "--force", "{package_id}")),
),
)()
captured: dict[str, tuple[str, ...]] = {}
monkeypatch.setattr(
manager,
"run_cli",
lambda *args, **kwargs: captured.update(install=args) or "",
)
manager.install("jq")
assert captured["install"] == ("install", "jq")
# upgrade_one_cli builds a command line off the resolved binary path.
monkeypatch.setattr(manager, "build_cli", lambda *args, **kwargs: args)
assert manager.upgrade_one_cli("jq") == ("install", "--force", "jq")
[docs]
def test_factory_operation_cli(monkeypatch):
"""An operation with its own `cli` routes the call through that binary.
It resolves the sibling binary. A missing sibling is an error, not a silent
fallback.
"""
manager = build_manager_class(
_definition(
remove=OperationSpec(args=("--auto", "{package_id}"), cli="urpme"),
),
)()
sibling = Path("/fake/bin/urpme")
captured: dict[str, object] = {}
monkeypatch.setattr(manager, "which", lambda name: sibling)
monkeypatch.setattr(
manager,
"run_cli",
lambda *args, **kwargs: captured.update(args=args, **kwargs) or "",
)
manager.remove("jq")
assert captured["args"] == ("--auto", "jq")
assert captured["override_cli_path"] == sibling
monkeypatch.setattr(manager, "which", lambda name: None)
with pytest.raises(FileNotFoundError, match="urpme not found"):
manager.remove("jq")
[docs]
def test_factory_orphans_query(stub_run_cli):
"""A config-defined manager can declare the `orphans` read-only query.
It is parsed like `installed` and advertised as a routable operation.
"""
manager = build_manager_class(
_definition(
orphans=OperationSpec(
args=("list", "--orphaned"),
parse_mode="regex",
regex=r"^(?P<package_id>\S+) (?P<installed_version>\S+)$",
),
remove=OperationSpec(args=("remove", "{package_id}")),
),
)()
stub_run_cli(manager, "libfoo 1.2\nlibbar 3.4")
assert implements(manager, Operations.orphans) is True
packages = {(p.id, str(p.installed_version)) for p in manager.orphans}
assert packages == {("libfoo", "1.2"), ("libbar", "3.4")}
# With orphans + remove declared and no native sweep, the synthesized
# system-wide sweep of `cleanup --orphans` kicks in for free.
assert cleanup_orphan_is_synthesized(manager) is True
[docs]
def test_factory_orphan_operations(monkeypatch):
"""A config-defined manager can declare the orphan refinements.
`remove_orphan` substitutes ``{package_id}`` like `remove`, `cleanup_orphan` runs a
bare command like `cleanup`. Both register as implemented capabilities.
"""
manager = build_manager_class(
_definition(
remove_orphan=OperationSpec(args=("remove", "--recursive", "{package_id}")),
cleanup_orphan=OperationSpec(args=("autoremove", "--yes")),
cleanup_cache=OperationSpec(args=("clean", "--cache")),
cleanup_repair=OperationSpec(args=("repair",)),
),
)()
captured: list[tuple[str, ...]] = []
def record_run_cli(*args, **kwargs):
captured.append(args)
return ""
monkeypatch.setattr(manager, "run_cli", record_run_cli)
manager.remove_orphan("jq")
assert captured.pop() == ("remove", "--recursive", "jq")
manager.cleanup_orphan()
assert captured.pop() == ("autoremove", "--yes")
manager.cleanup_cache()
assert captured.pop() == ("clean", "--cache")
manager.cleanup_repair()
assert captured.pop() == ("repair",)
assert implements_method(manager, "remove_orphan") is True
assert implements_method(manager, "cleanup_orphan") is True
assert implements_method(manager, "cleanup_cache") is True
assert implements_method(manager, "cleanup_repair") is True
[docs]
def test_factory_mark_explicit(monkeypatch):
"""A config-defined manager can declare `mark_explicit`.
It substitutes ``{package_id}`` like `remove` and registers as implemented.
"""
manager = build_manager_class(
_definition(mark_explicit=OperationSpec(args=("mark", "{package_id}"))),
)()
captured: list[tuple[str, ...]] = []
def record_run_cli(*args, **kwargs):
captured.append(args)
return ""
monkeypatch.setattr(manager, "run_cli", record_run_cli)
manager.mark_explicit("jq")
assert captured == [("mark", "jq")]
assert implements_method(manager, "mark_explicit") is True
[docs]
def test_factory_operation_sudo(monkeypatch):
"""A `sudo = true` operation is built privileged; unmarked ones are not."""
manager = build_manager_class(
_definition(
install=OperationSpec(args=("--auto", "{package_id}"), sudo=True),
sync=OperationSpec(args=("update",)),
upgrade_all=OperationSpec(args=("--auto-select",), sudo=True),
),
)()
captured: dict[str, object] = {}
monkeypatch.setattr(
manager,
"run_cli",
lambda *args, **kwargs: captured.update(kwargs) or "",
)
# Each phase clears the dict so its assert reads the state its own call
# captured, and reads through a local: mypy narrows the `captured["sudo"]`
# subscript on each assert and cannot see the monkeypatched methods mutate
# the dict, so re-asserting the same subscript with alternating values
# would mark the later phases unreachable.
manager.install("jq")
install_sudo = captured["sudo"]
assert install_sudo is True
captured.clear()
manager.sync()
sync_sudo = captured["sudo"]
assert sync_sudo is False
monkeypatch.setattr(
manager,
"build_cli",
lambda *args, **kwargs: captured.update(kwargs) or args,
)
captured.clear()
manager.upgrade_all_cli()
upgrade_all_sudo = captured["sudo"]
assert upgrade_all_sudo is True
[docs]
def test_factory_query_sudo(monkeypatch):
"""A query marked `sudo = true` runs privileged.
For tools that gate even their read-only listings behind root (deb-get's upgradable
check).
"""
manager = build_manager_class(
_definition(
outdated=OperationSpec(
args=("upgradable",),
sudo=True,
parse_mode="regex",
regex=r"^(?P<package_id>\S+) (?P<latest_version>\S+)$",
),
),
)()
captured: dict[str, object] = {}
monkeypatch.setattr(
manager,
"run_cli",
lambda *args, **kwargs: captured.update(kwargs) or "",
)
list(manager.outdated)
assert captured["sudo"] is True
[docs]
@pytest.mark.parametrize(
("extended", "exact", "expected_args"),
(
pytest.param(
False,
False,
("search", "--by-id-only", "--line", "vim"),
id="id-name-only-fuzzy",
),
pytest.param(
True,
False,
("search", "--description", "--line", "vim"),
id="extended-fuzzy",
),
pytest.param(
False,
True,
("search", "--by-id-only", "--exact", "--line", "vim"),
id="id-name-only-exact",
),
pytest.param(
True,
True,
("search", "--description", "--exact", "--line", "vim"),
id="extended-exact",
),
),
)
def test_factory_search_refinements(monkeypatch, extended, exact, expected_args):
"""Refinement markers expand in place to their argument lists when active.
They expand to nothing otherwise. Declaring them advertises native support.
"""
manager = build_manager_class(
_definition(
search=OperationSpec(
args=(
"search",
"{id_name_only_args}",
"{extended_args}",
"{exact_args}",
"--line",
"{query}",
),
exact_args=("--exact",),
extended_args=("--description",),
id_name_only_args=("--by-id-only",),
parse_mode="regex",
regex=r"^(?P<package_id>\S+)$",
),
),
)()
captured: dict[str, tuple[str, ...]] = {}
monkeypatch.setattr(
manager,
"run_cli",
lambda *args, **kwargs: captured.update(args=args) or "",
)
list(manager.search("vim", extended=extended, exact=exact))
assert captured["args"] == expected_args
search_func = type(manager).search
assert search_func.exact_support is True # type: ignore[attr-defined]
assert search_func.extended_support is True # type: ignore[attr-defined]
[docs]
def test_factory_search_without_refinements_relies_on_refiltering():
"""A plain search template advertises no native refinement support."""
manager = build_manager_class(
_definition(
search=OperationSpec(
args=("search", "{query}"),
parse_mode="regex",
regex=r"^(?P<package_id>\S+)$",
),
),
)()
search_func = type(manager).search
assert search_func.exact_support is False # type: ignore[attr-defined]
assert search_func.extended_support is False # type: ignore[attr-defined]
[docs]
def test_factory_json_array_field(stub_run_cli):
"""A `key[N]` selector picks one element out of a list-valued JSON field.
It resolves to nothing on non-list values or out-of-range indexes.
"""
manager = build_manager_class(
_definition(
installed=OperationSpec(
args=("list", "--json"),
parse_mode="json",
fields={
"package_id": "name",
"installed_version": "installed_versions[0]",
},
),
),
)()
stub_run_cli(
manager,
json.dumps([
{"name": "jq", "installed_versions": ["1.7.1", "1.6"]},
{"name": "empty", "installed_versions": []},
{"name": "scalar", "installed_versions": "oops"},
]),
)
packages = {p.id: p.installed_version for p in manager.installed}
assert str(packages["jq"]) == "1.7.1"
assert packages["empty"] is None
assert packages["scalar"] is None
[docs]
def test_factory_install_ignores_a_version_pin(capture_run_cli, caplog):
"""A config-defined install drops a requested version and says so.
Rather than failing, or passing a version its args have no place for.
"""
manager = build_manager_class(
_definition(install=OperationSpec(args=("install", "{package_id}"))),
)()
calls = capture_run_cli(manager)
manager.install("jq", version="1.7.1")
assert calls == [("install", "jq")]
assert "do not support version pinning" in caplog.text
[docs]
@pytest.mark.parametrize(
("list_path", "output", "expected", "warns"),
(
pytest.param(
None,
"Error: unknown command 'list'",
set(),
True,
id="not-json",
),
pytest.param(
"data.packages",
json.dumps({"data": [{"name": "jq", "version": "1.7.1"}]}),
set(),
False,
id="path-through-an-array",
),
pytest.param(
None,
json.dumps([
{"name": "jq", "version": "1.7.1"},
"jq",
{"version": "1.6"},
{"name": "", "version": "1.6"},
]),
{("jq", "1.7.1")},
False,
id="stray-items",
),
),
)
def test_factory_json_parsing_skips_unexpected_shapes(
stub_run_cli, caplog, list_path, output, expected, warns
):
"""A payload of an unexpected shape yields what it can, and never raises.
Output that is not JSON yields nothing and logs one warning. A `list_path`
that walks into an array instead of a table finds no package array. Items
that are not tables, or that carry no package ID, are skipped.
"""
manager = build_manager_class(
_definition(
installed=OperationSpec(
args=("list", "--json"),
parse_mode="json",
list_path=list_path,
fields={"package_id": "name", "installed_version": "version"},
),
),
)()
stub_run_cli(manager, output)
packages = {(p.id, str(p.installed_version)) for p in manager.installed}
assert packages == expected
assert ("Could not parse JSON output" in caplog.text) is warns
[docs]
def test_factory_regex_parsing_skips_a_match_without_id(stub_run_cli):
"""A line the pattern matches with an empty `package_id` group is skipped."""
manager = build_manager_class(
_definition(
installed=OperationSpec(
args=("list",),
parse_mode="regex",
regex=r"^(?P<package_id>\S*)@(?P<installed_version>\S+)$",
),
),
)()
stub_run_cli(manager, "[email protected]\n@24.1.0")
assert [(p.id, str(p.installed_version)) for p in manager.installed] == [
("ruff", "0.1.2"),
]
[docs]
@pytest.mark.parametrize(
("token", "expected"),
(
pytest.param("curl-8.7.1", ("curl", "8.7.1"), id="name-and-version"),
pytest.param("python311-3.11.9", ("python311", "3.11.9"), id="digits-in-name"),
pytest.param("curl", None, id="no-version"),
),
)
def test_split_name_version(token, expected):
"""The default splitter cuts at the last dash a digit follows.
It answers `None` for a token that carries no version.
"""
manager = build_manager_class(
_definition(sync=OperationSpec(args=("update",))),
)()
assert manager.split_name_version(token) == expected
# Trust gate.
[docs]
@skip_windows
@pytest.mark.parametrize(
("file_mode", "folder_mode", "trusted"),
(
pytest.param(0o600, 0o700, True, id="private"),
pytest.param(0o644, 0o755, True, id="readable-by-others"),
pytest.param(0o666, 0o755, False, id="world-writable-file"),
pytest.param(0o664, 0o755, False, id="group-writable-file"),
pytest.param(0o644, 0o777, False, id="world-writable-folder"),
pytest.param(0o644, 0o775, False, id="group-writable-folder"),
pytest.param(0o644, 0o1777, True, id="sticky-world-writable-folder"),
),
)
def test_config_file_is_trusted(tmp_path, request, file_mode, folder_mode, trusted):
"""A file others can write to, or swap through its folder, is not trusted.
The sticky bit stops others from replacing a file in a folder they can write
to, like `/tmp`, so such a folder is trusted.
"""
folder = tmp_path / "conf"
folder.mkdir()
config_file = folder / "config.toml"
config_file.write_text("", encoding="UTF-8")
config_file.chmod(file_mode)
folder.chmod(folder_mode)
request.addfinalizer(lambda: folder.chmod(0o700))
assert folder.stat().st_mode & 0o7777 == folder_mode
assert config_file_is_trusted(config_file) is trusted
[docs]
@skip_windows
def test_config_file_is_trusted_rejects_a_missing_file(tmp_path):
"""A file that cannot be read is never trusted."""
assert config_file_is_trusted(tmp_path / "absent.toml") is False
[docs]
@skip_windows
def test_config_file_is_trusted_rejects_another_owner(tmp_path, monkeypatch):
"""A private file owned by another account, and not root, is not trusted.
Another account than the current one.
"""
real_uid = os.getuid()
if real_uid == 0:
pytest.skip("A file root owns is always trusted.")
config_file = tmp_path / "config.toml"
config_file.write_text("", encoding="UTF-8")
config_file.chmod(0o600)
assert config_file_is_trusted(config_file) is True
monkeypatch.setattr(os, "getuid", lambda: real_uid + 1)
assert config_file_is_trusted(config_file) is False
[docs]
def test_config_file_is_trusted_without_posix_ownership(tmp_path, monkeypatch):
"""Without `os.getuid`, as on Windows, the ownership check does not apply."""
monkeypatch.delattr(os, "getuid", raising=False)
loose = tmp_path / "loose.toml"
loose.write_text("", encoding="UTF-8")
loose.chmod(0o666)
assert config_file_is_trusted(loose) is True
[docs]
@pytest.mark.parametrize(
("location", "remote"),
(
pytest.param("https://example.com/mpm/config.toml", True, id="https"),
pytest.param("http://example.com/mpm/config.toml", True, id="http"),
pytest.param("/etc/mpm/config.toml", False, id="absolute-path"),
pytest.param("config.toml", False, id="relative-path"),
),
)
def test_is_remote_url(location, remote):
assert _is_remote_url(location) is remote
[docs]
@pytest.mark.parametrize(
("location", "expected"),
(
pytest.param(None, (None, False), id="no-config"),
pytest.param("https://example.com/mpm/config.toml", (None, True), id="url"),
pytest.param(
"/etc/mpm/config.toml",
(Path("/etc/mpm/config.toml"), False),
id="local-file",
),
),
)
def test_config_source(location, expected):
"""The source of the trust gate is the location click-extra loaded."""
ctx = click.Context(click.Command("mpm"))
ctx.meta[CONF_SOURCE] = location
assert _config_source(ctx) == expected
[docs]
def test_collect_definitions_skips_overrides_and_invalid_sections():
"""Only the valid definitions of new managers come out.
A section of a shipped manager is an override, and an invalid one already failed the
load-time validator.
"""
sections = {
"pip": {"timeout": 60},
"mytool": {"platforms": ["linux"], "operations": {"sync": {"args": ["s"]}}},
"urpmi-like": {
"platforms": ["narnia"],
"operations": {"sync": {"args": ["s"]}},
},
}
assert list(_collect_definitions(pool, sections)) == ["mytool"]
# Eager pre-load of the definitions, ahead of the CLI construction.
[docs]
def test_candidate_config_path_reads_the_environment(
eager_inputs, monkeypatch, tmp_path
):
"""`MPM_CONFIG` names the file, which must exist."""
config_file = tmp_path / "config.toml"
config_file.write_text("", encoding="UTF-8")
monkeypatch.setenv("MPM_CONFIG", str(config_file))
assert _candidate_config_path() == (config_file, False)
monkeypatch.setenv("MPM_CONFIG", str(tmp_path / "absent.toml"))
assert _candidate_config_path() == (None, False)
[docs]
@pytest.mark.parametrize(
"args",
(
pytest.param(("--config", "{path}"), id="separate-value"),
pytest.param(("--config={path}",), id="joined-value"),
),
)
def test_candidate_config_path_reads_the_option(
eager_inputs, monkeypatch, tmp_path, args
):
"""`--config` wins over `MPM_CONFIG`, in both of its spellings."""
option_file = tmp_path / "option.toml"
option_file.write_text("", encoding="UTF-8")
env_file = tmp_path / "env.toml"
env_file.write_text("", encoding="UTF-8")
monkeypatch.setenv("MPM_CONFIG", str(env_file))
monkeypatch.setattr(
sys,
"argv",
["mpm", *(arg.format(path=option_file) for arg in args), "managers"],
)
assert _candidate_config_path() == (option_file, False)
[docs]
def test_candidate_config_path_defers_a_url(eager_inputs, monkeypatch):
"""A remote location is reported, not read: the CLI pass handles it."""
monkeypatch.setattr(
sys,
"argv",
["mpm", "--config", "https://example.com/mpm/config.toml", "managers"],
)
assert _candidate_config_path() == (None, True)
[docs]
@pytest.mark.parametrize(
("app_files", "expected"),
(
pytest.param(None, None, id="no-folder"),
pytest.param((), None, id="empty-folder"),
pytest.param(("mpm.json", "config.toml"), "config.toml", id="toml-first"),
pytest.param(("mpm.ini", "mpm.json"), "mpm.json", id="json-before-ini"),
),
)
def test_candidate_config_path_searches_the_default_folder(
eager_inputs, monkeypatch, app_files, expected
):
"""With no explicit location, the first file of the default folder wins, TOML first.
A trailing `--config` that carries no value is ignored.
"""
monkeypatch.setattr(sys, "argv", ["mpm", "managers", "--config"])
if app_files is not None:
eager_inputs.mkdir()
for name in app_files:
(eager_inputs / name).write_text("", encoding="UTF-8")
expected_path = eager_inputs / expected if expected else None
assert _candidate_config_path() == (expected_path, False)
[docs]
@pytest.mark.parametrize(
("file_name", "content", "found"),
(
pytest.param("config.toml", DEFINITION_TOML, True, id="mpm-table"),
pytest.param(
"pyproject.toml",
DEFINITION_TOML.replace("[mpm.", "[tool.mpm."),
True,
id="pyproject-table",
),
pytest.param(
"config.toml", '[mpm]\nverbosity = "INFO"\n', False, id="no-overrides"
),
pytest.param(
"config.toml", '[mpm]\noverrides = "none"\n', False, id="not-a-table"
),
pytest.param("config.toml", "[mpm.overrides.mytool\n", False, id="malformed"),
),
)
def test_discover_config_definitions(
eager_inputs, monkeypatch, tmp_path, file_name, content, found
):
"""The definitions of a readable file come out with their source.
Any failure yields nothing rather than breaking the start of the CLI.
"""
config_file = tmp_path / file_name
config_file.write_text(content, encoding="UTF-8")
monkeypatch.setenv("MPM_CONFIG", str(config_file))
definitions, source = discover_config_definitions(pool)
if found:
assert list(definitions) == ["mytool"]
assert source == config_file
else:
assert (definitions, source) == ({}, None)
[docs]
@pytest.mark.parametrize(
"argv",
(
pytest.param(["mpm", "managers"], id="no-config"),
pytest.param(
["mpm", "--config", "https://example.com/mpm/config.toml", "managers"],
id="url",
),
),
)
def test_discover_config_definitions_reads_nothing(eager_inputs, monkeypatch, argv):
"""No local file to read leaves the definitions to the CLI pass."""
monkeypatch.setattr(sys, "argv", argv)
assert discover_config_definitions(pool) == ({}, None)
[docs]
@pytest.mark.parametrize(
("file_mode", "registered"),
(
pytest.param(0o600, True, id="private"),
pytest.param(0o666, False, marks=skip_windows, id="world-writable"),
),
)
def test_register_eager_config_managers(
eager_inputs,
monkeypatch,
tmp_path,
reset_definitions,
caplog,
file_mode,
registered,
):
"""The eager pass registers the definitions of a trusted file.
It refuses those of a file others can write to.
"""
folder = tmp_path / "conf"
folder.mkdir()
folder.chmod(0o700)
config_file = folder / "config.toml"
config_file.write_text(DEFINITION_TOML, encoding="UTF-8")
config_file.chmod(file_mode)
monkeypatch.setenv("MPM_CONFIG", str(config_file))
register_eager_config_managers(pool)
assert ("mytool" in pool.register) is registered
assert ("Refusing to define manager 'mytool'" in caplog.text) is not registered
[docs]
def test_main_registers_definitions_before_the_cli(
eager_inputs, monkeypatch, tmp_path, reset_definitions, capsys
):
"""The entry point registers the definitions before the CLI runs.
`--version` exits ahead of the CLI pass, so only the eager pass can register here.
"""
config_file = tmp_path / "config.toml"
config_file.write_text(DEFINITION_TOML, encoding="UTF-8")
config_file.chmod(0o600)
monkeypatch.setenv("MPM_CONFIG", str(config_file))
monkeypatch.setattr(sys, "argv", ["mpm", "--version"])
with pytest.raises(SystemExit) as exited:
main()
assert exited.value.code == 0
assert capsys.readouterr().out.startswith("mpm, version ")
assert "mytool" in pool.register
[docs]
@skip_windows
def test_register_refuses_untrusted_file(tmp_path, reset_definitions, caplog):
loose = tmp_path / "loose.toml"
loose.write_text("")
loose.chmod(0o666)
definition = _definition(sync=OperationSpec(args=("update",)))
registered = register_config_managers(pool, {"mytool": definition}, source=loose)
assert registered == []
assert "mytool" not in pool.register
assert "unsafe config file" in caplog.text
[docs]
def test_register_refuses_url(reset_definitions, caplog):
definition = _definition(sync=OperationSpec(args=("update",)))
registered = register_config_managers(
pool, {"mytool": definition}, source=None, source_is_url=True
)
assert registered == []
assert "mytool" not in pool.register
assert "remote config URL" in caplog.text
[docs]
def test_register_rejects_builtin_collision(reset_definitions, caplog):
definition = _definition(
manager_id="pip",
name="Fake Pip",
cli_fields={},
sync=OperationSpec(args=("update",)),
)
registered = register_config_managers(pool, {"pip": definition}, source=None)
assert registered == []
assert "already uses this ID" in caplog.text
[docs]
def test_register_is_idempotent(reset_definitions):
definition = _definition(sync=OperationSpec(args=("update",)))
assert register_config_managers(pool, {"mytool": definition}, source=None) == [
"mytool",
]
assert "mytool" in pool.register
assert "mytool" in pool.config_defined_ids
# A second pass is a no-op (already registered).
assert register_config_managers(pool, {"mytool": definition}, source=None) == []
# End-to-end through the real subprocess and the CLI.
[docs]
@skip_windows
def test_factory_functional(tmp_path, fake_tool, reset_definitions):
definition = _definition(
cli_fields={
"cli_names": ("mytool",),
"cli_search_path": (str(tmp_path),),
"requirement": ">=1.0",
"version_regexes": (r"mytool (?P<version>\S+)",),
},
installed=OperationSpec(
args=("list",),
parse_mode="regex",
regex=r"^(?P<package_id>\S+)@(?P<installed_version>\S+)$",
),
search=OperationSpec(
args=("search", "{query}"),
parse_mode="regex",
regex=r"^(?P<package_id>\S+)@(?P<latest_version>\S+)$",
),
upgrade_one=OperationSpec(args=("install", "--force", "{package_id}")),
)
manager = build_manager_class(definition)()
assert manager.available is True
assert {p.id for p in manager.installed} == {"foo", "bar"}
assert [p.id for p in manager.search("foo", False, False)] == ["foo"]
assert "--force" in manager.upgrade_one_cli("foo")
[docs]
@skip_windows
def test_factory_functional_version_cli(tmp_path, fake_tool, reset_definitions):
"""The version probe runs the `version_cli` binary instead of the main CLI.
Like OS-versioned tool suites exposing no version flag of their own.
"""
probe = tmp_path / "myuname"
probe.write_text("#!/bin/sh\necho '7.7'\n")
probe.chmod(0o755)
definition = _definition(
cli_fields={
"cli_names": ("mytool",),
"cli_search_path": (str(tmp_path),),
"requirement": ">=7.0",
"version_cli": "myuname",
"version_cli_options": ("-r",),
"version_regexes": (r"(?P<version>[\d.]+)",),
},
sync=OperationSpec(args=("update",)),
)
manager = build_manager_class(definition)()
assert str(manager.version) == "7.7"
assert manager.available is True
[docs]
@skip_windows
def test_cli_lists_config_defined_manager(
invoke, create_config, tmp_path, fake_tool, reset_definitions
):
conf_path = create_config(
"conf.toml",
f"""
[mpm.overrides.mytool]
name = "My Tool"
platforms = ["all_platforms"]
cli_names = ["mytool"]
cli_search_path = ["{tmp_path}"]
requirement = ">=1.0"
version_regexes = ['mytool (?P<version>\\S+)']
[mpm.overrides.mytool.operations.installed]
args = ["list"]
regex = '^(?P<package_id>\\S+)@(?P<installed_version>\\S+)$'
""",
)
result = invoke("--config", str(conf_path), "managers")
assert result.exit_code == 0
assert "mytool" in result.stdout
[docs]
@skip_windows
def test_cli_refuses_definition_from_untrusted_file(
invoke, create_config, reset_definitions
):
"""The trust gate checks the file the CLI loads.
Not only the file the eager pass of `main()` finds.
"""
conf_path = create_config("conf.toml", DEFINITION_TOML)
conf_path.chmod(0o666)
result = invoke("--config", str(conf_path), "managers")
assert result.exit_code == 0
assert "unsafe config file ownership or permissions" in result.stderr
assert "mytool" not in pool.register
[docs]
def test_cli_rejects_bad_definition(invoke, create_config, reset_definitions):
conf_path = create_config(
"conf.toml",
"""
[mpm.overrides.mytool]
platforms = ["narnia"]
[mpm.overrides.mytool.operations.sync]
args = ["update"]
""",
)
result = invoke("--config", str(conf_path), "managers")
assert result.exit_code != 0
assert "narnia" in result.output
# Bundled manager definitions shipped with mpm as package data.
BUNDLED_DEFINITION_FILES = sorted(
(Path(inspect.getfile(meta_package_manager)).parent / "managers").glob("*.toml"),
)
"""Every definition file mpm ships, discovered exactly like the runtime loader."""
BUNDLED_FILE_DATA = {
toml_path: tomllib.loads(toml_path.read_text(encoding="UTF-8"))
for toml_path in BUNDLED_DEFINITION_FILES
}
"""Parsed content of each shipped file: the definition and its `[samples]`.
The samples are the source-derived output fixtures feeding
`test_bundled_version_regex` and `test_bundled_parsing`, so adding a bundled
manager (with its samples) extends those tests without touching this module.
"""
def _fresh_bundled(manager_id):
"""Build a throwaway config-defined manager instance.
So parsing tests can monkeypatch `run_cli` without mutating the shared pool
singleton.
"""
for definition, _ in load_bundled_definitions():
if definition.manager_id == manager_id:
return build_manager_class(definition)()
raise AssertionError(f"{manager_id} is not among the bundled definitions")
[docs]
def test_bundled_definitions_are_valid():
"""Every shipped definition parses and reports a repo-relative TOML source."""
bundled = load_bundled_definitions()
assert bundled, "mpm ships at least one bundled manager definition"
for definition, source in bundled:
assert isinstance(definition, ManagerDefinition)
assert source.startswith("meta_package_manager/managers/")
assert source.endswith(".toml")
[docs]
def test_bundled_loader_skips_broken_files(monkeypatch, tmp_path, caplog):
"""A broken bundled file is logged and skipped, and the valid files load.
A file that does not parse, or that defines an invalid manager: one bad resource
cannot break the start of `mpm`.
"""
(tmp_path / "broken.toml").write_text("[mpm.overrides.mytool\n", encoding="UTF-8")
(tmp_path / "invalid.toml").write_text(
DEFINITION_TOML.replace("mytool", "urpmi-like").replace(
'"all_platforms"', '"narnia"'
),
encoding="UTF-8",
)
(tmp_path / "valid.toml").write_text(DEFINITION_TOML, encoding="UTF-8")
(tmp_path / "readme.md").write_text("", encoding="UTF-8")
monkeypatch.setattr(
"meta_package_manager.definitions.importlib",
SimpleNamespace(resources=SimpleNamespace(files=lambda package: tmp_path)),
)
load_bundled_definitions.cache_clear()
try:
bundled = load_bundled_definitions()
finally:
# Drop the stand-in result, so the next caller reads the shipped files.
load_bundled_definitions.cache_clear()
assert [(d.manager_id, source) for d, source in bundled] == [
("mytool", "meta_package_manager/managers/valid.toml"),
]
assert (
"Skipping unreadable bundled definition meta_package_manager/managers/"
"broken.toml" in caplog.text
)
assert (
"Skipping invalid bundled manager 'urpmi-like' in meta_package_manager/"
"managers/invalid.toml" in caplog.text
)
[docs]
def test_bundled_ids_disjoint_from_builtins():
assert pool.bundled_manager_ids
assert pool.bundled_manager_ids.isdisjoint(pool.builtin_manager_ids)
assert pool.known_manager_ids == pool.builtin_manager_ids | pool.bundled_manager_ids
[docs]
def test_bundled_inventory():
"""The pool registers exactly one bundled manager per shipped TOML file.
Catches a loader-side skip: a malformed file is logged and dropped at load
time, so the pool silently missing an ID would be the only trace.
"""
assert BUNDLED_DEFINITION_FILES
file_ids = set()
for toml_path, data in BUNDLED_FILE_DATA.items():
sections = data["mpm"][OVERRIDES_SECTION]
assert len(sections) == 1, f"{toml_path.name} must define a single manager"
file_ids.update(sections)
assert file_ids == set(pool.bundled_manager_ids)
[docs]
@pytest.mark.parametrize("toml_path", BUNDLED_DEFINITION_FILES, ids=attrgetter("stem"))
def test_bundled_registered(toml_path):
"""Each shipped definition file lands in the pool as a config-defined manager.
Every expectation derives from the file itself, so a new bundled manager is
covered without extending any hardcoded list here. The file must carry
nothing but its definition and the test samples locking its parsers.
"""
data = BUNDLED_FILE_DATA[toml_path]
assert set(data) <= {"mpm", "samples"}, (
f"unexpected top-level keys in {toml_path.name}"
)
manager_id = next(iter(data["mpm"][OVERRIDES_SECTION]))
assert toml_path.stem == manager_id.replace("-", "_")
assert manager_id in pool.bundled_manager_ids
manager = pool[manager_id]
assert isinstance(manager, ConfigDrivenManager)
assert (
manager.definition_source == f"meta_package_manager/managers/{toml_path.name}"
)
assert manager.name
assert manager.homepage_url or manager.repository_url
# Validate the shape of the sample fixtures consumed by the tests below.
samples = data.get("samples", {})
assert set(samples) <= {"version", "installed", "outdated", "orphans", "search"}
assert "version" in samples, (
f"{toml_path.name} must ship a [samples.version] fixture"
)
assert set(samples["version"]) == {"output", "expected"}
for operation in ("installed", "outdated", "orphans", "search"):
for sample in samples.get(operation, ()):
assert set(sample) == {"output", "packages"}
assert sample["packages"], "a sample must expect at least one package"
for package in sample["packages"]:
assert set(package) <= {"id", "version"}
assert package["id"]
[docs]
@pytest.mark.parametrize("toml_path", BUNDLED_DEFINITION_FILES, ids=attrgetter("stem"))
def test_bundled_definition_carries_no_boilerplate(toml_path):
"""No shipped definition opens on a tag line saying it ships with mpm.
`_toml_definition_intro()` renders a definition's opening comment verbatim
as the manager page's intro, so such a line would print at the top of the
page. A file sitting under `meta_package_manager/managers/` already says
everything the line would.
A test rather than a review note, because a new definition is written by
copying an existing one: a single file carrying the line seeds every file
modelled on it.
"""
assert "definition shipped with mpm" not in toml_path.read_text(encoding="UTF-8"), (
f"{toml_path.name} carries the boilerplate tag line dropped from every "
"bundled definition, which now renders on the manager's page"
)
[docs]
@pytest.mark.parametrize(
("manager_id", "operation", "expected"),
(
("gh-ext", Operations.installed, True),
("gh-ext", Operations.search, True),
("gh-ext", Operations.install, True),
("gh-ext", Operations.remove, True),
("gh-ext", Operations.upgrade, True),
("gh-ext", Operations.upgrade_all, True),
("gh-ext", Operations.outdated, False),
("gh-ext", Operations.sync, False),
("gh-ext", Operations.cleanup, False),
("soar", Operations.installed, True),
("soar", Operations.search, True),
("soar", Operations.install, True),
("soar", Operations.remove, True),
("soar", Operations.upgrade, True),
("soar", Operations.upgrade_all, True),
("soar", Operations.sync, True),
("soar", Operations.cleanup, True),
("soar", Operations.outdated, False),
# SteamCMD exposes no inventory: only install and per-title upgrade
# exist, both mapping to the same idempotent `+app_update` command.
("steamcmd", Operations.install, True),
("steamcmd", Operations.installed, False),
("steamcmd", Operations.outdated, False),
("steamcmd", Operations.search, False),
("steamcmd", Operations.remove, False),
("steamcmd", Operations.sync, False),
("steamcmd", Operations.cleanup, False),
# The declared upgrade_one command is not enough: single-package upgrade
# also needs an `installed` listing to resolve which manager owns the
# package, and SteamCMD cannot provide one. Same verdict as the former
# Python class.
("steamcmd", Operations.upgrade, False),
("steamcmd", Operations.upgrade_all, False),
),
)
def test_bundled_definition_capabilities(manager_id, operation, expected):
"""A bundled definition implements the operations its file maps, and no other."""
assert implements(pool[manager_id], operation) is expected
def _version_sample_params():
"""One param per shipped definition file, from its `[samples.version]` fixture."""
params = []
for data in BUNDLED_FILE_DATA.values():
manager_id = next(iter(data["mpm"][OVERRIDES_SECTION]))
sample = data.get("samples", {}).get("version")
if sample:
params.append(
pytest.param(
manager_id, sample["output"], sample["expected"], id=manager_id
),
)
return params
[docs]
@pytest.mark.parametrize(("manager_id", "output", "expected"), _version_sample_params())
def test_bundled_version_regex(manager_id, output, expected):
"""Lock the version probe of each bundled definition to its output sample.
The sample ships in its TOML file, derived from the upstream tool's own source or
documentation.
"""
match = re.search(pool[manager_id].version_regexes[0], output)
assert match is not None
assert match.group("version") == expected
def _parsing_sample_params():
"""One param per query-operation sample shipped in the definition files."""
params = []
for data in BUNDLED_FILE_DATA.values():
manager_id = next(iter(data["mpm"][OVERRIDES_SECTION]))
samples = data.get("samples", {})
for operation in ("installed", "outdated", "orphans", "search"):
op_samples = samples.get(operation, ())
for index, sample in enumerate(op_samples):
expected = [
(package["id"], package.get("version"))
for package in sample["packages"]
]
param_id = f"{manager_id}-{operation}"
if len(op_samples) > 1:
param_id += f"-{index}"
params.append(
pytest.param(
manager_id, operation, sample["output"], expected, id=param_id
),
)
return params
[docs]
@pytest.mark.parametrize(
("manager_id", "operation", "output", "expected"),
_parsing_sample_params(),
)
def test_bundled_parsing(manager_id, operation, output, expected):
"""Lock the parsers of each bundled definition to its output samples.
The samples ship in its TOML file, derived from the upstream tools' own source code
or documentation.
Versions are compared as raw captured strings (omitted from the sample when
the regex captures no version group). `which` is stubbed so operations
declaring a sibling `cli` resolve without the real binary installed.
"""
manager = _fresh_bundled(manager_id)
manager.which = lambda name: Path("/fake/bin") / name
manager.run_cli = lambda *args, **kwargs: output
if operation == "search":
packages = manager.search("query", False, False)
else:
packages = getattr(manager, operation)
role = (
"installed_version"
if operation in ("installed", "orphans")
else "latest_version"
)
results = [
(p.id, str(version) if (version := getattr(p, role)) else None)
for p in packages
]
assert results == expected