# 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 logging
import re
from datetime import datetime, timedelta, timezone
from functools import partial
from pathlib import Path
from unittest.mock import patch
import pytest
import repomatic
from repomatic.github.actions import ReportAction
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,
_validate_notifications_token,
render_report,
unsubscribe_threads,
)
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=timezone.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 _thread_line(
thread_id: str,
subject_url: str,
*,
repo="fruits/apple",
title="Sunny",
) -> str:
"""Render one JSON line as `_fetch_notification_threads` receives it."""
return json.dumps({
"id": thread_id,
"repo": repo,
"subject_type": "Issue",
"subject_url": subject_url,
"title": title,
})
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[:4] == ["api", "--method", "GET", "/notifications"]:
if fail_notifications:
raise RuntimeError("notifications unavailable")
return 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 "{}"
# 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=timezone.utc),
1,
datetime(2026, 2, 28, 12, tzinfo=timezone.utc),
id="clamp-to-feb-28",
),
pytest.param(
datetime(2024, 3, 31, 12, tzinfo=timezone.utc),
1,
datetime(2024, 2, 29, 12, tzinfo=timezone.utc),
id="clamp-to-leap-feb-29",
),
pytest.param(
datetime(2026, 7, 16, 9, tzinfo=timezone.utc),
3,
datetime(2026, 4, 16, 9, tzinfo=timezone.utc),
id="simple-three-months",
),
pytest.param(
datetime(2026, 2, 15, tzinfo=timezone.utc),
3,
datetime(2025, 11, 15, tzinfo=timezone.utc),
id="year-rollover",
),
pytest.param(
datetime(2026, 7, 16, 9, tzinfo=timezone.utc),
12,
datetime(2025, 7, 16, 9, tzinfo=timezone.utc),
id="full-year",
),
pytest.param(
datetime(2026, 1, 31, 6, tzinfo=timezone.utc),
1,
datetime(2025, 12, 31, 6, tzinfo=timezone.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=timezone.utc))
table = _render_detail_table([row])
assert " | - | " not in table
assert "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.threads_unsubscribed = 5
result.phase1.threads_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(
("threads_deferred", "expect_warning"),
[
pytest.param(12, True, id="cap-reached"),
pytest.param(0, False, id="backlog-cleared"),
],
)
def test_render_report_backlog_warning(threads_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.threads_total = 100
p1.threads_deferred = threads_deferred
p1.oldest_updated = datetime(2026, 5, 1, tzinfo=timezone.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.threads_total = 100
p1.threads_deferred = 12
p1.oldest_updated = datetime(2026, 5, 1, tzinfo=timezone.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.items_total = 3
p2.items_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.
"""
lines = [
json.dumps({
"id": "t0",
"subject_url": "u0",
"updated_at": "2026-08-01T00:00:00Z",
}),
json.dumps({
"id": "t1",
"subject_url": "u1",
"updated_at": "2026-06-01T00:00:00Z",
}),
json.dumps({
"id": "t2",
"subject_url": "u2",
"updated_at": "2026-07-01T00:00:00Z",
}),
]
with patch_gh(return_value="\n".join(lines)):
threads = _fetch_notification_threads(CUTOFF)
assert [t["id"] for t in threads] == ["t1", "t2", "t0"]
[docs]
def test_fetch_notification_threads_gh_failure_returns_empty():
"""A gh failure degrades to an empty pool."""
with patch_gh(side_effect=RuntimeError("boom")):
assert _fetch_notification_threads(CUTOFF) == []
[docs]
def test_fetch_notification_threads_query_args():
"""The fetch requests both subject types, paginated, 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[:4] == ["api", "--method", "GET", "/notifications"]
assert "--paginate" in args
assert "all=true" in args
assert "before=2026-04-16T00:00:00Z" in args
assert f"per_page={NOTIFICATION_PAGE_SIZE}" in args
jq_filter = args[args.index("--jq") + 1]
# Without it the pool cannot be ordered: the response is not.
assert "updated_at" in jq_filter
assert '"Issue"' in jq_filter
assert '"PullRequest"' in jq_filter
[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))
assert "before=2026-04-15T21:00:00Z" in mock_gh.call_args.args[0]
# -- _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
# -- _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() 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
# No node budget: the cap counts unsubscribes, and GitHub's search stops at
# a thousand results on its own.
assert "max_nodes" not in kw_args
# -- _validate_notifications_token --------------------------------------------
[docs]
def test_validate_notifications_token_exact_scope(caplog):
"""A token scoped to exactly 'notifications' passes without a warning."""
with (
patch(
f"{_MODULE}.validate_classic_pat_scope",
return_value=["notifications"],
),
caplog.at_level(logging.WARNING),
):
_validate_notifications_token()
assert "more scopes than needed" not in caplog.text
[docs]
def test_validate_notifications_token_propagates():
"""A validation failure propagates to the CLI entry point."""
with (
patch(
f"{_MODULE}.validate_classic_pat_scope",
side_effect=RuntimeError("bad scope"),
),
pytest.raises(RuntimeError, match="bad scope"),
):
_validate_notifications_token()
# -- 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=_thread_line("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.threads_total == 1
assert p1.threads_unsubscribed == 1
assert p1.threads_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=_thread_line("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.threads_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),
"threads_skipped_open",
id="open",
),
pytest.param(
_closed_detail(state="closed", updated_at=RECENT),
"threads_skipped_recent",
id="recent",
),
pytest.param(
_closed_detail(state="unknown", updated_at=STALE),
"threads_skipped_unknown",
id="unknown-state",
),
pytest.param(
_closed_detail(state="closed", updated_at="not-a-date"),
"threads_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=_thread_line("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.threads_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=_thread_line("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.threads_skipped_unknown == 1
assert result.phase1.threads_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=_thread_line("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.threads_failed == 1
assert p1.threads_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"
lines = "\n".join(_thread_line(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=lines, 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.threads_total == 5
assert result.phase1.threads_unsubscribed == 2
assert result.phase1.threads_deferred == 3
[docs]
def test_phase1_fetch_failure_skips_phase():
"""A fetch failure leaves phase 1 empty while phase 2 still runs."""
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.threads_total == 0
assert result.phase1.threads_unsubscribed == 0
assert result.phase2.skipped is False
# -- 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(notifications="")
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.items_total == 3
assert p2.items_not_subscribed == 2
assert p2.items_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(notifications="")
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.items_total == 3
assert p2.items_skipped_recent == 2
assert p2.items_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(notifications="")
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.items_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(notifications="")
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.items_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(notifications="", 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.items_failed == 1
assert result.phase2.items_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(notifications="", 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(notifications="")
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=timezone.utc)
run = _dispatch_gh(notifications="", 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 details[urls[0]]["state"] == "open"
assert details[urls[1]]["state"] == "closed"
assert details[urls[1]]["html_url"] == "https://github.com/fruits/apple/pull/2"
assert urls[2] not in details
[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_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"
threads = [_thread_line(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="\n".join(threads), details=details)
) as mock_gh:
result = unsubscribe_threads(months=3, max_unsubscribes=10, dry_run=True)
assert result.phase1.threads_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