proj/pkgcore/pkgcore:master commit in: /, src/pkgcore/bugzilla/, src/pkgcore/pytest/, tests/bugzilla/
"Arthur Zamarin" <[email protected]>
| Newsgroups | gmane.linux.gentoo.cvs |
|---|---|
| Message-ID | <1786214878.c5606352abdfa9d4c5ef7e4e7e8e31255f762233.arthurzam@gentoo> |
commit: c5606352abdfa9d4c5ef7e4e7e8e31255f762233
Author: Arthur Zamarin <arthurzam <AT> gentoo <DOT> org>
AuthorDate: Sat Aug 8 10:08:37 2026 +0000
Commit: Arthur Zamarin <arthurzam <AT> gentoo <DOT> org>
CommitDate: Sat Aug 8 18:47:58 2026 +0000
URL: https://gitweb.gentoo.org/proj/pkgcore/pkgcore.git/commit/?id=c5606352
bugzilla: new module for interacting with Bugzilla instances
Signed-off-by: Arthur Zamarin <arthurzam <AT> gentoo.org>
NEWS.rst | 25 +++
pyproject.toml | 3 +
src/pkgcore/bugzilla/__init__.py | 60 ++++++
src/pkgcore/bugzilla/apikey.py | 169 ++++++++++++++++
src/pkgcore/bugzilla/bug.py | 304 ++++++++++++++++++++++++++++
src/pkgcore/bugzilla/changes.py | 389 +++++++++++++++++++++++++++++++++++
src/pkgcore/bugzilla/client.py | 259 ++++++++++++++++++++++++
src/pkgcore/bugzilla/enums.py | 209 +++++++++++++++++++
src/pkgcore/bugzilla/errors.py | 176 ++++++++++++++++
src/pkgcore/bugzilla/pkglist.py | 201 +++++++++++++++++++
src/pkgcore/bugzilla/query.py | 312 +++++++++++++++++++++++++++++
src/pkgcore/bugzilla/testing.py | 271 +++++++++++++++++++++++++
src/pkgcore/bugzilla/transport.py | 263 ++++++++++++++++++++++++
src/pkgcore/bugzilla/wire.py | 230 +++++++++++++++++++++
src/pkgcore/pytest/plugin.py | 15 ++
tests/bugzilla/__init__.py | 0
tests/bugzilla/conftest.py | 34 ++++
tests/bugzilla/test_apikey.py | 145 ++++++++++++++
tests/bugzilla/test_bug.py | 294 +++++++++++++++++++++++++++
tests/bugzilla/test_changes.py | 412 ++++++++++++++++++++++++++++++++++++++
tests/bugzilla/test_client.py | 394 ++++++++++++++++++++++++++++++++++++
tests/bugzilla/test_network.py | 97 +++++++++
tests/bugzilla/test_pkglist.py | 206 +++++++++++++++++++
tests/bugzilla/test_query.py | 285 ++++++++++++++++++++++++++
tests/bugzilla/test_testing.py | 141 +++++++++++++
tests/bugzilla/test_transport.py | 288 ++++++++++++++++++++++++++
26 files changed, 5182 insertions(+)
diff --git a/NEWS.rst b/NEWS.rst
index ebc16f818..455b78604 100644
--- a/NEWS.rst
+++ b/NEWS.rst
@@ -6,6 +6,31 @@ Release Notes
pkgcore 0.12.38 (unreleased)
----------------------------
+Features
+~~~~~~~~
+
+- ``pkgcore.bugzilla``: new fully annotated client for the bugs.gentoo.org REST
+ API, covering search, filing, updates and comments, on stdlib ``urllib``.
+ Searches compose (``BugQuery.category(...) & BugQuery.unresolved()``), with
+ Bugzilla's boolean-chart slots allocated at render time so any two queries
+ can be combined, and oversized queries split by measured URL length rather
+ than a guessed item count. Searches also page explicitly, instead of
+ silently stopping at ``max_search_results``. Updates are built from typed
+ ``ListChange`` objects, making the ``cc_add`` spelling that Bugzilla accepts
+ and ignores unwritable. ``cf_stabilisation_atoms`` parses to real atoms and
+ rewrites in place, preserving comments and line endings. The API key is
+ found via ``--api-key``, ``$BUGZ_API_KEY``, ``~/.bugzrc`` or
+ ``~/.bugz_token``, and is kept out of exception messages (Arthur Zamarin)
+
+- ``pkgcore.bugzilla.testing``: replay helpers for testing code that talks to
+ Bugzilla, also exposed as the ``bugzilla_cassette`` pytest fixture so
+ downstream projects get it without any conftest wiring. A cassette answers
+ queued responses and records the requests they answered; used as a context
+ manager it takes over the opener every client builds, capturing a client
+ constructed deep inside a CLI command without patching ``urlopen``
+ (Arthur Zamarin)
+
+
Fixes
~~~~~
diff --git a/pyproject.toml b/pyproject.toml
index de0b6c069..e2c6204ba 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -100,5 +100,8 @@ ignore = [
"S110", # try-except-pass
]
+[tool.ruff.lint.flake8-bugbear]
+extend-immutable-calls = ["pkgcore.bugzilla.bug._field"]
+
[tool.vulture]
paths = ["src/pkgcore"]
diff --git a/src/pkgcore/bugzilla/__init__.py b/src/pkgcore/bugzilla/__init__.py
new file mode 100644
index 000000000..63aa3f5e5
--- /dev/null
+++ b/src/pkgcore/bugzilla/__init__.py
@@ -0,0 +1,60 @@
+"""Typed client for the bugs.gentoo.org Bugzilla REST API.
+
+The public surface is re-exported here; the implementation is split across
+submodules so that :mod:`pkgcore.bugzilla.wire` and friends can be imported
+without dragging in the HTTP transport.
+
+ >>> from pkgcore.bugzilla import Bugzilla, BugQuery, Component, FlagStatus
+ >>> bz = Bugzilla() # doctest: +SKIP
+ >>> query = (BugQuery.component(Component.STABILIZATION)
+ ... & BugQuery.unresolved()
+ ... & BugQuery.flag("sanity-check", FlagStatus.GRANTED))
+ >>> bugs = bz.search(query) # doctest: +SKIP
+"""
+
+__all__ = (
+ "INCLUDE_FIELDS",
+ "AuthMode",
+ "Bug",
+ "BugCategory",
+ "BugChanges",
+ "BugQuery",
+ "BugUpdate",
+ "Bugzilla",
+ "BugzillaError",
+ "Comment",
+ "Component",
+ "Criterion",
+ "Flag",
+ "FlagChange",
+ "FlagStatus",
+ "ListChange",
+ "NewBug",
+ "NewComment",
+ "PackageList",
+ "PackageListEntry",
+ "Product",
+ "Resolution",
+ "RuntimeTesting",
+ "Severity",
+ "Status",
+ "User",
+)
+
+from .bug import INCLUDE_FIELDS, Bug, BugChanges, Comment, Flag, User
+from .changes import BugUpdate, FlagChange, ListChange, NewBug, NewComment
+from .client import Bugzilla
+from .enums import (
+ BugCategory,
+ Component,
+ FlagStatus,
+ Product,
+ Resolution,
+ RuntimeTesting,
+ Severity,
+ Status,
+)
+from .errors import BugzillaError
+from .pkglist import PackageList, PackageListEntry
+from .query import BugQuery, Criterion
+from .transport import AuthMode
diff --git a/src/pkgcore/bugzilla/apikey.py b/src/pkgcore/bugzilla/apikey.py
new file mode 100644
index 000000000..9f2c1bd0d
--- /dev/null
+++ b/src/pkgcore/bugzilla/apikey.py
@@ -0,0 +1,169 @@
+"""Locating a bugs.gentoo.org api key, and wiring it into an argparser.
+
+:func:`find_api_key` consults four sources in order, stopping at the first one
+holding a non-empty value:
+
+1. an explicit key, normally from ``--api-key``. Convenient, but visible to
+ anyone able to read ``ps``.
+2. the ``BUGZ_API_KEY`` environment variable. The only source usable from CI
+ without writing a secret to disk.
+3. ``~/.bugzrc``, an INI file whose name and layout come from pybugz. The
+ sections ``default``, ``gentoo`` and ``Gentoo`` are tried in that order::
+
+ [default]
+ key = AbCdEf0123456789AbCdEf0123456789AbCdEf01
+
+ Only the ``key`` option is read, so an existing pybugz config works as is.
+
+4. ``~/.bugz_token``, holding nothing but the key::
+
+ AbCdEf0123456789AbCdEf0123456789AbCdEf01
+
+Either file being group or world readable is warned about, but still used.
+
+Finding no key at all is not an error. The client then runs anonymously, which
+is read only, and which Bugzilla further degrades by truncating every email
+address it returns at the ``@``.
+"""
+
+__all__ = (
+ "API_KEY_ENV",
+ "BugzillaApiKey",
+ "BugzillaClientArgs",
+ "find_api_key",
+)
+
+import os
+import stat
+import typing
+from configparser import ConfigParser
+from configparser import Error as ConfigParserError
+from pathlib import Path
+
+from ..log import logger
+from .client import DEFAULT_URL, Bugzilla
+from .errors import BugzillaUsageError
+
+API_KEY_ENV: typing.Final = "BUGZ_API_KEY"
+
+_RC_FILE: typing.Final = ".bugzrc"
+_RC_SECTIONS: typing.Final = ("default", "gentoo", "Gentoo")
+_TOKEN_FILE: typing.Final = ".bugz_token"
+
+API_KEY_DOCS: typing.Final = """
+ The Bugzilla API key to use for authentication. WARNING: passing the key
+ here exposes it to every other user of the system, via ``ps``; prefer one
+ of the other sources below.
+
+ Four sources are consulted in order, and the first non-empty one wins:
+
+ 1. this option
+ 2. the ``BUGZ_API_KEY`` environment variable
+ 3. ``~/.bugzrc``, an INI file, looking at the ``default``, ``gentoo`` and
+ ``Gentoo`` sections in that order::
+
+ [default]
+ key = AbCdEf0123456789AbCdEf0123456789AbCdEf01
+
+ Only ``key`` is read, so an existing pybugz config can be left alone.
+
+ 4. ``~/.bugz_token``, holding nothing but the key::
+
+ AbCdEf0123456789AbCdEf0123456789AbCdEf01
+
+ A warning is emitted if either file is group or world readable.
+
+ Without a key the client is read only, and Bugzilla truncates every email
+ address it returns at the ``@``.
+"""
+
+
+def _warn_if_readable(path: Path) -> None:
+ try:
+ mode = path.stat().st_mode
+ except OSError:
+ return
+ if mode & (stat.S_IRWXG | stat.S_IRWXO):
+ logger.warning("%s holds an api key and is readable by others", path)
+
+
+def find_api_key(
+ explicit: str | None = None, *, allow_env: bool = True, home: Path | None = None
+) -> str | None:
+ """Locate an api key, per the precedence documented for this module.
+
+ :param explicit: a key supplied directly, normally from ``--api-key``;
+ blank or whitespace-only is treated as absent
+ :param allow_env: whether ``$BUGZ_API_KEY`` may be consulted
+ :param home: directory to look the dotfiles up in, defaulting to the
+ user's home
+ :return: the key, or None to run anonymously
+ :raises BugzillaUsageError: if ``~/.bugzrc`` exists but can't be parsed
+ """
+ if explicit and (explicit := explicit.strip()):
+ return explicit
+ if allow_env and (key := os.environ.get(API_KEY_ENV, "").strip()):
+ return key
+ home = home or Path.home()
+ if (rc := home / _RC_FILE).is_file():
+ _warn_if_readable(rc)
+ config = ConfigParser(default_section=_RC_SECTIONS[0])
+ try:
+ config.read(rc)
+ except (ConfigParserError, OSError, UnicodeDecodeError) as exc:
+ raise BugzillaUsageError(f"failed parsing {rc}: {exc}") from exc
+ for section in _RC_SECTIONS:
+ if config.has_option(section, "key") and (
+ key := config.get(section, "key").strip()
+ ):
+ return key
+ if (token := home / _TOKEN_FILE).is_file():
+ _warn_if_readable(token)
+ if key := token.read_text().strip():
+ return key
+ return None
+
+
+class BugzillaApiKey:
+ """Adds ``--api-key``, defaulting through :func:`find_api_key`"""
+
+ @classmethod
+ def mangle_argparser(cls, parser: typing.Any) -> None:
+ parser.add_argument(
+ "--api-key", metavar="TOKEN", help="Bugzilla API key", docs=API_KEY_DOCS
+ )
+ parser.bind_delayed_default(1000, "api_key")(cls._default_api_key)
+
+ @staticmethod
+ def _default_api_key(namespace: typing.Any, attr: str) -> None:
+ setattr(namespace, attr, find_api_key())
+
+
+class BugzillaClientArgs(BugzillaApiKey):
+ """As :class:`BugzillaApiKey`, plus ``--bugzilla-url`` and a ready client.
+
+ The client lands on ``namespace.bugzilla``; its delayed default runs after
+ the key's, so the key is resolved by the time it is built.
+ """
+
+ @classmethod
+ def mangle_argparser(cls, parser: typing.Any) -> None:
+ super().mangle_argparser(parser)
+ parser.add_argument(
+ "--bugzilla-url",
+ metavar="URL",
+ default=DEFAULT_URL,
+ help="base URL of the Bugzilla instance",
+ )
+ parser.bind_delayed_default(1001, "bugzilla")(cls._default_client)
+
+ @staticmethod
+ def _default_client(namespace: typing.Any, attr: str) -> None:
+ setattr(
+ namespace,
+ attr,
+ Bugzilla(
+ namespace.api_key,
+ base_url=getattr(namespace, "bugzilla_url", DEFAULT_URL),
+ ),
+ )
diff --git a/src/pkgcore/bugzilla/bug.py b/src/pkgcore/bugzilla/bug.py
new file mode 100644
index 000000000..dcda976e2
--- /dev/null
+++ b/src/pkgcore/bugzilla/bug.py
@@ -0,0 +1,304 @@
+"""Immutable value objects for what bugs.gentoo.org hands back.
+
+Each :class:`Bug` field carries the wire name it came from and how to decode
+it, so :data:`INCLUDE_FIELDS` and :func:`parse_bug` are both derived from the
+one declaration and can't drift apart.
+"""
+
+__all__ = (
+ "INCLUDE_FIELDS",
+ "Bug",
+ "BugChanges",
+ "Comment",
+ "FieldChange",
+ "Flag",
+ "User",
+ "parse_bug",
+ "parse_changes",
+ "parse_comment",
+ "parse_user",
+)
+
+import dataclasses
+import datetime
+import typing
+
+from ..log import logger
+from .enums import BugCategory, FlagStatus, Product, RuntimeTesting
+from .pkglist import PackageList
+from .wire import (
+ BugId,
+ CommentId,
+ FlagId,
+ FlagTypeId,
+ RawBug,
+ RawChanges,
+ RawComment,
+ RawFlag,
+ RawWhoami,
+)
+
+_EPOCH: typing.Final = datetime.datetime.fromtimestamp(0, datetime.UTC)
+_NO_FLAG_ID: typing.Final = FlagId(0)
+_NO_FLAG_TYPE_ID: typing.Final = FlagTypeId(0)
+
+
+def _field[T](wire: str, parse: typing.Callable[[typing.Any], T], default: T) -> T:
+ """Declare a field along with its wire name and decoder.
+
+ Returns ``T`` rather than ``Field[T]`` for the same reason
+ :func:`dataclasses.field` returns ``Any``, so the class body stays readable.
+ """
+ return dataclasses.field(default=default, metadata={"wire": wire, "parse": parse})
+
+
+def _strs(value: typing.Any) -> tuple[str, ...]:
+ return tuple(value)
+
+
+def _ids(value: typing.Any) -> tuple[BugId, ...]:
+ return tuple(BugId(x) for x in value)
+
+
+def _datetime(value: typing.Any) -> datetime.datetime:
+ return datetime.datetime.fromisoformat(value)
+
+
+def _date(value: typing.Any) -> datetime.date | None:
+ return datetime.date.fromisoformat(value) if value else None
+
+
+def _runtime_testing(value: typing.Any) -> RuntimeTesting:
+ try:
+ return RuntimeTesting(str(value).capitalize())
+ except ValueError:
+ return RuntimeTesting.UNSET
+
+
[email protected](frozen=True, slots=True)
+class Flag:
+ """A flag set on a bug"""
+
+ name: str
+ status: FlagStatus
+ id: FlagId = _NO_FLAG_ID
+ type_id: FlagTypeId = _NO_FLAG_TYPE_ID
+ setter: str = ""
+ requestee: str = ""
+
+ @property
+ def granted(self) -> bool | None:
+ """Tri-state view: True for ``+``, False for ``-``, None otherwise"""
+ if self.status is FlagStatus.GRANTED:
+ return True
+ if self.status is FlagStatus.DENIED:
+ return False
+ return None
+
+
+def _flags(value: typing.Any) -> tuple[Flag, ...]:
+ return tuple(parse_flag(x) for x in value)
+
+
+def parse_flag(raw: RawFlag) -> Flag:
+ return Flag(
+ name=raw["name"],
+ status=FlagStatus(raw["status"]),
+ id=raw.get("id", _NO_FLAG_ID),
+ type_id=raw.get("type_id", _NO_FLAG_TYPE_ID),
+ setter=raw.get("setter", ""),
+ requestee=raw.get("requestee", ""),
+ )
+
+
[email protected](frozen=True, slots=True, kw_only=True)
+class Bug:
+ """An immutable snapshot of a bug.
+
+ Anonymous requests get every email address truncated at the ``@``, so
+ :attr:`assigned_to`, :attr:`creator` and :attr:`cc` only hold full
+ addresses when the client was given an api key.
+ """
+
+ id: BugId = _field("id", BugId, BugId(0))
+ summary: str = _field("summary", str, "")
+ product: str = _field("product", str, "")
+ component: str = _field("component", str, "")
+ version: str = _field("version", str, "")
+ status: str = _field("status", str, "")
+ resolution: str = _field("resolution", str, "")
+ severity: str = _field("severity", str, "")
+ priority: str = _field("priority", str, "")
+ assigned_to: str = _field("assigned_to", str, "")
+ creator: str = _field("creator", str, "")
+ cc: tuple[str, ...] = _field("cc", _strs, ())
+ keywords: tuple[str, ...] = _field("keywords", _strs, ())
+ whiteboard: str = _field("whiteboard", str, "")
+ alias: tuple[str, ...] = _field("alias", _strs, ())
+ tags: tuple[str, ...] = _field("tags", _strs, ())
+ depends_on: tuple[BugId, ...] = _field("depends_on", _ids, ())
+ blocks: tuple[BugId, ...] = _field("blocks", _ids, ())
+ see_also: tuple[str, ...] = _field("see_also", _strs, ())
+ groups: tuple[str, ...] = _field("groups", _strs, ())
+ flags: tuple[Flag, ...] = _field("flags", _flags, ())
+ deadline: datetime.date | None = _field("deadline", _date, None)
+ creation_time: datetime.datetime = _field("creation_time", _datetime, _EPOCH)
+ last_change_time: datetime.datetime = _field("last_change_time", _datetime, _EPOCH)
+ package_list: PackageList = _field(
+ "cf_stabilisation_atoms", PackageList, PackageList()
+ )
+ runtime_testing_required: RuntimeTesting = _field(
+ "cf_runtime_testing_required", _runtime_testing, RuntimeTesting.UNSET
+ )
+
+ @property
+ def category(self) -> BugCategory | None:
+ return BugCategory.from_product_component(self.product, self.component)
+
+ @property
+ def resolved(self) -> bool:
+ return bool(self.resolution)
+
+ @property
+ def security(self) -> bool:
+ """Whether the bug lives in the security product.
+
+ Orthogonal to carrying the ``SECURITY`` keyword, which marks an
+ ordinary bug as blocking a security one.
+ """
+ return self.product == Product.GENTOO_SECURITY
+
+ @property
+ def sanity_check(self) -> bool | None:
+ return self.flag("sanity-check")
+
+ def flag(self, name: str) -> bool | None:
+ """Tri-state status of a named flag, None when it isn't set"""
+ for flag in self.flags:
+ if flag.name == name:
+ return flag.granted
+ return None
+
+ def arches(self, known_arches: typing.Container[str]) -> tuple[str, ...]:
+ """Arch names found in CC, tolerating truncated anonymous addresses"""
+ return tuple(
+ name
+ for entry in self.cc
+ if (entry.endswith("@gentoo.org") or "@" not in entry)
+ and (name := entry.split("@", 1)[0]) in known_arches
+ )
+
+ @property
+ def url(self) -> str:
+ return f"https://bugs.gentoo.org/{self.id}"
+
+
+_SPECS: typing.Final = tuple(
+ (field.name, field.metadata["wire"], field.metadata["parse"])
+ for field in dataclasses.fields(Bug)
+ if "wire" in field.metadata
+)
+
+INCLUDE_FIELDS: typing.Final[tuple[str, ...]] = tuple(
+ dict.fromkeys(wire for _, wire, _ in _SPECS)
+)
+
+
+def parse_bug(raw: RawBug) -> Bug:
+ """Build a :class:`Bug` from a raw response, skipping absent fields"""
+ data = typing.cast(dict[str, typing.Any], raw)
+ kwargs: dict[str, typing.Any] = {}
+ for name, wire, parse in _SPECS:
+ if wire not in data:
+ logger.debug("bug %s: field %r absent from response", data.get("id"), wire)
+ elif (value := data[wire]) is not None:
+ kwargs[name] = parse(value)
+ # tie the package list back to its bug, so a parse failure says which one
+ if (pkglist := kwargs.get("package_list")) is not None and "id" in kwargs:
+ kwargs["package_list"] = PackageList(pkglist.text, bug_id=kwargs["id"])
+ return Bug(**kwargs)
+
+
[email protected](frozen=True, slots=True)
+class Comment:
+ """A single comment on a bug"""
+
+ id: CommentId
+ bug_id: BugId
+ count: int
+ text: str
+ creator: str
+ creation_time: datetime.datetime
+ is_private: bool = False
+ tags: tuple[str, ...] = ()
+
+ @property
+ def obsolete(self) -> bool:
+ return "obsolete" in self.tags
+
+
+def parse_comment(raw: RawComment) -> Comment:
+ return Comment(
+ id=raw["id"],
+ bug_id=raw["bug_id"],
+ count=raw["count"],
+ text=raw["text"],
+ creator=raw["creator"],
+ creation_time=_datetime(raw["creation_time"]),
+ is_private=raw.get("is_private", False),
+ tags=tuple(raw.get("tags", ())),
+ )
+
+
[email protected](frozen=True, slots=True)
+class User:
+ """The account an api key belongs to"""
+
+ id: int
+ name: str
+ real_name: str = ""
+
+
+def parse_user(raw: RawWhoami) -> User:
+ return User(id=raw["id"], name=raw["name"], real_name=raw.get("real_name", ""))
+
+
[email protected](frozen=True, slots=True)
+class FieldChange:
+ """What one field gained and lost in an update"""
+
+ added: tuple[str, ...] = ()
+ removed: tuple[str, ...] = ()
+
+
[email protected](frozen=True, slots=True)
+class BugChanges:
+ """The result of updating a bug"""
+
+ id: BugId
+ last_change_time: datetime.datetime
+ changes: dict[str, FieldChange] = dataclasses.field(default_factory=dict)
+ alias: tuple[str, ...] = ()
+
+ def __bool__(self) -> bool:
+ return bool(self.changes)
+
+
+def _split_change(value: str) -> tuple[str, ...]:
+ # bugzilla joins these with ", " rather than returning a list
+ return tuple(x for x in (part.strip() for part in value.split(",")) if x)
+
+
+def parse_changes(raw: RawChanges) -> BugChanges:
+ return BugChanges(
+ id=raw["id"],
+ last_change_time=_datetime(raw["last_change_time"]),
+ changes={
+ name: FieldChange(
+ added=_split_change(change.get("added", "")),
+ removed=_split_change(change.get("removed", "")),
+ )
+ for name, change in raw.get("changes", {}).items()
+ },
+ alias=tuple(raw.get("alias", ())),
+ )
diff --git a/src/pkgcore/bugzilla/changes.py b/src/pkgcore/bugzilla/changes.py
new file mode 100644
index 000000000..8c6d8efc6
--- /dev/null
+++ b/src/pkgcore/bugzilla/changes.py
@@ -0,0 +1,389 @@
+"""Typed payloads for creating and updating bugs.
+
+Bugzilla mutates list valued fields through an ``{"add": [...], "remove": [...]}``
+object, and silently ignores anything it doesn't recognise, so a misspelt key
+like ``cc_add`` is accepted and does nothing. :class:`ListChange` and the frozen
+payload classes here make that shape the only one expressible.
+"""
+
+__all__ = (
+ "MAINTAINER_NEEDED",
+ "MAX_COMMENT_LENGTH",
+ "MAX_SUMMARY_LENGTH",
+ "TREECLEANER",
+ "BugUpdate",
+ "FlagChange",
+ "ListChange",
+ "NewBug",
+ "NewComment",
+ "summarise",
+)
+
+import dataclasses
+import datetime
+import typing
+
+from .enums import (
+ BugCategory,
+ Component,
+ FlagStatus,
+ Product,
+ Resolution,
+ RuntimeTesting,
+ Severity,
+ Status,
+)
+from .errors import BugzillaUsageError
+from .pkglist import PackageList
+from .wire import (
+ BugId,
+ RawBugUpdate,
+ RawFlagChange,
+ RawListChange,
+ RawNewBug,
+ RawNewComment,
+)
+
+# bugzilla rejects longer comments with error 114
+MAX_COMMENT_LENGTH: typing.Final = 65535
+
+# the length past which a package list summary is collapsed to "and friends"
+MAX_SUMMARY_LENGTH: typing.Final = 90
+
+MAINTAINER_NEEDED: typing.Final = "[email protected]"
+TREECLEANER: typing.Final = "[email protected]"
+
+
[email protected](frozen=True, slots=True)
+class ListChange[T]:
+ """An add/remove/set mutation of a list valued field"""
+
+ add: tuple[T, ...] = ()
+ remove: tuple[T, ...] = ()
+ replace: tuple[T, ...] | None = None
+
+ def __post_init__(self) -> None:
+ if self.replace is not None and (self.add or self.remove):
+ raise BugzillaUsageError("replace cannot be combined with add or remove")
+ if overlap := frozenset(self.add) & frozenset(self.remove):
+ raise BugzillaUsageError(
+ f"the same value is both added and removed: {sorted(map(str, overlap))}"
+ )
+
+ @classmethod
+ def adding(cls, *values: T) -> "ListChange[T]":
+ return cls(add=values)
+
+ @classmethod
+ def removing(cls, *values: T) -> "ListChange[T]":
+ return cls(remove=values)
+
+ @classmethod
+ def setting(cls, *values: T) -> "ListChange[T]":
+ """Replace the field wholesale, Bugzilla's ``set``"""
+ return cls(replace=values)
+
+ def __bool__(self) -> bool:
+ return bool(self.add or self.remove or self.replace is not None)
+
+ def __or__(self, other: "ListChange[T]") -> "ListChange[T]":
+ if other.replace is not None:
+ return other
+ return ListChange(
+ add=self.add + tuple(x for x in other.add if x not in self.add),
+ remove=self.remove + tuple(x for x in other.remove if x not in self.remove),
+ )
+
+ def to_wire(self) -> RawListChange:
+ if self.replace is not None:
+ return {"set": [str(x) for x in self.replace]}
+ wire: RawListChange = {}
+ if self.add:
+ wire["add"] = [str(x) for x in self.add]
+ if self.remove:
+ wire["remove"] = [str(x) for x in self.remove]
+ return wire
+
+
[email protected](frozen=True, slots=True)
+class FlagChange:
+ """Set or clear a flag"""
+
+ name: str
+ status: FlagStatus
+ requestee: str | None = None
+
+ def to_wire(self) -> RawFlagChange:
+ wire: RawFlagChange = {"name": self.name, "status": self.status.value}
+ if self.requestee is not None:
+ wire["requestee"] = self.requestee
+ return wire
+
+
[email protected](frozen=True, slots=True)
+class NewComment:
+ """A comment to leave alongside an update"""
+
+ body: str
+ is_private: bool = False
+
+ def __post_init__(self) -> None:
+ if len(self.body) > MAX_COMMENT_LENGTH:
+ raise BugzillaUsageError(
+ f"comment is {len(self.body)} characters, the limit is "
+ f"{MAX_COMMENT_LENGTH}; use NewComment.truncated()"
+ )
+
+ @classmethod
+ def truncated(
+ cls, body: str, limit: int = MAX_COMMENT_LENGTH, **kwargs: typing.Any
+ ) -> "NewComment":
+ """Build a comment, cutting an overlong body on a line boundary"""
+ if len(body) > limit:
+ marker = "\n...\n"
+ head = body[: limit - len(marker)]
+ body = head[: head.rfind("\n") + 1 or len(head)].rstrip() + marker
+ return cls(body, **kwargs)
+
+ def to_wire(self) -> RawNewComment:
+ wire: RawNewComment = {"body": self.body}
+ if self.is_private:
+ wire["is_private"] = True
+ return wire
+
+
+def summarise(package_list: PackageList, category: BugCategory) -> str:
+ """Build the conventional summary for an arch team request"""
+ names = [
+ pkg.cpvstr if pkg.op == "=" else str(pkg.key) for pkg in package_list.atoms
+ ]
+ if not names:
+ raise BugzillaUsageError("cannot summarise an empty package list")
+ summary = f"{', '.join(names)}: {category.summary_suffix}"
+ if len(summary) > MAX_SUMMARY_LENGTH and len(names) > 1:
+ summary = f"{names[0]} and friends: {category.summary_suffix}"
+ return summary
+
+
[email protected](frozen=True, slots=True, kw_only=True)
+class NewBug:
+ """A bug to file"""
+
+ summary: str
+ description: str
+ component: Component | str
+ product: Product | str = Product.GENTOO_LINUX
+ version: str = "unspecified"
+ severity: Severity | str = Severity.NORMAL
+ assigned_to: str | None = None
+ cc: tuple[str, ...] = ()
+ keywords: tuple[str, ...] = ()
+ depends_on: tuple[BugId, ...] = ()
+ blocks: tuple[BugId, ...] = ()
+ see_also: tuple[str, ...] = ()
+ deadline: datetime.date | None = None
+ package_list: PackageList | None = None
+ runtime_testing_required: RuntimeTesting | None = None
+
+ def __post_init__(self) -> None:
+ if not self.summary.strip():
+ raise BugzillaUsageError("a new bug needs a summary")
+
+ @classmethod
+ def arch_request(
+ cls,
+ category: BugCategory,
+ package_list: PackageList,
+ *,
+ maintainers: typing.Sequence[str] = (),
+ cc_arches: bool = False,
+ summary: str | None = None,
+ description: str | None = None,
+ **kwargs: typing.Any,
+ ) -> "NewBug":
+ """A keywordreq or stablereq, with the bgo conventions applied"""
+ assignee, *cc = maintainers or (MAINTAINER_NEEDED,)
+ return cls(
+ product=category.product,
+ component=category.component,
+ severity=Severity.ENHANCEMENT,
+ summary=summary or summarise(package_list, category),
+ description=description or f"Please {category.verb} the listed packages.",
+ keywords=("CC-ARCHES",) if cc_arches else (),
+ assigned_to=assignee,
+ cc=tuple(cc),
+ package_list=package_list,
+ **kwargs,
+ )
+
+ @classmethod
+ def package_mask(
+ cls,
+ summary: str,
+ description: str,
+ *,
+ rites: int,
+ maintainers: typing.Sequence[str] = (),
+ today: datetime.date | None = None,
+ **kwargs: typing.Any,
+ ) -> "NewBug":
+ """A last rites tracker, masked for ``rites`` days"""
+ assignee, *cc = maintainers or (MAINTAINER_NEEDED,)
+ today = today or datetime.datetime.now(datetime.UTC).date()
+ return cls(
+ component=Component.CURRENT_PACKAGES,
+ summary=summary,
+ description=description,
+ keywords=("PMASKED",),
+ assigned_to=assignee,
+ cc=(*cc, TREECLEANER),
+ deadline=today + datetime.timedelta(days=rites),
+ **kwargs,
+ )
+
+ def to_wire(self) -> RawNewBug:
+ wire: RawNewBug = {
+ "product": str(self.product),
+ "component": str(self.component),
+ "summary": self.summary,
+ "description": self.description,
+ "version": self.version,
+ "severity": str(self.severity),
+ }
+ if self.assigned_to:
+ wire["assigned_to"] = self.assigned_to
+ if self.cc:
+ wire["cc"] = list(self.cc)
+ if self.keywords:
+ wire["keywords"] = list(self.keywords)
+ if self.depends_on:
+ wire["depends_on"] = list(self.depends_on)
+ if self.blocks:
+ wire["blocks"] = list(self.blocks)
+ if self.see_also:
+ wire["see_also"] = list(self.see_also)
+ if self.deadline is not None:
+ wire["deadline"] = self.deadline.isoformat()
+ if self.package_list is not None:
+ wire["cf_stabilisation_atoms"] = str(self.package_list)
+ if self.runtime_testing_required is not None:
+ wire["cf_runtime_testing_required"] = str(self.runtime_testing_required)
+ return wire
+
+
[email protected](frozen=True, slots=True, kw_only=True)
+class BugUpdate:
+ """A patch to apply to one or more bugs.
+
+ Every field defaults to leaving the bug alone. List valued fields only
+ accept a :class:`ListChange`, so the ``cc_add`` shape Bugzilla ignores is
+ both a static and a runtime error.
+ """
+
+ status: Status | None = None
+ resolution: Resolution | None = None
+ dupe_of: BugId | None = None
+ summary: str | None = None
+ assigned_to: str | None = None
+ whiteboard: str | None = None
+ deadline: datetime.date | None = None
+ cc: ListChange[str] = ListChange()
+ keywords: ListChange[str] = ListChange()
+ blocks: ListChange[BugId] = ListChange()
+ depends_on: ListChange[BugId] = ListChange()
+ see_also: ListChange[str] = ListChange()
+ groups: ListChange[str] = ListChange()
+ flags: tuple[FlagChange, ...] = ()
+ comment: NewComment | None = None
+ package_list: PackageList | None = None
+ runtime_testing_required: RuntimeTesting | None = None
+
+ def __post_init__(self) -> None:
+ if self.resolution is not None and self.status is None:
+ raise BugzillaUsageError("a resolution needs an explicit status")
+ if self.status is Status.RESOLVED and self.resolution is None:
+ raise BugzillaUsageError("status=RESOLVED needs a resolution")
+ if (self.resolution is Resolution.DUPLICATE) != (self.dupe_of is not None):
+ raise BugzillaUsageError("DUPLICATE and dupe_of must be used together")
+
+ def __bool__(self) -> bool:
+ return any(
+ bool(getattr(self, field.name)) for field in dataclasses.fields(self)
+ )
+
+ @classmethod
+ def sanity_check(
+ cls, status: bool | None, *, comment: str | None = None, **kwargs: typing.Any
+ ) -> "BugUpdate":
+ """Set the sanity-check flag, None clearing it"""
+ flag = {
+ True: FlagStatus.GRANTED,
+ False: FlagStatus.DENIED,
+ None: FlagStatus.CLEARED,
+ }
+ return cls(
+ flags=(FlagChange("sanity-check", flag[status]),),
+ comment=NewComment(comment) if comment is not None else None,
+ **kwargs,
+ )
+
+ @classmethod
+ def resolve(
+ cls,
+ resolution: Resolution = Resolution.FIXED,
+ *,
+ comment: str | None = None,
+ **kwargs: typing.Any,
+ ) -> "BugUpdate":
+ return cls(
+ status=Status.RESOLVED,
+ resolution=resolution,
+ comment=NewComment(comment) if comment is not None else None,
+ **kwargs,
+ )
+
+ @classmethod
+ def obsoleted_by(cls, bug: BugId | int, **kwargs: typing.Any) -> "BugUpdate":
+ """Close as OBSOLETE, pointing at the bug that supersedes this one"""
+ return cls(
+ status=Status.RESOLVED,
+ resolution=Resolution.OBSOLETE,
+ see_also=ListChange.adding(f"https://bugs.gentoo.org/{bug}"),
+ **kwargs,
+ )
+
+ def to_wire(self, ids: typing.Sequence[BugId | int]) -> RawBugUpdate:
+ """Render the payload.
+
+ The complete id list is always sent, since Bugzilla lets the body
+ override the id in the request path rather than the other way around.
+ """
+ if not ids:
+ raise BugzillaUsageError("an update needs at least one bug id")
+ wire: RawBugUpdate = {"ids": [BugId(int(x)) for x in ids]}
+ if self.status is not None:
+ wire["status"] = str(self.status)
+ if self.resolution is not None:
+ wire["resolution"] = str(self.resolution)
+ if self.dupe_of is not None:
+ wire["dupe_of"] = self.dupe_of
+ if self.summary is not None:
+ wire["summary"] = self.summary
+ if self.assigned_to is not None:
+ wire["assigned_to"] = self.assigned_to
+ if self.whiteboard is not None:
+ wire["whiteboard"] = self.whiteboard
+ if self.deadline is not None:
+ wire["deadline"] = self.deadline.isoformat()
+ for name in ("cc", "keywords", "blocks", "depends_on", "see_also", "groups"):
+ if change := typing.cast(ListChange[typing.Any], getattr(self, name)):
+ wire[name] = change.to_wire()
+ if self.flags:
+ wire["flags"] = [x.to_wire() for x in self.flags]
+ if self.comment is not None:
+ wire["comment"] = self.comment.to_wire()
+ if self.package_list is not None:
+ wire["cf_stabilisation_atoms"] = str(self.package_list)
+ if self.runtime_testing_required is not None:
+ wire["cf_runtime_testing_required"] = str(self.runtime_testing_required)
+ return wire
diff --git a/src/pkgcore/bugzilla/client.py b/src/pkgcore/bugzilla/client.py
new file mode 100644
index 000000000..de9530464
--- /dev/null
+++ b/src/pkgcore/bugzilla/client.py
@@ -0,0 +1,259 @@
+"""The bugs.gentoo.org client."""
+
+__all__ = ("DEFAULT_URL", "EVERYTHING", "PAGE_SIZE", "Bugzilla")
+
+import typing
+import urllib.parse
+import urllib.request
+
+from ..log import logger
+from .bug import (
+ INCLUDE_FIELDS,
+ Bug,
+ BugChanges,
+ Comment,
+ User,
+ parse_bug,
+ parse_changes,
+ parse_comment,
+ parse_user,
+)
+from .changes import BugUpdate, ListChange, NewBug
+from .errors import BugzillaNotFound, BugzillaSchemaError, BugzillaUsageError
+from .query import BugQuery
+from .transport import AuthMode, Transport, UrllibTransport, expect_list, expect_object
+from .wire import BugId, CommentId, RawBug
+
+DEFAULT_URL: typing.Final = "https://bugs.gentoo.org"
+
+# an unconstrained search, the default for search()/raw_search()
+EVERYTHING: typing.Final = BugQuery()
+
+# bugzilla silently clamps results at max_search_results and reports no total,
+# so searches page explicitly until a short page comes back
+PAGE_SIZE: typing.Final = 500
+
+
+class Bugzilla:
+ """A Bugzilla instance, defaulting to bugs.gentoo.org.
+
+ Without an api key the client is read only, and Bugzilla truncates every
+ email address it returns at the ``@``, so anything matching on addresses
+ needs one.
+
+ ``user_agent`` names the calling tool; it is prepended to pkgcore's own
+ token rather than replacing it.
+ """
+
+ __slots__ = ("_transport", "_user", "base_url")
+
+ def __init__(
+ self,
+ api_key: str | None = None,
+ *,
+ base_url: str = DEFAULT_URL,
+ timeout: float = 30.0,
+ retries: int = 3,
+ auth_mode: AuthMode = AuthMode.QUERY,
+ retry_writes: bool = False,
+ user_agent: str | None = None,
+ opener: urllib.request.OpenerDirector | None = None,
+ transport: Transport | None = None,
+ ) -> None:
+ self.base_url = base_url.rstrip("/")
+ self._transport: Transport = transport or UrllibTransport(
+ self.base_url,
+ api_key,
+ timeout=timeout,
+ retries=retries,
+ auth_mode=auth_mode,
+ retry_writes=retry_writes,
+ user_agent=user_agent,
+ opener=opener,
+ )
+ self._user: User | None = None
+
+ def whoami(self) -> User:
+ """The account the api key belongs to, cached for the session"""
+ if self._user is None:
+ payload = self._transport.request("GET", "whoami")
+ self._user = parse_user(
+ typing.cast(typing.Any, expect_object(payload, "whoami"))
+ )
+ return self._user
+
+ @typing.overload
+ def get(self, bugs: BugId | int, /) -> Bug: ...
+
+ @typing.overload
+ def get(self, bugs: typing.Iterable[BugId | int], /) -> dict[BugId, Bug]: ...
+
+ def get(self, bugs: typing.Any, /) -> typing.Any:
+ """Fetch one bug, or a mapping for several.
+
+ :raises BugzillaNotFound: when a single requested bug doesn't exist
+ """
+ if isinstance(bugs, int):
+ found = self.search(BugQuery.ids((bugs,)))
+ if (bug := found.get(BugId(bugs))) is None:
+ raise BugzillaNotFound(f"bug {bugs} does not exist", code=101)
+ return bug
+ return self.search(BugQuery.ids(bugs))
+
+ def search(self, query: BugQuery = EVERYTHING, /) -> dict[BugId, Bug]:
+ """Run a search, batching and paging as needed"""
+ return {
+ BugId(raw["id"]): parse_bug(raw)
+ for raw in self.raw_search(query)
+ if "id" in raw
+ }
+
+ def raw_search(
+ self,
+ query: BugQuery = EVERYTHING,
+ /,
+ *,
+ fields: typing.Sequence[str] = INCLUDE_FIELDS,
+ ) -> tuple[RawBug, ...]:
+ """Run a search, returning the wire dicts rather than :class:`Bug`.
+
+ Use this for the narrow projections a full :class:`Bug` doesn't need;
+ anything left out of ``fields`` is simply absent from the results.
+ """
+ base = [("include_fields", ",".join(fields))]
+ overhead = len(f"{self.base_url}/rest/bug?") + len(urllib.parse.urlencode(base))
+ results: list[RawBug] = []
+ for batch in query.batches(base_length=overhead):
+ results.extend(self._paged_search(batch, base))
+ return tuple(results)
+
+ def _paged_search(
+ self, query: BugQuery, base: list[tuple[str, str]]
+ ) -> typing.Iterator[RawBug]:
+ offset = query.offset or 0
+ limit = query.limit
+ while True:
+ size = min(limit, PAGE_SIZE) if limit else PAGE_SIZE
+ page = query.paged(size, offset)
+ payload = self._transport.request("GET", "bug", params=base + page.params())
+ bugs = expect_list(payload, "bugs", "bug search")
+ yield from bugs
+ offset += len(bugs)
+ if limit is not None:
+ limit -= len(bugs)
+ if limit <= 0:
+ return
+ if len(bugs) < size:
+ return
+
+ def resolve_dependencies(self, bugs: dict[BugId, Bug]) -> dict[BugId, Bug]:
+ """Fetch the transitive closure of everything ``bugs`` depends on.
+
+ Dependencies that can't be fetched, because they were deleted or are
+ behind a security group, are logged and dropped rather than looping.
+ """
+ resolved = dict(bugs)
+ unreachable: set[BugId] = set()
+ while True:
+ missing = {
+ dep
+ for bug in resolved.values()
+ for dep in bug.depends_on
+ if dep not in resolved and dep not in unreachable
+ }
+ if not missing:
+ return resolved
+ fetched = self.search(BugQuery.ids(sorted(missing)))
+ if absent := missing - fetched.keys():
+ logger.warning(
+ "unreachable bug dependencies, skipping: %s", sorted(absent)
+ )
+ unreachable |= absent
+ resolved.update(fetched)
+
+ def comments(self, bug: BugId | int, /) -> tuple[Comment, ...]:
+ """Every comment on a bug, oldest first"""
+ payload = self._transport.request("GET", f"bug/{int(bug)}/comment")
+ # the bugs mapping is keyed by the *stringified* bug id
+ section = expect_object(payload, "comments").get("bugs", {})
+ if not isinstance(section, dict) or str(bug) not in section:
+ raise BugzillaSchemaError(f"comments: no entry for bug {bug}")
+ return tuple(
+ parse_comment(raw)
+ for raw in expect_list(section[str(bug)], "comments", "comments")
+ )
+
+ def latest_comment(
+ self, bug: BugId | int, /, *, creator: str | None = None
+ ) -> Comment | None:
+ """The newest comment, optionally restricted to one author.
+
+ ``creator`` defaults to the authenticated account.
+ """
+ creator = creator if creator is not None else self.whoami().name
+ for comment in reversed(self.comments(bug)):
+ if comment.creator == creator:
+ return comment
+ return None
+
+ def create(self, bug: NewBug, /) -> BugId:
+ """File a bug and return its id.
+
+ Never retried, since a retry after a timeout files a duplicate.
+ """
+ payload = self._transport.request("POST", "bug", body=bug.to_wire())
+ created = expect_object(payload, "bug creation")
+ if not isinstance(bug_id := created.get("id"), int):
+ raise BugzillaSchemaError(f"bug creation: no id in response {created!r}")
+ return BugId(bug_id)
+
+ @typing.overload
+ def update(self, bugs: BugId | int, /, update: BugUpdate) -> BugChanges: ...
+
+ @typing.overload
+ def update(
+ self, bugs: typing.Iterable[BugId | int], /, update: BugUpdate
+ ) -> tuple[BugChanges, ...]: ...
+
+ def update(self, bugs: typing.Any, /, update: BugUpdate) -> typing.Any:
+ """Apply an update to one or more bugs.
+
+ The full id list always goes in the body, because Bugzilla lets the
+ body override the id in the request path rather than the other way
+ around.
+ """
+ single = isinstance(bugs, int)
+ ids = [int(bugs)] if single else [int(x) for x in bugs]
+ if not ids:
+ raise BugzillaUsageError("update() needs at least one bug id")
+ payload = self._transport.request(
+ "PUT", f"bug/{ids[0]}", body=update.to_wire(ids)
+ )
+ changes = tuple(
+ parse_changes(raw) for raw in expect_list(payload, "bugs", "bug update")
+ )
+ return changes[0] if single else changes
+
+ def tag_comments(
+ self, comments: typing.Iterable[CommentId | int], /, tags: ListChange[str]
+ ) -> None:
+ """Add or remove tags on comments, one request per comment"""
+ for comment in comments:
+ self._transport.request(
+ "PUT", f"bug/comment/{int(comment)}/tags", body=tags.to_wire()
+ )
+
+ def mark_own_comments_obsolete(self, bug: BugId | int, /) -> int:
+ """Tag the authenticated user's comments obsolete, returning the count.
+
+ Deliberately separate from :meth:`update`, so a failed update doesn't
+ leave a bug with every comment obsoleted and no replacement.
+ """
+ username = self.whoami().name
+ stale = [
+ comment.id
+ for comment in self.comments(bug)
+ if comment.creator == username and not comment.obsolete
+ ]
+ self.tag_comments(stale, ListChange.adding("obsolete"))
+ return len(stale)
diff --git a/src/pkgcore/bugzilla/enums.py b/src/pkgcore/bugzilla/enums.py
new file mode 100644
index 000000000..459333525
--- /dev/null
+++ b/src/pkgcore/bugzilla/enums.py
@@ -0,0 +1,209 @@
+"""Bugzilla vocabularies used by bugs.gentoo.org.
+
+Only what pkgcore-adjacent tooling actually writes is enumerated; values read
+back off the wire stay plain strings, so anything Gentoo adds later never
+breaks parsing.
+"""
+
+__all__ = (
+ "OPEN_STATUSES",
+ "UNRESOLVED",
+ "BugCategory",
+ "ChartOp",
+ "Component",
+ "FlagStatus",
+ "Join",
+ "Product",
+ "Resolution",
+ "RuntimeTesting",
+ "Severity",
+ "Status",
+)
+
+import enum
+import typing
+
+
+class Product(enum.StrEnum):
+ """The bugs.gentoo.org products this module knows about"""
+
+ GENTOO_LINUX = "Gentoo Linux"
+ GENTOO_SECURITY = "Gentoo Security"
+
+
+class Component(enum.StrEnum):
+ """Components of Gentoo Linux, plus the security one that matters"""
+
+ STABILIZATION = "Stabilization"
+ KEYWORDING = "Keywording"
+ CURRENT_PACKAGES = "Current packages"
+ NEW_PACKAGES = "New packages"
+ ECLASSES = "Eclasses"
+ PROFILES = "Profiles"
+ VULNERABILITIES = "Vulnerabilities"
+
+
+class Status(enum.StrEnum):
+ """Bug workflow states, there is no NEW, ASSIGNED or CLOSED"""
+
+ UNCONFIRMED = "UNCONFIRMED"
+ CONFIRMED = "CONFIRMED"
+ IN_PROGRESS = "IN_PROGRESS"
+ RESOLVED = "RESOLVED"
+ VERIFIED = "VERIFIED"
+
+ @property
+ def is_open(self) -> bool:
+ return self in OPEN_STATUSES
+
+
+OPEN_STATUSES: typing.Final[tuple["Status", ...]] = (
+ Status.UNCONFIRMED,
+ Status.CONFIRMED,
+ Status.IN_PROGRESS,
+)
+
+# selects unresolved bugs in a search; an open bug stores the empty string
+UNRESOLVED: typing.Final = "---"
+
+
+class Resolution(enum.StrEnum):
+ """Resolutions enabled on bugs.gentoo.org.
+
+ LATER and REMIND are legacy values still set on old bugs, listed so parsing
+ round-trips rather than because anything should write them.
+ """
+
+ FIXED = "FIXED"
+ INVALID = "INVALID"
+ WONTFIX = "WONTFIX"
+ LATER = "LATER"
+ REMIND = "REMIND"
+ DUPLICATE = "DUPLICATE"
+ WORKSFORME = "WORKSFORME"
+ CANTFIX = "CANTFIX"
+ NEEDINFO = "NEEDINFO"
+ TEST_REQUEST = "TEST-REQUEST"
+ UPSTREAM = "UPSTREAM"
+ OBSOLETE = "OBSOLETE"
+ PKGREMOVED = "PKGREMOVED"
+
+
+class Severity(enum.StrEnum):
+ """Bug severities, QA being Gentoo specific"""
+
+ BLOCKER = "blocker"
+ CRITICAL = "critical"
+ MAJOR = "major"
+ NORMAL = "normal"
+ MINOR = "minor"
+ TRIVIAL = "trivial"
+ ENHANCEMENT = "enhancement"
+ QA = "QA"
+
+
+class FlagStatus(enum.StrEnum):
+ """Status of a Bugzilla flag, CLEARED being write only"""
+
+ REQUESTED = "?"
+ GRANTED = "+"
+ DENIED = "-"
+ CLEARED = "X"
+
+
+class RuntimeTesting(enum.StrEnum):
+ """Values of cf_runtime_testing_required.
+
+ The field only exists on the Keywording and Stabilization components, and
+ reads back as UNSET everywhere else.
+ """
+
+ UNSET = "---"
+ YES = "Yes"
+ NO = "No"
+ MANUAL = "Manual"
+
+
+class BugCategory(enum.StrEnum):
+ """Gentoo arch team bug categories, valued by their Bugzilla component"""
+
+ KEYWORDREQ = "Keywording"
+ STABLEREQ = "Stabilization"
+
+ @classmethod
+ def from_product_component(
+ cls, product: str, component: str
+ ) -> "BugCategory | None":
+ """Classify a bug, returning None if it's neither category"""
+ if product == Product.GENTOO_LINUX:
+ try:
+ return cls(component)
+ except ValueError:
+ pass
+ return None
+
+ @property
+ def product(self) -> Product:
+ return Product.GENTOO_LINUX
+
+ @property
+ def component(self) -> Component:
+ return Component(self.value)
+
+ @property
+ def summary_suffix(self) -> str:
+ """The conventional trailing word of the bug summary"""
+ return "keywordreq" if self is BugCategory.KEYWORDREQ else "stablereq"
+
+ @property
+ def verb(self) -> str:
+ """The verb used when describing the request"""
+ return "keyword" if self is BugCategory.KEYWORDREQ else "stabilize"
+
+
+class ChartOp(enum.StrEnum):
+ """Operators accepted by Bugzilla's boolean charts, the o<N> params.
+
+ MATCHES and NOT_MATCHES are only valid against the fulltext content field,
+ and IS_EMPTY/IS_NOT_EMPTY still need a v<N> value even though it's ignored.
+ """
+
+ EQUALS = "equals"
+ NOT_EQUALS = "notequals"
+ CASE_SUBSTRING = "casesubstring"
+ SUBSTRING = "substring"
+ NOT_SUBSTRING = "notsubstring"
+ REGEXP = "regexp"
+ NOT_REGEXP = "notregexp"
+ LESS_THAN = "lessthan"
+ LESS_THAN_EQ = "lessthaneq"
+ GREATER_THAN = "greaterthan"
+ GREATER_THAN_EQ = "greaterthaneq"
+ MATCHES = "matches"
+ NOT_MATCHES = "notmatches"
+ ANY_EXACT = "anyexact"
+ ANY_WORDS_SUBSTR = "anywordssubstr"
+ ALL_WORDS_SUBSTR = "allwordssubstr"
+ NO_WORDS_SUBSTR = "nowordssubstr"
+ ANY_WORDS = "anywords"
+ ALL_WORDS = "allwords"
+ NO_WORDS = "nowords"
+ CHANGED_BEFORE = "changedbefore"
+ CHANGED_AFTER = "changedafter"
+ CHANGED_FROM = "changedfrom"
+ CHANGED_TO = "changedto"
+ CHANGED_BY = "changedby"
+ IS_EMPTY = "isempty"
+ IS_NOT_EMPTY = "isnotempty"
+
+
+class Join(enum.StrEnum):
+ """How a boolean chart group combines its children, the j<N> params.
+
+ AND_G requires every condition to match the same row, which is what
+ constraining a single flag or attachment needs.
+ """
+
+ AND = "AND"
+ OR = "OR"
+ AND_G = "AND_G"
diff --git a/src/pkgcore/bugzilla/errors.py b/src/pkgcore/bugzilla/errors.py
new file mode 100644
index 000000000..1e88a93da
--- /dev/null
+++ b/src/pkgcore/bugzilla/errors.py
@@ -0,0 +1,176 @@
+"""Exceptions raised by pkgcore.bugzilla.
+
+Bugzilla reports failures as a JSON body {"error": true, "code": N, "message":
+...}, and derives the HTTP status from the code through a lossy table. Dispatch
+on the code, not the status: an invalid api key is code 306 delivered as HTTP
+400, permission denied is code 102 delivered as HTTP 401, and bugs.gentoo.org
+never emits 403 at all.
+"""
+
+__all__ = (
+ "BugzillaAuthError",
+ "BugzillaAuthRequired",
+ "BugzillaConnectionError",
+ "BugzillaError",
+ "BugzillaInvalidField",
+ "BugzillaNotFound",
+ "BugzillaPermissionDenied",
+ "BugzillaProtocolError",
+ "BugzillaResponseError",
+ "BugzillaSchemaError",
+ "BugzillaServerError",
+ "BugzillaUsageError",
+ "PackageListError",
+ "from_response",
+ "from_status",
+)
+
+import typing
+
+from ..exceptions import PkgcoreUserException
+
+
+class BugzillaError(PkgcoreUserException):
+ """Base for every failure raised by pkgcore.bugzilla"""
+
+
+class BugzillaUsageError(BugzillaError, ValueError):
+ """The request couldn't be built, raised locally without any network use"""
+
+
+class BugzillaConnectionError(BugzillaError):
+ """DNS, TCP, TLS or timeout failure, after any retries were exhausted"""
+
+
+class BugzillaProtocolError(BugzillaError):
+ """The server answered with something that isn't a Bugzilla REST reply"""
+
+
+class BugzillaSchemaError(BugzillaProtocolError):
+ """Valid JSON, but not the shape the endpoint documents"""
+
+
+class BugzillaResponseError(BugzillaError):
+ """Bugzilla returned an error body"""
+
+ def __init__(
+ self, message: str, *, code: int = 0, status: int = 0, url: str = ""
+ ) -> None:
+ super().__init__(message)
+ self.message = message
+ self.code = code
+ self.status = status
+ self.url = url
+
+ def __str__(self) -> str:
+ details = ", ".join(
+ f"{name}={value}"
+ for name, value in (("code", self.code), ("http", self.status))
+ if value
+ )
+ return f"{self.message} [{details}]" if details else self.message
+
+
+class BugzillaAuthError(BugzillaResponseError):
+ """Authentication failed, or the request requires being logged in"""
+
+
+class BugzillaAuthRequired(BugzillaAuthError):
+ """A write was attempted without an api key, raised without a round trip"""
+
+
+class BugzillaNotFound(BugzillaResponseError):
+ """The requested bug, comment or REST route doesn't exist"""
+
+
+class BugzillaPermissionDenied(BugzillaResponseError):
+ """The account is known but isn't allowed to see or change this"""
+
+
+class BugzillaInvalidField(BugzillaResponseError):
+ """A field name or value was rejected"""
+
+
+class BugzillaServerError(BugzillaResponseError):
+ """A 5xx from the web tier in front of Bugzilla"""
+
+ def __init__(
+ self, message: str, *, retry_after: float | None = None, **kwargs: typing.Any
+ ) -> None:
+ super().__init__(message, **kwargs)
+ self.retry_after = retry_after
+
+
+class PackageListError(BugzillaError):
+ """Malformed cf_stabilisation_atoms content"""
+
+ def __init__(
+ self,
+ message: str,
+ *,
+ bug_id: int | None = None,
+ lineno: int | None = None,
+ line: str = "",
+ ) -> None:
+ super().__init__(message)
+ self.message = message
+ self.bug_id = bug_id
+ self.lineno = lineno
+ self.line = line
+
+ def __str__(self) -> str:
+ where = ""
+ if self.bug_id is not None:
+ where += f"bug {self.bug_id}"
+ if self.lineno is not None:
+ where += f"{', ' if where else ''}line {self.lineno}"
+ return f"{where}: {self.message}" if where else self.message
+
+
+# codes per Bugzilla/WebService/Constants.pm, limited to what bgo emits
+_CODE_MAP: typing.Final[dict[int, type[BugzillaResponseError]]] = {
+ **dict.fromkeys(
+ (50, 53, 100, 104, 108, 111, 114, 115, *range(129, 135), *range(1101, 1106)),
+ BugzillaInvalidField,
+ ),
+ **dict.fromkeys((51, 101, 32614), BugzillaNotFound),
+ **dict.fromkeys((102, 106, 109, 110, 113, 120), BugzillaPermissionDenied),
+ **dict.fromkeys((*range(300, 308), 410, 504), BugzillaAuthError),
+}
+
+
+def from_response(
+ error: typing.Mapping[str, typing.Any],
+ status: int,
+ url: str,
+ retry_after: float | None = None,
+) -> BugzillaResponseError:
+ """Build the most specific exception for a Bugzilla error body"""
+ code = error.get("code")
+ code = code if isinstance(code, int) else 0
+ message = str(error.get("message") or "unknown Bugzilla error")
+ if (kls := _CODE_MAP.get(code)) is None:
+ kls = BugzillaServerError if status >= 500 else BugzillaResponseError
+ if kls is BugzillaServerError:
+ return BugzillaServerError(
+ message, code=code, status=status, url=url, retry_after=retry_after
+ )
+ return kls(message, code=code, status=status, url=url)
+
+
+def from_status(
+ status: int, url: str, payload: bytes = b"", retry_after: float | None = None
+) -> BugzillaResponseError:
+ """Build an exception for a failing response that carried no error body"""
+ message = f"HTTP {status} from {url}"
+ if snippet := payload[:200].decode("utf-8", "replace").strip():
+ message = f"{message}: {snippet}"
+ if status >= 500:
+ return BugzillaServerError(
+ message, status=status, url=url, retry_after=retry_after
+ )
+ if status in (401, 403):
+ return BugzillaAuthError(message, status=status, url=url)
+ if status == 404:
+ return BugzillaNotFound(message, status=status, url=url)
+ return BugzillaResponseError(message, status=status, url=url)
diff --git a/src/pkgcore/bugzilla/pkglist.py b/src/pkgcore/bugzilla/pkglist.py
new file mode 100644
index 000000000..e64a00baa
--- /dev/null
+++ b/src/pkgcore/bugzilla/pkglist.py
@@ -0,0 +1,201 @@
+"""Parsing and rewriting of the ``cf_stabilisation_atoms`` bug field.
+
+The field is a newline separated list of ``<package spec> [keyword...]`` lines,
+with ``#`` comments and three keyword sentinels: ``*`` expands to the keywords
+suggested for that package, ``^`` repeats the previous line's keywords, and
+``-`` marks a line as deliberately having none.
+
+:class:`PackageList` keeps the original text and rewrites only the lines it has
+to, so indentation and comments on untouched lines survive a round trip.
+"""
+
+__all__ = (
+ "ALL_KEYWORDS",
+ "NO_KEYWORDS",
+ "SAME_KEYWORDS",
+ "PackageList",
+ "PackageListEntry",
+ "parse_atom",
+)
+
+import dataclasses
+import re
+import typing
+
+from snakeoil.klass import immutable
+from snakeoil.klass.properties import jit_attr_none
+
+from ..ebuild.atom import atom
+from ..ebuild.errors import MalformedAtom
+from .errors import PackageListError
+from .wire import BugId
+
+ALL_KEYWORDS: typing.Final = "*"
+SAME_KEYWORDS: typing.Final = "^"
+NO_KEYWORDS: typing.Final = "-"
+
+_COMMENT_RE: typing.Final = re.compile(r"(?:^|\s)#")
+
+
+def parse_atom(token: str) -> atom:
+ """Parse a package list token into an atom.
+
+ Stabilization lines carry a bare ``cat/pkg-1.2.3`` rather than the
+ ``=cat/pkg-1.2.3`` an atom needs, so the versioned form is tried first.
+
+ :raises MalformedAtom: if the token isn't a usable package spec
+ """
+ for candidate in (f"={token}", token):
+ try:
+ pkg = atom(candidate)
+ except MalformedAtom:
+ continue
+ if pkg.blocks or pkg.use or pkg.slot_operator:
+ raise MalformedAtom(token, "blockers, use deps and slot operators")
+ return pkg
+ raise MalformedAtom(token)
+
+
[email protected](frozen=True, slots=True)
+class PackageListEntry:
+ """A single line of a package list"""
+
+ lineno: int
+ raw: str
+ pkg: atom | None
+ keywords: tuple[str, ...] = ()
+ comment: str = ""
+ eol: str = ""
+
+ @property
+ def is_blank(self) -> bool:
+ return self.pkg is None
+
+ def with_keywords(self, keywords: typing.Iterable[str]) -> "PackageListEntry":
+ """Return a copy with new keywords, keeping indentation and comment"""
+ if self.pkg is None:
+ return self
+ keywords = tuple(keywords)
+ indent = self.raw[: len(self.raw) - len(self.raw.lstrip())]
+ body = " ".join((str(self.pkg), *keywords))
+ tail = f" {self.comment}" if self.comment else ""
+ return dataclasses.replace(self, keywords=keywords, raw=f"{indent}{body}{tail}")
+
+
+class PackageList(immutable.Simple):
+ """A lazily parsed view of a bug's package list.
+
+ Parsing is deferred so fetching a bug with a malformed list never fails;
+ only code that looks at the list does.
+ """
+
+ __slots__ = ("_entries", "bug_id", "text")
+
+ def __init__(self, text: str = "", *, bug_id: BugId | None = None) -> None:
+ self.text = text
+ self.bug_id = bug_id
+
+ @classmethod
+ def build(
+ cls, entries: typing.Iterable[tuple[atom, typing.Iterable[str]]]
+ ) -> "PackageList":
+ """Construct a fresh list from atoms and their keywords"""
+ return cls(
+ "\n".join(
+ " ".join((str(pkg), *keywords)).rstrip() for pkg, keywords in entries
+ )
+ )
+
+ @jit_attr_none
+ def entries(self) -> tuple[PackageListEntry, ...]:
+ """Every line of the list, blanks and comments included"""
+ return tuple(self._parse())
+
+ def _parse(self) -> typing.Iterator[PackageListEntry]:
+ for lineno, line in enumerate(self.text.splitlines(keepends=True), start=1):
+ raw = line.rstrip("\r\n")
+ eol = line[len(raw) :]
+ body = raw
+ comment = ""
+ if match := _COMMENT_RE.search(body):
+ comment, body = body[match.end() - 1 :], body[: match.start()]
+ if not (tokens := body.split()):
+ yield PackageListEntry(lineno, raw, None, comment=comment, eol=eol)
+ continue
+ try:
+ pkg = parse_atom(tokens[0])
+ except MalformedAtom as exc:
+ raise PackageListError(
+ str(exc), bug_id=self.bug_id, lineno=lineno, line=raw
+ ) from exc
+ yield PackageListEntry(lineno, raw, pkg, tuple(tokens[1:]), comment, eol)
+
+ @property
+ def atoms(self) -> tuple[atom, ...]:
+ return tuple(x.pkg for x in self.entries if x.pkg is not None)
+
+ def keywords_for(self, pkg: atom) -> tuple[str, ...]:
+ """Keywords requested for an atom, as written"""
+ for entry in self.entries:
+ if entry.pkg == pkg:
+ return entry.keywords
+ return ()
+
+ def expand(
+ self, suggest: typing.Callable[[atom], typing.Sequence[str]]
+ ) -> "PackageList":
+ """Resolve the ``*`` and ``^`` sentinels.
+
+ ``suggest`` returns the keywords a package should be requested for, in
+ the order they should be written; returning nothing collapses the line
+ to ``-``.
+
+ :raises PackageListError: on ``^`` with nothing above it to copy
+ """
+ expanded: list[PackageListEntry] = []
+ previous: tuple[str, ...] | None = None
+ changed = False
+ for entry in self.entries:
+ if entry.pkg is None:
+ expanded.append(entry)
+ continue
+ keywords: list[str] = []
+ for keyword in entry.keywords:
+ if keyword == ALL_KEYWORDS:
+ keywords.extend(suggest(entry.pkg) or (NO_KEYWORDS,))
+ elif keyword == SAME_KEYWORDS:
+ if previous is None:
+ raise PackageListError(
+ f"{SAME_KEYWORDS!r} keyword with no line above it",
+ bug_id=self.bug_id,
+ lineno=entry.lineno,
+ line=entry.raw,
+ )
+ keywords.extend(previous)
+ else:
+ keywords.append(keyword)
+ previous = tuple(keywords)
+ if previous != entry.keywords:
+ entry = entry.with_keywords(previous)
+ changed = True
+ expanded.append(entry)
+ if not changed:
+ return self
+ return PackageList("".join(x.raw + x.eol for x in expanded), bug_id=self.bug_id)
+
+ def __str__(self) -> str:
+ return self.text
+
+ def __bool__(self) -> bool:
+ return bool(self.text.strip())
+
+ def __eq__(self, other: object) -> bool:
+ if isinstance(other, PackageList):
+ return self.text == other.text
+ return NotImplemented
+
+ def __hash__(self) -> int:
+ return hash(self.text)
+
+ def __repr__(self) -> str:
+ return f"<{self.__class__.__name__} {self.text!r}>"
diff --git a/src/pkgcore/bugzilla/query.py b/src/pkgcore/bugzilla/query.py
new file mode 100644
index 000000000..4c81fb8bf
--- /dev/null
+++ b/src/pkgcore/bugzilla/query.py
@@ -0,0 +1,312 @@
+"""Composable searches against ``GET /rest/bug``.
+
+Constraints come in two flavours. Plain equality goes in :attr:`BugQuery.simple`
+as repeated parameters, which Bugzilla ORs within a key and ANDs across keys.
+Anything needing an operator becomes a :class:`Criterion` rendered into the
+``f<N>``/``o<N>``/``v<N>`` boolean chart parameters, with the slot numbers
+allocated at render time so any two queries can be combined.
+"""
+
+__all__ = (
+ "MAX_URL_LENGTH",
+ "BugQuery",
+ "ChartGroup",
+ "Criterion",
+)
+
+import dataclasses
+import functools
+import typing
+import urllib.parse
+
+from .enums import (
+ UNRESOLVED,
+ BugCategory,
+ ChartOp,
+ Component,
+ FlagStatus,
+ Join,
+ Product,
+ Status,
+)
+from .errors import BugzillaUsageError
+from .wire import BugId
+
+# apache's default LimitRequestLine is 8190 for the whole request line, and
+# bgo's own frontend is stricter in practice
+MAX_URL_LENGTH: typing.Final = 6000
+
+
[email protected](frozen=True, slots=True)
+class Criterion:
+ """One boolean chart condition"""
+
+ field: str
+ op: ChartOp
+ values: tuple[str, ...] = ()
+ negate: bool = False
+ splittable: bool = False
+
+ def render(self, slot: int) -> list[tuple[str, str]]:
+ params = [(f"f{slot}", self.field), (f"o{slot}", str(self.op))]
+ params.extend((f"v{slot}", value) for value in self.values)
+ if self.negate:
+ params.append((f"n{slot}", "1"))
+ return params
+
+ def with_values(self, values: typing.Iterable[str]) -> "Criterion":
+ return dataclasses.replace(self, values=tuple(values))
+
+
[email protected](frozen=True, slots=True)
+class ChartGroup:
+ """Several conditions combined with an explicit join"""
+
+ join: Join
+ children: tuple["Criterion | ChartGroup", ...]
+
+ def render(self, slot: int) -> tuple[list[tuple[str, str]], int]:
+ params = [(f"f{slot}", "OP"), (f"j{slot}", str(self.join))]
+ slot += 1
+ for child in self.children:
+ rendered, slot = _render(child, slot)
+ params.extend(rendered)
+ params.append((f"f{slot}", "CP"))
+ return params, slot + 1
+
+
+def _render(
+ chart: Criterion | ChartGroup, slot: int
+) -> tuple[list[tuple[str, str]], int]:
+ if isinstance(chart, ChartGroup):
+ return chart.render(slot)
+ return chart.render(slot), slot + 1
+
+
+# the field name, its values, and how to rebuild the query around a subset
+type _SplitAxis = tuple[
+ str, tuple[str, ...], typing.Callable[[typing.Sequence[str]], "BugQuery"]
+]
+
+
+def _merge_simple(
+ left: tuple[tuple[str, tuple[str, ...]], ...],
+ right: tuple[tuple[str, tuple[str, ...]], ...],
+) -> tuple[tuple[str, tuple[str, ...]], ...]:
+ merged: dict[str, tuple[str, ...]] = dict(left)
+ for key, values in right:
+ existing = merged.get(key, ())
+ merged[key] = existing + tuple(x for x in values if x not in existing)
+ return tuple(merged.items())
+
+
[email protected](frozen=True, slots=True)
+class BugQuery:
+ """A search, built from named constructors and combined with ``&``"""
+
+ simple: tuple[tuple[str, tuple[str, ...]], ...] = ()
+ charts: tuple[Criterion | ChartGroup, ...] = ()
+ limit: int | None = None
+ offset: int | None = None
+ order: str | None = None
+
+ @classmethod
+ def ids(cls, bugs: typing.Iterable[BugId | int]) -> "BugQuery":
+ return cls(simple=(("id", tuple(str(x) for x in bugs)),))
+
+ @classmethod
+ def product(cls, *products: Product | str) -> "BugQuery":
+ return cls(simple=(("product", tuple(str(x) for x in products)),))
+
+ @classmethod
+ def component(cls, *components: Component | str) -> "BugQuery":
+ return cls(simple=(("component", tuple(str(x) for x in components)),))
+
+ @classmethod
+ def category(cls, *categories: BugCategory) -> "BugQuery":
+ """Restrict to keywordreqs and/or stablereqs"""
+ return cls(
+ simple=(
+ ("product", (str(Product.GENTOO_LINUX),)),
+ ("component", tuple(str(x.component) for x in categories)),
+ )
+ )
+
+ @classmethod
+ def unresolved(cls) -> "BugQuery":
+ """Match open bugs, i.e. those with no resolution set.
+
+ Selecting on the open statuses instead returns the same set, since
+ Bugzilla only leaves the resolution empty while the status is open, so
+ this is the one spelling worth having.
+ """
+ return cls(simple=(("resolution", (UNRESOLVED,)),))
+
+ @classmethod
+ def resolution(cls, *resolutions: str) -> "BugQuery":
+ return cls(simple=(("resolution", tuple(map(str, resolutions))),))
+
+ @classmethod
+ def status(cls, *statuses: Status | str) -> "BugQuery":
+ return cls(simple=(("bug_status", tuple(map(str, statuses))),))
+
+ @classmethod
+ def cc(cls, *emails: str) -> "BugQuery":
+ return cls(simple=(("cc", tuple(emails)),))
+
+ @classmethod
+ def assigned_to(cls, *emails: str) -> "BugQuery":
+ return cls(simple=(("assigned_to", tuple(emails)),))
+
+ @classmethod
+ def keywords(cls, *keywords: str) -> "BugQuery":
+ return cls(charts=(Criterion("keywords", ChartOp.ANY_WORDS, tuple(keywords)),))
+
+ @classmethod
+ def flag(cls, name: str, *statuses: FlagStatus | str) -> "BugQuery":
+ """Match a flag by status.
+
+ Bugzilla can't express "flag is absent"; those bugs have to be fetched
+ and filtered client side.
+ """
+ return cls(
+ charts=(
+ Criterion(
+ "flagtypes.name",
+ ChartOp.ANY_WORDS,
+ tuple(f"{name}{status}" for status in statuses),
+ ),
+ )
+ )
+
+ @classmethod
+ def without_tags(cls, *tags: str) -> "BugQuery":
+ """Exclude bugs carrying any of these personal tags"""
+ return cls(charts=(Criterion("tag", ChartOp.NO_WORDS_SUBSTR, tuple(tags)),))
+
+ @classmethod
+ def package_list_any(cls, packages: typing.Iterable[object]) -> "BugQuery":
+ """Match bugs whose package list mentions any of these packages"""
+ return cls(
+ charts=(
+ Criterion(
+ "cf_stabilisation_atoms",
+ ChartOp.ANY_WORDS,
+ tuple(str(x) for x in packages),
+ splittable=True,
+ ),
+ )
+ )
+
+ @classmethod
+ def any_of(cls, *queries: "BugQuery") -> "BugQuery":
+ """OR several chart-only queries together.
+
+ Simple parameters can't take part in a chart group, so a query holding
+ any is rejected rather than silently ANDed in.
+ """
+ charts: list[Criterion | ChartGroup] = []
+ for query in queries:
+ if query.simple:
+ raise BugzillaUsageError(
+ "any_of() only accepts chart based queries, got "
+ f"{[key for key, _ in query.simple]}"
+ )
+ charts.extend(query.charts)
+ return cls(charts=(ChartGroup(Join.OR, tuple(charts)),))
+
+ def __and__(self, other: "BugQuery") -> "BugQuery":
+ return BugQuery(
+ simple=_merge_simple(self.simple, other.simple),
+ charts=self.charts + other.charts,
+ limit=self.limit if other.limit is None else other.limit,
+ offset=self.offset if other.offset is None else other.offset,
+ order=other.order or self.order,
+ )
+
+ def paged(self, limit: int, offset: int = 0) -> "BugQuery":
+ """Return a copy with explicit paging.
+
+ Bugzilla rejects an offset without a limit, and treats ``limit=0`` as
+ unlimited while silently discarding the offset, so both are refused.
+ """
+ if limit <= 0:
+ raise BugzillaUsageError(f"limit must be positive, got {limit}")
+ if offset < 0:
+ raise BugzillaUsageError(f"offset must not be negative, got {offset}")
+ return dataclasses.replace(self, limit=limit, offset=offset)
+
+ def params(self) -> list[tuple[str, str]]:
+ """Render to ordered query parameters.
+
+ Ordered pairs rather than a mapping, since chart slots are positional
+ and duplicate keys are meaningful.
+ """
+ params: list[tuple[str, str]] = []
+ for key, values in self.simple:
+ params.extend((key, value) for value in values)
+ slot = 1
+ for chart in self.charts:
+ rendered, slot = _render(chart, slot)
+ params.extend(rendered)
+ if self.limit is not None:
+ params.append(("limit", str(self.limit)))
+ if self.offset:
+ params.append(("offset", str(self.offset)))
+ if self.order is not None:
+ params.append(("order", self.order))
+ return params
+
+ def batches(
+ self, base_length: int = 0, max_length: int = MAX_URL_LENGTH
+ ) -> typing.Iterator["BugQuery"]:
+ """Split into sub-queries whose encoded parameters each fit the budget.
+
+ Only the largest splittable axis is divided, either the ``id`` simple
+ parameter or a :class:`Criterion` marked splittable; everything else is
+ repeated in every batch. Sizing uses the encoded length rather than a
+ count, so it adapts to long atoms instead of guessing.
+ """
+ if (axis := self._split_axis()) is None:
+ yield self
+ return
+ key, values, rebuild = axis
+ empty = rebuild(())
+ budget = max_length - base_length - len(urllib.parse.urlencode(empty.params()))
+ batch: list[str] = []
+ used = 0
+ for value in values:
+ cost = len(urllib.parse.urlencode(((key, value),))) + 1
+ if batch and used + cost > budget:
+ yield rebuild(batch)
+ batch, used = [], 0
+ batch.append(value)
+ used += cost
+ yield rebuild(batch)
+
+ def _split_axis(self) -> "_SplitAxis | None":
+ """Pick the widest axis to spread across batches, if there is one"""
+ candidates: list[_SplitAxis] = [
+ (key, values, functools.partial(self._rebuild_simple, key))
+ for key, values in self.simple
+ if key == "id"
+ ]
+ candidates.extend(
+ (chart.field, chart.values, functools.partial(self._rebuild_chart, index))
+ for index, chart in enumerate(self.charts)
+ if isinstance(chart, Criterion) and chart.splittable
+ )
+ if not candidates:
+ return None
+ return max(candidates, key=lambda axis: len("".join(axis[1])))
+
+ def _rebuild_simple(self, key: str, values: typing.Sequence[str]) -> "BugQuery":
+ simple = tuple(x for x in self.simple if x[0] != key)
+ if values:
+ simple += ((key, tuple(values)),)
+ return dataclasses.replace(self, simple=simple)
+
+ def _rebuild_chart(self, index: int, values: typing.Sequence[str]) -> "BugQuery":
+ charts = list(self.charts)
+ charts[index] = typing.cast(Criterion, charts[index]).with_values(values)
+ return dataclasses.replace(self, charts=tuple(charts))
diff --git a/src/pkgcore/bugzilla/testing.py b/src/pkgcore/bugzilla/testing.py
new file mode 100644
index 000000000..b01a4971e
--- /dev/null
+++ b/src/pkgcore/bugzilla/testing.py
@@ -0,0 +1,271 @@
+"""Replay helpers for testing code that talks to Bugzilla.
+
+Subclassing the real urllib handler rather than patching ``urlopen`` keeps the
+genuine :class:`urllib.request.Request` in the loop, so header, method, body
+and encoding mistakes are caught, and an unexpected request fails loudly
+instead of reaching the network.
+
+Nothing here imports pytest, so it is usable from a plain ``unittest`` suite or
+a script; :mod:`pkgcore.pytest.plugin` wraps it in a ``bugzilla_cassette`` fixture that
+downstream projects get for free.
+
+A cassette used as a context manager takes over the opener every client builds,
+which is what lets it intercept a client constructed somewhere the test can't
+reach, such as inside a CLI command::
+
+ with Cassette().expect_created(12345) as cassette:
+ main(["pkgdev", "bugs", "..."])
+ assert cassette.calls[0].body["summary"] == "cat/pkg-1: stablereq"
+
+When the client is reachable, skip the patching and hand it the opener::
+
+ cassette = Cassette().expect_bugs({"id": 1, ...})
+ bugs = cassette.client().search()
+"""
+
+__all__ = ("Call", "Cassette", "Recording", "ReplayHandler", "response")
+
+import contextlib
+import dataclasses
+import email.message
+import io
+import itertools
+import json
+import types
+import typing
+import urllib.parse
+import urllib.request
+import urllib.response
+
+from . import transport
+from .client import Bugzilla
+
+API_KEY: typing.Final = "fake-api-key-for-tests"
+BASE_URL: typing.Final = "https://bugs.example.org"
+
+type _Body = typing.Any | typing.Callable[["Call"], typing.Any]
+
+
[email protected](frozen=True, slots=True)
+class Call:
+ """A request the code under test actually made"""
+
+ method: str
+ url: str
+ headers: dict[str, str]
+ body: dict[str, typing.Any] | None
+
+ @property
+ def path(self) -> str:
+ return urllib.parse.urlsplit(self.url).path
+
+ @property
+ def query(self) -> list[tuple[str, str]]:
+ """Query parameters in order, duplicates preserved"""
+ return urllib.parse.parse_qsl(
+ urllib.parse.urlsplit(self.url).query, keep_blank_values=True
+ )
+
+ @property
+ def params(self) -> dict[str, list[str]]:
+ """Query parameters grouped by name"""
+ grouped: dict[str, list[str]] = {}
+ for key, value in self.query:
+ grouped.setdefault(key, []).append(value)
+ return grouped
+
+ def header(self, name: str) -> str | None:
+ """Look a header up without caring how urllib cased it"""
+ lowered = name.lower()
+ return next((v for k, v in self.headers.items() if k.lower() == lowered), None)
+
+
[email protected](frozen=True, slots=True)
+class Recording:
+ """A canned response.
+
+ ``body`` is serialized to JSON, or may be a callable taking the
+ :class:`Call` for responses that vary per request. ``raw`` overrides it
+ with exact bytes, for testing non-JSON replies.
+ """
+
+ body: _Body = None
+ status: int = 200
+ raw: bytes | None = None
+ headers: dict[str, str] = dataclasses.field(default_factory=dict)
+ content_type: str = "application/json; charset=UTF-8"
+ reason: str = "OK"
+
+ def payload(self, call: Call) -> bytes:
+ if self.raw is not None:
+ return self.raw
+ body = self.body(call) if callable(self.body) else self.body
+ return json.dumps(body).encode()
+
+
+def response(body: _Body = None, **kwargs: typing.Any) -> Recording:
+ """Shorthand for :class:`Recording`"""
+ return Recording(body, **kwargs)
+
+
+class ReplayHandler(urllib.request.HTTPHandler, urllib.request.HTTPSHandler):
+ """Serve queued recordings in order, recording what was asked for.
+
+ Subclasses both handlers so :func:`urllib.request.build_opener` drops its
+ defaults for either scheme; inheriting only the https one leaves plain http
+ going to the network.
+ """
+
+ def __init__(self, cassette: "Cassette") -> None:
+ super().__init__()
+ self.cassette = cassette
+
+ def https_open(self, req: typing.Any) -> typing.Any:
+ call = Call(
+ req.get_method(),
+ req.full_url,
+ dict(req.headers),
+ json.loads(req.data.decode()) if req.data else None,
+ )
+ recording = self.cassette._consume(call)
+
+ headers = email.message.Message()
+ headers["Content-Type"] = recording.content_type
+ for key, value in recording.headers.items():
+ headers[key] = value
+ result = urllib.response.addinfourl(
+ io.BytesIO(recording.payload(call)),
+ headers,
+ req.full_url,
+ recording.status,
+ )
+ # HTTPErrorProcessor reads .msg when turning a 4xx/5xx into an error
+ result.msg = recording.reason # type: ignore[attr-defined]
+ return result
+
+ http_open = https_open
+
+
+class Cassette:
+ """Queued responses, the requests they answered, and a client to drive.
+
+ Recordings queued with :meth:`expect` are consumed in order; once they run
+ out an :meth:`always` fallback answers, or an unexpected request fails.
+ """
+
+ def __init__(
+ self,
+ *recordings: Recording,
+ api_key: str | None = API_KEY,
+ base_url: str = BASE_URL,
+ ) -> None:
+ self.api_key = api_key
+ self.base_url = base_url
+ self.calls: list[Call] = []
+ self.pending: list[Recording] = list(recordings)
+ self.fallback: Recording | None = None
+ self.opener = urllib.request.build_opener(ReplayHandler(self))
+
+ def _consume(self, call: Call) -> Recording:
+ self.calls.append(call)
+ if self.pending:
+ return self.pending.pop(0)
+ if self.fallback is not None:
+ return self.fallback
+ raise AssertionError(f"unexpected request: {call.method} {call.url}")
+
+ def expect(self, *recordings: Recording) -> "Cassette":
+ """Queue responses, returning self so calls can be chained"""
+ self.pending.extend(recordings)
+ return self
+
+ def always(self, recording: Recording) -> "Cassette":
+ """Answer any request the queue doesn't cover"""
+ self.fallback = recording
+ return self
+
+ def expect_bugs(self, *bugs: dict[str, typing.Any]) -> "Cassette":
+ """Queue a search result"""
+ return self.expect(response({"bugs": list(bugs)}))
+
+ def expect_created(self, *bug_ids: int) -> "Cassette":
+ """Queue replies to bug creation"""
+ return self.expect(*(response({"id": bug_id}) for bug_id in bug_ids))
+
+ def creates_bugs(self, first: int = 1) -> "Cassette":
+ """Answer every creation with the next id, for unbounded filing"""
+ counter = itertools.count(first)
+ return self.always(response(lambda call: {"id": next(counter)}))
+
+ def expect_changed(self, bug_id: int, **changes: dict[str, str]) -> "Cassette":
+ """Queue a reply to an update"""
+ return self.expect(
+ response(
+ {
+ "bugs": [
+ {
+ "id": bug_id,
+ "alias": [],
+ "last_change_time": "2024-01-02T03:04:05Z",
+ "changes": changes,
+ }
+ ]
+ }
+ )
+ )
+
+ def expect_error(
+ self, code: int, message: str = "error", status: int = 400
+ ) -> "Cassette":
+ """Queue a Bugzilla error body"""
+ return self.expect(
+ response({"error": True, "code": code, "message": message}, status=status)
+ )
+
+ def expect_whoami(
+ self, name: str = "[email protected]", real_name: str = "A Dev", id: int = 7
+ ) -> "Cassette":
+ return self.expect(response({"id": id, "name": name, "real_name": real_name}))
+
+ def transport(self, **kwargs: typing.Any) -> transport.UrllibTransport:
+ """A transport wired to this cassette"""
+ kwargs.setdefault("api_key", self.api_key)
+ kwargs.setdefault("retries", 1)
+ return transport.UrllibTransport(self.base_url, opener=self.opener, **kwargs)
+
+ def client(self, **kwargs: typing.Any) -> Bugzilla:
+ """A client wired to this cassette"""
+ kwargs.setdefault("api_key", self.api_key)
+ kwargs.setdefault("retries", 1)
+ api_key = kwargs.pop("api_key")
+ return Bugzilla(api_key, base_url=self.base_url, opener=self.opener, **kwargs)
+
+ @contextlib.contextmanager
+ def installed(self) -> typing.Generator[typing.Self, None, None]:
+ """Make every client built while active use this cassette"""
+ original = transport.build_opener
+ transport.build_opener = lambda: self.opener
+ try:
+ yield self
+ finally:
+ transport.build_opener = original
+
+ def __enter__(self) -> typing.Self:
+ self._installed = self.installed()
+ return self._installed.__enter__()
+
+ def __exit__(
+ self,
+ kls: type[BaseException] | None,
+ exc: BaseException | None,
+ traceback: types.TracebackType | None,
+ ) -> None:
+ self._installed.__exit__(kls, exc, traceback)
+
+ def assert_drained(self) -> None:
+ """Fail if queued responses went unused"""
+ if self.pending:
+ raise AssertionError(
+ f"{len(self.pending)} unused recordings after "
+ f"{len(self.calls)} requests"
+ )
diff --git a/src/pkgcore/bugzilla/transport.py b/src/pkgcore/bugzilla/transport.py
new file mode 100644
index 000000000..d8bb0412b
--- /dev/null
+++ b/src/pkgcore/bugzilla/transport.py
@@ -0,0 +1,263 @@
+"""HTTP plumbing for the Bugzilla REST API, on top of stdlib urllib."""
+
+__all__ = (
+ "USER_AGENT",
+ "AuthMode",
+ "Transport",
+ "UrllibTransport",
+ "build_opener",
+ "build_user_agent",
+ "expect_list",
+ "expect_object",
+ "redact",
+)
+
+import enum
+import http.client
+import json
+import random
+import time
+import typing
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from .. import __version__
+from ..log import logger
+from . import errors
+from .wire import JSONValue, RequestBody
+
+USER_AGENT: typing.Final = f"pkgcore/{__version__}"
+
+
+def build_user_agent(client: str | None = None) -> str:
+ """Compose the User-Agent, most specific product first.
+
+ A caller's own token goes in front of pkgcore's rather than replacing it,
+ so Gentoo infra can tell which tool the traffic came from while pkgcore
+ stays identifiable.
+ """
+ return f"{client} {USER_AGENT}" if client else USER_AGENT
+
+
+_RETRY_STATUSES: typing.Final = frozenset((429, 500, 502, 503, 504))
+_IDEMPOTENT: typing.Final = frozenset(("GET", "HEAD"))
+_API_KEY_PARAM: typing.Final = "Bugzilla_api_key"
+
+
+class AuthMode(enum.StrEnum):
+ """Where the api key is placed.
+
+ ``QUERY`` is the only mode bugs.gentoo.org honours today; ``HEADER`` exists
+ for instances running Bugzilla 6.
+ """
+
+ QUERY = "query"
+ HEADER = "header"
+
+
+class Transport(typing.Protocol):
+ """The seam a client talks through"""
+
+ def request(
+ self,
+ method: str,
+ path: str,
+ *,
+ params: typing.Sequence[tuple[str, str]] = (),
+ body: RequestBody | None = None,
+ ) -> JSONValue: ...
+
+
+def redact(url: str) -> str:
+ """Strip the api key out of a url before it reaches a log or traceback"""
+ split = urllib.parse.urlsplit(url)
+ if not split.query:
+ return url
+ query = [
+ (key, "<redacted>" if key == _API_KEY_PARAM else value)
+ for key, value in urllib.parse.parse_qsl(split.query, keep_blank_values=True)
+ ]
+ return urllib.parse.urlunsplit(split._replace(query=urllib.parse.urlencode(query)))
+
+
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+ """Refuse redirects, which would leak the key to another host"""
+
+ def redirect_request(self, req: typing.Any, *args: typing.Any) -> None:
+ return None
+
+
+def build_opener() -> urllib.request.OpenerDirector:
+ """The opener used when a transport isn't given one.
+
+ Indirected through a function so that :mod:`pkgcore.bugzilla.testing` can
+ replace it, and intercept clients built somewhere it can't reach.
+ """
+ return urllib.request.build_opener(_NoRedirect())
+
+
+class UrllibTransport:
+ """A Bugzilla transport built on :mod:`urllib.request`.
+
+ Reads are retried with exponential backoff, since bugs.gentoo.org signals
+ overload by resetting the connection rather than returning 429. Writes are
+ not, because a retried bug creation files a duplicate.
+ """
+
+ __slots__ = (
+ "_api_key",
+ "_auth_mode",
+ "_base_url",
+ "_opener",
+ "_retries",
+ "_retry_writes",
+ "_timeout",
+ "_user_agent",
+ )
+
+ def __init__(
+ self,
+ base_url: str = "https://bugs.gentoo.org",
+ api_key: str | None = None,
+ *,
+ timeout: float = 30.0,
+ retries: int = 3,
+ auth_mode: AuthMode = AuthMode.QUERY,
+ retry_writes: bool = False,
+ user_agent: str | None = None,
+ opener: urllib.request.OpenerDirector | None = None,
+ ) -> None:
+ self._base_url = base_url.rstrip("/")
+ self._user_agent = build_user_agent(user_agent)
+ self._api_key = api_key
+ self._timeout = timeout
+ self._retries = max(1, retries)
+ self._auth_mode = auth_mode
+ self._retry_writes = retry_writes
+ self._opener = opener or build_opener()
+
+ @property
+ def authenticated(self) -> bool:
+ return self._api_key is not None
+
+ def request(
+ self,
+ method: str,
+ path: str,
+ *,
+ params: typing.Sequence[tuple[str, str]] = (),
+ body: RequestBody | None = None,
+ ) -> JSONValue:
+ if method not in _IDEMPOTENT and self._api_key is None:
+ raise errors.BugzillaAuthRequired(
+ f"{method} {path} needs an api key, this client is anonymous"
+ )
+ attempts = self._retries if method in _IDEMPOTENT or self._retry_writes else 1
+ for attempt in range(attempts):
+ try:
+ return self._attempt(method, path, params, body)
+ except (errors.BugzillaServerError, errors.BugzillaConnectionError) as exc:
+ if attempt + 1 >= attempts:
+ raise
+ delay = getattr(exc, "retry_after", None)
+ if delay is None:
+ delay = (2**attempt) * 0.5 * (1.0 + random.random())
+ logger.debug("retrying %s %s in %.1fs: %s", method, path, delay, exc)
+ time.sleep(delay)
+ raise AssertionError("unreachable") # pragma: no cover
+
+ def _attempt(
+ self,
+ method: str,
+ path: str,
+ params: typing.Sequence[tuple[str, str]],
+ body: RequestBody | None,
+ ) -> JSONValue:
+ query = list(params)
+ headers = {"Accept": "application/json", "User-Agent": self._user_agent}
+ if self._api_key is not None:
+ if self._auth_mode is AuthMode.HEADER:
+ headers["X-BUGZILLA-API-KEY"] = self._api_key
+ elif body is not None:
+ body = {_API_KEY_PARAM: self._api_key, **body}
+ else:
+ query.append((_API_KEY_PARAM, self._api_key))
+
+ url = f"{self._base_url}/rest/{path.lstrip('/')}"
+ if query:
+ url = f"{url}?{urllib.parse.urlencode(query)}"
+
+ data = None
+ if body is not None:
+ data = json.dumps(body, separators=(",", ":")).encode()
+ headers["Content-Type"] = "application/json"
+
+ request = urllib.request.Request(url, data=data, headers=headers, method=method)
+ try:
+ with self._opener.open(request, timeout=self._timeout) as response:
+ payload, status = response.read(), response.status
+ retry_after = response.headers.get("Retry-After")
+ except urllib.error.HTTPError as exc:
+ payload, status, retry_after = (
+ exc.read(),
+ exc.code,
+ exc.headers.get("Retry-After"),
+ )
+ except (urllib.error.URLError, TimeoutError, http.client.HTTPException) as exc:
+ reason = getattr(exc, "reason", exc)
+ raise errors.BugzillaConnectionError(
+ f"{method} {redact(url)}: {reason}"
+ ) from exc
+ return _decode(method, redact(url), status, payload, _retry_after(retry_after))
+
+
+def _retry_after(value: str | None) -> float | None:
+ try:
+ return float(value) if value else None
+ except ValueError:
+ return None
+
+
+def _decode(
+ method: str,
+ url: str,
+ status: int,
+ payload: bytes,
+ retry_after: float | None,
+) -> JSONValue:
+ """Turn a response into JSON, or the most specific exception available"""
+ try:
+ decoded = json.loads(payload.decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+ if status in _RETRY_STATUSES:
+ raise errors.BugzillaServerError(
+ f"{method} {url}: HTTP {status}", status=status, retry_after=retry_after
+ ) from exc
+ raise errors.BugzillaProtocolError(
+ f"{method} {url}: HTTP {status}, response isn't JSON: {payload[:200]!r}"
+ ) from exc
+
+ # bugzilla reports some failures with a 2xx status, so the body wins
+ if isinstance(decoded, dict) and decoded.get("error"):
+ raise errors.from_response(decoded, status, url, retry_after)
+ if status >= 400:
+ raise errors.from_status(status, url, payload, retry_after)
+ return typing.cast(JSONValue, decoded)
+
+
+def expect_object(payload: JSONValue, context: str) -> dict[str, typing.Any]:
+ if not isinstance(payload, dict):
+ raise errors.BugzillaSchemaError(
+ f"{context}: expected an object, got {type(payload).__name__}"
+ )
+ return payload
+
+
+def expect_list(payload: JSONValue, key: str, context: str) -> list[typing.Any]:
+ value = expect_object(payload, context).get(key)
+ if not isinstance(value, list):
+ raise errors.BugzillaSchemaError(
+ f"{context}: expected a {key!r} list, got {type(value).__name__}"
+ )
+ return value
diff --git a/src/pkgcore/bugzilla/wire.py b/src/pkgcore/bugzilla/wire.py
new file mode 100644
index 000000000..ea0e83da6
--- /dev/null
+++ b/src/pkgcore/bugzilla/wire.py
@@ -0,0 +1,230 @@
+"""Literal mirrors of the bugs.gentoo.org REST JSON payloads.
+
+Nothing here has behaviour; it exists so the rest of pkgcore.bugzilla can be
+checked against the actual wire format. Note that RawBugUpdate has no cc_add
+key and RawNewBug has no ids key, so the shapes that Bugzilla silently ignores
+can't be constructed.
+"""
+
+__all__ = (
+ "BugId",
+ "CommentId",
+ "FlagId",
+ "FlagStatusRead",
+ "FlagStatusWrite",
+ "FlagTypeId",
+ "JSONValue",
+ "RawBug",
+ "RawBugUpdate",
+ "RawChanges",
+ "RawComment",
+ "RawComments",
+ "RawCommentsResponse",
+ "RawCreateResponse",
+ "RawError",
+ "RawFieldChange",
+ "RawFlag",
+ "RawFlagChange",
+ "RawGetResponse",
+ "RawListChange",
+ "RawNewBug",
+ "RawNewComment",
+ "RawSearchResponse",
+ "RawUpdateResponse",
+ "RawWhoami",
+ "RequestBody",
+)
+
+import typing
+
+type JSONValue = (
+ None | bool | int | float | str | list[JSONValue] | dict[str, JSONValue]
+)
+
+# what a request body may be; a Mapping rather than a dict so the TypedDicts
+# below are accepted, which an invariant dict[str, JSONValue] would not be
+type RequestBody = typing.Mapping[str, typing.Any]
+
+BugId = typing.NewType("BugId", int)
+CommentId = typing.NewType("CommentId", int)
+FlagId = typing.NewType("FlagId", int)
+FlagTypeId = typing.NewType("FlagTypeId", int)
+
+FlagStatusRead = typing.Literal["?", "+", "-"]
+FlagStatusWrite = typing.Literal["?", "+", "-", "X"]
+
+
+class RawFlag(typing.TypedDict):
+ id: FlagId
+ name: str
+ status: FlagStatusRead
+ type_id: FlagTypeId
+ setter: str
+ creation_date: str
+ modification_date: str
+ requestee: typing.NotRequired[str]
+
+
+class RawBug(typing.TypedDict, total=False):
+ """A bug as returned by Bugzilla.
+
+ Every key is optional because Bugzilla omits whatever isn't named in
+ include_fields; presence is guaranteed by bug.INCLUDE_FIELDS instead.
+ """
+
+ id: BugId
+ summary: str
+ product: str
+ component: str
+ version: str
+ status: str
+ resolution: str
+ severity: str
+ priority: str
+ assigned_to: str
+ creator: str
+ cc: list[str]
+ keywords: list[str]
+ whiteboard: str
+ alias: list[str]
+ tags: list[str]
+ depends_on: list[BugId]
+ blocks: list[BugId]
+ see_also: list[str]
+ groups: list[str]
+ flags: list[RawFlag]
+ deadline: str | None
+ creation_time: str
+ last_change_time: str
+ cf_stabilisation_atoms: str
+ cf_runtime_testing_required: str
+
+
+class RawSearchResponse(typing.TypedDict):
+ bugs: list[RawBug]
+
+
+class RawGetResponse(typing.TypedDict):
+ """GET /rest/bug/{id}, which unlike a search also carries faults"""
+
+ bugs: list[RawBug]
+ faults: list[dict[str, typing.Any]]
+
+
+class RawComment(typing.TypedDict):
+ id: CommentId
+ bug_id: BugId
+ count: int
+ text: str
+ creator: str
+ time: str
+ creation_time: str
+ is_private: bool
+ tags: list[str]
+ attachment_id: typing.NotRequired[int | None]
+ raw_text: typing.NotRequired[str]
+
+
+class RawComments(typing.TypedDict):
+ comments: list[RawComment]
+
+
+class RawCommentsResponse(typing.TypedDict):
+ """Note that the keys of bugs are stringified bug ids"""
+
+ bugs: dict[str, RawComments]
+ comments: dict[str, RawComment]
+
+
+class RawCreateResponse(typing.TypedDict):
+ id: BugId
+
+
+class RawFieldChange(typing.TypedDict):
+ """A single field's delta; both values are comma-and-space joined strings"""
+
+ added: str
+ removed: str
+
+
+class RawChanges(typing.TypedDict):
+ id: BugId
+ alias: list[str]
+ last_change_time: str
+ changes: dict[str, RawFieldChange]
+
+
+class RawUpdateResponse(typing.TypedDict):
+ bugs: list[RawChanges]
+
+
+class RawWhoami(typing.TypedDict):
+ id: int
+ name: str
+ real_name: str
+
+
+class RawError(typing.TypedDict):
+ error: bool
+ code: int
+ message: str
+ documentation: typing.NotRequired[str]
+
+
+class RawListChange(typing.TypedDict, total=False):
+ add: list[str]
+ remove: list[str]
+ set: list[str]
+
+
+class RawFlagChange(typing.TypedDict, total=False):
+ name: str
+ status: FlagStatusWrite
+ id: FlagId
+ type_id: FlagTypeId
+ requestee: str
+ new: bool
+
+
+class RawNewComment(typing.TypedDict, total=False):
+ body: str
+ is_private: bool
+
+
+class RawBugUpdate(typing.TypedDict, total=False):
+ ids: list[BugId]
+ status: str
+ resolution: str
+ dupe_of: BugId
+ summary: str
+ assigned_to: str
+ whiteboard: str
+ deadline: str
+ cc: RawListChange
+ keywords: RawListChange
+ blocks: RawListChange
+ depends_on: RawListChange
+ see_also: RawListChange
+ groups: RawListChange
+ flags: list[RawFlagChange]
+ comment: RawNewComment
+ cf_stabilisation_atoms: str
+ cf_runtime_testing_required: str
+
+
+class RawNewBug(typing.TypedDict, total=False):
+ product: str
+ component: str
+ summary: str
+ description: str
+ version: str
+ severity: str
+ assigned_to: str
+ cc: list[str]
+ keywords: list[str]
+ depends_on: list[BugId]
+ blocks: list[BugId]
+ see_also: list[str]
+ deadline: str
+ cf_stabilisation_atoms: str
+ cf_runtime_testing_required: str
diff --git a/src/pkgcore/pytest/plugin.py b/src/pkgcore/pytest/plugin.py
index 376bfb069..0b2252fd2 100644
--- a/src/pkgcore/pytest/plugin.py
+++ b/src/pkgcore/pytest/plugin.py
@@ -288,6 +288,21 @@ class EbuildRepo:
__dir__ = klass.DirProxy("_repo")
[email protected]
+def bugzilla_cassette():
+ """Intercept Bugzilla REST traffic with canned responses.
+
+ Active for the whole test, so a client built anywhere, including inside a
+ CLI command, is captured. Queue replies with ``expect``/``expect_bugs``,
+ then assert against ``cassette.calls``.
+ """
+ from ..bugzilla.testing import Cassette
+
+ with Cassette() as cassette:
+ yield cassette
+ cassette.assert_drained()
+
+
@pytest.fixture
def repo(tmp_path_factory):
"""Create a generic ebuild repository."""
diff --git a/tests/bugzilla/__init__.py b/tests/bugzilla/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/bugzilla/conftest.py b/tests/bugzilla/conftest.py
new file mode 100644
index 000000000..79b8b7d5c
--- /dev/null
+++ b/tests/bugzilla/conftest.py
@@ -0,0 +1,34 @@
+"""Thin adapters over the shipped replay helpers.
+
+The real machinery lives in :mod:`pkgcore.bugzilla.testing` so downstream
+projects can use it; the ``bugzilla_cassette`` fixture comes from pkgcore's
+pytest plugin. These fixtures only save the local tests a line or two.
+"""
+
+import pytest
+
+from pkgcore.bugzilla.testing import API_KEY, Recording
+
+
[email protected]
+def cassette(bugzilla_cassette):
+ """Queue recordings and hand back the cassette plus a transport"""
+
+ def build(*recordings: Recording, api_key: str | None = API_KEY, **kwargs):
+ bugzilla_cassette.api_key = api_key
+ bugzilla_cassette.expect(*recordings)
+ return bugzilla_cassette, bugzilla_cassette.transport(**kwargs)
+
+ return build
+
+
[email protected]
+def client(bugzilla_cassette):
+ """Queue recordings and hand back the cassette plus a client"""
+
+ def build(*recordings: Recording, api_key: str | None = API_KEY, **kwargs):
+ bugzilla_cassette.api_key = api_key
+ bugzilla_cassette.expect(*recordings)
+ return bugzilla_cassette, bugzilla_cassette.client(**kwargs)
+
+ return build
diff --git a/tests/bugzilla/test_apikey.py b/tests/bugzilla/test_apikey.py
new file mode 100644
index 000000000..28e1e9917
--- /dev/null
+++ b/tests/bugzilla/test_apikey.py
@@ -0,0 +1,145 @@
+import pytest
+from snakeoil.cli import arghparse
+
+from pkgcore.bugzilla.apikey import (
+ API_KEY_ENV,
+ BugzillaApiKey,
+ BugzillaClientArgs,
+ find_api_key,
+)
+from pkgcore.bugzilla.client import DEFAULT_URL, Bugzilla
+from pkgcore.bugzilla.errors import BugzillaUsageError
+
+
[email protected](autouse=True)
+def no_env(monkeypatch):
+ monkeypatch.delenv(API_KEY_ENV, raising=False)
+
+
+class TestFindApiKey:
+ def test_nothing_configured(self, tmp_path):
+ assert find_api_key(home=tmp_path) is None
+
+ def test_explicit_wins(self, tmp_path, monkeypatch):
+ monkeypatch.setenv(API_KEY_ENV, "from-env")
+ (tmp_path / ".bugz_token").write_text("from-file")
+ assert find_api_key("explicit", home=tmp_path) == "explicit"
+
+ def test_explicit_is_stripped(self, tmp_path):
+ assert find_api_key(" key ", home=tmp_path) == "key"
+
+ def test_blank_explicit_falls_through(self, tmp_path):
+ (tmp_path / ".bugz_token").write_text("from-file")
+ assert find_api_key(" ", home=tmp_path) == "from-file"
+
+ def test_env(self, tmp_path, monkeypatch):
+ monkeypatch.setenv(API_KEY_ENV, "from-env")
+ (tmp_path / ".bugz_token").write_text("from-file")
+ assert find_api_key(home=tmp_path) == "from-env"
+
+ def test_env_can_be_disabled(self, tmp_path, monkeypatch):
+ monkeypatch.setenv(API_KEY_ENV, "from-env")
+ assert find_api_key(allow_env=False, home=tmp_path) is None
+
+ def test_blank_env_falls_through(self, tmp_path, monkeypatch):
+ monkeypatch.setenv(API_KEY_ENV, " ")
+ (tmp_path / ".bugz_token").write_text("from-file")
+ assert find_api_key(home=tmp_path) == "from-file"
+
+ @pytest.mark.parametrize("section", ("default", "gentoo", "Gentoo"))
+ def test_bugzrc_sections(self, tmp_path, section):
+ (tmp_path / ".bugzrc").write_text(f"[{section}]\nkey = rc-key\n")
+ assert find_api_key(home=tmp_path) == "rc-key"
+
+ def test_bugzrc_beats_token_file(self, tmp_path):
+ (tmp_path / ".bugzrc").write_text("[default]\nkey = rc-key\n")
+ (tmp_path / ".bugz_token").write_text("token-key")
+ assert find_api_key(home=tmp_path) == "rc-key"
+
+ def test_bugzrc_without_a_key_falls_through(self, tmp_path):
+ (tmp_path / ".bugzrc").write_text("[default]\nuser = someone\n")
+ (tmp_path / ".bugz_token").write_text("token-key")
+ assert find_api_key(home=tmp_path) == "token-key"
+
+ def test_empty_bugzrc_value_is_not_a_key(self, tmp_path):
+ (tmp_path / ".bugzrc").write_text("[default]\nkey =\n")
+ (tmp_path / ".bugz_token").write_text("token-key")
+ assert find_api_key(home=tmp_path) == "token-key"
+
+ def test_unrelated_section_is_ignored(self, tmp_path):
+ (tmp_path / ".bugzrc").write_text("[somewhere-else]\nkey = nope\n")
+ assert find_api_key(home=tmp_path) is None
+
+ def test_malformed_bugzrc(self, tmp_path):
+ (tmp_path / ".bugzrc").write_text("this is not ini\n[unclosed\n")
+ with pytest.raises(BugzillaUsageError, match="failed parsing"):
+ find_api_key(home=tmp_path)
+
+ def test_token_file(self, tmp_path):
+ (tmp_path / ".bugz_token").write_text(" token-key\n")
+ assert find_api_key(home=tmp_path) == "token-key"
+
+ def test_empty_token_file(self, tmp_path):
+ (tmp_path / ".bugz_token").write_text("\n")
+ assert find_api_key(home=tmp_path) is None
+
+ def test_warns_on_world_readable_file(self, tmp_path, caplog):
+ token = tmp_path / ".bugz_token"
+ token.write_text("token-key")
+ token.chmod(0o644)
+ assert find_api_key(home=tmp_path) == "token-key"
+ assert "readable by others" in caplog.text
+
+ def test_no_warning_when_private(self, tmp_path, caplog):
+ token = tmp_path / ".bugz_token"
+ token.write_text("token-key")
+ token.chmod(0o600)
+ find_api_key(home=tmp_path)
+ assert "readable by others" not in caplog.text
+
+
+class TestBugzillaApiKey:
+ @pytest.fixture
+ def parser(self):
+ parser = arghparse.ArgumentParser(suppress=True)
+ BugzillaApiKey.mangle_argparser(parser)
+ return parser
+
+ def test_explicit_option(self, parser, tmp_path, monkeypatch):
+ monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
+ namespace = parser.parse_args(["--api-key", "cli-key"])
+ assert namespace.api_key == "cli-key"
+
+ def test_falls_back_to_discovery(self, parser, monkeypatch):
+ monkeypatch.setenv(API_KEY_ENV, "from-env")
+ assert parser.parse_args([]).api_key == "from-env"
+
+ def test_anonymous(self, parser, tmp_path, monkeypatch):
+ monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
+ assert parser.parse_args([]).api_key is None
+
+
+class TestBugzillaClientArgs:
+ @pytest.fixture
+ def parser(self):
+ parser = arghparse.ArgumentParser(suppress=True)
+ BugzillaClientArgs.mangle_argparser(parser)
+ return parser
+
+ def test_builds_a_client(self, parser, monkeypatch):
+ monkeypatch.setenv(API_KEY_ENV, "from-env")
+ namespace = parser.parse_args([])
+ assert isinstance(namespace.bugzilla, Bugzilla)
+ assert namespace.bugzilla.base_url == DEFAULT_URL
+
+ def test_custom_url(self, parser, monkeypatch):
+ monkeypatch.setenv(API_KEY_ENV, "from-env")
+ namespace = parser.parse_args(["--bugzilla-url", "https://bugs.example.org/"])
+ assert namespace.bugzilla.base_url == "https://bugs.example.org"
+
+ def test_api_key_is_resolved_first(self, parser, tmp_path, monkeypatch):
+ # the client's delayed default has a lower priority than the key's
+ monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
+ namespace = parser.parse_args(["--api-key", "cli-key"])
+ assert namespace.api_key == "cli-key"
+ assert isinstance(namespace.bugzilla, Bugzilla)
diff --git a/tests/bugzilla/test_bug.py b/tests/bugzilla/test_bug.py
new file mode 100644
index 000000000..4db9fa9fc
--- /dev/null
+++ b/tests/bugzilla/test_bug.py
@@ -0,0 +1,294 @@
+import dataclasses
+import datetime
+
+import pytest
+
+from pkgcore.bugzilla.bug import (
+ INCLUDE_FIELDS,
+ Bug,
+ Flag,
+ parse_bug,
+ parse_changes,
+ parse_comment,
+ parse_user,
+)
+from pkgcore.bugzilla.enums import BugCategory, FlagStatus, RuntimeTesting
+from pkgcore.bugzilla.errors import PackageListError
+from pkgcore.ebuild.atom import atom
+
+RAW_BUG = {
+ "id": 900001,
+ "summary": "dev-libs/a: stablereq",
+ "product": "Gentoo Linux",
+ "component": "Stabilization",
+ "version": "unspecified",
+ "status": "CONFIRMED",
+ "resolution": "",
+ "severity": "enhancement",
+ "priority": "Normal",
+ "assigned_to": "[email protected]",
+ "creator": "[email protected]",
+ "cc": ["[email protected]", "[email protected]", "[email protected]"],
+ "keywords": ["CC-ARCHES"],
+ "whiteboard": "",
+ "alias": [],
+ "tags": [],
+ "depends_on": [900000],
+ "blocks": [899999],
+ "see_also": [],
+ "groups": [],
+ "flags": [
+ {
+ "id": 51984,
+ "name": "sanity-check",
+ "status": "+",
+ "type_id": 6,
+ "setter": "[email protected]",
+ "creation_date": "2024-01-02T03:04:05Z",
+ "modification_date": "2024-01-02T03:04:05Z",
+ }
+ ],
+ "deadline": None,
+ "creation_time": "2024-01-01T00:00:00Z",
+ "last_change_time": "2024-01-02T03:04:05Z",
+ "cf_stabilisation_atoms": "=dev-libs/a-1.2 amd64 x86\r\n",
+ "cf_runtime_testing_required": "---",
+}
+
+
+class TestIncludeFields:
+ def test_matches_declared_fields(self):
+ declared = {
+ field.metadata["wire"]
+ for field in dataclasses.fields(Bug)
+ if "wire" in field.metadata
+ }
+ assert set(INCLUDE_FIELDS) == declared
+
+ def test_every_field_is_declared(self):
+ assert all("wire" in field.metadata for field in dataclasses.fields(Bug)), (
+ "a Bug field without a wire name is never populated"
+ )
+
+ def test_no_duplicates(self):
+ assert len(INCLUDE_FIELDS) == len(set(INCLUDE_FIELDS))
+
+
+class TestParseBug:
+ @pytest.fixture
+ def bug(self):
+ return parse_bug(RAW_BUG)
+
+ def test_scalars(self, bug):
+ assert bug.id == 900001
+ assert bug.summary == "dev-libs/a: stablereq"
+ assert bug.status == "CONFIRMED"
+ assert bug.severity == "enhancement"
+
+ def test_sequences_are_tuples(self, bug):
+ assert bug.cc == ("[email protected]", "[email protected]", "[email protected]")
+ assert bug.depends_on == (900000,)
+ assert bug.blocks == (899999,)
+ assert bug.keywords == ("CC-ARCHES",)
+
+ def test_timestamps(self, bug):
+ assert bug.last_change_time == datetime.datetime(
+ 2024, 1, 2, 3, 4, 5, tzinfo=datetime.UTC
+ )
+ assert bug.creation_time == datetime.datetime(2024, 1, 1, tzinfo=datetime.UTC)
+
+ def test_null_deadline(self, bug):
+ assert bug.deadline is None
+
+ def test_deadline(self):
+ bug = parse_bug(RAW_BUG | {"deadline": "2024-03-01"})
+ assert bug.deadline == datetime.date(2024, 3, 1)
+
+ def test_package_list(self, bug):
+ assert bug.package_list.atoms == (atom("=dev-libs/a-1.2"),)
+ assert str(bug.package_list) == "=dev-libs/a-1.2 amd64 x86\r\n"
+
+ def test_flags(self, bug):
+ assert bug.flags == (
+ Flag(
+ name="sanity-check",
+ status=FlagStatus.GRANTED,
+ id=51984,
+ type_id=6,
+ setter="[email protected]",
+ ),
+ )
+
+ def test_package_list_knows_its_bug(self, bug):
+ # so a malformed list says which bug it came from
+ assert bug.package_list.bug_id == 900001
+
+ def test_malformed_package_list_is_attributed(self):
+ bug = parse_bug(RAW_BUG | {"cf_stabilisation_atoms": "not an atom"})
+ with pytest.raises(PackageListError) as excinfo:
+ assert bug.package_list.atoms
+ assert excinfo.value.bug_id == 900001
+ assert "bug 900001" in str(excinfo.value)
+
+ def test_package_list_without_an_id(self):
+ raw = dict(RAW_BUG)
+ del raw["id"]
+ assert parse_bug(raw).package_list.bug_id is None
+
+ def test_absent_fields_fall_back_to_defaults(self):
+ bug = parse_bug({"id": 5})
+ assert bug.id == 5
+ assert bug.cc == ()
+ assert bug.package_list.atoms == ()
+ assert bug.runtime_testing_required is RuntimeTesting.UNSET
+ assert bug.deadline is None
+
+ def test_empty_response(self):
+ assert parse_bug({}) == Bug()
+
+ def test_immutable(self, bug):
+ with pytest.raises(dataclasses.FrozenInstanceError):
+ bug.summary = "nope"
+
+ def test_hashable(self, bug):
+ assert len({bug, parse_bug(RAW_BUG)}) == 1
+
+
+class TestDerived:
+ def test_category(self):
+ assert parse_bug(RAW_BUG).category is BugCategory.STABLEREQ
+ assert (
+ parse_bug(RAW_BUG | {"component": "Keywording"}).category
+ is BugCategory.KEYWORDREQ
+ )
+ assert parse_bug(RAW_BUG | {"component": "Eclasses"}).category is None
+ assert parse_bug(RAW_BUG | {"product": "Websites"}).category is None
+
+ def test_resolved(self):
+ assert not parse_bug(RAW_BUG).resolved
+ assert parse_bug(RAW_BUG | {"resolution": "FIXED"}).resolved
+ assert parse_bug(RAW_BUG | {"resolution": "OBSOLETE"}).resolved
+
+ def test_security(self):
+ assert not parse_bug(RAW_BUG).security
+ assert parse_bug(RAW_BUG | {"product": "Gentoo Security"}).security
+
+ @pytest.mark.parametrize(
+ ("status", "expected"),
+ (("+", True), ("-", False), ("?", None)),
+ )
+ def test_sanity_check(self, status, expected):
+ flags = [dict(RAW_BUG["flags"][0], status=status)]
+ assert parse_bug(RAW_BUG | {"flags": flags}).sanity_check is expected
+
+ def test_sanity_check_unset(self):
+ assert parse_bug(RAW_BUG | {"flags": []}).sanity_check is None
+
+ def test_unknown_flag(self):
+ assert parse_bug(RAW_BUG).flag("no-such-flag") is None
+
+ def test_url(self):
+ assert parse_bug(RAW_BUG).url == "https://bugs.gentoo.org/900001"
+
+ def test_arches(self):
+ known = frozenset(("amd64", "x86", "arm"))
+ assert parse_bug(RAW_BUG).arches(known) == ("amd64", "x86")
+
+ def test_arches_from_truncated_anonymous_cc(self):
+ bug = parse_bug(RAW_BUG | {"cc": ["amd64", "x86", "someone"]})
+ assert bug.arches(frozenset(("amd64", "x86"))) == ("amd64", "x86")
+
+ def test_arches_ignores_foreign_domains(self):
+ bug = parse_bug(RAW_BUG | {"cc": ["[email protected]"]})
+ assert bug.arches(frozenset(("amd64",))) == ()
+
+ def test_runtime_testing(self):
+ for value, expected in (
+ ("Yes", RuntimeTesting.YES),
+ ("no", RuntimeTesting.NO),
+ ("MANUAL", RuntimeTesting.MANUAL),
+ ("---", RuntimeTesting.UNSET),
+ ("bogus", RuntimeTesting.UNSET),
+ ):
+ bug = parse_bug(RAW_BUG | {"cf_runtime_testing_required": value})
+ assert bug.runtime_testing_required is expected
+
+
+class TestParseComment:
+ def test_parse(self):
+ comment = parse_comment(
+ {
+ "id": 503,
+ "bug_id": 900001,
+ "count": 0,
+ "text": "please stabilize",
+ "creator": "[email protected]",
+ "time": "2024-01-01T00:00:00Z",
+ "creation_time": "2024-01-01T00:00:00Z",
+ "is_private": False,
+ "tags": ["obsolete"],
+ }
+ )
+ assert comment.id == 503
+ assert comment.count == 0
+ assert comment.obsolete
+ assert comment.creation_time == datetime.datetime(
+ 2024, 1, 1, tzinfo=datetime.UTC
+ )
+
+ def test_not_obsolete(self):
+ comment = parse_comment(
+ {
+ "id": 1,
+ "bug_id": 2,
+ "count": 1,
+ "text": "hi",
+ "creator": "x",
+ "time": "2024-01-01T00:00:00Z",
+ "creation_time": "2024-01-01T00:00:00Z",
+ "is_private": False,
+ "tags": [],
+ }
+ )
+ assert not comment.obsolete
+
+
+class TestParseUser:
+ def test_parse(self):
+ user = parse_user({"id": 7, "name": "[email protected]", "real_name": "A Dev"})
+ assert (user.id, user.name, user.real_name) == (7, "[email protected]", "A Dev")
+
+
+class TestParseChanges:
+ def test_splits_comma_joined_values(self):
+ changes = parse_changes(
+ {
+ "id": 900001,
+ "alias": [],
+ "last_change_time": "2024-01-02T03:04:05Z",
+ "changes": {
+ "cc": {"added": "[email protected], [email protected]", "removed": ""},
+ "status": {"added": "RESOLVED", "removed": "CONFIRMED"},
+ },
+ }
+ )
+ assert changes.id == 900001
+ assert changes.changes["cc"].added == (
+ "[email protected]",
+ "[email protected]",
+ )
+ assert changes.changes["cc"].removed == ()
+ assert changes.changes["status"].added == ("RESOLVED",)
+ assert changes.changes["status"].removed == ("CONFIRMED",)
+ assert changes
+
+ def test_empty_changes_is_falsy(self):
+ changes = parse_changes(
+ {
+ "id": 1,
+ "alias": [],
+ "last_change_time": "2024-01-02T03:04:05Z",
+ "changes": {},
+ }
+ )
+ assert not changes
diff --git a/tests/bugzilla/test_changes.py b/tests/bugzilla/test_changes.py
new file mode 100644
index 000000000..183766638
--- /dev/null
+++ b/tests/bugzilla/test_changes.py
@@ -0,0 +1,412 @@
+import datetime
+
+import pytest
+
+from pkgcore.bugzilla.changes import (
+ MAX_COMMENT_LENGTH,
+ BugUpdate,
+ FlagChange,
+ ListChange,
+ NewBug,
+ NewComment,
+ summarise,
+)
+from pkgcore.bugzilla.enums import (
+ BugCategory,
+ Component,
+ FlagStatus,
+ Product,
+ Resolution,
+ RuntimeTesting,
+ Severity,
+ Status,
+)
+from pkgcore.bugzilla.errors import BugzillaUsageError
+from pkgcore.bugzilla.pkglist import PackageList
+
+
+class TestListChange:
+ def test_empty_is_falsy(self):
+ assert not ListChange()
+ assert ListChange().to_wire() == {}
+
+ def test_adding(self):
+ assert ListChange.adding("a", "b").to_wire() == {"add": ["a", "b"]}
+
+ def test_removing(self):
+ assert ListChange.removing("a").to_wire() == {"remove": ["a"]}
+
+ def test_setting(self):
+ assert ListChange.setting("a", "b").to_wire() == {"set": ["a", "b"]}
+
+ def test_setting_empty_still_serialises(self):
+ # an explicit "clear the field", distinct from leaving it alone
+ assert ListChange.setting().to_wire() == {"set": []}
+ assert ListChange.setting()
+
+ def test_add_and_remove(self):
+ assert ListChange(add=("a",), remove=("b",)).to_wire() == {
+ "add": ["a"],
+ "remove": ["b"],
+ }
+
+ def test_ints_are_stringified(self):
+ assert ListChange.adding(1, 2).to_wire() == {"add": ["1", "2"]}
+
+ def test_replace_conflicts_with_add(self):
+ with pytest.raises(BugzillaUsageError, match="replace cannot be combined"):
+ ListChange(add=("a",), replace=("b",))
+
+ def test_overlapping_add_and_remove(self):
+ with pytest.raises(BugzillaUsageError, match="both added and removed"):
+ ListChange(add=("a",), remove=("a",))
+
+ def test_merge(self):
+ merged = ListChange.adding("a") | ListChange(add=("b",), remove=("c",))
+ assert merged.to_wire() == {"add": ["a", "b"], "remove": ["c"]}
+
+ def test_merge_deduplicates(self):
+ assert (ListChange.adding("a") | ListChange.adding("a")).add == ("a",)
+
+ def test_merge_with_replace_wins(self):
+ merged = ListChange.adding("a") | ListChange.setting("z")
+ assert merged.to_wire() == {"set": ["z"]}
+
+ def test_frozen(self):
+ change = ListChange.adding("a")
+ with pytest.raises(AttributeError):
+ change.add = ()
+
+
+class TestNewComment:
+ def test_to_wire(self):
+ assert NewComment("hello").to_wire() == {"body": "hello"}
+
+ def test_private(self):
+ assert NewComment("hi", is_private=True).to_wire() == {
+ "body": "hi",
+ "is_private": True,
+ }
+
+ def test_rejects_overlong_body(self):
+ with pytest.raises(BugzillaUsageError, match="the limit is"):
+ NewComment("x" * (MAX_COMMENT_LENGTH + 1))
+
+ def test_truncated_leaves_short_bodies_alone(self):
+ assert NewComment.truncated("short").body == "short"
+
+ def test_truncated_cuts_on_a_line_boundary(self):
+ body = "\n".join(f"line {i}" for i in range(100))
+ comment = NewComment.truncated(body, limit=50)
+ assert len(comment.body) <= 50
+ assert comment.body.endswith("\n...\n")
+ assert "line 0" in comment.body
+
+ def test_truncated_without_newlines(self):
+ comment = NewComment.truncated("x" * 100, limit=20)
+ assert len(comment.body) <= 20
+ assert comment.body.endswith("\n...\n")
+
+
+class TestSummarise:
+ def test_single_package_keeps_the_version(self):
+ pkglist = PackageList("=dev-libs/a-1 amd64")
+ assert summarise(pkglist, BugCategory.STABLEREQ) == "dev-libs/a-1: stablereq"
+
+ def test_keywordreq_suffix(self):
+ pkglist = PackageList("dev-libs/a ~amd64")
+ assert summarise(pkglist, BugCategory.KEYWORDREQ) == "dev-libs/a: keywordreq"
+
+ def test_several_packages(self):
+ pkglist = PackageList("=dev-libs/a-1\n=dev-libs/b-2")
+ assert summarise(pkglist, BugCategory.STABLEREQ) == (
+ "dev-libs/a-1, dev-libs/b-2: stablereq"
+ )
+
+ def test_long_list_collapses(self):
+ pkglist = PackageList(
+ "\n".join(f"=dev-libs/averylongpackagename{i}-1" for i in range(10))
+ )
+ assert summarise(pkglist, BugCategory.STABLEREQ) == (
+ "dev-libs/averylongpackagename0-1 and friends: stablereq"
+ )
+
+ def test_single_long_package_is_not_collapsed(self):
+ pkglist = PackageList(f"=dev-libs/{'a' * 120}-1")
+ assert summarise(pkglist, BugCategory.STABLEREQ).startswith("dev-libs/aaa")
+
+ def test_empty_list(self):
+ with pytest.raises(BugzillaUsageError, match="empty package list"):
+ summarise(PackageList(), BugCategory.STABLEREQ)
+
+
+class TestNewBug:
+ def test_minimal(self):
+ bug = NewBug(
+ summary="a summary",
+ description="a description",
+ component=Component.CURRENT_PACKAGES,
+ )
+ assert bug.to_wire() == {
+ "product": "Gentoo Linux",
+ "component": "Current packages",
+ "summary": "a summary",
+ "description": "a description",
+ "version": "unspecified",
+ "severity": "normal",
+ }
+
+ def test_rejects_blank_summary(self):
+ with pytest.raises(BugzillaUsageError, match="needs a summary"):
+ NewBug(summary=" ", description="x", component=Component.ECLASSES)
+
+ def test_full(self):
+ bug = NewBug(
+ summary="s",
+ description="d",
+ component=Component.VULNERABILITIES,
+ product=Product.GENTOO_SECURITY,
+ severity=Severity.QA,
+ assigned_to="[email protected]",
+ cc=("[email protected]",),
+ keywords=("SECURITY",),
+ depends_on=(1,),
+ blocks=(2,),
+ see_also=("https://bugs.gentoo.org/3",),
+ deadline=datetime.date(2024, 3, 1),
+ package_list=PackageList("=dev-libs/a-1"),
+ runtime_testing_required=RuntimeTesting.MANUAL,
+ )
+ assert bug.to_wire() == {
+ "product": "Gentoo Security",
+ "component": "Vulnerabilities",
+ "summary": "s",
+ "description": "d",
+ "version": "unspecified",
+ "severity": "QA",
+ "assigned_to": "[email protected]",
+ "cc": ["[email protected]"],
+ "keywords": ["SECURITY"],
+ "depends_on": [1],
+ "blocks": [2],
+ "see_also": ["https://bugs.gentoo.org/3"],
+ "deadline": "2024-03-01",
+ "cf_stabilisation_atoms": "=dev-libs/a-1",
+ "cf_runtime_testing_required": "Manual",
+ }
+
+ def test_no_ids_key(self):
+ # ids belongs to updates; sending it on create would be silently ignored
+ with pytest.raises(TypeError):
+ NewBug(summary="s", description="d", component="x", ids=[1])
+
+
+class TestArchRequest:
+ @pytest.fixture
+ def pkglist(self):
+ return PackageList("=dev-libs/a-1 amd64\n=dev-libs/b-2 x86")
+
+ def test_stablereq(self, pkglist):
+ wire = NewBug.arch_request(
+ BugCategory.STABLEREQ, pkglist, maintainers=("[email protected]", "[email protected]")
+ ).to_wire()
+ assert wire["product"] == "Gentoo Linux"
+ assert wire["component"] == "Stabilization"
+ assert wire["severity"] == "enhancement"
+ assert wire["summary"] == "dev-libs/a-1, dev-libs/b-2: stablereq"
+ assert wire["assigned_to"] == "[email protected]"
+ assert wire["cc"] == ["[email protected]"]
+ assert "keywords" not in wire
+ assert wire["cf_stabilisation_atoms"] == str(pkglist)
+
+ def test_keywordreq(self, pkglist):
+ wire = NewBug.arch_request(BugCategory.KEYWORDREQ, pkglist).to_wire()
+ assert wire["component"] == "Keywording"
+ assert wire["description"] == "Please keyword the listed packages."
+
+ def test_unmaintained_falls_back(self, pkglist):
+ wire = NewBug.arch_request(BugCategory.STABLEREQ, pkglist).to_wire()
+ assert wire["assigned_to"] == "[email protected]"
+ assert "cc" not in wire
+
+ def test_cc_arches(self, pkglist):
+ wire = NewBug.arch_request(
+ BugCategory.STABLEREQ, pkglist, cc_arches=True
+ ).to_wire()
+ assert wire["keywords"] == ["CC-ARCHES"]
+
+ def test_explicit_summary_and_description(self, pkglist):
+ wire = NewBug.arch_request(
+ BugCategory.STABLEREQ, pkglist, summary="custom", description="why"
+ ).to_wire()
+ assert wire["summary"] == "custom"
+ assert wire["description"] == "why"
+
+ def test_extra_kwargs_pass_through(self, pkglist):
+ wire = NewBug.arch_request(
+ BugCategory.STABLEREQ, pkglist, depends_on=(7,)
+ ).to_wire()
+ assert wire["depends_on"] == [7]
+
+
+class TestPackageMask:
+ def test_deadline_and_cc(self):
+ wire = NewBug.package_mask(
+ "dev-libs/a: removal",
+ "unmaintained",
+ rites=30,
+ maintainers=("[email protected]", "[email protected]"),
+ today=datetime.date(2024, 1, 1),
+ ).to_wire()
+ assert wire["component"] == "Current packages"
+ assert wire["keywords"] == ["PMASKED"]
+ assert wire["assigned_to"] == "[email protected]"
+ assert wire["cc"] == ["[email protected]", "[email protected]"]
+ assert wire["deadline"] == "2024-01-31"
+
+ def test_unmaintained(self):
+ wire = NewBug.package_mask(
+ "s", "d", rites=30, today=datetime.date(2024, 1, 1)
+ ).to_wire()
+ assert wire["assigned_to"] == "[email protected]"
+ assert wire["cc"] == ["[email protected]"]
+
+
+class TestBugUpdate:
+ def test_empty_is_falsy(self):
+ assert not BugUpdate()
+
+ def test_empty_still_carries_ids(self):
+ assert BugUpdate().to_wire([1]) == {"ids": [1]}
+
+ def test_needs_ids(self):
+ with pytest.raises(BugzillaUsageError, match="at least one bug id"):
+ BugUpdate(summary="x").to_wire([])
+
+ def test_full_id_list_is_always_sent(self):
+ # bugzilla lets the body override the path id, so it must be complete
+ assert BugUpdate(summary="x").to_wire([1, 2, 3])["ids"] == [1, 2, 3]
+
+ def test_cc_add_is_rejected(self):
+ with pytest.raises(TypeError):
+ BugUpdate(cc_add=["[email protected]"])
+
+ def test_list_fields(self):
+ wire = BugUpdate(
+ cc=ListChange.adding("[email protected]"),
+ keywords=ListChange(add=("ALLARCHES",), remove=("CC-ARCHES",)),
+ blocks=ListChange.adding(5),
+ depends_on=ListChange.removing(6),
+ see_also=ListChange.adding("https://bugs.gentoo.org/7"),
+ groups=ListChange.setting("gentoo-security"),
+ ).to_wire([1])
+ assert wire["cc"] == {"add": ["[email protected]"]}
+ assert wire["keywords"] == {"add": ["ALLARCHES"], "remove": ["CC-ARCHES"]}
+ assert wire["blocks"] == {"add": ["5"]}
+ assert wire["depends_on"] == {"remove": ["6"]}
+ assert wire["see_also"] == {"add": ["https://bugs.gentoo.org/7"]}
+ assert wire["groups"] == {"set": ["gentoo-security"]}
+
+ def test_empty_list_fields_are_omitted(self):
+ assert BugUpdate(summary="x").to_wire([1]).keys() == {"ids", "summary"}
+
+ def test_scalars(self):
+ wire = BugUpdate(
+ summary="new summary",
+ assigned_to="[email protected]",
+ whiteboard="B3 [ebuild]",
+ deadline=datetime.date(2024, 3, 1),
+ package_list=PackageList("=dev-libs/a-1 amd64"),
+ runtime_testing_required=RuntimeTesting.YES,
+ ).to_wire([1])
+ assert wire["summary"] == "new summary"
+ assert wire["assigned_to"] == "[email protected]"
+ assert wire["whiteboard"] == "B3 [ebuild]"
+ assert wire["deadline"] == "2024-03-01"
+ assert wire["cf_stabilisation_atoms"] == "=dev-libs/a-1 amd64"
+ assert wire["cf_runtime_testing_required"] == "Yes"
+
+ def test_flags(self):
+ wire = BugUpdate(
+ flags=(FlagChange("sanity-check", FlagStatus.DENIED),)
+ ).to_wire([1])
+ assert wire["flags"] == [{"name": "sanity-check", "status": "-"}]
+
+ def test_flag_requestee(self):
+ change = FlagChange("review", FlagStatus.REQUESTED, requestee="[email protected]")
+ assert change.to_wire() == {
+ "name": "review",
+ "status": "?",
+ "requestee": "[email protected]",
+ }
+
+
+class TestBugUpdateValidation:
+ def test_resolution_needs_status(self):
+ with pytest.raises(BugzillaUsageError, match="needs an explicit status"):
+ BugUpdate(resolution=Resolution.FIXED)
+
+ def test_resolved_needs_resolution(self):
+ with pytest.raises(BugzillaUsageError, match="needs a resolution"):
+ BugUpdate(status=Status.RESOLVED)
+
+ def test_duplicate_needs_dupe_of(self):
+ with pytest.raises(BugzillaUsageError, match="dupe_of"):
+ BugUpdate(status=Status.RESOLVED, resolution=Resolution.DUPLICATE)
+
+ def test_dupe_of_needs_duplicate(self):
+ with pytest.raises(BugzillaUsageError, match="dupe_of"):
+ BugUpdate(status=Status.RESOLVED, resolution=Resolution.FIXED, dupe_of=5)
+
+ def test_duplicate_pair(self):
+ wire = BugUpdate(
+ status=Status.RESOLVED, resolution=Resolution.DUPLICATE, dupe_of=5
+ ).to_wire([1])
+ assert wire["resolution"] == "DUPLICATE"
+ assert wire["dupe_of"] == 5
+
+ def test_status_alone_is_fine(self):
+ assert BugUpdate(status=Status.IN_PROGRESS).to_wire([1])["status"] == (
+ "IN_PROGRESS"
+ )
+
+
+class TestShorthands:
+ @pytest.mark.parametrize(
+ ("status", "expected"), ((True, "+"), (False, "-"), (None, "X"))
+ )
+ def test_sanity_check(self, status, expected):
+ wire = BugUpdate.sanity_check(status).to_wire([1])
+ assert wire["flags"] == [{"name": "sanity-check", "status": expected}]
+ assert "comment" not in wire
+
+ def test_sanity_check_with_comment(self):
+ wire = BugUpdate.sanity_check(False, comment="broken").to_wire([1])
+ assert wire["comment"] == {"body": "broken"}
+
+ def test_sanity_check_with_extra_fields(self):
+ wire = BugUpdate.sanity_check(
+ True, cc=ListChange.adding("[email protected]")
+ ).to_wire([1])
+ assert wire["cc"] == {"add": ["[email protected]"]}
+
+ def test_resolve(self):
+ wire = BugUpdate.resolve(comment="all arches done").to_wire([1])
+ assert wire["status"] == "RESOLVED"
+ assert wire["resolution"] == "FIXED"
+ assert wire["comment"] == {"body": "all arches done"}
+
+ def test_resolve_with_uncc(self):
+ wire = BugUpdate.resolve(
+ comment="done", cc=ListChange.removing("[email protected]")
+ ).to_wire([1])
+ assert wire["cc"] == {"remove": ["[email protected]"]}
+
+ def test_obsoleted_by(self):
+ wire = BugUpdate.obsoleted_by(999).to_wire([1, 2])
+ assert wire == {
+ "ids": [1, 2],
+ "status": "RESOLVED",
+ "resolution": "OBSOLETE",
+ "see_also": {"add": ["https://bugs.gentoo.org/999"]},
+ }
diff --git a/tests/bugzilla/test_client.py b/tests/bugzilla/test_client.py
new file mode 100644
index 000000000..804f16904
--- /dev/null
+++ b/tests/bugzilla/test_client.py
@@ -0,0 +1,394 @@
+import pytest
+
+from pkgcore.bugzilla import errors
+from pkgcore.bugzilla.changes import BugUpdate, ListChange, NewBug
+from pkgcore.bugzilla.client import PAGE_SIZE
+from pkgcore.bugzilla.enums import BugCategory, Component, FlagStatus, Resolution
+from pkgcore.bugzilla.query import BugQuery
+from pkgcore.bugzilla.testing import API_KEY, response
+
+WHOAMI = response({"id": 7, "name": "[email protected]", "real_name": "A Dev"})
+
+
+def raw_bug(bug_id, **kwargs):
+ return {
+ "id": bug_id,
+ "product": "Gentoo Linux",
+ "component": "Stabilization",
+ "resolution": "",
+ "summary": f"bug {bug_id}",
+ "cc": [],
+ "keywords": [],
+ "depends_on": [],
+ "blocks": [],
+ "flags": [],
+ "alias": [],
+ "cf_stabilisation_atoms": "",
+ "cf_runtime_testing_required": "---",
+ "last_change_time": "2024-01-02T03:04:05Z",
+ "creation_time": "2024-01-01T00:00:00Z",
+ **kwargs,
+ }
+
+
+def raw_comment(comment_id, bug_id, creator, tags=(), text="hi"):
+ return {
+ "id": comment_id,
+ "bug_id": bug_id,
+ "count": comment_id,
+ "text": text,
+ "creator": creator,
+ "time": "2024-01-01T00:00:00Z",
+ "creation_time": "2024-01-01T00:00:00Z",
+ "is_private": False,
+ "tags": list(tags),
+ }
+
+
+class TestWhoami:
+ def test_whoami(self, client):
+ handler, bz = client(WHOAMI)
+ user = bz.whoami()
+ assert user.name == "[email protected]"
+ assert handler.calls[0].path == "/rest/whoami"
+
+ def test_cached(self, client):
+ handler, bz = client(WHOAMI)
+ assert bz.whoami() is bz.whoami()
+ assert len(handler.calls) == 1
+
+
+class TestGet:
+ def test_single(self, client):
+ handler, bz = client(response({"bugs": [raw_bug(900001)]}))
+ bug = bz.get(900001)
+ assert bug.id == 900001
+ assert bug.category is BugCategory.STABLEREQ
+ assert ("id", "900001") in handler.calls[0].query
+
+ def test_single_missing(self, client):
+ _, bz = client(response({"bugs": []}))
+ with pytest.raises(errors.BugzillaNotFound, match="900001"):
+ bz.get(900001)
+
+ def test_several(self, client):
+ _, bz = client(response({"bugs": [raw_bug(1), raw_bug(2)]}))
+ bugs = bz.get([1, 2])
+ assert sorted(bugs) == [1, 2]
+ assert bugs[1].summary == "bug 1"
+
+ def test_several_tolerates_missing(self, client):
+ _, bz = client(response({"bugs": [raw_bug(1)]}))
+ assert list(bz.get([1, 2])) == [1]
+
+
+class TestSearch:
+ def test_include_fields_is_always_sent(self, client):
+ handler, bz = client(response({"bugs": []}))
+ bz.search(BugQuery.unresolved())
+ fields = dict(handler.calls[0].query)["include_fields"]
+ assert "cf_stabilisation_atoms" in fields
+ assert "id" in fields
+
+ def test_query_params_are_forwarded(self, client):
+ handler, bz = client(response({"bugs": []}))
+ bz.search(
+ BugQuery.component(Component.STABILIZATION)
+ & BugQuery.flag("sanity-check", FlagStatus.GRANTED)
+ )
+ query = handler.calls[0].query
+ assert ("component", "Stabilization") in query
+ assert ("f1", "flagtypes.name") in query
+ assert ("v1", "sanity-check+") in query
+
+ def test_paging_stops_on_a_short_page(self, client):
+ handler, bz = client(response({"bugs": [raw_bug(1)]}))
+ assert list(bz.search()) == [1]
+ assert len(handler.calls) == 1
+ assert ("limit", str(PAGE_SIZE)) in handler.calls[0].query
+
+ def test_paging_follows_full_pages(self, client):
+ handler, bz = client(
+ response({"bugs": [raw_bug(i) for i in range(PAGE_SIZE)]}),
+ response({"bugs": [raw_bug(PAGE_SIZE)]}),
+ )
+ assert len(bz.search()) == PAGE_SIZE + 1
+ assert len(handler.calls) == 2
+ assert ("offset", str(PAGE_SIZE)) in handler.calls[1].query
+
+ def test_explicit_limit_is_honoured(self, client):
+ handler, bz = client(response({"bugs": [raw_bug(1), raw_bug(2)]}))
+ assert len(bz.search(BugQuery().paged(2))) == 2
+ assert ("limit", "2") in handler.calls[0].query
+ assert len(handler.calls) == 1
+
+ def test_batching_splits_long_id_lists(self, client):
+ ids = list(range(900000, 903000))
+ handler, bz = client(
+ *(
+ response({"bugs": []})
+ for _ in BugQuery.ids(ids).batches(base_length=500)
+ )
+ )
+ bz.search(BugQuery.ids(ids))
+ assert len(handler.calls) > 1
+ seen = [v for call in handler.calls for k, v in call.query if k == "id"]
+ assert sorted(map(int, seen)) == ids
+
+ def test_raw_search_narrow_projection(self, client):
+ handler, bz = client(response({"bugs": [{"id": 1, "summary": "s"}]}))
+ raw = bz.raw_search(BugQuery.unresolved(), fields=("id", "summary"))
+ assert raw == ({"id": 1, "summary": "s"},)
+ assert dict(handler.calls[0].query)["include_fields"] == "id,summary"
+
+ def test_malformed_response(self, client):
+ _, bz = client(response({"unexpected": 1}))
+ with pytest.raises(errors.BugzillaSchemaError, match="'bugs' list"):
+ bz.search()
+
+
+class TestResolveDependencies:
+ def test_transitive_closure(self, client):
+ _, bz = client(
+ response({"bugs": [raw_bug(2, depends_on=[3])]}),
+ response({"bugs": [raw_bug(3)]}),
+ )
+ start = {1: bz.get(2)}
+ resolved = bz.resolve_dependencies({2: start[1]})
+ assert sorted(resolved) == [2, 3]
+
+ def test_nothing_to_do(self, client):
+ handler, bz = client(response({"bugs": [raw_bug(1)]}))
+ bugs = bz.get([1])
+ assert bz.resolve_dependencies(bugs) == bugs
+ assert len(handler.calls) == 1
+
+ def test_unreachable_dependency_is_dropped(self, client, caplog):
+ # a deleted or security restricted dep must not spin forever
+ _, bz = client(
+ response({"bugs": [raw_bug(1, depends_on=[999])]}),
+ response({"bugs": []}),
+ )
+ bugs = bz.get([1])
+ assert list(bz.resolve_dependencies(bugs)) == [1]
+ assert "unreachable bug dependencies" in caplog.text
+
+ def test_input_is_not_mutated(self, client):
+ _, bz = client(
+ response({"bugs": [raw_bug(1, depends_on=[2])]}),
+ response({"bugs": [raw_bug(2)]}),
+ )
+ bugs = bz.get([1])
+ bz.resolve_dependencies(bugs)
+ assert list(bugs) == [1]
+
+
+class TestComments:
+ def test_comments(self, client):
+ _, bz = client(
+ response(
+ {
+ "bugs": {
+ "5": {
+ "comments": [
+ raw_comment(1, 5, "[email protected]"),
+ raw_comment(2, 5, "[email protected]"),
+ ]
+ }
+ },
+ "comments": {},
+ }
+ )
+ )
+ comments = bz.comments(5)
+ assert [x.id for x in comments] == [1, 2]
+
+ def test_missing_bug_key(self, client):
+ _, bz = client(response({"bugs": {}, "comments": {}}))
+ with pytest.raises(errors.BugzillaSchemaError, match="no entry for bug 5"):
+ bz.comments(5)
+
+ def test_latest_comment_defaults_to_the_current_user(self, client):
+ _, bz = client(
+ WHOAMI,
+ response(
+ {
+ "bugs": {
+ "5": {
+ "comments": [
+ raw_comment(1, 5, "[email protected]", text="old"),
+ raw_comment(2, 5, "[email protected]"),
+ raw_comment(3, 5, "[email protected]", text="new"),
+ ]
+ }
+ },
+ "comments": {},
+ }
+ ),
+ )
+ assert bz.latest_comment(5).text == "new"
+
+ def test_latest_comment_for_another_creator(self, client):
+ _, bz = client(
+ response(
+ {
+ "bugs": {"5": {"comments": [raw_comment(1, 5, "[email protected]")]}},
+ "comments": {},
+ }
+ )
+ )
+ assert bz.latest_comment(5, creator="[email protected]").id == 1
+
+ def test_latest_comment_when_there_is_none(self, client):
+ _, bz = client(
+ response(
+ {
+ "bugs": {"5": {"comments": [raw_comment(1, 5, "[email protected]")]}},
+ "comments": {},
+ }
+ )
+ )
+ assert bz.latest_comment(5, creator="[email protected]") is None
+
+
+class TestCreate:
+ def test_create(self, client):
+ handler, bz = client(response({"id": 900123}))
+ bug = NewBug(
+ summary="dev-libs/a: stablereq",
+ description="please",
+ component=Component.STABILIZATION,
+ )
+ assert bz.create(bug) == 900123
+ (call,) = handler.calls
+ assert call.method == "POST"
+ assert call.path == "/rest/bug"
+ assert call.body["summary"] == "dev-libs/a: stablereq"
+ assert call.body["Bugzilla_api_key"] == API_KEY
+
+ def test_missing_id_in_response(self, client):
+ _, bz = client(response({}))
+ bug = NewBug(summary="s", description="d", component=Component.ECLASSES)
+ with pytest.raises(errors.BugzillaSchemaError, match="no id in response"):
+ bz.create(bug)
+
+ def test_anonymous_client_cannot_create(self, client):
+ handler, bz = client(api_key=None)
+ bug = NewBug(summary="s", description="d", component=Component.ECLASSES)
+ with pytest.raises(errors.BugzillaAuthRequired):
+ bz.create(bug)
+ assert handler.calls == []
+
+ def test_not_retried(self, client, monkeypatch):
+ monkeypatch.setattr("pkgcore.bugzilla.transport.time.sleep", lambda _: None)
+ handler, bz = client(response(raw=b"", status=503), retries=3)
+ bug = NewBug(summary="s", description="d", component=Component.ECLASSES)
+ with pytest.raises(errors.BugzillaServerError):
+ bz.create(bug)
+ assert len(handler.calls) == 1
+
+
+class TestUpdate:
+ CHANGED = response(
+ {
+ "bugs": [
+ {
+ "id": 5,
+ "alias": [],
+ "last_change_time": "2024-01-02T03:04:05Z",
+ "changes": {
+ "status": {"added": "RESOLVED", "removed": "CONFIRMED"}
+ },
+ }
+ ]
+ }
+ )
+
+ def test_single(self, client):
+ handler, bz = client(self.CHANGED)
+ changes = bz.update(5, BugUpdate.resolve(comment="done"))
+ assert changes.id == 5
+ assert changes.changes["status"].added == ("RESOLVED",)
+ (call,) = handler.calls
+ assert call.method == "PUT"
+ assert call.path == "/rest/bug/5"
+ assert call.body["ids"] == [5]
+
+ def test_several_send_the_full_id_list(self, client):
+ # the body overrides the path id, so a partial list would lose bugs
+ handler, bz = client(self.CHANGED)
+ result = bz.update([5, 6, 7], BugUpdate(summary="x"))
+ assert isinstance(result, tuple)
+ (call,) = handler.calls
+ assert call.path == "/rest/bug/5"
+ assert call.body["ids"] == [5, 6, 7]
+
+ def test_needs_ids(self, client):
+ _, bz = client()
+ with pytest.raises(errors.BugzillaUsageError, match="at least one bug id"):
+ bz.update([], BugUpdate(summary="x"))
+
+ def test_obsoleting(self, client):
+ handler, bz = client(self.CHANGED)
+ bz.update(5, BugUpdate.obsoleted_by(900999))
+ assert handler.calls[0].body["resolution"] == Resolution.OBSOLETE
+ assert handler.calls[0].body["see_also"] == {
+ "add": ["https://bugs.gentoo.org/900999"]
+ }
+
+ def test_sanity_check(self, client):
+ handler, bz = client(self.CHANGED)
+ bz.update(
+ 5, BugUpdate.sanity_check(True, cc=ListChange.adding("[email protected]"))
+ )
+ body = handler.calls[0].body
+ assert body["flags"] == [{"name": "sanity-check", "status": "+"}]
+ assert body["cc"] == {"add": ["[email protected]"]}
+
+
+class TestCommentTags:
+ def test_tag_comments(self, client):
+ handler, bz = client(response(["obsolete"]), response(["obsolete"]))
+ bz.tag_comments([11, 12], ListChange.adding("obsolete"))
+ assert [x.path for x in handler.calls] == [
+ "/rest/bug/comment/11/tags",
+ "/rest/bug/comment/12/tags",
+ ]
+ assert handler.calls[0].body["add"] == ["obsolete"]
+
+ def test_mark_own_comments_obsolete(self, client):
+ handler, bz = client(
+ WHOAMI,
+ response(
+ {
+ "bugs": {
+ "5": {
+ "comments": [
+ raw_comment(1, 5, "[email protected]"),
+ raw_comment(2, 5, "[email protected]"),
+ raw_comment(3, 5, "[email protected]", tags=["obsolete"]),
+ ]
+ }
+ },
+ "comments": {},
+ }
+ ),
+ response(["obsolete"]),
+ )
+ assert bz.mark_own_comments_obsolete(5) == 1
+ assert handler.calls[-1].path == "/rest/bug/comment/1/tags"
+
+ def test_nothing_to_obsolete(self, client):
+ handler, bz = client(
+ WHOAMI,
+ response(
+ {
+ "bugs": {
+ "5": {"comments": [raw_comment(1, 5, "[email protected]")]}
+ },
+ "comments": {},
+ }
+ ),
+ )
+ assert bz.mark_own_comments_obsolete(5) == 0
+ assert len(handler.calls) == 2
diff --git a/tests/bugzilla/test_network.py b/tests/bugzilla/test_network.py
new file mode 100644
index 000000000..6669eb786
--- /dev/null
+++ b/tests/bugzilla/test_network.py
@@ -0,0 +1,97 @@
+"""Read only smoke tests against the real bugs.gentoo.org.
+
+These catch schema drift that no cassette can. Run with ``pytest --network``.
+"""
+
+import pytest
+
+from pkgcore.bugzilla import errors
+from pkgcore.bugzilla.bug import INCLUDE_FIELDS
+from pkgcore.bugzilla.client import Bugzilla
+from pkgcore.bugzilla.enums import OPEN_STATUSES, BugCategory, Component
+from pkgcore.bugzilla.query import BugQuery
+from pkgcore.bugzilla.transport import AuthMode
+
+
[email protected]
+def bugzilla():
+ return Bugzilla()
+
+
+# an ancient, public, long settled bug; bug 1 itself isn't reachable
+KNOWN_BUG = 100
+
+
[email protected]_network
+class TestLive:
+ def test_get_a_bug(self, bugzilla):
+ bug = bugzilla.get(KNOWN_BUG)
+ assert bug.id == KNOWN_BUG
+ assert bug.summary
+ assert bug.creation_time.year == 2002
+
+ def test_missing_bug(self, bugzilla):
+ with pytest.raises(errors.BugzillaNotFound):
+ bugzilla.get(999999999)
+
+ def test_every_include_field_comes_back(self, bugzilla):
+ raw = bugzilla.raw_search(BugQuery.component(Component.STABILIZATION).paged(1))
+ assert raw, "no open stabilization bugs, which should never happen"
+ missing = set(INCLUDE_FIELDS) - set(raw[0])
+ assert not missing, f"bgo no longer returns {sorted(missing)}"
+
+ def test_search_and_parse(self, bugzilla):
+ bugs = bugzilla.search(
+ BugQuery.category(BugCategory.STABLEREQ)
+ & BugQuery.unresolved()
+ & BugQuery().paged(5)
+ )
+ assert bugs
+ for bug in bugs.values():
+ assert bug.category is BugCategory.STABLEREQ
+ assert not bug.resolved
+ # exercises the package list parser against real content
+ assert bug.package_list.atoms is not None
+
+ def test_unresolved_matches_the_open_statuses(self, bugzilla):
+ base = BugQuery.component(Component.KEYWORDING)
+ fields = ("id",)
+ unresolved = {
+ b["id"]
+ for b in bugzilla.raw_search(base & BugQuery.unresolved(), fields=fields)
+ }
+ open_status = {
+ b["id"]
+ for b in bugzilla.raw_search(
+ base & BugQuery.status(*OPEN_STATUSES), fields=fields
+ )
+ }
+ assert unresolved and unresolved == open_status
+
+ def test_boolean_charts_filter(self, bugzilla):
+ query = (
+ BugQuery.component(Component.STABILIZATION)
+ & BugQuery.flag("sanity-check", "+")
+ & BugQuery().paged(5)
+ )
+ for bug in bugzilla.search(query).values():
+ assert bug.sanity_check is True
+
+ def test_anonymous_truncates_emails(self, bugzilla):
+ # documents why anything matching on addresses needs an api key
+ bug = bugzilla.get(KNOWN_BUG)
+ assert "@" not in bug.assigned_to
+
+ def test_whoami_needs_a_key(self, bugzilla):
+ with pytest.raises(errors.BugzillaAuthError):
+ bugzilla.whoami()
+
+ def test_header_auth_is_still_unsupported(self):
+ # if this ever starts passing, bgo has moved past bugzilla 5.2 and
+ # AuthMode.HEADER can become the default
+ client = Bugzilla("invalid-key", auth_mode=AuthMode.HEADER)
+ with pytest.raises(errors.BugzillaAuthError) as excinfo:
+ client.whoami()
+ assert excinfo.value.code == 410, (
+ "bgo now reads X-BUGZILLA-API-KEY; expected it to be ignored"
+ )
diff --git a/tests/bugzilla/test_pkglist.py b/tests/bugzilla/test_pkglist.py
new file mode 100644
index 000000000..0d5d2effa
--- /dev/null
+++ b/tests/bugzilla/test_pkglist.py
@@ -0,0 +1,206 @@
+import pytest
+
+from pkgcore.bugzilla.errors import PackageListError
+from pkgcore.bugzilla.pkglist import PackageList, PackageListEntry, parse_atom
+from pkgcore.ebuild.atom import atom
+from pkgcore.ebuild.errors import MalformedAtom
+
+
+class TestParseAtom:
+ @pytest.mark.parametrize(
+ ("token", "expected"),
+ (
+ ("dev-python/foo-1.2", "=dev-python/foo-1.2"),
+ ("=dev-python/foo-1.2", "=dev-python/foo-1.2"),
+ ("dev-python/foo", "dev-python/foo"),
+ ("dev-python/foo:3", "dev-python/foo:3"),
+ ("=dev-libs/x-1.2.3_p1-r2", "=dev-libs/x-1.2.3_p1-r2"),
+ (">=dev-libs/x-1.2", ">=dev-libs/x-1.2"),
+ ),
+ )
+ def test_valid(self, token, expected):
+ assert parse_atom(token) == atom(expected)
+
+ @pytest.mark.parametrize(
+ "token",
+ (
+ "",
+ "not-an-atom",
+ "!dev-libs/foo",
+ "dev-libs/foo[bar]",
+ "dev-libs/foo:*",
+ ),
+ )
+ def test_invalid(self, token):
+ with pytest.raises(MalformedAtom):
+ parse_atom(token)
+
+
+class TestPackageList:
+ def test_empty(self):
+ pkglist = PackageList()
+ assert not pkglist
+ assert pkglist.entries == ()
+ assert pkglist.atoms == ()
+ assert str(pkglist) == ""
+
+ def test_blank_is_falsy(self):
+ assert not PackageList("\n \r\n")
+
+ def test_parse(self):
+ pkglist = PackageList(
+ " dev-python/foo-1.2 amd64 x86 # careful\n"
+ "dev-libs/bar\n"
+ "\n"
+ "# standalone comment\n"
+ )
+ first, second, blank, comment = pkglist.entries
+ assert (first.lineno, first.pkg, first.keywords) == (
+ 1,
+ atom("=dev-python/foo-1.2"),
+ ("amd64", "x86"),
+ )
+ assert first.comment == "# careful"
+ assert not first.is_blank
+ assert (second.pkg, second.keywords, second.comment) == (
+ atom("dev-libs/bar"),
+ (),
+ "",
+ )
+ assert blank.is_blank and blank.comment == ""
+ assert comment.is_blank and comment.comment == "# standalone comment"
+ assert pkglist.atoms == (atom("=dev-python/foo-1.2"), atom("dev-libs/bar"))
+
+ def test_hash_needs_leading_whitespace_to_start_a_comment(self):
+ (entry,) = PackageList("dev-libs/a amd64#x86").entries
+ assert entry.comment == ""
+ assert entry.keywords == ("amd64#x86",)
+
+ @pytest.mark.parametrize("eol", ("\n", "\r\n"))
+ def test_round_trip(self, eol):
+ text = eol.join(
+ (" dev-libs/a amd64 # note", "dev-libs/b *", "", "dev-libs/c")
+ )
+ assert str(PackageList(text)) == text
+
+ def test_round_trip_trailing_newline(self):
+ text = "dev-libs/a amd64\r\n"
+ assert str(PackageList(text)) == text
+
+ def test_malformed_atom_reports_line(self):
+ pkglist = PackageList("dev-libs/a\nnot an atom\n", bug_id=42)
+ with pytest.raises(PackageListError) as excinfo:
+ assert pkglist.entries
+ assert excinfo.value.lineno == 2
+ assert excinfo.value.bug_id == 42
+ assert "bug 42, line 2" in str(excinfo.value)
+
+ def test_parse_is_lazy(self):
+ # constructing must not raise, only looking at the entries does
+ pkglist = PackageList("not an atom")
+ with pytest.raises(PackageListError):
+ assert pkglist.entries
+
+ def test_entries_cached(self):
+ pkglist = PackageList("dev-libs/a amd64")
+ assert pkglist.entries is pkglist.entries
+
+ def test_keywords_for(self):
+ pkglist = PackageList("dev-libs/a amd64 x86\ndev-libs/b arm")
+ assert pkglist.keywords_for(atom("dev-libs/a")) == ("amd64", "x86")
+ assert pkglist.keywords_for(atom("dev-libs/b")) == ("arm",)
+ assert pkglist.keywords_for(atom("dev-libs/nope")) == ()
+
+ def test_build(self):
+ pkglist = PackageList.build(
+ (
+ (atom("=dev-libs/a-1"), ("amd64", "x86")),
+ (atom("dev-libs/b"), ()),
+ )
+ )
+ assert str(pkglist) == "=dev-libs/a-1 amd64 x86\ndev-libs/b"
+ assert pkglist.atoms == (atom("=dev-libs/a-1"), atom("dev-libs/b"))
+
+ def test_equality_and_hash(self):
+ assert PackageList("dev-libs/a") == PackageList("dev-libs/a")
+ assert PackageList("dev-libs/a") != PackageList("dev-libs/b")
+ assert PackageList("dev-libs/a") != "dev-libs/a"
+ assert len({PackageList("dev-libs/a"), PackageList("dev-libs/a")}) == 1
+
+ def test_immutable(self):
+ pkglist = PackageList("dev-libs/a")
+ with pytest.raises(AttributeError):
+ pkglist.text = "dev-libs/b"
+
+
+class TestExpand:
+ def test_all_keywords(self):
+ pkglist = PackageList("dev-libs/a *\n")
+ assert (
+ str(pkglist.expand(lambda pkg: ("alpha", "hppa")))
+ == "dev-libs/a alpha hppa\n"
+ )
+
+ def test_all_keywords_empty_collapses_to_dash(self):
+ assert str(PackageList("dev-libs/a *").expand(lambda pkg: ())) == "dev-libs/a -"
+
+ def test_same_keywords(self):
+ pkglist = PackageList("dev-libs/a amd64 x86\ndev-libs/b ^\n")
+ expanded = pkglist.expand(lambda pkg: ())
+ assert str(expanded) == "dev-libs/a amd64 x86\ndev-libs/b amd64 x86\n"
+
+ def test_same_keywords_chains(self):
+ pkglist = PackageList("dev-libs/a *\ndev-libs/b ^\ndev-libs/c ^")
+ expanded = pkglist.expand(lambda pkg: ("arm",))
+ assert str(expanded) == "dev-libs/a arm\ndev-libs/b arm\ndev-libs/c arm"
+
+ def test_same_keywords_on_first_line(self):
+ with pytest.raises(PackageListError, match="no line above"):
+ PackageList("dev-libs/a ^", bug_id=7).expand(lambda pkg: ())
+
+ def test_mixed_sentinel_and_literal(self):
+ pkglist = PackageList("dev-libs/a * ppc")
+ assert str(pkglist.expand(lambda pkg: ("amd64",))) == "dev-libs/a amd64 ppc"
+
+ def test_preserves_line_endings(self):
+ pkglist = PackageList("dev-libs/a *\r\ndev-libs/b ^\r\n")
+ assert str(pkglist.expand(lambda pkg: ("arm",))) == (
+ "dev-libs/a arm\r\ndev-libs/b arm\r\n"
+ )
+
+ def test_preserves_untouched_lines(self):
+ text = " dev-libs/a amd64 # keep me\ndev-libs/b *\n"
+ expanded = PackageList(text).expand(lambda pkg: ("arm",))
+ assert str(expanded) == " dev-libs/a amd64 # keep me\ndev-libs/b arm\n"
+
+ def test_keeps_comment_on_rewritten_line(self):
+ expanded = PackageList("dev-libs/a * # why").expand(lambda pkg: ("arm",))
+ assert str(expanded) == "dev-libs/a arm # why"
+
+ def test_no_sentinels_returns_self(self):
+ pkglist = PackageList("dev-libs/a amd64\n")
+ assert pkglist.expand(lambda pkg: ("arm",)) is pkglist
+
+ def test_blank_lines_do_not_reset_previous(self):
+ pkglist = PackageList("dev-libs/a amd64\n\n# note\ndev-libs/b ^")
+ assert str(pkglist.expand(lambda pkg: ())) == (
+ "dev-libs/a amd64\n\n# note\ndev-libs/b amd64"
+ )
+
+
+class TestPackageListEntry:
+ def test_with_keywords_keeps_indent_and_comment(self):
+ (entry,) = PackageList(" dev-libs/a amd64 # note").entries
+ updated = entry.with_keywords(("arm", "ppc"))
+ assert updated.raw == " dev-libs/a arm ppc # note"
+ assert updated.keywords == ("arm", "ppc")
+ assert updated.lineno == entry.lineno
+
+ def test_with_keywords_on_blank_is_a_noop(self):
+ (entry,) = PackageList("# just a comment").entries
+ assert entry.with_keywords(("arm",)) is entry
+
+ def test_frozen(self):
+ entry = PackageListEntry(1, "dev-libs/a", atom("dev-libs/a"))
+ with pytest.raises(AttributeError):
+ entry.lineno = 2
diff --git a/tests/bugzilla/test_query.py b/tests/bugzilla/test_query.py
new file mode 100644
index 000000000..9ea7eb63c
--- /dev/null
+++ b/tests/bugzilla/test_query.py
@@ -0,0 +1,285 @@
+import urllib.parse
+
+import pytest
+
+from pkgcore.bugzilla.enums import (
+ BugCategory,
+ ChartOp,
+ Component,
+ FlagStatus,
+ Join,
+ Product,
+ Status,
+)
+from pkgcore.bugzilla.errors import BugzillaUsageError
+from pkgcore.bugzilla.query import MAX_URL_LENGTH, BugQuery, ChartGroup, Criterion
+
+
+def encoded(query):
+ return urllib.parse.urlencode(query.params())
+
+
+class TestSimpleParams:
+ def test_empty(self):
+ assert BugQuery().params() == []
+
+ def test_ids(self):
+ assert BugQuery.ids((1, 2, 3)).params() == [
+ ("id", "1"),
+ ("id", "2"),
+ ("id", "3"),
+ ]
+
+ def test_component(self):
+ query = BugQuery.component(Component.STABILIZATION, Component.KEYWORDING)
+ assert query.params() == [
+ ("component", "Stabilization"),
+ ("component", "Keywording"),
+ ]
+
+ def test_category(self):
+ query = BugQuery.category(BugCategory.STABLEREQ, BugCategory.KEYWORDREQ)
+ assert query.params() == [
+ ("product", "Gentoo Linux"),
+ ("component", "Stabilization"),
+ ("component", "Keywording"),
+ ]
+
+ def test_unresolved(self):
+ assert BugQuery.unresolved().params() == [("resolution", "---")]
+
+ def test_status(self):
+ query = BugQuery.status(Status.CONFIRMED, Status.IN_PROGRESS)
+ assert query.params() == [
+ ("bug_status", "CONFIRMED"),
+ ("bug_status", "IN_PROGRESS"),
+ ]
+
+ def test_cc_and_assigned_to(self):
+ query = BugQuery.cc("[email protected]") & BugQuery.assigned_to("[email protected]")
+ assert query.params() == [
+ ("cc", "[email protected]"),
+ ("assigned_to", "[email protected]"),
+ ]
+
+ def test_product(self):
+ assert BugQuery.product(Product.GENTOO_SECURITY).params() == [
+ ("product", "Gentoo Security")
+ ]
+
+
+class TestCharts:
+ def test_flag(self):
+ query = BugQuery.flag("sanity-check", FlagStatus.GRANTED, FlagStatus.DENIED)
+ assert query.params() == [
+ ("f1", "flagtypes.name"),
+ ("o1", "anywords"),
+ ("v1", "sanity-check+"),
+ ("v1", "sanity-check-"),
+ ]
+
+ def test_without_tags(self):
+ assert BugQuery.without_tags("nattka:skip").params() == [
+ ("f1", "tag"),
+ ("o1", "nowordssubstr"),
+ ("v1", "nattka:skip"),
+ ]
+
+ def test_negate(self):
+ query = BugQuery(
+ charts=(Criterion("keywords", ChartOp.ANY_WORDS, ("x",), negate=True),)
+ )
+ assert query.params() == [
+ ("f1", "keywords"),
+ ("o1", "anywords"),
+ ("v1", "x"),
+ ("n1", "1"),
+ ]
+
+ def test_slots_are_allocated_at_render_time(self):
+ query = (
+ BugQuery.flag("sanity-check", FlagStatus.GRANTED)
+ & BugQuery.without_tags("nattka:skip")
+ & BugQuery.keywords("ALLARCHES")
+ )
+ slots = [key for key, _ in query.params() if key.startswith("f")]
+ assert slots == ["f1", "f2", "f3"]
+
+ def test_combining_never_collides(self):
+ # each half is authored as f1; combining must renumber
+ combined = BugQuery.flag("sanity-check", FlagStatus.GRANTED) & BugQuery.flag(
+ "other", FlagStatus.DENIED
+ )
+ assert combined.params() == [
+ ("f1", "flagtypes.name"),
+ ("o1", "anywords"),
+ ("v1", "sanity-check+"),
+ ("f2", "flagtypes.name"),
+ ("o2", "anywords"),
+ ("v2", "other-"),
+ ]
+
+ def test_any_of(self):
+ query = BugQuery.any_of(
+ BugQuery.keywords("ALLARCHES"),
+ BugQuery.flag("sanity-check", FlagStatus.DENIED),
+ )
+ assert query.params() == [
+ ("f1", "OP"),
+ ("j1", "OR"),
+ ("f2", "keywords"),
+ ("o2", "anywords"),
+ ("v2", "ALLARCHES"),
+ ("f3", "flagtypes.name"),
+ ("o3", "anywords"),
+ ("v3", "sanity-check-"),
+ ("f4", "CP"),
+ ]
+
+ def test_any_of_rejects_simple_params(self):
+ with pytest.raises(BugzillaUsageError, match="chart based"):
+ BugQuery.any_of(BugQuery.unresolved(), BugQuery.keywords("x"))
+
+ def test_nested_groups(self):
+ inner = ChartGroup(Join.AND, (Criterion("a", ChartOp.EQUALS, ("1",)),))
+ query = BugQuery(charts=(ChartGroup(Join.OR, (inner,)),))
+ assert query.params() == [
+ ("f1", "OP"),
+ ("j1", "OR"),
+ ("f2", "OP"),
+ ("j2", "AND"),
+ ("f3", "a"),
+ ("o3", "equals"),
+ ("v3", "1"),
+ ("f4", "CP"),
+ ("f5", "CP"),
+ ]
+
+ def test_group_after_criterion_continues_numbering(self):
+ query = BugQuery.keywords("x") & BugQuery.any_of(BugQuery.keywords("y"))
+ assert [key for key, _ in query.params() if key[0] in "fj"] == [
+ "f1",
+ "f2",
+ "j2",
+ "f3",
+ "f4",
+ ]
+
+
+class TestCombining:
+ def test_documented_example(self):
+ query = (
+ BugQuery.component(Component.STABILIZATION, Component.KEYWORDING)
+ & BugQuery.unresolved()
+ & BugQuery.flag("sanity-check", FlagStatus.GRANTED)
+ & BugQuery.without_tags("nattka:skip")
+ )
+ assert encoded(query) == (
+ "component=Stabilization&component=Keywording&resolution=---"
+ "&f1=flagtypes.name&o1=anywords&v1=sanity-check%2B"
+ "&f2=tag&o2=nowordssubstr&v2=nattka%3Askip"
+ )
+
+ def test_same_key_values_are_unioned(self):
+ query = BugQuery.component(Component.STABILIZATION) & BugQuery.component(
+ Component.KEYWORDING
+ )
+ assert query.params() == [
+ ("component", "Stabilization"),
+ ("component", "Keywording"),
+ ]
+
+ def test_duplicate_values_are_dropped(self):
+ query = BugQuery.ids((1, 2)) & BugQuery.ids((2, 3))
+ assert query.params() == [("id", "1"), ("id", "2"), ("id", "3")]
+
+ def test_right_hand_limit_wins(self):
+ left = BugQuery().paged(10)
+ assert (left & BugQuery().paged(20)).limit == 20
+ assert (left & BugQuery()).limit == 10
+
+ def test_operands_are_unchanged(self):
+ left = BugQuery.ids((1,))
+ right = BugQuery.unresolved()
+ left & right
+ assert left.params() == [("id", "1")]
+ assert right.params() == [("resolution", "---")]
+
+
+class TestPaging:
+ def test_paged(self):
+ query = BugQuery.unresolved().paged(100, 200)
+ assert query.params()[-2:] == [("limit", "100"), ("offset", "200")]
+
+ def test_offset_zero_is_omitted(self):
+ assert ("offset", "0") not in BugQuery().paged(100).params()
+
+ @pytest.mark.parametrize("limit", (0, -1))
+ def test_rejects_non_positive_limit(self, limit):
+ # bugzilla treats limit=0 as unlimited and silently drops the offset
+ with pytest.raises(BugzillaUsageError, match="limit must be positive"):
+ BugQuery().paged(limit)
+
+ def test_rejects_negative_offset(self):
+ with pytest.raises(BugzillaUsageError, match="offset"):
+ BugQuery().paged(10, -1)
+
+ def test_order(self):
+ query = BugQuery(order="bug_id")
+ assert query.params() == [("order", "bug_id")]
+
+
+class TestBatches:
+ def test_no_splittable_axis_yields_itself(self):
+ query = BugQuery.unresolved()
+ assert list(query.batches()) == [query]
+
+ def test_small_query_is_a_single_batch(self):
+ query = BugQuery.ids((1, 2, 3))
+ assert [b.params() for b in query.batches()] == [query.params()]
+
+ def test_ids_are_split(self):
+ query = BugQuery.ids(range(900000, 902000))
+ batches = list(query.batches())
+ assert len(batches) > 1
+ for batch in batches:
+ assert len(urllib.parse.urlencode(batch.params())) <= MAX_URL_LENGTH
+
+ def test_split_is_lossless_and_ordered(self):
+ ids = list(range(900000, 902000))
+ batches = BugQuery.ids(ids).batches()
+ assert [int(v) for b in batches for k, v in b.params() if k == "id"] == ids
+
+ def test_fixed_params_are_repeated(self):
+ query = BugQuery.ids(range(900000, 902000)) & BugQuery.unresolved()
+ for batch in query.batches():
+ assert ("resolution", "---") in batch.params()
+
+ def test_package_list_is_split(self):
+ packages = [
+ f"=dev-libs/verylongpackagename{i}-1.2.3_p20240101-r3" for i in range(200)
+ ]
+ batches = list(BugQuery.package_list_any(packages).batches())
+ assert len(batches) > 1
+ assert [
+ v for b in batches for k, v in b.params() if k.startswith("v")
+ ] == packages
+
+ def test_base_length_shrinks_batches(self):
+ ids = list(range(900000, 902000))
+ wide = len(list(BugQuery.ids(ids).batches(base_length=0)))
+ narrow = len(list(BugQuery.ids(ids).batches(base_length=4000)))
+ assert narrow > wide
+
+ def test_widest_axis_is_chosen(self):
+ query = BugQuery.ids((1, 2)) & BugQuery.package_list_any(
+ f"=dev-libs/pkg{i}-1" for i in range(400)
+ )
+ for batch in query.batches():
+ # the narrow axis rides along in every batch
+ assert [v for k, v in batch.params() if k == "id"] == ["1", "2"]
+
+ def test_max_length_is_honoured(self):
+ query = BugQuery.ids(range(900000, 902000))
+ for batch in query.batches(max_length=500):
+ assert len(urllib.parse.urlencode(batch.params())) <= 500
diff --git a/tests/bugzilla/test_testing.py b/tests/bugzilla/test_testing.py
new file mode 100644
index 000000000..98fc41bf2
--- /dev/null
+++ b/tests/bugzilla/test_testing.py
@@ -0,0 +1,141 @@
+"""Coverage for the replay helpers other projects are meant to reuse."""
+
+import pytest
+
+from pkgcore.bugzilla import errors, transport
+from pkgcore.bugzilla.changes import NewBug
+from pkgcore.bugzilla.client import Bugzilla
+from pkgcore.bugzilla.enums import Component
+from pkgcore.bugzilla.testing import Cassette, response
+
+
+def new_bug(summary="cat/pkg-1: stablereq"):
+ return NewBug(
+ summary=summary, description="please", component=Component.STABILIZATION
+ )
+
+
+class TestCassette:
+ def test_expect_bugs(self):
+ cassette = Cassette().expect_bugs({"id": 1}, {"id": 2})
+ assert sorted(cassette.client().search()) == [1, 2]
+
+ def test_expect_created(self):
+ cassette = Cassette().expect_created(900123)
+ assert cassette.client().create(new_bug()) == 900123
+
+ def test_expect_changed(self):
+ cassette = Cassette().expect_changed(
+ 5, status={"added": "RESOLVED", "removed": "CONFIRMED"}
+ )
+ from pkgcore.bugzilla.changes import BugUpdate
+
+ changes = cassette.client().update(5, BugUpdate.resolve(comment="done"))
+ assert changes.changes["status"].added == ("RESOLVED",)
+
+ def test_expect_error(self):
+ cassette = Cassette().expect_error(101, "no such bug", status=404)
+ with pytest.raises(errors.BugzillaNotFound):
+ cassette.client().search()
+
+ def test_expect_whoami(self):
+ cassette = Cassette().expect_whoami(name="[email protected]")
+ assert cassette.client().whoami().name == "[email protected]"
+
+ def test_chaining(self):
+ cassette = Cassette().expect_whoami().expect_bugs({"id": 1})
+ client = cassette.client()
+ client.whoami()
+ assert list(client.search()) == [1]
+ cassette.assert_drained()
+
+ def test_recordings_may_be_passed_to_the_constructor(self):
+ cassette = Cassette(response({"bugs": [{"id": 3}]}))
+ assert list(cassette.client().search()) == [3]
+
+ def test_intercepts_plain_http(self):
+ cassette = Cassette(base_url="http://bugs.example.org").expect_bugs({"id": 1})
+ assert list(cassette.client().search()) == [1]
+ assert cassette.calls[0].url.startswith("http://")
+
+ def test_unexpected_request(self):
+ with pytest.raises(AssertionError, match="unexpected request"):
+ Cassette().client().search()
+
+ def test_assert_drained(self):
+ cassette = Cassette().expect_bugs()
+ with pytest.raises(AssertionError, match="unused recordings"):
+ cassette.assert_drained()
+
+ def test_anonymous_client(self):
+ cassette = Cassette(api_key=None).expect_bugs()
+ assert not cassette.client().search()
+ assert "Bugzilla_api_key" not in cassette.calls[0].params
+
+
+class TestCall:
+ def test_inspection(self):
+ cassette = Cassette().expect_created(1)
+ cassette.client().create(new_bug("cat/pkg-1: stablereq"))
+ (call,) = cassette.calls
+ assert call.method == "POST"
+ assert call.path == "/rest/bug"
+ assert call.body["summary"] == "cat/pkg-1: stablereq"
+ assert call.header("content-type") == "application/json"
+ assert call.header("Content-Type") == "application/json"
+ assert call.header("no-such-header") is None
+
+ def test_params_group_repeated_keys(self):
+ from pkgcore.bugzilla.query import BugQuery
+
+ cassette = Cassette().expect_bugs()
+ cassette.client().search(BugQuery.ids((1, 2, 3)))
+ assert cassette.calls[0].params["id"] == ["1", "2", "3"]
+
+
+class TestDynamicResponses:
+ def test_callable_body_sees_the_call(self):
+ cassette = Cassette().always(
+ response(lambda call: {"id": len(call.body["summary"])})
+ )
+ assert cassette.client().create(new_bug("abcde")) == 5
+
+ def test_creates_bugs_answers_every_filing(self):
+ cassette = Cassette().creates_bugs()
+ client = cassette.client()
+ assert [client.create(new_bug()) for _ in range(3)] == [1, 2, 3]
+
+ def test_fallback_only_applies_once_the_queue_is_empty(self):
+ cassette = Cassette().expect_created(100).creates_bugs(first=7)
+ client = cassette.client()
+ assert [client.create(new_bug()) for _ in range(2)] == [100, 7]
+
+
+class TestGlobalInstall:
+ """The path downstream CLI tests need, where the client is out of reach"""
+
+ def build_a_client_somewhere_unreachable(self):
+ return Bugzilla("some-key", base_url="https://bugs.example.org")
+
+ def test_context_manager_intercepts_clients_it_did_not_build(self):
+ with Cassette().expect_created(42) as cassette:
+ bug_id = self.build_a_client_somewhere_unreachable().create(new_bug())
+ assert bug_id == 42
+ assert cassette.calls[0].path == "/rest/bug"
+
+ def test_the_opener_is_restored_afterwards(self):
+ original = transport.build_opener
+ with Cassette().creates_bugs():
+ assert transport.build_opener is not original
+ assert transport.build_opener is original
+
+ def test_restored_even_when_the_body_raises(self):
+ original = transport.build_opener
+ with pytest.raises(ValueError), Cassette():
+ raise ValueError("boom")
+ assert transport.build_opener is original
+
+ def test_fixture_installs_by_default(self, bugzilla_cassette):
+ # the plugin fixture is already active, so no opener needs passing
+ bugzilla_cassette.expect_created(7)
+ assert self.build_a_client_somewhere_unreachable().create(new_bug()) == 7
diff --git a/tests/bugzilla/test_transport.py b/tests/bugzilla/test_transport.py
new file mode 100644
index 000000000..dfbc5969e
--- /dev/null
+++ b/tests/bugzilla/test_transport.py
@@ -0,0 +1,288 @@
+import urllib.error
+import urllib.request
+
+import pytest
+
+from pkgcore.bugzilla import errors
+from pkgcore.bugzilla.testing import API_KEY, Cassette, response
+from pkgcore.bugzilla.transport import (
+ USER_AGENT,
+ AuthMode,
+ UrllibTransport,
+ build_user_agent,
+ expect_list,
+ expect_object,
+ redact,
+)
+
+
+class TestRedact:
+ def test_removes_the_key(self):
+ url = "https://bugs.example.org/rest/bug?id=1&Bugzilla_api_key=secret"
+ assert "secret" not in redact(url)
+ assert "id=1" in redact(url)
+
+ def test_no_query(self):
+ assert redact("https://bugs.example.org/rest/bug") == (
+ "https://bugs.example.org/rest/bug"
+ )
+
+
+class TestRequests:
+ def test_get(self, cassette):
+ handler, transport = cassette(response({"bugs": []}))
+ assert transport.request("GET", "bug", params=(("id", "1"),)) == {"bugs": []}
+ (call,) = handler.calls
+ assert call.method == "GET"
+ assert call.path == "/rest/bug"
+ assert ("id", "1") in call.query
+ assert call.body is None
+
+ def test_leading_slash_is_tolerated(self, cassette):
+ handler, transport = cassette(response({}))
+ transport.request("GET", "/bug/1")
+ assert handler.calls[0].path == "/rest/bug/1"
+
+ def test_headers(self, cassette):
+ handler, transport = cassette(response({}))
+ transport.request("GET", "bug")
+ headers = handler.calls[0].headers
+ assert headers["Accept"] == "application/json"
+ assert headers["User-agent"] == USER_AGENT
+ assert "Content-type" not in headers
+
+ def test_custom_user_agent(self, cassette):
+ handler, transport = cassette(response({}), user_agent="gentoo/1.2")
+ transport.request("GET", "bug")
+ agent = handler.calls[0].header("user-agent")
+ assert agent == f"gentoo/1.2 {USER_AGENT}"
+
+ def test_put_body(self, cassette):
+ handler, transport = cassette(response({"bugs": []}))
+ transport.request("PUT", "bug/1", body={"ids": [1], "summary": "x"})
+ (call,) = handler.calls
+ assert call.method == "PUT"
+ assert call.headers["Content-type"] == "application/json"
+ assert call.body["ids"] == [1]
+
+ def test_post_body(self, cassette):
+ handler, transport = cassette(response({"id": 5}))
+ assert transport.request("POST", "bug", body={"summary": "x"}) == {"id": 5}
+ assert handler.calls[0].method == "POST"
+
+
+class TestAuth:
+ def test_key_in_query_for_reads(self, cassette):
+ handler, transport = cassette(response({}))
+ transport.request("GET", "bug")
+ assert ("Bugzilla_api_key", API_KEY) in handler.calls[0].query
+ assert "X-bugzilla-api-key" not in handler.calls[0].headers
+
+ def test_key_in_body_for_writes(self, cassette):
+ handler, transport = cassette(response({}))
+ transport.request("PUT", "bug/1", body={"ids": [1]})
+ (call,) = handler.calls
+ assert call.body["Bugzilla_api_key"] == API_KEY
+ assert "Bugzilla_api_key" not in dict(call.query)
+
+ def test_header_mode(self, cassette):
+ handler, transport = cassette(response({}), auth_mode=AuthMode.HEADER)
+ transport.request("GET", "bug")
+ (call,) = handler.calls
+ assert call.headers["X-bugzilla-api-key"] == API_KEY
+ assert "Bugzilla_api_key" not in dict(call.query)
+
+ def test_anonymous_reads(self, cassette):
+ handler, transport = cassette(response({"bugs": []}), api_key=None)
+ transport.request("GET", "bug")
+ assert "Bugzilla_api_key" not in dict(handler.calls[0].query)
+ assert not transport.authenticated
+
+ def test_anonymous_writes_fail_without_a_round_trip(self, cassette):
+ handler, transport = cassette(api_key=None)
+ with pytest.raises(errors.BugzillaAuthRequired, match="anonymous"):
+ transport.request("PUT", "bug/1", body={"ids": [1]})
+ assert handler.calls == []
+
+
+class TestUserAgent:
+ def test_default(self):
+ assert build_user_agent() == USER_AGENT
+
+ def test_client_token_comes_first(self):
+ assert build_user_agent("glibc/1.0") == f"glibc/1.0 {USER_AGENT}"
+
+ def test_blank_client_is_ignored(self):
+ assert build_user_agent("") == USER_AGENT
+
+
+class TestErrors:
+ def test_error_body_with_a_2xx_status(self, cassette):
+ # bugzilla does this, so the body has to win over the status
+ _, transport = cassette(
+ response({"error": True, "code": 101, "message": "Bug #1 does not exist."})
+ )
+ with pytest.raises(errors.BugzillaNotFound) as excinfo:
+ transport.request("GET", "bug/1")
+ assert excinfo.value.code == 101
+
+ def test_invalid_api_key_arrives_as_400(self, cassette):
+ _, transport = cassette(
+ response(
+ {"error": True, "code": 306, "message": "The API key is invalid."},
+ status=400,
+ )
+ )
+ with pytest.raises(errors.BugzillaAuthError) as excinfo:
+ transport.request("GET", "bug/1")
+ assert (excinfo.value.code, excinfo.value.status) == (306, 400)
+
+ def test_login_required_arrives_as_401(self, cassette):
+ _, transport = cassette(
+ response({"error": True, "code": 410, "message": "log in"}, status=401)
+ )
+ with pytest.raises(errors.BugzillaAuthError):
+ transport.request("GET", "whoami")
+
+ def test_permission_denied(self, cassette):
+ _, transport = cassette(
+ response({"error": True, "code": 102, "message": "nope"}, status=401)
+ )
+ with pytest.raises(errors.BugzillaPermissionDenied):
+ transport.request("GET", "bug/1")
+
+ def test_unknown_code_stays_generic(self, cassette):
+ _, transport = cassette(
+ response({"error": True, "code": 999, "message": "?"}, status=400)
+ )
+ with pytest.raises(errors.BugzillaResponseError) as excinfo:
+ transport.request("GET", "bug/1")
+ assert type(excinfo.value) is errors.BugzillaResponseError
+
+ def test_html_body(self, cassette):
+ _, transport = cassette(
+ response(raw=b"<html>blocked</html>", content_type="text/html")
+ )
+ with pytest.raises(errors.BugzillaProtocolError, match="isn't JSON"):
+ transport.request("GET", "bug/1")
+
+ def test_failing_status_without_an_error_body(self, cassette):
+ _, transport = cassette(response({"unexpected": True}, status=404))
+ with pytest.raises(errors.BugzillaNotFound):
+ transport.request("GET", "bug/1")
+
+ def test_api_key_is_redacted_from_messages(self, cassette):
+ _, transport = cassette(
+ response({"error": True, "code": 101, "message": "nope"})
+ )
+ with pytest.raises(errors.BugzillaNotFound) as excinfo:
+ transport.request("GET", "bug/1")
+ assert API_KEY not in excinfo.value.url
+ assert API_KEY not in str(excinfo.value)
+
+ def test_connection_error(self):
+ class Broken(urllib.request.HTTPSHandler):
+ def https_open(self, req):
+ raise urllib.error.URLError("connection reset by peer")
+
+ http_open = https_open
+
+ transport = UrllibTransport(
+ "https://bugs.example.org",
+ API_KEY,
+ retries=1,
+ opener=urllib.request.build_opener(Broken()),
+ )
+ with pytest.raises(errors.BugzillaConnectionError, match="reset by peer"):
+ transport.request("GET", "bug/1")
+
+
+class TestRetries:
+ @pytest.fixture(autouse=True)
+ def no_sleep(self, monkeypatch):
+ monkeypatch.setattr("pkgcore.bugzilla.transport.time.sleep", lambda _: None)
+
+ def test_reads_are_retried(self, cassette):
+ handler, transport = cassette(
+ response(raw=b"", status=503),
+ response(raw=b"", status=503),
+ response({"bugs": []}),
+ retries=3,
+ )
+ assert transport.request("GET", "bug") == {"bugs": []}
+ assert len(handler.calls) == 3
+
+ def test_retries_are_bounded(self, cassette):
+ handler, transport = cassette(
+ *(response(raw=b"", status=503) for _ in range(3)), retries=3
+ )
+ with pytest.raises(errors.BugzillaServerError):
+ transport.request("GET", "bug")
+ assert len(handler.calls) == 3
+
+ def test_writes_are_not_retried(self, cassette):
+ # a retried bug creation would file a duplicate
+ handler, transport = cassette(response(raw=b"", status=503), retries=3)
+ with pytest.raises(errors.BugzillaServerError):
+ transport.request("POST", "bug", body={"summary": "x"})
+ assert len(handler.calls) == 1
+
+ def test_writes_retry_when_asked(self, cassette):
+ handler, transport = cassette(
+ response(raw=b"", status=503),
+ response({"id": 5}),
+ retries=3,
+ retry_writes=True,
+ )
+ assert transport.request("POST", "bug", body={"summary": "x"}) == {"id": 5}
+ assert len(handler.calls) == 2
+
+ def test_client_errors_are_not_retried(self, cassette):
+ handler, transport = cassette(
+ response({"error": True, "code": 101, "message": "nope"}, status=404),
+ retries=3,
+ )
+ with pytest.raises(errors.BugzillaNotFound):
+ transport.request("GET", "bug/1")
+ assert len(handler.calls) == 1
+
+ def test_connection_errors_are_retried(self, monkeypatch):
+ attempts = []
+
+ class Flaky(urllib.request.HTTPSHandler):
+ def https_open(self, req):
+ attempts.append(req)
+ if len(attempts) < 3:
+ raise urllib.error.URLError("reset")
+ return Cassette().expect_bugs().opener.open(req)
+
+ http_open = https_open
+
+ transport = UrllibTransport(
+ "https://bugs.example.org",
+ API_KEY,
+ retries=3,
+ opener=urllib.request.build_opener(Flaky()),
+ )
+ assert transport.request("GET", "bug") == {"bugs": []}
+ assert len(attempts) == 3
+
+
+class TestShapeGuards:
+ def test_expect_object(self):
+ assert expect_object({"a": 1}, "ctx") == {"a": 1}
+
+ def test_expect_object_rejects_a_list(self):
+ with pytest.raises(errors.BugzillaSchemaError, match="expected an object"):
+ expect_object([], "ctx")
+
+ def test_expect_list(self):
+ assert expect_list({"bugs": [1]}, "bugs", "ctx") == [1]
+
+ def test_expect_list_missing_key(self):
+ with pytest.raises(errors.BugzillaSchemaError, match="'bugs' list"):
+ expect_list({}, "bugs", "ctx")
+
+ def test_expect_list_wrong_type(self):
+ with pytest.raises(errors.BugzillaSchemaError):
+ expect_list({"bugs": {}}, "bugs", "ctx")