mirror of
https://github.com/hansjone/netx.git
synced 2026-10-15 02:33:23 +08:00
Let users pick and order their workbench module tiles.
Per-user layout lives in the new app_user.prefs JSON column, read via /v1/auth/me and written via PUT /v1/auth/me/prefs/workbench. The workbench gets an in-place edit mode (gear -> drag to reorder, x to remove, tray to add back, reset/done). Reset stores empty prefs so registry changes still apply; ids of modules the user can't currently see are kept on save. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
313706595d
commit
2307f19779
16 changed files with 906 additions and 23 deletions
193
tests/test_user_prefs.py
Normal file
193
tests/test_user_prefs.py
Normal file
|
|
@ -0,0 +1,193 @@
|
|||
"""Per-user prefs API: workbench layout read/write, whitelist, validation, isolation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from netx_api.auth_middleware import AuthAuditMiddleware
|
||||
from netx_api.auth_router import router as auth_router
|
||||
from netx_api.auth_service import bootstrap_admin_if_needed, create_user
|
||||
from netx_api.db import Base, get_db
|
||||
from netx_api.models import AppUser
|
||||
|
||||
|
||||
class UserPrefsApiTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine(
|
||||
"sqlite+pysqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=self.engine)
|
||||
self.Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
|
||||
|
||||
self.app = FastAPI()
|
||||
self.app.add_middleware(AuthAuditMiddleware)
|
||||
self.app.include_router(auth_router)
|
||||
|
||||
def _override_db():
|
||||
db = self.Session()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
self.app.dependency_overrides[get_db] = _override_db
|
||||
|
||||
self._sess_patch = patch("netx_api.auth_middleware.SessionLocal", self.Session)
|
||||
self._sess_patch.start()
|
||||
self._settings_patches = [
|
||||
patch("netx_api.auth_middleware.settings.auth_enabled", True),
|
||||
patch("netx_api.audit_async.settings.audit_async", False),
|
||||
patch("netx_api.auth_tokens.settings.auth_secret", "unit-test-auth-secret-32bytes!!"),
|
||||
patch("netx_api.auth_tokens.settings.auth_token_ttl_sec", 3600),
|
||||
patch("netx_api.auth_service.settings.bootstrap_admin_username", "admin"),
|
||||
patch("netx_api.auth_service.settings.bootstrap_admin_password", "adminpass"),
|
||||
patch("netx_api.auth_deps.settings.auth_enabled", True),
|
||||
]
|
||||
for p in self._settings_patches:
|
||||
p.start()
|
||||
|
||||
db = self.Session()
|
||||
try:
|
||||
bootstrap_admin_if_needed(db)
|
||||
admin = db.query(AppUser).filter(AppUser.username == "admin").one()
|
||||
admin.must_change_password = False
|
||||
create_user(db, username="alice", password="alice123", role="user", actor=admin)
|
||||
create_user(db, username="bob", password="bob12345", role="user", actor=admin)
|
||||
for name in ("alice", "bob"):
|
||||
u = db.query(AppUser).filter(AppUser.username == name).one()
|
||||
u.must_change_password = False
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
self.client = TestClient(self.app)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self._sess_patch.stop()
|
||||
for p in self._settings_patches:
|
||||
p.stop()
|
||||
self.app.dependency_overrides.clear()
|
||||
self.engine.dispose()
|
||||
|
||||
def _login(self, username: str, password: str) -> str:
|
||||
r = self.client.post("/v1/auth/login", json={"username": username, "password": password})
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
return str(r.json()["access_token"])
|
||||
|
||||
def test_me_includes_empty_prefs(self) -> None:
|
||||
token = self._login("alice", "alice123")
|
||||
me = self.client.get("/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
|
||||
self.assertEqual(me.status_code, 200, me.text)
|
||||
self.assertEqual(me.json().get("prefs"), {})
|
||||
login = self.client.post("/v1/auth/login", json={"username": "alice", "password": "alice123"})
|
||||
self.assertEqual(login.json().get("prefs"), {})
|
||||
|
||||
def test_get_and_put_workbench(self) -> None:
|
||||
token = self._login("alice", "alice123")
|
||||
empty = self.client.get("/v1/auth/me/prefs", headers={"Authorization": f"Bearer {token}"})
|
||||
self.assertEqual(empty.status_code, 200, empty.text)
|
||||
self.assertEqual(empty.json(), {"prefs": {}})
|
||||
|
||||
body = {"order": ["ne", "topology"], "hidden": ["webcrt"]}
|
||||
put = self.client.put(
|
||||
"/v1/auth/me/prefs/workbench",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
json=body,
|
||||
)
|
||||
self.assertEqual(put.status_code, 200, put.text)
|
||||
self.assertEqual(put.json()["prefs"]["workbench"], body)
|
||||
|
||||
got = self.client.get("/v1/auth/me/prefs", headers={"Authorization": f"Bearer {token}"})
|
||||
self.assertEqual(got.json()["prefs"]["workbench"], body)
|
||||
me = self.client.get("/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
|
||||
self.assertEqual(me.json()["prefs"]["workbench"], body)
|
||||
|
||||
def test_put_strips_unknown_fields(self) -> None:
|
||||
token = self._login("alice", "alice123")
|
||||
put = self.client.put(
|
||||
"/v1/auth/me/prefs/workbench",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
json={"order": ["ne"], "hidden": [], "extra": True, "panels": {}},
|
||||
)
|
||||
self.assertEqual(put.status_code, 200, put.text)
|
||||
self.assertEqual(put.json()["prefs"]["workbench"], {"order": ["ne"], "hidden": []})
|
||||
|
||||
def test_unknown_key_404(self) -> None:
|
||||
token = self._login("alice", "alice123")
|
||||
r = self.client.put(
|
||||
"/v1/auth/me/prefs/panels",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
json={"order": [], "hidden": []},
|
||||
)
|
||||
self.assertEqual(r.status_code, 404)
|
||||
|
||||
def test_validation_422(self) -> None:
|
||||
token = self._login("alice", "alice123")
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
bad_char = self.client.put(
|
||||
"/v1/auth/me/prefs/workbench",
|
||||
headers=headers,
|
||||
json={"order": ["NE_Admin"], "hidden": []},
|
||||
)
|
||||
self.assertEqual(bad_char.status_code, 422)
|
||||
|
||||
too_long = self.client.put(
|
||||
"/v1/auth/me/prefs/workbench",
|
||||
headers=headers,
|
||||
json={"order": [f"m{i}" for i in range(65)], "hidden": []},
|
||||
)
|
||||
self.assertEqual(too_long.status_code, 422)
|
||||
|
||||
bad_type = self.client.put(
|
||||
"/v1/auth/me/prefs/workbench",
|
||||
headers=headers,
|
||||
json={"order": "ne", "hidden": []},
|
||||
)
|
||||
self.assertEqual(bad_type.status_code, 422)
|
||||
|
||||
def test_user_isolation(self) -> None:
|
||||
alice = self._login("alice", "alice123")
|
||||
bob = self._login("bob", "bob12345")
|
||||
layout = {"order": ["topology", "ne"], "hidden": ["ume"]}
|
||||
put = self.client.put(
|
||||
"/v1/auth/me/prefs/workbench",
|
||||
headers={"Authorization": f"Bearer {alice}"},
|
||||
json=layout,
|
||||
)
|
||||
self.assertEqual(put.status_code, 200, put.text)
|
||||
|
||||
bob_me = self.client.get("/v1/auth/me", headers={"Authorization": f"Bearer {bob}"})
|
||||
self.assertEqual(bob_me.status_code, 200, bob_me.text)
|
||||
self.assertEqual(bob_me.json().get("prefs"), {})
|
||||
|
||||
alice_me = self.client.get("/v1/auth/me", headers={"Authorization": f"Bearer {alice}"})
|
||||
self.assertEqual(alice_me.json()["prefs"]["workbench"], layout)
|
||||
|
||||
def test_auth_disabled_does_not_insert_system_user(self) -> None:
|
||||
with patch("netx_api.auth_deps.settings.auth_enabled", False), patch(
|
||||
"netx_api.auth_middleware.settings.auth_enabled", False
|
||||
):
|
||||
r = self.client.put(
|
||||
"/v1/auth/me/prefs/workbench",
|
||||
json={"order": ["ne"], "hidden": []},
|
||||
)
|
||||
self.assertEqual(r.status_code, 409, r.text)
|
||||
db = self.Session()
|
||||
try:
|
||||
self.assertIsNone(db.get(AppUser, "system"))
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue