Add admin user email management
Magent CI/CD / verify (push) Successful in 10m59s
Magent CI/CD / deploy-prod (push) Skipped
Magent CI/CD / deploy-beta (push) Successful in 20s

This commit is contained in:
2026-09-01 22:09:20 +12:00
parent 0ac53b7f59
commit 06d944c9d9
3 changed files with 153 additions and 0 deletions
+29
View File
@@ -1358,6 +1358,35 @@ async def update_user_role(username: str, payload: Dict[str, Any]) -> Dict[str,
return {"status": "ok", "username": username, "role": role}
@router.post("/users/{username}/email")
async def update_user_email(username: str, payload: Dict[str, Any]) -> Dict[str, Any]:
user = get_user_by_username(username)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if not isinstance(payload, dict):
raise HTTPException(status_code=400, detail="Invalid payload")
email = _optional_recipient_email(payload.get("email"))
if email:
duplicate = next(
(
candidate
for candidate in get_all_users()
if str(candidate.get("username") or "").casefold() != username.casefold()
and str(candidate.get("email") or "").strip().casefold() == email.casefold()
),
None,
)
if duplicate:
raise HTTPException(status_code=409, detail="That email address is already assigned to another user")
if not set_user_email(username, email):
raise HTTPException(status_code=404, detail="User not found")
refreshed = get_user_by_username(username)
logger.info("Admin updated user contact email: username=%s email_set=%s", username, bool(email))
return {"status": "ok", "user": refreshed, "email": email}
@router.post("/users/{username}/auto-search")
async def update_user_auto_search(username: str, payload: Dict[str, Any]) -> Dict[str, Any]:
enabled = payload.get("enabled") if isinstance(payload, dict) else None
+33
View File
@@ -1320,6 +1320,39 @@ class DatabaseEmailTests(TempDatabaseMixin, unittest.TestCase):
self.assertEqual(stored.get("email"), "mixed@example.com")
class AdminUserEmailTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
async def test_admin_can_add_and_remove_user_email(self) -> None:
db.create_user_if_missing("Viewer", "password123", auth_provider="local")
saved = await admin_router.update_user_email("viewer", {"email": "viewer@example.com"})
self.assertEqual(saved["user"]["email"], "viewer@example.com")
cleared = await admin_router.update_user_email("VIEWER", {"email": None})
self.assertIsNone(cleared["user"]["email"])
async def test_admin_cannot_assign_duplicate_user_email(self) -> None:
db.create_user_if_missing(
"FirstViewer", "password123", email="shared@example.com", auth_provider="local"
)
db.create_user_if_missing("SecondViewer", "password123", auth_provider="local")
with self.assertRaises(HTTPException) as context:
await admin_router.update_user_email(
"SecondViewer", {"email": "SHARED@example.com"}
)
self.assertEqual(context.exception.status_code, 409)
self.assertIn("another user", str(context.exception.detail))
async def test_admin_user_email_requires_valid_address(self) -> None:
db.create_user_if_missing("Viewer", "password123", auth_provider="local")
with self.assertRaises(HTTPException) as context:
await admin_router.update_user_email("Viewer", {"email": "not-an-email"})
self.assertEqual(context.exception.status_code, 400)
class SnapshotHistoryTests(TempDatabaseMixin, unittest.TestCase):
def test_duplicate_snapshots_are_not_saved_and_download_evidence_is_retained(self) -> None:
snapshot = Snapshot(