150 lines
5.0 KiB
Python
150 lines
5.0 KiB
Python
"""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 = "<html><body><h1>Hello Static</h1></body></html>"
|
|
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 "<h1>Hello Static</h1>" 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
|