b6a14ca8e0
The fleet row named the wave by its short name only ("PILOT-2") while the
detail view and the filter carry the full label ("P2 — PILOT-2"). An operator
matching a row against a work order speaks in codes, so the row carries the
full label now -- still named once, which the row test asserts.
Also takes the console agent's leftover: a dead state_style helper removed and
the unit convention on ALL_TOKENS documented (lengths are unitless; Tk padding
is already pixels, the parity test strips the CSS unit).
1292 lines
58 KiB
Python
1292 lines
58 KiB
Python
#!/usr/bin/env python3
|
||
"""web panel — static and contract tests (stdlib unittest).
|
||
|
||
No port is bound, no browser is opened and no server process is started: the
|
||
whole HTTP surface is covered through ``web.panel.route()``, which is a pure
|
||
function, and every rendering claim is checked on the rendered bytes.
|
||
|
||
What these tests defend (the things that could silently regress):
|
||
|
||
* the Turkish status vocabulary still matches docs/DESIGN.md exactly, so the
|
||
device console and the central panel cannot drift apart;
|
||
* every user-visible string is non-empty, Turkish and free of placeholders;
|
||
* no external URL, no webfont, no hex colour and no new colour token anywhere
|
||
in the templates or stylesheets: colour comes from tokens.css only;
|
||
* the panel imports nothing outside the standard library and can therefore be
|
||
installed on the air-gapped fleet it manages;
|
||
* only GET/HEAD are served, static assets are allow-listed, and a full sweep
|
||
over every endpoint leaves the filesystem untouched;
|
||
* the first viewport is ordered and populated as docs/DESIGN.md requires.
|
||
|
||
Run:
|
||
python3 tests/test_web_static.py -v
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import contextlib
|
||
import io
|
||
import json
|
||
import re
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||
WEB_DIR = REPO_ROOT / "web"
|
||
TEMPLATES_DIR = WEB_DIR / "templates"
|
||
STATIC_DIR = WEB_DIR / "static"
|
||
TOKENS_CSS = STATIC_DIR / "tokens.css"
|
||
DESIGN_DOC = REPO_ROOT / "docs" / "DESIGN.md"
|
||
|
||
if str(REPO_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
|
||
import web # noqa: E402 (path set above)
|
||
from web import fleet_data, panel, render, server # noqa: E402
|
||
|
||
# The panel must be the one in this checkout, not some unrelated "web" package.
|
||
assert Path(web.__file__).resolve().parent == WEB_DIR, f"yanlis paket: {web.__file__}"
|
||
|
||
NOW = "2026-09-16T22:00:00+03:00"
|
||
STATE_WORDS = ("KURULDU", "KAYITLI", "HAZIR", "HATA", "YOK")
|
||
|
||
# Everything an air-gapped, framework-free surface must never contain.
|
||
FORBIDDEN_ASSET_MARKERS = (
|
||
"http://",
|
||
"https://",
|
||
"//cdn",
|
||
"cdn.jsdelivr",
|
||
"cdnjs",
|
||
"unpkg",
|
||
"fonts.googleapis",
|
||
"fonts.gstatic",
|
||
"@import",
|
||
"@font-face",
|
||
"src=",
|
||
"xlink:href=",
|
||
)
|
||
FORBIDDEN_IMPORT_ROOTS = ("fastapi", "flask", "jinja2", "uvicorn", "django", "werkzeug", "starlette", "requests")
|
||
FORBIDDEN_IMPORTS = ("urllib.request", "http.client", "socket", "subprocess", "shutil", "ftplib")
|
||
FORBIDDEN_SOURCE_MARKERS = ("os.system(", "subprocess.", "urllib.request", "requests.", "eval(", "exec(")
|
||
|
||
HEX_COLOUR = re.compile(r"#[0-9a-fA-F]{3,8}\b")
|
||
CUSTOM_PROPERTY = re.compile(r"^\s*--[A-Za-z0-9-]+\s*:", re.M)
|
||
# A custom-property declaration, e.g. `--state-ready: var(--ok);`
|
||
CUSTOM_PROPERTY_DECL = re.compile(r"^\s*--([A-Za-z0-9-]+)\s*:\s*([^;]+);", re.M)
|
||
# A single `var(--name)` reference inside a value.
|
||
VAR_REFERENCE = re.compile(r"var\(\s*--([A-Za-z0-9-]+)\s*\)")
|
||
DEFINED_TOKEN = re.compile(r"^\s*(--[a-z0-9-]+)\s*:", re.M)
|
||
USED_TOKEN = re.compile(r"var\((--[a-z0-9-]+)\)")
|
||
TEXT_TAGS = ("td", "th", "span", "h1", "h2", "h3", "p", "li", "option", "dd", "dt", "button", "a", "caption", "label")
|
||
ELEMENT = re.compile(r"<(%s)\b[^>]*>(.*?)</\1>" % "|".join(TEXT_TAGS), re.S)
|
||
ROW_BADGE = re.compile(r'<span class="badge badge--state-([a-z]+)"[^>]*>([^<]*)</span>')
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# helpers
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def token_values(text: str) -> dict[str, str]:
|
||
"""Every custom-property declaration in a stylesheet, name -> raw value."""
|
||
return {
|
||
name: value.strip()
|
||
for name, value in CUSTOM_PROPERTY_DECL.findall(text)
|
||
}
|
||
|
||
|
||
def resolved_colour(name: str, values: dict[str, str], limit: int = 12) -> str | None:
|
||
"""Follow ``var(--x)`` aliases to the colour a token actually paints.
|
||
|
||
Comparing raw declarations is not enough to prove two tokens differ: a
|
||
mutation writing ``--state-provisioned: var(--accent)`` leaves the strings
|
||
unequal while painting exactly the same blue. Only the resolved value is
|
||
evidence.
|
||
"""
|
||
seen: set[str] = set()
|
||
while limit > 0:
|
||
value = values.get(name)
|
||
if value is None:
|
||
return None
|
||
reference = VAR_REFERENCE.fullmatch(value)
|
||
if reference is None:
|
||
return value
|
||
name = reference.group(1)
|
||
if name in seen:
|
||
return value # a cycle: report what was declared rather than loop
|
||
seen.add(name)
|
||
limit -= 1
|
||
return values.get(name)
|
||
|
||
|
||
def design_vocabulary() -> list[tuple[str, str]]:
|
||
"""Read the status vocabulary table out of docs/DESIGN.md."""
|
||
text = DESIGN_DOC.read_text(encoding="utf-8")
|
||
row = re.compile(r"^\|\s*`([^`]+)`\s*\|\s*`(--state-[a-z-]+)`\s*\|", re.M)
|
||
return row.findall(text)
|
||
|
||
|
||
def text_files() -> list[Path]:
|
||
"""Templates and stylesheets: everything the browser receives from disk."""
|
||
return sorted(TEMPLATES_DIR.glob("*.html")) + sorted(STATIC_DIR.glob("*.css"))
|
||
|
||
|
||
def empty_elements(markup: str) -> list[str]:
|
||
"""Visible elements that carry no text at all (an empty label is a bug)."""
|
||
problems: list[str] = []
|
||
for match in ELEMENT.finditer(markup):
|
||
inner = re.sub(r"<[^>]+>", "", match.group(2))
|
||
if not inner.strip():
|
||
problems.append(match.group(0)[:140])
|
||
return problems
|
||
|
||
|
||
def tree_fingerprint(root: Path) -> tuple:
|
||
"""Sizes and mtimes of every source file, ignoring interpreter caches."""
|
||
entries = []
|
||
for path in sorted(root.rglob("*")):
|
||
if "__pycache__" in path.parts or not path.is_file():
|
||
continue
|
||
stat = path.stat()
|
||
entries.append((str(path.relative_to(root)), stat.st_size, stat.st_mtime_ns))
|
||
return tuple(entries)
|
||
|
||
|
||
class PanelCase(unittest.TestCase):
|
||
"""Shared fixture: the labelled demo snapshot and both privilege levels."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.snapshot = fleet_data.demo_snapshot(NOW)
|
||
cls.read_only = render.privilege_from_level("read-only")
|
||
cls.full = render.privilege_from_level("full")
|
||
cls.info = render.ServerInfo(
|
||
host="127.0.0.1",
|
||
port=8787,
|
||
privilege=cls.read_only,
|
||
python_version="3.13.0",
|
||
started_at=NOW,
|
||
)
|
||
cls.context = panel.PanelContext(
|
||
snapshot=cls.snapshot, privilege=cls.read_only, server_info=cls.info
|
||
)
|
||
full_info = render.ServerInfo(
|
||
host="127.0.0.1",
|
||
port=8787,
|
||
privilege=cls.full,
|
||
python_version="3.13.0",
|
||
started_at=NOW,
|
||
)
|
||
cls.full_context = panel.PanelContext(
|
||
snapshot=cls.snapshot, privilege=cls.full, server_info=full_info
|
||
)
|
||
|
||
# -- request helpers ---------------------------------------------------
|
||
def get(self, target: str, method: str = "GET", context: panel.PanelContext | None = None):
|
||
return panel.route(context or self.context, method, target)
|
||
|
||
def page(self, target: str = "/", method: str = "GET", context: panel.PanelContext | None = None) -> str:
|
||
response = self.get(target, method, context)
|
||
self.assertEqual(response.status, 200, f"{target}: {response.status} (beklenen 200)")
|
||
return response.body.decode("utf-8")
|
||
|
||
def list_region(self, page: str) -> str:
|
||
"""The device table only, so about/detail badges never leak into a scan."""
|
||
start = page.index('data-region="list"')
|
||
end = page.index('data-region="detail"')
|
||
return page[start:end]
|
||
|
||
def row_words(self, page: str) -> list[str]:
|
||
return [word for _, word in ROW_BADGE.findall(self.list_region(page))]
|
||
|
||
def snapshot_from(self, document: object, *, now: str = NOW, filename: str = "fleet.json"):
|
||
"""Write a temporary inventory document and load it through the loader."""
|
||
directory = tempfile.TemporaryDirectory(prefix="bf-panel-test-")
|
||
self.addCleanup(directory.cleanup)
|
||
path = Path(directory.name) / filename
|
||
if isinstance(document, str):
|
||
path.write_text(document, encoding="utf-8")
|
||
else:
|
||
path.write_text(json.dumps(document, ensure_ascii=False), encoding="utf-8")
|
||
return fleet_data.load_fleet(path, env={}, now=now)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# vocabulary and design contract
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestVocabularyMatchesDesign(PanelCase):
|
||
def test_vocabulary_matches_docs_design_exactly(self) -> None:
|
||
"""One word per state, shared with the device console; no fifth word."""
|
||
self.assertEqual(
|
||
design_vocabulary(),
|
||
[(spec.word, spec.token) for spec in fleet_data.STATE_VOCABULARY],
|
||
)
|
||
|
||
def test_design_document_still_has_the_five_words(self) -> None:
|
||
words = [word for word, _ in design_vocabulary()]
|
||
self.assertEqual(words, list(STATE_WORDS))
|
||
|
||
def test_every_word_is_upper_case_turkish_and_non_empty(self) -> None:
|
||
for spec in fleet_data.STATE_VOCABULARY:
|
||
with self.subTest(state=spec.state):
|
||
self.assertTrue(spec.word.strip(), spec.state)
|
||
self.assertEqual(spec.word, spec.word.upper())
|
||
self.assertTrue(spec.meaning.strip(), spec.word)
|
||
self.assertNotIn("None", spec.meaning)
|
||
|
||
def test_tokens_exist_in_tokens_css(self) -> None:
|
||
css = TOKENS_CSS.read_text(encoding="utf-8")
|
||
for spec in fleet_data.STATE_VOCABULARY:
|
||
with self.subTest(state=spec.state):
|
||
self.assertIn(f"{spec.token}:", css)
|
||
self.assertIn(f"{spec.soft_token}:", css)
|
||
|
||
def test_non_ready_states_sort_before_ready(self) -> None:
|
||
self.assertEqual(fleet_data.STATE_ORDER[-1], "READY")
|
||
self.assertEqual(set(fleet_data.STATE_ORDER), set(fleet_data.STATE_BY_NAME))
|
||
|
||
def test_demo_data_exercises_every_state_word(self) -> None:
|
||
counts = self.snapshot.counts
|
||
self.assertEqual(list(counts), list(STATE_WORDS))
|
||
for word, count in counts.items():
|
||
with self.subTest(word=word):
|
||
self.assertGreaterEqual(count, 1, f"{word} demo veride yok; sözlük test edilemez")
|
||
|
||
def test_demo_data_exercises_every_wave(self) -> None:
|
||
counts = fleet_data.wave_counts(self.snapshot.devices)
|
||
self.assertEqual(list(counts), [wave.code for wave in fleet_data.WAVES])
|
||
for code, count in counts.items():
|
||
with self.subTest(wave=code):
|
||
self.assertGreaterEqual(count, 1)
|
||
|
||
def test_waves_match_the_rollout_document(self) -> None:
|
||
"""docs/18: LAB(2) -> P1(5) -> P2(20) -> W1(50) -> W2(100) -> PROD."""
|
||
self.assertEqual(
|
||
[(wave.code, wave.target) for wave in fleet_data.WAVES],
|
||
[("LAB", 2), ("P1", 5), ("P2", 20), ("W1", 50), ("W2", 100), ("PROD", 523)],
|
||
)
|
||
|
||
def test_four_documented_channels(self) -> None:
|
||
self.assertEqual(
|
||
[channel.label for channel in fleet_data.CHANNELS],
|
||
["SSH-over-WG", "xRDP", "RustDesk", "MeshCentral"],
|
||
)
|
||
self.assertEqual(
|
||
[channel.key for channel in fleet_data.CHANNELS],
|
||
["ssh_over_wg", "xrdp", "rustdesk", "meshcentral"],
|
||
)
|
||
|
||
def test_channel_status_words_are_turkish_and_non_empty(self) -> None:
|
||
for status, word in fleet_data.CHANNEL_STATUS_WORDS.items():
|
||
with self.subTest(status=status):
|
||
self.assertTrue(word.strip())
|
||
self.assertNotIn("None", word)
|
||
self.assertEqual(set(fleet_data.CHANNEL_STATUS_WORDS), set(fleet_data.CHANNEL_STATUSES))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# no external assets, no invented colour
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestAssetsAndColour(PanelCase):
|
||
def test_no_external_url_or_remote_asset_in_any_template_or_stylesheet(self) -> None:
|
||
for path in text_files():
|
||
text = path.read_text(encoding="utf-8").lower()
|
||
for marker in FORBIDDEN_ASSET_MARKERS:
|
||
with self.subTest(file=path.name, marker=marker):
|
||
self.assertNotIn(marker, text)
|
||
|
||
def test_no_javascript_ships_with_the_panel(self) -> None:
|
||
for path in text_files():
|
||
text = path.read_text(encoding="utf-8").lower()
|
||
with self.subTest(file=path.name):
|
||
self.assertNotIn("<script", text)
|
||
self.assertNotIn("onclick=", text)
|
||
self.assertNotIn("javascript:", text)
|
||
|
||
def test_hex_colour_literals_live_only_in_tokens_css(self) -> None:
|
||
for path in text_files():
|
||
text = path.read_text(encoding="utf-8")
|
||
if path == TOKENS_CSS:
|
||
self.assertTrue(HEX_COLOUR.search(text), "tokens.css palet içermiyor")
|
||
continue
|
||
with self.subTest(file=path.name):
|
||
self.assertEqual(HEX_COLOUR.findall(text), [], f"{path.name} hex renk içeriyor")
|
||
|
||
def test_templates_and_app_css_declare_no_custom_property(self) -> None:
|
||
for path in text_files():
|
||
if path == TOKENS_CSS:
|
||
continue
|
||
text = path.read_text(encoding="utf-8")
|
||
with self.subTest(file=path.name):
|
||
self.assertEqual(CUSTOM_PROPERTY.findall(text), [], f"{path.name} belirteç tanımlıyor")
|
||
|
||
def test_every_colour_token_used_is_defined_in_tokens_css(self) -> None:
|
||
defined = set(DEFINED_TOKEN.findall(TOKENS_CSS.read_text(encoding="utf-8")))
|
||
self.assertTrue(defined)
|
||
used: dict[str, str] = {}
|
||
for path in text_files():
|
||
if path == TOKENS_CSS:
|
||
continue
|
||
for token in USED_TOKEN.findall(path.read_text(encoding="utf-8")):
|
||
used[token] = path.name
|
||
unknown = {token: name for token, name in used.items() if token not in defined}
|
||
self.assertEqual(unknown, {}, f"tokens.css'te tanımsız belirteç: {unknown}")
|
||
self.assertTrue(used, "hiç belirteç kullanılmıyor")
|
||
|
||
def test_the_action_accent_is_not_a_device_state_colour(self) -> None:
|
||
"""The accent means "the action you came to take" and "the selected row".
|
||
|
||
A state that borrows it makes a KURULDU badge and a selected row the same
|
||
blue, so the list can no longer say which of the two it is showing. Aliases
|
||
are resolved before comparing: a token may chain (``--state-ready`` to
|
||
``--ok``), and ``var(--accent)`` is the accent whatever the string says.
|
||
"""
|
||
values = token_values(TOKENS_CSS.read_text(encoding="utf-8"))
|
||
self.assertIn("accent", values, "tokens.css'te --accent yok")
|
||
accent = resolved_colour("accent", values)
|
||
states = {n: v for n, v in values.items() if n.startswith("state-")}
|
||
self.assertTrue(states, "tokens.css'te cihaz durumu belirteci yok")
|
||
for name in states:
|
||
with self.subTest(token=name):
|
||
resolved = resolved_colour(name, values)
|
||
self.assertIsNotNone(resolved, f"--{name} çözümlenemedi")
|
||
self.assertNotEqual(
|
||
resolved, accent,
|
||
f"--{name} ile --accent aynı rengi veriyor ({resolved}); "
|
||
"seçili satır ile durum rozeti ayırt edilemez",
|
||
)
|
||
|
||
def test_wave_is_rendered_once_and_not_as_the_bare_code_plus_full_label(self) -> None:
|
||
"""``detail.html`` printed ``$wave_code`` and then ``$wave_label``, and the
|
||
label already begins with the code, so one device read "P2 P2 — PILOT-2"
|
||
(the list row was the same). A wave cell must name each wave once.
|
||
"""
|
||
device = next(
|
||
d for d in self.snapshot.devices if d.wave_code == "P2"
|
||
)
|
||
target = f"/?cihaz={device.device_id}"
|
||
body = panel.route(self.context, "GET", target).body.decode("utf-8")
|
||
self.assertNotIn("$wave_name", body, "şablon değişkeni değiştirilmedi")
|
||
self.assertNotIn("$wave_code", body, "şablon değişkeni değiştirilmedi")
|
||
self.assertNotIn(device.wave_code * 2, body, f"dalga kodu iki kez basılıyor: {device.wave_code}")
|
||
self.assertIn(device.wave_label, body, "dalga tam etiketi görünmüyor")
|
||
|
||
def test_only_local_stylesheets_are_linked(self) -> None:
|
||
base = (TEMPLATES_DIR / "base.html").read_text(encoding="utf-8")
|
||
linked = re.findall(r'<link[^>]*href="([^"]+)"', base)
|
||
self.assertEqual(linked, ["/static/tokens.css", "/static/app.css"])
|
||
|
||
def test_templates_are_utf8_with_turkish_characters(self) -> None:
|
||
for path in sorted(TEMPLATES_DIR.glob("*.html")):
|
||
text = path.read_text(encoding="utf-8")
|
||
with self.subTest(file=path.name):
|
||
self.assertTrue(text.strip())
|
||
base = (TEMPLATES_DIR / "base.html").read_text(encoding="utf-8")
|
||
self.assertIn('<html lang="tr"', base)
|
||
self.assertIn('<meta charset="utf-8">', base)
|
||
|
||
def test_state_badge_styles_use_the_shared_tokens(self) -> None:
|
||
css = (STATIC_DIR / "app.css").read_text(encoding="utf-8")
|
||
for spec in fleet_data.STATE_VOCABULARY:
|
||
with self.subTest(state=spec.state):
|
||
self.assertIn(f".badge--state-{spec.token_suffix}", css)
|
||
self.assertIn("var(--accent-soft)", css)
|
||
self.assertIn(".fleet-row--selected", css)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# standard library only: the fleet installs offline
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestStdlibOnly(PanelCase):
|
||
def web_sources(self) -> list[Path]:
|
||
return sorted(WEB_DIR.glob("*.py"))
|
||
|
||
def test_web_modules_exist(self) -> None:
|
||
names = {path.name for path in self.web_sources()}
|
||
self.assertEqual(
|
||
names, {"__init__.py", "fleet_data.py", "render.py", "panel.py", "server.py"}
|
||
)
|
||
|
||
def test_every_import_resolves_to_the_standard_library(self) -> None:
|
||
stdlib = set(sys.stdlib_module_names)
|
||
for path in self.web_sources():
|
||
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.Import):
|
||
roots = [alias.name.split(".")[0] for alias in node.names]
|
||
elif isinstance(node, ast.ImportFrom):
|
||
if node.level:
|
||
continue
|
||
roots = [(node.module or "").split(".")[0]]
|
||
else:
|
||
continue
|
||
for root in roots:
|
||
with self.subTest(file=path.name, module=root):
|
||
self.assertIn(root, stdlib | {"web"})
|
||
|
||
def test_no_third_party_web_framework_is_referenced(self) -> None:
|
||
for path in self.web_sources():
|
||
text = path.read_text(encoding="utf-8").lower()
|
||
for marker in FORBIDDEN_IMPORT_ROOTS:
|
||
with self.subTest(file=path.name, marker=marker):
|
||
self.assertNotIn(f"import {marker}", text)
|
||
self.assertNotIn(f"from {marker}", text)
|
||
|
||
def test_no_outbound_client_and_no_process_execution(self) -> None:
|
||
for path in self.web_sources():
|
||
text = path.read_text(encoding="utf-8")
|
||
for marker in FORBIDDEN_IMPORTS:
|
||
with self.subTest(file=path.name, marker=marker):
|
||
self.assertNotIn(f"import {marker}", text)
|
||
lowered = text.lower()
|
||
for marker in FORBIDDEN_SOURCE_MARKERS:
|
||
with self.subTest(file=path.name, marker=marker):
|
||
self.assertNotIn(marker, lowered)
|
||
|
||
def test_route_is_a_pure_function_with_a_small_signature(self) -> None:
|
||
import inspect
|
||
|
||
signature = inspect.signature(panel.route)
|
||
self.assertEqual(list(signature.parameters), ["context", "method", "target"])
|
||
|
||
def test_server_module_takes_host_and_port_from_the_command_line(self) -> None:
|
||
args = server.parse_args([])
|
||
self.assertEqual(args.host, "127.0.0.1")
|
||
self.assertEqual(args.host, web.DEFAULT_HOST)
|
||
self.assertEqual(args.port, web.DEFAULT_PORT)
|
||
self.assertEqual(args.privilege, "read-only")
|
||
self.assertEqual(args.data, None)
|
||
|
||
def test_check_mode_reports_without_binding_a_port(self) -> None:
|
||
buffer = io.StringIO()
|
||
with contextlib.redirect_stdout(buffer):
|
||
code = server.main(["--check"])
|
||
self.assertEqual(code, 0)
|
||
output = buffer.getvalue()
|
||
self.assertIn(web.PANEL_VERSION, output)
|
||
self.assertIn("Veri kaynağı", output)
|
||
self.assertIn("Yetki", output)
|
||
self.assertIn("sunucu başlatılmadı", output)
|
||
|
||
def test_version_and_privilege_flags(self) -> None:
|
||
with contextlib.redirect_stdout(io.StringIO()):
|
||
with self.assertRaises(SystemExit) as raised:
|
||
server.parse_args(["--version"])
|
||
self.assertEqual(raised.exception.code, 0)
|
||
with contextlib.redirect_stderr(io.StringIO()):
|
||
with self.assertRaises(SystemExit):
|
||
server.parse_args(["--privilege", "root"])
|
||
|
||
def test_data_flag_selects_the_inventory_file(self) -> None:
|
||
snapshot = self.snapshot_from(self.example_document())
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
path = Path(directory) / "fleet.json"
|
||
path.write_text(self.example_document(), encoding="utf-8")
|
||
context = server.build_context(server.parse_args(["--data", str(path)]))
|
||
self.assertTrue(context.snapshot.is_file)
|
||
self.assertEqual(context.snapshot.total, snapshot.total)
|
||
self.assertEqual(context.server_info.data_path, str(path))
|
||
|
||
def example_document(self) -> str:
|
||
return (WEB_DIR / "data" / "fleet.example.json").read_text(encoding="utf-8")
|
||
|
||
def test_example_document_matches_the_panel_schema(self) -> None:
|
||
"""The shipped example must load cleanly, or it teaches the wrong schema."""
|
||
snapshot = self.snapshot_from(self.example_document())
|
||
self.assertTrue(snapshot.is_file, snapshot.error or "örnek belge okunamadı")
|
||
self.assertEqual(snapshot.invalid, ())
|
||
self.assertGreaterEqual(snapshot.total, 5)
|
||
self.assertGreater(snapshot.ready_count, 0)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# loader: state machines, refusal paths, provenance
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoader(PanelCase):
|
||
READY_RECORD: dict[str, Any] = {
|
||
"dealer_id": "12010193",
|
||
"wave": "P1",
|
||
"state": "READY",
|
||
"channels": {
|
||
"ssh_over_wg": "verified",
|
||
"xrdp": "verified",
|
||
"rustdesk": "verified",
|
||
"meshcentral": "verified",
|
||
},
|
||
}
|
||
|
||
def document(self, *records: object) -> dict:
|
||
return {"schema_version": 1, "generated_at": NOW, "devices": list(records)}
|
||
|
||
def test_missing_default_file_falls_back_to_labelled_demo(self) -> None:
|
||
snapshot = fleet_data.load_fleet(None, env={}, now=NOW, default_path="/yok/fleet.json")
|
||
self.assertTrue(snapshot.is_demo)
|
||
self.assertEqual(snapshot.total, len(fleet_data.DEMO_DEVICES))
|
||
self.assertIn("demo", snapshot.source_label.lower())
|
||
self.assertEqual(snapshot.banner_word, "DEMO VERİ")
|
||
self.assertIn("sentetik", snapshot.banner_text)
|
||
|
||
def test_missing_explicit_file_is_an_error_and_does_not_become_demo(self) -> None:
|
||
snapshot = fleet_data.load_fleet("/yok/boyle/bir/dosya.json", env={}, now=NOW)
|
||
self.assertTrue(snapshot.is_error)
|
||
self.assertEqual(snapshot.total, 0)
|
||
self.assertEqual(snapshot.banner_word, "VERİ YOK")
|
||
self.assertIn("bulunamadı", snapshot.error or "")
|
||
self.assertIn("Demo veriye düşülmedi", snapshot.banner_text)
|
||
|
||
def test_environment_variable_selects_the_document(self) -> None:
|
||
document = json.dumps(self.document(self.READY_RECORD))
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
path = Path(directory) / "fleet.json"
|
||
path.write_text(document, encoding="utf-8")
|
||
snapshot = fleet_data.load_fleet(None, env={fleet_data.DATA_PATH_ENV: str(path)}, now=NOW)
|
||
self.assertTrue(snapshot.is_file)
|
||
self.assertEqual(snapshot.total, 1)
|
||
|
||
def test_identity_is_derived_from_the_dealer_number(self) -> None:
|
||
snapshot = self.snapshot_from(self.document(self.READY_RECORD))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(device.device_id, "BF-12010193")
|
||
self.assertEqual(device.hostname, "bf-12010193")
|
||
self.assertEqual(device.dealer_id, "12010193")
|
||
self.assertEqual(snapshot.schema_version, 1)
|
||
self.assertEqual(snapshot.generated_at, NOW)
|
||
|
||
def test_wave_aliases_resolve_to_the_rollout_codes(self) -> None:
|
||
for raw, expected in (
|
||
("pilot_1", "P1"),
|
||
("LAB", "LAB"),
|
||
("wave_2", "W2"),
|
||
("production", "PROD"),
|
||
):
|
||
with self.subTest(wave=raw):
|
||
record = dict(self.READY_RECORD, wave=raw)
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
self.assertEqual(snapshot.devices[0].wave_code, expected)
|
||
|
||
def test_turkish_state_words_are_accepted_as_input(self) -> None:
|
||
for raw, expected in (("HAZIR", "READY"), ("Hazır", "READY"), ("hazir", "READY")):
|
||
with self.subTest(state=raw):
|
||
record = dict(self.READY_RECORD, state=raw)
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
self.assertEqual(snapshot.devices[0].state, expected)
|
||
|
||
def test_broken_json_is_an_error_that_names_the_document(self) -> None:
|
||
snapshot = self.snapshot_from('{"schema_version": 1, "devices": [')
|
||
self.assertTrue(snapshot.is_error)
|
||
self.assertIn("JSON", snapshot.error or "")
|
||
|
||
def test_unsupported_schema_version_is_refused(self) -> None:
|
||
for version in (2, 99):
|
||
with self.subTest(version=version):
|
||
document = {"schema_version": version, "devices": []}
|
||
snapshot = self.snapshot_from(document)
|
||
self.assertTrue(snapshot.is_error)
|
||
self.assertIn("şema sürümü", snapshot.error or "")
|
||
|
||
def test_missing_schema_version_or_devices_list_is_refused(self) -> None:
|
||
for document in ({"devices": []}, {"schema_version": 1}):
|
||
with self.subTest(document=document):
|
||
snapshot = self.snapshot_from(document)
|
||
self.assertTrue(snapshot.is_error)
|
||
|
||
def test_bad_dealer_number_is_shown_not_dropped(self) -> None:
|
||
record = dict(self.READY_RECORD, dealer_id="1201019")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
self.assertEqual(snapshot.total, 0)
|
||
self.assertEqual(len(snapshot.invalid), 1)
|
||
reason = snapshot.invalid[0].reason
|
||
self.assertIn("8 hane", reason)
|
||
self.assertIn("1201019", reason)
|
||
|
||
def test_duplicate_dealer_number_is_refused(self) -> None:
|
||
snapshot = self.snapshot_from(self.document(self.READY_RECORD, dict(self.READY_RECORD)))
|
||
self.assertEqual(snapshot.total, 1)
|
||
self.assertEqual(len(snapshot.invalid), 1)
|
||
self.assertIn("Yinelenen", snapshot.invalid[0].reason)
|
||
|
||
def test_unknown_wave_is_refused_with_a_turkish_reason(self) -> None:
|
||
record = dict(self.READY_RECORD, wave="WAVE-9")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
self.assertEqual(snapshot.total, 0)
|
||
self.assertIn("Bilinmeyen dalga", snapshot.invalid[0].reason)
|
||
|
||
def test_mismatching_device_id_is_refused(self) -> None:
|
||
record = dict(self.READY_RECORD, device_id="BF-12010999")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
self.assertEqual(snapshot.total, 0)
|
||
self.assertIn("kurala uymuyor", snapshot.invalid[0].reason)
|
||
|
||
def test_unknown_state_becomes_hata_with_the_reason(self) -> None:
|
||
record = dict(self.READY_RECORD, state="BOZUK")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(device.state_label, "HATA")
|
||
self.assertEqual(device.claimed_state, "BOZUK")
|
||
self.assertIn("tanınmıyor", device.reason or "")
|
||
|
||
def test_missing_state_becomes_hata_and_is_never_invented(self) -> None:
|
||
record = {key: value for key, value in self.READY_RECORD.items() if key != "state"}
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(device.state_label, "HATA")
|
||
self.assertIn("Durum alanı yok", device.reason or "")
|
||
|
||
def test_ready_claim_without_verified_channels_is_refused(self) -> None:
|
||
"""docs/27: READY is refused unless every channel reports verified."""
|
||
record = dict(self.READY_RECORD)
|
||
record["channels"] = dict(self.READY_RECORD["channels"], xrdp="pending")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(device.state_label, "HATA")
|
||
self.assertEqual(device.claimed_state, "READY")
|
||
self.assertIn("HAZIR iddiası reddedildi", device.reason or "")
|
||
self.assertIn("xRDP", device.reason or "")
|
||
self.assertIn("xRDP", device.unverified_channels)
|
||
|
||
def test_ready_claim_with_every_channel_verified_is_ready(self) -> None:
|
||
snapshot = self.snapshot_from(self.document(self.READY_RECORD))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(device.state_label, "HAZIR")
|
||
self.assertEqual(device.verified_channel_count, 4)
|
||
self.assertEqual(device.unverified_channels, ())
|
||
self.assertIn("dört kanalın dördü de doğrulandı", device.ready_gate_note)
|
||
|
||
def test_failed_channel_marks_the_device_hata(self) -> None:
|
||
record = dict(self.READY_RECORD, state="ENROLLED")
|
||
record["channels"] = dict(self.READY_RECORD["channels"], rustdesk="failed")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(device.state_label, "HATA")
|
||
self.assertIn("başarısız", device.reason or "")
|
||
|
||
def test_panel_never_promotes_a_device(self) -> None:
|
||
record = dict(self.READY_RECORD, state="ENROLLED")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(device.state_label, "KAYITLI")
|
||
self.assertTrue(any("yükseltmez" in note for note in device.notes), device.notes)
|
||
|
||
def test_missing_channels_are_all_reported_as_missing(self) -> None:
|
||
record = {"dealer_id": "12010193", "wave": "P1", "state": "PROVISIONED_OFFLINE"}
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
device = snapshot.devices[0]
|
||
self.assertEqual(len(device.channels), 4)
|
||
self.assertEqual({check.status for check in device.channels}, {"missing"})
|
||
self.assertEqual({check.status_word for check in device.channels}, {"yok"})
|
||
self.assertTrue(any("Kanallar bildirilmemiş" in note for note in device.notes))
|
||
|
||
def test_unknown_channel_keys_are_reported_not_silently_ignored(self) -> None:
|
||
record = dict(self.READY_RECORD)
|
||
record["channels"] = dict(self.READY_RECORD["channels"], telnet="verified")
|
||
snapshot = self.snapshot_from(self.document(record))
|
||
device = snapshot.devices[0]
|
||
self.assertTrue(any("telnet" in note for note in device.notes), device.notes)
|
||
self.assertEqual(len(device.channels), 4)
|
||
|
||
def test_loader_is_deterministic(self) -> None:
|
||
document = self.document(self.READY_RECORD)
|
||
first = self.snapshot_from(document)
|
||
second = self.snapshot_from(document)
|
||
self.assertEqual(first.devices, second.devices)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# ordering and filtering
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestOrderingAndFiltering(PanelCase):
|
||
def sorted_demo(self):
|
||
return fleet_data.sort_devices(self.snapshot.devices)
|
||
|
||
def test_non_ready_devices_come_first(self) -> None:
|
||
flags = [device.is_ready for device in self.sorted_demo()]
|
||
self.assertEqual(flags, sorted(flags), "hazır cihaz hazır olmayanın önünde")
|
||
self.assertIn(True, flags)
|
||
self.assertIn(False, flags)
|
||
|
||
def test_failed_devices_are_at_the_top(self) -> None:
|
||
ordered = self.sorted_demo()
|
||
self.assertEqual(ordered[0].state_label, "HATA")
|
||
|
||
def test_sort_keeps_every_device_exactly_once(self) -> None:
|
||
ordered = self.sorted_demo()
|
||
self.assertEqual(len(ordered), self.snapshot.total)
|
||
self.assertEqual(
|
||
{device.device_id for device in ordered},
|
||
{device.device_id for device in self.snapshot.devices},
|
||
)
|
||
|
||
def test_query_filter_matches_identity_hostname_and_dealer(self) -> None:
|
||
for needle in ("BF-12010193", "bf-12010193", "12010193"):
|
||
with self.subTest(needle=needle):
|
||
criteria, warnings = fleet_data.parse_filter({"q": [needle]})
|
||
shown = fleet_data.filter_devices(self.snapshot.devices, criteria)
|
||
self.assertEqual(len(shown), 1, needle)
|
||
self.assertEqual(shown[0].device_id, "BF-12010193")
|
||
self.assertEqual(warnings, ())
|
||
|
||
def test_filter_query_is_case_insensitive(self) -> None:
|
||
criteria, _ = fleet_data.parse_filter({"q": ["bf-1201019"]})
|
||
shown = fleet_data.filter_devices(self.snapshot.devices, criteria)
|
||
self.assertEqual(
|
||
{device.device_id for device in shown},
|
||
{"BF-12010193", "BF-12010194", "BF-12010195"},
|
||
)
|
||
|
||
def test_state_filter_accepts_word_machine_name_and_folded_input(self) -> None:
|
||
for raw in ("HAZIR", "ready", "hazir", "Hazır"):
|
||
with self.subTest(raw=raw):
|
||
criteria, warnings = fleet_data.parse_filter({"durum": [raw]})
|
||
self.assertEqual(criteria.state, "READY")
|
||
self.assertEqual(warnings, ())
|
||
shown = fleet_data.filter_devices(self.snapshot.devices, criteria)
|
||
self.assertTrue(shown)
|
||
self.assertEqual({device.state_label for device in shown}, {"HAZIR"})
|
||
|
||
def test_wave_filter_accepts_repo_group_names(self) -> None:
|
||
for raw, code in (("pilot_1", "P1"), ("W1", "W1"), ("production", "PROD")):
|
||
with self.subTest(raw=raw):
|
||
criteria, _ = fleet_data.parse_filter({"dalga": [raw]})
|
||
self.assertEqual(criteria.wave, code)
|
||
shown = fleet_data.filter_devices(self.snapshot.devices, criteria)
|
||
self.assertTrue(shown)
|
||
self.assertEqual({device.wave_code for device in shown}, {code})
|
||
|
||
def test_unknown_filter_values_are_ignored_with_a_turkish_warning(self) -> None:
|
||
criteria, warnings = fleet_data.parse_filter({"durum": ["MOR"], "dalga": ["W9"]})
|
||
self.assertEqual(criteria.state, "")
|
||
self.assertEqual(criteria.wave, "")
|
||
self.assertEqual(len(warnings), 2)
|
||
for warning in warnings:
|
||
self.assertIn("yok sayıldı", warning)
|
||
page = self.page("/?durum=MOR&dalga=W9")
|
||
self.assertIn("yok sayıldı", page)
|
||
self.assertIn("Uyarılar", page)
|
||
|
||
def test_only_not_ready_filter(self) -> None:
|
||
criteria, _ = fleet_data.parse_filter({"hazir_degil": ["1"]})
|
||
shown = fleet_data.filter_devices(self.snapshot.devices, criteria)
|
||
self.assertTrue(shown)
|
||
self.assertTrue(all(not device.is_ready for device in shown))
|
||
|
||
def test_counts_are_zero_filled_and_add_up(self) -> None:
|
||
counts = fleet_data.state_counts(self.snapshot.devices)
|
||
self.assertEqual(list(counts), list(STATE_WORDS))
|
||
self.assertEqual(sum(counts.values()), self.snapshot.total)
|
||
self.assertEqual(counts, self.snapshot.counts)
|
||
self.assertEqual(self.snapshot.ready_count + self.snapshot.not_ready_count, self.snapshot.total)
|
||
|
||
def test_filter_matches_the_rendered_count(self) -> None:
|
||
criteria, _ = fleet_data.parse_filter({"durum": ["YOK"]})
|
||
shown = fleet_data.filter_devices(self.snapshot.devices, criteria)
|
||
page = self.page("/?durum=YOK")
|
||
self.assertEqual(len(self.row_words(page)), len(shown))
|
||
self.assertTrue(all(word == "YOK" for word in self.row_words(page)))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# the first viewport
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestFleetPage(PanelCase):
|
||
def test_served_as_utf8_turkish_html(self) -> None:
|
||
response = self.get("/")
|
||
self.assertEqual(response.status, 200)
|
||
self.assertEqual(response.content_type, "text/html; charset=utf-8")
|
||
markup = response.body.decode("utf-8")
|
||
self.assertIn('<html lang="tr"', markup)
|
||
self.assertIn('<meta charset="utf-8">', markup)
|
||
# Turkish characters must survive the byte round trip, not just the string.
|
||
self.assertIn("doğrulandı".encode("utf-8"), response.body)
|
||
self.assertEqual(response.content_length, len(response.body))
|
||
|
||
def test_first_viewport_regions_are_ordered(self) -> None:
|
||
page = self.page("/")
|
||
positions = [
|
||
page.index('data-region="header"'),
|
||
page.index('data-region="summary"'),
|
||
page.index('data-region="list"'),
|
||
page.index('data-region="detail"'),
|
||
]
|
||
self.assertEqual(positions, sorted(positions), "bölge sırası DESIGN.md'ye uymuyor")
|
||
|
||
def test_header_carries_product_target_and_privilege(self) -> None:
|
||
page = self.page("/")
|
||
header = page[: page.index('data-region="summary"')]
|
||
self.assertIn(web.PANEL_NAME, header)
|
||
self.assertIn(web.PANEL_SURFACE, header)
|
||
self.assertIn("Hedef", header)
|
||
self.assertIn(f"Tüm filo · {self.snapshot.total} cihaz", header)
|
||
self.assertIn("Yetki", header)
|
||
self.assertIn("Salt-okunur", header)
|
||
self.assertIn(self.read_only.reason, header)
|
||
|
||
def test_privilege_level_is_the_right_most_header_readout(self) -> None:
|
||
"""docs/DESIGN.md: 'privilege level on the right -- never a buried sentence'."""
|
||
page = self.page("/")
|
||
header = page[: page.index('data-region="summary"')]
|
||
privilege = header.index("app-header__privilege")
|
||
self.assertGreater(privilege, header.index("app-header__identity"))
|
||
self.assertGreater(privilege, header.index("app-header__target"))
|
||
self.assertGreater(privilege, header.index("app-header__actions"))
|
||
|
||
def test_primary_action_sits_in_the_header_not_below_the_fold(self) -> None:
|
||
page = self.page("/")
|
||
self.assertLess(
|
||
page.index("Filo durumunu yenile"),
|
||
page.index('data-region="summary"'),
|
||
"birincil eylem başlık bandında değil",
|
||
)
|
||
self.assertIn('class="button button--primary"', page)
|
||
|
||
def test_unprivileged_write_action_is_disabled_with_the_reason_on_the_control(self) -> None:
|
||
page = self.page("/")
|
||
self.assertIn('disabled aria-disabled="true" aria-describedby="gated-reason"', page)
|
||
self.assertIn('id="gated-reason"', page)
|
||
self.assertIn(self.read_only.reason, page)
|
||
|
||
def test_every_state_word_appears_in_the_summary_strip(self) -> None:
|
||
page = self.page("/")
|
||
summary = page[page.index('data-region="summary"') : page.index('data-region="list"')]
|
||
for word in STATE_WORDS:
|
||
with self.subTest(word=word):
|
||
self.assertIn(word, summary)
|
||
|
||
def test_summary_strip_has_exactly_one_large_figure(self) -> None:
|
||
page = self.page("/")
|
||
self.assertEqual(page.count("summary__lead-figure"), 1)
|
||
self.assertEqual(
|
||
page.count("summary__cell-value"), len(fleet_data.STATE_VOCABULARY)
|
||
)
|
||
counts = self.snapshot.counts
|
||
summary = page[page.index('data-region="summary"') : page.index('data-region="list"')]
|
||
self.assertIn(f'>{counts["HAZIR"]}<', summary)
|
||
|
||
def test_every_row_carries_a_state_word_in_monospace_columns(self) -> None:
|
||
page = self.page("/")
|
||
self.assertEqual(len(self.row_words(page)), self.snapshot.total)
|
||
for device in self.snapshot.devices:
|
||
with self.subTest(device=device.device_id):
|
||
self.assertIn(
|
||
f'<td class="mono"><a class="row-select" href="/?cihaz={device.device_id}">'
|
||
f"{device.device_id}</a></td>",
|
||
page,
|
||
)
|
||
self.assertIn(f'class="badge badge--state-{device.state_token_suffix}"', page)
|
||
self.assertIn(f'<td class="mono">{device.hostname}</td>', page)
|
||
|
||
def test_wave_column_names_the_wave_once_in_monospace(self) -> None:
|
||
"""The Dalga column is a data cell, and it names each wave exactly once."""
|
||
region = self.list_region(self.page("/"))
|
||
rows = re.findall(r'<tr class="fleet-row.*?</tr>', region, flags=re.S)
|
||
self.assertEqual(len(rows), self.snapshot.total)
|
||
by_id = {device.device_id: device for device in self.snapshot.devices}
|
||
for row in rows:
|
||
match = re.search(r">(BF-[0-9]{8})</a>", row)
|
||
if match is None:
|
||
self.fail(f"satırda cihaz kimliği yok: {row[:120]}")
|
||
device = by_id[match.group(1)]
|
||
cells = re.findall(r"<td[^>]*>(.*?)</td>", row, flags=re.S)
|
||
self.assertEqual(len(cells), 5, row[:200])
|
||
self.assertEqual(
|
||
cells[3], f'<span class="mono">{device.wave_label}</span>', row[:200]
|
||
)
|
||
|
||
def test_rows_are_sorted_non_ready_first(self) -> None:
|
||
words = self.row_words(self.page("/"))
|
||
ready_flags = [
|
||
fleet_data.STATE_BY_WORD[word].state == "READY" for word in words
|
||
]
|
||
self.assertEqual(ready_flags, sorted(ready_flags))
|
||
self.assertEqual(ready_flags[0], False)
|
||
self.assertEqual(ready_flags[-1], True)
|
||
|
||
def test_selected_row_uses_the_accent_soft_fill(self) -> None:
|
||
page = self.page("/")
|
||
self.assertEqual(page.count("fleet-row--selected"), 1)
|
||
selected_row = page[page.index("fleet-row--selected") :]
|
||
selected_row = selected_row[: selected_row.index("</tr>")]
|
||
detail = page[page.index('data-region="detail"') :]
|
||
device_id = re.search(r">(BF-[0-9]{8})</a>", selected_row).group(1)
|
||
self.assertIn(device_id, detail)
|
||
css = (STATIC_DIR / "app.css").read_text(encoding="utf-8")
|
||
self.assertIn(".fleet-row--selected td { background: var(--accent-soft); }", css)
|
||
|
||
def test_default_selection_is_the_first_row_of_the_list(self) -> None:
|
||
page = self.page("/")
|
||
first_row = self.list_region(page)
|
||
first_row = first_row[first_row.index("<tbody>") :]
|
||
device_id = re.search(r">(BF-[0-9]{8})</a>", first_row).group(1)
|
||
detail = page[page.index('data-region="detail"') :]
|
||
self.assertIn(device_id, detail)
|
||
self.assertIn("Hedef kimliği", detail)
|
||
|
||
def test_detail_region_shows_state_identity_wave_and_channels(self) -> None:
|
||
page = self.page("/?cihaz=BF-12010193")
|
||
detail = page[page.index('data-region="detail"') :]
|
||
device = next(d for d in self.snapshot.devices if d.device_id == "BF-12010193")
|
||
self.assertIn(device.state_label, detail)
|
||
self.assertIn(device.device_id, detail)
|
||
self.assertIn(device.hostname, detail)
|
||
self.assertIn(device.wave_code, detail)
|
||
self.assertIn("Kanal doğrulaması", detail)
|
||
for channel in fleet_data.CHANNELS:
|
||
with self.subTest(channel=channel.label):
|
||
self.assertIn(channel.label, detail)
|
||
self.assertIn("Doğrulanan kanal: 4/4", detail)
|
||
self.assertIn("doğrulandı", detail)
|
||
|
||
def test_channel_status_words_are_shown_in_turkish(self) -> None:
|
||
"""Every channel status word reaches the screen, not only a colour."""
|
||
cases = {
|
||
"BF-12010193": ("doğrulandı",), # all four verified
|
||
"BF-12010194": ("bekliyor",), # RustDesk pending
|
||
"BF-12010205": ("başarısız", "bilinmiyor"), # failed + unknown
|
||
"BF-12010195": ("yok",), # never checked
|
||
}
|
||
for device_id, words in cases.items():
|
||
with self.subTest(device=device_id):
|
||
page = self.page(f"/?cihaz={device_id}")
|
||
detail = page[page.index('data-region="detail"') :]
|
||
for word in words:
|
||
self.assertIn(f">{word}</span>", detail)
|
||
|
||
def test_ready_device_detail_confirms_the_gate(self) -> None:
|
||
page = self.page("/?cihaz=BF-12010193")
|
||
self.assertIn("dört kanalın dördü de doğrulandı", page)
|
||
self.assertNotIn("HAZIR için eksik kanallar", page)
|
||
|
||
def test_not_ready_device_detail_names_the_blocking_channels(self) -> None:
|
||
page = self.page("/?cihaz=BF-12010194")
|
||
self.assertIn("HAZIR için eksik kanallar: RustDesk", page)
|
||
self.assertIn("Doğrulanan kanal: 3/4", page)
|
||
|
||
def test_hostname_selects_the_same_device_as_device_id(self) -> None:
|
||
by_id = self.page("/?cihaz=BF-12010193")
|
||
by_host = self.page("/?cihaz=bf-12010193")
|
||
self.assertEqual(self.row_words(by_id), self.row_words(by_host))
|
||
self.assertIn("BF-12010193", by_host[by_host.index('data-region="detail"') :])
|
||
|
||
def test_unknown_device_id_is_reported_and_not_invented(self) -> None:
|
||
page = self.page("/?cihaz=BF-99999999")
|
||
detail = page[page.index('data-region="detail"') :]
|
||
self.assertIn("BF-99999999", detail)
|
||
self.assertIn("envanterde yok", detail.lower())
|
||
self.assertIn("uydurmaz", detail)
|
||
|
||
def test_filter_is_reflected_in_the_list_and_the_header(self) -> None:
|
||
page = self.page("/?durum=HAZIR")
|
||
self.assertEqual(set(self.row_words(page)), {"HAZIR"})
|
||
self.assertIn("Durum HAZIR", page)
|
||
self.assertIn(f"Gösterilen: <span class=\"mono\">{self.snapshot.ready_count}</span>", page)
|
||
|
||
def test_selection_outside_the_filter_is_flagged_not_hidden(self) -> None:
|
||
page = self.page("/?durum=HAZIR&cihaz=BF-12010205")
|
||
detail = page[page.index('data-region="detail"') :]
|
||
self.assertIn("BF-12010205", detail)
|
||
self.assertIn("geçerli filtreye uymuyor", detail)
|
||
|
||
def test_empty_result_gets_a_turkish_note_and_keeps_the_header(self) -> None:
|
||
page = self.page("/?durum=YOK&dalga=PROD")
|
||
self.assertEqual(self.row_words(page), [])
|
||
self.assertIn("Bu filtreye uyan cihaz yok", page)
|
||
self.assertIn("Filtreyi temizle", page)
|
||
self.assertIn('data-region="header"', page)
|
||
|
||
def test_wave_strip_shows_every_wave_against_its_target(self) -> None:
|
||
page = self.page("/")
|
||
counts = fleet_data.wave_counts(self.snapshot.devices)
|
||
for wave in fleet_data.WAVES:
|
||
with self.subTest(wave=wave.code):
|
||
self.assertIn(
|
||
f'<span class="mono">{wave.code}</span> {counts[wave.code]}/{wave.target}',
|
||
page,
|
||
)
|
||
|
||
def test_demo_data_is_labelled_on_the_page(self) -> None:
|
||
page = self.page("/")
|
||
self.assertIn("DEMO VERİ", page)
|
||
self.assertIn("sentetik", page)
|
||
self.assertIn("hiçbir cihazı temsil etmez", page)
|
||
self.assertIn("Veri kaynağı: sentetik demo veri", page)
|
||
|
||
def test_inventory_file_source_is_labelled_on_the_page(self) -> None:
|
||
document = {
|
||
"schema_version": 1,
|
||
"generated_at": NOW,
|
||
"devices": [
|
||
{
|
||
"dealer_id": "12010193",
|
||
"wave": "P1",
|
||
"state": "ENROLLED",
|
||
"channels": {"ssh_over_wg": "verified"},
|
||
}
|
||
],
|
||
}
|
||
snapshot = self.snapshot_from(document)
|
||
context = panel.PanelContext(
|
||
snapshot=snapshot, privilege=self.read_only, server_info=self.info
|
||
)
|
||
page = self.page("/", context=context)
|
||
self.assertIn("ENVANTER", page)
|
||
self.assertIn(snapshot.source_path, page)
|
||
self.assertNotIn("DEMO VERİ", page)
|
||
|
||
def test_error_source_says_veri_yok_and_never_shows_demo_data(self) -> None:
|
||
snapshot = fleet_data.load_fleet("/yok/boyle/bir/dosya.json", env={}, now=NOW)
|
||
context = panel.PanelContext(
|
||
snapshot=snapshot, privilege=self.read_only, server_info=self.info
|
||
)
|
||
page = self.page("/", context=context)
|
||
self.assertIn("VERİ YOK", page)
|
||
self.assertIn("Demo veriye düşülmedi", page)
|
||
self.assertNotIn("DEMO VERİ", page)
|
||
self.assertEqual(self.row_words(page), [])
|
||
self.assertIn("Envanterde gösterilecek cihaz yok", page)
|
||
|
||
def test_invalid_records_are_surfaced_with_their_reason(self) -> None:
|
||
document = {
|
||
"schema_version": 1,
|
||
"devices": [{"dealer_id": "1201019", "wave": "P1", "state": "READY"}],
|
||
}
|
||
snapshot = self.snapshot_from(document)
|
||
context = panel.PanelContext(
|
||
snapshot=snapshot, privilege=self.read_only, server_info=self.info
|
||
)
|
||
page = self.page("/", context=context)
|
||
self.assertIn("Geçersiz kayıtlar", page)
|
||
self.assertIn("8 hane", page)
|
||
|
||
def test_every_visible_string_is_non_empty(self) -> None:
|
||
for target in ("/", "/?durum=HAZIR", "/?cihaz=BF-99999999", "/hakkinda"):
|
||
with self.subTest(target=target):
|
||
self.assertEqual(empty_elements(self.page(target)), [])
|
||
|
||
def test_error_pages_keep_the_frame_and_stay_readable(self) -> None:
|
||
for target, status in (("/yok", 404),):
|
||
with self.subTest(target=target):
|
||
response = self.get(target)
|
||
self.assertEqual(response.status, status)
|
||
markup = response.body.decode("utf-8")
|
||
self.assertEqual(empty_elements(markup), [])
|
||
self.assertIn('data-region="header"', markup)
|
||
|
||
def test_no_placeholder_or_python_none_leaks_into_the_page(self) -> None:
|
||
for target in ("/", "/hakkinda", "/?cihaz=BF-12010193", "/?durum=MOR"):
|
||
with self.subTest(target=target):
|
||
page = self.page(target)
|
||
self.assertNotIn("$", page)
|
||
self.assertNotIn("None", page)
|
||
self.assertNotIn("KeyError", page)
|
||
|
||
def test_page_renders_without_a_live_server_context(self) -> None:
|
||
"""render_fleet_page is usable on its own (tests, --check, future CLI)."""
|
||
page = render.render_fleet_page(self.snapshot)
|
||
self.assertIn('<html lang="tr"', page)
|
||
self.assertIn("Salt-okunur", page)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# about / durum
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestAboutPage(PanelCase):
|
||
def test_reports_version_privilege_and_python(self) -> None:
|
||
page = self.page("/hakkinda")
|
||
self.assertIn(web.PANEL_VERSION, page)
|
||
self.assertIn(render.PANEL_VERSION, page)
|
||
self.assertIn("Panel sürümü", page)
|
||
self.assertIn("127.0.0.1:8787", page)
|
||
self.assertIn("Salt-okunur", page)
|
||
self.assertIn(self.read_only.reason, page)
|
||
self.assertIn(self.info.python_version, page)
|
||
|
||
def test_reports_the_data_source_and_its_limits(self) -> None:
|
||
page = self.page("/hakkinda")
|
||
self.assertIn("Veri kaynağı", page)
|
||
self.assertIn("sentetik demo veri", page)
|
||
self.assertIn("kullanılmadı (demo veri)", page)
|
||
self.assertIn(f"{self.snapshot.ready_count} hazır", page)
|
||
self.assertIn("Okuma hatası", page)
|
||
self.assertIn("yok", page)
|
||
|
||
def test_lists_the_vocabulary_with_tokens_and_meanings(self) -> None:
|
||
page = self.page("/hakkinda")
|
||
for spec in fleet_data.STATE_VOCABULARY:
|
||
with self.subTest(state=spec.state):
|
||
self.assertIn(spec.word, page)
|
||
self.assertIn(spec.token, page)
|
||
self.assertIn(spec.meaning, page)
|
||
|
||
def test_lists_every_wave_and_channel(self) -> None:
|
||
page = self.page("/hakkinda")
|
||
for wave in fleet_data.WAVES:
|
||
with self.subTest(wave=wave.code):
|
||
self.assertIn(wave.code, page)
|
||
for channel in fleet_data.CHANNELS:
|
||
with self.subTest(channel=channel.key):
|
||
self.assertIn(channel.key, page)
|
||
self.assertIn(channel.label, page)
|
||
|
||
def test_states_the_panel_rules(self) -> None:
|
||
page = self.page("/hakkinda")
|
||
self.assertGreaterEqual(len(render.PANEL_RULES), 5)
|
||
for rule in render.PANEL_RULES:
|
||
with self.subTest(rule=rule[:30]):
|
||
self.assertTrue(rule.strip())
|
||
self.assertIn(rule[:24], page)
|
||
self.assertIn("Panel kuralları", page)
|
||
|
||
def test_full_privilege_level_is_distinct_and_honest(self) -> None:
|
||
page = self.page("/", context=self.full_context)
|
||
self.assertIn("badge--privilege-full", page)
|
||
self.assertIn("Tam yetki", page)
|
||
self.assertIn("yazma uç noktası uygulanmadı", page)
|
||
self.assertNotIn("badge--privilege-read-only", page)
|
||
|
||
def test_about_page_has_no_external_url(self) -> None:
|
||
page = self.page("/hakkinda")
|
||
self.assertNotIn("http://", page)
|
||
self.assertNotIn("https://", page)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# routing and the privilege gate
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class TestRoutingAndGate(PanelCase):
|
||
def test_only_read_methods_are_served(self) -> None:
|
||
for method in ("POST", "PUT", "PATCH", "DELETE", "OPTIONS", "TRACE"):
|
||
with self.subTest(method=method):
|
||
response = self.get("/", method)
|
||
self.assertEqual(response.status, 405)
|
||
markup = response.body.decode("utf-8")
|
||
self.assertIn("Yöntem reddedildi", markup)
|
||
self.assertIn(method, markup)
|
||
self.assertIn("yalnız okuma yapar", markup)
|
||
headers = dict(response.headers)
|
||
self.assertEqual(headers.get("Allow"), "GET, HEAD")
|
||
|
||
def test_write_methods_are_refused_even_with_full_privilege(self) -> None:
|
||
response = self.get("/", "POST", self.full_context)
|
||
self.assertEqual(response.status, 405)
|
||
|
||
def test_head_returns_the_same_length_without_a_body(self) -> None:
|
||
head = self.get("/", "HEAD")
|
||
get = self.get("/", "GET")
|
||
self.assertEqual(head.status, 200)
|
||
self.assertEqual(head.body, b"")
|
||
self.assertEqual(head.content_length, get.content_length)
|
||
self.assertEqual(head.content_type, get.content_type)
|
||
|
||
def test_static_assets_are_served_from_disk_byte_for_byte(self) -> None:
|
||
for name in ("tokens.css", "app.css"):
|
||
with self.subTest(name=name):
|
||
response = self.get(f"/static/{name}")
|
||
self.assertEqual(response.status, 200)
|
||
self.assertEqual(response.content_type, "text/css; charset=utf-8")
|
||
self.assertEqual(response.body, (STATIC_DIR / name).read_bytes())
|
||
|
||
def test_unknown_static_asset_is_a_404(self) -> None:
|
||
response = self.get("/static/gizli.txt")
|
||
self.assertEqual(response.status, 404)
|
||
self.assertIn("Sayfa yok", response.body.decode("utf-8"))
|
||
|
||
def test_path_traversal_is_refused(self) -> None:
|
||
for target in (
|
||
"/static/../server.py",
|
||
"/static/..%2fserver.py",
|
||
"/static/%2e%2e/fleet_data.py",
|
||
"/../web/server.py",
|
||
"/static/.%2e/%2e%2e/etc/passwd",
|
||
):
|
||
with self.subTest(target=target):
|
||
response = self.get(target)
|
||
self.assertEqual(response.status, 404)
|
||
body = response.body.decode("utf-8")
|
||
self.assertNotIn("BaseHTTPRequestHandler", body)
|
||
self.assertNotIn("route(context", body)
|
||
|
||
def test_unknown_path_is_a_turkish_404(self) -> None:
|
||
response = self.get("/boyle-bir-sayfa-yok")
|
||
self.assertEqual(response.status, 404)
|
||
markup = response.body.decode("utf-8")
|
||
self.assertIn("Sayfa yok", markup)
|
||
self.assertIn("/boyle-bir-sayfa-yok", markup)
|
||
|
||
def test_security_headers_keep_the_panel_local_and_script_free(self) -> None:
|
||
headers = dict(self.get("/").headers)
|
||
self.assertEqual(headers["X-Content-Type-Options"], "nosniff")
|
||
self.assertEqual(headers["Referrer-Policy"], "no-referrer")
|
||
csp = headers["Content-Security-Policy"]
|
||
self.assertIn("default-src 'self'", csp)
|
||
self.assertIn("script-src 'none'", csp)
|
||
self.assertIn("form-action 'self'", csp)
|
||
self.assertNotIn("http", csp)
|
||
self.assertEqual(headers["Cache-Control"], "no-store")
|
||
|
||
def test_route_is_pure_and_repeatable(self) -> None:
|
||
targets = ("/", "/hakkinda", "/static/app.css", "/?durum=HAZIR&q=1201", "/yok", "/static/../x")
|
||
for target in targets:
|
||
with self.subTest(target=target):
|
||
first = self.get(target)
|
||
second = self.get(target)
|
||
self.assertEqual(first, second)
|
||
|
||
def test_bogus_query_values_do_not_crash_the_page(self) -> None:
|
||
for target in (
|
||
"/?q=&durum=&dalga=&cihaz=&hazir_degil=abc",
|
||
"/?durum=READY&durum=YOK",
|
||
"/?q=%C4%B0stanbul%20%2F%20test",
|
||
"/?cihaz=" + "x" * 500,
|
||
):
|
||
with self.subTest(target=target[:40]):
|
||
response = self.get(target)
|
||
self.assertEqual(response.status, 200)
|
||
self.assertEqual(empty_elements(response.body.decode("utf-8")), [])
|
||
|
||
def test_a_full_sweep_of_every_endpoint_writes_nothing(self) -> None:
|
||
"""Read-only is a property of the surface, not a promise in a docstring."""
|
||
before = tree_fingerprint(WEB_DIR)
|
||
requests = (
|
||
("GET", "/"),
|
||
("GET", "/filo"),
|
||
("GET", "/fleet"),
|
||
("GET", "/hakkinda"),
|
||
("GET", "/hakkinda/"),
|
||
("GET", "/about"),
|
||
("GET", "/static/tokens.css"),
|
||
("GET", "/static/app.css"),
|
||
("GET", "/?q=1201&durum=HAZIR&dalga=P1&hazir_degil=1&cihaz=BF-12010193"),
|
||
("GET", "/yok"),
|
||
("POST", "/"),
|
||
("PUT", "/hakkinda"),
|
||
("DELETE", "/static/app.css"),
|
||
("HEAD", "/"),
|
||
)
|
||
for method, target in requests:
|
||
with self.subTest(method=method, target=target):
|
||
response = self.get(target, method)
|
||
self.assertIn(response.status, (200, 404, 405))
|
||
self.assertEqual(tree_fingerprint(WEB_DIR), before)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|