roboco/tests/integration/services/test_env_sync_engine.py

"""EnvSyncEngine — cascade prod->head; conflict opens a PR + task, never prod.

Mirrors the ci-watch engine test: seeds projects + agents, mocks the GitService
(the merges API + PR open are GitHub calls) so the cascade is driven by the
fake's queued statuses, and asserts the engine's contract:

* clean cascade (merged / already_ancestor) opens nothing,
* a conflict opens ONE sync PR + tracked task and stops the cascade,
* a tokenless project is skipped,
* per-cycle + rolling caps + per-repo dedup are honoured,
* the cascade target is never prod (the lower rung of every pair is non-prod).
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any
from uuid import uuid4

import pytest
from roboco.config import settings
from roboco.db.tables import AgentTable, ProjectTable
from roboco.foundation import identity as _foundation
from roboco.models.base import AgentRole, AgentStatus, TaskStatus, Team
from roboco.services import env_sync_engine as env_sync_module
from roboco.services.env_sync_engine import get_env_sync_engine
from roboco.services.task import ENV_SYNC_SOURCE, get_task_service

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession

SYSTEM_UUID = _foundation.AGENTS["system"].uuid
MAIN_PM_UUID = _foundation.AGENTS["main-pm"].uuid

# A two-rung ladder: head=dev (PR target), prod=master (release target).
_LADDER = [
    {"name": "head", "branch": "dev"},
    {"name": "prod", "branch": "master"},
]


class _FakeGit:
    """Stand-in for GitService: drives the cascade from queued statuses."""

    def __init__(self, statuses: list[str], *, pr_number: int = 42) -> None:
        self._statuses = list(statuses)
        self.sync_calls: list[tuple[str, str, str]] = []
        self.pr_calls: list[tuple[str, str, str, str]] = []
        self._pr_number = pr_number

    async def sync_env_branch(
        self, slug: str, target_branch: str, source_branch: str
    ) -> dict[str, Any]:
        self.sync_calls.append((slug, target_branch, source_branch))
        status = self._statuses.pop(0) if self._statuses else "already_ancestor"
        return {"status": status}

    async def open_sync_pr(
        self, slug: str, source_branch: str, target_branch: str, body: str
    ) -> dict[str, Any] | None:
        self.pr_calls.append((slug, source_branch, target_branch, body))
        return {
            "number": self._pr_number,
            "url": f"https://github.com/x/{slug}/pull/{self._pr_number}",
        }


async def _get_or_create_agent(
    db: AsyncSession, agent_id: object, role: AgentRole, slug: str
) -> None:
    if await db.get(AgentTable, agent_id) is None:
        db.add(
            AgentTable(
                id=agent_id,
                name=slug,
                slug=f"{slug}-{uuid4().hex[:8]}",
                role=role,
                team=None,
                status=AgentStatus.ACTIVE,
                model_config={},
                system_prompt="x",
                capabilities=[],
                permissions={},
                metrics={},
            )
        )
        await db.flush()


async def _seed_project(
    db: AsyncSession,
    slug: str,
    git_url: str,
    *,
    environments: list[dict[str, str]] | None = _LADDER,
    token: str | None = "fake-encrypted-token",
) -> ProjectTable:
    project = ProjectTable(
        id=uuid4(),
        name=slug,
        slug=slug,
        git_url=git_url,
        assigned_cell=Team.BACKEND,
        created_by=SYSTEM_UUID,
        environments=environments,
        git_token_encrypted=token,
    )
    db.add(project)
    await db.flush()
    return project


@pytest.fixture(autouse=True)
async def _enabled(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "env_sync_enabled", True)
    monkeypatch.setattr(settings, "env_sync_max_per_cycle", 5)
    monkeypatch.setattr(settings, "env_sync_max_open_tasks", 5)
    await _get_or_create_agent(db_session, SYSTEM_UUID, AgentRole.SYSTEM, "system")
    await _get_or_create_agent(db_session, MAIN_PM_UUID, AgentRole.MAIN_PM, "main-pm")


def _patch_git(monkeypatch: pytest.MonkeyPatch, fake: _FakeGit) -> None:
    monkeypatch.setattr(env_sync_module, "get_git_service", lambda _session: fake)


@pytest.mark.asyncio
async def test_clean_cascade_opens_nothing(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    proj = await _seed_project(db_session, "clean-a", "https://github.com/x/a.git")
    fake = _FakeGit(["merged"])
    _patch_git(monkeypatch, fake)
    created = await get_env_sync_engine(db_session).run_cycle([proj])
    assert created == []
    # prod(master) merged into head(dev) — one cascade step for the 2-rung ladder.
    assert fake.sync_calls == [("clean-a", "dev", "master")]
    assert fake.pr_calls == []


@pytest.mark.asyncio
async def test_already_ancestor_is_clean(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    proj = await _seed_project(db_session, "anc-a", "https://github.com/x/anc.git")
    fake = _FakeGit(["already_ancestor"])
    _patch_git(monkeypatch, fake)
    assert await get_env_sync_engine(db_session).run_cycle([proj]) == []
    assert fake.sync_calls == [("anc-a", "dev", "master")]


@pytest.mark.asyncio
async def test_conflict_opens_pr_and_task_then_stops(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    # 4-rung ladder so a conflict on the FIRST (topmost) pair stops the cascade
    # before reaching head — proving it does not cascade a dirty merge downward.
    proj = await _seed_project(
        db_session,
        "conf-a",
        "https://github.com/x/conf.git",
        environments=[
            {"name": "head", "branch": "dev"},
            {"name": "qa", "branch": "qa"},
            {"name": "stag", "branch": "stag"},
            {"name": "prod", "branch": "master"},
        ],
    )
    fake = _FakeGit(["conflict", "merged"])  # only the first is consumed
    _patch_git(monkeypatch, fake)
    created = await get_env_sync_engine(db_session).run_cycle([proj])

    assert len(created) == 1
    task = created[0]
    assert task.source == ENV_SYNC_SOURCE
    assert task.status == TaskStatus.PENDING
    assert task.project_id == proj.id
    # The cascade stopped at the first conflict: only one sync_env_branch call.
    assert len(fake.sync_calls) == 1
    # The sync PR targets the lower (non-prod) rung of the conflicted pair.
    assert len(fake.pr_calls) == 1
    _slug, _source_branch, target_branch, _body = fake.pr_calls[0]
    assert target_branch != "master"  # never prod


@pytest.mark.asyncio
async def test_missing_ref_skips_without_pr(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    proj = await _seed_project(db_session, "miss-a", "https://github.com/x/miss.git")
    fake = _FakeGit(["missing_ref"])
    _patch_git(monkeypatch, fake)
    created = await get_env_sync_engine(db_session).run_cycle([proj])
    assert created == []
    assert fake.pr_calls == []


@pytest.mark.asyncio
async def test_tokenless_project_skipped(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    proj = await _seed_project(
        db_session, "notok", "https://github.com/x/notok.git", token=None
    )
    fake = _FakeGit(["merged"])
    _patch_git(monkeypatch, fake)
    created = await get_env_sync_engine(db_session).run_cycle([proj])
    assert created == []
    assert fake.sync_calls == []  # never attempted


@pytest.mark.asyncio
async def test_degenerate_ladder_skipped(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    # Single-rung ladder (head==prod) has no pairs to cascade.
    proj = await _seed_project(
        db_session,
        "single",
        "https://github.com/x/single.git",
        environments=[{"name": "prod", "branch": "master"}],
    )
    fake = _FakeGit(["merged"])
    _patch_git(monkeypatch, fake)
    created = await get_env_sync_engine(db_session).run_cycle([proj])
    assert created == []
    assert fake.sync_calls == []


@pytest.mark.asyncio
async def test_per_cycle_cap(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(settings, "env_sync_max_per_cycle", 1)
    p1 = await _seed_project(db_session, "cap-1", "https://github.com/x/c1.git")
    p2 = await _seed_project(db_session, "cap-2", "https://github.com/x/c2.git")
    fake = _FakeGit(["conflict", "conflict"])
    _patch_git(monkeypatch, fake)
    created = await get_env_sync_engine(db_session).run_cycle([p1, p2])
    assert len(created) == 1  # capped at one per cycle


@pytest.mark.asyncio
async def test_deduped_per_repo(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A repo with an open env_sync task is skipped until the PR resolves."""
    proj = await _seed_project(db_session, "dedup", "https://github.com/x/d.git")
    fake = _FakeGit(["conflict"])
    _patch_git(monkeypatch, fake)
    first = await get_env_sync_engine(db_session).run_cycle([proj])
    assert len(first) == 1
    # Second cycle, same repo still has the open task -> deduped (no new PR).
    fake2 = _FakeGit(["conflict"])
    _patch_git(monkeypatch, fake2)
    second = await get_env_sync_engine(db_session).run_cycle([proj])
    assert second == []
    assert fake2.sync_calls == []  # cascade paused at the conflicted rung
    assert len(await get_task_service(db_session).list_open_env_sync_tasks()) == 1


@pytest.mark.asyncio
async def test_disabled_is_noop(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(settings, "env_sync_enabled", False)
    proj = await _seed_project(db_session, "off", "https://github.com/x/off.git")
    fake = _FakeGit(["merged"])
    _patch_git(monkeypatch, fake)
    assert await get_env_sync_engine(db_session).run_cycle([proj]) == []
    assert fake.sync_calls == []