import json import os import tempfile import unittest from pathlib import Path from unittest.mock import MagicMock, patch from werkzeug.security import generate_password_hash from sqlalchemy import text from app import create_app, db from app.models import AutoUpdateSettings, SyncClient, WebUser from config import TestingConfig class AdminPanelTestConfig(TestingConfig): _TMP_DIR = tempfile.mkdtemp(prefix="wesp-admin-panel-tests-") BASE_DIR = _TMP_DIR DATA_DIR = os.path.join(_TMP_DIR, "data") SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'recipes_test.db')}" SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports_test.db')}"} AUTH_LOGIN = "admin" AUTH_PASSWORD = "admin-secret" SECRET_KEY = "unit-test-secret-key-not-default-32chars" KIOSK_AUTH_COOKIE_DAYS = 365 WESP_ADMIN_RESTART_CMD = "" WESP_ADMIN_STOP_CMD = "" WESP_SYSTEMD_UNIT = "" class AdminPanelAccessAndUsersTests(unittest.TestCase): def setUp(self) -> None: self._prev_security_settings_path = os.environ.get("WESP_SECURITY_SETTINGS_PATH") self._prev_network_diagnostics_path = os.environ.get("WESP_NETWORK_DIAGNOSTICS_PATH") _sec_dir = tempfile.mkdtemp(prefix="wesp-admin-security-") os.environ["WESP_SECURITY_SETTINGS_PATH"] = os.path.join(_sec_dir, "wesp_security_settings.json") _diag_dir = tempfile.mkdtemp(prefix="wesp-admin-diagnostics-") os.environ["WESP_NETWORK_DIAGNOSTICS_PATH"] = os.path.join( _diag_dir, "wesp_network_diagnostics.json" ) self.app = create_app(AdminPanelTestConfig) self.client = self.app.test_client() self.ctx = self.app.app_context() self.ctx.push() os.makedirs(AdminPanelTestConfig.DATA_DIR, exist_ok=True) db.create_all() admin = WebUser( login="admin", password_hash=generate_password_hash("admin-secret"), is_superuser=True, ) operator = WebUser( login="operator", password_hash=generate_password_hash("operator-secret"), is_superuser=False, ) db.session.add(admin) db.session.add(operator) db.session.commit() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() if self._prev_security_settings_path is None: os.environ.pop("WESP_SECURITY_SETTINGS_PATH", None) else: os.environ["WESP_SECURITY_SETTINGS_PATH"] = self._prev_security_settings_path if self._prev_network_diagnostics_path is None: os.environ.pop("WESP_NETWORK_DIAGNOSTICS_PATH", None) else: os.environ["WESP_NETWORK_DIAGNOSTICS_PATH"] = self._prev_network_diagnostics_path def _login(self, login: str, password: str) -> None: response = self.client.post( "/api/auth/login", json={"login": login, "password": password} ) self.assertEqual(response.status_code, 200) def test_admin_page_redirects_to_login_without_session(self) -> None: response = self.client.get("/admin", follow_redirects=False) self.assertEqual(response.status_code, 302) self.assertIn("/login", response.headers.get("Location", "")) def test_non_superuser_blocked_from_admin(self) -> None: self._login("operator", "operator-secret") html_response = self.client.get("/admin", follow_redirects=False) self.assertEqual(html_response.status_code, 302) self.assertIn("/recipes", html_response.headers.get("Location", "")) api_response = self.client.get("/api/admin/summary") self.assertEqual(api_response.status_code, 403) def test_superuser_can_open_admin(self) -> None: self._login("admin", "admin-secret") auth_check = self.client.get("/api/auth/check") self.assertEqual(auth_check.status_code, 200) self.assertTrue(auth_check.get_json().get("is_superuser")) html_response = self.client.get("/admin") self.assertEqual(html_response.status_code, 200) api_response = self.client.get("/api/admin/summary") self.assertEqual(api_response.status_code, 200) payload = api_response.get_json() self.assertEqual(payload.get("status"), "success") self.assertIn("counts", payload) app = payload.get("app") or {} self.assertIn("first_launch_at", app) self.assertIn("warranty_until", app) self.assertIn("warranty_days_remaining", app) self.assertIn("machine", payload) machine = payload.get("machine") or {} self.assertIn("os", machine) self.assertIn("cpu_model", machine) sync = payload.get("sync") or {} self.assertIn("queue_errors", sync) self.assertIsInstance(sync.get("queue_errors"), list) self.assertIn("initial_sync_progress", sync) self.assertIsInstance(sync.get("initial_sync_progress"), dict) security = payload.get("security") or {} self.assertIn("mac_lock_enabled", security) self.assertIn("session_remember_days", security) self.assertIn("kiosk_pair_token_ttl_seconds", security) au = payload.get("auto_update") or {} self.assertIn("enabled", au) self.assertIn("check_interval_sec", au) self.assertEqual(au.get("storage"), "sqlite:auto_update_settings") def test_auto_update_settings_patch_writes_recipes_db(self) -> None: self._login("admin", "admin-secret") cfg_path = os.path.join(AdminPanelTestConfig.DATA_DIR, "config.json") os.makedirs(AdminPanelTestConfig.DATA_DIR, exist_ok=True) with open(cfg_path, "w", encoding="utf-8") as f: json.dump({"version": "9.9.9"}, f) r = self.client.patch( "/api/admin/auto-update-settings", json={ "enabled": True, "auto_install": False, "gitea_url": "https://git.example.com", "gitea_owner": "farm", "gitea_repo": "wesp", "check_interval_sec": 120, }, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") self.assertTrue((body.get("auto_update") or {}).get("enabled")) self.assertEqual((body.get("auto_update") or {}).get("storage"), "sqlite:auto_update_settings") with open(cfg_path, encoding="utf-8") as f: disk = json.load(f) self.assertEqual(disk.get("version"), "9.9.9") self.assertNotIn("auto_update", disk) row = db.session.get(AutoUpdateSettings, 1) self.assertIsNotNone(row) assert row is not None self.assertTrue(row.enabled) self.assertFalse(row.auto_install) self.assertEqual(row.gitea_url, "https://git.example.com") self.assertEqual(row.gitea_owner, "farm") self.assertEqual(row.gitea_repo, "wesp") self.assertEqual(row.check_interval_sec, 120) def test_superuser_patch_gitea_secrets_writes_secret_file(self) -> None: self._login("admin", "admin-secret") r = self.client.patch( "/api/admin/gitea-secrets", json={"gitea_token": "test-token-xyz", "gitea_username": "", "gitea_password": ""}, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") self.assertTrue((body.get("auto_update") or {}).get("gitea_file_has_token")) secret_path = os.path.join(AdminPanelTestConfig.BASE_DIR, ".secret", "gitea_secrets.json") self.assertTrue(os.path.isfile(secret_path)) with open(secret_path, encoding="utf-8") as f: raw = json.load(f) self.assertIn("gitea_token", raw) def test_system_metrics_requires_superuser(self) -> None: self._login("operator", "operator-secret") denied = self.client.get("/api/admin/system-metrics") self.assertEqual(denied.status_code, 403) self.client.post("/api/auth/logout") self._login("admin", "admin-secret") ok = self.client.get("/api/admin/system-metrics") self.assertEqual(ok.status_code, 200) body = ok.get_json() self.assertEqual(body.get("status"), "success") self.assertIn("metrics", body) metrics = body["metrics"] self.assertIn("available", metrics) def test_hardware_metrics_history_requires_superuser(self) -> None: self._login("operator", "operator-secret") self.assertEqual( self.client.get("/api/admin/hardware-metrics-history").status_code, 403 ) self.client.post("/api/auth/logout") self._login("admin", "admin-secret") ok = self.client.get("/api/admin/hardware-metrics-history") self.assertEqual(ok.status_code, 200) body = ok.get_json() self.assertEqual(body.get("status"), "success") self.assertIn("points", body) self.assertEqual(body.get("points"), []) def test_hardware_metrics_history_reads_jsonl(self) -> None: import json import tempfile import time log_dir = tempfile.mkdtemp(prefix="wesp-hw-log-") log_path = os.path.join(log_dir, "h.jsonl") now = int(time.time()) with open(log_path, "w", encoding="utf-8") as f: f.write( json.dumps( { "u": now - 3600, "cpu_pct": 12.5, "ram_pct": 40.0, "swap_pct": 0.0, "disk_pct": 21.0, }, separators=(",", ":"), ) + "\n" ) self._login("admin", "admin-secret") self.app.config["WESP_ADMIN_HARDWARE_LOG_PATH"] = log_path r = self.client.get("/api/admin/hardware-metrics-history") self.assertEqual(r.status_code, 200) body = r.get_json() self.assertEqual(body.get("status"), "success") pts = body.get("points") or [] self.assertEqual(len(pts), 1) self.assertEqual(pts[0].get("cpu"), 12.5) self.assertEqual(pts[0].get("ram"), 40.0) self.assertEqual(pts[0].get("disk"), 21.0) def test_server_log_and_backup_require_superuser(self) -> None: self._login("operator", "operator-secret") self.assertEqual(self.client.get("/api/admin/server-log").status_code, 403) self.assertEqual(self.client.get("/api/admin/server-log/download").status_code, 403) self.assertEqual(self.client.get("/api/admin/backup.sqlite").status_code, 403) self.assertEqual( self.client.get("/api/admin/sqlite-snapshot?bind=recipes").status_code, 403 ) self.assertEqual( self.client.post("/api/admin/service-control", json={"action": "restart"}).status_code, 403, ) self.assertEqual(self.client.get("/api/admin/db/tree").status_code, 403) self.assertEqual( self.client.get("/api/admin/db/table?bind=recipes&table=web_user").status_code, 403, ) self.assertEqual(self.client.patch("/api/admin/db/row", json={}).status_code, 403) self.assertEqual( self.client.post("/api/admin/db/row/delete-preview", json={}).status_code, 403 ) self.assertEqual(self.client.post("/api/admin/db/row/delete", json={}).status_code, 403) self.assertEqual(self.client.post("/api/admin/restore.sqlite").status_code, 403) self.assertEqual( self.client.get("/api/admin/hardware-metrics-history").status_code, 403 ) self.assertEqual( self.client.patch("/api/admin/sync-settings", json={"role": "client"}).status_code, 403, ) self.assertEqual( self.client.patch( "/api/admin/security-settings", json={"mac_lock_enabled": True} ).status_code, 403, ) self.assertEqual(self.client.get("/api/admin/diagnostics/report").status_code, 403) self.assertEqual(self.client.get("/api/admin/diagnostics/settings").status_code, 403) self.assertEqual( self.client.patch("/api/admin/diagnostics/settings", json={"extra_targets": []}).status_code, 403, ) self.assertEqual( self.client.post("/api/admin/diagnostics/network-run", json={"scope": "all"}).status_code, 403, ) self.assertEqual(self.client.get("/api/admin/sync-diagnostics").status_code, 403) for _path in ( "/api/admin/sync-actions/requeue-stuck", "/api/admin/sync-actions/restart-local-sync", "/api/admin/sync-actions/refresh-runtime", "/api/admin/sync-actions/retry-report-push", ): self.assertEqual(self.client.post(_path, json={}).status_code, 403) def test_superuser_sync_diagnostics_contract(self) -> None: self._login("admin", "admin-secret") r = self.client.get("/api/admin/sync-diagnostics") self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") for key in ( "generated_at", "overview", "runtime", "queues", "bootstrap", "report_push", "clients", "conflicts", "consistency", "actions_capabilities", ): self.assertIn(key, body, msg=key) ov = body["overview"] self.assertIn("effective_role", ov) self.assertIn("connection", ov) self.assertIn("client_state", ov) self.assertIn("config", ov) rt = body["runtime"] self.assertEqual(rt.get("scope"), "process_local") self.assertIn("note", rt) self.assertIn("queues", body) self.assertIn("stats", body["queues"]) cap = body["actions_capabilities"] self.assertTrue(cap.get("requeue_stuck")) self.assertTrue(cap.get("refresh_diagnostics")) def test_superuser_sync_actions_smoke(self) -> None: self._login("admin", "admin-secret") r = self.client.post("/api/admin/sync-actions/requeue-stuck", json={}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) self.assertEqual((r.get_json() or {}).get("status"), "success") r2 = self.client.post("/api/admin/sync-actions/restart-local-sync", json={}) self.assertEqual(r2.status_code, 200) self.assertEqual((r2.get_json() or {}).get("status"), "success") self.assertFalse((r2.get_json() or {}).get("reloaded")) r3 = self.client.post("/api/admin/sync-actions/refresh-runtime", json={}) self.assertEqual(r3.status_code, 200) self.assertIn("runtime", r3.get_json() or {}) r4 = self.client.post("/api/admin/sync-actions/retry-report-push", json={}) self.assertEqual(r4.status_code, 200) self.assertFalse((r4.get_json() or {}).get("scheduled")) def test_superuser_patch_security_settings_updates_config(self) -> None: import json self._login("admin", "admin-secret") path = os.environ.get("WESP_SECURITY_SETTINGS_PATH", "") self.assertTrue(path) r = self.client.patch( "/api/admin/security-settings", json={ "mac_lock_enabled": True, "allowed_mac_addresses": "AA:BB:CC:DD:EE:FF", "kiosk_enforce_paired_only": False, "session_remember_days": 14, }, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") sec = body.get("security") or {} self.assertTrue(sec.get("mac_lock_enabled")) self.assertEqual(sec.get("allowed_mac_addresses"), "aa:bb:cc:dd:ee:ff") self.assertFalse(sec.get("kiosk_enforce_paired_only")) self.assertEqual(sec.get("session_remember_days"), 14) self.assertTrue(self.app.config.get("MAC_LOCK_ENABLED")) self.assertEqual(self.app.config.get("ALLOWED_MAC_ADDRESSES"), ["aa:bb:cc:dd:ee:ff"]) self.assertFalse(self.app.config.get("KIOSK_ENFORCE_PAIRED_ONLY")) with open(path, encoding="utf-8") as f: stored = json.load(f) self.assertTrue(stored.get("mac_lock_enabled")) def test_superuser_patch_security_llm_autostart(self) -> None: import json self._login("admin", "admin-secret") self.assertFalse(self.app.config.get("WESP_LLM_AUTOSTART")) r = self.client.patch("/api/admin/security-settings", json={"llm_autostart": True}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) sec = (r.get_json() or {}).get("security") or {} self.assertTrue(sec.get("llm_autostart")) self.assertTrue(self.app.config.get("WESP_LLM_AUTOSTART")) path = os.environ.get("WESP_SECURITY_SETTINGS_PATH", "") self.assertTrue(path) with open(path, encoding="utf-8") as f: stored = json.load(f) self.assertTrue(stored.get("llm_autostart")) def test_superuser_patch_security_admin_llm_enabled(self) -> None: import json self._login("admin", "admin-secret") self.assertFalse(self.app.config.get("WESP_ADMIN_LLM_ENABLED")) r = self.client.patch("/api/admin/security-settings", json={"admin_llm_enabled": True}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) sec = (r.get_json() or {}).get("security") or {} self.assertTrue(sec.get("admin_llm_enabled")) self.assertTrue(self.app.config.get("WESP_ADMIN_LLM_ENABLED")) path = os.environ.get("WESP_SECURITY_SETTINGS_PATH", "") with open(path, encoding="utf-8") as f: stored = json.load(f) self.assertTrue(stored.get("admin_llm_enabled")) def test_superuser_patch_security_llm_chat_db_context(self) -> None: import json self._login("admin", "admin-secret") self.assertFalse(self.app.config.get("WESP_LLM_CHAT_INCLUDE_DB_CONTEXT")) r = self.client.patch("/api/admin/security-settings", json={"llm_chat_db_context": True}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) sec = (r.get_json() or {}).get("security") or {} self.assertTrue(sec.get("llm_chat_db_context")) self.assertTrue(self.app.config.get("WESP_LLM_CHAT_INCLUDE_DB_CONTEXT")) path = os.environ.get("WESP_SECURITY_SETTINGS_PATH", "") with open(path, encoding="utf-8") as f: stored = json.load(f) self.assertTrue(stored.get("llm_chat_db_context")) def test_superuser_patch_security_llm_tools_enabled(self) -> None: import json self._login("admin", "admin-secret") self.assertFalse(self.app.config.get("WESP_LLM_TOOLS_ENABLED")) r = self.client.patch("/api/admin/security-settings", json={"llm_tools_enabled": True}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) sec = (r.get_json() or {}).get("security") or {} self.assertTrue(sec.get("llm_tools_enabled")) self.assertTrue(self.app.config.get("WESP_LLM_TOOLS_ENABLED")) path = os.environ.get("WESP_SECURITY_SETTINGS_PATH", "") with open(path, encoding="utf-8") as f: stored = json.load(f) self.assertTrue(stored.get("llm_tools_enabled")) def test_superuser_patch_sync_settings_writes_state_file(self) -> None: import json import tempfile sync_dir = tempfile.mkdtemp(prefix="wesp-admin-sync-settings-") sync_path = os.path.join(sync_dir, "sync_state.json") prev = os.environ.get("WESP_SYNC_CLIENT_STATE_PATH") os.environ["WESP_SYNC_CLIENT_STATE_PATH"] = sync_path try: self._login("admin", "admin-secret") r = self.client.patch( "/api/admin/sync-settings", json={"role": "client", "server_url": "http://10.0.0.12:5000"}, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") self.assertFalse(body.get("restart_scheduled")) self.assertFalse(body.get("sync_reloaded")) conn = body.get("connection") or {} self.assertEqual(conn.get("role"), "client") self.assertEqual(conn.get("server_url"), "http://10.0.0.12:5000") self.assertTrue(conn.get("sync_server_url_explicit")) with open(sync_path, encoding="utf-8") as f: data = json.load(f) self.assertEqual(data.get("role"), "client") self.assertEqual(data.get("server_url"), "http://10.0.0.12:5000") finally: if prev is None: os.environ.pop("WESP_SYNC_CLIENT_STATE_PATH", None) else: os.environ["WESP_SYNC_CLIENT_STATE_PATH"] = prev def test_superuser_patch_sync_settings_client_identity_fields(self) -> None: import json import tempfile sync_dir = tempfile.mkdtemp(prefix="wesp-admin-sync-state-fields-") sync_path = os.path.join(sync_dir, "sync_state.json") prev = os.environ.get("WESP_SYNC_CLIENT_STATE_PATH") os.environ["WESP_SYNC_CLIENT_STATE_PATH"] = sync_path try: self._login("admin", "admin-secret") nid = "11111111-1111-4111-8111-111111111111" r = self.client.patch( "/api/admin/sync-settings", json={ "client_id": nid, "client_name": "Node A", "first_bootstrap_done": True, }, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") st = body.get("state") or {} self.assertEqual(st.get("client_id"), nid) self.assertEqual(st.get("client_name"), "Node A") self.assertTrue(st.get("first_bootstrap_done")) with open(sync_path, encoding="utf-8") as f: data = json.load(f) self.assertEqual(data.get("client_id"), nid) self.assertEqual(data.get("client_name"), "Node A") self.assertTrue(data.get("first_bootstrap_done")) bad = self.client.patch( "/api/admin/sync-settings", json={"client_id": "not-a-uuid"}, ) self.assertEqual(bad.status_code, 400) finally: if prev is None: os.environ.pop("WESP_SYNC_CLIENT_STATE_PATH", None) else: os.environ["WESP_SYNC_CLIENT_STATE_PATH"] = prev def test_superuser_server_log_no_path_returns_error_payload(self) -> None: self._login("admin", "admin-secret") r = self.client.get("/api/admin/server-log") self.assertEqual(r.status_code, 200) body = r.get_json() self.assertEqual(body.get("status"), "error") self.assertEqual(body.get("code"), "no_path") dl = self.client.get("/api/admin/server-log/download") self.assertEqual(dl.status_code, 400) dlb = dl.get_json() self.assertEqual(dlb.get("code"), "no_path") def test_superuser_client_uploaded_logs_index_and_tail(self) -> None: self._login("admin", "admin-secret") upload_root = Path(self.app.config["WESP_CLIENT_LOG_UPLOAD_DIR"]) node_id = "11111111-1111-1111-1111-111111111111" d = upload_root / node_id d.mkdir(parents=True, exist_ok=True) (d / "20990101_120000_wesp.log").write_text("alpha_line\nbeta_line\n", encoding="utf-8") idx = self.client.get("/api/admin/client-uploaded-logs") self.assertEqual(idx.status_code, 200, idx.get_data(as_text=True)) data = idx.get_json() self.assertEqual(data.get("status"), "success") clients = data.get("clients") or [] self.assertTrue(any((c.get("api_client_id") == node_id) for c in clients)) tail = self.client.get( f"/api/admin/client-uploaded-logs/tail?client_id={node_id}&lines=50" ) self.assertEqual(tail.status_code, 200) tj = tail.get_json() self.assertEqual(tj.get("status"), "success") blob = "\n".join(tj.get("lines") or []) self.assertIn("alpha_line", blob) def test_operator_denied_client_uploaded_logs(self) -> None: self._login("operator", "operator-secret") r = self.client.get("/api/admin/client-uploaded-logs") self.assertEqual(r.status_code, 403) def test_superuser_diagnostics_settings_and_report(self) -> None: self._login("admin", "admin-secret") save = self.client.patch( "/api/admin/diagnostics/settings", json={ "extra_targets": [ {"role": "gateway", "host": "192.168.0.1", "title": "Главный роутер"}, {"role": "receiver", "host": "192.168.0.21", "title": "Приемник"}, ] }, ) self.assertEqual(save.status_code, 200, save.get_data(as_text=True)) saved = save.get_json() self.assertEqual(saved.get("status"), "success") self.assertEqual(len(saved.get("extra_targets") or []), 2) report_resp = self.client.get("/api/admin/diagnostics/report") self.assertEqual(report_resp.status_code, 200) report = report_resp.get_json() self.assertEqual(report.get("status"), "success") self.assertIn("node_role", report) self.assertIn("primary_targets", report) self.assertIn("security_checks", report) ids = {c.get("id") for c in (report.get("security_checks") or []) if isinstance(c, dict)} self.assertIn("secret_key_non_default", ids) self.assertIn("weak_credential_blocklist", ids) self.assertIn("web_users_login_not_weak_blocklist", ids) self.assertIn("web_users_password_not_in_weak_blocklist", ids) self.assertGreaterEqual(len(ids), 10) self.assertEqual(len(report.get("extra_targets") or []), 2) hw = report.get("hardware") or {} self.assertIn("rpi_gpio_available", hw) self.assertIn("hx711_driver_ready", hw) self.assertIn("simulation_mode", hw) self.assertIn("drivers_absent", hw) def test_superuser_traceroute_analyze_endpoint(self) -> None: self._login("admin", "admin-secret") text = ( " 1 192.168.1.1 (192.168.1.1) 2.1 ms 1.9 ms\n" " 2 8.8.8.8 (8.8.8.8) 10.0 ms 9.5 ms\n" ) r = self.client.post("/api/admin/diagnostics/traceroute-analyze", json={"text": text}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") self.assertEqual(body.get("stats", {}).get("total_hops"), 2) self.assertEqual(len(body.get("hops") or []), 2) bad = self.client.post("/api/admin/diagnostics/traceroute-analyze", json={"text": ""}) self.assertEqual(bad.status_code, 400) def test_operator_blocked_from_traceroute_analyze(self) -> None: self._login("operator", "operator-secret") r = self.client.post( "/api/admin/diagnostics/traceroute-analyze", json={"text": " 1 8.8.8.8 1 ms\n"}, ) self.assertEqual(r.status_code, 403) def test_superuser_traceroute_by_hosts_mocked(self) -> None: self._login("admin", "admin-secret") fake_lines = [ " 1 192.168.1.1 (192.168.1.1) 2.1 ms 1.9 ms", " 2 8.8.8.8 (8.8.8.8) 10.0 ms 9.5 ms", ] with patch( "app.routes.admin.trace_host", return_value={ "ok": True, "host": "8.8.8.8", "tool": "traceroute", "exit_code": 0, "lines": fake_lines, "command": "traceroute -n 8.8.8.8", }, ), patch("app.routes.admin.ping_host", return_value={"ok": True, "host": "8.8.8.8"}), patch( "app.routes.admin.tcp_probe", return_value={"ok": True, "host": "8.8.8.8", "port": 5000}, ): r = self.client.post( "/api/admin/diagnostics/traceroute-by-hosts", json={"hosts": "8.8.8.8\n8.8.8.8"}, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") results = body.get("results") or [] self.assertEqual(len(results), 1) self.assertEqual(results[0].get("host"), "8.8.8.8") self.assertIsNotNone(results[0].get("parse")) self.assertTrue((results[0].get("ping") or {}).get("ok")) self.assertTrue((results[0].get("tcp") or {}).get("ok")) def test_superuser_patch_sync_client_ip(self) -> None: with self.app.app_context(): row = SyncClient( node_id="client-node-1", client_name="Клиент 1", ip_address="10.10.0.5", port=5000, status="active", is_enabled=True, ) db.session.add(row) db.session.commit() self._login("admin", "admin-secret") r = self.client.patch( "/api/admin/sync-clients/client-node-1/ip", json={"ip_address": "10.10.0.42"}, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") self.assertEqual((body.get("client") or {}).get("ip_address"), "10.10.0.42") def test_superuser_network_run_uses_probe_helpers(self) -> None: self._login("admin", "admin-secret") tcp_mock = MagicMock(return_value={"ok": True, "host": "127.0.0.1", "port": 5000}) with patch("app.routes.admin.ping_host", return_value={"ok": True}), patch( "app.routes.admin.tcp_probe", tcp_mock ), patch( "app.routes.admin.trace_host", return_value={"ok": True, "tool": "traceroute"} ), patch( "app.routes.admin.http_probe", side_effect=lambda url, timeout_sec=3: {"ok": True, "url": url, "status_code": 204}, ): r = self.client.post( "/api/admin/diagnostics/network-run", json={"scope": "custom", "custom_targets": ["127.0.0.1"], "run_trace": True}, ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") rows = body.get("results") or [] self.assertEqual(len(rows), 1) self.assertTrue(rows[0].get("ok")) tcp_mock.assert_called() self.assertEqual(tcp_mock.call_args[0][1], 80) self.assertTrue((rows[0].get("probes") or {}).get("tcp", {}).get("ok")) def test_superuser_network_run_rejects_bad_tcp_port(self) -> None: self._login("admin", "admin-secret") r = self.client.post( "/api/admin/diagnostics/network-run", json={"scope": "custom", "custom_targets": ["127.0.0.1"], "tcp_port": 70000}, ) self.assertEqual(r.status_code, 400) def test_superuser_backup_sqlite_returns_zip(self) -> None: self._login("admin", "admin-secret") r = self.client.get("/api/admin/backup.sqlite") self.assertEqual(r.status_code, 200) self.assertEqual(r.mimetype, "application/zip") def test_superuser_sqlite_snapshot_returns_sqlite_file(self) -> None: self._login("admin", "admin-secret") r = self.client.get("/api/admin/sqlite-snapshot?bind=recipes") self.assertEqual(r.status_code, 200) self.assertTrue(r.data.startswith(b"SQLite format 3\x00")) r2 = self.client.get("/api/admin/sqlite-snapshot?bind=reports") self.assertEqual(r2.status_code, 200) self.assertTrue(r2.data.startswith(b"SQLite format 3\x00")) def test_sqlite_snapshot_bad_bind_400(self) -> None: self._login("admin", "admin-secret") r = self.client.get("/api/admin/sqlite-snapshot?bind=unknown") self.assertEqual(r.status_code, 400) def test_admin_db_browser_tree_table_and_patch_row(self) -> None: self._login("admin", "admin-secret") with self.app.app_context(): db.session.execute( text( "CREATE TABLE IF NOT EXISTS wesp_admin_browser_qa " "(id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT)" ) ) db.session.execute(text("DELETE FROM wesp_admin_browser_qa")) db.session.execute(text("INSERT INTO wesp_admin_browser_qa (label) VALUES ('alpha')")) db.session.commit() tr = self.client.get("/api/admin/db/tree") self.assertEqual(tr.status_code, 200) tree = tr.get_json() self.assertEqual(tree.get("status"), "success") binds = {b["bind"]: b for b in tree.get("binds", [])} self.assertIn("recipes", binds) names = {t["name"] for t in binds["recipes"].get("tables", [])} self.assertIn("wesp_admin_browser_qa", names) tb = self.client.get( "/api/admin/db/table?bind=recipes&table=wesp_admin_browser_qa&limit=10&offset=0" ) self.assertEqual(tb.status_code, 200, tb.get_data(as_text=True)) body = tb.get_json() self.assertGreaterEqual(len(body.get("rows") or []), 1) row = body["rows"][0] self.assertEqual(row.get("label"), "alpha") rk = {"_wesp_rowid": row["_wesp_rowid"]} patch = self.client.patch( "/api/admin/db/row", json={ "bind": "recipes", "table": "wesp_admin_browser_qa", "row_key": rk, "changes": {"label": "beta"}, }, ) self.assertEqual(patch.status_code, 200, patch.get_data(as_text=True)) tb2 = self.client.get( "/api/admin/db/table?bind=recipes&table=wesp_admin_browser_qa&limit=10&offset=0" ) row2 = tb2.get_json()["rows"][0] self.assertEqual(row2.get("label"), "beta") def test_admin_db_table_bad_identifier_400(self) -> None: self._login("admin", "admin-secret") r = self.client.get("/api/admin/db/table?bind=recipes&table=evil;drop") self.assertEqual(r.status_code, 400) def test_superuser_restore_sqlite_from_backup_zip(self) -> None: self._login("admin", "admin-secret") from flask import current_app from app.services.admin_dashboard_service import build_sqlite_backup_zip with self.app.app_context(): db.session.execute( text( "CREATE TABLE IF NOT EXISTS wesp_restore_marker " "(id INTEGER PRIMARY KEY, v TEXT)" ) ) db.session.execute(text("DELETE FROM wesp_restore_marker")) db.session.execute(text("INSERT INTO wesp_restore_marker VALUES (1, 'A')")) db.session.commit() zip_bytes, _ = build_sqlite_backup_zip(current_app) with self.app.app_context(): db.session.execute(text("UPDATE wesp_restore_marker SET v='B' WHERE id=1")) db.session.commit() db.session.remove() db.engine.dispose() from io import BytesIO r = self.client.post( "/api/admin/restore.sqlite", data={"file": (BytesIO(zip_bytes), "backup.zip")}, content_type="multipart/form-data", ) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) body = r.get_json() self.assertEqual(body.get("status"), "success") self.assertIn("restored", body) import sqlite3 uri = AdminPanelTestConfig.SQLALCHEMY_DATABASE_URI path = uri.replace("sqlite:///", "") con = sqlite3.connect(path) try: row = con.execute("SELECT v FROM wesp_restore_marker WHERE id=1").fetchone() self.assertIsNotNone(row) self.assertEqual(row[0], "A") finally: con.close() def test_admin_db_cascade_delete_preview_and_execute(self) -> None: self._login("admin", "admin-secret") with self.app.app_context(): db.session.execute(text("DROP TABLE IF EXISTS wesp_del_child")) db.session.execute(text("DROP TABLE IF EXISTS wesp_del_parent")) db.session.execute( text( "CREATE TABLE wesp_del_parent (id INTEGER PRIMARY KEY AUTOINCREMENT, n TEXT)" ) ) db.session.execute( text( "CREATE TABLE wesp_del_child (id INTEGER PRIMARY KEY AUTOINCREMENT, " "pid INTEGER NOT NULL REFERENCES wesp_del_parent(id))" ) ) db.session.execute(text("INSERT INTO wesp_del_parent (n) VALUES ('p')")) db.session.execute(text("INSERT INTO wesp_del_child (pid) VALUES (1)")) db.session.commit() pv = self.client.post( "/api/admin/db/row/delete-preview", json={ "bind": "recipes", "table": "wesp_del_parent", "row_key": {"_wesp_rowid": 1}, }, ) self.assertEqual(pv.status_code, 200, pv.get_data(as_text=True)) prev_body = pv.get_json() self.assertEqual(prev_body.get("total"), 2) dl = self.client.post( "/api/admin/db/row/delete", json={ "bind": "recipes", "table": "wesp_del_parent", "row_key": {"_wesp_rowid": 1}, }, ) self.assertEqual(dl.status_code, 200, dl.get_data(as_text=True)) left = self.client.get( "/api/admin/db/table?bind=recipes&table=wesp_del_parent&limit=10&offset=0" ) self.assertEqual(len(left.get_json().get("rows") or []), 0) left_c = self.client.get( "/api/admin/db/table?bind=recipes&table=wesp_del_child&limit=10&offset=0" ) self.assertEqual(len(left_c.get_json().get("rows") or []), 0) def test_service_control_without_cmd_returns_400(self) -> None: self._login("admin", "admin-secret") r = self.client.post("/api/admin/service-control", json={"action": "restart"}) self.assertEqual(r.status_code, 400) body = r.get_json() self.assertEqual(body.get("status"), "error") self.assertEqual(body.get("code"), "not_configured") def test_superuser_can_crud_users_with_safety_limits(self) -> None: self._login("admin", "admin-secret") create_response = self.client.post( "/api/admin/users", json={ "login": "qa-user", "password": "qa-secret", "is_superuser": False, }, ) self.assertEqual(create_response.status_code, 201) created_id = create_response.get_json().get("id") self.assertTrue(created_id) promote_response = self.client.patch( f"/api/admin/users/{created_id}", json={"is_superuser": True}, ) self.assertEqual(promote_response.status_code, 200) demote_response = self.client.patch( f"/api/admin/users/{created_id}", json={"is_superuser": False}, ) self.assertEqual(demote_response.status_code, 200) list_response = self.client.get("/api/admin/users") users = list_response.get_json().get("users", []) admin_row = next((user for user in users if user["login"] == "admin"), None) self.assertIsNotNone(admin_row) delete_admin_response = self.client.delete(f"/api/admin/users/{admin_row['id']}") self.assertEqual(delete_admin_response.status_code, 400) delete_created_response = self.client.delete(f"/api/admin/users/{created_id}") self.assertEqual(delete_created_response.status_code, 200) # После удаления/создания пользователь должен уметь войти. create_again = self.client.post( "/api/admin/users", json={"login": "qa-user", "password": "qa-secret", "is_superuser": False}, ) self.assertEqual(create_again.status_code, 201) self.client.post("/api/auth/logout") relogin = self.client.post( "/api/auth/login", json={"login": "qa-user", "password": "qa-secret"}, ) self.assertEqual(relogin.status_code, 200) def test_superuser_cannot_create_user_with_short_password(self) -> None: self._login("admin", "admin-secret") r = self.client.post( "/api/admin/users", json={"login": "short-pw-user", "password": "a1b2c3", "is_superuser": False}, ) self.assertEqual(r.status_code, 400, r.get_data(as_text=True)) self.assertIn("8", r.get_json().get("message", "")) def test_superuser_cannot_create_user_with_weak_password(self) -> None: self._login("admin", "admin-secret") r = self.client.post( "/api/admin/users", json={"login": "safe-login-1", "password": "password", "is_superuser": False}, ) self.assertEqual(r.status_code, 400, r.get_data(as_text=True)) msg = (r.get_json().get("message") or "").lower() self.assertTrue( "слаб" in msg or "пароль" in msg, msg=r.get_json().get("message"), ) def test_superuser_cannot_create_user_with_weak_login(self) -> None: self._login("admin", "admin-secret") r = self.client.post( "/api/admin/users", json={"login": "password", "password": "qa-secret-99", "is_superuser": False}, ) self.assertEqual(r.status_code, 400, r.get_data(as_text=True)) msg = (r.get_json().get("message") or "").lower() self.assertTrue( "логин" in msg or "слаб" in msg or "списк" in msg, msg=r.get_json().get("message"), ) def test_superuser_cannot_patch_user_to_weak_password(self) -> None: self._login("admin", "admin-secret") created = self.client.post( "/api/admin/users", json={"login": "patchme", "password": "qa-secret", "is_superuser": False}, ) self.assertEqual(created.status_code, 201, created.get_data(as_text=True)) cid = created.get_json().get("id") r = self.client.patch(f"/api/admin/users/{cid}", json={"password": "12345678"}) self.assertEqual(r.status_code, 400, r.get_data(as_text=True)) def test_superuser_cannot_patch_user_to_short_password(self) -> None: self._login("admin", "admin-secret") created = self.client.post( "/api/admin/users", json={"login": "patchshort", "password": "qa-secret-9", "is_superuser": False}, ) self.assertEqual(created.status_code, 201, created.get_data(as_text=True)) cid = created.get_json().get("id") r = self.client.patch(f"/api/admin/users/{cid}", json={"password": "x9y8z7"}) self.assertEqual(r.status_code, 400, r.get_data(as_text=True)) self.assertIn("8", r.get_json().get("message", "")) def test_superuser_cannot_create_user_with_letters_only_password(self) -> None: self._login("admin", "admin-secret") r = self.client.post( "/api/admin/users", json={ "login": "letters-only-user", "password": "helloworld", "is_superuser": False, }, ) self.assertEqual(r.status_code, 400, r.get_data(as_text=True)) self.assertIn("только из", r.get_json().get("message", "")) def test_superuser_cannot_patch_user_to_digits_only_password(self) -> None: self._login("admin", "admin-secret") created = self.client.post( "/api/admin/users", json={"login": "diguser", "password": "qa-secret-9", "is_superuser": False}, ) self.assertEqual(created.status_code, 201, created.get_data(as_text=True)) cid = created.get_json().get("id") r = self.client.patch(f"/api/admin/users/{cid}", json={"password": "87654321"}) self.assertEqual(r.status_code, 400, r.get_data(as_text=True)) self.assertIn("только из", r.get_json().get("message", "")) if __name__ == "__main__": unittest.main()