1034 lines
44 KiB
Python
1034 lines
44 KiB
Python
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()
|