76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
from dataclasses import FrozenInstanceError
|
|
|
|
import pytest
|
|
|
|
from agentci.domain.events import (
|
|
CommandReceived,
|
|
CommentLinked,
|
|
JobCompleted,
|
|
JobStarted,
|
|
PermissionGranted,
|
|
ServiceRestarted,
|
|
)
|
|
from agentci.domain.models import JobKind, JobStatus
|
|
from agentci.domain.state_machine import InvalidTransition, next_state, render_job_comment
|
|
|
|
|
|
def received(body: str = "/agent plan message"):
|
|
return next_state(
|
|
None,
|
|
CommandReceived(
|
|
job_id="job",
|
|
delivery_id="delivery",
|
|
receive_sequence=1,
|
|
command_body=body,
|
|
target_key="org/repo:issue:1",
|
|
repo_owner="org",
|
|
repo_name="repo",
|
|
issue_number=1,
|
|
pr_number=None,
|
|
requester="alice",
|
|
comment_id=4,
|
|
),
|
|
)
|
|
|
|
|
|
def test_permission_parses_and_queues_execution() -> None:
|
|
transition = next_state(received().state, PermissionGranted(job_id="job"))
|
|
assert transition.state.status is JobStatus.QUEUED
|
|
assert transition.state.kind is JobKind.PLAN
|
|
assert transition.state.message == "message"
|
|
assert [(item.listener, item.queue) for item in transition.notifications] == [
|
|
("execute", "jobs"),
|
|
("reconcile_comment", "control"),
|
|
]
|
|
|
|
|
|
def test_invalid_syntax_is_rejected_after_permission() -> None:
|
|
transition = next_state(received("/agent nonsense").state, PermissionGranted(job_id="job"))
|
|
assert transition.state.status is JobStatus.REJECTED
|
|
assert "Unknown" in (transition.state.error or "")
|
|
|
|
|
|
def test_running_completion_and_restart_are_explicit() -> None:
|
|
queued = next_state(received().state, PermissionGranted(job_id="job")).state
|
|
running = next_state(queued, JobStarted(job_id="job")).state
|
|
completed = next_state(running, JobCompleted(job_id="job", comment_body="# Result")).state
|
|
assert completed.status is JobStatus.SUCCEEDED
|
|
assert "# Result" in render_job_comment(completed)
|
|
assert next_state(completed, ServiceRestarted(job_id="job")).state == completed
|
|
|
|
|
|
def test_comment_link_is_allowed_on_terminal_state() -> None:
|
|
queued = next_state(received().state, PermissionGranted(job_id="job")).state
|
|
running = next_state(queued, JobStarted(job_id="job")).state
|
|
completed = next_state(running, JobCompleted(job_id="job", comment_body="ok")).state
|
|
linked = next_state(completed, CommentLinked(job_id="job", comment_id=9)).state
|
|
assert linked.accepted_comment_id == 9
|
|
|
|
|
|
def test_state_is_immutable_and_invalid_transitions_fail() -> None:
|
|
state = received().state
|
|
with pytest.raises(FrozenInstanceError):
|
|
state.stage = "changed" # type: ignore[misc]
|
|
with pytest.raises(InvalidTransition):
|
|
next_state(state, JobStarted(job_id="job"))
|