diff --git a/pipeline-runner/run_tests.py b/pipeline-runner/run_tests.py new file mode 100644 index 0000000..17bb86c --- /dev/null +++ b/pipeline-runner/run_tests.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +"""Run integration tests for pipeline-runner.""" + +import os +import subprocess +import sys + + +def main(): + project_dir = os.path.dirname(os.path.abspath(__file__)) + tests_dir = os.path.join(project_dir, "tests") + + # Build in FAKE mode first + build_script = os.path.join(project_dir, "build.sh") + print("==> Building in FAKE mode...") + result = subprocess.run([build_script, "--fake"], cwd=project_dir) + if result.returncode != 0: + print("Build failed", file=sys.stderr) + sys.exit(1) + + # Run pytest + print("==> Running integration tests...") + result = subprocess.run( + [sys.executable, "-m", "pytest", tests_dir, "-v"], + cwd=project_dir, + ) + sys.exit(result.returncode) + + +if __name__ == "__main__": + main() diff --git a/pipeline-runner/tests/conftest.py b/pipeline-runner/tests/conftest.py new file mode 100644 index 0000000..3554eb4 --- /dev/null +++ b/pipeline-runner/tests/conftest.py @@ -0,0 +1,102 @@ +"""Pytest fixture for managing the pipeline-runner server lifecycle during tests.""" + +import json +import os +import shutil +import socket +import subprocess +import tempfile +import time + +import pytest + + +def get_free_port(): + """Get a free port for the test server.""" + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("", 0)) + port = s.getsockname()[1] + s.close() + return port + + +class PipelineServer: + """Manages a pipeline-runner server instance for tests.""" + + def __init__(self, port, work_dir, verbose=False): + self.port = port + self.work_dir = work_dir + self.verbose = verbose + + test_dir = os.path.dirname(os.path.abspath(__file__)) + project_root = os.path.dirname(test_dir) + self.bin_path = os.path.join(project_root, "build", "pipeline-runner") + + config = { + "port": port, + "pipelines_dir": os.path.join(work_dir, "pipelines") + "/", + "logs_dir": os.path.join(work_dir, "logs") + "/", + } + self.config_path = os.path.join(work_dir, "pipeline-runner.conf") + with open(self.config_path, "w") as f: + json.dump(config, f) + + self.process = None + + @property + def base_url(self): + return f"http://localhost:{self.port}" + + def start(self, timeout=30): + """Start the server and wait until it responds.""" + self.process = subprocess.Popen( + [self.bin_path, self.config_path], + cwd=self.work_dir, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + + deadline = time.time() + timeout + while time.time() < deadline: + try: + resp = __import__("requests").get( + f"{self.base_url}/api/pipelines", timeout=1 + ) + if resp.status_code == 200: + return True + except Exception: + pass + time.sleep(0.3) + + return False + + def stop(self): + """Stop the server.""" + if self.process: + self.process.terminate() + try: + self.process.wait(timeout=10) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait(timeout=5) + + +@pytest.fixture(scope="session") +def pipeline_server(): + """Session-scoped fixture that builds, starts, and stops the pipeline-runner server.""" + project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + build_script = os.path.join(project_dir, "build.sh") + + subprocess.check_call([build_script, "--fake"], cwd=project_dir) + + work_dir = tempfile.mkdtemp(prefix="pipeline-test-") + port = get_free_port() + + server = PipelineServer(port, work_dir) + if not server.start(): + pytest.fail("Could not start pipeline-runner server") + + yield server + + server.stop() + shutil.rmtree(work_dir, ignore_errors=True) diff --git a/pipeline-runner/tests/test_integration.py b/pipeline-runner/tests/test_integration.py new file mode 100644 index 0000000..01a30c0 --- /dev/null +++ b/pipeline-runner/tests/test_integration.py @@ -0,0 +1,85 @@ +"""Integration tests for the full pipeline CRUD + execution flow.""" + +import time +import uuid + +import requests + + +def _unique_id(): + return str(uuid.uuid4()) + + +def test_full_pipeline_lifecycle(pipeline_server): + """Create, list, get, run, poll, log, delete a pipeline end-to-end.""" + base = pipeline_server.base_url + pipeline_id = _unique_id() + + # --- Create pipeline --- + pipeline = { + "id": pipeline_id, + "name": "integration-test-pipeline", + "working_dir": "/tmp", + "prompts": [ + {"id": _unique_id(), "text": "echo step1", "title": "Step 1", "order": 0}, + {"id": _unique_id(), "text": "echo step2", "title": "Step 2", "order": 1}, + {"id": _unique_id(), "text": "echo step3", "title": "Step 3", "order": 2}, + ], + } + resp = requests.post(f"{base}/api/pipelines", json=pipeline) + assert resp.status_code == 200 + created = resp.json() + assert created["id"] == pipeline_id + assert created["name"] == "integration-test-pipeline" + assert len(created["prompts"]) == 3 + + # --- List pipelines --- + resp = requests.get(f"{base}/api/pipelines") + assert resp.status_code == 200 + pipelines = resp.json() + assert any(p["id"] == pipeline_id for p in pipelines) + + # --- Get single pipeline --- + resp = requests.get(f"{base}/api/pipelines/{pipeline_id}") + assert resp.status_code == 200 + fetched = resp.json() + assert fetched["id"] == pipeline_id + assert fetched["name"] == "integration-test-pipeline" + assert len(fetched["prompts"]) == 3 + assert fetched["prompts"][0]["order"] == 0 + + # --- Start a run --- + resp = requests.post(f"{base}/api/runs", json={"pipeline_id": pipeline_id}) + assert resp.status_code == 200 + run_data = resp.json() + run_id = run_data["run_id"] + assert run_id + assert run_data["status"] == "running" + + # --- Poll run status until completed --- + for _ in range(60): + resp = requests.get(f"{base}/api/runs/{run_id}/status") + assert resp.status_code == 200 + status_data = resp.json() + if status_data["status"] in ("completed", "error"): + break + time.sleep(0.5) + else: + pytest.fail("Run did not complete within timeout") + + assert status_data["status"] == "completed" + assert len(status_data["steps"]) == 3 + + # --- Get log --- + resp = requests.get(f"{base}/api/runs/{run_id}/log") + assert resp.status_code == 200 + assert len(resp.text) > 0 + + # --- Delete pipeline --- + resp = requests.delete(f"{base}/api/pipelines/{pipeline_id}") + assert resp.status_code == 200 + assert resp.json()["deleted"] == pipeline_id + + # --- Verify deletion --- + resp = requests.get(f"{base}/api/pipelines/{pipeline_id}") + assert resp.status_code == 404 diff --git a/pipeline-runner/tests/test_smoke.py b/pipeline-runner/tests/test_smoke.py new file mode 100644 index 0000000..ff99458 --- /dev/null +++ b/pipeline-runner/tests/test_smoke.py @@ -0,0 +1,26 @@ +"""Smoke tests for pipeline-runner REST API endpoints.""" + +import pytest +import requests + + +test_get_endpoints = [ + ("api/pipelines", 200, list), + ("api/runs/nonexistent-id/status", 404, dict), +] + + +@pytest.mark.parametrize("endpoint,expected_status,expected_type", test_get_endpoints) +def test_get_endpoints(pipeline_server, endpoint, expected_status, expected_type): + """Check HTTP status code and response content-type for GET endpoints.""" + response = requests.get(f"{pipeline_server.base_url}/{endpoint}") + assert response.status_code == expected_status + assert "application/json" in response.headers.get("Content-Type", "") + data = response.json() + assert isinstance(data, expected_type) + + +def test_cors_headers(pipeline_server): + """Check that CORS headers are present on responses.""" + response = requests.get(f"{pipeline_server.base_url}/api/pipelines") + assert response.headers.get("Access-Control-Allow-Origin") == "*"