"""Smoke tests for pipeline-runner REST API endpoints."""
import os
import shutil
import socket
import tempfile
import pytest
import requests
REQUEST_TIMEOUT = 5
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}", timeout=REQUEST_TIMEOUT
)
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", timeout=REQUEST_TIMEOUT
)
assert response.headers.get("Access-Control-Allow-Origin") == "*"
def _get_free_port():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("", 0))
port = s.getsockname()[1]
s.close()
return port
@pytest.fixture(scope="module")
def static_file_server():
"""Fixture that starts a server with web_root configured for static file tests."""
work_dir = tempfile.mkdtemp(prefix="pipeline-static-test-")
web_root = os.path.join(work_dir, "web")
os.makedirs(web_root)
os.makedirs(os.path.join(web_root, "subdir"))
html_content = "Hello Static
"
with open(os.path.join(web_root, "index.html"), "w") as f:
f.write(html_content)
css_content = "body { color: red; }"
with open(os.path.join(web_root, "style.css"), "w") as f:
f.write(css_content)
js_content = "console.log('hello');"
with open(os.path.join(web_root, "app.js"), "w") as f:
f.write(js_content)
nested_content = "nested file content"
with open(os.path.join(web_root, "subdir", "nested.txt"), "w") as f:
f.write(nested_content)
port = _get_free_port()
test_dir = os.path.dirname(os.path.abspath(__file__))
from conftest import PipelineServer
server = PipelineServer(port, work_dir, web_root=web_root)
if not server.start():
pytest.fail("Could not start pipeline-runner server with static files")
yield server, web_root
server.stop()
shutil.rmtree(work_dir, ignore_errors=True)
class TestStaticFiles:
def test_serve_html(self, static_file_server):
server, _ = static_file_server
resp = requests.get(f"{server.base_url}/index.html", timeout=REQUEST_TIMEOUT)
assert resp.status_code == 200
assert "text/html" in resp.headers.get("Content-Type", "")
assert "Hello Static
" in resp.text
assert "ETag" in resp.headers
def test_serve_css(self, static_file_server):
server, _ = static_file_server
resp = requests.get(f"{server.base_url}/style.css", timeout=REQUEST_TIMEOUT)
assert resp.status_code == 200
assert "text/css" in resp.headers.get("Content-Type", "")
assert resp.text == "body { color: red; }"
def test_serve_js(self, static_file_server):
server, _ = static_file_server
resp = requests.get(f"{server.base_url}/app.js", timeout=REQUEST_TIMEOUT)
assert resp.status_code == 200
assert "text/javascript" in resp.headers.get("Content-Type", "")
assert resp.text == "console.log('hello');"
def test_serve_nested_file(self, static_file_server):
server, _ = static_file_server
resp = requests.get(
f"{server.base_url}/subdir/nested.txt", timeout=REQUEST_TIMEOUT
)
assert resp.status_code == 200
assert "text/plain" in resp.headers.get("Content-Type", "")
assert resp.text == "nested file content"
def test_etag_304(self, static_file_server):
server, _ = static_file_server
resp1 = requests.get(
f"{server.base_url}/index.html", timeout=REQUEST_TIMEOUT
)
assert resp1.status_code == 200
etag = resp1.headers.get("ETag")
assert etag is not None
resp2 = requests.get(
f"{server.base_url}/index.html",
headers={"If-None-Match": etag},
timeout=REQUEST_TIMEOUT,
)
assert resp2.status_code == 304
assert resp2.text == ""
def test_cache_control_header(self, static_file_server):
server, _ = static_file_server
resp = requests.get(
f"{server.base_url}/index.html", timeout=REQUEST_TIMEOUT
)
assert resp.status_code == 200
assert "max-age" in resp.headers.get("Cache-Control", "")
def test_static_404(self, static_file_server):
server, _ = static_file_server
resp = requests.get(
f"{server.base_url}/nonexistent.html", timeout=REQUEST_TIMEOUT
)
assert resp.status_code == 404