Raise package test coverage to 98% (#878)

tests/test_cla.py

@@ -155,7 +155,7 @@ def test_persist_honors_retry_after_for_transient_write(monkeypatch):
 
 @pytest.mark.parametrize(
     ("statuses", "message"),
-    [([502, 502, 502, 502], "502 Bad Gateway"), ([502, 502, 502, 409], "409 Conflict")],
+    [([502, 502, 502, 502], "502 Bad Gateway"), ([502, 502, 502, 409], "409 Conflict"), ([422], "422 Unprocessable")],
 )
 def test_persist_surfaces_final_exhausted_error(monkeypatch, statuses, message):
     """Preserve the final HTTP error when mixed write retries are exhausted."""
@@ -168,6 +168,7 @@ def test_persist_surfaces_final_exhausted_error(monkeypatch, statuses, message):
 
     with pytest.raises(RuntimeError, match=message):
         cla._persist(store, [{"name": "new", "id": 2}], source, 7)
+    assert store.put.call_count == len(statuses)
 
 
 def test_run_records_exact_sentence_and_updates_legacy_comment():
@@ -340,3 +341,68 @@ def test_rerun_leaves_queued_exact_head_to_complete(monkeypatch):
     cla._rerun_pr_check(source, 7)
 
     source.post.assert_not_called()
+
+
+def test_read_paginate_and_ledger_guards(monkeypatch):
+    """Surface exhausted read retries, oversized collections, and invalid ledger schemas."""
+    source = action()
+    monkeypatch.setattr("actions.cla.time.sleep", MagicMock())
+    source.get.return_value = response(502)
+    source.get.return_value.raise_for_status.side_effect = RuntimeError("502")
+    with pytest.raises(RuntimeError, match="502"):
+        cla._read(source, "get", "url")
+    assert source.get.call_count == 4
+
+    source.get.return_value = response(data=[{}] * 100)
+    with pytest.raises(RuntimeError, match="exceeded 10,000 items"):
+        cla._paginate(source, "url")
+
+    source.get.return_value = ledger_response({})
+    with pytest.raises(TypeError, match="invalid schema"):
+        cla._ledger(source)
+
+
+def test_persist_comment_and_rerun_edge_cases(monkeypatch):
+    """Skip already-signed records, fail missing created comments, and require a PR-head run to rerun."""
+    source, store = action(), action()
+    store.get.return_value = ledger_response([{"id": 2}])
+    cla._persist(store, [{"name": "new", "id": 2}], source, 7)
+    store.put.assert_not_called()
+
+    source.post.return_value = response(201)
+    source.get.return_value = response(data=[])
+    with pytest.raises(RuntimeError, match="did not return"):
+        cla._update_comment(source, 7, [], "body")
+
+    source.get.reset_mock()
+    cla._rerun_pr_check(source, 7)
+    source.get.assert_not_called()
+
+    source = action("issue_comment")
+    monkeypatch.setenv("GITHUB_WORKFLOW_REF", "ultralytics/example/.github/workflows/cla.yml@refs/heads/main")
+    source.get.side_effect = [
+        response(data={"head": {"ref": "feature", "sha": "exact"}}),
+        response(data={"workflow_runs": []}),
+    ]
+    with pytest.raises(RuntimeError, match="Could not find"):
+        cla._rerun_pr_check(source, 7)
+
+
+def test_main_builds_source_and_ledger_actions(monkeypatch):
+    """Run the CLA check with separate source and ledger tokens from the environment."""
+    monkeypatch.setenv("GITHUB_TOKEN", "token")
+    monkeypatch.setenv("CLA_TOKEN", "cla-token")
+    source, store = action(), action()
+    run = MagicMock()
+    monkeypatch.setattr("actions.cla.run", run)
+    monkeypatch.setattr("actions.cla.Action", MagicMock(side_effect=[source, store]))
+
+    cla.main()
+
+    assert cla.Action.call_args_list[0].kwargs == {"token": "token"}
+    assert cla.Action.call_args_list[1].kwargs == {
+        "token": "cla-token",
+        "event_name": source.event_name,
+        "event_data": source.event_data,
+    }
+    run.assert_called_once_with(source, store)

tests/test_format_code.py

@@ -39,8 +39,8 @@ def test_markdown_prettier_skips_symlinks():
 
 
 @patch("actions.format_code.subprocess.run")
-def test_format_main_runs_all_formatters(mock_run, monkeypatch):
-    """Test format CLI runs every formatter by default and respects INPUTS_* opt-outs."""
+def test_format_main_runs_all_formatters(mock_run, monkeypatch, capsys):
+    """Test format CLI runs every formatter by default, respects INPUTS_* opt-outs, and continues on failure."""
     format_code.main()
     assert mock_run.call_args_list == [
         call(format_code.RUFF_CHECK, shell=False, check=True),
@@ -55,3 +55,10 @@ def test_format_main_runs_all_formatters(mock_run, monkeypatch):
     monkeypatch.setenv("INPUTS_SPELLING", "false")
     format_code.main()
     assert mock_run.call_args_list == [call(format_code.PRETTIER, shell=True, check=True)]
+
+    mock_run.reset_mock()
+    mock_run.side_effect = RuntimeError("boom")
+    monkeypatch.setenv("INPUTS_SPELLING", "true")
+    format_code.main()
+    assert mock_run.call_count == 2
+    assert "Formatter script failed but continuing: boom" in capsys.readouterr().out

tests/test_github_report.py

@@ -1,7 +1,9 @@
 # Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license
 
+import json
 import urllib.error
 from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 
 from actions import failed_scheduled_actions, github_report
 
@@ -522,3 +524,75 @@ def test_failed_scheduled_actions_summary_appends_section_break(tmp_path, monkey
     failed_scheduled_actions.run()
 
     assert "Skipped 0\n\n# Failed Default Branch Actions" in summary.read_text()
+
+
+def test_github_report_helpers_and_open_prs(monkeypatch):
+    """Status check shapes, age phases, and open PR filtering are handled for each input form."""
+    assert github_report.get_status_checks({"contexts": [{"state": "SUCCESS"}]}) == [{"state": "SUCCESS"}]
+    assert github_report.get_status_checks(None) == []
+    assert [github_report.get_phase_emoji(days)[0] for days in (0, 7, 30, 31)] == ["🆕", "🟢", "🟡", "🔴"]
+    assert github_report.get_age_days(datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")) == 0
+    stdout = '[{"repository": {"name": "repo"}}, {"repository": {"name": "other"}}]'
+    monkeypatch.setattr(github_report.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(stdout=stdout))
+    assert github_report.collect_open_prs("ultralytics", {"repo": "url"}) == [{"repository": {"name": "repo"}}]
+
+
+def test_format_pr_report_empty_and_truncated(monkeypatch):
+    """The PR report handles no PRs and truncates long per-repo lists."""
+    assert "No open PRs found" in github_report.format_pr_report([], {}, "public")
+    monkeypatch.setattr(github_report, "get_age_days", lambda created_at: 0)
+    prs = [{"repository": {"name": "repo"}, "number": i, "title": "t", "url": "u", "createdAt": ""} for i in range(31)]
+    assert "- ... 1 more PRs" in github_report.format_pr_report(prs, {"repo": "url"}, "public")
+
+
+def test_run_pr_report_writes_step_summary(tmp_path, monkeypatch):
+    """The PR report is appended to the step summary with auto-merge enabled or disabled."""
+    summary = tmp_path / "summary.md"
+    monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary))
+    monkeypatch.setenv("AUTO_MERGE_ACTIONS_PRS", "false")
+    monkeypatch.setattr(github_report, "collect_repos", lambda *args: ({}, "public"))
+    github_report.run_pr_report()
+    assert "Auto-merge disabled" in summary.read_text()
+
+    monkeypatch.setenv("AUTO_MERGE_ACTIONS_PRS", "true")
+    monkeypatch.setattr(github_report, "collect_repos", lambda *args: ({"repo": "url"}, "public"))
+    monkeypatch.setattr(github_report, "collect_open_prs", lambda *args: [])
+    monkeypatch.setattr(github_report, "auto_merge_actions_prs", lambda *args: "merged\n")
+    github_report.run_pr_report()
+    assert summary.read_text().endswith("merged\n")
+
+
+def test_auto_merge_actions_prs_conflicts_failures_and_repo_limit(monkeypatch):
+    """Conflicting, failed, unrelated, and second-per-repo PRs are skipped while one eligible PR merges."""
+
+    def pr(number, mergeable, title="Bump actions/cache in /.github/workflows/ci.yml"):
+        rollup = {"contexts": [{"context": "CI", "state": "SUCCESS"}]}
+        return {
+            "number": number,
+            "title": title,
+            "files": [{"path": "action.yml"}],
+            "mergeable": mergeable,
+            "statusCheckRollup": rollup,
+        }
+
+    def fake_run(cmd, **kwargs):
+        if cmd[:3] == ["gh", "pr", "list"]:
+            if "app/dependabot" not in cmd:
+                return SimpleNamespace(returncode=1, stdout="", stderr="")
+            prs = [
+                pr(1, "CONFLICTING"),
+                pr(2, "MERGEABLE"),
+                pr(3, "MERGEABLE"),
+                pr(4, "MERGEABLE"),
+                pr(5, "MERGEABLE", "Other"),
+            ]
+            return SimpleNamespace(returncode=0, stdout=json.dumps(prs), stderr="")
+        return SimpleNamespace(returncode=int(cmd[3] == "2"), stdout="", stderr="boom")
+
+    monkeypatch.setattr(github_report.subprocess, "run", fake_run)
+    report = github_report.auto_merge_actions_prs("ultralytics", {"repo": "url"})
+
+    assert "- ❌ ultralytics/repo#1: merge conflicts" in report
+    assert "- ❌ ultralytics/repo#2: boom" in report
+    assert "- ✅ Merged ultralytics/repo#3" in report
+    assert "**Summary:** Found 4 | Merged 1 | Skipped 3" in report

tests/test_update_file_headers.py

@@ -0,0 +1,72 @@
+# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from actions import update_file_headers
+
+AGPL = "Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license"
+YEAR = datetime.now(timezone.utc).year
+
+
+@pytest.mark.parametrize(
+    ("name", "text", "expected"),
+    [
+        ("a.py", "import os\n", f"# {AGPL}\n\nimport os\n"),
+        ("b.py", '"""Doc."""\n', f'# {AGPL}\n"""Doc."""\n'),
+        (
+            "c.py",
+            "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\nx = 1\n",
+            f"#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n# {AGPL}\n\nx = 1\n",
+        ),
+        ("d.py", "# Ultralytics 🚀 AGPL-3.0 License - old\n\nx = 1\n", f"# {AGPL}\n\nx = 1\n"),
+        ("e.py", "# © 2014-2020 Ultralytics Inc. 🚀 old\nx = 1\n", f"# {AGPL}\n\nx = 1\n"),
+        ("f.css", "body {}\n", f"/* {AGPL} */\n\nbody {{}}\n"),
+        ("g.html", "<!DOCTYPE html>\n<html></html>\n", f"<!DOCTYPE html>\n<!-- {AGPL} -->\n\n<html></html>\n"),
+        ("h.py", f"# {AGPL}\n\nimport os\n", None),
+        ("i.py", "", None),
+        ("missing.py", None, None),
+    ],
+)
+def test_update_file(tmp_path, name, text, expected):
+    """Test headers are inserted, replaced, or left alone for each supported file shape."""
+    path = tmp_path / name
+    if text is not None:
+        path.write_text(text, encoding="utf-8")
+    prefix, start, end = update_file_headers.COMMENT_MAP[path.suffix]
+    assert update_file_headers.update_file(path, prefix, start, end, AGPL) is (expected is not None)
+    if text is not None:
+        assert path.read_text(encoding="utf-8") == (expected or text)
+
+
+def test_main_selects_header_by_repository(tmp_path, monkeypatch, capsys):
+    """Test main picks the confidential, AGPL, or custom header and skips ignored paths."""
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "a.py").write_text("x = 1\n")
+    (tmp_path / "node_modules").mkdir()
+    (tmp_path / "node_modules" / "b.js").write_text("x\n")
+    event = MagicMock(repository="ultralytics/repo")
+    with patch("actions.update_file_headers.Action", return_value=event):
+        event.is_repo_private.return_value = True
+        update_file_headers.main()
+        assert f"# © 2014-{YEAR} Ultralytics Inc. 🚀 All rights reserved." in (tmp_path / "a.py").read_text()
+        assert "Headers: 1, Updated: 1, Unchanged: 0" in capsys.readouterr().out
+
+        event.is_repo_private.return_value = False
+        update_file_headers.main()
+        assert f"# {AGPL}\n\nx = 1\n" == (tmp_path / "a.py").read_text()
+        capsys.readouterr()
+
+        event.repository = "other/repo"
+        monkeypatch.setattr(update_file_headers, "HEADER", "true")
+        update_file_headers.main()
+        assert capsys.readouterr().out == ""
+
+        monkeypatch.setattr(update_file_headers, "HEADER", "© 2014-2020 Custom")
+        update_file_headers.main()
+        assert f"# © 2014-{YEAR} Custom\n\nx = 1\n" == (tmp_path / "a.py").read_text()
+        update_file_headers.main()
+        assert "Headers: 1, Updated: 0, Unchanged: 1" in capsys.readouterr().out
+    assert (tmp_path / "node_modules" / "b.js").read_text() == "x\n"