92 lines
2.4 KiB
Python
92 lines
2.4 KiB
Python
import pytest
|
|
import os
|
|
import tempfile
|
|
|
|
# Override DB path before importing app
|
|
_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator.db")
|
|
os.environ["ORCH_DB_PATH"] = _test_db
|
|
|
|
from fastapi.testclient import TestClient
|
|
from src.main import app
|
|
from src.db import init_db
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_db():
|
|
"""Ensure DB tables exist before each test."""
|
|
# Remove old test db if exists
|
|
if os.path.exists(_test_db):
|
|
os.unlink(_test_db)
|
|
init_db()
|
|
yield
|
|
if os.path.exists(_test_db):
|
|
os.unlink(_test_db)
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_health():
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
assert resp.json()["service"] == "orchestrator"
|
|
|
|
|
|
def test_plane_webhook_accepts():
|
|
resp = client.post("/webhook/plane", json={
|
|
"event": "work_item.created",
|
|
"data": {"id": "test-123", "name": "Test task", "project": "proj-1"}
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "accepted"
|
|
|
|
|
|
def test_plane_webhook_comment():
|
|
resp = client.post("/webhook/plane", json={
|
|
"event": "comment.created",
|
|
"data": {"comment": "LGTM :approved:"}
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "accepted"
|
|
|
|
|
|
def test_gitea_webhook_push():
|
|
resp = client.post(
|
|
"/webhook/gitea",
|
|
json={"ref": "refs/heads/feature/test", "repository": {"name": "enduro-trails"}},
|
|
headers={"X-Gitea-Event": "push"}
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "accepted"
|
|
|
|
|
|
def test_gitea_webhook_pr():
|
|
resp = client.post(
|
|
"/webhook/gitea",
|
|
json={
|
|
"action": "reviewed",
|
|
"pull_request": {"state": "approved", "number": 1}
|
|
},
|
|
headers={"X-Gitea-Event": "pull_request"}
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "accepted"
|
|
|
|
|
|
def test_status_endpoint():
|
|
resp = client.get("/status")
|
|
assert resp.status_code == 200
|
|
assert "active_tasks" in resp.json()
|
|
|
|
|
|
def test_plane_webhook_creates_task():
|
|
"""Verify that work_item.created actually inserts a task."""
|
|
client.post("/webhook/plane", json={
|
|
"event": "work_item.created",
|
|
"data": {"id": "task-456", "name": "New feature", "project": "proj-2"}
|
|
})
|
|
resp = client.get("/status")
|
|
tasks = resp.json()["active_tasks"]
|
|
assert any(t["plane_id"] == "task-456" for t in tasks)
|