144 lines
4.0 KiB
Python
144 lines
4.0 KiB
Python
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from agentci.adapters.gitea_models import IssueInfo, PullRequestInfo, RepositoryInfo
|
|
from agentci.domain.models import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus
|
|
from agentci.workflows.implement import ImplementWorkflow
|
|
from agentci.workflows.pull_request import PullRequestWorkflow
|
|
|
|
|
|
class SetupReached(RuntimeError):
|
|
pass
|
|
|
|
|
|
class FakeDevelopment:
|
|
description = "python"
|
|
|
|
def __init__(self, events: list[str]) -> None:
|
|
self.events = events
|
|
|
|
async def prepare(self, _workspace: Path) -> None:
|
|
self.events.append("prepare")
|
|
raise SetupReached
|
|
|
|
|
|
class FakeGit:
|
|
def __init__(self, events: list[str]) -> None:
|
|
self.events = events
|
|
|
|
async def clone(self, *_args) -> str:
|
|
self.events.append("clone")
|
|
return "base-sha"
|
|
|
|
async def create_branch(self, *_args) -> None:
|
|
self.events.append("create branch")
|
|
|
|
async def sync_branch(self, *_args) -> None:
|
|
self.events.append("sync branch")
|
|
|
|
|
|
class FakeStorage:
|
|
def __init__(self, workflow: Workflow | None = None) -> None:
|
|
self.workflow = workflow
|
|
|
|
async def implementation_workflows(self, *_args):
|
|
return []
|
|
|
|
async def create_workflow(self, _workflow) -> None:
|
|
return None
|
|
|
|
async def update_job(self, *_args, **_kwargs) -> None:
|
|
return None
|
|
|
|
async def workflow_for_pr(self, *_args):
|
|
return self.workflow
|
|
|
|
|
|
class FakeContext:
|
|
def __init__(self, pull: PullRequestInfo) -> None:
|
|
self.pull = pull
|
|
|
|
async def pull_request_context(self, *_args):
|
|
return self.pull, "context"
|
|
|
|
|
|
class FakeGitea:
|
|
async def repository(self, *_args) -> RepositoryInfo:
|
|
return RepositoryInfo("org", "repo", "org/repo", "main")
|
|
|
|
async def issue(self, *_args) -> IssueInfo:
|
|
return IssueInfo(1, "Issue", "Body", "open")
|
|
|
|
|
|
def job(kind: JobKind, *, pr_number: int | None = None) -> Job:
|
|
return Job(
|
|
id="job",
|
|
kind=kind,
|
|
target_key="org/repo:target",
|
|
repo_owner="org",
|
|
repo_name="repo",
|
|
issue_number=1,
|
|
pr_number=pr_number,
|
|
requester="alice",
|
|
message="",
|
|
comment_id=1,
|
|
)
|
|
|
|
|
|
def pull() -> PullRequestInfo:
|
|
return PullRequestInfo(2, "PR", "Body", "open", False, "main", "agent", "sha", "org", "repo")
|
|
|
|
|
|
async def test_initial_implementation_prepares_after_clone_and_branch(tmp_path) -> None:
|
|
events: list[str] = []
|
|
settings = SimpleNamespace(branch_prefix="agent", workspaces_dir=tmp_path)
|
|
deps = SimpleNamespace(
|
|
settings=settings,
|
|
storage=FakeStorage(),
|
|
gitea=FakeGitea(),
|
|
git=FakeGit(events),
|
|
development=FakeDevelopment(events),
|
|
)
|
|
|
|
with pytest.raises(SetupReached):
|
|
await ImplementWorkflow(deps).run(job(JobKind.IMPLEMENT)) # type: ignore[arg-type]
|
|
|
|
assert events == ["clone", "create branch", "prepare"]
|
|
|
|
|
|
@pytest.mark.parametrize("operation", ["iterate", "fix"])
|
|
async def test_pull_request_implementation_prepares_after_checkout(
|
|
tmp_path, operation: str
|
|
) -> None:
|
|
events: list[str] = []
|
|
existing = Workflow(
|
|
id="flow",
|
|
kind=WorkflowKind.IMPLEMENT,
|
|
repo_owner="org",
|
|
repo_name="repo",
|
|
issue_number=1,
|
|
workspace_path=tmp_path / "repo",
|
|
base_sha="base",
|
|
branch="agent",
|
|
primary_session_id="primary",
|
|
reviewer_session_id="reviewer",
|
|
status=WorkflowStatus.COMPLETED,
|
|
)
|
|
settings = SimpleNamespace(workspaces_dir=tmp_path)
|
|
deps = SimpleNamespace(
|
|
settings=settings,
|
|
storage=FakeStorage(existing),
|
|
context=FakeContext(pull()),
|
|
git=FakeGit(events),
|
|
development=FakeDevelopment(events),
|
|
)
|
|
workflow = PullRequestWorkflow(deps) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(SetupReached):
|
|
await getattr(workflow, operation)(job(JobKind.FIX, pr_number=2))
|
|
|
|
expected_checkout = "sync branch" if operation == "iterate" else "clone"
|
|
assert events == [expected_checkout, "prepare"]
|