2bad7fbb7f
The console was a flat form: every control on the root window, 48 operations as undifferentiated rows, a giant output textarea, privilege as a buried sentence. It now has a shell: a fixed header band (product, target, device state, privilege) plus a page body, and in detailed mode a sidebar with five operation categories and an Activity page. What changed (presentation layer only; _execute, gates, run_plan, audit and redaction untouched): - simple mode is an Overview: four state cards (target/device/privilege/ inventory) plus Install/Status/Exit, instead of the raw form; - the target is one segmented control (Device/Wave/Entire fleet) with a single scope source of truth; the dealer text can no longer silently override the wave, and the scope stays visible across pages; - operations are browsed, not dumped: icon + name + explanation + scope/risk badges, with a detail panel (meta grid, warnings, options form, backend-built command preview on demand); - destructive confirmations are dedicated dialogs showing OPERATION/TARGET/ SCOPE/EFFECT with a fleet/production strip and focus on Cancel; the token dialog is styled; secrets never enter either; - the output textarea is an execution drawer (IDLE/RUNNING with mm:ss / DONE-FAILED with duration and exit code) streaming the same backend output; - write operations stay visible but disabled with their reason when unprivileged, in both entry points; - new module bfos/ui_widgets.py (~490 lines, stdlib + tkinter only, no colour literals): canvas line icons, segmented control, collapsible panel, meta grid, timer, the two safety dialogs. Tests: test_bfos_gui_static.py 95 OK, test_bfos.py 155 OK, run-all.sh 12/12. The colour-literal test now also scans ui_widgets.py (proven by mutation: a #ff0000 probe goes red, green after removal). Live QA: zero geometry errors over 712 widgets, all 48 operations listed, 13 write ops disabled with reasons without root, TUI and --list smoke-tested. Known limits: no dark theme (tokens ready); the detail page's target line does not refresh when the target changes mid-form (the confirmation dialog always shows the current one); visual QA was geometry + OCR, not a human eye.
1103 lines
50 KiB
Python
1103 lines
50 KiB
Python
#!/usr/bin/env python3
|
||
"""bfos.ui_gui -- statik testler (stdlib unittest, pencere ACMAZ).
|
||
|
||
Bu testler GUI'yi baslatmaz ve hicbir Tk penceresi olusturmaz; yalnizca
|
||
tkinter'siz calisan yardimcilari ve backend sozlesmesini dogrular:
|
||
|
||
* bayi no dogrulamasi (^[0-9]{8}$) ve Turkce hata metinleri,
|
||
* komut onizleme metni (calistirilacak komut onizlemede GORUNUR),
|
||
* yikici islem siniflandirmasi (katalog + playbook ile capraz kontrol),
|
||
* headless (SSH) tespiti ve tkinter eksikligi mesaji,
|
||
* hedef/kapi kurallari: release -> tek cihaz yok, production -> token,
|
||
* --check kipi, salt-okunur rapor plani,
|
||
* komut calistirma: canli satir akisi + cikis kodu (gercek subprocess),
|
||
* tasarim sozlesmesi: tema token'lari web/static/tokens.css ile birebir ayni,
|
||
durum sozlugu (KURULDU/KAYITLI/HAZIR/HATA/YOK) ve yetki kapisi gorunurlugu.
|
||
|
||
Calistirma:
|
||
python3 tests/test_bfos_gui_static.py -v
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import contextlib
|
||
import importlib.util
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import textwrap
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||
UI_GUI_PATH = REPO_ROOT / "bfos" / "ui_gui.py"
|
||
|
||
|
||
def _load_ui_gui():
|
||
"""ui_gui.py'yi paket __init__'ini calistirmadan yukler (izole import)."""
|
||
if not UI_GUI_PATH.is_file():
|
||
raise unittest.SkipTest(f"bulunamadi: {UI_GUI_PATH}")
|
||
if str(REPO_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
name = "bfos_ui_gui_static_test"
|
||
spec = importlib.util.spec_from_file_location(name, UI_GUI_PATH)
|
||
assert spec is not None and spec.loader is not None
|
||
module = importlib.util.module_from_spec(spec)
|
||
# dataclasses kayitli bir modul bekler; bu kayit zorunludur.
|
||
sys.modules[name] = module
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
ui = _load_ui_gui()
|
||
|
||
INVENTORY_YAML = textwrap.dedent(
|
||
"""
|
||
# gecici test envanteri (gercek cihaz degil)
|
||
all:
|
||
children:
|
||
blueforce:
|
||
children:
|
||
lab:
|
||
hosts:
|
||
bf-12010101:
|
||
ansible_host: 10.42.0.11
|
||
pilot_1:
|
||
hosts:
|
||
bf-12010193:
|
||
ansible_host: 10.42.1.11
|
||
production:
|
||
hosts:
|
||
bf-12010999:
|
||
ansible_host: 10.42.9.11
|
||
"""
|
||
).strip() + "\n"
|
||
|
||
|
||
class StaticCase(unittest.TestCase):
|
||
"""Tum testler icin ortak kurulum: gecici envanter + backend."""
|
||
|
||
tmpdir: tempfile.TemporaryDirectory
|
||
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.tmpdir = tempfile.TemporaryDirectory(prefix="bfos-gui-test-")
|
||
cls.inventory = Path(cls.tmpdir.name) / "hosts.yml"
|
||
cls.inventory.write_text(INVENTORY_YAML, encoding="utf-8")
|
||
cls.ctx = ui.GuiContext.detect(inventory=str(cls.inventory))
|
||
cls.backend, cls.backend_note = ui.load_backend(cls.ctx)
|
||
|
||
@classmethod
|
||
def tearDownClass(cls) -> None:
|
||
cls.tmpdir.cleanup()
|
||
|
||
# -- kucuk yardimcilar -------------------------------------------------
|
||
def operation(self, operation_id: str):
|
||
for op in self.backend.list_operations():
|
||
if op.id == operation_id:
|
||
return op
|
||
self.fail(f"katalogda islem yok: {operation_id}")
|
||
|
||
def available_fleet_operation(self, scope: str = "simple"):
|
||
"""Playbook'u DISKTE olan bir filo islemi (depo es zamanli degisebilir)."""
|
||
for op in self.backend.list_operations():
|
||
if op.is_fleet and op.scope == scope and op.playbook and \
|
||
ui.playbook_path(REPO_ROOT, op.playbook).is_file():
|
||
return op
|
||
self.skipTest(f"kullanilabilir {scope} kapsamli filo islemi yok")
|
||
|
||
SAMPLE_VALUES = {
|
||
"script": "/opt/blueforce/bin/deneme.sh",
|
||
"packages": "docker-ce=5:27.0.3-1~ubuntu.24.04~noble",
|
||
"service": "docker",
|
||
"only": "01-precheck",
|
||
"grace": "300",
|
||
"log_lines": "500",
|
||
}
|
||
|
||
def valid_options(self, operation) -> dict:
|
||
"""Zorunlu secenekler icin gecerli ornek degerlerle sozluk uretir."""
|
||
options: dict = {}
|
||
for spec in operation.options:
|
||
value = self.SAMPLE_VALUES.get(spec.name, spec.default)
|
||
if spec.required and not value:
|
||
self.skipTest(f"{operation.id}: '{spec.name}' icin ornek deger yok")
|
||
options[spec.name] = value
|
||
return options
|
||
|
||
def dealer_target(self, dealer: str = "12010193"):
|
||
return ui.Target("dealer", dealer)
|
||
|
||
def group_target(self, group: str = "pilot_1"):
|
||
return ui.Target("group", group)
|
||
|
||
|
||
class TestDealerValidation(StaticCase):
|
||
def test_valid_dealer_numbers(self) -> None:
|
||
for value in ("12010193", "00000000", "99999999"):
|
||
self.assertTrue(ui.validate_dealer(value), value)
|
||
self.assertIsNone(ui.dealer_error(value), value)
|
||
|
||
def test_invalid_dealer_numbers(self) -> None:
|
||
for value in ("", "1201019", "120101933", "1201019a", "1201 0193",
|
||
" 12010193", "12010193 ", "bf-12010193", "１２３４５６７８"):
|
||
self.assertFalse(ui.validate_dealer(value), repr(value))
|
||
self.assertIsNotNone(ui.dealer_error(value), repr(value))
|
||
|
||
def test_error_message_is_turkish_and_actionable(self) -> None:
|
||
message = ui.dealer_error("1201019")
|
||
self.assertIn("8 hane", message)
|
||
self.assertIn("1201019", message)
|
||
self.assertIn("12010193", message) # ornek deger
|
||
self.assertIn("rakam", ui.dealer_error("abcdefgh"))
|
||
|
||
def test_dealer_derived_names(self) -> None:
|
||
self.assertEqual(ui.dealer_to_host("12010193"), "bf-12010193")
|
||
self.assertEqual(ui.dealer_to_device_id("12010193"), "BF-12010193")
|
||
self.assertEqual(ui.host_to_dealer("bf-12010193"), "12010193")
|
||
self.assertIsNone(ui.host_to_dealer("sunucu-1"))
|
||
|
||
def test_dealer_host_matches_repo_conventions(self) -> None:
|
||
# docs/02: hostname bf-<8 hane>, insan kimligi BF-<8 hane>
|
||
self.assertEqual(ui.HOST_PATTERN.pattern, r"^bf-[0-9]{8}$")
|
||
self.assertEqual(ui.DEALER_PATTERN.pattern, r"^[0-9]{8}$")
|
||
|
||
|
||
class TestHeadlessAndTkinter(StaticCase):
|
||
def test_detects_headless_without_display(self) -> None:
|
||
message = ui.headless_error({})
|
||
self.assertIsNotNone(message)
|
||
self.assertIn("DISPLAY", message)
|
||
self.assertIn("SSH", message)
|
||
# Grafik arayüz açılamadığında önerilen yol aynı uygulamanın terminal
|
||
# konsoludur; bash menü artık yedek (bkz. TERMINAL_MENU_HINT).
|
||
self.assertIn("bf --tui", message)
|
||
self.assertIn("python3 -m bfos.ui_gui", message)
|
||
|
||
def test_display_and_wayland_are_accepted(self) -> None:
|
||
self.assertIsNone(ui.headless_error({"DISPLAY": ":0"}))
|
||
self.assertIsNone(ui.headless_error({"WAYLAND_DISPLAY": "wayland-0"}))
|
||
self.assertIsNone(ui.headless_error({"DISPLAY": ":99",
|
||
"WAYLAND_DISPLAY": "wayland-9"}))
|
||
|
||
def test_tkinter_missing_message_mentions_package(self) -> None:
|
||
message = ui.tkinter_missing_message("No module named '_tkinter'")
|
||
self.assertIn("python3-tk", message)
|
||
self.assertIn("apt", message)
|
||
self.assertIn("bf --tui", message)
|
||
self.assertIn("_tkinter", message)
|
||
|
||
def test_window_start_error_is_turkish(self) -> None:
|
||
message = ui.window_start_error(RuntimeError("no display name and no $DISPLAY"))
|
||
self.assertIn("Pencere açılamadı", message)
|
||
self.assertIn("ssh -X", message)
|
||
self.assertIn("bf --tui", message)
|
||
|
||
def test_import_does_not_create_a_window(self) -> None:
|
||
# tkinter varsa bile modul import'u bir Tk koku olusturmamalidir:
|
||
# BfGui yalnizca acikca verilen bir root ile kurulabilir.
|
||
self.assertIsInstance(ui.HAVE_TKINTER, bool)
|
||
import inspect
|
||
|
||
signature = inspect.signature(ui.BfGui.__init__)
|
||
self.assertIn("root", signature.parameters)
|
||
# Yardimcilar tkinter olmadan da cagrilabilir.
|
||
self.assertIsNone(ui.dealer_error("12010193"))
|
||
|
||
def test_main_list_prints_catalogue_without_a_window(self) -> None:
|
||
buffer = io.StringIO()
|
||
with contextlib.redirect_stdout(buffer):
|
||
code = ui.main(["--list", "--inventory", str(self.inventory)])
|
||
self.assertEqual(code, 0)
|
||
output = buffer.getvalue()
|
||
self.assertIn("işlem kataloğu", output)
|
||
for category in ui.CATEGORIES:
|
||
self.assertIn(category, output)
|
||
|
||
def test_main_reports_headless_with_exit_3(self) -> None:
|
||
env = {k: v for k, v in os.environ.items()
|
||
if k not in ("DISPLAY", "WAYLAND_DISPLAY")}
|
||
stderr = io.StringIO()
|
||
import unittest.mock as mock
|
||
|
||
with mock.patch.dict(os.environ, env, clear=True), \
|
||
contextlib.redirect_stderr(stderr):
|
||
code = ui.main([])
|
||
self.assertEqual(code, 3)
|
||
# Aynı uygulamanın terminal konsoluna yönlendirmeli (bf --tui).
|
||
self.assertIn("bf --tui", stderr.getvalue())
|
||
|
||
|
||
class TestRootGate(StaticCase):
|
||
"""Tam yetkili GUI: yazma işlemleri root ister, raporlar her zaman açık.
|
||
|
||
Katalogun kendi `destructive` işareti ölçüt alınır (bkz. ui_gui
|
||
_normalize_operation); bu yüzden testler katalogdan örnek seçer, sabit
|
||
isimlere bağlanmaz.
|
||
"""
|
||
|
||
def test_is_root_reflects_the_effective_uid(self) -> None:
|
||
original = os.geteuid
|
||
try:
|
||
os.geteuid = lambda: 0
|
||
self.assertTrue(ui.is_root())
|
||
self.assertTrue(ui.write_operations_enabled())
|
||
os.geteuid = lambda: 1000
|
||
self.assertFalse(ui.is_root())
|
||
self.assertFalse(ui.write_operations_enabled())
|
||
finally:
|
||
os.geteuid = original
|
||
|
||
def test_destructive_operations_require_root(self) -> None:
|
||
writes = [op for op in self.backend.list_operations() if op.destructive]
|
||
self.assertTrue(writes, "katalogda yazma işlemi yok; kapı test edilemez")
|
||
for op in writes:
|
||
self.assertTrue(op.needs_root,
|
||
f"{op.id} yıkıcı ama needs_root değil")
|
||
|
||
def test_read_only_operations_do_not_require_root(self) -> None:
|
||
reports = [op for op in self.backend.list_operations() if not op.destructive]
|
||
self.assertTrue(reports, "katalogda salt-okunur işlem yok")
|
||
frozen = [op.id for op in reports if op.needs_root]
|
||
self.assertEqual([], frozen,
|
||
"salt-okunur işlemler root istememeli: " + ", ".join(frozen))
|
||
|
||
def test_notice_names_the_command_to_get_privileges(self) -> None:
|
||
self.assertIn("sudo bf", ui.ROOT_REQUIRED_NOTICE)
|
||
self.assertIn("sudo bf", ui.NON_ROOT_DISABLED_HINT)
|
||
|
||
def test_run_plan_refuses_a_root_plan_without_root(self) -> None:
|
||
"""Son savunma hattı: çalıştırma tek noktadan geçer."""
|
||
plan = ui.CommandPlan(
|
||
commands=(("true",),),
|
||
destructive=True,
|
||
requires_root=True,
|
||
operation_id="install-local",
|
||
)
|
||
original = os.geteuid
|
||
try:
|
||
os.geteuid = lambda: 1000
|
||
result = ui.run_plan(plan)
|
||
finally:
|
||
os.geteuid = original
|
||
self.assertFalse(result.ok)
|
||
self.assertEqual(2, result.exit_code)
|
||
self.assertIn("yönetici yetkisi", result.output.lower())
|
||
|
||
def test_run_plan_allows_a_root_plan_when_root(self) -> None:
|
||
plan = ui.CommandPlan(
|
||
commands=(("true",),),
|
||
destructive=True,
|
||
requires_root=True,
|
||
operation_id="install-local",
|
||
)
|
||
original = os.geteuid
|
||
try:
|
||
os.geteuid = lambda: 0
|
||
result = ui.run_plan(plan)
|
||
finally:
|
||
os.geteuid = original
|
||
self.assertTrue(result.ok, result.output)
|
||
|
||
|
||
class TestSimpleModeLayout(StaticCase):
|
||
def test_three_big_buttons_in_simple_mode(self) -> None:
|
||
self.assertEqual(ui.SIMPLE_MODE_BUTTONS, ("Kur", "Durum", "Çıkış"))
|
||
self.assertEqual(ui.SIMPLE_MODE_LABEL, "Detaylı mod")
|
||
|
||
def test_window_title_is_turkish(self) -> None:
|
||
self.assertEqual(ui.WINDOW_TITLE, "Blueforce Field OS — Yönetim")
|
||
|
||
def test_five_category_tabs(self) -> None:
|
||
self.assertEqual(
|
||
ui.CATEGORIES,
|
||
("Kurulum", "Cihaz işlemleri", "Merkez hazırlığı", "Araçlar", "Cihaz bilgisi"),
|
||
)
|
||
|
||
|
||
class TestCatalogue(StaticCase):
|
||
def test_operations_have_unique_ids_and_known_categories(self) -> None:
|
||
operations = self.backend.list_operations()
|
||
self.assertGreaterEqual(len(operations), 21)
|
||
ids = [op.id for op in operations]
|
||
self.assertEqual(len(ids), len(set(ids)), "islem id'leri tekil olmali")
|
||
for op in operations:
|
||
self.assertIn(op.category, ui.CATEGORIES, op.id)
|
||
self.assertTrue(op.label.strip(), op.id)
|
||
|
||
def test_every_category_has_at_least_one_operation(self) -> None:
|
||
for category in ui.CATEGORIES:
|
||
with self.subTest(category=category):
|
||
self.assertTrue(
|
||
[op for op in self.backend.list_operations() if op.category == category],
|
||
f"{category} kategorisi bos",
|
||
)
|
||
|
||
def test_fleet_operation_playbooks_match_declared_scope(self) -> None:
|
||
"""Konsol tablosu playbook'la uyusmali; playbook yoksa CALISTIRMA reddedilir.
|
||
|
||
(bf_playbook_check + fail-closed kuralinin statik karsiligi.)
|
||
"""
|
||
checked = 0
|
||
missing: list[str] = []
|
||
for op in self.backend.list_operations():
|
||
if not op.is_fleet:
|
||
continue
|
||
book = ui.playbook_path(REPO_ROOT, op.playbook)
|
||
if not book.is_file():
|
||
# Eksik playbook SESSIZ gecilemez: komut uretimi reddedilmeli.
|
||
missing.append(f"{op.id} ({op.playbook})")
|
||
with self.subTest(operation=op.id):
|
||
with self.assertRaises(ui.CommandError):
|
||
self.backend.build_command(op, self.group_target())
|
||
continue
|
||
with self.subTest(operation=op.id):
|
||
self.assertEqual(
|
||
ui.playbook_scope(book), op.scope,
|
||
f"{op.playbook}: tablo '{op.scope}' derken playbook "
|
||
f"'{ui.playbook_scope(book)}' beyan ediyor",
|
||
)
|
||
checked += 1
|
||
self.assertGreaterEqual(checked, 15)
|
||
if missing:
|
||
print(f"\n[UYARI] bu checkout'ta playbook dosyası eksik: {', '.join(missing)}"
|
||
" -> ilgili işlemler fail-closed olarak reddediliyor")
|
||
|
||
def test_every_ansible_playbook_is_covered_or_known(self) -> None:
|
||
"""Katalog disinda kalan playbook varsa gorunur olmali (sessiz kapsam disi yok)."""
|
||
catalogue = {op.playbook for op in self.backend.list_operations() if op.is_fleet}
|
||
on_disk = {p.name for p in (REPO_ROOT / "ansible" / "playbooks").glob("bf-*.yml")}
|
||
self.assertEqual(on_disk - catalogue, set())
|
||
|
||
def test_operation_ids_match_admin_bf_menu_catalogue(self) -> None:
|
||
"""id'ler admin/bf-menu tablosuyla ayni olmali (iki giris noktasi ayrisamaz)."""
|
||
menu = (REPO_ROOT / "admin" / "bf-menu").read_text(encoding="utf-8")
|
||
for op in self.backend.list_operations():
|
||
if op.is_fleet:
|
||
with self.subTest(operation=op.id):
|
||
self.assertIn(f'"{op.id}|', menu)
|
||
|
||
|
||
class TestDestructiveClassification(StaticCase):
|
||
YIKICI = ("rdp-restart", "rustdesk-restart", "wireguard-restart", "service-restart",
|
||
"gui-off", "gui-on", "run-script", "reboot", "deploy-update",
|
||
"install-local", "install-resume", "install-module", "tools-git-pull")
|
||
SALT_OKUNUR = ("ping", "uptime", "disk", "docker", "packages", "security-updates",
|
||
"provisioning-status", "enrollment-status", "fieldos-version",
|
||
"remote-channels", "offline-ready", "collect-logs", "install-preflight",
|
||
"install-state", "install-log", "central-targets", "info-summary",
|
||
"tools-release", "tools-support-bundle")
|
||
|
||
def test_destructive_ids_are_flagged(self) -> None:
|
||
flagged = set(ui.destructive_operations(self.backend.list_operations()))
|
||
for operation_id in self.YIKICI:
|
||
with self.subTest(operation=operation_id):
|
||
self.assertIn(operation_id, flagged)
|
||
self.assertTrue(ui.operation_is_destructive(operation_id))
|
||
|
||
def test_read_only_ids_are_not_flagged(self) -> None:
|
||
flagged = set(ui.destructive_operations(self.backend.list_operations()))
|
||
for operation_id in self.SALT_OKUNUR:
|
||
with self.subTest(operation=operation_id):
|
||
self.assertNotIn(operation_id, flagged)
|
||
self.assertFalse(ui.operation_is_destructive(operation_id))
|
||
|
||
def test_destructive_flag_reaches_the_plan(self) -> None:
|
||
plan = self.backend.build_command(self.operation("reboot"), self.group_target(),
|
||
{"grace": "300"})
|
||
self.assertTrue(plan.destructive)
|
||
plan = self.backend.build_command(self.operation("ping"), self.group_target())
|
||
self.assertFalse(plan.destructive)
|
||
|
||
def test_unknown_operation_is_not_destructive(self) -> None:
|
||
self.assertFalse(ui.operation_is_destructive("boyle-bir-islem-yok"))
|
||
|
||
|
||
class TestCommandPreview(StaticCase):
|
||
def test_fleet_preview_shows_the_exact_command(self) -> None:
|
||
operation = self.available_fleet_operation()
|
||
target = self.dealer_target()
|
||
plan = self.backend.build_command(operation, target, self.valid_options(operation))
|
||
preview = ui.preview_text(plan, operation=operation, target=target)
|
||
self.assertTrue(plan.lines, "en az bir komut satiri uretilmeli")
|
||
for line in plan.lines:
|
||
self.assertIn(line, preview)
|
||
self.assertIn("ansible-playbook", preview)
|
||
self.assertIn("--limit", preview)
|
||
self.assertIn("bf-12010193", preview)
|
||
self.assertIn(operation.playbook, preview)
|
||
self.assertIn("Çalıştırılacak komut", preview)
|
||
|
||
def test_preview_marks_destructive_operations(self) -> None:
|
||
operation = self.operation("deploy-update")
|
||
plan = self.backend.build_command(
|
||
operation, self.group_target(),
|
||
{"packages": "docker-ce=5:27.0.3-1~ubuntu.24.04~noble"},
|
||
)
|
||
preview = ui.preview_text(plan, operation=operation)
|
||
self.assertIn("UYARI", preview)
|
||
self.assertIn("YIKICI", preview)
|
||
self.assertIn("update_packages=", preview)
|
||
|
||
def test_preview_lists_target_and_scope_in_turkish(self) -> None:
|
||
operation = self.operation("provisioning-status")
|
||
target = self.group_target()
|
||
plan = self.backend.build_command(operation, target)
|
||
preview = ui.preview_text(plan, operation=operation, target=target)
|
||
self.assertIn("dalga", preview.lower())
|
||
self.assertIn("dalga kapılı", preview)
|
||
|
||
def test_preview_of_report_only_plan_says_nothing_runs(self) -> None:
|
||
operation = self.operation("central-targets")
|
||
plan = self.backend.build_command(operation, None)
|
||
self.assertTrue(plan.is_report_only)
|
||
preview = ui.preview_text(plan, operation=operation)
|
||
self.assertIn("komut çalıştırmaz", preview)
|
||
self.assertNotIn("Çalıştırılacak komut(lar)", preview)
|
||
|
||
def test_check_mode_is_visible_in_command_and_preview(self) -> None:
|
||
operation = self.available_fleet_operation()
|
||
target = self.group_target()
|
||
plan = self.backend.build_command(
|
||
operation, target, self.valid_options(operation), check_mode=True
|
||
)
|
||
self.assertTrue(plan.check_mode)
|
||
self.assertIn("--check", " ".join(plan.lines))
|
||
preview = ui.preview_text(plan, operation=operation, target=target)
|
||
self.assertIn("--check", preview)
|
||
self.assertIn("değişiklik uygulanmaz", preview)
|
||
|
||
def test_options_change_the_previewed_command(self) -> None:
|
||
operation = self.operation("disk")
|
||
options = {spec.name: "75" for spec in operation.options if "disk" in spec.name}
|
||
if not options:
|
||
self.skipTest("disk isleminin esik secenegi yok")
|
||
plan = self.backend.build_command(operation, self.group_target(), options)
|
||
self.assertIn("disk_warn_pct=75", " ".join(plan.lines))
|
||
|
||
def test_invalid_option_is_refused_with_turkish_message(self) -> None:
|
||
operation = self.operation("service-restart")
|
||
name = "service" if any(s.name == "service" for s in operation.options) else "service_name"
|
||
with self.assertRaises(ui.CommandError) as ctx:
|
||
self.backend.build_command(operation, self.group_target(), {name: "kritik-servis"})
|
||
self.assertIn("izinli", str(ctx.exception).lower())
|
||
|
||
def test_missing_required_option_is_refused(self) -> None:
|
||
operation = self.operation("deploy-update")
|
||
name = "packages" if any(s.name == "packages" for s in operation.options) \
|
||
else "update_packages"
|
||
with self.assertRaises(ui.CommandError):
|
||
self.backend.build_command(operation, self.group_target(), {name: ""})
|
||
|
||
|
||
class TestGatesAndTargets(StaticCase):
|
||
def test_wave_operation_requires_wave_token(self) -> None:
|
||
# A wave-scoped READ-ONLY report asks for no gate token -- the bash
|
||
# console behaves the same way (see bfos/operations.confirmation_plan:
|
||
# only release-gated runs and the production wave ask for a written
|
||
# token). What must hold is that the built command carries the wave
|
||
# scope and the confirmation flag, so the playbook's own gate accepts it.
|
||
steps = self.backend.gate_steps(self.operation("provisioning-status"),
|
||
self.group_target())
|
||
tokens = [s.token for s in steps if s.kind == "token"]
|
||
self.assertEqual([], tokens,
|
||
"a read-only wave report must not ask for a gate token")
|
||
plan = self.backend.build_command(self.operation("provisioning-status"),
|
||
self.group_target(), {}, {})
|
||
argv = list(plan.commands[0])
|
||
self.assertIn("operation_wave=pilot_1", argv)
|
||
self.assertIn("operation_confirmed=true", argv)
|
||
|
||
def test_release_gated_operation_requires_wave_token(self) -> None:
|
||
# The gate that must never disappear: reboot/deploy-update are
|
||
# release-gated, so the wave name has to be typed before anything runs.
|
||
steps = self.backend.gate_steps(self.operation("reboot"),
|
||
self.group_target())
|
||
tokens = [s.token for s in steps if s.kind == "token"]
|
||
self.assertIn("pilot_1", tokens)
|
||
|
||
def test_destructive_operation_requires_second_approval(self) -> None:
|
||
steps = self.backend.gate_steps(self.operation("gui-off"), self.group_target())
|
||
self.assertTrue([s for s in steps if s.kind == "approval"], steps)
|
||
|
||
def test_production_wave_requires_production_token(self) -> None:
|
||
steps = self.backend.gate_steps(self.operation("gui-off"),
|
||
self.group_target("production"))
|
||
tokens = [s.token for s in steps if s.kind == "token"]
|
||
self.assertIn("production", tokens)
|
||
unlocks = {name for step in steps for name in step.unlocks}
|
||
self.assertIn("production_override", unlocks)
|
||
|
||
def test_release_scope_refuses_single_device_and_fleet(self) -> None:
|
||
operation = self.operation("reboot")
|
||
with self.assertRaises(ui.CommandError) as ctx:
|
||
self.backend.build_command(operation, self.dealer_target(), {"grace": "300"})
|
||
self.assertIn("dalga", str(ctx.exception).lower())
|
||
with self.assertRaises(ui.CommandError):
|
||
self.backend.build_command(operation, ui.Target("all", "all"), {"grace": "300"})
|
||
|
||
def test_wave_scope_refuses_a_non_wave_group(self) -> None:
|
||
operation = self.operation("gui-on")
|
||
with self.assertRaises(ui.CommandError):
|
||
self.backend.build_command(operation, self.group_target("blueforce"))
|
||
|
||
def test_wave_scope_on_a_dealer_resolves_the_wave(self) -> None:
|
||
plan = self.backend.build_command(self.operation("gui-off"), self.dealer_target())
|
||
line = " ".join(plan.lines)
|
||
self.assertIn("operation_wave=pilot_1", line)
|
||
self.assertIn("operation_confirmed=true", line)
|
||
self.assertIn("bf-12010193", line)
|
||
|
||
def test_unknown_dealer_outside_inventory_is_refused(self) -> None:
|
||
operation = self.available_fleet_operation()
|
||
with self.assertRaises(ui.CommandError) as ctx:
|
||
self.backend.build_command(operation, self.dealer_target("12010000"))
|
||
self.assertIn("12010000", str(ctx.exception))
|
||
|
||
def test_invalid_dealer_never_builds_a_command(self) -> None:
|
||
operation = self.available_fleet_operation()
|
||
with self.assertRaises(ui.CommandError):
|
||
self.backend.build_command(operation, self.dealer_target("123"))
|
||
|
||
def test_grants_change_the_gated_command(self) -> None:
|
||
operation = self.operation("reboot")
|
||
target = self.group_target("production")
|
||
options = {"grace": "120"}
|
||
if not any(s.name == "grace" for s in operation.options):
|
||
options = {"reboot_grace_seconds": "120"}
|
||
draft = self.backend.build_command(operation, target, options)
|
||
final = self.backend.build_command(
|
||
operation, target, options,
|
||
grants={"wave_gate_confirmed": True, "production_override": True,
|
||
"allow_production_reboot": True},
|
||
)
|
||
joined = " ".join(final.lines)
|
||
self.assertIn("allow_production_reboot=true", joined)
|
||
self.assertNotIn("allow_production_reboot=true", " ".join(draft.lines))
|
||
self.assertNotEqual(draft.lines, final.lines)
|
||
|
||
def test_no_gate_flag_is_ever_added_silently(self) -> None:
|
||
"""Onaysiz hicbir komut kapi bayragi tasimamali (fail-closed)."""
|
||
checked = 0
|
||
for operation in self.backend.list_operations():
|
||
if not operation.is_fleet or operation.scope == "simple":
|
||
continue
|
||
target = self.group_target("production" if operation.scope == "release"
|
||
else "pilot_1")
|
||
options = self.valid_options(operation)
|
||
try:
|
||
plan = self.backend.build_command(operation, target, options)
|
||
except ui.CommandError:
|
||
# Playbook eksik/kapi reddi: komut URETILMEDI, bu da guvenli.
|
||
continue
|
||
with self.subTest(operation=operation.id):
|
||
joined = " ".join(plan.lines)
|
||
for flag in ("wave_gate_confirmed=true", "production_override=true",
|
||
"allow_production_reboot=true"):
|
||
self.assertNotIn(flag, joined,
|
||
f"{operation.id}: onaysiz kapi bayragi uretildi")
|
||
checked += 1
|
||
self.assertGreaterEqual(checked, 1)
|
||
|
||
|
||
class TestRunnerStreaming(StaticCase):
|
||
def test_run_command_streams_lines_and_returns_exit_code(self) -> None:
|
||
lines: list[str] = []
|
||
result = ui.run_command(
|
||
[sys.executable, "-c", "print('merhaba'); print('dunya')"],
|
||
on_line=lines.append,
|
||
)
|
||
self.assertEqual(result.exit_code, 0)
|
||
self.assertTrue(result.ok)
|
||
self.assertEqual(lines, ["merhaba", "dunya"])
|
||
self.assertIn("merhaba", result.output)
|
||
|
||
def test_run_command_reports_nonzero_exit_code(self) -> None:
|
||
result = ui.run_command([sys.executable, "-c", "import sys; sys.exit(3)"])
|
||
self.assertEqual(result.exit_code, 3)
|
||
self.assertFalse(result.ok)
|
||
|
||
def test_missing_command_gives_turkish_message(self) -> None:
|
||
result = ui.run_command(["boyle-bir-komut-yok-12345"])
|
||
self.assertEqual(result.exit_code, 127)
|
||
self.assertFalse(result.ok)
|
||
self.assertIn("bulunamadı", result.output)
|
||
self.assertIn("bf-bootstrap", result.output)
|
||
|
||
def test_run_plan_stops_at_the_first_failure(self) -> None:
|
||
plan = ui.CommandPlan(
|
||
commands=(
|
||
(sys.executable, "-c", "import sys; print('ilk'); sys.exit(4)"),
|
||
(sys.executable, "-c", "print('IKINCI-CALISMAMALI')"),
|
||
)
|
||
)
|
||
lines: list[str] = []
|
||
result = ui.run_plan(plan, on_line=lines.append)
|
||
self.assertEqual(result.exit_code, 4)
|
||
self.assertFalse(result.ok)
|
||
self.assertIn("ilk", lines)
|
||
self.assertNotIn("IKINCI-CALISMAMALI", lines)
|
||
|
||
def test_run_plan_executes_nothing_for_a_report_plan(self) -> None:
|
||
plan = ui.CommandPlan(report_lines=("satır 1", "satır 2"))
|
||
lines: list[str] = []
|
||
result = ui.run_plan(plan, on_line=lines.append)
|
||
self.assertTrue(result.ok)
|
||
self.assertEqual(lines, ["satır 1", "satır 2"])
|
||
self.assertEqual(result.exit_code, 0)
|
||
|
||
def test_dry_run_plan_does_not_execute(self) -> None:
|
||
plan = ui.CommandPlan(
|
||
commands=((sys.executable, "-c", "print('CALISMAMALI')"),),
|
||
dry_run=True,
|
||
)
|
||
lines: list[str] = []
|
||
result = ui.run_plan(plan, on_line=lines.append)
|
||
self.assertTrue(result.ok)
|
||
# Yalnizca "[deneme] <komut>" satiri yazilir; komutun CIKTISI yazilmaz.
|
||
self.assertTrue(lines)
|
||
self.assertTrue(all(line.startswith("[deneme] ") for line in lines), lines)
|
||
self.assertNotIn("CALISMAMALI", result.output)
|
||
self.assertIn("[deneme]", result.output)
|
||
|
||
|
||
class TestInventoryHelpers(StaticCase):
|
||
def test_parses_groups_and_hosts(self) -> None:
|
||
groups = ui.parse_inventory(self.inventory)
|
||
self.assertEqual(groups.get("pilot_1"), ["bf-12010193"])
|
||
self.assertEqual(groups.get("lab"), ["bf-12010101"])
|
||
self.assertEqual(groups.get("production"), ["bf-12010999"])
|
||
|
||
def test_wave_list_follows_rollout_order(self) -> None:
|
||
groups = ui.parse_inventory(self.inventory)
|
||
self.assertEqual(ui.waves_in_inventory(groups), ("lab", "pilot_1", "production"))
|
||
self.assertEqual(ui.WAVES,
|
||
("lab", "pilot_1", "pilot_2", "wave_1", "wave_2", "production"))
|
||
|
||
def test_device_wave_lookup(self) -> None:
|
||
groups = ui.parse_inventory(self.inventory)
|
||
self.assertEqual(ui.device_wave("bf-12010193", groups), "pilot_1")
|
||
self.assertIsNone(ui.device_wave("bf-99999999", groups))
|
||
|
||
def test_missing_inventory_does_not_crash(self) -> None:
|
||
self.assertEqual(ui.parse_inventory(Path("/yok/boyle/bir/dosya.yml")), {})
|
||
|
||
def test_missing_machine_inventory_is_reported_in_turkish(self) -> None:
|
||
backend = ui.CatalogueBackend(ui.GuiContext(repo_root=REPO_ROOT, inventory=None))
|
||
note = backend.inventory_note()
|
||
self.assertIn("envanter", note.lower())
|
||
self.assertIn("hosts.yml", note)
|
||
|
||
def test_inventory_note_survives_and_warns_without_ansible(self) -> None:
|
||
"""Not uretimi hicbir durumda patlamamali (eksik import regresyonu)."""
|
||
for backend in (
|
||
ui.CatalogueBackend(ui.GuiContext(repo_root=REPO_ROOT,
|
||
inventory=self.inventory)),
|
||
self.backend,
|
||
):
|
||
with self.subTest(backend=type(backend).__name__):
|
||
note = backend.inventory_note()
|
||
self.assertIn("Envanter", note)
|
||
if shutil.which("ansible-playbook") is None:
|
||
self.assertIn("ansible-playbook bulunamadı", note)
|
||
|
||
def test_dealer_wave_gate_requires_inventory(self) -> None:
|
||
backend = ui.CatalogueBackend(ui.GuiContext(repo_root=REPO_ROOT, inventory=None))
|
||
problem = backend.validate_target(
|
||
ui.operation_by_id("gui-off") or ui.OPERATIONS[0], self.dealer_target()
|
||
)
|
||
self.assertIsNotNone(problem)
|
||
|
||
|
||
class TestCatalogueBackendFallback(StaticCase):
|
||
"""Gercek moduller yoksa da ayni kurallar gecerli olmali (ayna backend)."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
super().setUpClass()
|
||
cls.fallback = ui.CatalogueBackend(cls.ctx)
|
||
|
||
def test_fallback_has_the_same_fleet_operations(self) -> None:
|
||
fleet = {op.id for op in self.fallback.list_operations() if op.is_fleet}
|
||
self.assertGreaterEqual(len(fleet), 21)
|
||
self.assertIn("ping", fleet)
|
||
self.assertIn("deploy-update", fleet)
|
||
|
||
def test_fallback_builds_the_same_core_arguments(self) -> None:
|
||
operation = None
|
||
for candidate in self.fallback.list_operations():
|
||
if candidate.is_fleet and candidate.scope == "simple" and candidate.playbook \
|
||
and ui.playbook_path(REPO_ROOT, candidate.playbook).is_file():
|
||
operation = candidate
|
||
break
|
||
if operation is None:
|
||
self.skipTest("kullanilabilir basit kapsamli filo islemi yok")
|
||
options = {spec.name: self.SAMPLE_VALUES[spec.name]
|
||
for spec in operation.options if spec.name in self.SAMPLE_VALUES}
|
||
plan = self.fallback.build_command(operation, self.dealer_target(), options)
|
||
joined = " ".join(plan.lines)
|
||
self.assertIn("ansible-playbook", joined)
|
||
self.assertIn(f"-i {self.ctx.inventory}", joined)
|
||
self.assertIn("--limit bf-12010193", joined)
|
||
|
||
def test_fallback_uses_the_same_destructive_table(self) -> None:
|
||
destructive = set(ui.destructive_operations(self.fallback.list_operations()))
|
||
self.assertEqual(destructive, set(ui.destructive_operations()))
|
||
|
||
def test_fallback_refuses_missing_inventory(self) -> None:
|
||
backend = ui.CatalogueBackend(ui.GuiContext(repo_root=REPO_ROOT, inventory=None))
|
||
with self.assertRaises(ui.CommandError) as ctx:
|
||
backend.build_command(ui.operation_by_id("uptime"), self.dealer_target())
|
||
self.assertIn("envanter", str(ctx.exception).lower())
|
||
|
||
def test_fallback_refuses_scope_mismatch_with_the_playbook(self) -> None:
|
||
"""Tablo playbook'la celisirse komut URETILMEZ (bf_playbook_check kurali)."""
|
||
wrong = ui.Operation(
|
||
id="uptime", label="Yanlış kapsam", category="Cihaz işlemleri",
|
||
destructive=False, scope="wave", playbook="bf-uptime.yml",
|
||
)
|
||
with self.assertRaises(ui.CommandError) as ctx:
|
||
self.fallback.build_command(wrong, self.group_target())
|
||
self.assertIn("bf-uptime.yml", str(ctx.exception))
|
||
self.assertIn("kapsam", str(ctx.exception).lower())
|
||
|
||
def test_fallback_installs_only_with_a_valid_dealer(self) -> None:
|
||
installer = REPO_ROOT / ui.INSTALLER_REL
|
||
if not installer.is_file():
|
||
self.skipTest("kurulum betigi bu checkout'ta yok")
|
||
plan = self.fallback.build_command(
|
||
ui.operation_by_id("install-preflight"), self.dealer_target()
|
||
)
|
||
joined = " ".join(plan.lines)
|
||
self.assertIn("--dealer-id 12010193", joined)
|
||
self.assertIn("--check", joined)
|
||
self.assertFalse(plan.destructive)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tasarim sozlesmesi (docs/DESIGN.md + web/static/tokens.css)
|
||
# ---------------------------------------------------------------------------
|
||
def _srgb_channel(value: float) -> float:
|
||
return value / 12.92 if value <= 0.03928 else ((value + 0.055) / 1.055) ** 2.4
|
||
|
||
|
||
def relative_luminance(colour: str) -> float:
|
||
"""WCAG bagil parlaklik (hex token rengi)."""
|
||
r, g, b = (int(colour[index:index + 2], 16) / 255 for index in (1, 3, 5))
|
||
return 0.2126 * _srgb_channel(r) + 0.7152 * _srgb_channel(g) + 0.0722 * _srgb_channel(b)
|
||
|
||
|
||
def contrast_ratio(first: str, second: str) -> float:
|
||
"""Iki token rengi arasindaki WCAG kontrast orani."""
|
||
a, b = relative_luminance(first), relative_luminance(second)
|
||
lighter, darker = max(a, b), min(a, b)
|
||
return (lighter + 0.05) / (darker + 0.05)
|
||
|
||
|
||
class TestThemeMirrorsTokensCss(StaticCase):
|
||
"""bfos/ui_theme.py, web/static/tokens.css'in BIREBIR aynasi olmali.
|
||
|
||
Iki yuzey (cihaz konsolu + merkez panel) ayni sozlugu konusur; bir tarafta
|
||
degistirilip digerinde unutulan bir token sessiz bir ayrisma olurdu. Bu
|
||
yuzden aynalama testtir (docs/DESIGN.md, "Rules").
|
||
"""
|
||
|
||
CSS_PATH = REPO_ROOT / "web" / "static" / "tokens.css"
|
||
|
||
#: Tk'de karsiligi olmayan token'lar (golge, yazi tipi listesi, satir
|
||
#: yuksekligi orani, yuvarlaklik): konsol bunlari kullanmaz, ama CSS'te
|
||
#: kalirlar cunku merkez panel kullanir.
|
||
NOT_MIRRORED = (
|
||
"--font-ui", "--font-data",
|
||
"--leading-tight", "--leading-normal",
|
||
"--shadow-sm", "--shadow-md",
|
||
"--radius", "--radius-lg",
|
||
)
|
||
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
super().setUpClass()
|
||
cls.theme = ui.theme
|
||
cls.css = cls._parse_tokens(cls.CSS_PATH.read_text(encoding="utf-8"))
|
||
|
||
@staticmethod
|
||
def _parse_tokens(text: str) -> dict:
|
||
"""`--ad: deger;` ciftlerini toplar, var(--x) referanslarini cozer."""
|
||
raw = {
|
||
match.group(1): match.group(2).strip()
|
||
for match in re.finditer(r"(--[a-z0-9-]+)\s*:\s*([^;]+);", text)
|
||
}
|
||
resolved = {}
|
||
for name, value in raw.items():
|
||
reference = re.fullmatch(r"var\((--[a-z0-9-]+)\)", value)
|
||
resolved[name] = raw.get(reference.group(1), value) if reference else value
|
||
return resolved
|
||
|
||
def test_the_mirror_source_exists(self) -> None:
|
||
self.assertTrue(self.CSS_PATH.is_file(), str(self.CSS_PATH))
|
||
self.assertTrue(self.css, "tokens.css icinde token bulunamadi")
|
||
|
||
def test_every_mirrored_value_equals_the_css(self) -> None:
|
||
for name, value in self.theme.ALL_TOKENS.items():
|
||
with self.subTest(token=name):
|
||
self.assertIn(name, self.css, f"{name} tokens.css'te yok")
|
||
css_value = self.css[name]
|
||
if css_value.endswith("px"):
|
||
css_value = css_value[:-2]
|
||
self.assertEqual(value, css_value.strip(),
|
||
f"{name}: ui_theme.py ile tokens.css ayrismis")
|
||
|
||
def test_the_colour_palette_is_fully_mirrored(self) -> None:
|
||
"""CSS'teki her RENK tokeninin Python karsiligi olmali (fazla renk = hata)."""
|
||
css_colours = {name for name, value in self.css.items() if value.startswith("#")}
|
||
python_colours = {name for name, value in self.theme.TOKENS.items()
|
||
if value.startswith("#")}
|
||
self.assertEqual(css_colours, python_colours)
|
||
self.assertTrue(css_colours)
|
||
|
||
def test_state_and_severity_pairs_have_readable_contrast(self) -> None:
|
||
"""Durum renkleri gunesli atolyede okunmali; rozet sozcugu de yaninda."""
|
||
self.assertGreaterEqual(contrast_ratio(self.theme.TEXT, self.theme.BG_APP), 7.0)
|
||
self.assertGreaterEqual(contrast_ratio(self.theme.TEXT, self.theme.BG_PANEL), 7.0)
|
||
self.assertGreaterEqual(contrast_ratio(self.theme.TEXT_MUTED,
|
||
self.theme.BG_PANEL), 4.5)
|
||
self.assertGreaterEqual(contrast_ratio(self.theme.TEXT_FAINT,
|
||
self.theme.BG_PANEL), 3.0)
|
||
self.assertGreaterEqual(contrast_ratio(self.theme.TEXT_ON_DARK,
|
||
self.theme.BG_HEADER), 7.0)
|
||
self.assertGreaterEqual(contrast_ratio(self.theme.TEXT_ON_ACCENT,
|
||
self.theme.ACCENT), 4.5)
|
||
# En dusuk cift "yesil zemin + yesil yazi" (4.42:1); esik bunun altina inmez.
|
||
for key, word in self.theme.STATE_WORDS.items():
|
||
with self.subTest(state=word):
|
||
self.assertGreaterEqual(
|
||
contrast_ratio(self.theme.STATE_TEXT_COLOUR[key],
|
||
self.theme.STATE_FILL[key]), 4.4)
|
||
for severity in self.theme.SEVERITIES:
|
||
with self.subTest(severity=severity):
|
||
self.assertGreaterEqual(
|
||
contrast_ratio(self.theme.SEVERITY_TEXT[severity],
|
||
self.theme.SEVERITY_FILL[severity]), 4.4)
|
||
|
||
def test_fonts_resolve_to_platform_families_without_downloads(self) -> None:
|
||
"""Webfont yok: aday listesi platformda VAR OLAN bir yuzeye duser."""
|
||
self.assertTrue(self.theme.FONT_UI_CANDIDATES)
|
||
self.assertTrue(self.theme.FONT_DATA_CANDIDATES)
|
||
self.assertTrue(self.theme.FONT_UI_FALLBACK)
|
||
self.assertTrue(self.theme.FONT_DATA_FALLBACK)
|
||
for family in (list(self.theme.FONT_UI_CANDIDATES)
|
||
+ list(self.theme.FONT_DATA_CANDIDATES)):
|
||
with self.subTest(family=family):
|
||
self.assertNotIn("/", family)
|
||
self.assertNotIn(".ttf", family.lower())
|
||
self.assertNotIn("http", family.lower())
|
||
|
||
def test_px_tokens_convert_to_tk_point_sizes(self) -> None:
|
||
sizes = (self.theme.SIZE_XS, self.theme.SIZE_SM, self.theme.SIZE_MD,
|
||
self.theme.SIZE_LG, self.theme.SIZE_XL, self.theme.SIZE_2XL)
|
||
for size in sizes:
|
||
with self.subTest(size=size):
|
||
self.assertLess(self.theme.point_size(size), size)
|
||
self.assertGreaterEqual(self.theme.point_size(size), 7)
|
||
# Siralama korunur: buyuk token daha buyuk puan uretir (esitlik yok).
|
||
points = [self.theme.point_size(size) for size in sizes]
|
||
self.assertEqual(points, sorted(points))
|
||
self.assertEqual(len(set(points)), len(points))
|
||
|
||
def test_the_space_rhythm_is_the_committed_one(self) -> None:
|
||
self.assertEqual(
|
||
[self.theme.SP_1, self.theme.SP_2, self.theme.SP_3, self.theme.SP_4,
|
||
self.theme.SP_5, self.theme.SP_6, self.theme.SP_7],
|
||
[4, 8, 12, 16, 24, 32, 48],
|
||
)
|
||
self.assertEqual(self.theme.ROW_H, 44)
|
||
self.assertEqual(self.theme.HEADER_H, 56)
|
||
|
||
|
||
class TestStateVocabulary(StaticCase):
|
||
"""Durum sozlugu: TEK sozcuk, renk tek basina anlam tasimaz."""
|
||
|
||
def test_the_five_words_are_exactly_the_committed_ones(self) -> None:
|
||
self.assertEqual(ui.DEVICE_STATES,
|
||
("KURULDU", "KAYITLI", "HAZIR", "HATA", "YOK"))
|
||
self.assertEqual(set(ui.DEVICE_STATES), set(ui.theme.STATE_WORDS.values()))
|
||
|
||
def test_every_word_has_its_own_badge_style(self) -> None:
|
||
styles = [ui.state_badge_style(word) for word in ui.DEVICE_STATES]
|
||
self.assertEqual(len(set(styles)), len(styles), styles)
|
||
for style in styles:
|
||
self.assertTrue(style.endswith(".Badge.TLabel"), style)
|
||
for key, style in ui.theme.STATE_BADGE_STYLE.items():
|
||
with self.subTest(state=key):
|
||
self.assertEqual(style,
|
||
ui.state_badge_style(ui.theme.STATE_WORDS[key]))
|
||
|
||
def test_every_word_has_its_own_text_colour(self) -> None:
|
||
self.assertEqual(len(set(ui.theme.STATE_TEXT_COLOUR.values())),
|
||
len(ui.DEVICE_STATES))
|
||
|
||
def test_badge_fills_come_from_the_token_set(self) -> None:
|
||
"""Dolgu renkleri token kumesinden gelir; uydurulmus renk yok."""
|
||
allowed = set(ui.theme.TOKENS.values()) | set(ui.theme.SEVERITY_FILL.values())
|
||
for key, fill in ui.theme.STATE_FILL.items():
|
||
with self.subTest(state=key):
|
||
self.assertIn(fill, allowed)
|
||
self.assertIn(ui.theme.STATE_TEXT_COLOUR[key], allowed)
|
||
|
||
def test_provisioned_is_not_the_action_accent(self) -> None:
|
||
"""KURULDU ile secili satir ayni maviye boyanamaz (tokens.css)."""
|
||
self.assertNotEqual(ui.theme.STATE_TEXT_COLOUR["provisioned"],
|
||
ui.theme.ACCENT)
|
||
self.assertNotEqual(ui.theme.STATE_FILL["provisioned"],
|
||
ui.theme.ACCENT_SOFT)
|
||
|
||
def test_unknown_word_falls_back_to_a_neutral_badge(self) -> None:
|
||
self.assertEqual(ui.state_badge_style("BOYLE-BIR-DURUM-YOK"),
|
||
ui.theme.STATE_BADGE_STYLE["absent"])
|
||
|
||
def test_the_state_machine_maps_to_one_word(self) -> None:
|
||
cases = (
|
||
({}, "YOK"),
|
||
({"phase": "PROVISIONED_OFFLINE"}, "KURULDU"),
|
||
({"phase": "PROVISIONED_OFFLINE",
|
||
"enrollment_status": "ENROLLED"}, "KAYITLI"),
|
||
({"fleet_status": "READY"}, "HAZIR"),
|
||
({"phase": "FAILED"}, "HATA"),
|
||
({"enrollment_status": "NOT_ENROLLED"}, "KURULDU"),
|
||
)
|
||
for values, expected in cases:
|
||
with self.subTest(values=values):
|
||
self.assertEqual(ui.state_word_from_values(values), expected)
|
||
|
||
def test_missing_or_broken_state_file_is_the_absent_state(self) -> None:
|
||
self.assertEqual(ui.device_state_word("/yok/boyle/bir/state.json"), "YOK")
|
||
self.assertEqual(ui.read_state_values("/yok/boyle/bir/state.json"), {})
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
broken = Path(tmp) / "state.json"
|
||
broken.write_text("{bu json degil", encoding="utf-8")
|
||
self.assertEqual(ui.device_state_word(str(broken)), "YOK")
|
||
|
||
def test_state_json_is_read_through_a_whitelist(self) -> None:
|
||
"""Jeton ve anahtar alanlari ekrana/loga ASLA okunmaz (K-14)."""
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
path = Path(tmp) / "state.json"
|
||
path.write_text(json.dumps({
|
||
"phase": "READY",
|
||
"device_id": "BF-12010193",
|
||
"enrollment_token": "SEKRET-JETON",
|
||
"private_key": "-----BEGIN PRIVATE KEY-----",
|
||
}), encoding="utf-8")
|
||
values = ui.read_state_values(str(path))
|
||
self.assertEqual(ui.device_state_word(str(path)), "HAZIR")
|
||
self.assertIn("device_id", values)
|
||
self.assertNotIn("enrollment_token", values)
|
||
self.assertNotIn("private_key", values)
|
||
self.assertNotIn("SEKRET-JETON", json.dumps(values))
|
||
self.assertNotIn("BEGIN PRIVATE KEY", json.dumps(values))
|
||
|
||
|
||
class TestPrivilegeIsVisible(StaticCase):
|
||
"""Yetki bir durumdur: yazma islemleri gizlenmez, KAPATILIR ve nedeni yazar."""
|
||
|
||
def test_privilege_levels_have_wording_and_a_badge(self) -> None:
|
||
self.assertEqual(ui.privilege_level(True), ui.PRIVILEGE_FULL)
|
||
self.assertEqual(ui.privilege_level(False), ui.PRIVILEGE_READ_ONLY)
|
||
self.assertEqual("SALT-OKUNUR", ui.privilege_word(ui.PRIVILEGE_READ_ONLY))
|
||
self.assertEqual("TAM YETKİ", ui.privilege_word(ui.PRIVILEGE_FULL))
|
||
self.assertEqual(set(ui.theme.PRIVILEGE_BADGE_STYLE),
|
||
{ui.PRIVILEGE_FULL, ui.PRIVILEGE_READ_ONLY})
|
||
|
||
def test_unprivileged_writes_stay_in_the_list_with_a_reason(self) -> None:
|
||
writes = [op for op in self.backend.list_operations() if op.needs_root]
|
||
reports = [op for op in self.backend.list_operations() if not op.needs_root]
|
||
self.assertTrue(writes, "katalogda yazma islemi yok; kapi test edilemez")
|
||
original = os.geteuid
|
||
try:
|
||
os.geteuid = lambda: 1000
|
||
for op in writes:
|
||
with self.subTest(operation=op.id):
|
||
self.assertEqual(ui.write_blocked_reason(op),
|
||
ui.NON_ROOT_DISABLED_HINT)
|
||
self.assertTrue(ui.operation_is_write(op))
|
||
for op in reports:
|
||
with self.subTest(operation=op.id):
|
||
self.assertIsNone(ui.write_blocked_reason(op))
|
||
self.assertFalse(ui.operation_is_write(op))
|
||
os.geteuid = lambda: 0
|
||
for op in writes:
|
||
with self.subTest(operation=op.id):
|
||
self.assertIsNone(ui.write_blocked_reason(op))
|
||
finally:
|
||
os.geteuid = original
|
||
|
||
def test_the_blocked_reason_names_the_command_to_get_privileges(self) -> None:
|
||
self.assertIn("sudo bf", ui.NON_ROOT_DISABLED_HINT)
|
||
self.assertIn("yönetici", ui.NON_ROOT_DISABLED_HINT.lower())
|
||
|
||
|
||
class TestConsoleLayoutContract(StaticCase):
|
||
"""Olculen sorunlar geri gelmesin: kaynak duzeyinde baglayici kurallar."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
super().setUpClass()
|
||
cls.source = UI_GUI_PATH.read_text(encoding="utf-8")
|
||
|
||
def test_the_view_has_no_hardcoded_colour(self) -> None:
|
||
"""Renk token'dir, literal degil (docs/DESIGN.md kural 1)."""
|
||
self.assertEqual(re.findall(r"#[0-9a-fA-F]{3,8}\b", self.source), [])
|
||
widgets_path = REPO_ROOT / "bfos" / "ui_widgets.py"
|
||
widgets_source = widgets_path.read_text(encoding="utf-8")
|
||
self.assertEqual(
|
||
re.findall(r"#[0-9a-fA-F]{3,8}\b", widgets_source), [])
|
||
|
||
def test_there_is_exactly_one_exit_control(self) -> None:
|
||
"""'Cik' tek kontrol: ya basit modda ya alt cubukta, ayni anda degil."""
|
||
self.assertEqual(self.source.count('text="Çıkış"'), 1)
|
||
self.assertIn("Çıkış", ui.SIMPLE_MODE_BUTTONS)
|
||
for needle in ("self.exit_button.pack(side=\"right\")",
|
||
"self.exit_button.pack_forget()"):
|
||
with self.subTest(needle=needle):
|
||
self.assertIn(needle, self.source)
|
||
|
||
def test_the_header_carries_the_privilege_element(self) -> None:
|
||
"""Yetki gomulu bir cumle degil, basliktaki gorunur bir rozet."""
|
||
for needle in ("PRIVILEGE_BADGE_STYLE", "self.privilege_badge",
|
||
'text="Yetki"', "ROOT_REQUIRED_NOTICE",
|
||
'style=self.styles["warn_band"]'):
|
||
with self.subTest(needle=needle):
|
||
self.assertIn(needle, self.source)
|
||
|
||
def test_the_header_carries_the_device_state_and_target(self) -> None:
|
||
for needle in ("self.state_badge", "self.header_target",
|
||
"device_state_word()", "state_badge_style("):
|
||
with self.subTest(needle=needle):
|
||
self.assertIn(needle, self.source)
|
||
|
||
def test_write_operations_are_disabled_not_hidden(self) -> None:
|
||
for needle in ('state="disabled"', "BLOCKED_LABEL", "AVAILABLE_LABEL",
|
||
"write_blocked_reason"):
|
||
with self.subTest(needle=needle):
|
||
self.assertIn(needle, self.source)
|
||
|
||
def test_the_status_strip_is_a_state_pill_not_a_plain_label(self) -> None:
|
||
self.assertIn("theme.severity_style", self.source)
|
||
self.assertNotIn('text="Hazır."', self.source)
|
||
|
||
def test_the_output_area_is_a_read_only_well(self) -> None:
|
||
for needle in ('state="disabled"', "BG_SUNKEN", "data_font",
|
||
'cursor="arrow"'):
|
||
with self.subTest(needle=needle):
|
||
self.assertIn(needle, self.source)
|
||
|
||
def test_the_view_reads_its_values_from_the_theme_module(self) -> None:
|
||
self.assertIn("ui_theme as theme", self.source)
|
||
self.assertTrue((REPO_ROOT / "bfos" / "ui_theme.py").is_file())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|