diff --git a/api/ADMIN_NOTIFICATIONS.md b/api/ADMIN_NOTIFICATIONS.md index 50eba38..fbf1845 100644 --- a/api/ADMIN_NOTIFICATIONS.md +++ b/api/ADMIN_NOTIFICATIONS.md @@ -5,7 +5,8 @@ | Event | When | Where | |---|---|---| | New user | A user signs in for the first time (including a user an admin invited earlier) | Teams + email to admins | -| Equipment awaiting approval | A non-admin registers equipment, or submits changes that need review (one notification each time it enters review; resubmitting while it is still pending does not notify again) | Teams + email to admins | +| Equipment auto-approved | A non-admin registers equipment or saves changes while `EQUIPMENT_AUTO_APPROVE` is on (the default); the change goes live immediately and each one is reported | Teams + email to admins | +| Equipment awaiting approval | Only when `EQUIPMENT_AUTO_APPROVE=false`: a non-admin registers equipment, or submits changes that need review (one notification each time it enters review; resubmitting while it is still pending does not notify again) | Teams + email to admins | | Role request | A user asks for a different role from their Settings page | Teams + email to admins | | Role decision | An admin approves or rejects a role request | Email to the requester (only if email is configured; they always see it on their Settings page) | diff --git a/api/metadata_pg.py b/api/metadata_pg.py index c1bcf06..0d3be9a 100644 --- a/api/metadata_pg.py +++ b/api/metadata_pg.py @@ -1,3 +1,4 @@ +import fcntl import hashlib import json import logging @@ -1445,6 +1446,50 @@ def get_equipment_pg(domain_id: str) -> dict[str, Any] | None: conn.close() +def equipment_auto_approve_enabled() -> bool: + """Whether equipment registrations and edits go live without admin review. + + On by default; set ``EQUIPMENT_AUTO_APPROVE=false`` to restore the review + queue. Admins still get a notice for every auto-approved change. + """ + raw = os.getenv("EQUIPMENT_AUTO_APPROVE", "true").strip().lower() + return raw not in {"0", "false", "no", "off"} + + +def _enqueue_equipment_auto_approved_notification( + cur, + *, + domain_id: str, + equipment_name: str, + user: PlatformUser, + is_change: bool, +) -> None: + """Tell administrators an equipment change went live without review.""" + from platform_notifications import enqueue_notification + + action = "changes to" if is_change else "a new registration for" + enqueue_notification( + cur, + kind="equipment_auto_approved", + # The unique part goes first: keys are truncated to 255 characters. + dedupe_key=f"equipment_auto_approved:{uuid.uuid4().hex}:{domain_id}", + subject=f"Equipment auto-approved: {equipment_name}", + message=( + f"{user.name or user.id} submitted {action} {equipment_name}. " + "It was approved automatically and is live." + ), + facts=[ + ("Equipment", equipment_name), + ("Equipment ID", domain_id), + ("Submitted by", user.name or user.id), + ("Email", user.email), + ("Organization", user.organization), + ("Type", "Changes to existing equipment" if is_change else "New registration"), + ], + link_path=f"/equipment/{domain_id}", + ) + + def _enqueue_equipment_review_notification( cur, *, @@ -1532,7 +1577,8 @@ def register_equipment_pg( ), } - record_status = "approved" if user.role == "admin" else "pending" + auto_approved = user.role != "admin" and equipment_auto_approve_enabled() + record_status = "approved" if user.role == "admin" or auto_approved else "pending" conn = get_pg_connection() try: @@ -1727,7 +1773,8 @@ def register_equipment_pg( # Notify when a registration enters review. A resubmission while it is # still pending coalesces into the open review the admins were # already told about. - if record_status == "pending" and registration_row and not already_pending_review: + notify_review = record_status == "pending" and registration_row and not already_pending_review + if notify_review: with conn.cursor() as cur: _enqueue_equipment_review_notification( cur, @@ -1737,8 +1784,17 @@ def register_equipment_pg( user=user, is_change=False, ) + elif auto_approved: + with conn.cursor() as cur: + _enqueue_equipment_auto_approved_notification( + cur, + domain_id=domain_id, + equipment_name=str(payload.get("name") or domain_id), + user=user, + is_change=False, + ) conn.commit() - if record_status == "pending" and registration_row and not already_pending_review: + if notify_review or auto_approved: from platform_notifications import kick_notification_delivery kick_notification_delivery() @@ -1888,7 +1944,8 @@ def update_equipment_pg( "definition_changed": False, } - record_status = "pending" + auto_approved = equipment_auto_approve_enabled() + record_status = "approved" if auto_approved else "pending" features_count = len(features) cur.execute( """ @@ -1915,8 +1972,8 @@ def update_equipment_pg( primary_target_json = %s, secondary_target_json = %s, config_json = %s, - approved_at = NULL, - reviewed_by = NULL, + approved_at = CASE WHEN %s = 'approved' THEN CURRENT_TIMESTAMP ELSE NULL END, + reviewed_by = CASE WHEN %s = 'approved' THEN %s ELSE NULL END, updated_at = CURRENT_TIMESTAMP WHERE domain_id = %s """, @@ -1942,6 +1999,9 @@ def update_equipment_pg( Json(primary_target), Json(secondary_target), Json(config), + record_status, + record_status, + user.id, domain_id, ), ) @@ -1969,8 +2029,8 @@ def update_equipment_pg( source_type = %s, features_count = %s, status = %s, - reviewed_by = NULL, - reviewed_at = NULL + reviewed_by = CASE WHEN %s = 'approved' THEN %s ELSE NULL END, + reviewed_at = CASE WHEN %s = 'approved' THEN CURRENT_TIMESTAMP ELSE NULL END WHERE id = %s RETURNING id """, @@ -1982,6 +2042,9 @@ def update_equipment_pg( payload.get("source_type", "postgresql"), features_count, record_status, + record_status, + user.id, + record_status, existing_registration["id"], ), ) @@ -1997,9 +2060,15 @@ def update_equipment_pg( location, source_type, features_count, - status + status, + reviewed_by, + reviewed_at + ) + VALUES ( + %s, %s, %s, %s, %s, %s, %s, %s, + CASE WHEN %s = 'approved' THEN %s ELSE NULL END, + CASE WHEN %s = 'approved' THEN CURRENT_TIMESTAMP ELSE NULL END ) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( @@ -2011,14 +2080,18 @@ def update_equipment_pg( payload.get("source_type", "postgresql"), features_count, record_status, + record_status, + user.id, + record_status, ), ) registration_row = cur.fetchone() # Notify when a registration enters review. A resubmission while it is # still pending coalesces into the open review the admins were - # already told about. - if record_status == "pending" and registration_row and not already_pending_review: + # already told about. Auto-approved changes get a notice every time. + notify_review = record_status == "pending" and registration_row and not already_pending_review + if notify_review: with conn.cursor() as cur: _enqueue_equipment_review_notification( cur, @@ -2028,8 +2101,17 @@ def update_equipment_pg( user=user, is_change=True, ) + elif auto_approved: + with conn.cursor() as cur: + _enqueue_equipment_auto_approved_notification( + cur, + domain_id=domain_id, + equipment_name=str(payload.get("name") or domain_id), + user=user, + is_change=True, + ) conn.commit() - if record_status == "pending" and registration_row and not already_pending_review: + if notify_review or auto_approved: from platform_notifications import kick_notification_delivery kick_notification_delivery() @@ -2039,7 +2121,11 @@ def update_equipment_pg( "status": record_status, "config_json": config, "config_file": f"{domain_id}.json", - "message": f"Equipment '{payload.get('name', domain_id)}' changes submitted for admin review.", + "message": ( + f"Equipment '{payload.get('name', domain_id)}' changes saved and auto-approved." + if auto_approved + else f"Equipment '{payload.get('name', domain_id)}' changes submitted for admin review." + ), "changed": True, } finally: @@ -2071,6 +2157,26 @@ def delete_config_snapshot( return True +def sync_config_snapshot(*, configs_dir: Path, domain_id: str) -> tuple[str, bool]: + """Make the equipment's config snapshot match its latest committed row. + + Writes the snapshot when the equipment is approved and removes it + otherwise (including when the equipment no longer exists). All snapshot + changes go through here under a per-equipment file lock, and each one + re-reads the database, so concurrent saves, approvals and deletions can't + leave an older definition on disk. Returns (written path, removed). + """ + configs_dir.mkdir(parents=True, exist_ok=True) + with (configs_dir / f".{domain_id}.lock").open("w") as lock_file: + fcntl.flock(lock_file, fcntl.LOCK_EX) + current = get_equipment_pg(domain_id) + config = _coerce_json(current.get("config_json"), {}) if current else {} + if current and current.get("status") == "approved" and config: + path = write_config_snapshot(configs_dir=configs_dir, domain_id=domain_id, config_json=config) + return path, False + return "", delete_config_snapshot(configs_dir=configs_dir, domain_id=domain_id) + + def delete_equipment_pg( *, domain_id: str, diff --git a/api/routers/admin.py b/api/routers/admin.py index ea4a12a..b02395f 100644 --- a/api/routers/admin.py +++ b/api/routers/admin.py @@ -11,7 +11,7 @@ list_registration_reviews, update_execution_queue_status_pg, update_review_status_pg, - write_config_snapshot, + sync_config_snapshot, ) from security import ( PlatformUser, @@ -334,12 +334,8 @@ def update_review_status( status=payload.status, reviewer_id=reviewer.id, ) - if payload.status == "approved" and result["config_json"]: - write_config_snapshot( - configs_dir=CONFIGS_DIR, - domain_id=result["review"]["domain_id"], - config_json=result["config_json"], - ) + # Approvals publish the snapshot; rejections remove any stale one. + sync_config_snapshot(configs_dir=CONFIGS_DIR, domain_id=result["review"]["domain_id"]) return result["review"] except LookupError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc diff --git a/api/routers/equipment.py b/api/routers/equipment.py index 8067d8b..4c77122 100644 --- a/api/routers/equipment.py +++ b/api/routers/equipment.py @@ -22,7 +22,6 @@ approved_equipment_available_to_user, create_experiment_definition_pg, create_experiment_type_pg, - delete_config_snapshot, delete_equipment_pg, delete_experiment_type_pg, equipment_visible_to_user, @@ -42,7 +41,7 @@ update_experiment_type_pg, update_equipment_pg, update_recipe_proposal_status_pg, - write_config_snapshot, + sync_config_snapshot, ) from security import PlatformUser, get_platform_user @@ -362,13 +361,8 @@ def register_equipment( except PermissionError as exc: raise HTTPException(status_code=403, detail=str(exc)) from exc - config_path = "" - if result["status"] == "approved": - config_path = write_config_snapshot( - configs_dir=CONFIGS_DIR, - domain_id=result["domain_id"], - config_json=result["config_json"], - ) + config_path, _ = sync_config_snapshot(configs_dir=CONFIGS_DIR, domain_id=result["domain_id"]) + if config_path: logger.info("Wrote approved equipment config snapshot for %s", result["domain_id"]) return { @@ -390,9 +384,9 @@ def update_equipment( """ Edit an existing registered equipment record. - Any material change returns the equipment to pending review. This keeps the - visible form values editable without letting approved machine definitions - silently change underneath projects and optimization jobs. + A material change goes live immediately when auto-approval is on (the + default, see ``equipment_auto_approve_enabled``) and notifies admins; + otherwise it returns the equipment to pending review. """ try: result = update_equipment_pg( @@ -407,11 +401,17 @@ def update_equipment( raise HTTPException(status_code=404, detail="Equipment not found") snapshot_removed = False + config_path = "" if result.get("changed"): - snapshot_removed = delete_config_snapshot( + # Approved (auto-approved) definitions refresh the snapshot; pending + # ones remove it until an admin approves. + config_path, snapshot_removed = sync_config_snapshot( configs_dir=CONFIGS_DIR, domain_id=result["domain_id"], ) + if result.get("changed") and result.get("status") == "approved": + logger.info("Equipment %s edited by %s; auto-approved", result["domain_id"], user.id) + elif result.get("changed"): logger.info( "Equipment %s edited by %s; moved to pending review (snapshot_removed=%s)", result["domain_id"], @@ -423,7 +423,7 @@ def update_equipment( "status": "updated" if result.get("changed") else "unchanged", "domain_id": result["domain_id"], "config_file": result["config_file"], - "config_path": "", + "config_path": config_path, "review_status": result["status"], "changed": result.get("changed", False), "snapshot_removed": snapshot_removed, @@ -808,10 +808,7 @@ def delete_equipment( ), ) - snapshot_removed = delete_config_snapshot( - configs_dir=CONFIGS_DIR, - domain_id=domain_id, - ) + _, snapshot_removed = sync_config_snapshot(configs_dir=CONFIGS_DIR, domain_id=domain_id) logger.info( "Deleted equipment %s (owner=%s, by=%s, snapshot_removed=%s)", domain_id, result.get("display_name"), user.id, snapshot_removed, diff --git a/api/tests/test_admin_notifications_pg.py b/api/tests/test_admin_notifications_pg.py index 6ee4e12..d5b647b 100644 --- a/api/tests/test_admin_notifications_pg.py +++ b/api/tests/test_admin_notifications_pg.py @@ -190,6 +190,7 @@ def test_invited_user_first_sign_in_notifies_but_existing_user_does_not(self): # ── equipment awaiting review ─────────────────────────────────────── + @patch.dict(os.environ, {"EQUIPMENT_AUTO_APPROVE": "false"}) def test_equipment_registration_awaiting_review_notifies_admins(self): from metadata_pg import register_equipment_pg @@ -650,6 +651,7 @@ def test_merging_accounts_moves_requests_and_keeps_one_pending(self): rows = self._owner_sql("SELECT user_id, status FROM role_change_requests WHERE id = %s", (older["id"],)) self.assertEqual(rows, [(merged["id"], "pending")]) + @patch.dict(os.environ, {"EQUIPMENT_AUTO_APPROVE": "false"}) def test_rapid_equipment_resubmissions_notify_once(self): from metadata_pg import register_equipment_pg, update_equipment_pg @@ -666,6 +668,7 @@ def test_rapid_equipment_resubmissions_notify_once(self): rows = self._notifications("equipment_review_requested", f"equipment_review_requested:{created['registration_id']}:%") self.assertEqual(len(rows), 1) + @patch.dict(os.environ, {"EQUIPMENT_AUTO_APPROVE": "false"}) def test_resubmission_after_approval_starts_a_new_review_alert(self): from metadata_pg import register_equipment_pg, update_equipment_pg @@ -687,6 +690,79 @@ def test_resubmission_after_approval_starts_a_new_review_alert(self): rows = self._notifications("equipment_review_requested", f"equipment_review_requested:{created['registration_id']}:%") self.assertEqual(len(rows), 2) + @patch.dict(os.environ, {"EQUIPMENT_AUTO_APPROVE": "true"}) + def test_equipment_auto_approval_goes_live_and_notifies(self): + from metadata_pg import register_equipment_pg, update_equipment_pg + + name = f"Auto Etcher {self.suffix}" + payload = {"name": name, "columns": [{"name": "p", "type": "float"}], "primary_target": "rate"} + created = register_equipment_pg(payload=payload, user=self.owner) + self.assertEqual(created["status"], "approved") + domain_id = created["domain_id"] + self.assertEqual( + self._notifications("equipment_review_requested", f"equipment_review_requested:{created['registration_id']}:%"), + [], + ) + notices = self._notifications("equipment_auto_approved", f"equipment_auto_approved:%:{domain_id}") + self.assertEqual(len(notices), 1) + self.assertEqual(dict(notices[0][6])["Type"], "New registration") + + revised = update_equipment_pg( + domain_id=domain_id, + payload={**payload, "columns": [{"name": "p", "type": "float"}, {"name": "q", "type": "float"}]}, + user=self.owner, + ) + self.assertEqual(revised["status"], "approved") + self.assertIn("auto-approved", revised["message"]) + notices = self._notifications("equipment_auto_approved", f"equipment_auto_approved:%:{domain_id}") + self.assertEqual(len(notices), 2) + self.assertEqual(dict(notices[1][6])["Type"], "Changes to existing equipment") + + metadata = self._owner_sql( + "SELECT status, reviewed_by, approved_at IS NOT NULL FROM equipment_metadata WHERE domain_id = %s", + (domain_id,), + ) + self.assertEqual(metadata, [("approved", self.owner.id, True)]) + registration = self._owner_sql( + "SELECT status, reviewed_by, reviewed_at IS NOT NULL FROM equipment_registrations WHERE domain_id = %s", + (domain_id,), + ) + self.assertEqual(registration, [("approved", self.owner.id, True)]) + + @patch.dict(os.environ, {"EQUIPMENT_AUTO_APPROVE": "true"}) + def test_auto_approved_edit_refreshes_config_snapshot(self): + import tempfile + + from routers import equipment as equipment_router + + name = f"Snapshot Etcher {self.suffix}" + payload = { + "name": name, + "domain_id": f"snapshot_etcher_{self.suffix}", + "columns": [{"name": "p", "type": "float"}], + "primary_target": "rate", + } + with tempfile.TemporaryDirectory() as configs_dir, patch.object( + equipment_router, "CONFIGS_DIR", Path(configs_dir) + ): + client = self._client() + created = client.post("/api/equipment/register", json=payload, headers=self._headers(self.owner)) + self.assertEqual(created.status_code, 200, created.text) + domain_id = created.json()["domain_id"] + self.assertEqual(created.json()["review_status"], "approved") + + edited = client.patch( + f"/api/equipment/{domain_id}", + json={**payload, "columns": [{"name": "p", "type": "float"}, {"name": "z", "type": "float"}]}, + headers=self._headers(self.owner), + ) + self.assertEqual(edited.status_code, 200, edited.text) + self.assertEqual(edited.json()["review_status"], "approved") + self.assertFalse(edited.json()["snapshot_removed"]) + snapshot = Path(configs_dir) / f"{domain_id}.json" + self.assertTrue(snapshot.exists()) + self.assertIn("z", snapshot.read_text()) + def test_subjects_are_single_line(self): import data_loader_pg from platform_notifications import enqueue_notification diff --git a/api/tests/test_phase3_security.py b/api/tests/test_phase3_security.py index d6e30b5..3ba64b4 100644 --- a/api/tests/test_phase3_security.py +++ b/api/tests/test_phase3_security.py @@ -321,8 +321,9 @@ def test_equipment_registration_uses_authenticated_owner_context(self): }, ) as register_equipment_pg, patch.object( equipment, - "write_config_snapshot", - ) as write_config_snapshot: + "sync_config_snapshot", + return_value=("", False), + ) as sync_config_snapshot: response = self.client.post( "/api/equipment/register", headers=OWNER_HEADERS, @@ -380,7 +381,7 @@ def test_equipment_registration_uses_authenticated_owner_context(self): register_equipment_pg.call_args.kwargs["user"].organization, "Birck", ) - self.assertEqual(write_config_snapshot.call_count, 0) + sync_config_snapshot.assert_called_once() def test_trusted_maintainer_can_edit_equipment_with_researcher_role(self): connection = FakeConnection( diff --git a/web/app/equipment/register/page.tsx b/web/app/equipment/register/page.tsx index f5f27c3..01477b9 100644 --- a/web/app/equipment/register/page.tsx +++ b/web/app/equipment/register/page.tsx @@ -1512,8 +1512,8 @@ function EquipmentRegisterContent() { Review the equipment definition carefully {isEditMode - ? "Editing approved equipment sends the changed definition back to admin review before it can be used again." - : "This registration will persist equipment identity, status, inputs, and outputs in Postgres. Admin-submitted registrations are auto-approved; equipment-owner submissions go into admin review."} + ? "Saving applies the changed definition to projects and uploads right away, and administrators are notified. If admin review is turned on, the change waits for approval instead." + : "This registration will persist equipment identity, status, inputs, and outputs in Postgres. It goes live right away and administrators are notified; if admin review is turned on, it waits for approval instead."} @@ -1532,7 +1532,7 @@ function EquipmentRegisterContent() {

{isEditMode - ? "Update the submitted equipment definition. Changed approved records require review again." + ? "Update the submitted equipment definition. Changes go live when saved, and administrators are notified." : "Add a durable, Postgres-backed equipment definition with identity, status, inputs, and outputs."}