add support for collection/resource permissions in admin TUI (5b7f359b) · Commits · certic / Jama

src/jama/adminui/app.py

+233 −5

Original line number	Diff line number	Diff line
		@@ -25,6 +25,7 @@ from textual.widgets import (
		ListView,
		Select,
		Static,
		Tree,
		)

		from jama.adminui.i18n import DEFAULT_LANGUAGE, normalize_language, translate
		@@ -41,6 +42,7 @@ from resources.models import (
		FileType,
		Metadata,
		MetadataSet,
		ObjectPermission,
		Project,
		ProjectAccess,
		Role,
		@@ -317,6 +319,138 @@ class ProjectRoleForm(ModalScreen[dict \| None]):
		self.dismiss(None)


		class ResourceCollectionPermissionScreen(ModalScreen[dict \| None]):
		BINDINGS = [("escape", "cancel", "Cancel")]

		def __init__(self, role: Role, permission: ObjectPermission \| None = None):
		super().__init__()
		self.role = role
		self.permission = permission
		self.selected_item: access_services.ProjectTreeItem \| None = None
		if permission and permission.object_pk:
		self.selected_item = access_services.project_tree_item_for_permission(
		role.project, permission.object_class, permission.object_pk
		)

		def compose(self) -> ComposeResult:
		title = "Edit scoped permission" if self.permission else "Add scoped permission"
		with Vertical(id="dialog"):
		with VerticalScroll(classes="dialog-body"):
		yield Label(
		translate_for_widget(self, title),
		classes="dialog-title",
		)
		yield Label(
		self.selected_item_label(),
		id="selected-permission-target",
		)
		yield Tree(
		f"{self.role.project.pk}: {self.role.project.label}",
		id="project-tree",
		)
		yield Checkbox(
		translate_for_widget(self, "Create"),
		value=bool(self.permission and self.permission.object_create),
		id="field-create",
		)
		yield Checkbox(
		translate_for_widget(self, "Read"),
		value=bool(self.permission and self.permission.object_read),
		id="field-read",
		)
		yield Checkbox(
		translate_for_widget(self, "Update"),
		value=bool(self.permission and self.permission.object_update),
		id="field-update",
		)
		yield Checkbox(
		translate_for_widget(self, "Delete"),
		value=bool(self.permission and self.permission.object_delete),
		id="field-delete",
		)
		yield Label("", id="permission-error")
		with Horizontal(classes="buttons dialog-actions"):
		yield Button(
		translate_for_widget(self, "Save"), variant="primary", id="save"
		)
		yield Button(translate_for_widget(self, "Cancel"), id="cancel")

		def on_mount(self) -> None:
		tree = self.query_one("#project-tree", Tree)
		self.load_tree_node(tree.root)
		tree.root.expand()

		def selected_item_label(self) -> str:
		if not self.selected_item:
		return translate_for_widget(self, "Selected: none")
		return translate_for_widget(
		self,
		(
		"Selected: "
		f"{self.selected_item.object_class} "
		f"{self.selected_item.object_pk}: {self.selected_item.label}"
		),
		)

		@staticmethod
		def item_node_label(item: access_services.ProjectTreeItem) -> str:
		return f"{item.object_class} {item.object_pk}: {item.label}"

		def load_tree_node(self, node) -> None:
		node_data = node.data
		if node_data is None:
		if node.children:
		return
		children = access_services.project_tree_roots(self.role.project)
		else:
		if getattr(node_data, "loaded", False):
		return
		children = access_services.project_tree_children(
		self.role.project, node_data
		)
		node_data.loaded = True
		for item in children:
		node.add(
		self.item_node_label(item),
		data=item,
		allow_expand=item.has_children,
		)

		def on_tree_node_expanded(self, event: Tree.NodeExpanded) -> None:
		self.load_tree_node(event.node)

		def on_tree_node_selected(self, event: Tree.NodeSelected) -> None:
		if isinstance(event.node.data, access_services.ProjectTreeItem):
		self.selected_item = event.node.data
		self.query_one("#selected-permission-target", Label).update(
		self.selected_item_label()
		)

		def on_button_pressed(self, event: Button.Pressed) -> None:
		if event.button.id == "cancel":
		self.dismiss(None)
		return
		if event.button.id == "save":
		if not self.selected_item:
		self.query_one("#permission-error", Label).update(
		translate_for_widget(self, "Select a Resource or Collection.")
		)
		return
		self.dismiss(
		{
		"object_class": self.selected_item.object_class,
		"object_pk": self.selected_item.object_pk,
		"create": self.query_one("#field-create", Checkbox).value,
		"read": self.query_one("#field-read", Checkbox).value,
		"update": self.query_one("#field-update", Checkbox).value,
		"delete": self.query_one("#field-delete", Checkbox).value,
		}
		)

		def action_cancel(self) -> None:
		self.dismiss(None)


		class ConfirmScreen(ModalScreen[bool]):
		BINDINGS = [("escape", "cancel", "Cancel")]

		@@ -443,6 +577,11 @@ class JamaAdminApp(App):
		height: 24;
		margin-top: 1;
		}

		#project-tree {
		height: 18;
		margin: 1 0;
		}
		"""

		BINDINGS = [
		@@ -498,7 +637,9 @@ class JamaAdminApp(App):
		],
		"role_matrix": [
		FooterCommand("Space/Enter toggle cell", "toggle"),
		FooterCommand("A add scoped", "add"),
		FooterCommand("E edit row", "edit"),
		FooterCommand("D delete scoped", "delete"),
		FooterCommand("Esc back", "back"),
		FooterCommand("R refresh", "refresh"),
		FooterCommand("Q quit", "quit"),
		@@ -689,6 +830,7 @@ class JamaAdminApp(App):
		"users": self.add_user,
		"project_roles": self.add_role,
		"project_members": self.add_project_member,
		"role_matrix": self.add_scoped_role_permission,
		"user_accesses": self.add_user_access,
		"metadata_sets": self.add_metadata_set,
		"metadata": self.add_metadata,
		@@ -719,6 +861,7 @@ class JamaAdminApp(App):
		handlers: dict[str, Callable[[], None]] = {
		"users": self.delete_user,
		"project_roles": self.delete_role,
		"role_matrix": self.delete_scoped_role_permission,
		"project_members": self.remove_project_member,
		"user_accesses": self.remove_user_access,
		"metadata_sets": self.delete_metadata_set,
		@@ -1092,14 +1235,15 @@ class JamaAdminApp(App):
		self.set_title(
		f"{self.translate('Permissions')}: {role.project.label} / {role.label}"
		)
		self.reset_table(["object class", "create", "read", "update", "delete"])
		self.reset_table(["scope", "target", "create", "read", "update", "delete"])
		matrix = access_services.role_permission_matrix(role)
		for object_class in acl.PERM_CLASSES:
		permissions = matrix[object_class]
		self.add_row(
		object_class,
		f"class:{object_class}",
		object_class,
		[
		"class",
		object_class,
		permissions[acl.CRUD_CREATE],
		permissions[acl.CRUD_READ],
		@@ -1107,7 +1251,22 @@ class JamaAdminApp(App):
		permissions[acl.CRUD_DELETE],
		],
		)
		self.set_status("Space/Enter toggles selected CRUD cell. E edits whole row.")
		for permission in access_services.list_role_object_permissions(role):
		self.add_row(
		f"permission:{permission.pk}",
		permission,
		[
		permission.object_class,
		permission.serialization_label(),
		permission.object_create,
		permission.object_read,
		permission.object_update,
		permission.object_delete,
		],
		)
		self.set_status(
		"A adds Resource/Collection permission. Space/Enter toggles CRUD cell. E edits row. D deletes scoped row."
		)

		def selected_cell_key(self):
		coordinate = self.table().cursor_coordinate
		@@ -1127,7 +1286,24 @@ class JamaAdminApp(App):
		"delete": acl.CRUD_DELETE,
		}
		crud = crud_by_column.get(cell_key.column_key.value)
		if not isinstance(object_class, str) or crud is None:
		if crud is None:
		self.set_status("Select a create/read/update/delete cell to toggle.")
		return
		if isinstance(object_class, ObjectPermission):
		field_name = f"object_{crud}"
		new_value = not getattr(object_class, field_name)
		setattr(object_class, field_name, new_value)
		object_class.save(update_fields=[field_name])
		self.table().update_cell(
		cell_key.row_key,
		cell_key.column_key,
		Text("[x]" if new_value else "[ ]"),
		)
		self.set_status(
		f"Toggled {object_class.object_class}({object_class.object_pk}).{crud}."
		)
		return
		if not isinstance(object_class, str):
		self.set_status("Select a create/read/update/delete cell to toggle.")
		return
		role = get_object_or_404(
		@@ -1150,7 +1326,17 @@ class JamaAdminApp(App):
		self.set_status(f"Toggled {object_class}.{crud}.")

		def edit_role_matrix_row(self) -> None:
		object_class = self.selected_object()
		selected = self.selected_object()
		if isinstance(selected, ObjectPermission):
		role = get_object_or_404(
		Role.objects.select_related("project").only(
		"id", "label", "project_id", "project__id", "project__label"
		),
		pk=self.context["role_id"],
		)
		self.open_scoped_permission_form(role, selected)
		return
		object_class = selected
		if not isinstance(object_class, str):
		return
		role = get_object_or_404(
		@@ -1184,6 +1370,48 @@ class JamaAdminApp(App):
		role.project, role.label, access_services.matrix_to_permissions(matrix)
		)

		def add_scoped_role_permission(self) -> None:
		role = get_object_or_404(
		Role.objects.select_related("project").only(
		"id", "label", "project_id", "project__id", "project__label"
		),
		pk=self.context["role_id"],
		)
		self.open_scoped_permission_form(role)

		def open_scoped_permission_form(
		self, role: Role, permission: ObjectPermission \| None = None
		) -> None:
		def handle(values):
		if values is None:
		return
		try:
		access_services.set_object_permission_on_role(
		role,
		values["object_class"],
		int(values["object_pk"]),
		can_create=values["create"],
		can_read=values["read"],
		can_update=values["update"],
		can_delete=values["delete"],
		)
		self.refresh_table()
		self.set_status("Saved scoped permission.")
		except (ValueError, ValidationError, IntegrityError) as error:
		self.set_status(f"Error: {error}")

		self.push_screen(ResourceCollectionPermissionScreen(role, permission), handle)

		def delete_scoped_role_permission(self) -> None:
		permission = self.selected_object()
		if isinstance(permission, ObjectPermission):
		self.confirm(
		f'Delete scoped permission "{permission.serialization_label()}"?',
		lambda _: access_services.delete_object_permission(permission),
		)
		return
		self.set_status("Select a scoped Resource or Collection permission to delete.")

		def load_project_members(self) -> None:
		project = get_object_or_404(
		Project.objects.only("id", "label"), pk=self.context["project_id"]

src/jama/adminui/i18n.py

+9 −0

Original line number	Diff line number	Diff line
		@@ -39,7 +39,9 @@ FR_TRANSLATIONS = {
		"E rename role": "E renommer rôle",
		"D delete role": "D supprimer rôle",
		"Space/Enter toggle cell": "Espace/Entrée basculer case",
		"A add scoped": "A ajouter ciblée",
		"E edit row": "E modifier ligne",
		"D delete scoped": "D supprimer ciblée",
		"A add member": "A ajouter membre",
		"D remove selected role": "D retirer rôle sélectionné",
		"Enter metadata": "Entrée métadonnées",
		@@ -135,6 +137,10 @@ FR_TRANSLATIONS = {
		"Collection pipelines JSON": "Pipelines collections JSON",
		"Add role": "Ajouter un rôle",
		"Rename role": "Renommer le rôle",
		"Add scoped permission": "Ajouter une permission ciblée",
		"Edit scoped permission": "Modifier la permission ciblée",
		"Selected: none": "Sélection : aucune",
		"Select a Resource or Collection.": "Sélectionnez une ressource ou collection.",
		"Create": "Créer",
		"Read": "Lire",
		"Update": "Modifier",
		@@ -199,7 +205,10 @@ FR_TRANSLATIONS = {
		"Esc back. Ctrl+R roles, Ctrl+M metadata, Ctrl+U members.": "Esc retour. Ctrl+R rôles, Ctrl+M métadonnées, Ctrl+U membres.",
		"Enter edits permission matrix. A add, E rename, D delete.": "Entrée modifie la matrice de permissions. A ajouter, E renommer, D supprimer.",
		"Space/Enter toggles selected CRUD cell. E edits whole row.": "Espace/Entrée bascule la case CRUD sélectionnée. E modifie toute la ligne.",
		"A adds Resource/Collection permission. Space/Enter toggles CRUD cell. E edits row. D deletes scoped row.": "A ajoute une permission ressource/collection. Espace/Entrée bascule la case CRUD. E modifie la ligne. D supprime la ligne ciblée.",
		"Select a create/read/update/delete cell to toggle.": "Sélectionnez une case créer/lire/modifier/supprimer à basculer.",
		"Select a scoped Resource or Collection permission to delete.": "Sélectionnez une permission ciblée ressource ou collection à supprimer.",
		"Saved scoped permission.": "Permission ciblée enregistrée.",
		"A add user by ids. D removes selected role from user.": "A ajoute un utilisateur. D retire le rôle sélectionné de l'utilisateur.",
		"Enter opens metadata. A add set, E rename, D delete.": "Entrée ouvre les métadonnées. A ajouter jeu, E renommer, D supprimer.",
		"A add metadata. E edit. D delete.": "A ajouter métadonnée. E modifier. D supprimer.",

src/jama/tests.py

+62 −2

Original line number	Diff line number	Diff line
		@@ -596,6 +596,32 @@ class AdminUiAppTestCase(TransactionTestCase):
		with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}):
		asyncio.run(run_app())

		def test_scoped_permission_dialog_uses_project_tree(self):
		project = Project.objects.create(label="Tree dialog", description="")
		project.root_collection
		role = Role.objects.create(project=project, label="reader")

		async def run_app():
		from jama.adminui.app import JamaAdminApp
		from textual.widgets import Tree

		app = JamaAdminApp()
		async with app.run_test() as pilot:
		app.open_mode(
		"role_matrix",
		{"project_id": project.pk, "role_id": role.pk},
		push_history=False,
		)
		await pilot.pause()
		app.add_scoped_role_permission()
		await pilot.pause()

		tree = app.screen.query_one("#project-tree", Tree)
		self.assertEqual(len(tree.root.children), 1)

		with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}):
		asyncio.run(run_app())

		def test_clear_selected_cache_dialog_uses_french(self):
		async def run_app():
		from jama.adminui.app import JamaAdminApp
		@@ -749,10 +775,10 @@ class AdminUiAppTestCase(TransactionTestCase):
		push_history=False,
		)
		await pilot.pause()
		app.table().move_cursor(row=0, column=2)
		app.table().move_cursor(row=0, column=3)
		await pilot.press("space")
		await pilot.pause()
		cell = app.table().get_cell(acl.PERM_CLASS_RESOURCE, "read")
		cell = app.table().get_cell(f"class:{acl.PERM_CLASS_RESOURCE}", "read")
		self.assertEqual(getattr(cell, "plain", str(cell)), "[x]")

		with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}):
		@@ -763,6 +789,40 @@ class AdminUiAppTestCase(TransactionTestCase):
		)
		self.assertTrue(permission.object_read)

		def test_role_matrix_scoped_crud_cells_toggle_inline(self):
		project = Project.objects.create(label="Scoped permissions", description="")
		role = Role.objects.create(project=project, label="editor")
		root = project.root_collection
		permission = ObjectPermission.objects.create(
		role=role,
		object_class=acl.PERM_CLASS_COLLECTION,
		object_pk=root.pk,
		object_read=True,
		)

		async def run_app():
		from jama.adminui.app import JamaAdminApp

		app = JamaAdminApp()
		async with app.run_test() as pilot:
		app.open_mode(
		"role_matrix",
		{"project_id": project.pk, "role_id": role.pk},
		push_history=False,
		)
		await pilot.pause()
		app.table().move_cursor(row=6, column=4)
		await pilot.press("space")
		await pilot.pause()
		cell = app.table().get_cell(f"permission:{permission.pk}", "update")
		self.assertEqual(getattr(cell, "plain", str(cell)), "[x]")

		with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}):
		asyncio.run(run_app())

		permission.refresh_from_db()
		self.assertTrue(permission.object_update)


		class ApiKeyCsrfMiddlewareTestCase(TestCase):
		def setUp(self):

src/resources/admin_services/access.py

+138 −4

Original line number	Diff line number	Diff line
		from dataclasses import dataclass

		from django.contrib.auth.models import User
		from django.db import IntegrityError, transaction
		from django.db.models import Count

		from resources import acl
		from resources.models import ObjectPermission, Project, ProjectAccess, Role
		from resources.models import (
		Collection,
		CollectionMembership,
		ObjectPermission,
		Project,
		ProjectAccess,
		Resource,
		Role,
		)


		@dataclass
		class ProjectTreeItem:
		object_class: str
		object_pk: int
		label: str
		has_children: bool
		loaded: bool = False


		def list_roles(project: Project):
		@@ -28,13 +47,13 @@ def set_role_permissions(project: Project, role_label: str, permissions: list[di
		raise ValueError("Role label is empty")
		with transaction.atomic():
		role, _ = Role.objects.get_or_create(label=role_label, project=project)
		# The Vue-style matrix edits object-level permissions as one replaceable set.
		ObjectPermission.objects.filter(role=role).delete()
		# Matrix edits are class-wide. Keep Resource/Collection-scoped permissions.
		ObjectPermission.objects.filter(role=role, object_pk__isnull=True).delete()
		for permission in permissions:
		ObjectPermission.objects.create(
		role=role,
		object_class=str(permission.get("object_class")),
		object_pk=permission.get("object_pk"),
		object_pk=None,
		object_create=bool(permission.get("object_create", False)),
		object_read=bool(permission.get("object_read", False)),
		object_update=bool(permission.get("object_update", False)),
		@@ -103,6 +122,14 @@ def role_permission_matrix(role: Role) -> dict[str, dict[str, bool]]:
		return matrix


		def list_role_object_permissions(role: Role):
		return ObjectPermission.objects.filter(
		role=role,
		object_class__in=[acl.PERM_CLASS_COLLECTION, acl.PERM_CLASS_RESOURCE],
		object_pk__isnull=False,
		).order_by("object_class", "object_pk", "id")


		def matrix_to_permissions(matrix: dict[str, dict[str, bool]]) -> list[dict]:
		return [
		{
		@@ -115,3 +142,110 @@ def matrix_to_permissions(matrix: dict[str, dict[str, bool]]) -> list[dict]:
		}
		for object_class, permissions in matrix.items()
		]


		def _project_tree_item_for_collection(collection: Collection) -> ProjectTreeItem:
		has_children = (
		Collection.objects.filter(parent=collection, deleted_at__isnull=True).exists()
		or CollectionMembership.objects.filter(
		collection=collection, resource__deleted_at__isnull=True
		).exists()
		)
		return ProjectTreeItem(
		object_class=acl.PERM_CLASS_COLLECTION,
		object_pk=collection.pk,
		label=collection.title,
		has_children=has_children,
		)


		def _project_tree_item_for_resource(resource: Resource) -> ProjectTreeItem:
		return ProjectTreeItem(
		object_class=acl.PERM_CLASS_RESOURCE,
		object_pk=resource.pk,
		label=resource.title,
		has_children=False,
		)


		def project_tree_roots(project: Project) -> list[ProjectTreeItem]:
		return [_project_tree_item_for_collection(project.root_collection)]


		def project_tree_children(project: Project, parent_item: ProjectTreeItem):
		if parent_item.object_class != acl.PERM_CLASS_COLLECTION:
		return []
		collection = Collection.objects.filter(
		project=project, pk=parent_item.object_pk, deleted_at__isnull=True
		).first()
		if not collection:
		return []
		children = [
		_project_tree_item_for_collection(child)
		for child in Collection.objects.filter(
		project=project, parent=collection, deleted_at__isnull=True
		).order_by("title", "id")
		]
		children.extend(
		_project_tree_item_for_resource(membership.resource)
		for membership in CollectionMembership.objects.filter(
		collection=collection, resource__deleted_at__isnull=True
		)
		.select_related("resource")
		.order_by("rank", "resource__title", "resource__id")
		)
		return children


		def project_tree_item_for_permission(
		project: Project, object_class: str, object_pk: int
		) -> ProjectTreeItem:
		if object_class == acl.PERM_CLASS_COLLECTION:
		collection = Collection.objects.filter(
		project=project, pk=object_pk, deleted_at__isnull=True
		).first()
		if not collection:
		raise ValueError("Collection does not belong to project")
		return _project_tree_item_for_collection(collection)
		if object_class == acl.PERM_CLASS_RESOURCE:
		resource = Resource.objects.filter(
		ptr_project=project, pk=object_pk, deleted_at__isnull=True
		).first()
		if not resource:
		raise ValueError("Resource does not belong to project")
		return _project_tree_item_for_resource(resource)
		raise ValueError("Permission target must be a Resource or Collection")


		def set_object_permission_on_role(
		role: Role,
		object_class: str,
		object_pk: int,
		*,
		can_create: bool = False,
		can_read: bool = False,
		can_update: bool = False,
		can_delete: bool = False,
		) -> ObjectPermission:
		object_class = str(object_class)
		object_pk = int(object_pk)
		project_tree_item_for_permission(role.project, object_class, object_pk)
		with transaction.atomic():
		ObjectPermission.objects.filter(
		role=role, object_class=object_class, object_pk=object_pk
		).delete()
		return ObjectPermission.objects.create(
		role=role,
		object_class=object_class,
		object_pk=object_pk,
		object_create=can_create,
		object_read=can_read,
		object_update=can_update,
		object_delete=can_delete,
		)


		def delete_object_permission(permission: ObjectPermission) -> tuple[int, dict]:
		if permission.object_pk is None:
		raise ValueError("Only Resource or Collection permissions can be deleted here")
		return permission.delete()

src/resources/tests.py

+86 −0

File changed.

Preview size limit exceeded, changes collapsed.