bot-bottle/tests/unit/orchestrator/test_lifecycle.py

"""Unit: the orchestration lifecycle."""

from __future__ import annotations

import unittest
from typing import cast

from bot_bottle.orchestrator.lifecycle import Orchestrator
from bot_bottle.orchestrator.model import (
    STATUS_FROZEN,
    STATUS_RUNNING,
    CommentCreated,
    IssueAssigned,
    PullRequestClosed,
)
from bot_bottle.orchestrator.store import InMemoryStateStore

from ._fakes import FakeForge, FakeRunner


def _assigned(
    labels: tuple[str, ...] = ("bot-bottle:impl",),
    assignees: tuple[str, ...] = ("agent-bot",),
) -> IssueAssigned:
    return IssueAssigned(
        owner="didericis", repo="bot-bottle", issue_number=17,
        title="t", body="the task", assignees=tuple(assignees), labels=tuple(labels),
    )


class LifecycleTest(unittest.TestCase):
    def setUp(self):
        self.forge = FakeForge(members=("agent-bot",))
        self.store = InMemoryStateStore()
        self.runner = FakeRunner()
        self.orch = Orchestrator(
            forge=self.forge, store=self.store, runner=self.runner,
            org="bot-bottle", gitea_api="https://g/api/v1",
            now=lambda: "2026-07-01T00:00:00-04:00",
        )

    def _record(self):
        return self.store.get("didericis", "bot-bottle", 17)

    def test_assigned_targeted_launches(self):
        self.orch.handle(_assigned())
        rec = self._record()
        assert rec is not None
        self.assertEqual(STATUS_RUNNING, rec.status)
        self.assertEqual("impl-didericis-bot-bottle-17", rec.slug)
        self.assertEqual("start", self.runner.calls[0][0])
        # forge context injected into the child env.
        env = cast("dict[str, str]", self.runner.calls[0][5])
        self.assertEqual("didericis", env["FORGE_OWNER"])
        self.assertEqual("17", env["FORGE_ISSUE_NUMBER"])

    def test_untargeted_ignored(self):
        self.orch.handle(_assigned(labels=("bug",)))
        self.assertIsNone(self._record())
        self.assertEqual([], self.runner.calls)

    def test_assigned_is_idempotent(self):
        self.orch.handle(_assigned())
        self.orch.handle(_assigned())  # redelivery
        starts = [c for c in self.runner.calls if c[0] == "start"]
        self.assertEqual(1, len(starts))

    def test_done_signal_freezes(self):
        self.orch.handle(_assigned())
        self.orch.on_done_signal("didericis", "bot-bottle", 17, "success", "done")
        rec = self._record()
        assert rec is not None
        self.assertEqual(STATUS_FROZEN, rec.status)
        self.assertIn(("freeze", "impl-didericis-bot-bottle-17"), self.runner.calls)

    def test_done_signal_ignored_when_not_running(self):
        # No record yet -> no freeze.
        self.orch.on_done_signal("didericis", "bot-bottle", 17, "s", "")
        self.assertEqual([], self.runner.calls)

    def test_comment_on_frozen_resumes(self):
        self.orch.handle(_assigned())
        self.orch.on_done_signal("didericis", "bot-bottle", 17, "s", "")
        self.orch.handle(CommentCreated(
            owner="didericis", repo="bot-bottle", issue_number=17,
            comment_id=1, author="reviewer", body="please redo", is_pull=False,
        ))
        rec = self._record()
        assert rec is not None
        self.assertEqual(STATUS_RUNNING, rec.status)
        self.assertIn(("resume", "impl-didericis-bot-bottle-17", "please redo"),
                      self.runner.calls)

    def test_comment_echo_guard(self):
        self.orch.handle(_assigned())
        self.orch.on_done_signal("didericis", "bot-bottle", 17, "s", "")
        rec = self._record()
        assert rec is not None
        rec.agent_git_user = "agent-bot"
        self.store.upsert(rec)
        self.orch.handle(CommentCreated(
            owner="didericis", repo="bot-bottle", issue_number=17,
            comment_id=2, author="agent-bot", body="I finished", is_pull=False,
        ))
        # Still frozen, no resume triggered by the agent's own comment.
        self.assertEqual(STATUS_FROZEN, self._record().status)  # type: ignore[union-attr]
        self.assertNotIn("resume", [c[0] for c in self.runner.calls])

    def test_comment_on_running_ignored(self):
        self.orch.handle(_assigned())  # running
        self.orch.handle(CommentCreated(
            owner="didericis", repo="bot-bottle", issue_number=17,
            comment_id=1, author="reviewer", body="hi", is_pull=False,
        ))
        self.assertNotIn("resume", [c[0] for c in self.runner.calls])

    def test_pr_comment_routes_via_link(self):
        self.orch.handle(_assigned())
        self.orch.on_done_signal("didericis", "bot-bottle", 17, "s", "")
        self.orch.link_pr("didericis", "bot-bottle", 17, 42)
        # Comment arrives on PR #42 (issue_number == PR number in Gitea).
        self.orch.handle(CommentCreated(
            owner="didericis", repo="bot-bottle", issue_number=42,
            comment_id=9, author="reviewer", body="fix", is_pull=True,
        ))
        self.assertIn(("resume", "impl-didericis-bot-bottle-17", "fix"),
                      self.runner.calls)

    def test_pr_closed_destroys_and_removes(self):
        self.orch.handle(_assigned())
        self.orch.link_pr("didericis", "bot-bottle", 17, 42)
        self.orch.handle(PullRequestClosed(
            owner="didericis", repo="bot-bottle", pr_number=42, merged=True,
        ))
        self.assertIn(("destroy", "impl-didericis-bot-bottle-17"), self.runner.calls)
        self.assertIsNone(self._record())


if __name__ == "__main__":
    unittest.main()