tegwick
f22faaa815
Expose list/get/claim/close for agent-harness intake against the same auth as POST /issues/ ingestion (HARNESS-WP-0001-T03).
111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
"""Tests for GET/PATCH /issues/ worker poll/claim surface."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("fastapi")
|
|
pytest.importorskip("httpx")
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from issue_core.api.app import create_app
|
|
|
|
API_KEY = "test-key-not-a-real-secret-only-for-pytest"
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_issue_store(monkeypatch, tmp_path):
|
|
config_dir = tmp_path / "issue-core"
|
|
config_dir.mkdir()
|
|
db_path = str(config_dir / "issues.db")
|
|
configs = {"default": "local", "local": {"type": "local", "db_path": db_path}}
|
|
monkeypatch.setattr(
|
|
"issue_core.cli.utils.get_config_dir", lambda: config_dir, raising=True
|
|
)
|
|
monkeypatch.setattr(
|
|
"issue_core.api.ingest.get_config_dir", lambda: config_dir, raising=True
|
|
)
|
|
monkeypatch.setattr(
|
|
"issue_core.cli.utils.load_backend_configs", lambda: configs, raising=True
|
|
)
|
|
monkeypatch.setattr(
|
|
"issue_core.api.ingest.load_backend_configs", lambda: configs, raising=True
|
|
)
|
|
return config_dir
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch, tmp_issue_store):
|
|
monkeypatch.setenv("ISSUE_CORE_API_KEY", API_KEY)
|
|
return TestClient(create_app())
|
|
|
|
|
|
def _payload(**overrides):
|
|
base = {
|
|
"title": "Run Binky daily rhythm",
|
|
"description": "Queue hygiene + brief",
|
|
"target_repo": "binky-control",
|
|
"priority": "medium",
|
|
"labels": ["binky", "rhythm", "automated"],
|
|
"source_type": "rule",
|
|
"source_id": "emit-daily-rhythm-task",
|
|
"triggering_event_id": str(uuid.uuid4()),
|
|
"activity_definition_id": "binky-daily-rhythm",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
def _auth():
|
|
return {"Authorization": f"Bearer {API_KEY}"}
|
|
|
|
|
|
def test_list_and_claim_flow(client):
|
|
created = client.post("/issues/", json=_payload(), headers=_auth())
|
|
assert created.status_code == 201, created.text
|
|
issue_id = created.json()["issue_id"]
|
|
|
|
listed = client.get("/issues/", params={"state": "open"}, headers=_auth())
|
|
assert listed.status_code == 200
|
|
items = listed.json()
|
|
assert len(items) == 1
|
|
assert items[0]["issue_id"] == issue_id
|
|
assert items[0]["target_repo"] == "binky-control"
|
|
assert "automated" in items[0]["labels"]
|
|
assert items[0]["activity_definition_id"] == "binky-daily-rhythm"
|
|
|
|
filtered = client.get(
|
|
"/issues/",
|
|
params=[("state", "open"), ("label", "automated")],
|
|
headers=_auth(),
|
|
)
|
|
assert filtered.status_code == 200
|
|
assert len(filtered.json()) == 1
|
|
|
|
claimed = client.patch(
|
|
f"/issues/{issue_id}",
|
|
json={"state": "in_progress", "assignee": "agent-harness"},
|
|
headers=_auth(),
|
|
)
|
|
assert claimed.status_code == 200, claimed.text
|
|
assert claimed.json()["state"] == "in_progress"
|
|
assert claimed.json()["assignee"] == "agent-harness"
|
|
|
|
open_after = client.get("/issues/", params={"state": "open"}, headers=_auth())
|
|
assert open_after.json() == []
|
|
|
|
closed = client.patch(
|
|
f"/issues/{issue_id}",
|
|
json={"state": "closed"},
|
|
headers=_auth(),
|
|
)
|
|
assert closed.status_code == 200
|
|
assert closed.json()["state"] == "closed"
|
|
|
|
|
|
def test_get_issue_not_found(client):
|
|
resp = client.get("/issues/does-not-exist", headers=_auth())
|
|
assert resp.status_code == 404
|