diff --git a/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.js b/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.js
new file mode 100644
index 0000000..b126fae
--- /dev/null
+++ b/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.js
@@ -0,0 +1,8 @@
+// Copyright (c) 2025, Aakvatech and contributors
+// For license information, please see license.txt
+
+frappe.ui.form.on('Temporary Role Access Grant Log', {
+ // refresh: function(frm) {
+
+ // }
+});
diff --git a/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.json b/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.json
new file mode 100644
index 0000000..ff95c69
--- /dev/null
+++ b/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.json
@@ -0,0 +1,124 @@
+{
+ "actions": [],
+ "allow_rename": 1,
+ "autoname": "TRAL-.YYYY.-.#####",
+ "creation": "2025-01-01 00:00:00.000000",
+ "doctype": "DocType",
+ "document_type": "Document",
+ "editable_grid": 1,
+ "engine": "InnoDB",
+ "field_order": [
+ "user",
+ "role",
+ "request",
+ "column_break_4",
+ "granted_on",
+ "revoked_on",
+ "was_existing_role",
+ "section_break_8",
+ "remarks"
+ ],
+ "fields": [
+ {
+ "fieldname": "user",
+ "fieldtype": "Link",
+ "in_list_view": 1,
+ "in_standard_filter": 1,
+ "label": "User",
+ "options": "User",
+ "reqd": 1,
+ "search_index": 1
+ },
+ {
+ "fieldname": "role",
+ "fieldtype": "Link",
+ "in_list_view": 1,
+ "in_standard_filter": 1,
+ "label": "Role",
+ "options": "Role",
+ "reqd": 1,
+ "search_index": 1
+ },
+ {
+ "fieldname": "request",
+ "fieldtype": "Link",
+ "in_list_view": 1,
+ "in_standard_filter": 1,
+ "label": "Request",
+ "options": "Temporary Role Access Request",
+ "reqd": 1,
+ "search_index": 1
+ },
+ {
+ "fieldname": "column_break_4",
+ "fieldtype": "Column Break"
+ },
+ {
+ "fieldname": "granted_on",
+ "fieldtype": "Datetime",
+ "in_list_view": 1,
+ "label": "Granted On",
+ "reqd": 1
+ },
+ {
+ "fieldname": "revoked_on",
+ "fieldtype": "Datetime",
+ "label": "Revoked On"
+ },
+ {
+ "default": "0",
+ "fieldname": "was_existing_role",
+ "fieldtype": "Check",
+ "label": "Was Existing Role"
+ },
+ {
+ "fieldname": "section_break_8",
+ "fieldtype": "Section Break"
+ },
+ {
+ "fieldname": "remarks",
+ "fieldtype": "Small Text",
+ "label": "Remarks"
+ }
+ ],
+ "index_web_pages_for_search": 1,
+ "is_submittable": 0,
+ "links": [],
+ "modified": "2025-01-01 00:00:00.000000",
+ "modified_by": "Administrator",
+ "module": "Av Tools",
+ "name": "Temporary Role Access Grant Log",
+ "naming_rule": "Expression (old style)",
+ "owner": "Administrator",
+ "permissions": [
+ {
+ "create": 1,
+ "delete": 1,
+ "email": 1,
+ "export": 1,
+ "print": 1,
+ "read": 1,
+ "report": 1,
+ "role": "System Manager",
+ "share": 1,
+ "write": 1
+ },
+ {
+ "create": 1,
+ "email": 1,
+ "export": 1,
+ "print": 1,
+ "read": 1,
+ "report": 1,
+ "role": "Role Access Approver",
+ "share": 1,
+ "write": 1
+ }
+ ],
+ "sort_field": "modified",
+ "sort_order": "DESC",
+ "states": [],
+ "track_changes": 1,
+ "track_seen": 0,
+ "track_views": 0
+}
diff --git a/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.py b/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.py
new file mode 100644
index 0000000..e3c1cd6
--- /dev/null
+++ b/av_tools/av_tools/doctype/temporary_role_access_grant_log/temporary_role_access_grant_log.py
@@ -0,0 +1,14 @@
+# Copyright (c) 2025, Aakvatech and contributors
+# For license information, please see license.txt
+
+import frappe
+from frappe import _
+from frappe.model.document import Document
+from frappe.utils import now_datetime, get_datetime, getdate
+
+
+class TemporaryRoleAccessGrantLog(Document):
+ def validate(self):
+ """Validate the grant log entry."""
+ if self.revoked_on and self.granted_on and get_datetime(self.revoked_on) < get_datetime(self.granted_on):
+ frappe.throw(_("Revoked On cannot be earlier than Granted On"))
diff --git a/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.js b/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.js
new file mode 100644
index 0000000..68cc753
--- /dev/null
+++ b/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.js
@@ -0,0 +1,8 @@
+// Copyright (c) 2025, Aakvatech and contributors
+// For license information, please see license.txt
+
+frappe.ui.form.on('Temporary Role Access Request', {
+ // refresh: function(frm) {
+
+ // }
+});
diff --git a/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.json b/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.json
new file mode 100644
index 0000000..197eea3
--- /dev/null
+++ b/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.json
@@ -0,0 +1,227 @@
+{
+ "actions": [],
+ "allow_rename": 1,
+ "autoname": "TRA-.YYYY.-.#####",
+ "creation": "2025-01-01 00:00:00.000000",
+ "default_view": "List",
+ "doctype": "DocType",
+ "document_type": "Document",
+ "editable_grid": 1,
+ "engine": "InnoDB",
+ "field_order": [
+ "requested_for",
+ "requested_by",
+ "role_requested",
+ "column_break_4",
+ "from_datetime",
+ "to_datetime",
+ "duration_hours",
+ "section_break_8",
+ "reason",
+ "status",
+ "column_break_11",
+ "role_granted",
+ "granted_on",
+ "revoked_on",
+ "section_break_15",
+ "approver",
+ "approval_remarks",
+ "amended_from"
+ ],
+ "fields": [
+ {
+ "fieldname": "requested_for",
+ "fieldtype": "Link",
+ "in_list_view": 1,
+ "in_standard_filter": 1,
+ "label": "Requested For",
+ "options": "User",
+ "reqd": 1,
+ "search_index": 1
+ },
+ {
+ "fetch_from": "requested_for.full_name",
+ "fieldname": "requested_for_name",
+ "fieldtype": "Read Only",
+ "in_list_view": 1,
+ "label": "Requested For (Name)",
+ "print_hide": 1
+ },
+ {
+ "fieldname": "requested_by",
+ "fieldtype": "Link",
+ "in_list_view": 1,
+ "in_standard_filter": 1,
+ "label": "Requested By",
+ "options": "User",
+ "reqd": 1,
+ "search_index": 1
+ },
+ {
+ "fieldname": "role_requested",
+ "fieldtype": "Link",
+ "in_list_view": 1,
+ "in_standard_filter": 1,
+ "label": "Role Requested",
+ "options": "Role",
+ "reqd": 1,
+ "search_index": 1
+ },
+ {
+ "fieldname": "column_break_4",
+ "fieldtype": "Column Break"
+ },
+ {
+ "fieldname": "from_datetime",
+ "fieldtype": "Datetime",
+ "in_list_view": 1,
+ "label": "From Date/Time",
+ "reqd": 1
+ },
+ {
+ "fieldname": "to_datetime",
+ "fieldtype": "Datetime",
+ "in_list_view": 1,
+ "label": "To Date/Time",
+ "reqd": 1
+ },
+ {
+ "fieldname": "duration_hours",
+ "fieldtype": "Float",
+ "label": "Duration (Hours)",
+ "read_only": 1
+ },
+ {
+ "fieldname": "section_break_8",
+ "fieldtype": "Section Break"
+ },
+ {
+ "fieldname": "reason",
+ "fieldtype": "Small Text",
+ "label": "Reason",
+ "reqd": 1
+ },
+ {
+ "fieldname": "status",
+ "fieldtype": "Select",
+ "in_list_view": 1,
+ "in_standard_filter": 1,
+ "label": "Status",
+ "no_copy": 1,
+ "options": "Draft\nPending Approval\nApproved\nRejected\nActive\nExpired\nCancelled",
+ "read_only": 1
+ },
+ {
+ "fieldname": "column_break_11",
+ "fieldtype": "Column Break"
+ },
+ {
+ "fieldname": "role_granted",
+ "fieldtype": "Check",
+ "label": "Role Granted",
+ "no_copy": 1,
+ "read_only": 1
+ },
+ {
+ "fieldname": "granted_on",
+ "fieldtype": "Datetime",
+ "label": "Granted On",
+ "no_copy": 1,
+ "read_only": 1
+ },
+ {
+ "fieldname": "revoked_on",
+ "fieldtype": "Datetime",
+ "label": "Revoked On",
+ "no_copy": 1,
+ "read_only": 1
+ },
+ {
+ "fieldname": "section_break_15",
+ "fieldtype": "Section Break"
+ },
+ {
+ "fieldname": "approver",
+ "fieldtype": "Link",
+ "label": "Approver",
+ "no_copy": 1,
+ "options": "User",
+ "read_only": 1
+ },
+ {
+ "fieldname": "approval_remarks",
+ "fieldtype": "Small Text",
+ "label": "Approval Remarks",
+ "no_copy": 1
+ },
+ {
+ "fieldname": "amended_from",
+ "fieldtype": "Link",
+ "label": "Amended From",
+ "no_copy": 1,
+ "options": "Temporary Role Access Request",
+ "print_hide": 1,
+ "read_only": 1,
+ "search_index": 1
+ }
+ ],
+ "index_web_pages_for_search": 1,
+ "is_submittable": 1,
+ "allow_attach": 1,
+ "links": [
+ {
+ "link_doctype": "Temporary Role Access Grant Log",
+ "link_fieldname": "request"
+ }
+ ],
+ "modified": "2025-01-01 00:00:00.000000",
+ "modified_by": "Administrator",
+ "module": "Av Tools",
+ "name": "Temporary Role Access Request",
+ "naming_rule": "Expression (old style)",
+ "owner": "Administrator",
+ "permissions": [
+ {
+ "create": 1,
+ "delete": 1,
+ "email": 1,
+ "export": 1,
+ "print": 1,
+ "read": 1,
+ "report": 1,
+ "role": "System Manager",
+ "share": 1,
+ "submit": 1,
+ "write": 1
+ },
+ {
+ "create": 1,
+ "email": 1,
+ "export": 1,
+ "print": 1,
+ "read": 1,
+ "report": 1,
+ "role": "Role Access Approver",
+ "share": 1,
+ "submit": 1,
+ "write": 1
+ },
+ {
+ "create": 1,
+ "email": 1,
+ "export": 1,
+ "print": 1,
+ "read": 1,
+ "report": 1,
+ "role": "All",
+ "share": 1,
+ "write": 1
+ }
+ ],
+ "sort_field": "modified",
+ "sort_order": "DESC",
+ "states": [],
+ "track_changes": 1,
+ "track_seen": 1,
+ "track_views": 1
+}
diff --git a/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.py b/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.py
new file mode 100644
index 0000000..59931b8
--- /dev/null
+++ b/av_tools/av_tools/doctype/temporary_role_access_request/temporary_role_access_request.py
@@ -0,0 +1,577 @@
+# Copyright (c) 2025, Aakvatech and contributors
+# For license information, please see license.txt
+
+import frappe
+from frappe import _
+from frappe.model.document import Document
+from frappe.utils import now_datetime, get_datetime, flt
+
+
+class TemporaryRoleAccessRequest(Document):
+ def before_insert(self):
+ """Set defaults before insert."""
+ if not self.requested_by:
+ self.requested_by = frappe.session.user
+ self.status = "Draft"
+
+ def on_update(self):
+ """Detect status changes and send notifications."""
+ # Get the document state before this save
+ doc_before_save = self.get_doc_before_save()
+
+ if doc_before_save:
+ old_status = doc_before_save.get("status")
+ # Notify approvers when status changes to 'Pending Approval'
+ if old_status == "Draft" and self.status == "Pending Approval":
+ self._notify_approvers_of_new_request()
+
+ def validate(self):
+ """Main validation logic."""
+ self._validate_requested_for_user()
+ self._validate_role_requested()
+ self._validate_dates()
+ self._validate_no_overlapping_requests()
+ self._validate_user_does_not_already_have_role()
+ self._calculate_duration()
+
+ def before_submit(self):
+ """Run before submit (docstatus 0→1).
+ This is triggered by Workflow when transitioning to Approved state."""
+ # Set the approver
+ self.approver = frappe.session.user
+
+ def before_update_after_submit(self):
+ """Called when workflow updates after submit (e.g., from Approved to Active/Expired)."""
+ pass
+
+ def on_submit(self):
+ """Handle role granting when request is approved (docstatus=1).
+
+ If from_datetime has already passed, activate immediately.
+ Otherwise, the scheduler will activate it when from_datetime arrives.
+ """
+ now = now_datetime()
+ from_dt = get_datetime(self.from_datetime)
+
+ if from_dt <= now:
+ self._grant_role()
+ else:
+ self.db_set("status", "Approved")
+
+ def on_cancel(self):
+ """Handle cancellation - revoke role if currently granted.
+
+ Note: The workflow sets self.status to 'Cancelled' before calling
+ on_cancel, so we check role_granted (a separate field) instead.
+ Notification is sent inside _revoke_role().
+ """
+ if self.role_granted:
+ self._revoke_role(reason="cancelled")
+ self.db_set("status", "Cancelled")
+
+ def before_cancel(self):
+ """Validate before cancellation.
+
+ Note: At this point, the workflow has already updated self.status to
+ the target state's value, so we check the DB state instead."""
+ doc_before = self.get_doc_before_save()
+ if doc_before:
+ old_status = doc_before.get("status")
+ allowed_statuses = ("Pending Approval", "Approved", "Active")
+ if old_status not in allowed_statuses:
+ frappe.throw(
+ _("Only requests with status: {0} can be cancelled").format(
+ ", ".join(allowed_statuses)
+ )
+ )
+
+ # ---------- Validation Methods ----------
+
+ def _validate_requested_for_user(self):
+ """Ensure requested_for is an enabled user."""
+ user_enabled = frappe.db.get_value("User", self.requested_for, "enabled")
+ if user_enabled == 0:
+ frappe.throw(
+ _("User {0} is disabled. Please select an enabled user.").format(
+ frappe.bold(self.requested_for)
+ )
+ )
+
+ def _validate_role_requested(self):
+ """Ensure the role exists and is a valid role."""
+ if not frappe.db.exists("Role", self.role_requested):
+ frappe.throw(
+ _("Role {0} does not exist").format(frappe.bold(self.role_requested))
+ )
+
+ def _validate_dates(self):
+ """Validate date requirements."""
+ if not self.from_datetime or not self.to_datetime:
+ frappe.throw(_("From Date/Time and To Date/Time are required"))
+
+ from_dt = get_datetime(self.from_datetime)
+ to_dt = get_datetime(self.to_datetime)
+
+ if to_dt <= from_dt:
+ frappe.throw(_("To Date/Time must be greater than From Date/Time"))
+
+ if to_dt <= now_datetime() and not frappe.flags.in_test:
+ frappe.throw(_("To Date/Time must be in the future"))
+
+ def _validate_no_overlapping_requests(self):
+ """Prevent overlapping active/approved requests for same user and role."""
+ from_dt = get_datetime(self.from_datetime)
+ to_dt = get_datetime(self.to_datetime)
+
+ filters = [
+ ["requested_for", "=", self.requested_for],
+ ["role_requested", "=", self.role_requested],
+ ["status", "in", ["Pending Approval", "Approved", "Active"]],
+ ["docstatus", "<", 2],
+ ]
+
+ if self.name:
+ filters.append(["name", "!=", self.name])
+
+ overlapping = frappe.db.get_all(
+ "Temporary Role Access Request",
+ filters=filters,
+ fields=["name", "from_datetime", "to_datetime", "status"],
+ )
+
+ for req in overlapping:
+ req_from = get_datetime(req.from_datetime)
+ req_to = get_datetime(req.to_datetime)
+
+ # Check if date ranges overlap
+ if from_dt < req_to and to_dt > req_from:
+ frappe.throw(
+ _(
+ "An overlapping {0} request ({1}) already exists for User {2} and Role {3} "
+ "from {4} to {5}."
+ ).format(
+ req.status,
+ frappe.bold(req.name),
+ frappe.bold(self.requested_for),
+ frappe.bold(self.role_requested),
+ req.from_datetime,
+ req.to_datetime,
+ )
+ )
+
+ def _validate_user_does_not_already_have_role(self):
+ """Check if user already permanently has the requested role.
+ Skip this check if configured to allow re-granting (via AV Tools Settings).
+
+ Note: AV Tools Settings is a separate doctype that may not exist on all
+ sites. We gracefully handle the case where the table or column is missing."""
+ try:
+ allow_existing = frappe.db.get_single_value(
+ "AV Tools Settings", "allow_existing_role_temporary_access"
+ )
+ if allow_existing:
+ return
+ except Exception:
+ # AV Tools Settings table or column may not exist on this site
+ pass
+
+ has_role = frappe.db.get_value(
+ "Has Role",
+ {"parent": self.requested_for, "role": self.role_requested},
+ )
+
+ if has_role:
+ frappe.msgprint(
+ _(
+ "Warning: User {0} already has the role {1}. "
+ "The existing permanent role will not be removed on expiry."
+ ).format(
+ frappe.bold(self.requested_for),
+ frappe.bold(self.role_requested),
+ ),
+ alert=True,
+ indicator="orange",
+ )
+
+ def _calculate_duration(self):
+ """Auto-calculate duration in hours."""
+ if self.from_datetime and self.to_datetime:
+ from_dt = get_datetime(self.from_datetime)
+ to_dt = get_datetime(self.to_datetime)
+ delta = to_dt - from_dt
+ self.duration_hours = flt(delta.total_seconds() / 3600, 2)
+
+ # ---------- Role Management ----------
+
+ def _grant_role(self):
+ """Grant the requested role to the user and log the grant."""
+ now = now_datetime()
+
+ # Check if user is still enabled before granting
+ user_enabled = frappe.db.get_value("User", self.requested_for, "enabled")
+ if user_enabled == 0:
+ frappe.log_error(
+ message=f"Cannot grant role {self.role_requested} to disabled user {self.requested_for}",
+ title="Temporary Role Access - User Disabled",
+ )
+ self.db_set("status", "Cancelled")
+ return
+
+ # Check if user already has this role (existing permanent role)
+ has_role = frappe.db.get_value(
+ "Has Role",
+ {"parent": self.requested_for, "role": self.role_requested},
+ )
+ was_existing = bool(has_role)
+
+ if not has_role:
+ # Add the role to the user using Frappe's utility
+ user_doc = frappe.get_doc("User", self.requested_for)
+ user_doc.append("roles", {"role": self.role_requested})
+ user_doc.save(ignore_permissions=True)
+
+ # Create grant log entry
+ grant_log = frappe.get_doc(
+ {
+ "doctype": "Temporary Role Access Grant Log",
+ "user": self.requested_for,
+ "role": self.role_requested,
+ "request": self.name,
+ "granted_on": now,
+ "was_existing_role": 1 if was_existing else 0,
+ "remarks": f"Role granted via Temporary Role Access Request {self.name}",
+ }
+ )
+ grant_log.insert(ignore_permissions=True)
+
+ # Update the request document
+ self.db_set(
+ {
+ "role_granted": 1,
+ "granted_on": now,
+ "status": "Active",
+ }
+ )
+
+ # Send notification
+ self._send_notification(
+ recipients=[self.requested_for, self.requested_by],
+ subject=_("Temporary Role Access - Role {0} is now Active").format(self.role_requested),
+ message=_(
+ "The role {0} has been granted to {1} from {2} to {3}."
+ ).format(
+ frappe.bold(self.role_requested),
+ frappe.bold(self.requested_for),
+ self.from_datetime,
+ self.to_datetime,
+ ),
+ )
+
+ frappe.msgprint(
+ _("Role {0} has been granted to {1}").format(
+ frappe.bold(self.role_requested),
+ frappe.bold(self.requested_for),
+ ),
+ alert=True,
+ )
+
+ def _revoke_role(self, reason="expired"):
+ """Revoke the requested role from the user (only if it wasn't pre-existing).
+
+ Args:
+ reason: Either "expired" (to_datetime passed) or "cancelled" (manual cancellation).
+ """
+ now = now_datetime()
+
+ # Check the grant log for this request
+ grant_logs = frappe.db.get_all(
+ "Temporary Role Access Grant Log",
+ filters={"request": self.name, "role": self.role_requested, "user": self.requested_for},
+ fields=["name", "was_existing_role"],
+ order_by="creation desc",
+ limit=1,
+ )
+
+ was_existing = False
+ if grant_logs:
+ was_existing = grant_logs[0].was_existing_role
+
+ # Only remove if it wasn't a pre-existing role
+ if not was_existing:
+ user_doc = frappe.get_doc("User", self.requested_for)
+ roles_to_keep = [r for r in user_doc.roles if r.role != self.role_requested]
+
+ if len(roles_to_keep) != len(user_doc.roles):
+ user_doc.roles = roles_to_keep
+ user_doc.save(ignore_permissions=True)
+
+ # Update grant log with revocation
+ if grant_logs:
+ revoke_remarks = (
+ f"Role revoked at {now} via cancellation"
+ if reason == "cancelled"
+ else f"Role revoked at {now} via scheduler"
+ )
+ frappe.db.set_value(
+ "Temporary Role Access Grant Log",
+ grant_logs[0].name,
+ {
+ "revoked_on": now,
+ "remarks": revoke_remarks,
+ },
+ )
+
+ # Update the request document
+ self.db_set(
+ {
+ "role_granted": 0,
+ "revoked_on": now,
+ "status": "Expired",
+ }
+ )
+
+ # Send notification
+ if reason == "cancelled":
+ subject = _("Temporary Role Access - Role {0} Revoked (Cancelled)").format(self.role_requested)
+ message = _(
+ "The temporary access to role {0} for {1} has been revoked at {2} due to cancellation."
+ ).format(
+ frappe.bold(self.role_requested),
+ frappe.bold(self.requested_for),
+ now,
+ )
+ else:
+ subject = _("Temporary Role Access - Role {0} has Expired").format(self.role_requested)
+ message = _(
+ "The temporary access to role {0} for {1} has expired at {2}."
+ ).format(
+ frappe.bold(self.role_requested),
+ frappe.bold(self.requested_for),
+ now,
+ )
+
+ self._send_notification(
+ recipients=[self.requested_for, self.requested_by],
+ subject=subject,
+ message=message,
+ )
+
+ frappe.msgprint(
+ _("Role {0} has been revoked from {1}").format(
+ frappe.bold(self.role_requested),
+ frappe.bold(self.requested_for),
+ ),
+ alert=True,
+ )
+ def _notify_approvers_of_new_request(self):
+ """Notify approvers that a new request has been submitted."""
+ approvers = frappe.db.get_all(
+ "Has Role",
+ filters={"role": "Role Access Approver", "parenttype": "User"},
+ pluck="parent",
+ )
+ if not approvers:
+ return
+
+ self._send_notification(
+ recipients=approvers,
+ subject=_("New Temporary Role Access Request: {0}").format(self.name),
+ message=_(
+ "User {0} has requested temporary access to role {1} "
+ "from {2} to {3}.
Reason: {4}"
+ ).format(
+ frappe.bold(self.requested_by),
+ frappe.bold(self.role_requested),
+ self.from_datetime,
+ self.to_datetime,
+ self.reason,
+ ),
+ )
+
+ def _send_notification(self, recipients, subject, message):
+ """Send a notification to recipients."""
+ if isinstance(recipients, str):
+ recipients = [recipients]
+
+ # Only send if we have valid recipients
+ valid_recipients = [r for r in recipients if r and frappe.db.exists("User", r)]
+
+ if not valid_recipients:
+ return
+
+ try:
+ frappe.sendmail(
+ recipients=valid_recipients,
+ subject=subject,
+ message=message,
+ reference_doctype=self.doctype,
+ reference_name=self.name,
+ )
+ except Exception as e:
+ frappe.log_error(
+ message=f"Failed to send notification for {self.name}: {str(e)}",
+ title="Temporary Role Access - Notification Error",
+ )
+
+
+# ---------- Whitelisted Methods ----------
+
+@frappe.whitelist()
+def approve_request(name, approval_remarks=None):
+ """Approve a temporary role access request via Workflow action."""
+ doc = frappe.get_doc("Temporary Role Access Request", name)
+
+ if doc.status != "Pending Approval":
+ frappe.throw(_("Only requests in 'Pending Approval' status can be approved"))
+
+ if doc.requested_by == frappe.session.user:
+ frappe.throw(_("You cannot approve your own request"))
+
+ doc.approval_remarks = approval_remarks
+ doc.approver = frappe.session.user
+ doc.save(ignore_permissions=True)
+
+ # Use Workflow to approve - this triggers the workflow transition
+ # which sets docstatus=1 and calls on_submit
+ from frappe.model.workflow import apply_workflow
+
+ apply_workflow(doc, "Approve")
+
+ # Send notification
+ _send_status_notification(
+ doc,
+ _("Approved"),
+ _("Your request for role {0} has been approved by {1}.").format(
+ frappe.bold(doc.role_requested),
+ frappe.bold(frappe.session.user),
+ ),
+ )
+
+ frappe.msgprint(_("Request {0} has been approved").format(frappe.bold(name)), alert=True)
+
+
+@frappe.whitelist()
+def reject_request(name, approval_remarks=None):
+ """Reject a temporary role access request via Workflow action."""
+ doc = frappe.get_doc("Temporary Role Access Request", name)
+
+ if doc.status != "Pending Approval":
+ frappe.throw(_("Only requests in 'Pending Approval' status can be rejected"))
+
+ doc.approval_remarks = approval_remarks
+ doc.approver = frappe.session.user
+ doc.save(ignore_permissions=True)
+
+ # Use Workflow to reject
+ from frappe.model.workflow import apply_workflow
+
+ apply_workflow(doc, "Reject")
+
+ # Send notification
+ _send_status_notification(
+ doc,
+ _("Rejected"),
+ _("Your request for role {0} has been rejected by {1}.").format(
+ frappe.bold(doc.role_requested),
+ frappe.bold(frappe.session.user),
+ )
+ + (" Reason: " + approval_remarks if approval_remarks else ""),
+ )
+
+ frappe.msgprint(_("Request {0} has been rejected").format(frappe.bold(name)), alert=True)
+
+
+def _send_status_notification(doc, action_label, message):
+ """Send status change notification."""
+ recipients = [doc.requested_by]
+ if doc.requested_by != doc.requested_for:
+ recipients.append(doc.requested_for)
+
+ valid_recipients = [r for r in recipients if r and frappe.db.exists("User", r)]
+ if not valid_recipients:
+ return
+
+ try:
+ frappe.sendmail(
+ recipients=valid_recipients,
+ subject=_("Temporary Role Access Request {0} - {1}").format(doc.name, action_label),
+ message=message,
+ reference_doctype=doc.doctype,
+ reference_name=doc.name,
+ )
+ except Exception as e:
+ frappe.log_error(
+ message=f"Failed to send status notification for {doc.name}: {str(e)}",
+ title="Temporary Role Access - Notification Error",
+ )
+
+
+# ---------- Scheduler Method ----------
+
+def process_temporary_role_access():
+ """Scheduled job to process role activation and expiry.
+
+ Runs periodically (every 15 minutes by default) to:
+ 1. Activate approved requests where from_datetime <= now
+ 2. Expire active requests where to_datetime < now
+ """
+ now = now_datetime()
+
+ # 1. Activate approved future requests
+ approved_requests = frappe.db.get_all(
+ "Temporary Role Access Request",
+ filters={
+ "status": "Approved",
+ "docstatus": 1,
+ "from_datetime": ("<=", now),
+ "role_granted": 0,
+ },
+ pluck="name",
+ )
+
+ for name in approved_requests:
+ try:
+ doc = frappe.get_doc("Temporary Role Access Request", name)
+ # Double-check user is still enabled
+ user_enabled = frappe.db.get_value("User", doc.requested_for, "enabled")
+ if user_enabled == 0:
+ doc.db_set("status", "Cancelled")
+ frappe.log_error(
+ message=f"Cannot activate - user {doc.requested_for} is disabled",
+ title="Temporary Role Access - User Disabled",
+ )
+ frappe.db.commit()
+ continue
+
+ doc._grant_role()
+ frappe.db.commit()
+ except Exception as e:
+ frappe.db.rollback()
+ frappe.log_error(
+ message=f"Failed to activate role for request {name}: {str(e)}",
+ title="Temporary Role Access Activation Error",
+ )
+
+ # 2. Expire active requests that have passed their to_datetime
+ active_requests = frappe.db.get_all(
+ "Temporary Role Access Request",
+ filters={
+ "status": "Active",
+ "docstatus": 1,
+ "to_datetime": ("<", now),
+ "role_granted": 1,
+ },
+ pluck="name",
+ )
+
+ for name in active_requests:
+ try:
+ doc = frappe.get_doc("Temporary Role Access Request", name)
+ doc._revoke_role()
+ frappe.db.commit()
+ except Exception as e:
+ frappe.db.rollback()
+ frappe.log_error(
+ message=f"Failed to expire role for request {name}: {str(e)}",
+ title="Temporary Role Access Expiry Error",
+ )
diff --git a/av_tools/av_tools/doctype/temporary_role_access_request/test_temporary_role_access_request.py b/av_tools/av_tools/doctype/temporary_role_access_request/test_temporary_role_access_request.py
new file mode 100644
index 0000000..e55672b
--- /dev/null
+++ b/av_tools/av_tools/doctype/temporary_role_access_request/test_temporary_role_access_request.py
@@ -0,0 +1,657 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2025, Aakvatech and Contributors
+# See license.txt
+from __future__ import unicode_literals
+
+import frappe
+from frappe.tests.utils import FrappeTestCase
+from frappe.utils import now_datetime, add_to_date
+
+from av_tools.av_tools.doctype.temporary_role_access_request.temporary_role_access_request import (
+ approve_request,
+ reject_request,
+ process_temporary_role_access,
+)
+from av_tools.setup.temporary_role_access import (
+ _create_roles,
+ _create_workflow,
+)
+
+
+# Prevent test runner from auto-creating test records for User and Role
+# since they trigger deep ERPNext dependency chains (User → Employee → Company → Cost Center)
+# that fail on sites without properly configured accounting setup.
+# Our setUpClass handles all necessary fixtures manually.
+test_ignore = ["User", "Role"]
+
+
+class TestTemporaryRoleAccessRequest(FrappeTestCase):
+ """Test suite for Temporary Role Access Request feature."""
+
+ test_users = []
+ test_roles_created = []
+ _workflow_created = False
+
+ @classmethod
+ def setUpClass(cls):
+ """Set up test fixtures once before all tests."""
+ super().setUpClass()
+
+ # Create test users
+ for email, name in [
+ ("test_requestor@example.com", "Test Requestor"),
+ ("test_approver@example.com", "Test Approver"),
+ ("test_target@example.com", "Test Target User"),
+ ]:
+ if not frappe.db.exists("User", email):
+ user = frappe.get_doc({
+ "doctype": "User",
+ "email": email,
+ "first_name": name,
+ "send_welcome_email": 0,
+ "enabled": 1,
+ "roles": [{"role": "System Manager"}],
+ })
+ user.insert(ignore_permissions=True)
+ cls.test_users.append(email)
+
+ # Setup the role and workflow
+ _create_roles()
+ _create_workflow()
+
+ # Assign Role Access Approver role to the test approver
+ approver_user = frappe.get_doc("User", "test_approver@example.com")
+ has_role = any(r.role == "Role Access Approver" for r in approver_user.roles)
+ if not has_role:
+ approver_user.append("roles", {"role": "Role Access Approver"})
+ approver_user.save(ignore_permissions=True)
+
+ @classmethod
+ def tearDownClass(cls):
+ """Clean up all fixtures."""
+ # Delete test requests and grant logs
+ frappe.db.delete("Temporary Role Access Grant Log", {
+ "user": ("in", cls.test_users)
+ })
+ frappe.db.delete("Temporary Role Access Request", {
+ "requested_by": ("in", cls.test_users)
+ })
+
+ # Remove test roles from target user
+ target_user = frappe.get_doc("User", "test_target@example.com")
+ target_user.roles = [r for r in target_user.roles
+ if r.role not in ("Test Role 1", "Test Role 2")]
+ target_user.save(ignore_permissions=True)
+
+ # Delete test roles
+ for role_name in ("Test Role 1", "Test Role 2"):
+ if frappe.db.exists("Role", role_name):
+ frappe.db.delete("Has Role", {"role": role_name})
+ frappe.delete_doc("Role", role_name)
+
+ # Delete Workflow Action records (auto-created by workflow transitions) that reference test users
+ for email in cls.test_users:
+ if frappe.db.exists("User", email):
+ frappe.db.delete("Workflow Action", {"user": email})
+ frappe.db.delete("Workflow Action", {"completed_by": email})
+
+ # Delete test users - wrap in try/except in case other system links remain
+ for email in cls.test_users:
+ if frappe.db.exists("User", email):
+ frappe.db.delete("Has Role", {"parent": email})
+ try:
+ frappe.delete_doc("User", email)
+ except frappe.LinkExistsError:
+ # User may be linked to Workflow Actions or other system records
+ # Disable as fallback
+ frappe.db.set_value("User", email, "enabled", 0)
+ frappe.msgprint(f"Could not delete {email}, disabled instead")
+
+ # Clean up workflow
+ workflow_name = "Temporary Role Access Request Workflow"
+ if frappe.db.exists("Workflow", workflow_name):
+ frappe.delete_doc("Workflow", workflow_name)
+
+ # Clean up role - must delete dependent records first
+ # Also delete any Workflow Action records linked to this role
+ frappe.db.delete("Workflow Action", {"completed_by_role": "Role Access Approver"})
+ if frappe.db.exists("Role", "Role Access Approver"):
+ frappe.db.delete("Temporary Role Access Grant Log", {
+ "role": ("in", ["Role Access Approver", "Test Role 1", "Test Role 2"])
+ })
+ frappe.db.delete("Temporary Role Access Request", {
+ "role_requested": ("in", ["Role Access Approver", "Test Role 1", "Test Role 2"])
+ })
+ frappe.db.delete("Has Role", {"role": "Role Access Approver"})
+ try:
+ frappe.delete_doc("Role", "Role Access Approver")
+ except (frappe.LinkValidationError, frappe.LinkExistsError):
+ pass # Role might be linked to other system data, skip deletion
+
+ super().tearDownClass()
+
+ def setUp(self):
+ """Reset state before each test."""
+ frappe.set_user("Administrator")
+ frappe.flags.in_test = True
+ # Ensure test user is enabled (may have been left disabled by a failed test run)
+ frappe.db.set_value("User", "test_target@example.com", "enabled", 1)
+
+ def tearDown(self):
+ """Clean up after each test to ensure isolation."""
+ # Remove any test roles granted during the test from target user
+ target = frappe.get_doc("User", "test_target@example.com")
+ modified = False
+ for role_name in ("Test Role 1", "Test Role 2"):
+ roles_to_keep = [r for r in target.roles if r.role != role_name]
+ if len(roles_to_keep) != len(target.roles):
+ target.roles = roles_to_keep
+ modified = True
+ if modified:
+ target.save(ignore_permissions=True)
+
+ # Clean up grant logs and requests created during this test
+ frappe.db.delete("Temporary Role Access Grant Log", {
+ "user": ("in", self.test_users)
+ })
+ frappe.db.delete("Temporary Role Access Request", {
+ "requested_by": ("in", self.test_users)
+ })
+
+ # ========== Test Data Helpers ==========
+
+ def _make_request(self, **kwargs):
+ """Create a draft Temporary Role Access Request."""
+ defaults = {
+ "doctype": "Temporary Role Access Request",
+ "requested_for": "test_target@example.com",
+ "requested_by": "test_requestor@example.com",
+ "role_requested": "Test Role 1",
+ "from_datetime": add_to_date(now_datetime(), hours=1),
+ "to_datetime": add_to_date(now_datetime(), hours=5),
+ "reason": "Need temporary access for a task",
+ }
+ defaults.update(kwargs)
+
+ doc = frappe.get_doc(defaults)
+ doc.insert(ignore_permissions=True)
+ doc.reload()
+ return doc
+
+ def _submit_for_approval(self, doc):
+ """Transition from Draft to Pending Approval via workflow."""
+ from frappe.model.workflow import apply_workflow
+ frappe.set_user("test_requestor@example.com")
+ try:
+ apply_workflow(doc, "Submit for Approval")
+ finally:
+ frappe.set_user("Administrator")
+ doc.reload()
+
+ def _ensure_role_exists(self, role_name):
+ """Create a test role if it doesn't exist."""
+ if not frappe.db.exists("Role", role_name):
+ role = frappe.get_doc({
+ "doctype": "Role",
+ "role_name": role_name,
+ "desk_access": 1,
+ })
+ role.insert(ignore_permissions=True)
+
+ # ========== Validation Tests ==========
+
+ def test_disabled_user_cannot_be_requested_for(self):
+ """Request for disabled user should throw validation error."""
+ frappe.db.set_value("User", "test_target@example.com", "enabled", 0)
+ try:
+ with self.assertRaises(frappe.ValidationError):
+ self._make_request()
+ finally:
+ frappe.db.set_value("User", "test_target@example.com", "enabled", 1)
+
+ def test_non_existent_role_throws_error(self):
+ """Request with non-existent role should throw validation error."""
+ with self.assertRaises(frappe.ValidationError):
+ self._make_request(role_requested="NonExistentRole__XYZ")
+
+ def test_to_datetime_must_be_after_from_datetime(self):
+ """to_datetime must be greater than from_datetime."""
+ base_time = add_to_date(now_datetime(), hours=1)
+ with self.assertRaises(frappe.ValidationError):
+ self._make_request(
+ from_datetime=base_time,
+ to_datetime=add_to_date(base_time, hours=-2),
+ )
+
+ def test_to_datetime_must_be_in_future(self):
+ """to_datetime must be in the future."""
+ past_time = add_to_date(now_datetime(), hours=-1)
+ frappe.flags.in_test = False
+ try:
+ with self.assertRaises(frappe.ValidationError):
+ self._make_request(
+ from_datetime=add_to_date(past_time, hours=-2),
+ to_datetime=past_time,
+ )
+ finally:
+ frappe.flags.in_test = True
+
+ def test_no_overlapping_requests_same_user_role(self):
+ """Same user cannot have overlapping active/approved requests for same role."""
+ doc1 = self._make_request()
+ self._submit_for_approval(doc1)
+ approve_request(doc1.name)
+
+ with self.assertRaises(frappe.ValidationError):
+ self._make_request(
+ from_datetime=add_to_date(now_datetime(), hours=2),
+ to_datetime=add_to_date(now_datetime(), hours=4),
+ )
+
+ def test_non_overlapping_requests_allowed(self):
+ """Non-overlapping requests for same user+role should be allowed."""
+ self._ensure_role_exists("Test Role 1")
+ doc1 = self._make_request(
+ from_datetime=add_to_date(now_datetime(), hours=1),
+ to_datetime=add_to_date(now_datetime(), hours=3),
+ )
+ self._submit_for_approval(doc1)
+ approve_request(doc1.name)
+
+ # Create a non-overlapping request (after the first one ends)
+ doc2 = self._make_request(
+ from_datetime=add_to_date(now_datetime(), hours=10),
+ to_datetime=add_to_date(now_datetime(), hours=15),
+ )
+ self.assertEqual(doc2.status, "Draft")
+
+ def test_warning_when_user_already_has_role(self):
+ """Warning should be shown when user already has the requested role."""
+ self._ensure_role_exists("Test Role 1")
+ target = frappe.get_doc("User", "test_target@example.com")
+ target.append("roles", {"role": "Test Role 1"})
+ target.save(ignore_permissions=True)
+
+ doc = self._make_request()
+ self.assertEqual(doc.status, "Draft")
+
+ def test_duration_auto_calculated(self):
+ """Duration should be auto-calculated from dates."""
+ from_dt = add_to_date(now_datetime(), hours=1)
+ to_dt = add_to_date(from_dt, hours=3)
+ doc = self._make_request(from_datetime=from_dt, to_datetime=to_dt)
+ self.assertAlmostEqual(doc.duration_hours, 3.0, places=1)
+
+ # ========== Workflow Transition Tests ==========
+
+ def test_draft_to_pending_approval(self):
+ """Workflow should transition from Draft to Pending Approval."""
+ doc = self._make_request()
+
+ from frappe.model.workflow import apply_workflow
+ frappe.set_user("test_requestor@example.com")
+ apply_workflow(doc, "Submit for Approval")
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Pending Approval")
+ self.assertEqual(doc.docstatus, 0)
+
+ def test_approve_request_grants_role_immediately(self):
+ """Approving with from_datetime <= now should grant the role immediately."""
+ self._ensure_role_exists("Test Role 1")
+ doc = self._make_request(
+ from_datetime=add_to_date(now_datetime(), hours=-1),
+ to_datetime=add_to_date(now_datetime(), hours=3),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Active")
+ self.assertEqual(doc.docstatus, 1)
+ self.assertEqual(doc.role_granted, 1)
+ self.assertIsNotNone(doc.granted_on)
+
+ # Verify the role was assigned
+ has_role = frappe.db.get_value("Has Role", {
+ "parent": "test_target@example.com",
+ "role": "Test Role 1",
+ })
+ self.assertIsNotNone(has_role)
+
+ # Verify grant log was created
+ grant_logs = frappe.db.get_all("Temporary Role Access Grant Log", {
+ "request": doc.name,
+ "role": "Test Role 1",
+ "user": "test_target@example.com",
+ })
+ self.assertEqual(len(grant_logs), 1)
+
+ def test_approve_future_request_keeps_approved(self):
+ """Approving a future-dated request should keep it as Approved (not Active)."""
+ self._ensure_role_exists("Test Role 2")
+ doc = self._make_request(
+ role_requested="Test Role 2",
+ from_datetime=add_to_date(now_datetime(), hours=2),
+ to_datetime=add_to_date(now_datetime(), hours=5),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Approved")
+ self.assertEqual(doc.docstatus, 1)
+ self.assertEqual(doc.role_granted, 0)
+
+ def test_reject_request(self):
+ """Rejecting a request should set status to Rejected."""
+ doc = self._make_request()
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ reject_request(doc.name, "Not needed right now")
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Rejected")
+ self.assertEqual(doc.docstatus, 0)
+
+ def test_cancel_approved_request(self):
+ """Cancelling an approved request (no role granted yet) should work."""
+ doc = self._make_request(
+ from_datetime=add_to_date(now_datetime(), hours=2),
+ to_datetime=add_to_date(now_datetime(), hours=5),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Approved")
+
+ from frappe.model.workflow import apply_workflow
+ frappe.set_user("test_approver@example.com")
+ apply_workflow(doc, "Cancel")
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Cancelled")
+ self.assertEqual(doc.docstatus, 2)
+
+ def test_cancel_active_request_revokes_role(self):
+ """Cancelling an active request should revoke the role."""
+ self._ensure_role_exists("Test Role 1")
+ doc = self._make_request(
+ from_datetime=add_to_date(now_datetime(), hours=-1),
+ to_datetime=add_to_date(now_datetime(), hours=5),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Active")
+ self.assertEqual(doc.role_granted, 1)
+ self.assertIsNotNone(frappe.db.get_value("Has Role", {
+ "parent": "test_target@example.com",
+ "role": "Test Role 1",
+ }))
+
+ from frappe.model.workflow import apply_workflow
+ frappe.set_user("test_approver@example.com")
+ apply_workflow(doc, "Cancel")
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Cancelled")
+ self.assertEqual(doc.role_granted, 0)
+ self.assertIsNotNone(doc.revoked_on)
+ self.assertIsNone(frappe.db.get_value("Has Role", {
+ "parent": "test_target@example.com",
+ "role": "Test Role 1",
+ }))
+
+ def test_resubmit_from_rejected(self):
+ """Rejected request should be resubmittable back to Draft."""
+ doc = self._make_request()
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ reject_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Rejected")
+
+ from frappe.model.workflow import apply_workflow
+ frappe.set_user("test_requestor@example.com")
+ apply_workflow(doc, "Resubmit")
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Draft")
+ self.assertEqual(doc.docstatus, 0)
+
+ def test_requester_cannot_approve_own_request(self):
+ """User should not be able to approve their own request."""
+ doc = self._make_request()
+ self._submit_for_approval(doc)
+
+ # Switch to the requester and try to approve own request
+ frappe.set_user("test_requestor@example.com")
+ with self.assertRaises(frappe.ValidationError):
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ def test_cancel_rejected_request_blocked(self):
+ """Cancelling a rejected request should be blocked."""
+ doc = self._make_request()
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ reject_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Rejected")
+
+ # Only approved/active/pending documents should be cancellable
+ from frappe.model.workflow import apply_workflow
+ frappe.set_user("test_approver@example.com")
+ with self.assertRaises(frappe.ValidationError):
+ apply_workflow(doc, "Cancel")
+ frappe.set_user("Administrator")
+
+ # ========== Scheduler Tests ==========
+
+ def test_scheduler_activates_future_requests(self):
+ """Scheduler should activate approved requests whose from_datetime has passed."""
+ self._ensure_role_exists("Test Role 1")
+ doc = self._make_request(
+ role_requested="Test Role 1",
+ from_datetime=add_to_date(now_datetime(), hours=1),
+ to_datetime=add_to_date(now_datetime(), hours=5),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Approved")
+ self.assertEqual(doc.role_granted, 0)
+
+ # Simulate time passing by updating from_datetime
+ frappe.db.set_value("Temporary Role Access Request", doc.name,
+ "from_datetime", add_to_date(now_datetime(), hours=-1))
+ frappe.db.set_value("Temporary Role Access Request", doc.name,
+ "to_datetime", add_to_date(now_datetime(), hours=5))
+
+ process_temporary_role_access()
+
+ doc.reload()
+ self.assertEqual(doc.status, "Active")
+ self.assertEqual(doc.role_granted, 1)
+
+ def test_scheduler_expires_overdue_requests(self):
+ """Scheduler should expire active requests whose to_datetime has passed."""
+ self._ensure_role_exists("Test Role 1")
+ doc = self._make_request(
+ role_requested="Test Role 1",
+ from_datetime=add_to_date(now_datetime(), hours=-3),
+ to_datetime=add_to_date(now_datetime(), hours=-1),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Active")
+ self.assertEqual(doc.role_granted, 1)
+
+ process_temporary_role_access()
+
+ doc.reload()
+ self.assertEqual(doc.status, "Expired")
+ self.assertEqual(doc.role_granted, 0)
+
+ def test_scheduler_skips_disabled_user(self):
+ """Scheduler should not activate a request for a disabled user."""
+ self._ensure_role_exists("Test Role 1")
+ # Use future from_datetime so approval doesn't auto-activate
+ doc = self._make_request(
+ role_requested="Test Role 1",
+ from_datetime=add_to_date(now_datetime(), hours=2),
+ to_datetime=add_to_date(now_datetime(), hours=5),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Approved")
+
+ # Simulate time passing - make from_datetime in the past
+ frappe.db.set_value("Temporary Role Access Request", doc.name,
+ "from_datetime", add_to_date(now_datetime(), hours=-1))
+
+ # Disable the user
+ frappe.db.set_value("User", "test_target@example.com", "enabled", 0)
+ try:
+ process_temporary_role_access()
+ doc.reload()
+ self.assertEqual(doc.status, "Cancelled")
+ self.assertEqual(doc.role_granted, 0)
+ finally:
+ frappe.db.set_value("User", "test_target@example.com", "enabled", 1)
+
+ # ========== Existing Role Protection Tests ==========
+
+ def test_existing_role_not_removed_on_expiry(self):
+ """Permanent existing role should NOT be removed when temporary access expires."""
+ self._ensure_role_exists("Test Role 1")
+
+ # Give the user a permanent Test Role 1
+ target = frappe.get_doc("User", "test_target@example.com")
+ target.append("roles", {"role": "Test Role 1"})
+ target.save(ignore_permissions=True)
+
+ doc = self._make_request(
+ role_requested="Test Role 1",
+ from_datetime=add_to_date(now_datetime(), hours=-3),
+ to_datetime=add_to_date(now_datetime(), hours=-1),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Active")
+
+ # Grant log should record was_existing_role = 1
+ grant_logs = frappe.db.get_all("Temporary Role Access Grant Log",
+ filters={"request": doc.name},
+ fields=["was_existing_role"])
+ self.assertEqual(len(grant_logs), 1)
+ self.assertEqual(grant_logs[0].was_existing_role, 1)
+
+ process_temporary_role_access()
+
+ doc.reload()
+ self.assertEqual(doc.status, "Expired")
+
+ # Verify the permanent role was NOT removed
+ has_role = frappe.db.get_value("Has Role", {
+ "parent": "test_target@example.com",
+ "role": "Test Role 1",
+ })
+ self.assertIsNotNone(has_role, "Permanent role should not be removed on expiry")
+
+ # ========== Grant Log Tests ==========
+
+ def test_grant_log_created_on_grant(self):
+ """Grant log entry should be created when role is granted."""
+ self._ensure_role_exists("Test Role 1")
+ doc = self._make_request(
+ role_requested="Test Role 1",
+ from_datetime=add_to_date(now_datetime(), hours=-1),
+ to_datetime=add_to_date(now_datetime(), hours=3),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ grant_logs = frappe.db.get_all("Temporary Role Access Grant Log",
+ filters={"request": doc.name, "user": "test_target@example.com", "role": "Test Role 1"},
+ fields=["name", "granted_on", "was_existing_role"])
+ self.assertEqual(len(grant_logs), 1)
+ self.assertIsNotNone(grant_logs[0].granted_on)
+
+ def test_grant_log_updated_on_revoke(self):
+ """Grant log should be updated with revoked_on when role is revoked."""
+ self._ensure_role_exists("Test Role 1")
+ doc = self._make_request(
+ role_requested="Test Role 1",
+ from_datetime=add_to_date(now_datetime(), hours=-3),
+ to_datetime=add_to_date(now_datetime(), hours=-1),
+ )
+ self._submit_for_approval(doc)
+
+ frappe.set_user("test_approver@example.com")
+ approve_request(doc.name)
+ frappe.set_user("Administrator")
+
+ doc.reload()
+ self.assertEqual(doc.status, "Active")
+
+ process_temporary_role_access()
+
+ doc.reload()
+ self.assertEqual(doc.status, "Expired")
+
+ grant_logs = frappe.db.get_all("Temporary Role Access Grant Log",
+ filters={"request": doc.name},
+ fields=["revoked_on"])
+ self.assertEqual(len(grant_logs), 1)
+ self.assertIsNotNone(grant_logs[0].revoked_on,
+ "revoked_on should be set after expiry")
diff --git a/av_tools/hooks.py b/av_tools/hooks.py
index 768adc3..7e2f487 100644
--- a/av_tools/hooks.py
+++ b/av_tools/hooks.py
@@ -123,6 +123,7 @@
"av_tools.weigh_bridge.custom_fields.setup_custom_fields",
"av_tools.utils.create_custom_fields.execute",
"av_tools.utils.create_property_setter.execute",
+ "av_tools.setup.temporary_role_access.setup_temporary_role_access",
"av_tools.patches.v1_0.migrate_ai_integration_site_data.execute",
"av_tools.patches.v1_0.migrate_report_extension_site_data.execute",
]
@@ -241,7 +242,10 @@
"cron": {
"0 */6 * * *": [
"av_tools.av_tools.doctype.parking_bill.parking_bill.check_bills_all_vehicles",
- ]
+ ],
+ "*/15 * * * *": [
+ "av_tools.av_tools.doctype.temporary_role_access_request.temporary_role_access_request.process_temporary_role_access",
+ ],
},
"daily": [
"av_tools.av_tools.doctype.visibility.visibility.trigger_daily_alerts",
diff --git a/av_tools/modules.txt b/av_tools/modules.txt
index b7b2297..99abd8d 100644
--- a/av_tools/modules.txt
+++ b/av_tools/modules.txt
@@ -5,3 +5,4 @@ Feedback
AI Integration
Compliance
Trade In
+Temporary Role Access
diff --git a/av_tools/setup/__init__.py b/av_tools/setup/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/av_tools/setup/temporary_role_access.py b/av_tools/setup/temporary_role_access.py
new file mode 100644
index 0000000..0f4f95e
--- /dev/null
+++ b/av_tools/setup/temporary_role_access.py
@@ -0,0 +1,207 @@
+# Copyright (c) 2025, Aakvatech and contributors
+# For license information, please see license.txt
+
+import frappe
+from frappe import _
+
+
+def setup_temporary_role_access():
+ """Setup workflow, roles, and required data for Temporary Role Access feature."""
+ _create_roles()
+ _create_workflow_states()
+ _create_workflow_action_masters()
+ _create_workflow()
+
+
+def _create_roles():
+ """Create the Role Access Approver role if it doesn't exist."""
+ if not frappe.db.exists("Role", "Role Access Approver"):
+ role = frappe.get_doc(
+ {
+ "doctype": "Role",
+ "role_name": "Role Access Approver",
+ "desk_access": 1,
+ }
+ )
+ role.insert(ignore_permissions=True)
+ frappe.msgprint(_("Role 'Role Access Approver' created"))
+
+
+def _create_workflow_states():
+ """Create Workflow State records required by the workflow."""
+ states = [
+ "Draft",
+ "Pending Approval",
+ "Approved",
+ "Active",
+ "Expired",
+ "Rejected",
+ "Cancelled",
+ ]
+ for state_name in states:
+ if not frappe.db.exists("Workflow State", state_name):
+ state = frappe.get_doc(
+ {
+ "doctype": "Workflow State",
+ "workflow_state_name": state_name,
+ }
+ )
+ state.insert(ignore_permissions=True)
+
+
+def _create_workflow_action_masters():
+ """Create Workflow Action Master records required by the workflow."""
+ actions = [
+ "Submit for Approval",
+ "Approve",
+ "Reject",
+ "Cancel",
+ "Resubmit",
+ ]
+ for action_name in actions:
+ if not frappe.db.exists("Workflow Action Master", action_name):
+ action = frappe.get_doc(
+ {
+ "doctype": "Workflow Action Master",
+ "workflow_action_name": action_name,
+ }
+ )
+ action.insert(ignore_permissions=True)
+
+
+def _create_workflow():
+ """Create the workflow for Temporary Role Access Request.
+
+ Note: The 'allowed' field in Workflow Transition is a Link to Role,
+ so only ONE role per transition row is allowed. For transitions that
+ need multiple roles, create separate rows.
+ """
+ workflow_name = "Temporary Role Access Request Workflow"
+
+ if frappe.db.exists("Workflow", workflow_name):
+ return
+
+ workflow = frappe.get_doc(
+ {
+ "doctype": "Workflow",
+ "workflow_name": workflow_name,
+ "document_type": "Temporary Role Access Request",
+ "is_active": 1,
+ "override_status": 0,
+ "send_email_alert": 0,
+ "states": [
+ {
+ "state": "Draft",
+ "doc_status": 0,
+ "update_field": "status",
+ "update_value": "Draft",
+ "allow_edit": "All",
+ },
+ {
+ "state": "Pending Approval",
+ "doc_status": 0,
+ "update_field": "status",
+ "update_value": "Pending Approval",
+ "allow_edit": "All",
+ },
+ {
+ "state": "Approved",
+ "doc_status": 1,
+ "update_field": "status",
+ "update_value": "Approved",
+ "allow_edit": "Role Access Approver",
+ },
+ {
+ "state": "Active",
+ "doc_status": 1,
+ "update_field": "status",
+ "update_value": "Active",
+ "allow_edit": "System Manager",
+ },
+ {
+ "state": "Expired",
+ "doc_status": 1,
+ "update_field": "status",
+ "update_value": "Expired",
+ "allow_edit": "System Manager",
+ },
+ {
+ "state": "Rejected",
+ "doc_status": 0,
+ "update_field": "status",
+ "update_value": "Rejected",
+ "allow_edit": "Role Access Approver",
+ },
+ {
+ "state": "Cancelled",
+ "doc_status": 2,
+ "update_field": "status",
+ "update_value": "Cancelled",
+ "allow_edit": "System Manager",
+ },
+ ],
+ "transitions": [
+ {
+ "state": "Draft",
+ "action": "Submit for Approval",
+ "next_state": "Pending Approval",
+ "allowed": "All",
+ "allow_self_approval": 1,
+ "condition": "doc.requested_by == frappe.session.user",
+ },
+ {
+ "state": "Pending Approval",
+ "action": "Approve",
+ "next_state": "Approved",
+ "allowed": "Role Access Approver",
+ "allow_self_approval": 0,
+ },
+ {
+ "state": "Pending Approval",
+ "action": "Reject",
+ "next_state": "Rejected",
+ "allowed": "Role Access Approver",
+ "allow_self_approval": 0,
+ },
+ # Split multi-role Cancel into separate rows (one per role)
+ {
+ "state": "Approved",
+ "action": "Cancel",
+ "next_state": "Cancelled",
+ "allowed": "System Manager",
+ "allow_self_approval": 0,
+ },
+ {
+ "state": "Approved",
+ "action": "Cancel",
+ "next_state": "Cancelled",
+ "allowed": "Role Access Approver",
+ "allow_self_approval": 0,
+ },
+ {
+ "state": "Active",
+ "action": "Cancel",
+ "next_state": "Cancelled",
+ "allowed": "System Manager",
+ "allow_self_approval": 0,
+ },
+ {
+ "state": "Active",
+ "action": "Cancel",
+ "next_state": "Cancelled",
+ "allowed": "Role Access Approver",
+ "allow_self_approval": 0,
+ },
+ {
+ "state": "Rejected",
+ "action": "Resubmit",
+ "next_state": "Draft",
+ "allowed": "All",
+ "allow_self_approval": 1,
+ "condition": "doc.requested_by == frappe.session.user",
+ },
+ ],
+ }
+ )
+ workflow.insert(ignore_permissions=True)
+ frappe.msgprint(_("Workflow 'Temporary Role Access Request Workflow' created"))
diff --git a/av_tools/temporary_role_access/__init__.py b/av_tools/temporary_role_access/__init__.py
new file mode 100644
index 0000000..e69de29