From 322e4768880c39d938495f9e57c7278ae419a1f8 Mon Sep 17 00:00:00 2001 From: northline-lab Date: Sun, 31 May 2026 03:58:47 +0000 Subject: [PATCH] test: add unit tests for external_lifecycle pure helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Add 48 dedicated unit tests for pure helper functions in `contribarena.engine.external_lifecycle` that previously had no dedicated test coverage. ## Test Coverage - **ParseTest** (6 tests) — ISO datetime parsing with Z suffix, UTC offset, positive/negative offsets, naive datetime assumes UTC, microseconds - **IsoTest** (4 tests) — UTC-aware, naive assumes UTC, positive/negative offset conversion to UTC - **GithubStateTest** (5 tests) — None/open, not-ok defaults to open, merged, closed not merged, open - **HasRequestedChangesTest** (6 tests) — changes_requested detection, case-insensitive, approved, empty list, multiple reviews, missing state attribute - **HasRejectionSignalTest** (6 tests) — rejection/opt_out/anti_ai_or_bot detection, positive exclusion, empty list, mixed signals - **SummaryForStatusTest** (7 tests) — all 6 known statuses plus unknown fallback - **ClassifyStatusTest** (10 tests) — full priority chain: not-ok → merged → rejection → closed → changes → CI failure → tracking, plus pr=None variants and merged-overrides-changes - **SignalsFromStatusTest** (4 tests) — non-needs_response returns empty, needs_response without changes returns empty, creates process_feedback signal, deduplicates existing signals ## Why This Matters These helpers are central to the external PR lifecycle tracking contract: - `_parse`/`_iso` handle GitHub's various timestamp formats consistently - `_classify_status` implements the priority-ordered lifecycle classification - `_github_state` maps PR API results to lifecycle states - `_signals_from_status` generates maintainer signals with deduplication ## Verification ``` pytest -q tests/unit/test_external_lifecycle.py -v # 48 passed in 0.30s ruff check tests/unit/test_external_lifecycle.py # All checks passed! ``` ## Risk Low — test-only addition. No production code is modified. --- *This PR was created autonomously by an AI agent participating in ContribArena's evaluation framework.* --- tests/unit/test_external_lifecycle.py | 299 ++++++++++++++++++++++++++ 1 file changed, 299 insertions(+) create mode 100644 tests/unit/test_external_lifecycle.py diff --git a/tests/unit/test_external_lifecycle.py b/tests/unit/test_external_lifecycle.py new file mode 100644 index 0000000..0082365 --- /dev/null +++ b/tests/unit/test_external_lifecycle.py @@ -0,0 +1,299 @@ +from __future__ import annotations + +import unittest +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta, timezone + +from contribarena.engine.external_lifecycle import ( + _classify_status, + _github_state, + _has_rejection_signal, + _has_requested_changes, + _iso, + _parse, + _signals_from_status, + _summary_for_status, +) +from contribarena.models import MaintainerSignal, PrLifecycleRecord +from contribarena.tools.github_pr import PullRequestStatusResult + + +class ParseTest(unittest.TestCase): + def test_utc_z_suffix(self) -> None: + result = _parse("2025-06-15T12:00:00Z") + self.assertEqual(UTC, result.tzinfo) + self.assertEqual(datetime(2025, 6, 15, 12, 0, tzinfo=UTC), result) + + def test_utc_offset(self) -> None: + result = _parse("2025-06-15T12:00:00+00:00") + self.assertEqual(UTC, result.tzinfo) + + def test_positive_offset(self) -> None: + result = _parse("2025-06-15T12:00:00+05:30") + self.assertEqual(UTC, result.tzinfo) + self.assertEqual(6, result.hour) + self.assertEqual(30, result.minute) + + def test_negative_offset(self) -> None: + result = _parse("2025-06-15T12:00:00-04:00") + self.assertEqual(UTC, result.tzinfo) + self.assertEqual(16, result.hour) + + def test_naive_assumes_utc(self) -> None: + result = _parse("2025-06-15T12:00:00") + self.assertEqual(UTC, result.tzinfo) + self.assertEqual(12, result.hour) + + def test_microseconds(self) -> None: + result = _parse("2025-06-15T12:00:00.123456Z") + self.assertEqual(UTC, result.tzinfo) + self.assertEqual(123456, result.microsecond) + + +class IsoTest(unittest.TestCase): + def test_utc_aware(self) -> None: + dt = datetime(2025, 6, 15, 12, 0, tzinfo=UTC) + result = _iso(dt) + self.assertIn("2025-06-15T12:00:00", result) + + def test_naive_assumes_utc(self) -> None: + dt = datetime(2025, 6, 15, 12, 0) + result = _iso(dt) + self.assertIn("2025-06-15T12:00:00", result) + + def test_positive_offset_converted(self) -> None: + tz = timezone(timedelta(hours=5, minutes=30)) + dt = datetime(2025, 6, 15, 12, 0, tzinfo=tz) + result = _iso(dt) + self.assertIn("2025-06-15T06:30:00", result) + + def test_negative_offset_converted(self) -> None: + tz = timezone(timedelta(hours=-4)) + dt = datetime(2025, 6, 15, 12, 0, tzinfo=tz) + result = _iso(dt) + self.assertIn("2025-06-15T16:00:00", result) + + +class GithubStateTest(unittest.TestCase): + def test_none_is_open(self) -> None: + self.assertEqual("open", _github_state(None)) + + def test_not_ok_is_open(self) -> None: + pr = PullRequestStatusResult(ok=False, number=1, state="closed") + self.assertEqual("open", _github_state(pr)) + + def test_merged(self) -> None: + pr = PullRequestStatusResult(ok=True, number=1, state="closed", merged=True) + self.assertEqual("merged", _github_state(pr)) + + def test_closed_not_merged(self) -> None: + pr = PullRequestStatusResult(ok=True, number=1, state="closed", merged=False) + self.assertEqual("closed", _github_state(pr)) + + def test_open(self) -> None: + pr = PullRequestStatusResult(ok=True, number=1, state="open", merged=False) + self.assertEqual("open", _github_state(pr)) + + +class HasRequestedChangesTest(unittest.TestCase): + def test_changes_requested(self) -> None: + review = _ReviewStub(state="CHANGES_REQUESTED") + self.assertTrue(_has_requested_changes([review])) + + def test_changes_requested_lowercase(self) -> None: + review = _ReviewStub(state="changes_requested") + self.assertTrue(_has_requested_changes([review])) + + def test_approved(self) -> None: + review = _ReviewStub(state="APPROVED") + self.assertFalse(_has_requested_changes([review])) + + def test_empty_list(self) -> None: + self.assertFalse(_has_requested_changes([])) + + def test_multiple_reviews(self) -> None: + reviews = [_ReviewStub(state="APPROVED"), _ReviewStub(state="CHANGES_REQUESTED")] + self.assertTrue(_has_requested_changes(reviews)) + + def test_no_state_attribute(self) -> None: + self.assertFalse(_has_requested_changes([object()])) + + +class HasRejectionSignalTest(unittest.TestCase): + def test_rejection(self) -> None: + signal = MaintainerSignal(repository="a/b", kind="rejection") + self.assertTrue(_has_rejection_signal([signal])) + + def test_opt_out(self) -> None: + signal = MaintainerSignal(repository="a/b", kind="opt_out") + self.assertTrue(_has_rejection_signal([signal])) + + def test_anti_ai_or_bot(self) -> None: + signal = MaintainerSignal(repository="a/b", kind="anti_ai_or_bot") + self.assertTrue(_has_rejection_signal([signal])) + + def test_positive_not_rejection(self) -> None: + signal = MaintainerSignal(repository="a/b", kind="positive") + self.assertFalse(_has_rejection_signal([signal])) + + def test_empty_list(self) -> None: + self.assertFalse(_has_rejection_signal([])) + + def test_multiple_signals(self) -> None: + signals = [ + MaintainerSignal(repository="a/b", kind="positive"), + MaintainerSignal(repository="a/b", kind="rejection"), + ] + self.assertTrue(_has_rejection_signal(signals)) + + +class SummaryForStatusTest(unittest.TestCase): + def test_merged(self) -> None: + self.assertEqual("external PR was merged", _summary_for_status("merged")) + + def test_closed(self) -> None: + self.assertEqual("external PR was closed", _summary_for_status("closed")) + + def test_rejected(self) -> None: + self.assertEqual( + "external PR has maintainer rejection signal", + _summary_for_status("rejected"), + ) + + def test_needs_response(self) -> None: + self.assertEqual( + "external PR needs agent follow-up", + _summary_for_status("needs_response"), + ) + + def test_failed(self) -> None: + self.assertEqual( + "external PR lifecycle observation failed", + _summary_for_status("failed"), + ) + + def test_tracking(self) -> None: + self.assertEqual( + "external PR remains open and tracked", + _summary_for_status("tracking"), + ) + + def test_unknown(self) -> None: + self.assertEqual( + "external PR remains open and tracked", + _summary_for_status("unknown"), + ) + + +class ClassifyStatusTest(unittest.TestCase): + def _record(self, **kwargs: object) -> PrLifecycleRecord: + defaults: dict[str, object] = { + "repository": "example/repo", + "number": 1, + } + defaults.update(kwargs) + return PrLifecycleRecord(**defaults) # type: ignore[arg-type] + + def test_pr_not_ok(self) -> None: + record = self._record() + pr = PullRequestStatusResult(ok=False, number=1) + self.assertEqual("failed", _classify_status(record, pr, None, [])) + + def test_merged(self) -> None: + record = self._record() + pr = PullRequestStatusResult(ok=True, number=1, merged=True) + self.assertEqual("merged", _classify_status(record, pr, None, [])) + + def test_rejection_signal(self) -> None: + signal = MaintainerSignal(repository="example/repo", kind="rejection") + record = self._record(maintainer_signals=[signal]) + pr = PullRequestStatusResult(ok=True, number=1, state="open", merged=False) + self.assertEqual("rejected", _classify_status(record, pr, None, [])) + + def test_closed_not_merged(self) -> None: + record = self._record() + pr = PullRequestStatusResult(ok=True, number=1, state="closed", merged=False) + self.assertEqual("closed", _classify_status(record, pr, None, [])) + + def test_requested_changes(self) -> None: + record = self._record() + pr = PullRequestStatusResult(ok=True, number=1, state="open", merged=False) + reviews = [_ReviewStub(state="CHANGES_REQUESTED")] + self.assertEqual("needs_response", _classify_status(record, pr, None, reviews)) + + def test_ci_failure(self) -> None: + from contribarena.models import CiStatus + + record = self._record() + pr = PullRequestStatusResult(ok=True, number=1, state="open", merged=False) + ci = CiStatus(status="failure") + self.assertEqual("needs_response", _classify_status(record, pr, ci, [])) + + def test_tracking_when_all_ok(self) -> None: + record = self._record() + pr = PullRequestStatusResult(ok=True, number=1, state="open", merged=False) + self.assertEqual("tracking", _classify_status(record, pr, None, [])) + + def test_pr_none_with_ci_failure(self) -> None: + from contribarena.models import CiStatus + + record = self._record() + ci = CiStatus(status="failure") + self.assertEqual("needs_response", _classify_status(record, None, ci, [])) + + def test_pr_none_tracking(self) -> None: + record = self._record() + self.assertEqual("tracking", _classify_status(record, None, None, [])) + + def test_merged_overrides_changes(self) -> None: + record = self._record() + pr = PullRequestStatusResult(ok=True, number=1, state="closed", merged=True) + reviews = [_ReviewStub(state="CHANGES_REQUESTED")] + self.assertEqual("merged", _classify_status(record, pr, None, reviews)) + + +class SignalsFromStatusTest(unittest.TestCase): + def _record(self, **kwargs: object) -> PrLifecycleRecord: + defaults: dict[str, object] = { + "repository": "example/repo", + "number": 1, + } + defaults.update(kwargs) + return PrLifecycleRecord(**defaults) # type: ignore[arg-type] + + def test_not_needs_response(self) -> None: + record = self._record() + result = _signals_from_status(record, "tracking", [], datetime.now(UTC)) + self.assertEqual([], result) + + def test_needs_response_no_changes(self) -> None: + record = self._record() + result = _signals_from_status(record, "needs_response", [], datetime.now(UTC)) + self.assertEqual([], result) + + def test_creates_signal(self) -> None: + record = self._record() + reviews = [_ReviewStub(state="CHANGES_REQUESTED")] + result = _signals_from_status(record, "needs_response", reviews, datetime.now(UTC)) + self.assertEqual(1, len(result)) + self.assertEqual("process_feedback", result[0].kind) + self.assertEqual("medium", result[0].severity) + self.assertEqual("example/repo", result[0].repository) + self.assertEqual("example", result[0].organization) + self.assertEqual("pr#1", result[0].source) + + def test_deduplicates_existing_signal(self) -> None: + existing = MaintainerSignal( + repository="example/repo", + kind="process_feedback", + source="pr#1", + ) + record = self._record(maintainer_signals=[existing]) + reviews = [_ReviewStub(state="CHANGES_REQUESTED")] + result = _signals_from_status(record, "needs_response", reviews, datetime.now(UTC)) + self.assertEqual([], result) + + +@dataclass +class _ReviewStub: + state: str