# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Tests for :mod:`repomatic.github.unsubscribe`."""
from __future__ import annotations
import json
import re
from datetime import UTC, datetime, timedelta, timezone
from functools import partial
from pathlib import Path
from unittest.mock import patch
from urllib.parse import parse_qs
import pytest
import repomatic
from repomatic.github.gh import GhCommandError
from repomatic.github.unsubscribe import (
GRAPHQL_PAGE_SIZE,
NOTIFICATION_PAGE_SIZE,
THREADLESS_SEARCH_QUERY,
UNSUBSCRIBE_WORKFLOW,
DetailRow,
UnsubscribeResult,
_batch_subject_details,
_compute_cutoff,
_fetch_notification_threads,
_fetch_subject_details,
_format_link,
_get_authenticated_username,
_get_thread_details,
_graphql_unsubscribe,
_iter_closed_items,
_render_detail_table,
_unsubscribe_rest_thread,
render_report,
unsubscribe_threads,
)
from repomatic.tabular import ReportAction
from tests.conftest import patch_gh as shared_patch_gh
TYPE_CHECKING = False
if TYPE_CHECKING:
from typing import Any
_MODULE = "repomatic.github.unsubscribe"
# A subject URL well in the past: parses below any realistic cutoff.
STALE = "2020-01-01T00:00:00Z"
# A subject URL far in the future: parses above any realistic cutoff.
RECENT = "2099-01-01T00:00:00Z"
# A stable cutoff for report-rendering tests; its isoformat is asserted below.
CUTOFF = datetime(2026, 4, 16, tzinfo=UTC)
CUTOFF_ISO = "2026-04-16T00:00:00+00:00"
# -- Test helpers -------------------------------------------------------------
def _make_row(
*,
action=ReportAction.UNSUBSCRIBED,
html_url="https://example.com/apple/1",
number=1,
repo="fruits/apple",
title="Sunny afternoon",
updated_at=None,
) -> DetailRow:
"""Build a `DetailRow` with domain-neutral defaults."""
return DetailRow(
action=action,
html_url=html_url,
number=number,
repo=repo,
title=title,
updated_at=updated_at,
)
def _notification(
thread_id: str,
subject_url: str,
*,
repo="fruits/apple",
title="Sunny",
updated_at="",
subject_type="Issue",
) -> dict[str, object]:
"""Build one notification as `GET /notifications` returns it."""
return {
"id": thread_id,
"updated_at": updated_at,
"repository": {"full_name": repo},
"subject": {"type": subject_type, "url": subject_url, "title": title},
}
def _listing(*notifications: dict[str, object]) -> str:
"""One `--slurp` answer: a single page holding *notifications*."""
return json.dumps([list(notifications)])
def _closed_detail(
*,
state="closed",
updated_at=STALE,
number=1,
html_url="https://github.com/fruits/apple/issues/1",
) -> dict[str, object]:
"""Build a thread-detail dict as `_get_thread_details` returns it."""
return {
"state": state,
"updated_at": updated_at,
"html_url": html_url,
"number": number,
}
def _gql_item(
node_id: str,
*,
subscription="SUBSCRIBED",
number=1,
repo="fruits/apple",
title="Sunny",
updated_at=STALE,
url="https://github.com/fruits/apple/issues/1",
) -> dict[str, object]:
"""Build a GraphQL search node as `_iter_closed_items` yields it."""
return {
"id": node_id,
"number": number,
"title": title,
"repository": {"nameWithOwner": repo},
"updatedAt": updated_at,
"url": url,
"viewerSubscription": subscription,
}
_SUBJECT_ALIAS = re.compile(
r'(?P<alias>s\d+): repository\(owner: "(?P<owner>[^"]+)", name: "(?P<repo>[^"]+)"\)'
r"[^{]*\{\s*issueOrPullRequest\(number: (?P<number>\d+)\)"
)
"""Alias, coordinates and number, read back out of a batched subject query."""
_REST_TO_GRAPHQL_STATE = {"open": "OPEN", "closed": "CLOSED"}
"""The detail fixtures speak REST; the batched lookup answers in GraphQL."""
def _subject_batch_response(query, detail_map):
"""Answer a batched subject query from the REST-shaped detail fixtures.
Keeps one fixture vocabulary for both lookup routes: a test declares REST
details, and this renders whichever shape the code under test asked for.
An alias whose subject is not in *detail_map* answers `null`, which is how
GitHub reports a repository the token cannot see.
"""
data: dict[str, dict[str, Any]] = {}
for match in _SUBJECT_ALIAS.finditer(query):
detail = None
for kind in ("issues", "pulls"):
url = (
f"https://api.github.com/repos/{match['owner']}"
f"/{match['repo']}/{kind}/{match['number']}"
)
if url in detail_map:
detail = detail_map[url]
break
if detail is None:
data[match["alias"]] = {"issueOrPullRequest": None}
continue
state = detail.get("state", "")
data[match["alias"]] = {
"issueOrPullRequest": {
"__typename": "Issue",
"state": _REST_TO_GRAPHQL_STATE.get(state, state),
"updatedAt": detail.get("updated_at", ""),
"number": detail.get("number"),
"url": detail.get("html_url", ""),
}
}
return data
def _dispatch_gh(
*,
notifications=(),
details=None,
username="octocat",
fail_notifications=False,
fail_username=False,
fail_delete=frozenset(),
fail_patch=frozenset(),
fail_mutation=frozenset(),
fail_subject_batch=False,
):
"""Build a `run_gh_command` side effect that answers by argument shape.
Routes each gh call to a canned response so `unsubscribe_threads` runs
end-to-end without a network. `details` maps a subject URL to its detail
dict; an unlisted URL raises, modeling an inaccessible subject.
"""
detail_map = details or {}
def _run(args):
if args[:3] == ["api", "--paginate", "--slurp"]:
if fail_notifications:
raise RuntimeError("notifications unavailable")
return _listing(*notifications)
if args[:2] == ["api", "/user"]:
if fail_username:
raise RuntimeError("no user")
return f"{username}\n"
if args[:3] == ["api", "--method", "DELETE"]:
thread_id = args[3].split("/")[3]
if thread_id in fail_delete:
raise RuntimeError("delete failed")
return ""
if args[:3] == ["api", "--method", "PATCH"]:
thread_id = args[3].split("/")[-1]
if thread_id in fail_patch:
raise RuntimeError("patch failed")
return ""
if args[:2] == ["api", "graphql"]:
query = next(a for a in args if a.startswith("query="))
if "issueOrPullRequest" in query:
if fail_subject_batch:
raise RuntimeError("graphql unavailable")
return json.dumps({"data": _subject_batch_response(query, detail_map)})
node_id = next(a[len("id=") :] for a in args if a.startswith("id="))
if node_id in fail_mutation:
raise RuntimeError("mutation failed")
return json.dumps({"data": {"updateSubscription": None}})
# Remaining shape: a thread-detail lookup keyed by subject URL.
subject_url = args[1]
if subject_url not in detail_map:
raise RuntimeError("subject inaccessible")
return json.dumps(detail_map[subject_url])
return _run
patch_gh = partial(shared_patch_gh, _MODULE)
"""The conftest `patch_gh`, bound to this module's dispatch points."""
def _gh_calls_matching(mock_gh, prefix):
"""Return the argument lists of gh calls whose args start with *prefix*."""
return [
entry.args[0]
for entry in mock_gh.call_args_list
if entry.args[0][: len(prefix)] == prefix
]
# -- _compute_cutoff ----------------------------------------------------------
[docs]
@pytest.mark.parametrize(
("fixed_now", "months", "expected"),
[
pytest.param(
datetime(2026, 3, 31, 12, tzinfo=UTC),
1,
datetime(2026, 2, 28, 12, tzinfo=UTC),
id="clamp-to-feb-28",
),
pytest.param(
datetime(2024, 3, 31, 12, tzinfo=UTC),
1,
datetime(2024, 2, 29, 12, tzinfo=UTC),
id="clamp-to-leap-feb-29",
),
pytest.param(
datetime(2026, 7, 16, 9, tzinfo=UTC),
3,
datetime(2026, 4, 16, 9, tzinfo=UTC),
id="simple-three-months",
),
pytest.param(
datetime(2026, 2, 15, tzinfo=UTC),
3,
datetime(2025, 11, 15, tzinfo=UTC),
id="year-rollover",
),
pytest.param(
datetime(2026, 7, 16, 9, tzinfo=UTC),
12,
datetime(2025, 7, 16, 9, tzinfo=UTC),
id="full-year",
),
pytest.param(
datetime(2026, 1, 31, 6, tzinfo=UTC),
1,
datetime(2025, 12, 31, 6, tzinfo=UTC),
id="january-to-december",
),
],
)
def test_compute_cutoff(fixed_now, months, expected):
"""Subtracting months clamps the day and rolls the year over correctly."""
with patch(f"{_MODULE}.datetime") as mock_dt:
mock_dt.now.return_value = fixed_now
assert _compute_cutoff(months) == expected
# -- _format_link -------------------------------------------------------------
# -- _render_detail_table -----------------------------------------------------
[docs]
def test_render_detail_table_empty():
"""No rows renders to an empty string, not a headerless table."""
assert _render_detail_table([]) == ""
[docs]
def test_render_detail_table_exact():
"""Rows render as a headed markdown table with one line each."""
rows = [
_make_row(
action=ReportAction.UNSUBSCRIBED,
html_url="https://example.com/apple/7",
number=7,
repo="fruits/apple",
title="Sunny afternoon",
updated_at=None,
),
_make_row(
action=ReportAction.FAILED,
html_url="",
number=None,
repo="cities/paris",
title="Rainy morning",
updated_at=None,
),
]
expected_lines = [
"### \U0001f4dd Details",
"",
(
"| \U0001f4ac Title | \U0001f517 Link"
" | \U0001f550 Last activity | \u26a1 Action |"
),
"| --- | --- | --- | --- |",
(
"| Sunny afternoon | [`fruits/apple#7`](https://example.com/apple/7)"
" | - | \U0001f515 Unsubscribed |"
),
"| Rainy morning | cities/paris | - | \u26a0\ufe0f Failed |",
]
expected = "\n".join(expected_lines)
assert _render_detail_table(rows) == expected
[docs]
def test_render_detail_table_humanizes_timestamp():
"""A real timestamp renders a relative string, not the '-' placeholder."""
row = _make_row(updated_at=datetime(2020, 1, 1, tzinfo=UTC))
table = _render_detail_table([row], now=datetime(2020, 3, 1, tzinfo=UTC))
assert " | 2020-01-01 (2 months ago) | " in table
# -- render_report ------------------------------------------------------------
[docs]
def test_render_report_phase1_dry_run_summary():
"""Dry-run phase 1 counts candidate rows and marks the mode."""
result = UnsubscribeResult(dry_run=True, months=3)
result.phase1.cutoff = CUTOFF
result.phase1.rows = [_make_row(), _make_row()]
report = render_report(result)
assert "Unsubscribe report (dry-run)" in report
assert (
"\U0001f50d **Candidates found:** 2"
f" \u2014 cutoff: `{CUTOFF_ISO}`"
" (inactive for more than 3 months, dry-run)"
) in report
[docs]
def test_render_report_phase1_live_summary():
"""Live phase 1 reports unsubscribed and failed counts."""
result = UnsubscribeResult(dry_run=False, months=6)
result.phase1.cutoff = CUTOFF
result.phase1.unsubscribed = 5
result.phase1.failed = 1
report = render_report(result)
assert "Unsubscribe report (live)" in report
assert (
"\U0001f515 **Unsubscribed:** 5"
" | \u26a0\ufe0f **Failed:** 1"
f" \u2014 cutoff: `{CUTOFF_ISO}`"
" (inactive for more than 6 months)"
) in report
[docs]
@pytest.mark.parametrize(
("deferred", "expect_warning"),
[
pytest.param(12, True, id="cap-reached"),
pytest.param(0, False, id="backlog-cleared"),
],
)
def test_render_report_backlog_warning(deferred, expect_warning):
"""The warning fires on work the cap deferred, and on nothing else.
Inspection is not capped, so no thread goes uninspected: what a run leaves
behind is eligible threads it declined to act on. That is a remainder a
higher cap really does clear, which is what the warning advises.
"""
result = UnsubscribeResult(dry_run=False, months=3)
p1 = result.phase1
p1.cutoff = CUTOFF
p1.max_unsubscribes = 200
p1.total = 100
p1.deferred = deferred
p1.oldest_updated = datetime(2026, 5, 1, tzinfo=UTC)
report = render_report(result)
assert ("eligible threads were left alone" in report) is expect_warning
def _backlog_report(repo_url: str | None = None) -> str:
"""A report whose cap deferred a dozen eligible threads."""
result = UnsubscribeResult(dry_run=False, months=3)
p1 = result.phase1
p1.cutoff = CUTOFF
p1.max_unsubscribes = 200
p1.total = 100
p1.deferred = 12
p1.oldest_updated = datetime(2026, 5, 1, tzinfo=UTC)
return render_report(result, repo_url)
[docs]
def test_render_report_backlog_warning_links_the_workflow():
"""The manual-run advice links the workflow whose run carries the report."""
url = "https://example.com/orchard/papaya"
assert (
f"[run it manually]({url}/actions/workflows/{UNSUBSCRIBE_WORKFLOW})"
in _backlog_report(url)
)
# No repository in hand: the advice stays, the link goes.
assert "run it manually with a higher cap" in _backlog_report()
[docs]
def test_backlog_warning_breaks_only_between_sentences():
"""A job summary renders every newline as a line break.
Wrapping the prose to a source column therefore leaks those wraps into the
rendered warning, which is what made this one read as randomly broken. Each
line has to end a sentence instead.
"""
warning = [
line
for line in _backlog_report().splitlines()
if line.startswith(">") and line != "> [!WARNING]"
]
assert warning
assert all(line.endswith(".") for line in warning), warning
[docs]
def test_unsubscribe_workflow_constant_names_a_bundled_workflow():
"""The link is only worth rendering while repomatic ships that workflow."""
bundled = Path(repomatic.__file__).parent / "data" / UNSUBSCRIBE_WORKFLOW
assert bundled.is_file()
[docs]
def test_render_report_phase2_skipped():
"""A skipped phase 2 renders its reason inside a warning admonition."""
result = UnsubscribeResult(dry_run=True, months=3)
result.phase2.skipped = True
result.phase2.skip_reason = "Fine-grained PAT unsupported."
report = render_report(result)
assert "Threadless subscriptions (dry-run)" in report
assert "> [!WARNING]" in report
assert "Fine-grained PAT unsupported." in report
[docs]
def test_render_report_phase2_content():
"""An active phase 2 renders its summary, search table, and query."""
result = UnsubscribeResult(dry_run=True, months=3)
p2 = result.phase2
p2.cutoff = CUTOFF
p2.max_unsubscribes = 200
p2.total = 3
p2.not_subscribed = 1
p2.search_query = "involves:fruitbot is:closed updated:<2026-04-16"
p2.rows = [_make_row(action=ReportAction.DRY_RUN)]
report = render_report(result)
assert "Search details" in report
assert "involves:fruitbot is:closed updated:<2026-04-16" in report
assert (
"\U0001f50d **Candidates found:** 1"
f" \u2014 cutoff: `{CUTOFF_ISO}`"
" (inactive for more than 3 months, dry-run)"
) in report
[docs]
def test_render_report_empty_result():
"""A default result renders both phases plus the attribution footer."""
report = render_report(UnsubscribeResult())
assert "Unsubscribe report (live)" in report
assert "Threadless subscriptions (live)" in report
assert "Generated with" in report
# -- _fetch_notification_threads ----------------------------------------------
[docs]
def test_fetch_notification_threads_sorts_oldest_first():
"""The pool comes back oldest first, whatever order the API used.
The order is imposed here, not inherited: the endpoint calls itself sorted
by most recent update, and a probe against 1637 real threads found the
response ordered by neither `updated_at` nor id. The arrival order below is
deliberately none of those.
"""
answer = _listing(
_notification("t0", "u0", updated_at="2026-08-01T00:00:00Z"),
_notification("t1", "u1", updated_at="2026-06-01T00:00:00Z"),
_notification("t2", "u2", updated_at="2026-07-01T00:00:00Z"),
)
with patch_gh(return_value=answer):
threads = _fetch_notification_threads(CUTOFF)
assert [t["id"] for t in threads] == ["t1", "t2", "t0"]
[docs]
def test_fetch_notification_threads_keeps_issues_and_pull_requests():
"""Only the subject types a stale check applies to become candidates."""
answer = _listing(
_notification("t0", "u0", subject_type="Issue"),
_notification("t1", "u1", subject_type="Release"),
_notification("t2", "u2", subject_type="PullRequest"),
)
with patch_gh(return_value=answer):
threads = _fetch_notification_threads(CUTOFF)
assert [t["id"] for t in threads] == ["t0", "t2"]
assert threads[0] == {
"id": "t0",
"updated_at": "",
"repo": "fruits/apple",
"subject_type": "Issue",
"subject_url": "u0",
"title": "Sunny",
}
[docs]
def test_fetch_notification_threads_raises_on_a_gh_failure():
"""A gh failure surfaces, for the phase to report it skipped."""
with (
patch_gh(side_effect=RuntimeError("boom")),
pytest.raises(RuntimeError, match="boom"),
):
_fetch_notification_threads(CUTOFF)
[docs]
def test_fetch_notification_threads_query_args():
"""The fetch reads every page, gated on the cutoff."""
with patch_gh(return_value="[]") as mock_gh:
_fetch_notification_threads(CUTOFF)
args = mock_gh.call_args.args[0]
assert args[:3] == ["api", "--paginate", "--slurp"]
endpoint, _, query = args[3].partition("?")
assert endpoint == "/notifications"
params = parse_qs(query)
assert params["all"] == ["true"]
assert params["before"] == ["2026-04-16T00:00:00Z"]
assert params["per_page"] == [str(NOTIFICATION_PAGE_SIZE)]
[docs]
def test_fetch_notification_threads_sends_the_cutoff_in_utc():
"""A cutoff from another zone must not stamp its wall clock with a `Z`.
The suffix is a claim about the instant, so formatting without converting
first would shift the window by the offset and silently fetch the wrong
threads.
"""
elsewhere = timezone(timedelta(hours=5))
with patch_gh(return_value="[]") as mock_gh:
_fetch_notification_threads(datetime(2026, 4, 16, 2, tzinfo=elsewhere))
params = parse_qs(mock_gh.call_args.args[0][3].partition("?")[2])
assert params["before"] == ["2026-04-15T21:00:00Z"]
# -- _get_thread_details ------------------------------------------------------
[docs]
def test_get_thread_details_success():
"""A well-formed response is parsed and returned verbatim."""
detail = _closed_detail()
with patch_gh(return_value=json.dumps(detail)):
assert _get_thread_details("https://api.github.com/x") == detail
[docs]
def test_get_thread_details_inaccessible_returns_none():
"""A gh failure (inaccessible subject) yields None."""
with patch_gh(side_effect=RuntimeError("404")):
assert _get_thread_details("https://api.github.com/x") is None
[docs]
def test_get_thread_details_non_object_returns_none():
"""A JSON answer that is not an object yields None, like a malformed one."""
with patch_gh(return_value='["closed"]'):
assert _get_thread_details("https://api.github.com/x") is None
# -- _unsubscribe_rest_thread -------------------------------------------------
[docs]
def test_unsubscribe_rest_thread_success():
"""A successful unsubscribe deletes the subscription then marks it read."""
with patch_gh() as mock_gh:
assert _unsubscribe_rest_thread("42") is True
assert mock_gh.call_args_list[0].args[0] == [
"api",
"--method",
"DELETE",
"/notifications/threads/42/subscription",
]
assert mock_gh.call_args_list[1].args[0] == [
"api",
"--method",
"PATCH",
"/notifications/threads/42",
]
[docs]
def test_unsubscribe_rest_thread_delete_failure():
"""A failed DELETE returns False and never attempts the PATCH."""
with patch_gh(side_effect=RuntimeError("no")) as mock_gh:
assert _unsubscribe_rest_thread("42") is False
assert mock_gh.call_count == 1
[docs]
def test_unsubscribe_rest_thread_patch_failure():
"""A failed PATCH (after a successful DELETE) returns False."""
with patch_gh(
side_effect=[None, RuntimeError("no")],
) as mock_gh:
assert _unsubscribe_rest_thread("42") is False
assert mock_gh.call_count == 2
# -- _graphql_unsubscribe -----------------------------------------------------
[docs]
def test_graphql_unsubscribe_success():
"""A successful mutation passes the node id and query, returning True."""
with patch_gh(return_value='{"data": {"updateSubscription": null}}') as mock_gh:
assert _graphql_unsubscribe("node-1") is True
args = mock_gh.call_args.args[0]
assert args[:2] == ["api", "graphql"]
assert "id=node-1" in args
assert any(a.startswith("query=") for a in args)
[docs]
def test_graphql_unsubscribe_failure():
"""A failed mutation returns False."""
with patch_gh(side_effect=RuntimeError("no")):
assert _graphql_unsubscribe("node-1") is False
# -- _get_authenticated_username ----------------------------------------------
[docs]
def test_get_authenticated_username_strips():
"""The login is trimmed of trailing whitespace."""
with patch_gh(return_value="octocat\n") as mock_gh:
assert _get_authenticated_username() == "octocat"
assert mock_gh.call_args.args[0] == ["api", "/user", "--jq", ".login"]
[docs]
def test_get_authenticated_username_propagates_error():
"""A gh failure propagates so the caller can skip phase 2."""
with (
patch_gh(side_effect=RuntimeError("boom")),
pytest.raises(RuntimeError, match="boom"),
):
_get_authenticated_username()
# -- _iter_closed_items -------------------------------------------------------
[docs]
def test_iter_closed_items_delegates_to_paginator():
"""The search wraps the shared paginator, bounded by GitHub's own ceiling."""
with patch(f"{_MODULE}.iter_graphql_nodes") as mock_iter:
mock_iter.return_value = iter([{"id": "n1"}])
_iter_closed_items("involves:fruitbot is:closed")
mock_iter.assert_called_once()
pos_args = mock_iter.call_args.args
kw_args = mock_iter.call_args.kwargs
assert pos_args[0] == THREADLESS_SEARCH_QUERY
assert pos_args[1] == ("search",)
assert pos_args[2] == {"searchQuery": "involves:fruitbot is:closed"}
assert kw_args["page_size_var"] == "pageSize"
assert kw_args["page_size"] == GRAPHQL_PAGE_SIZE
# -- unsubscribe_threads: phase 1 (REST) --------------------------------------
[docs]
def test_phase1_live_unsubscribes_closed_stale():
"""A closed, stale thread is unsubscribed via DELETE + PATCH in live mode."""
url = "https://api.github.com/repos/fruits/apple/issues/1"
run = _dispatch_gh(
notifications=[_notification("t1", url)],
details={url: _closed_detail()},
)
with (
patch_gh(side_effect=run) as mock_gh,
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=False)
p1 = result.phase1
assert p1.total == 1
assert p1.unsubscribed == 1
assert p1.failed == 0
assert p1.rows[0].action == ReportAction.UNSUBSCRIBED
assert len(_gh_calls_matching(mock_gh, ["api", "--method", "DELETE"])) == 1
assert len(_gh_calls_matching(mock_gh, ["api", "--method", "PATCH"])) == 1
[docs]
def test_phase1_dry_run_records_candidate_without_mutations():
"""Dry-run records a candidate row but issues no DELETE or PATCH."""
url = "https://api.github.com/repos/fruits/apple/issues/1"
run = _dispatch_gh(
notifications=[_notification("t1", url)],
details={url: _closed_detail()},
)
with (
patch_gh(side_effect=run) as mock_gh,
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
assert result.phase1.unsubscribed == 1
assert result.phase1.rows[0].action == ReportAction.DRY_RUN
assert _gh_calls_matching(mock_gh, ["api", "--method", "DELETE"]) == []
assert _gh_calls_matching(mock_gh, ["api", "--method", "PATCH"]) == []
[docs]
@pytest.mark.parametrize(
("detail", "counter"),
[
pytest.param(
_closed_detail(state="open", updated_at=STALE),
"skipped_open",
id="open",
),
pytest.param(
_closed_detail(state="closed", updated_at=RECENT),
"skipped_recent",
id="recent",
),
pytest.param(
_closed_detail(state="unknown", updated_at=STALE),
"skipped_unknown",
id="unknown-state",
),
pytest.param(
_closed_detail(state="closed", updated_at="not-a-date"),
"skipped_unknown",
id="bad-timestamp",
),
],
)
def test_phase1_skip_scenarios(detail, counter):
"""Open, recent, unknown-state, and unparsable threads are each skipped."""
url = "https://api.github.com/repos/fruits/apple/issues/1"
run = _dispatch_gh(notifications=[_notification("t1", url)], details={url: detail})
with (
patch_gh(side_effect=run) as mock_gh,
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=False)
assert getattr(result.phase1, counter) == 1
assert result.phase1.unsubscribed == 0
assert _gh_calls_matching(mock_gh, ["api", "--method", "DELETE"]) == []
[docs]
def test_phase1_inaccessible_subject_skipped():
"""An inaccessible subject (no details) counts as unknown, not a failure."""
url = "https://api.github.com/repos/fruits/apple/issues/1"
run = _dispatch_gh(notifications=[_notification("t1", url)], details={})
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=False)
assert result.phase1.skipped_unknown == 1
assert result.phase1.unsubscribed == 0
[docs]
def test_phase1_delete_failure_records_failed():
"""A failed DELETE records a FAILED row and skips the PATCH."""
url = "https://api.github.com/repos/fruits/apple/issues/1"
run = _dispatch_gh(
notifications=[_notification("t1", url)],
details={url: _closed_detail()},
fail_delete={"t1"},
)
with (
patch_gh(side_effect=run) as mock_gh,
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=False)
p1 = result.phase1
assert p1.failed == 1
assert p1.unsubscribed == 0
assert p1.rows[0].action == ReportAction.FAILED
assert len(_gh_calls_matching(mock_gh, ["api", "--method", "DELETE"])) == 1
assert _gh_calls_matching(mock_gh, ["api", "--method", "PATCH"]) == []
[docs]
def test_phase1_cap_defers_but_reports_total():
"""The cap bounds the unsubscribes, yet the full total is reported."""
base = "https://api.github.com/repos/fruits/apple/issues"
notifications = [_notification(f"t{i}", f"{base}/{i}") for i in range(5)]
details = {f"{base}/{i}": _closed_detail(number=i) for i in range(5)}
run = _dispatch_gh(notifications=notifications, details=details)
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=2, dry_run=True)
# Every candidate is inspected; the cap only stops the acting.
assert result.phase1.total == 5
assert result.phase1.unsubscribed == 2
assert result.phase1.deferred == 3
[docs]
def test_phase1_fetch_failure_skips_phase():
"""A fetch failure marks phase 1 skipped while phase 2 still runs.
The report then says the phase did not run, rather than reporting an empty
pool as a clean one.
"""
run = _dispatch_gh(fail_notifications=True)
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
assert result.phase1.skipped is True
assert "notification threads" in result.phase1.skip_reason
assert result.phase1.total == 0
assert result.phase2.skipped is False
report = render_report(result)
assert report.count("> [!WARNING]") == 1
assert "Failed to fetch notification threads." in report
# -- unsubscribe_threads: phase 2 (GraphQL) -----------------------------------
[docs]
def test_phase2_filters_non_subscribed_items():
"""Only items with viewerSubscription SUBSCRIBED are acted on."""
items = [
_gql_item("n1", subscription="SUBSCRIBED"),
_gql_item("n2", subscription="IGNORED"),
_gql_item("n3", subscription="UNSUBSCRIBED"),
]
run = _dispatch_gh()
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes", return_value=items),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
p2 = result.phase2
assert p2.total == 3
assert p2.not_subscribed == 2
assert p2.unsubscribed == 1
assert p2.skipped is False
[docs]
def test_phase2_skips_items_active_since_cutoff():
"""Fresh or timestamp-less items are skipped despite the search filter.
The search query's day-granular `updated:<` filter is served by GitHub's
search index, which can lag; the client-side re-check mirrors phase 1.
"""
items = [
_gql_item("n1", updated_at=STALE),
_gql_item("n2", updated_at=RECENT),
_gql_item("n3", updated_at="not-a-timestamp"),
]
run = _dispatch_gh()
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes", return_value=items),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
p2 = result.phase2
assert p2.total == 3
assert p2.skipped_recent == 2
assert p2.unsubscribed == 1
[docs]
def test_phase2_dry_run_no_mutation_calls():
"""Dry-run records a candidate row but issues no mutation."""
items = [_gql_item("n1", subscription="SUBSCRIBED")]
run = _dispatch_gh()
with (
patch_gh(side_effect=run) as mock_gh,
patch(f"{_MODULE}.iter_graphql_nodes", return_value=items),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
assert result.phase2.unsubscribed == 1
assert result.phase2.rows[0].action == ReportAction.DRY_RUN
assert _gh_calls_matching(mock_gh, ["api", "graphql"]) == []
[docs]
def test_phase2_live_unsubscribes_subscribed():
"""Live mode issues the mutation for a subscribed item."""
items = [_gql_item("n1", subscription="SUBSCRIBED")]
run = _dispatch_gh()
with (
patch_gh(side_effect=run) as mock_gh,
patch(f"{_MODULE}.iter_graphql_nodes", return_value=items),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=False)
assert result.phase2.unsubscribed == 1
assert result.phase2.rows[0].action == ReportAction.UNSUBSCRIBED
gql_calls = _gh_calls_matching(mock_gh, ["api", "graphql"])
assert len(gql_calls) == 1
assert "id=n1" in gql_calls[0]
[docs]
def test_phase2_mutation_failure_records_failed():
"""A failed mutation records a FAILED row."""
items = [_gql_item("n1", subscription="SUBSCRIBED")]
run = _dispatch_gh(fail_mutation={"n1"})
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes", return_value=items),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=False)
assert result.phase2.failed == 1
assert result.phase2.unsubscribed == 0
assert result.phase2.rows[0].action == ReportAction.FAILED
[docs]
def test_phase2_username_failure_skips_search():
"""A username lookup failure skips phase 2 before any search."""
run = _dispatch_gh(fail_username=True)
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes") as mock_iter,
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
assert result.phase2.skipped is True
assert "authenticated username" in result.phase2.skip_reason
mock_iter.assert_not_called()
[docs]
def test_phase2_search_failure_skips():
"""A GraphQL search failure marks phase 2 skipped with a reason."""
run = _dispatch_gh()
with (
patch_gh(side_effect=run),
patch(
f"{_MODULE}.iter_graphql_nodes",
side_effect=RuntimeError("no graphql"),
),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
assert result.phase2.skipped is True
assert "GraphQL search" in result.phase2.skip_reason
[docs]
def test_phase2_search_query_built():
"""The search query embeds the username and the computed cutoff date."""
with patch(f"{_MODULE}.datetime") as mock_dt:
mock_dt.now.return_value = datetime(2026, 7, 16, tzinfo=UTC)
run = _dispatch_gh(username="fruitbot")
with (
patch_gh(side_effect=run),
patch(f"{_MODULE}.iter_graphql_nodes", return_value=[]),
):
result = unsubscribe_threads(months=3, max_unsubscribes=200, dry_run=True)
assert result.phase2.search_query == (
"involves:fruitbot is:closed updated:<2026-04-16"
)
# --- Batched subject lookup ---
[docs]
def test_batch_subject_details_maps_graphql_onto_the_rest_vocabulary():
"""One round trip answers many subjects, in the spelling the phase reads.
REST calls a merged pull request `closed`; GraphQL distinguishes `MERGED`.
The phase compares one vocabulary, so the batch collapses the two.
"""
urls = [
"https://api.github.com/repos/fruits/apple/issues/1",
"https://api.github.com/repos/fruits/apple/pulls/2",
"https://api.github.com/repos/cities/lisbon/issues/3",
]
response = {
"s0": {
"issueOrPullRequest": {
"state": "OPEN",
"updatedAt": STALE,
"number": 1,
"url": "https://github.com/fruits/apple/issues/1",
}
},
"s1": {
"issueOrPullRequest": {
"state": "MERGED",
"updatedAt": STALE,
"number": 2,
"url": "https://github.com/fruits/apple/pull/2",
}
},
# A repository the token cannot see answers null, not an error.
"s2": {"issueOrPullRequest": None},
}
with patch(f"{_MODULE}.gh_graphql", return_value=response) as mock_graphql:
details = _batch_subject_details(urls)
assert mock_graphql.call_count == 1
assert mock_graphql.call_args.kwargs == {"partial": True}
assert details == {
urls[0]: {
"state": "open",
"updated_at": STALE,
"html_url": "https://github.com/fruits/apple/issues/1",
"number": 1,
},
urls[1]: {
"state": "closed",
"updated_at": STALE,
"html_url": "https://github.com/fruits/apple/pull/2",
"number": 2,
},
# The null answer is final: no REST call would answer differently.
urls[2]: None,
}
[docs]
def test_fetch_subject_details_falls_back_to_one_call_each():
"""A failed batch falls back to one REST call per subject, losing no result."""
urls = ["https://api.github.com/repos/fruits/apple/issues/1"]
detail = _closed_detail()
with (
patch(f"{_MODULE}.gh_graphql", side_effect=RuntimeError("graphql down")),
patch(f"{_MODULE}._get_thread_details", return_value=detail) as mock_single,
):
details = _fetch_subject_details(urls)
assert details == {urls[0]: detail}
assert mock_single.call_count == 1
[docs]
def test_fetch_subject_details_falls_back_for_an_unparsable_url():
"""A subject URL the pattern cannot read still gets resolved, one call."""
urls = ["https://api.github.com/repos/fruits/apple/discussions/1"]
detail = _closed_detail()
with (
patch(f"{_MODULE}.gh_graphql", return_value={}) as mock_graphql,
patch(f"{_MODULE}._get_thread_details", return_value=detail) as mock_single,
):
details = _fetch_subject_details(urls)
assert details == {urls[0]: detail}
assert mock_graphql.call_count == 0
assert mock_single.call_count == 1
[docs]
def test_fetch_subject_details_keeps_a_partial_answer():
"""A query failing in part keeps its answers, and re-reads nothing.
`gh` exits non-zero when any subject of the batch errors, and its standard
output still holds the `data` of the others. Reading that answer spares one
REST call per subject of the batch.
"""
urls = [
"https://api.github.com/repos/fruits/apple/issues/1",
"https://api.github.com/repos/fruits/hidden/issues/2",
]
answer = {
"data": {
"s0": {
"issueOrPullRequest": {
"state": "CLOSED",
"updatedAt": STALE,
"number": 1,
"url": "https://github.com/fruits/apple/issues/1",
}
},
"s1": None,
},
"errors": [{"message": "Could not resolve to a Repository"}],
}
with (
patch(
"repomatic.github.gh.run_gh_command",
side_effect=GhCommandError("GraphQL: partial", stdout=json.dumps(answer)),
),
patch(f"{_MODULE}._get_thread_details") as mock_single,
):
details = _fetch_subject_details(urls)
assert list(details) == [urls[0]]
assert details[urls[0]]["state"] == "closed"
mock_single.assert_not_called()
[docs]
def test_phase1_batches_every_subject_into_one_round_trip():
"""The whole pool costs one request, not one per thread."""
root = "https://api.github.com/repos/fruits/apple/issues"
notifications = [_notification(f"t{n}", f"{root}/{n}") for n in range(3)]
details = {f"{root}/{n}": _closed_detail(number=n) for n in range(3)}
with patch_gh(
side_effect=_dispatch_gh(notifications=notifications, details=details)
) as mock_gh:
result = unsubscribe_threads(months=3, max_unsubscribes=10, dry_run=True)
assert result.phase1.total == 3
graphql_calls = [
call
for call in mock_gh.call_args_list
if call.args[0][:2] == ["api", "graphql"]
and any("issueOrPullRequest" in arg for arg in call.args[0])
]
assert len(graphql_calls) == 1