293 lines
9.6 KiB
Python
293 lines
9.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from agentci.engine.model import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus
|
|
from agentci.integrations.gitea.models import CommentInfo, IssueInfo, PullRequestInfo
|
|
from agentci.workflows.implementation import implement
|
|
from agentci.workflows.model import AgentResult, ReviewReport
|
|
from agentci.workflows.render import JobRejected
|
|
from tests.workflow_support import make_workflow_harness
|
|
|
|
|
|
class FakeRepository:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
implementations: list[Workflow] | None = None,
|
|
plan: Workflow | None = None,
|
|
) -> None:
|
|
self.implementations = implementations or []
|
|
self.plan = plan
|
|
self.saved_workflows: list[Workflow] = []
|
|
|
|
async def implementation_workflows(self, *_args: object) -> list[Workflow]:
|
|
return self.implementations
|
|
|
|
async def latest_workflow(self, *_args: object) -> Workflow | None:
|
|
return self.plan
|
|
|
|
async def operational_comment_ids(self, *_args: object) -> set[int]:
|
|
return {99}
|
|
|
|
async def save_workflow(self, workflow: Workflow) -> None:
|
|
self.saved_workflows.append(workflow)
|
|
|
|
|
|
class FakeGitea:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
existing_pulls: dict[int, PullRequestInfo] | None = None,
|
|
) -> None:
|
|
self.existing_pulls = existing_pulls or {}
|
|
self.created_pulls: list[tuple[str, str, dict[str, str]]] = []
|
|
self.default_branch_calls: list[tuple[str, str]] = []
|
|
|
|
async def default_branch(self, owner: str, repo: str) -> str:
|
|
self.default_branch_calls.append((owner, repo))
|
|
return "main"
|
|
|
|
async def issue(self, *_args: object) -> IssueInfo:
|
|
return IssueInfo(7, "Fix widget", "The widget is broken.", "open")
|
|
|
|
async def issue_comments(self, *_args: object) -> list[CommentInfo]:
|
|
return [
|
|
CommentInfo(12, "alice", "Please cover empty input.", "2026-07-01"),
|
|
CommentInfo(99, "agentci", "Job queued", "2026-07-02"),
|
|
]
|
|
|
|
async def pull_request(self, _owner: str, _repo: str, number: int) -> PullRequestInfo:
|
|
return self.existing_pulls[number]
|
|
|
|
async def create_pull_request(self, owner: str, repo: str, **values: str) -> PullRequestInfo:
|
|
self.created_pulls.append((owner, repo, values))
|
|
return pull(number=42, branch=values["head"])
|
|
|
|
|
|
def job() -> Job:
|
|
return Job(
|
|
id="job-1",
|
|
kind=JobKind.IMPLEMENT,
|
|
target_key="org/repo:issue:7",
|
|
repo_owner="org",
|
|
repo_name="repo",
|
|
issue_number=7,
|
|
pr_number=None,
|
|
requester="alice",
|
|
comment_id=10,
|
|
delivery_id="delivery-1",
|
|
receive_sequence=1,
|
|
command_body="/agent implement",
|
|
message="Keep the change focused.",
|
|
)
|
|
|
|
|
|
def workflow(*, pr_number: int | None = None) -> Workflow:
|
|
return Workflow(
|
|
id=f"old-{pr_number}",
|
|
kind=WorkflowKind.IMPLEMENT,
|
|
repo_owner="org",
|
|
repo_name="repo",
|
|
issue_number=7,
|
|
workspace_path=Path("/old/repo"),
|
|
base_sha="old-sha",
|
|
pr_number=pr_number,
|
|
status=WorkflowStatus.COMPLETED,
|
|
)
|
|
|
|
|
|
def pull(
|
|
*,
|
|
number: int = 8,
|
|
state: str = "open",
|
|
merged: bool = False,
|
|
branch: str = "agent/old",
|
|
) -> PullRequestInfo:
|
|
return PullRequestInfo(
|
|
number=number,
|
|
title="Existing PR",
|
|
body="Body",
|
|
state=state,
|
|
merged=merged,
|
|
base_branch="main",
|
|
head_branch=branch,
|
|
head_sha="head-sha",
|
|
head_owner="org",
|
|
head_repo="repo",
|
|
)
|
|
|
|
|
|
def implementation_settings(tmp_path: Path) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
branch_prefix="agent",
|
|
workspaces_dir=tmp_path / "workspaces",
|
|
gitea_url="https://git.example.test",
|
|
implement_model="provider/model",
|
|
implement_variant="high",
|
|
implement_review_rounds=2,
|
|
)
|
|
|
|
|
|
async def test_initial_implementation_completes_review_commit_push_and_pr(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
stale = workflow(pr_number=8)
|
|
repository = FakeRepository(implementations=[stale])
|
|
gitea = FakeGitea(existing_pulls={8: pull(state="closed")})
|
|
harness = make_workflow_harness(
|
|
settings=implementation_settings(tmp_path),
|
|
repository=repository,
|
|
gitea=gitea,
|
|
responses=[
|
|
AgentResult(
|
|
summary_markdown="# Implement widget\n\nHandled empty input.",
|
|
tests=["pytest: passed"],
|
|
),
|
|
ReviewReport(summary="Ready", findings=[]),
|
|
],
|
|
)
|
|
|
|
body = await implement(job(), harness.run, harness.services)
|
|
|
|
assert len(harness.run.created_workflows) == 1
|
|
created, created_stage = harness.run.created_workflows[0]
|
|
assert created_stage == "installing development environment"
|
|
assert created.status is WorkflowStatus.ACTIVE
|
|
assert created.branch == f"agent/issue-7-{created.id[:8]}"
|
|
assert created.workspace_path == tmp_path / "workspaces" / created.id / "repo"
|
|
assert created.base_sha == "base-sha"
|
|
assert harness.development.workspaces == [created.workspace_path]
|
|
assert harness.run.linked_sessions == ["implementation-session"]
|
|
assert harness.opencode.created_sessions == [
|
|
(created.workspace_path, "implementation"),
|
|
(created.workspace_path, "implementation-review"),
|
|
]
|
|
assert [call["result_type"] for call in harness.opencode.resume_calls] == [
|
|
AgentResult,
|
|
ReviewReport,
|
|
]
|
|
|
|
initial_prompt = harness.prompts.calls[0]
|
|
assert initial_prompt[0] == "implement_initial"
|
|
assert initial_prompt[1]["artifact"] == "(no canonical plan)"
|
|
assert initial_prompt[1]["request"] == "Keep the change focused."
|
|
assert initial_prompt[1]["context"].startswith("Repository: org/repo\n")
|
|
|
|
assert harness.git.calls == [
|
|
("clone", "org", "repo", "main", created.workspace_path),
|
|
("create_branch", created.workspace_path, created.branch),
|
|
("has_changes", created.workspace_path),
|
|
("diff_check", created.workspace_path),
|
|
("commit", created.workspace_path, "agent: Implement widget"),
|
|
("push", created.workspace_path, created.branch, True),
|
|
]
|
|
assert gitea.created_pulls == [
|
|
(
|
|
"org",
|
|
"repo",
|
|
{
|
|
"title": "Agent: Fix widget",
|
|
"body": (
|
|
"Closes #7\n\n"
|
|
"## Implementation\n\n# Implement widget\n\nHandled empty input.\n\n"
|
|
"## Validation\n\n- pytest: passed\n\n_Created by Agent CI._"
|
|
),
|
|
"head": created.branch,
|
|
"base": "main",
|
|
},
|
|
)
|
|
]
|
|
|
|
completed = repository.saved_workflows[-1]
|
|
assert completed.id == created.id
|
|
assert completed.status is WorkflowStatus.COMPLETED
|
|
assert completed.pr_number == 42
|
|
assert completed.primary_session_id == "implementation-session"
|
|
assert completed.reviewer_session_id == "implementation-review-session"
|
|
assert AgentResult.model_validate_json(completed.artifact or "").tests == ["pytest: passed"]
|
|
assert json.loads(completed.review_json or "") == {
|
|
"summary": "Ready",
|
|
"findings": [],
|
|
}
|
|
assert body == (
|
|
f"<!-- agentci:implementation workflow={created.id} -->\n"
|
|
"Pull request created: https://git.example.test/org/repo/pulls/42\n\n"
|
|
"## Agent result\n\n# Implement widget\n\nHandled empty input.\n\n"
|
|
"## Validation\n\n- pytest: passed\n\nCommit: `commit-sha`"
|
|
)
|
|
|
|
|
|
async def test_initial_implementation_rejects_clean_worktree_before_commit_or_pr(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
repository = FakeRepository()
|
|
gitea = FakeGitea()
|
|
harness = make_workflow_harness(
|
|
settings=implementation_settings(tmp_path),
|
|
repository=repository,
|
|
gitea=gitea,
|
|
changed=False,
|
|
responses=[
|
|
AgentResult(
|
|
summary_markdown="# Implement widget\n\nHandled empty input.",
|
|
tests=["pytest: passed"],
|
|
),
|
|
ReviewReport(summary="Ready", findings=[]),
|
|
],
|
|
)
|
|
|
|
with pytest.raises(
|
|
JobRejected,
|
|
match="OpenCode completed without producing any file changes",
|
|
):
|
|
await implement(job(), harness.run, harness.services)
|
|
|
|
assert [call[0] for call in harness.git.calls] == [
|
|
"clone",
|
|
"create_branch",
|
|
"has_changes",
|
|
]
|
|
assert gitea.created_pulls == []
|
|
assert repository.saved_workflows
|
|
assert all(item.status is WorkflowStatus.ACTIVE for item in repository.saved_workflows)
|
|
assert "creating pull request" not in harness.run.stages
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("existing_pull", "message"),
|
|
[
|
|
(
|
|
pull(state="open"),
|
|
"Agent PR #8 is already open. Use `/agent iterate` on that pull request.",
|
|
),
|
|
(
|
|
pull(state="closed", merged=True),
|
|
"Agent PR #8 has already been merged for this issue.",
|
|
),
|
|
],
|
|
)
|
|
async def test_initial_implementation_rejects_duplicate_agent_pr_before_clone(
|
|
tmp_path: Path,
|
|
existing_pull: PullRequestInfo,
|
|
message: str,
|
|
) -> None:
|
|
repository = FakeRepository(implementations=[workflow(pr_number=8)])
|
|
gitea = FakeGitea(existing_pulls={8: existing_pull})
|
|
harness = make_workflow_harness(
|
|
settings=implementation_settings(tmp_path),
|
|
repository=repository,
|
|
gitea=gitea,
|
|
)
|
|
|
|
with pytest.raises(JobRejected) as error:
|
|
await implement(job(), harness.run, harness.services)
|
|
|
|
assert str(error.value) == message
|
|
assert harness.git.calls == []
|
|
assert harness.development.workspaces == []
|
|
assert harness.opencode.resume_calls == []
|
|
assert gitea.default_branch_calls == []
|