205 lines
7.7 KiB
Python
205 lines
7.7 KiB
Python
"""Setup wizard: guard, API, completion."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from app import create_app, db
|
|
from app.models import WebUser
|
|
from app.services.setup_state import is_setup_completed, read_install_state
|
|
from config import ProductionConfig
|
|
from tests.kiosk_test_env import kiosk_client_kwargs
|
|
|
|
|
|
class SetupWizardTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.mkdtemp(prefix="wesp-setup-wizard-")
|
|
self._base = Path(self._tmp) / "wesp_home"
|
|
self._base.mkdir()
|
|
data = self._base / "data"
|
|
data.mkdir()
|
|
recipes = data / "recipes.db"
|
|
reports = data / "reports.db"
|
|
|
|
class SetupConfig(ProductionConfig):
|
|
BASE_DIR = str(self._base)
|
|
DATA_DIR = str(data)
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{recipes}"
|
|
SQLALCHEMY_BINDS = {"reports": f"sqlite:///{reports}"}
|
|
SECRET_KEY = "setup-wizard-test-secret"
|
|
TESTING = True
|
|
WESP_TESTING_BYPASS_SETUP_GUARD = False
|
|
SYNC_BACKGROUND_REQUEUE = False
|
|
SYNC_CLIENT_AUTOSTART = False
|
|
WESP_ADMIN_LOG_PATH = ""
|
|
WESP_ADMIN_HARDWARE_LOG_PATH = ""
|
|
WESP_ADMIN_UI_ACTIVITY_LOG_PATH = ""
|
|
WESP_LLM_AUTOSTART = False
|
|
|
|
self.app = create_app(SetupConfig, run_migrations=True)
|
|
self.client = self.app.test_client()
|
|
self._ctx = self.app.app_context()
|
|
self._ctx.push()
|
|
db.create_all()
|
|
|
|
def tearDown(self) -> None:
|
|
self._ctx.pop()
|
|
shutil.rmtree(self._tmp, ignore_errors=True)
|
|
|
|
def test_fresh_install_redirects_to_setup(self) -> None:
|
|
self.assertFalse(is_setup_completed(self.app))
|
|
resp = self.client.get("/")
|
|
self.assertEqual(resp.status_code, 302)
|
|
self.assertIn("/setup", resp.headers.get("Location", ""))
|
|
|
|
def test_setup_page_ok(self) -> None:
|
|
resp = self.client.get("/setup")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"setup-wizard.js", resp.data)
|
|
|
|
def test_setup_status_defaults(self) -> None:
|
|
resp = self.client.get("/api/setup/status")
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.get_json()
|
|
self.assertEqual(body.get("status"), "success")
|
|
self.assertFalse(body.get("setup", {}).get("setup_completed"))
|
|
|
|
def test_setup_api_accessible_from_lan_while_incomplete(self) -> None:
|
|
self.app.config["TESTING"] = False
|
|
try:
|
|
resp = self.client.get("/api/setup/status", **kiosk_client_kwargs())
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertFalse(resp.get_json().get("setup", {}).get("setup_completed"))
|
|
|
|
post = self.client.post(
|
|
"/api/setup/device-role",
|
|
json={"device_role": "server"},
|
|
**kiosk_client_kwargs(),
|
|
)
|
|
self.assertEqual(post.status_code, 200)
|
|
finally:
|
|
self.app.config["TESTING"] = True
|
|
|
|
def test_setup_status_from_lan_blocked_after_complete(self) -> None:
|
|
self.client.post("/api/setup/device-role", json={"device_role": "client"})
|
|
self.client.post(
|
|
"/api/setup/sync",
|
|
json={"server_url": "http://127.0.0.1", "client_name": "c1"},
|
|
)
|
|
self.client.post("/api/setup/complete")
|
|
self.assertTrue(is_setup_completed(self.app))
|
|
|
|
self.app.config["TESTING"] = False
|
|
try:
|
|
resp = self.client.get("/api/setup/status", **kiosk_client_kwargs())
|
|
self.assertEqual(resp.status_code, 403)
|
|
finally:
|
|
self.app.config["TESTING"] = True
|
|
|
|
def test_server_flow_complete(self) -> None:
|
|
r1 = self.client.post(
|
|
"/api/setup/device-role",
|
|
json={"device_role": "server"},
|
|
)
|
|
self.assertEqual(r1.status_code, 200)
|
|
|
|
r2 = self.client.post(
|
|
"/api/setup/network",
|
|
json={
|
|
"local_hostname": "komton_test_1",
|
|
},
|
|
)
|
|
self.assertEqual(r2.status_code, 200)
|
|
net = r2.get_json().get("network") or {}
|
|
self.assertEqual(net.get("local_hostname"), "komton_test_1.local")
|
|
self.assertTrue(str(net.get("public_base_url") or "").startswith("http://komton_test_1.local"))
|
|
|
|
r3 = self.client.post(
|
|
"/api/setup/users",
|
|
json={
|
|
"zootech": {
|
|
"login": "setup_zootech",
|
|
"password": "ZootechPass1",
|
|
"confirm": "ZootechPass1",
|
|
},
|
|
},
|
|
)
|
|
self.assertEqual(r3.status_code, 201 if r3.status_code != 200 else 200)
|
|
self.assertEqual(r3.get_json().get("status"), "success")
|
|
|
|
users = db.session.query(WebUser).all()
|
|
self.assertEqual(len(users), 2)
|
|
supers = [u for u in users if u.is_superuser]
|
|
self.assertEqual(len(supers), 1)
|
|
self.assertEqual(supers[0].login, "admin")
|
|
zootechs = [u for u in users if not u.is_superuser]
|
|
self.assertEqual(len(zootechs), 1)
|
|
self.assertEqual(zootechs[0].login, "setup_zootech")
|
|
|
|
r4 = self.client.post("/api/setup/complete")
|
|
self.assertEqual(r4.status_code, 200)
|
|
self.assertTrue(is_setup_completed(self.app))
|
|
|
|
r5 = self.client.get("/")
|
|
self.assertNotEqual(r5.status_code, 302)
|
|
|
|
def test_setup_mutation_blocked_after_complete(self) -> None:
|
|
self.client.post("/api/setup/device-role", json={"device_role": "client"})
|
|
self.client.post(
|
|
"/api/setup/sync",
|
|
json={"server_url": "http://127.0.0.1", "client_name": "c1"},
|
|
)
|
|
self.client.post("/api/setup/complete")
|
|
self.assertTrue(is_setup_completed(self.app))
|
|
|
|
resp = self.client.post("/api/setup/device-role", json={"device_role": "server"})
|
|
self.assertEqual(resp.status_code, 403)
|
|
|
|
def test_client_sync_progress(self) -> None:
|
|
self.client.post("/api/setup/device-role", json={"device_role": "client"})
|
|
resp = self.client.get("/api/setup/sync/progress")
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.get_json()
|
|
self.assertIn("first_bootstrap_done", body)
|
|
self.assertIn("initial_sync_progress", body)
|
|
|
|
def test_setup_users_update_existing_zootech(self) -> None:
|
|
self.client.post("/api/setup/device-role", json={"device_role": "server"})
|
|
payload = {
|
|
"zootech": {
|
|
"login": "setup_zootech",
|
|
"password": "ZootechPass1",
|
|
"confirm": "ZootechPass1",
|
|
},
|
|
}
|
|
r1 = self.client.post("/api/setup/users", json=payload)
|
|
self.assertEqual(r1.status_code, 200)
|
|
self.assertEqual(r1.get_json().get("action"), "created")
|
|
|
|
payload["zootech"]["password"] = "ZootechPass2"
|
|
payload["zootech"]["confirm"] = "ZootechPass2"
|
|
r2 = self.client.post("/api/setup/users", json=payload)
|
|
self.assertEqual(r2.status_code, 200)
|
|
self.assertEqual(r2.get_json().get("action"), "updated")
|
|
self.assertEqual(db.session.query(WebUser).filter_by(is_superuser=False).count(), 1)
|
|
|
|
payload["zootech"]["login"] = "setup_zootech_2"
|
|
payload["zootech"]["password"] = "ZootechPass3"
|
|
payload["zootech"]["confirm"] = "ZootechPass3"
|
|
r3 = self.client.post("/api/setup/users", json=payload)
|
|
self.assertEqual(r3.status_code, 200)
|
|
self.assertEqual(r3.get_json().get("action"), "created")
|
|
self.assertEqual(db.session.query(WebUser).filter_by(is_superuser=False).count(), 2)
|
|
|
|
def test_install_state_has_first_launch(self) -> None:
|
|
state = read_install_state(self.app)
|
|
self.assertIn("first_launch_at", state)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|