Coverage for tests/streamwise_app/app_test_helpers.py: 100%
66 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1"""
2Shared helper functions for StreamWise app HTTP endpoint tests.
4Each helper encapsulates a common assertion so that individual app test files
5can delegate to a single implementation rather than repeating the same code.
6"""
8import os
10from http import HTTPStatus
12from quart import Quart
15async def check_app_root(test_app: Quart, app_title: str) -> None:
16 """Check that GET / returns 200 with the correct HTML and app title."""
17 client = test_app.test_client()
18 response = await client.get("/")
19 assert response is not None
20 assert response.status_code == HTTPStatus.OK
21 assert "text/html; charset=utf-8" == response.content_type
22 response_html = await response.get_data(as_text=True)
23 assert response_html.startswith("<!DOCTYPE html>\n<html lang=\"en\">")
24 assert app_title in response_html
27async def check_health(test_app: Quart) -> None:
28 """Check that /health returns the expected status dict with no jobs."""
29 client = test_app.test_client()
30 response = await client.get("/health")
31 assert response is not None
32 assert response.status_code == HTTPStatus.OK
33 response_json = await response.get_json()
34 assert response_json == {
35 "host": None,
36 "jobs": {},
37 "k8s_cluster": None,
38 "port": None,
39 "services": {},
40 "status": "ok"
41 }
44async def check_files(test_app: Quart, app_name: str) -> None:
45 """Check the /files, /file, /file_stream and /file_view endpoints."""
46 client = test_app.test_client()
48 # Ensure the tmp directory exists so listing always works
49 os.makedirs(f"/tmp/{app_name}", exist_ok=True)
51 response = await client.get("/files")
52 assert response.status_code == HTTPStatus.OK
53 response_json = await response.get_json()
54 assert "files" in response_json
56 response = await client.get("/file/testfile.txt")
57 assert response.status_code == HTTPStatus.NOT_FOUND
58 response_json = await response.get_json()
59 assert response_json == {"error": f"File '/tmp/{app_name}/testfile.txt' not found"}
61 response = await client.get("/file_stream/job_id/testfile2.txt")
62 assert response.status_code == HTTPStatus.NOT_FOUND
63 response_json = await response.get_json()
64 assert response_json == {"error": "File not found"}
66 response = await client.get("/file_view/job_id/testfile3.txt")
67 assert response.status_code == HTTPStatus.OK
68 assert response.content_type == "text/html; charset=utf-8"
69 response_text = await response.get_data(as_text=True)
70 assert response_text.startswith("<!DOCTYPE html>\n<html>")
71 assert "<title>File viewer: testfile3.txt</title>" in response_text
74async def check_unknown_route(test_app: Quart) -> None:
75 """Check that an unknown route returns 404."""
76 client = test_app.test_client()
77 response = await client.get("/does-not-exist")
78 assert response.status_code == HTTPStatus.NOT_FOUND
81async def check_job_submit_page(test_app: Quart) -> None:
82 """Check that GET /job returns the job submission page."""
83 client = test_app.test_client()
84 response = await client.get("/job")
85 assert response.status_code == HTTPStatus.OK
86 text = await response.get_data(as_text=True)
87 assert len(text) > 0
90async def check_job_status_page(test_app: Quart) -> None:
91 """Check that GET /job/<job_id> returns the job status page."""
92 client = test_app.test_client()
93 response = await client.get("/job/testjobid")
94 assert response.status_code == HTTPStatus.OK
95 text = await response.get_data(as_text=True)
96 assert len(text) > 0
99async def check_api_job_status(test_app: Quart) -> None:
100 """Check that the API job status endpoint returns a status dict."""
101 client = test_app.test_client()
102 response = await client.get("/api/job/nonexistent_job_id/status")
103 assert response.status_code == HTTPStatus.OK
104 response_json = await response.get_json()
105 assert "status" in response_json
108async def check_api_job_requests(test_app: Quart) -> None:
109 """Check that the API job requests endpoint returns 200."""
110 client = test_app.test_client()
111 response = await client.get("/api/job/nonexistent_job_id/requests")
112 assert response.status_code == HTTPStatus.OK