b545665e2d
- handle_ci_status: fallback git branch -r --contains when branches[] empty - webhook router: handle pull_request_approved event type - handle_pr: map review.type to review.state for new Gitea format - launcher: auto-advance stage after agent completion (_try_advance_stage) - plane_sync: notify Plane on stage changes - stages.py: stage machine with QG definitions - notifications.py: stage change notifications - safe.directory fix for container git operations
189 lines
6.6 KiB
Python
189 lines
6.6 KiB
Python
import pytest
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import patch, MagicMock
|
|
import httpx
|
|
|
|
# Override DB path before importing app
|
|
_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator.db")
|
|
os.environ["ORCH_DB_PATH"] = _test_db
|
|
os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
|
|
os.environ["ORCH_GITEA_TOKEN"] = "test-token"
|
|
os.environ["ORCH_PLANE_API_TOKEN"] = "test-token"
|
|
|
|
from src.qg.checks import (
|
|
check_analysis_complete,
|
|
check_architecture_done,
|
|
check_ci_green,
|
|
check_review_approved,
|
|
check_tests_passed,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_work_item_dir(tmp_path, monkeypatch):
|
|
"""Create temp repo structure for filesystem checks."""
|
|
monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path))
|
|
repo_dir = tmp_path / "enduro-trails"
|
|
repo_dir.mkdir()
|
|
return repo_dir
|
|
|
|
|
|
class TestCheckAnalysisComplete:
|
|
def test_all_files_present(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "01-brd.md").write_text("# BRD")
|
|
(wi_dir / "02-trz.md").write_text("# TRZ")
|
|
(wi_dir / "03-acceptance-criteria.md").write_text("# AC")
|
|
(wi_dir / "04-test-plan.yaml").write_text("tests: []")
|
|
|
|
passed, reason = check_analysis_complete("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_missing_files(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-002"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "01-brd.md").write_text("# BRD")
|
|
|
|
passed, reason = check_analysis_complete("enduro-trails", "ET-002")
|
|
assert passed is False
|
|
assert "Missing files" in reason
|
|
|
|
def test_no_directory(self, setup_work_item_dir):
|
|
passed, reason = check_analysis_complete("enduro-trails", "ET-999")
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckArchitectureDone:
|
|
def test_adr_directory_with_files(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
adr_dir = repo_dir / "docs" / "work-items" / "ET-001" / "06-adr"
|
|
adr_dir.mkdir(parents=True)
|
|
(adr_dir / "001-use-postgres.md").write_text("# ADR")
|
|
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_infra_requirements(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "07-infra-requirements.md").write_text("# Infra")
|
|
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_empty_adr_directory(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
adr_dir = repo_dir / "docs" / "work-items" / "ET-001" / "06-adr"
|
|
adr_dir.mkdir(parents=True)
|
|
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
|
|
def test_nothing_present(self, setup_work_item_dir):
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckCIGreen:
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_ci_success(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"state": "success"}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test")
|
|
assert passed is True
|
|
assert "green" in reason.lower()
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_ci_pending(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"state": "pending"}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test")
|
|
assert passed is False
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_ci_branch_not_found(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 404
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_ci_green("enduro-trails", "nonexistent")
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckReviewApproved:
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_approved(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = [
|
|
{"state": "APPROVED", "user": {"login": "reviewer1"}}
|
|
]
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_review_approved("enduro-trails", 1)
|
|
assert passed is True
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_changes_requested(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = [
|
|
{"state": "REQUEST_CHANGES", "user": {"login": "reviewer1"}}
|
|
]
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_review_approved("enduro-trails", 1)
|
|
assert passed is False
|
|
assert "Changes requested" in reason
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_no_reviews(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = []
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_review_approved("enduro-trails", 1)
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckTestsPassed:
|
|
def test_report_with_pass(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "13-test-report.md").write_text("# Test Report\n\nResult: PASS\n")
|
|
|
|
passed, reason = check_tests_passed("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_report_without_pass(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "13-test-report.md").write_text("# Test Report\n\nResult: FAIL\n")
|
|
|
|
passed, reason = check_tests_passed("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
|
|
def test_no_report(self, setup_work_item_dir):
|
|
passed, reason = check_tests_passed("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
assert "not found" in reason.lower()
|