diff --git a/src/openedx_authz/engine/schema/compilation.py b/src/openedx_authz/engine/schema/compilation.py new file mode 100644 index 00000000..84b8a65e --- /dev/null +++ b/src/openedx_authz/engine/schema/compilation.py @@ -0,0 +1,540 @@ +"""Resolve documents into one set of static definitions (the ``compile`` step). + +Compilation (ADR 0018 §1) merges base definitions across all documents and +applies ``role_extensions`` per ADR 0023: + + * Extensions resolve only after every role and permission is loaded. + * An extension changes only the fields it includes; absent fields keep + their current value; it cannot change a role ID. + * Different fields from different contributions combine. + * ``priority`` resolves conflicts on the same metadata field or the same + permission (higher wins). Equal priority with disagreeing values raises + :class:`SchemaCompileError` so deployment stops before the database + changes. + * Adding a permission the role already has, or removing one it lacks, is a + no-op logged as a warning. + +Every resulting :class:`CompiledDefinition` retains all contributing +:class:`SourceRecord` values, and each role-permission grant is attributed at +the (role, permission) grain with its origin (base vs extension) for ADR 0025 +source tracking. Output is deterministic regardless of discovery order. No +Casbin/Django imports. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, replace +from enum import Enum + +from openedx_authz.constants import SchemaOriginKind +from openedx_authz.engine.schema.exceptions import SchemaCompileError +from openedx_authz.engine.schema.types import ( + CompiledDefinition, + CompiledSchema, + RelationshipSource, + SchemaDocument, + SourceRecord, +) + +logger = logging.getLogger(__name__) + + +class RoleMetadataField(str, Enum): + """Metadata fields a ``RoleExtension`` may replace on a role (ADR 0023).""" + + DISPLAY_NAME = "display_name" + DESCRIPTION = "description" + ICON = "icon" + HIDDEN = "hidden" + + +class PermissionAction(str, Enum): + """How a ``RoleExtension`` changes a single permission on a role (ADR 0023).""" + + ADD = "add" + REMOVE = "remove" + + +class DefinitionKind(Enum): + """A family of base definitions compiled from schema documents. + + ``attr`` is the :class:`SchemaDocument` field holding this family's + definitions (populated by the loader from the YAML blocks). ``label`` is the + singular noun used for this family in operator-facing error and warning + messages. + """ + + CATEGORY = ("categories", "category") + PERMISSION = ("permissions", "permission") + ROLE = ("roles", "role") + + def __init__(self, attr: str, label: str) -> None: + self.attr = attr + self.label = label + + +@dataclass(frozen=True) +class _Tracked: + """A base definition plus the sources and priority that produced it. + + Immutable: resolution steps return a new ``_Tracked`` via + :func:`dataclasses.replace` rather than mutating one in place. + """ + + definition: object + sources: tuple[SourceRecord, ...] = () + priority: int = 0 + + +class SchemaCompiler: + """Merges validated documents into a :class:`CompiledSchema`.""" + + def compile(self, documents: list[SchemaDocument]) -> CompiledSchema: + """Resolve categories, permissions, roles, and extensions. + + Assumes ``documents`` already passed validation. + + Raises: + SchemaCompileError: On an unresolvable equal-priority conflict. + """ + categories = self._collect(documents, DefinitionKind.CATEGORY, key=lambda c: c.id) + permissions = self._collect(documents, DefinitionKind.PERMISSION, key=lambda p: p.identifier) + roles = self._collect(documents, DefinitionKind.ROLE, key=lambda r: r.id) + + resolved_roles, role_permission_sources = self._resolve_roles_and_provenance(roles, documents) + + return CompiledSchema( + categories=self._finalize(categories), + permissions=self._finalize(permissions), + roles=self._finalize(resolved_roles), + role_permission_sources=role_permission_sources, + ) + + def _collect(self, documents: list[SchemaDocument], kind: DefinitionKind, key) -> dict[str, _Tracked]: + """Gather base definitions keyed by identifier, resolving by priority. + + Higher priority wins on conflict; equal priority with differing content + raises; identical duplicates merge their sources. + """ + tracked: dict[str, _Tracked] = {} + for document in documents: + for definition in getattr(document, kind.attr): + identifier = key(definition) + existing = tracked.get(identifier) + if existing is None: + tracked[identifier] = _Tracked( + definition=definition, + sources=(document.source,), + priority=document.priority, + ) + continue + + tracked[identifier] = self._resolve_priority( + kind.label, + identifier, + existing=existing, + incoming=_Tracked( + definition=definition, + sources=(document.source,), + priority=document.priority, + ), + ) + return tracked + + def _resolve_priority( + self, + kind: str, + identifier: str, + *, + existing: _Tracked, + incoming: _Tracked, + ) -> _Tracked: + """Resolve two competing definitions for one identifier by priority. + + Single responsibility: decide which of two :class:`_Tracked` entries for + the same identifier survives. + + * Identical definitions merge their sources (both files contributed the + same thing). + * Otherwise higher priority wins; the loser is warned about (ADR 0017 + §4) so a valid-but-ineffective file does not look like it took effect. + * Equal priority with disagreeing definitions is unresolvable and raises + :class:`SchemaCompileError` so deployment stops before the database + changes. + + Each ``_Tracked`` carries a single source; ``existing`` and ``incoming`` + are symmetric, so this also serves conflicts between extension + contributions where only priority (not load order) decides the winner. + """ + if existing.definition == incoming.definition: + return replace(existing, sources=existing.sources + incoming.sources) + if incoming.priority == existing.priority: + raise SchemaCompileError( + f"Conflicting {kind} definition for {identifier!r} at equal priority " + f"{incoming.priority} " + f"({existing.sources[0].source_id} vs {incoming.sources[0].source_id})." + ) + winner, loser = (incoming, existing) if incoming.priority > existing.priority else (existing, incoming) + self._warn_discarded( + kind, + identifier, + loser=loser.sources[0], + loser_priority=loser.priority, + winner=winner.sources[0], + winner_priority=winner.priority, + ) + return winner + + @staticmethod + def _warn_discarded( + kind: str, + identifier: str, + *, + loser: SourceRecord, + loser_priority: int, + winner: SourceRecord, + winner_priority: int, + ) -> None: + """Report a contribution that lost to a higher-priority one. + + Priority silently picking a winner is the behavior operators find hardest + to debug: the losing file is valid, was loaded, and simply has no effect. + ADR 0017 §4 requires warning about exactly this. + """ + logger.warning( + "authz schema: %s %r from %s (priority %s) has no effect; %s (priority %s) takes precedence.", + kind, + identifier, + loser.source_id, + loser_priority, + winner.source_id, + winner_priority, + ) + + def _resolve_roles_and_provenance( + self, roles: dict[str, _Tracked], documents: list[SchemaDocument] + ) -> tuple[dict[str, _Tracked], dict[tuple[str, str], list[RelationshipSource]]]: + """Apply extensions and build per-(role, permission) provenance. + + Seeds base provenance from each role's own definition, then folds in + ``role_extensions`` (metadata replacement + permission add/remove), + honoring priority. Returns the resolved roles (a new mapping; inputs are + left untouched) alongside the relationship provenance map. + """ + metadata_changes, permission_changes = self._gather_extension_changes(roles, documents) + resolved_roles: dict[str, _Tracked] = {} + role_permission_sources: dict[tuple[str, str], list[RelationshipSource]] = {} + + for role_id, tracked in roles.items(): + base_sources = tracked.sources + base_priority = tracked.priority + + tracked = self._apply_metadata_changes(role_id, tracked, metadata_changes.get(role_id, {})) + + tracked, provenance = self._apply_permission_changes( + role_id, + tracked, + base_sources, + base_priority, + permission_changes.get(role_id), + ) + + resolved_roles[role_id] = tracked + for perm, sources in provenance.items(): + role_permission_sources[(role_id, perm)] = sources + + return resolved_roles, role_permission_sources + + def _apply_metadata_changes( + self, + role_id: str, + tracked: _Tracked, + metadata_changes_for_role: dict[str, list[tuple[object, int, SourceRecord]]], + ) -> _Tracked: + """Return the role with winning metadata applied and its sources recorded. + + Single responsibility: resolve the winning metadata values for this role + and produce a new :class:`_Tracked` carrying the updated definition and + the sources that contributed them. Returns ``tracked`` unchanged when the + role has no metadata extensions. + """ + if not metadata_changes_for_role: + return tracked + new_values, contributing_sources = self._resolve_metadata(role_id, metadata_changes_for_role) + merged_sources = tracked.sources + tuple(src for src in contributing_sources if src not in tracked.sources) + return replace( + tracked, + definition=replace(tracked.definition, **new_values), + sources=merged_sources, + ) + + @staticmethod + def _seed_base_provenance( + permissions: tuple[str, ...], + base_sources: tuple[SourceRecord, ...], + base_priority: int, + ) -> dict[str, list[RelationshipSource]]: + """Attribute each of a role's declared permissions to its base sources. + + Every permission the role declares in its own definition is a ``BASE`` + grant from each contributing source. Extension-driven add/remove layers + on top of this seed. + """ + return { + perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources] + for perm in permissions + } + + def _apply_permission_changes( + self, + role_id: str, + tracked: _Tracked, + base_sources: tuple[SourceRecord, ...], + base_priority: int, + permission_changes_for_role: dict[PermissionAction, list[tuple[str, int, SourceRecord]]] | None, + ) -> tuple[_Tracked, dict[str, list[RelationshipSource]]]: + """Return the role with permission changes applied and its provenance. + + Single responsibility: own the per-permission provenance for this role. + Seeds base provenance from the role's declared permissions, then, if the + role has permission extensions, resolves the final permission set, + produces a new :class:`_Tracked`, and reflects the add/remove in + provenance. Returns ``(tracked, base_provenance)`` unchanged when the + role has no permission extensions. + """ + base_provenance = self._seed_base_provenance(tracked.definition.permissions, base_sources, base_priority) + if not permission_changes_for_role or not ( + permission_changes_for_role[PermissionAction.ADD] or permission_changes_for_role[PermissionAction.REMOVE] + ): + return tracked, base_provenance + final_perms, provenance = self._resolve_permissions( + role_id, + tracked.definition.permissions, + base_sources, + base_priority, + permission_changes_for_role, + ) + tracked = replace(tracked, definition=replace(tracked.definition, permissions=final_perms)) + return tracked, provenance + + def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[SchemaDocument]): + """Collect per-role metadata and permission changes from all extensions. + + Thin orchestrator over the two independent concerns; see + :meth:`_gather_metadata_changes` and :meth:`_gather_permission_changes`. + """ + return ( + self._gather_metadata_changes(roles, documents), + self._gather_permission_changes(roles, documents), + ) + + @staticmethod + def _extensions_for_known_roles(roles: dict[str, _Tracked], documents: list[SchemaDocument]): + """Yield ``(document, extension)`` for extensions targeting a known role. + + Extensions naming an unknown role are skipped defensively; validation is + expected to have already errored on them. + """ + for document in documents: + for extension in document.role_extensions: + if extension.role_id in roles: + yield document, extension + + def _gather_metadata_changes( + self, roles: dict[str, _Tracked], documents: list[SchemaDocument] + ) -> dict[str, dict[str, list[tuple[object, int, SourceRecord]]]]: + """Collect per-role metadata field contributions from all extensions. + + Entries carry the full :class:`SourceRecord` and priority so provenance + and conflict resolution have everything they need. + """ + metadata_changes: dict[str, dict[str, list[tuple[object, int, SourceRecord]]]] = {} + for document, extension in self._extensions_for_known_roles(roles, documents): + metadata_changes_for_role = metadata_changes.setdefault(extension.role_id, {}) + for field in RoleMetadataField: + field_name = field.value + value = getattr(extension, field_name) + if value is not None: + metadata_changes_for_role.setdefault(field_name, []).append( + (value, document.priority, document.source) + ) + return metadata_changes + + def _gather_permission_changes( + self, roles: dict[str, _Tracked], documents: list[SchemaDocument] + ) -> dict[str, dict[PermissionAction, list[tuple[str, int, SourceRecord]]]]: + """Collect per-role permission add/remove contributions from all extensions. + + Entries carry the full :class:`SourceRecord` and priority so provenance + and conflict resolution have everything they need. + """ + permission_changes: dict[str, dict[PermissionAction, list[tuple[str, int, SourceRecord]]]] = {} + for document, extension in self._extensions_for_known_roles(roles, documents): + permission_changes_for_role = permission_changes.setdefault( + extension.role_id, {PermissionAction.ADD: [], PermissionAction.REMOVE: []} + ) + for perm in extension.add_permissions: + permission_changes_for_role[PermissionAction.ADD].append((perm, document.priority, document.source)) + for perm in extension.remove_permissions: + permission_changes_for_role[PermissionAction.REMOVE].append((perm, document.priority, document.source)) + return permission_changes + + def _resolve_contributions_based_on_priority( + self, + identifier: str, + entries: list[tuple[object, int, SourceRecord]], + *, + loser_kind, + on_tie, + ) -> tuple[object, int, list[SourceRecord]]: + """Pick the winning value among competing extension contributions. + + The batch counterpart to :meth:`_resolve_priority`: where that method + decides between two base definitions pairwise, this decides among any + number of ``(value, priority, source)`` contributions to the same + extension field or permission. + + The highest priority wins. Every contribution in the top-priority group + must agree; if they do not, ``on_tie(top_values)`` builds the message + for a :class:`SchemaCompileError` (the caller knows how to describe its + own conflict). Lower-priority contributions are warned about so a + valid-but-ineffective file is not mistaken for one that took effect + (ADR 0017 §4); ``loser_kind(value)`` labels each loser so the warning + can name what that contribution tried to do. + + Returns the winning value, the winning priority, and every source in the + winning group (for provenance). + """ + max_priority = max(priority for _, priority, _ in entries) + top_values = {value for value, priority, _ in entries if priority == max_priority} + if len(top_values) > 1: + raise SchemaCompileError(on_tie(top_values)) + + winner_value = next(iter(top_values)) + winning_sources = [src for value, priority, src in entries if priority == max_priority] + for value, priority, src in entries: + if priority < max_priority: + self._warn_discarded( + loser_kind(value), + identifier, + loser=src, + loser_priority=priority, + winner=winning_sources[0], + winner_priority=max_priority, + ) + return winner_value, max_priority, winning_sources + + def _resolve_metadata(self, role_id: str, metadata_changes: dict[str, list[tuple[object, int, SourceRecord]]]): + """Pick winning metadata values by priority; error on equal-priority ties.""" + new_values: dict[str, object] = {} + contributing: set[SourceRecord] = set() + for field_name, entries in metadata_changes.items(): + value, _, winning_sources = self._resolve_contributions_based_on_priority( + role_id, + entries, + loser_kind=lambda _value, _field=field_name: f"role_extension {_field}", + on_tie=lambda top, _field=field_name, _entries=entries: ( + f"Conflicting {_field!r} for role {role_id!r} at equal priority " + f"{max(p for _, p, _ in _entries)}: {sorted(map(str, top))}." + ), + ) + new_values[field_name] = value + contributing.update(winning_sources) + return new_values, contributing + + def _resolve_permissions( + self, + role_id: str, + base: tuple[str, ...], + base_sources: tuple[SourceRecord, ...], + base_priority: int, + permission_changes: dict[PermissionAction, list[tuple[str, int, SourceRecord]]], + ): + """Apply add/remove per permission, returning (final_perms, provenance). + + Add-vs-remove conflicts resolve by priority; equal priority raises. + Provenance keeps base attribution and appends extension attribution for + added permissions. + """ + current = set(base) + provenance = self._seed_base_provenance(base, base_sources, base_priority) + + # Regroup from action-keyed to permission-keyed so every contribution + # touching one permission (both adds and removes, across extensions) is + # resolved together: priority can pick a winner and an equal-priority + # add-vs-remove is caught as a conflict. + actions: dict[str, list[tuple[PermissionAction, int, SourceRecord]]] = {} + for perm, priority, src in permission_changes[PermissionAction.ADD]: + actions.setdefault(perm, []).append((PermissionAction.ADD, priority, src)) + for perm, priority, src in permission_changes[PermissionAction.REMOVE]: + actions.setdefault(perm, []).append((PermissionAction.REMOVE, priority, src)) + + for perm, entries in actions.items(): + action, max_priority, winning_sources = self._resolve_contributions_based_on_priority( + role_id, + entries, + loser_kind=lambda act, _perm=perm: f"role_extension {act.value} of {_perm!r} on role", + on_tie=lambda _top, _perm=perm, _entries=entries: ( + f"Conflicting add/remove for permission {_perm!r} on role {role_id!r} " + f"at equal priority {max(p for _, p, _ in _entries)}." + ), + ) + + match action: + case PermissionAction.ADD: + extension_sources = [ + RelationshipSource(src, SchemaOriginKind.EXTENSION, max_priority) for src in winning_sources + ] + self._apply_added_permission(role_id, perm, current, provenance, extension_sources) + case PermissionAction.REMOVE: + self._apply_removed_permission(role_id, perm, current, provenance) + + return tuple(sorted(current)), provenance + + @staticmethod + def _apply_added_permission( + role_id: str, + perm: str, + current: set[str], + provenance: dict[str, list[RelationshipSource]], + extension_sources: list[RelationshipSource], + ) -> None: + """Grant ``perm`` to the role, attributing it to ``extension_sources``. + + Mutates ``current`` and ``provenance`` in place. Adding a permission the + role already has is a no-op, logged as a warning (ADR 0023 §3). + """ + if perm in current: + logger.warning("role_extension adds %r already on role %r; no-op.", perm, role_id) + current.add(perm) + provenance.setdefault(perm, []) + provenance[perm].extend(extension_sources) + + @staticmethod + def _apply_removed_permission( + role_id: str, + perm: str, + current: set[str], + provenance: dict[str, list[RelationshipSource]], + ) -> None: + """Revoke ``perm`` from the role and drop its provenance. + + Mutates ``current`` and ``provenance`` in place. Removing a permission + the role does not have is a no-op, logged as a warning (ADR 0023 §3). + """ + if perm not in current: + logger.warning("role_extension removes %r not on role %r; no-op.", perm, role_id) + current.discard(perm) + provenance.pop(perm, None) + + def _finalize(self, tracked: dict[str, _Tracked]) -> dict[str, CompiledDefinition]: + """Turn tracked definitions into CompiledDefinition entries.""" + return { + identifier: CompiledDefinition( + key=identifier, + definition=entry.definition, + sources=entry.sources, + ) + for identifier, entry in tracked.items() + } diff --git a/src/openedx_authz/tests/schema/test_compilation.py b/src/openedx_authz/tests/schema/test_compilation.py new file mode 100644 index 00000000..d23ee835 --- /dev/null +++ b/src/openedx_authz/tests/schema/test_compilation.py @@ -0,0 +1,388 @@ +"""Unit tests for the schema compilation step (merge + extensions + priority). + +Grouped by concern: base compilation, extensions, priority resolution, +provenance, discarded-contribution warnings, no-op extension warnings, and the +defensive branches guarding states validation is expected to have rejected. +""" + +import logging +from dataclasses import fields + +import pytest + +from openedx_authz.constants import SchemaOriginKind +from openedx_authz.engine.schema.compilation import RoleMetadataField, SchemaCompiler +from openedx_authz.engine.schema.exceptions import SchemaCompileError +from openedx_authz.engine.schema.types import RoleExtension + +from .factories import category, extension, make_document, make_source, permission, role + +PERMS = [ + permission(name="view_course", cat="cat"), + permission(name="export_course", cat="cat"), + permission(name="manage_tags", cat="cat"), +] + + +def _base(**role_kwargs): + return make_document( + "base", + priority=100, + categories=[category("cat")], + permissions=PERMS, + roles=[role(rid="course_editor", permissions=("courses.view_course", "courses.manage_tags"), **role_kwargs)], + ) + + +class TestRoleMetadataFieldInvariant: + """``RoleMetadataField`` must stay in lockstep with ``RoleExtension``. + + ``_gather_extension_changes`` reads each member via ``getattr`` with no + default, so a member naming a field that ``RoleExtension`` does not declare + would raise at compile time. This guards that coupling at the enum level, + turning a rename drift into a fast, obvious test failure. + """ + + def test_every_member_is_a_role_extension_field(self): + """Each enum value names a real ``RoleExtension`` field.""" + extension_fields = {f.name for f in fields(RoleExtension)} + enum_values = {member.value for member in RoleMetadataField} + assert enum_values <= extension_fields + + def test_members_exclude_non_metadata_fields(self): + """Permission and identity fields are not metadata an extension replaces.""" + enum_values = {member.value for member in RoleMetadataField} + assert enum_values.isdisjoint({"role_id", "add_permissions", "remove_permissions"}) + + +class TestBaseCompilation: + """Compiling base definitions with no extensions applied.""" + + def test_base_definitions_compile(self): + """A single base document yields its roles and permissions verbatim.""" + schema = SchemaCompiler().compile([_base()]) + assert set(schema.roles) == {"course_editor"} + assert len(schema.permissions) == 3 + # Base definitions keep their declared order; rendering sorts later. + assert schema.roles["course_editor"].definition.permissions == ( + "courses.view_course", + "courses.manage_tags", + ) + + +class TestExtensions: + """Applying ``role_extensions`` on top of a base role definition.""" + + def test_extension_adds_and_removes_permissions_and_metadata(self): + """One extension can add, remove, rename, and hide in a single pass.""" + ext = make_document( + "ext", + priority=200, + role_extensions=[ + extension( + "course_editor", + add_permissions=("courses.export_course",), + remove_permissions=("courses.manage_tags",), + display_name="Author", + hidden=True, + ) + ], + ) + definition = SchemaCompiler().compile([_base(), ext]).roles["course_editor"].definition + assert "courses.export_course" in definition.permissions + assert "courses.manage_tags" not in definition.permissions + assert definition.display_name == "Author" + assert definition.hidden is True + + def test_extension_sources_are_retained(self): + """A role touched by an extension keeps both the base and extension sources.""" + ext = make_document("ext", priority=200, role_extensions=[extension("course_editor", display_name="X")]) + compiled = SchemaCompiler().compile([_base(), ext]) + assert len(compiled.roles["course_editor"].sources) == 2 + + +class TestPriorityResolution: + """How priority resolves conflicting contributions (higher wins; ties fail).""" + + def test_equal_priority_metadata_conflict_raises(self): + """Two extensions setting the same field at equal priority is an error.""" + a = make_document("a", priority=200, role_extensions=[extension("course_editor", display_name="A")]) + b = make_document("b", priority=200, role_extensions=[extension("course_editor", display_name="B")]) + with pytest.raises(SchemaCompileError): + SchemaCompiler().compile([_base(), a, b]) + + def test_higher_priority_metadata_wins(self): + """The higher-priority extension's metadata value takes effect.""" + lo = make_document("lo", priority=150, role_extensions=[extension("course_editor", display_name="Lo")]) + hi = make_document("hi", priority=300, role_extensions=[extension("course_editor", display_name="Hi")]) + definition = SchemaCompiler().compile([_base(), lo, hi]).roles["course_editor"].definition + assert definition.display_name == "Hi" + + def test_equal_priority_add_remove_conflict_raises(self): + """An add and a remove of the same permission at equal priority is an error.""" + add = make_document( + "add", + priority=200, + role_extensions=[extension("course_editor", add_permissions=("courses.export_course",))], + ) + rem = make_document( + "rem", + priority=200, + role_extensions=[extension("course_editor", remove_permissions=("courses.export_course",))], + ) + with pytest.raises(SchemaCompileError): + SchemaCompiler().compile([_base(), add, rem]) + + def test_conflicting_base_definition_equal_priority_raises(self): + """Two base definitions of the same role at equal priority is an error.""" + a = make_document("a", priority=100, roles=[role(rid="dup", display_name="A", permissions=())]) + b = make_document("b", priority=100, roles=[role(rid="dup", display_name="B", permissions=())]) + with pytest.raises(SchemaCompileError): + SchemaCompiler().compile([a, b]) + + def test_higher_priority_base_definition_wins(self): + """The higher-priority base definition replaces the lower one.""" + lo = make_document("lo", priority=100, roles=[role(rid="dup", display_name="Lo", permissions=())]) + hi = make_document("hi", priority=200, roles=[role(rid="dup", display_name="Hi", permissions=())]) + compiled = SchemaCompiler().compile([lo, hi]) + assert compiled.roles["dup"].definition.display_name == "Hi" + + +class TestProvenance: + """Each role-permission grant records where it came from (ADR 0025).""" + + def test_base_permissions_get_base_provenance(self): + """Permissions from the role's own definition are tagged ``BASE``.""" + schema = SchemaCompiler().compile([_base()]) + for perm in ("courses.view_course", "courses.manage_tags"): + prov = schema.role_permission_sources[("course_editor", perm)] + assert [(rs.source.distribution, rs.origin_kind) for rs in prov] == [("test-dist", SchemaOriginKind.BASE)] + + def test_extension_grant_is_attributed_to_the_module_not_core(self): + """An extension-added grant is tagged ``EXTENSION``, base grants stay ``BASE``.""" + ext = make_document( + "modx", + priority=200, + role_extensions=[extension("course_editor", add_permissions=("courses.export_course",))], + ) + schema = SchemaCompiler().compile([_base(), ext]) + + core = schema.role_permission_sources[("course_editor", "courses.view_course")] + added = schema.role_permission_sources[("course_editor", "courses.export_course")] + + # Both permissions coexist on the role, but their origins remain distinct. + assert [rs.origin_kind for rs in core] == [SchemaOriginKind.BASE] + assert [rs.origin_kind for rs in added] == [SchemaOriginKind.EXTENSION] + + def test_removed_permission_has_no_provenance(self): + """A permission removed by an extension leaves no provenance entry.""" + ext = make_document( + "modx", + priority=200, + role_extensions=[extension("course_editor", remove_permissions=("courses.manage_tags",))], + ) + schema = SchemaCompiler().compile([_base(), ext]) + assert ("course_editor", "courses.manage_tags") not in schema.role_permission_sources + + +class TestDiscardedContributionWarnings: + """Priority silently picks a winner; the loser must be reported. + + ADR 0017 §4 requires warning about contributions that do not take effect + because another file has a higher priority. A losing file is valid and was + loaded, so without a warning it looks like it applied. + """ + + @staticmethod + def _compile(*documents): + return SchemaCompiler().compile(list(documents)) + + def test_lower_priority_base_definition_warns(self, caplog): + """A base definition that loses on priority is reported as having no effect.""" + low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")]) + high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")]) + + with caplog.at_level(logging.WARNING): + compiled = self._compile(high, low) + + assert compiled.roles["course_editor"].definition.display_name == "Author" + assert "has no effect" in caplog.text + assert make_source("low").source_id in caplog.text + assert make_source("high").source_id in caplog.text + + def test_warning_names_both_priorities(self, caplog): + """The discard warning names both the losing and winning priorities.""" + low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")]) + high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")]) + + with caplog.at_level(logging.WARNING): + self._compile(low, high) + + assert "priority 100" in caplog.text + assert "priority 200" in caplog.text + + def test_warns_regardless_of_document_order(self, caplog): + """Discovery order must not decide whether the operator is told.""" + low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")]) + high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")]) + + with caplog.at_level(logging.WARNING): + self._compile(low, high) + ascending = caplog.text + caplog.clear() + with caplog.at_level(logging.WARNING): + self._compile(high, low) + + assert "has no effect" in ascending + assert "has no effect" in caplog.text + + def test_identical_duplicate_does_not_warn(self): + """An identical definition merges sources; nothing is discarded.""" + first = make_document("first", priority=100, roles=[role(rid="course_editor")]) + second = make_document("second", priority=200, roles=[role(rid="course_editor")]) + + compiled = self._compile(first, second) + + assert len(compiled.roles["course_editor"].sources) == 2 + + def test_uses_singular_kind_label(self, caplog): + """Messages say 'category', not the truncated attribute name.""" + low = make_document("low", priority=100, categories=[category("cat", display_name="Low")]) + high = make_document("high", priority=200, categories=[category("cat", display_name="High")]) + + with caplog.at_level(logging.WARNING): + self._compile(low, high) + + assert "category 'cat'" in caplog.text + assert "categorie" not in caplog.text + + def test_conflict_error_uses_singular_kind_label(self): + """A conflict error uses the singular kind label ('category').""" + left = make_document("left", priority=100, categories=[category("cat", display_name="Left")]) + right = make_document("right", priority=100, categories=[category("cat", display_name="Right")]) + + with pytest.raises(SchemaCompileError, match="Conflicting category definition"): + self._compile(left, right) + + def test_losing_metadata_extension_warns(self, caplog): + """A metadata extension that loses on priority is reported, naming its source.""" + base = make_document("base", priority=100, roles=[role(rid="course_editor")]) + low = make_document("low", priority=100, role_extensions=[extension("course_editor", display_name="Low")]) + high = make_document("high", priority=200, role_extensions=[extension("course_editor", display_name="High")]) + + with caplog.at_level(logging.WARNING): + compiled = self._compile(base, low, high) + + assert compiled.roles["course_editor"].definition.display_name == "High" + assert "role_extension display_name" in caplog.text + assert make_source("low").source_id in caplog.text + + def test_losing_permission_extension_warns(self, caplog): + """A permission-changing extension that loses on priority is reported.""" + base = make_document( + "base", + priority=100, + permissions=[permission(cat="cat")], + roles=[role(rid="course_editor", permissions=("courses.view_course",))], + ) + low = make_document( + "low", + priority=100, + role_extensions=[extension("course_editor", remove_permissions=("courses.view_course",))], + ) + high = make_document( + "high", + priority=200, + role_extensions=[extension("course_editor", add_permissions=("courses.view_course",))], + ) + + with caplog.at_level(logging.WARNING): + compiled = self._compile(base, low, high) + + # The higher-priority add wins, so the permission stays. + assert "courses.view_course" in compiled.roles["course_editor"].definition.permissions + assert "role_extension remove of 'courses.view_course'" in caplog.text + + +class TestNoOpExtensionWarnings: + """ADR 0023 §3: a no-op add/remove warns and leaves the result unchanged.""" + + @staticmethod + def _compile(*documents): + return SchemaCompiler().compile(list(documents)) + + def test_adding_an_existing_permission_warns(self, caplog): + """Adding a permission the role already has warns and is a no-op.""" + base = make_document( + "base", + priority=100, + permissions=[permission(cat="cat")], + roles=[role(rid="course_editor", permissions=("courses.view_course",))], + ) + ext = make_document( + "ext", priority=200, role_extensions=[extension("course_editor", add_permissions=("courses.view_course",))] + ) + + with caplog.at_level(logging.WARNING): + compiled = self._compile(base, ext) + + assert compiled.roles["course_editor"].definition.permissions == ("courses.view_course",) + assert "already on role" in caplog.text + + def test_removing_an_absent_permission_warns(self, caplog): + """Removing a permission the role does not have warns and is a no-op.""" + base = make_document("base", priority=100, roles=[role(rid="course_editor", permissions=())]) + ext = make_document( + "ext", + priority=200, + role_extensions=[extension("course_editor", remove_permissions=("courses.manage_tags",))], + ) + + with caplog.at_level(logging.WARNING): + compiled = self._compile(base, ext) + + assert compiled.roles["course_editor"].definition.permissions == () + assert "not on role" in caplog.text + + +class TestDefensiveBranches: + """Paths guarded against states validation is expected to have rejected.""" + + def test_extension_for_an_unknown_role_is_skipped(self): + """Validation errors on this; compilation must not raise on it.""" + ext = make_document( + "ext", priority=200, role_extensions=[extension("ghost", add_permissions=("courses.view_course",))] + ) + + compiled = SchemaCompiler().compile([ext]) + + assert not compiled.roles + assert not compiled.role_permission_sources + + def test_identical_duplicate_categories_merge_sources(self): + """Two identical category definitions merge into one, keeping both sources.""" + first = make_document("first", categories=[category("cat")]) + second = make_document("second", categories=[category("cat")]) + + compiled = SchemaCompiler().compile([first, second]) + + assert len(compiled.categories["cat"].sources) == 2 + + def test_identical_duplicate_permissions_merge_sources(self): + """Two identical permission definitions merge into one, keeping both sources.""" + first = make_document("first", permissions=[permission(cat="cat")]) + second = make_document("second", permissions=[permission(cat="cat")]) + + compiled = SchemaCompiler().compile([first, second]) + + assert len(compiled.permissions["courses.view_course"].sources) == 2 + + def test_lower_priority_base_definition_is_kept_out(self): + """The 'keep existing' branch: a later, lower-priority file loses.""" + high = make_document("high", priority=200, roles=[role(rid="course_editor", display_name="Author")]) + low = make_document("low", priority=100, roles=[role(rid="course_editor", display_name="Editor")]) + + compiled = SchemaCompiler().compile([high, low]) + + assert compiled.roles["course_editor"].definition.display_name == "Author" + assert [s.source_id for s in compiled.roles["course_editor"].sources] == [make_source("high").source_id]