494 lines
19 KiB
Python
494 lines
19 KiB
Python
import re
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.frontend.router import router
|
|
from app.identity import (
|
|
get_current_user,
|
|
router as identity_router,
|
|
safe_next_path,
|
|
install_identity_middleware,
|
|
protected_page,
|
|
required_page_permission,
|
|
)
|
|
|
|
|
|
ROUTES = (
|
|
"/dashboard",
|
|
"/monitoring/map",
|
|
"/operations/repairs",
|
|
"/assets/miners",
|
|
)
|
|
STATIC_ROOT = Path(__file__).parents[1] / "app" / "frontend" / "static"
|
|
|
|
|
|
class FrontendShellTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.app = FastAPI()
|
|
self.app.include_router(router)
|
|
self.client = TestClient(self.app)
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
|
|
def authenticate(self, permissions=None, role="Read Only"):
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": "00000000-0000-0000-0000-000000000321",
|
|
"username": "operator",
|
|
"display_name": "Floor Operator",
|
|
"role": role,
|
|
"permissions": list(
|
|
permissions
|
|
or (
|
|
"dashboard.view",
|
|
"locations.view",
|
|
"repairs.view",
|
|
"miners.view",
|
|
)
|
|
),
|
|
}
|
|
|
|
def test_routes_require_authentication(self):
|
|
for path in ROUTES:
|
|
with self.subTest(path=path):
|
|
self.assertEqual(self.client.get(path).status_code, 401)
|
|
|
|
def test_all_direct_routes_render_authenticated_shell(self):
|
|
self.authenticate()
|
|
for path in ROUTES:
|
|
with self.subTest(path=path):
|
|
response = self.client.get(path)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("MinerOps", response.text)
|
|
self.assertIn(
|
|
"Operational Intelligence for Mining",
|
|
response.text,
|
|
)
|
|
self.assertIn("Floor Operator", response.text)
|
|
self.assertIn(
|
|
'aria-label="Primary navigation"',
|
|
response.text,
|
|
)
|
|
|
|
def test_navigation_connects_all_internal_alpha_views(self):
|
|
self.authenticate()
|
|
response = self.client.get("/dashboard")
|
|
for path in ROUTES:
|
|
self.assertIn(f'href="{path}"', response.text)
|
|
self.assertIn('aria-current="page"', response.text)
|
|
|
|
def test_each_route_requires_its_specific_permission(self):
|
|
cases = (
|
|
("/dashboard", "dashboard.view"),
|
|
("/monitoring/map", "locations.view"),
|
|
("/operations/repairs", "repairs.view"),
|
|
("/assets/miners", "miners.view"),
|
|
)
|
|
for path, permission in cases:
|
|
with self.subTest(path=path):
|
|
self.authenticate(permissions=("unrelated.permission",))
|
|
self.assertEqual(self.client.get(path).status_code, 403)
|
|
self.authenticate(permissions=(permission,))
|
|
self.assertEqual(self.client.get(path).status_code, 200)
|
|
|
|
def test_main_authentication_gate_redirects_to_existing_login(self):
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
install_identity_middleware(app)
|
|
with TestClient(app) as client, patch(
|
|
"app.identity.request_user_or_none",
|
|
return_value=None,
|
|
):
|
|
for path in ROUTES:
|
|
with self.subTest(path=path):
|
|
response = client.get(path, follow_redirects=False)
|
|
self.assertEqual(response.status_code, 303)
|
|
self.assertTrue(
|
|
response.headers["location"].startswith(
|
|
"/login?next="
|
|
)
|
|
)
|
|
|
|
def test_page_permission_mapping_covers_new_routes(self):
|
|
expected = {
|
|
"/dashboard": "dashboard.view",
|
|
"/monitoring/map": "locations.view",
|
|
"/operations/repairs": "repairs.view",
|
|
"/assets/miners": "miners.view",
|
|
}
|
|
for path, permission in expected.items():
|
|
with self.subTest(path=path):
|
|
self.assertTrue(protected_page(path))
|
|
self.assertEqual(
|
|
required_page_permission(path),
|
|
permission,
|
|
)
|
|
|
|
def test_dashboard_does_not_fabricate_operational_values(self):
|
|
self.authenticate()
|
|
response = self.client.get("/dashboard")
|
|
self.assertIn(
|
|
"Live health summary not yet connected",
|
|
response.text,
|
|
)
|
|
self.assertIn("No events are fabricated", response.text)
|
|
self.assertNotIn("100 TH/s", response.text)
|
|
|
|
def test_map_is_a_structured_operational_view(self):
|
|
self.authenticate()
|
|
response = self.client.get("/monitoring/map")
|
|
for label in (
|
|
"Site",
|
|
"Room",
|
|
"Operational Status",
|
|
"Miner details",
|
|
"Find miner",
|
|
):
|
|
self.assertIn(label, response.text)
|
|
for value in (
|
|
'id="mapSite"', 'id="mapRoom"', 'id="mapLayer"',
|
|
'id="mapProblemsOnly"',
|
|
'id="mapViewport"', 'id="mapInspector"',
|
|
'/assets/minerops-mine-map.js',
|
|
):
|
|
self.assertIn(value, response.text)
|
|
self.assertNotIn("Live layout and telemetry data are not", response.text)
|
|
|
|
def test_map_script_dom_contract_matches_rendered_page(self):
|
|
self.authenticate()
|
|
response = self.client.get("/monitoring/map")
|
|
javascript = (STATIC_ROOT / "mine-map.js").read_text(encoding="utf-8")
|
|
declared_ids = re.search(
|
|
r"const ids = \[(.*?)\];",
|
|
javascript,
|
|
re.DOTALL,
|
|
)
|
|
self.assertIsNotNone(declared_ids)
|
|
required = set(
|
|
re.findall(
|
|
r'"([A-Za-z][A-Za-z0-9]+)"',
|
|
declared_ids.group(1),
|
|
)
|
|
)
|
|
required.update(
|
|
re.findall(
|
|
r'document\.getElementById\("([^"]+)"\)',
|
|
javascript,
|
|
)
|
|
)
|
|
rendered = set(re.findall(r'\bid="([^"]+)"', response.text))
|
|
self.assertEqual(required - rendered, set())
|
|
|
|
def test_data_views_include_all_request_states(self):
|
|
self.authenticate()
|
|
for path in ("/operations/repairs", "/assets/miners"):
|
|
with self.subTest(path=path):
|
|
response_text = self.client.get(path).text
|
|
for state in (
|
|
"loading",
|
|
"empty",
|
|
"error",
|
|
"denied",
|
|
"ready",
|
|
):
|
|
self.assertIn(
|
|
f'data-view-state="{state}"',
|
|
response_text,
|
|
)
|
|
|
|
def test_frontend_assets_are_responsive_and_accessible(self):
|
|
css = (STATIC_ROOT / "app.css").read_text(encoding="utf-8")
|
|
javascript = (STATIC_ROOT / "app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("@media (max-width: 560px)", css)
|
|
self.assertIn(":focus-visible", css)
|
|
self.assertIn(".skip-link", css)
|
|
repair_center = (STATIC_ROOT / "repair-center.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn('response.status === 401', repair_center)
|
|
self.assertIn('response.status === 403', repair_center)
|
|
self.assertIn('credentials: "same-origin"', javascript)
|
|
|
|
def test_static_assets_are_served_without_an_api_route(self):
|
|
css_response = self.client.get("/assets/minerops-app.css")
|
|
script_response = self.client.get("/assets/minerops-app.js")
|
|
self.assertEqual(css_response.status_code, 200)
|
|
self.assertEqual(
|
|
css_response.headers["content-type"],
|
|
"text/css; charset=utf-8",
|
|
)
|
|
self.assertEqual(script_response.status_code, 200)
|
|
paths = {route.path for route in router.routes}
|
|
self.assertFalse(any(path.startswith("/api/") for path in paths))
|
|
|
|
def test_live_app_prefers_new_shell_and_preserves_legacy_pages(self):
|
|
from app.main import app
|
|
|
|
first_handlers = {}
|
|
for route in app.routes:
|
|
if route.path not in first_handlers:
|
|
first_handlers[route.path] = route.endpoint
|
|
|
|
for path in ROUTES:
|
|
with self.subTest(path=path):
|
|
self.assertEqual(
|
|
first_handlers[path].__module__,
|
|
"app.frontend.router",
|
|
)
|
|
for path in (
|
|
"/repairs",
|
|
"/locations",
|
|
"/miners/{miner_id}/detail",
|
|
):
|
|
with self.subTest(legacy_path=path):
|
|
self.assertIn(path, first_handlers)
|
|
|
|
|
|
def test_safe_next_path_accepts_only_protected_local_pages(self):
|
|
valid = (
|
|
"/dashboard",
|
|
"/assets/miners?query=S19%20Pro",
|
|
"/miners/00000000-0000-0000-0000-000000000123/detail",
|
|
)
|
|
for value in valid:
|
|
with self.subTest(valid=value):
|
|
self.assertEqual(safe_next_path(value), value)
|
|
|
|
invalid = (
|
|
None,
|
|
"",
|
|
"https://example.invalid/dashboard",
|
|
"//example.invalid/dashboard",
|
|
r"\\example.invalid\dashboard",
|
|
"/../login",
|
|
"/assets/%2f%2fexample.invalid",
|
|
"/dashboard#fragment",
|
|
"/login",
|
|
"/api/inventory/miners",
|
|
" /dashboard",
|
|
"/dashboard\n",
|
|
)
|
|
for value in invalid:
|
|
with self.subTest(invalid=value):
|
|
self.assertEqual(safe_next_path(value), "/dashboard")
|
|
|
|
def test_login_redirect_preserves_direct_link_query(self):
|
|
app = FastAPI()
|
|
app.include_router(identity_router)
|
|
app.include_router(router)
|
|
install_identity_middleware(app)
|
|
with TestClient(app) as client, patch(
|
|
"app.identity.request_user_or_none",
|
|
return_value=None,
|
|
):
|
|
response = client.get(
|
|
"/assets/miners?query=S19%20Pro",
|
|
follow_redirects=False,
|
|
)
|
|
self.assertEqual(response.status_code, 303)
|
|
location = urlsplit(response.headers["location"])
|
|
self.assertEqual(location.path, "/login")
|
|
self.assertEqual(
|
|
parse_qs(location.query)["next"],
|
|
["/assets/miners?query=S19%20Pro"],
|
|
)
|
|
|
|
def test_authenticated_login_redirect_rejects_unsafe_next(self):
|
|
app = FastAPI()
|
|
app.include_router(identity_router)
|
|
install_identity_middleware(app)
|
|
with TestClient(app) as client, patch(
|
|
"app.identity.request_user_or_none",
|
|
return_value={"id": "user"},
|
|
):
|
|
for value in (
|
|
"https://example.invalid",
|
|
"//example.invalid",
|
|
r"\\example.invalid\dashboard",
|
|
"/../login",
|
|
):
|
|
with self.subTest(value=value):
|
|
response = client.get(
|
|
"/login",
|
|
params={"next": value},
|
|
follow_redirects=False,
|
|
)
|
|
self.assertEqual(response.status_code, 303)
|
|
self.assertEqual(
|
|
response.headers["location"],
|
|
"/dashboard",
|
|
)
|
|
|
|
def test_login_template_escapes_safe_query_destination(self):
|
|
app = FastAPI()
|
|
app.include_router(identity_router)
|
|
hostile = '/assets/miners?query="><script>alert(1)</script>'
|
|
with TestClient(app) as client:
|
|
response = client.get("/login", params={"next": hostile})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertNotIn(hostile, response.text)
|
|
self.assertIn(""><script>", response.text)
|
|
self.assertNotIn("new URLSearchParams(location.search)", response.text)
|
|
|
|
def test_user_values_are_html_escaped(self):
|
|
hostile = '<img src=x onerror=alert(1)>'
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": "00000000-0000-0000-0000-000000000321",
|
|
"username": "operator",
|
|
"display_name": hostile,
|
|
"role": "Read Only",
|
|
"permissions": ["dashboard.view", "miners.view"],
|
|
}
|
|
response = self.client.get("/dashboard")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertNotIn(hostile, response.text)
|
|
self.assertIn("<img src=x onerror=alert(1)>", response.text)
|
|
|
|
def test_navigation_and_search_are_permission_aware(self):
|
|
self.authenticate(permissions=("dashboard.view",))
|
|
response = self.client.get("/dashboard")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('href="/dashboard"', response.text)
|
|
for path in ROUTES[1:]:
|
|
self.assertNotIn(f'href="{path}"', response.text)
|
|
self.assertNotIn('id="globalSearchForm"', response.text)
|
|
self.assertIn('data-can-view-miners="false"', response.text)
|
|
|
|
def test_assets_disable_caching_and_missing_assets_are_404(self):
|
|
for path in (
|
|
"/assets/minerops-app.css",
|
|
"/assets/minerops-app.js",
|
|
"/assets/minerops-mine-map.js",
|
|
"/assets/minerops-fleet.js",
|
|
):
|
|
with self.subTest(path=path):
|
|
response = self.client.get(path)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.headers["cache-control"], "no-store")
|
|
self.assertEqual(
|
|
self.client.get("/assets/missing-minerops-file.js").status_code,
|
|
404,
|
|
)
|
|
|
|
def test_trailing_slash_redirects_are_consistent(self):
|
|
self.authenticate()
|
|
for path in ROUTES:
|
|
with self.subTest(path=path):
|
|
response = self.client.get(
|
|
path + "/",
|
|
follow_redirects=False,
|
|
)
|
|
self.assertEqual(response.status_code, 307)
|
|
self.assertEqual(
|
|
urlsplit(response.headers["location"]).path,
|
|
path,
|
|
)
|
|
|
|
def test_shell_markup_has_accessibility_landmarks(self):
|
|
self.authenticate()
|
|
for path in ROUTES:
|
|
with self.subTest(path=path):
|
|
text = self.client.get(path).text
|
|
self.assertEqual(text.count("<h1>"), 1)
|
|
self.assertIn('class="skip-link"', text)
|
|
self.assertIn('<main class="main-content"', text)
|
|
self.assertIn('aria-label="Primary navigation"', text)
|
|
self.assertIn('aria-controls="primaryNav"', text)
|
|
|
|
def test_javascript_uses_safe_dom_and_strict_failure_states(self):
|
|
javascript = (STATIC_ROOT / "app.js").read_text(encoding="utf-8")
|
|
repair_center = (STATIC_ROOT / "repair-center.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
for source in (javascript, repair_center):
|
|
for unsafe in ("innerHTML", "insertAdjacentHTML", "document.write"):
|
|
self.assertNotIn(unsafe, source)
|
|
self.assertIn("window.location.replace", source)
|
|
self.assertNotIn("/api/repairs/queue", javascript)
|
|
self.assertIn("new AbortController()", repair_center)
|
|
self.assertIn("cell.textContent = text(content)", repair_center)
|
|
self.assertIn("encodeURIComponent(item.miner_id)", repair_center)
|
|
self.assertIn("invalid Repair Center response", repair_center)
|
|
|
|
def test_fleet_page_is_semantic_responsive_and_permission_aware(self):
|
|
self.authenticate()
|
|
response = self.client.get("/assets/miners")
|
|
self.assertEqual(response.status_code, 200)
|
|
for value in (
|
|
'id="fleetFilters"', 'id="fleetRows"', 'id="fleetResultSummary"',
|
|
'<table class="fleet-table">', '<th>Operational</th>',
|
|
'id="fleetPrevious"', 'id="fleetNext"',
|
|
'/assets/minerops-fleet.js', 'data-can-view-locations="true"',
|
|
'id="fleetSite"', 'id="fleetRoom"',
|
|
):
|
|
self.assertIn(value, response.text)
|
|
|
|
self.authenticate(permissions=("miners.view",))
|
|
restricted = self.client.get("/assets/miners").text
|
|
self.assertIn('data-can-view-locations="false"', restricted)
|
|
self.assertNotIn('id="fleetSite"', restricted)
|
|
self.assertNotIn('id="fleetRoom"', restricted)
|
|
|
|
def test_fleet_reset_action_has_a_visible_label(self):
|
|
self.authenticate()
|
|
response = self.client.get("/assets/miners")
|
|
self.assertIn(
|
|
'<button class="quiet-button" id="fleetReset" '
|
|
'type="button">Reset</button>',
|
|
response.text,
|
|
)
|
|
css = (STATIC_ROOT / "app.css").read_text(encoding="utf-8")
|
|
self.assertIn(
|
|
".fleet-filter-actions .quiet-button { color: var(--ink); }",
|
|
css,
|
|
)
|
|
|
|
def test_fleet_script_dom_contract_matches_authorized_page(self):
|
|
self.authenticate()
|
|
response = self.client.get("/assets/miners")
|
|
source = (STATIC_ROOT / "fleet.js").read_text(encoding="utf-8")
|
|
declared = re.search(r"const ids = \[(.*?)\];", source, re.DOTALL)
|
|
self.assertIsNotNone(declared)
|
|
required = set(re.findall(r'"([A-Za-z][A-Za-z0-9]+)"', declared.group(1)))
|
|
required.update(("fleetSite", "fleetRoom"))
|
|
rendered = set(re.findall(r'\bid="([^"]+)"', response.text))
|
|
self.assertEqual(required - rendered, set())
|
|
|
|
def test_fleet_script_is_bounded_read_only_and_xss_safe(self):
|
|
source = (STATIC_ROOT / "fleet.js").read_text(encoding="utf-8")
|
|
self.assertEqual(source.count('fetch("/api/inventory/miners?"'), 1)
|
|
self.assertNotIn("/api/location/map", source)
|
|
self.assertNotIn("setInterval", source)
|
|
for method in ('method: "POST"', 'method: "PATCH"', 'method: "PUT"', 'method: "DELETE"'):
|
|
self.assertNotIn(method, source)
|
|
for unsafe in ("innerHTML", "insertAdjacentHTML", "document.write"):
|
|
self.assertNotIn(unsafe, source)
|
|
self.assertIn("textContent", source)
|
|
self.assertIn('"/miners/" + encodeURIComponent(miner.id) + "/detail"', source)
|
|
self.assertIn('"/monitoring/map?" + params.toString()', source)
|
|
self.assertNotRegex(source.lower(), r"stale.{0,20}(threshold|minutes|hours)")
|
|
self.assertNotIn("lowHashrate", source)
|
|
|
|
def test_fleet_css_has_status_text_mobile_cards_and_visible_focus(self):
|
|
css = (STATIC_ROOT / "app.css").read_text(encoding="utf-8")
|
|
self.assertIn('.fleet-row[data-status="offline"]', css)
|
|
self.assertIn("content: attr(data-label)", css)
|
|
self.assertIn(".fleet-pagination", css)
|
|
self.assertIn(":focus-visible", css)
|
|
|
|
def test_styles_have_no_motion_to_reduce(self):
|
|
css = (STATIC_ROOT / "app.css").read_text(encoding="utf-8")
|
|
self.assertNotIn("animation:", css)
|
|
self.assertNotIn("transition:", css)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|