From abc8d60ed9506df0f9810fd2a28f5ac7f7315620 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= Date: Wed, 23 Sep 2026 10:06:18 -0600 Subject: [PATCH 1/2] feat: add authz schema pipeline and load_authz_schema command --- CHANGELOG.rst | 45 ++ src/openedx_authz/engine/schema/loading.py | 41 +- src/openedx_authz/engine/schema/pipeline.py | 110 +++++ .../management/commands/load_authz_schema.py | 223 +++++++++ .../schema/test_load_authz_schema_command.py | 457 ++++++++++++++++++ .../tests/schema/test_loading.py | 287 ++++++++++- .../tests/schema/test_pipeline.py | 200 ++++++++ 7 files changed, 1350 insertions(+), 13 deletions(-) create mode 100644 src/openedx_authz/engine/schema/pipeline.py create mode 100644 src/openedx_authz/management/commands/load_authz_schema.py create mode 100644 src/openedx_authz/tests/schema/test_load_authz_schema_command.py create mode 100644 src/openedx_authz/tests/schema/test_pipeline.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 88bfebdf..88e0bc11 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -14,6 +14,51 @@ Change Log Unreleased ********** +Added +===== + +* Added the static authorization schema, a versioned YAML format for declaring permissions, + permission categories, roles, and role extensions (ADR 0017). The permissions and roles that + ``authz.policy`` defines are now also expressed as schema files under + ``openedx_authz/authz/schema/``. +* Added the schema loading pipeline in ``openedx_authz/engine/schema/``, covering the discover, + load, validate and compile phases of the lifecycle (ADR 0018), plus render and apply in + ``openedx_authz/engine/renderer.py``. +* Added schema discovery through the ``authz.schema`` entry-point group and the + ``OPENEDX_AUTHZ_SCHEMA_DIRECTORIES`` setting, so applications can ship authorization definitions + with their code and operators can contribute them through deployment configuration (ADR 0019). +* Added the ``load_authz_schema`` management command, the single non-interactive deployment entry + point, with ``--dry-run`` to print the change report without writing, ``--force`` to allow + removing roles that still have assignments, and repeatable ``--dir`` for CI and local runs. +* Added ``role_extensions`` support: an application or deployment can add or remove permissions and + replace the display metadata or ``hidden`` flag of an existing static role without copying its + definition. ``priority`` resolves conflicts; an unresolvable equal-priority conflict stops the run + before any database change (ADR 0023). +* Added first-class tables for compiled definitions and their provenance, in migration + ``0011_authz_schema_definitions``: permission categories, permission definitions, role + definitions, role-permission grants, schema sources, and one source-link table per definition kind + recording whether a contribution was a base definition or an extension (ADR 0025). +* Added source attribution at the role-permission grain, so contributions from different + applications to the same role remain distinguishable and queryable through + ``origins_for_role``, ``origins_for_permission``, ``origins_for_category`` and + ``origin_for_role_permission``. +* Added a change report before any write: the command lists the policy rows and the definitions that + would be added, updated or removed, including metadata-only edits that change no policy row + (ADR 0018 §6). + +Notes +===== + +* No authorization behavior changes in this release. Loading ``authz.policy`` works as before, and + the new pipeline runs only when ``load_authz_schema`` is invoked. +* Applying a schema is idempotent and preserves data the loader does not own: user assignments, + dynamic roles, legacy ``g2`` action-inheritance rows, and pre-existing policy rows that no schema + declares. Rows that already exist are adopted, gaining definition and source records rather than + being rewritten (ADR 0025 §6). +* Removing a static role that still has user assignments stops the deployment and reports the + assignments. ``--force`` removes the role together with its assignments and writes a + ``RoleAssignmentAudit`` record for each one. + 1.24.0 - 2026-09-14 ******************* diff --git a/src/openedx_authz/engine/schema/loading.py b/src/openedx_authz/engine/schema/loading.py index 6fa7f8f1..c5c70f68 100644 --- a/src/openedx_authz/engine/schema/loading.py +++ b/src/openedx_authz/engine/schema/loading.py @@ -41,6 +41,8 @@ class SchemaLoader: """ _UNKNOWN_DISTRIBUTION = "unknown" + _distribution_ambiguity_warned: set[tuple[str, str, str]] = set() + """Tracks warned (package, resource_path, selected) combinations to deduplicate logs.""" def load(self, resources: list[DiscoveredResource]) -> list[SchemaDocument]: """Load every discovered resource into a :class:`SchemaDocument`. @@ -110,7 +112,13 @@ def _resolve_distribution(cls, resource: DiscoveredResource) -> tuple[str, str]: mapping = metadata.packages_distributions() except Exception: # pylint: disable=broad-exception-caught # noqa: BLE001 - defensive; metadata quirks across envs mapping = {} - candidates = mapping.get(top_level) or [] + # ``packages_distributions()`` can list the same distribution more than + # once for one top-level package (observed with editable installs and + # overlapping metadata). Those are not competing owners, so collapse to + # the distinct names before resolving — otherwise a lone real owner that + # happens to be listed twice looks "ambiguous" and the fallback path + # warns on every resource for a non-problem. + candidates = sorted(set(mapping.get(top_level) or [])) if not candidates: return top_level, cls._UNKNOWN_DISTRIBUTION @@ -130,7 +138,8 @@ def _select_owning_distribution(cls, candidates: list[str], resource: Discovered it. If exactly zero or more than one distribution claims the file (or the file lists are unavailable), we cannot know the true owner, so we return the first candidate in sorted order — a stable choice across environments - — and log the ambiguity. + — and log the ambiguity once per distinct (package, resource_path, selected) + combination. """ if len(candidates) == 1: return candidates[0] @@ -141,16 +150,24 @@ def _select_owning_distribution(cls, candidates: list[str], resource: Discovered return owners[0] fallback = sorted(candidates)[0] - logger.warning( - "Could not uniquely resolve the distribution that ships a schema resource; selecting deterministically.", - extra={ - "top_level": resource.package, - "resource_path": resource.resource_path, - "candidates": sorted(candidates), - "matched_owners": sorted(owners), - "selected": fallback, - }, - ) + # Deduplicate warnings by tracking (package, resource_path, selected) combinations + warn_key = (resource.package, resource.resource_path, fallback) + if warn_key not in cls._distribution_ambiguity_warned: + cls._distribution_ambiguity_warned.add(warn_key) + logger.info( + "Schema resource for package '%s' is claimed by multiple distributions %s; " + "deterministically selected '%s'.", + resource.package, + sorted(candidates), + fallback, + extra={ + "package": resource.package, + "resource_path": resource.resource_path, + "candidates": sorted(candidates), + "matched_owners": sorted(owners), + "selected": fallback, + }, + ) return fallback @staticmethod diff --git a/src/openedx_authz/engine/schema/pipeline.py b/src/openedx_authz/engine/schema/pipeline.py new file mode 100644 index 00000000..5d437d8e --- /dev/null +++ b/src/openedx_authz/engine/schema/pipeline.py @@ -0,0 +1,110 @@ +"""End-to-end orchestration of the authz schema lifecycle (ADR 0018). + +:class:`SchemaPipeline` wires the steps together: + + discover -> load -> validate -> compile -> render -> (plan) -> apply + +The Casbin-free steps (discover..compile) live in :mod:`openedx_authz.engine.schema`; +render/apply live in :mod:`openedx_authz.engine.renderer`. This orchestrator is +the single entry point used by the deployment management command and by tests. + +Deployment runs discover-through-apply before the application serves traffic +(ADR 0018 §2). CI/local runs may stop after ``plan`` for a dry run, or pass +explicit resources. +""" + +from __future__ import annotations + +import logging + +from openedx_authz.engine.renderer import ( + ApplyResult, + ChangePlan, + PolicyRenderer, + SchemaApplier, +) +from openedx_authz.engine.schema.compilation import SchemaCompiler +from openedx_authz.engine.schema.discovery import SchemaDiscovery +from openedx_authz.engine.schema.exceptions import SchemaValidationError +from openedx_authz.engine.schema.loading import SchemaLoader +from openedx_authz.engine.schema.types import CompiledSchema +from openedx_authz.engine.schema.validation import SchemaValidator, ValidationIssue + +logger = logging.getLogger(__name__) + + +class SchemaPipeline: + """Runs the schema lifecycle from discovery through apply. + + Components are injected for testability; each defaults to its standard + implementation. + """ + + def __init__( + self, + *, + discovery: SchemaDiscovery | None = None, + loader: SchemaLoader | None = None, + validator: SchemaValidator | None = None, + compiler: SchemaCompiler | None = None, + renderer: PolicyRenderer | None = None, + applier: SchemaApplier | None = None, + ): + self._discovery = discovery or SchemaDiscovery() + self._loader = loader or SchemaLoader() + self._validator = validator or SchemaValidator() + self._compiler = compiler or SchemaCompiler() + self._renderer = renderer or PolicyRenderer() + self._applier = applier or SchemaApplier() + + def compile(self) -> CompiledSchema: + """Run discover -> load -> validate -> compile and return the result. + + Validation gates twice: once on the loaded documents, then again on the + compiled schema, because extensions and priority resolution can only be + checked after they are applied (ADR 0017 §4). + + Raises: + SchemaValidationError: If either validation pass finds error-level + issues. + SchemaCompileError: On an unresolvable conflict. + """ + resources = self._discovery.discover() + documents = self._loader.load(resources) + + self._gate(self._validator.validate(documents)) + schema = self._compiler.compile(documents) + self._gate(self._validator.validate_compiled(schema)) + + return schema + + def _gate(self, issues: list[ValidationIssue]) -> None: + """Report every issue, then stop the run if any is error-level. + + Warnings are logged and the run continues; errors are logged and raised + together so the deployment report lists all of them at once. + """ + for issue in issues: + log = logger.error if issue.is_error else logger.warning + log("authz schema %s: %s [%s]", issue.level, issue.message, issue.source_id or "-") + if self._validator.has_errors(issues): + raise SchemaValidationError([i for i in issues if i.is_error]) + + def plan(self) -> ChangePlan: + """Run through render and produce the change report without writing. + + Used for dry-run / CI review (ADR 0018 §6). + """ + schema = self.compile() + rendered = self._renderer.render(schema) + return self._applier.plan(rendered, schema) + + def apply(self, *, force: bool = False) -> ApplyResult: + """Run the full lifecycle and persist the result transactionally. + + Args: + force: Allow removal of roles that still have assignments (ADR 0018). + """ + schema = self.compile() + rendered = self._renderer.render(schema) + return self._applier.apply(rendered, schema, force=force) diff --git a/src/openedx_authz/management/commands/load_authz_schema.py b/src/openedx_authz/management/commands/load_authz_schema.py new file mode 100644 index 00000000..0d2598d2 --- /dev/null +++ b/src/openedx_authz/management/commands/load_authz_schema.py @@ -0,0 +1,223 @@ +"""Discover, validate, compile, report, and apply the static authz schema. + +This is the single non-interactive deployment command described in ADR 0019 §3. +Tutor (via a plugin init task) and other deployment systems invoke it before the +application serves traffic; all integrations share this one compiler/pipeline. + +Usage:: + + python manage.py load_authz_schema # full apply + python manage.py load_authz_schema --dry-run # report only, no writes + python manage.py load_authz_schema --force # allow role removals + python manage.py load_authz_schema \\ + --dir openedx_authz/authz/schema # explicit directory (CI/local) + +The command must run at a point where all contributing packages are installed +and Django settings/DB are available (ADR 0018 / plugin timing constraint). +""" + +from __future__ import annotations + +from typing import Callable + +from django.core.management.base import BaseCommand, CommandError + +from openedx_authz.engine.schema.discovery import SchemaDiscovery, SchemaDiscoveryError +from openedx_authz.engine.schema.exceptions import SchemaError +from openedx_authz.engine.schema.pipeline import SchemaPipeline + + +class Command(BaseCommand): + """Management command wrapper around :class:`SchemaPipeline`.""" + + help = "Discover, validate, compile, and apply the static authorization schema." + + def add_arguments(self, parser) -> None: + """Register command-line options.""" + parser.add_argument( + "--dry-run", + action="store_true", + help="Run discover through render and print the change report without writing to the database.", + ) + parser.add_argument( + "--force", + action="store_true", + help="Allow removing static roles that still have user assignments (ADR 0018).", + ) + parser.add_argument( + "--dir", + action="append", + default=None, + dest="directories", + metavar="DIRECTORY", + help=( + "Explicitly include a schema directory (repeatable), in addition to discovered " + "entry points and settings. The loader reads every .yaml file in it. " + "Path format is 'top_level_package/sub/dir' (e.g. 'openedx_authz/authz/schema'). " + "Intended for CI and local development." + ), + ) + + def handle(self, *args, **options) -> None: + """Build the pipeline and run the requested operation. + + Validation/compile/apply errors surface as CommandError so deployment + stops before (or without partially applying) any database change. + """ + self._write_banner() + + directories = options.get("directories") or [] + discovery = SchemaDiscovery(passed_in_directories=directories) if directories else SchemaDiscovery() + pipeline = SchemaPipeline(discovery=discovery) + + try: + if options.get("dry_run"): + plan = pipeline.plan() + self._report_plan(plan) + self.stdout.write(self.style.SUCCESS("\nDry run: no changes were applied.")) + return + + result = pipeline.apply(force=options.get("force", False)) + except (SchemaError, SchemaDiscoveryError) as exc: + raise CommandError(str(exc)) from exc + + if result.unchanged: + self.stdout.write(self.style.SUCCESS("Authz schema unchanged; no rows written.")) + return + + # Print the same detailed breakdown a dry run would, so an operator can + # see exactly which Casbin policy rows and definition records changed, + # then close with the applied summary. + if result.plan is not None: + self._report_plan(result.plan) + self.stdout.write(self.style.SUCCESS(f"\n{self._apply_summary(result)}")) + + def _apply_summary(self, result) -> str: + """One-line recap of what apply wrote, across both layers. + + Reports the Casbin ``p`` row counts and the definition-metadata counts + (roles/permissions/categories/grants) together, since either layer can + change on its own — a metadata-only edit writes 0 policy rows but is + still a real change the operator should see reflected here. + """ + summary = f"Authz schema applied: {result.added} Casbin policy row(s) added, {result.removed} removed" + + plan = result.plan + if plan is not None and not plan.definitions_unchanged: + parts = [f"{len(diff)} {label}" for label, diff in plan.definition_diffs if not diff.is_empty] + summary += f"; definition changes: {', '.join(parts)}" + + return f"{summary}." + + def _write_banner(self) -> None: + """Write the command banner that opens every run. + + This output is emitted mid-deployment (e.g. interleaved into a Tutor + launch log), so every path — dry-run, apply, and both "unchanged" + short-circuits — must be attributable to ``load_authz_schema`` without + surrounding context. Emitting it once at the top of ``handle`` covers + them all, including the early returns that never reach the change report. + ``MIGRATE_HEADING`` is bold on a tty and plain text otherwise. + """ + self.stdout.write(self.style.MIGRATE_HEADING("[load_authz_schema] Loading authorization schema")) + + def _write_marked_line(self, marker: str, text: str, style: Callable[[str], str]) -> None: + """Write one `` {marker} {text}`` line, colorized via Django's ``style``. + + Centralizing this keeps the literal marker character and the color that + decorates it paired in one place, so the two layers below (Casbin rows, + definitions) can't drift out of sync with each other on how a given kind + of change is marked. ``style`` is one of ``self.style.SUCCESS`` / + ``WARNING`` / ``ERROR``; Django's style wrappers pass text through + unchanged when stdout isn't a tty, so the marker stays readable without + color (requirement: markers are real text, not color-only). + """ + self.stdout.write(style(f" {marker} {text}")) + + def _write_section_header(self, text: str) -> None: + """Write a bold section header preceded by a blank line. + + ``MIGRATE_HEADING`` is Django's built-in bold heading style; like the + other ``self.style`` wrappers it emits bold ANSI on a tty and plain text + otherwise, so the header stays readable (and the leading blank line + separates each section) with or without color. + """ + self.stdout.write(self.style.MIGRATE_HEADING(f"\n{text}")) + + @staticmethod + def _format_policy_row(row) -> str: + """Render a Casbin row as its policy line, e.g. ``p, role^x, act^y, scope^*, allow``. + + ``row.as_policy()`` returns the enforcer arg form (a ``list[str]``) used + when talking to Casbin; for the human-facing report we prefix the row's + policy type (``row.PTYPE`` — ``p`` for a permission row, ``g`` for a + grouping row) and comma-join the fields, matching how a Casbin policy + line is written in a model/policy file. The underlying ``as_policy`` + contract is left untouched since the enforcer depends on the list form. + """ + return ", ".join([row.PTYPE, *row.as_policy()]) + + def _report_plan(self, plan) -> None: + """Print the change report (ADR 0018 §6). + + Covers the definition tables (roles, permissions, categories, and + role-permission grants) as well as the Casbin ``p`` policy rows. The two + are reported as separate top-level sections because they are distinct + layers: apply syncs the definition metadata even when no ``p`` row + changes, so a metadata-only edit is a real change the operator needs to + see. Each layer is one compact, interleaved diff (additions and removals + together) rather than split by direction, so a reviewer can scan one + block per layer instead of hunting across separate add/remove sections. + """ + if plan.unchanged: + self.stdout.write(self.style.SUCCESS("Authz schema unchanged; nothing would be written.")) + return + + self.stdout.write("\nChanges to apply:") + + total_rows = len(plan.added_rows) + len(plan.removed_rows) + if total_rows: + self._write_section_header(f"Casbin policies ({total_rows})") + for row in plan.added_rows: + self._write_marked_line("+", self._format_policy_row(row), self.style.SUCCESS) + for row in plan.removed_rows: + self._write_marked_line("-", self._format_policy_row(row), self.style.ERROR) + + self._report_definitions(plan) + + if plan.blocking_assignments: + self.stdout.write( + self.style.WARNING( + f"{len(plan.blocking_assignments)} role(s) with existing assignments would be " + "removed; apply requires --force:" + ) + ) + for role, subject in plan.blocking_assignments: + self.stdout.write(f" ! {role} assigned to {subject}") + + def _report_definitions(self, plan) -> None: + """Print the definition-metadata changes as one compact, interleaved diff. + + These are the role/permission/category/grant records, a separate layer + from the Casbin policy rows above: they can change on their own (e.g. a + display-name edit) without adding or removing any ``p`` row. The single + header no longer names the kind (there's only one header for the whole + section), so each line is made self-describing with an inline + ``{label}: `` prefix, keeping the existing per-kind iteration order and + the existing added-then-updated-then-removed order within each kind. + """ + if plan.definitions_unchanged: + self.stdout.write("Role/permission/category definitions unchanged.") + return + + total = sum(len(diff) for _, diff in plan.definition_diffs) + self._write_section_header(f"Definitions ({total})") + for label, diff in plan.definition_diffs: + if diff.is_empty: + continue + for key in diff.added: + self._write_marked_line("+", f"{label}: {key}", self.style.SUCCESS) + for key in diff.updated: + self._write_marked_line("~", f"{label}: {key}", self.style.WARNING) + for key in diff.removed: + self._write_marked_line("-", f"{label}: {key}", self.style.ERROR) diff --git a/src/openedx_authz/tests/schema/test_load_authz_schema_command.py b/src/openedx_authz/tests/schema/test_load_authz_schema_command.py new file mode 100644 index 00000000..ef5374ae --- /dev/null +++ b/src/openedx_authz/tests/schema/test_load_authz_schema_command.py @@ -0,0 +1,457 @@ +"""Unit tests for the ``load_authz_schema`` management command. + +The command is a thin wrapper over :class:`SchemaPipeline`. These tests mock the +pipeline (and, where relevant, discovery) at the command module so the command's +own logic — option handling, apply vs. dry-run branching, report formatting, and +error translation to CommandError — is verified without a database. +""" + +from io import StringIO +from unittest import mock + +import pytest +from django.core.management import call_command +from django.core.management.base import CommandError + +from openedx_authz.engine.renderer import ( + ApplyResult, + ChangePlan, + DefinitionDiff, + PolicyRow, +) +from openedx_authz.engine.schema.discovery import SchemaDiscoveryError +from openedx_authz.engine.schema.exceptions import ( + SchemaApplyError, + SchemaCompileError, + SchemaValidationError, +) + +COMMAND = "load_authz_schema" +PIPELINE_PATH = "openedx_authz.management.commands.load_authz_schema.SchemaPipeline" +DISCOVERY_PATH = "openedx_authz.management.commands.load_authz_schema.SchemaDiscovery" + + +def _run(*args): + """Invoke the command, capturing stdout; returns the printed text.""" + out = StringIO() + call_command(COMMAND, *args, stdout=out) + return out.getvalue() + + +class TestApplyMode: + """Cover the default (apply) mode of the command.""" + + def test_apply_reports_changes(self): + """Apply prints the added/removed policy-row summary.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(added=3, removed=1, unchanged=False) + output = _run() + + pipeline_cls.return_value.apply.assert_called_once_with(force=False) + assert "3 Casbin policy row(s) added, 1 removed" in output + + def test_apply_prints_detailed_change_report(self): + """Apply prints the same breakdown as a dry run when a plan is attached.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + removed_rows=[PolicyRow("role^old", "act^courses.manage_tags", "course-v1^*", "allow")], + unchanged=False, + roles=DefinitionDiff(updated=["course_editor"]), + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(added=1, removed=1, unchanged=False, plan=plan) + output = _run() + + # Detailed policy-row and definition sections, as one compact diff each. + assert "Casbin policies (2)" in output + assert "role^r" in output + assert "role^old" in output + assert "Definitions (1)" in output + assert "~ role: course_editor" in output + # And still closes with the applied summary, now including definitions, + # preceded by a blank line for consistency with the dry-run closing note. + assert "\nAuthz schema applied: 1 Casbin policy row(s) added, 1 removed; definition changes: 1 role" in output + + def test_apply_summary_counts_definition_changes_with_zero_policy_rows(self): + """A metadata-only apply writes 0 p rows but still recaps definitions.""" + plan = ChangePlan( + added_rows=[], + removed_rows=[], + unchanged=False, + categories=DefinitionDiff(added=["course_content", "library"]), + roles=DefinitionDiff(updated=["course_editor"]), + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(added=0, removed=0, unchanged=False, plan=plan) + output = _run() + + assert "0 Casbin policy row(s) added, 0 removed; definition changes: 2 category, 1 role" in output + + def test_apply_summary_omits_definitions_when_unchanged(self): + """Row-only changes don't tack on an empty definition recap.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(added=1, removed=0, unchanged=False, plan=plan) + output = _run() + + assert "1 Casbin policy row(s) added, 0 removed." in output + assert "definition changes:" not in output + + def test_apply_reports_unchanged(self): + """An apply that changes nothing reports 'unchanged'.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(added=0, removed=0, unchanged=True) + output = _run() + + assert "unchanged" in output.lower() + + def test_apply_output_always_opens_with_the_command_banner(self): + """The banner precedes every apply path, including the unchanged short-circuit. + + The ``result.unchanged`` branch returns before the change report, so the + banner must come from ``handle`` itself to label this ("no rows written") + output in a Tutor launch log. + """ + banner = "[load_authz_schema] Loading authorization schema" + + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(added=0, removed=0, unchanged=True) + unchanged_output = _run() + + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(added=1, removed=0, unchanged=False, plan=plan) + applied_output = _run() + + assert banner in unchanged_output + assert "Authz schema unchanged; no rows written." in unchanged_output + assert banner in applied_output + + def test_force_flag_is_forwarded(self): + """``--force`` is passed through to ``pipeline.apply``.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(unchanged=True) + _run("--force") + + pipeline_cls.return_value.apply.assert_called_once_with(force=True) + + +class TestDryRunMode: + """Cover the --dry-run mode and its change report formatting.""" + + def test_dry_run_calls_plan_not_apply(self): + """``--dry-run`` calls ``plan`` and never ``apply``.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = ChangePlan(unchanged=True) + _run("--dry-run") + + pipeline_cls.return_value.plan.assert_called_once_with() + pipeline_cls.return_value.apply.assert_not_called() + + def test_dry_run_unchanged_report(self): + """A dry run with no diff reports 'unchanged'.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = ChangePlan(unchanged=True) + output = _run("--dry-run") + + assert "unchanged" in output.lower() + + def test_dry_run_closes_with_a_no_changes_applied_message(self): + """A dry run ends with an explicit 'no changes were applied' note, after a blank line.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Dry run: no changes were applied." in output + assert "\nDry run: no changes were applied." in output + assert output.rstrip().endswith("Dry run: no changes were applied.") + + def test_dry_run_closing_message_also_shown_when_unchanged(self): + """Even an unchanged dry run states it was a dry run with nothing applied.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = ChangePlan(unchanged=True) + output = _run("--dry-run") + + assert "Dry run: no changes were applied." in output + + def test_report_opens_with_a_command_banner(self): + """A dry run opens with a banner naming the command, changed or unchanged. + + The output is emitted mid-deployment (e.g. in a Tutor launch log), so it + must be attributable to ``load_authz_schema`` without surrounding context. + """ + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = ChangePlan(unchanged=True) + unchanged_output = _run("--dry-run") + + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + changed_output = _run("--dry-run") + + assert "[load_authz_schema] Loading authorization schema" in unchanged_output + assert "[load_authz_schema] Loading authorization schema" in changed_output + + def test_dry_run_reports_added_and_removed_rows(self): + """A dry run lists the policy rows it would add and remove.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + removed_rows=[PolicyRow("role^old", "act^courses.manage_tags", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Casbin policies (2)" in output + assert "role^r" in output + assert "role^old" in output + + def test_dry_run_reports_compact_interleaved_casbin_section(self): + """Added and removed Casbin rows share one 'Casbin policies (N)' header, interleaved +/-.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^course_editor", "act^courses.view_course", "course-v1^*", "allow")], + removed_rows=[PolicyRow("role^old_editor", "act^courses.manage_tags", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Casbin policies (2)" in output + assert "role^course_editor" in output + assert "role^old_editor" in output + assert "Definitions" not in output + + def test_report_opens_with_a_changes_to_apply_title(self): + """The change report is introduced by a 'Changes to apply:' title line.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Changes to apply:" in output + + def test_casbin_rows_render_as_comma_joined_policy_lines(self): + """Each Casbin row prints as a `` p, `` policy line, not a list repr.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^course_editor", "act^courses.view_course", "course-v1^*", "allow")], + removed_rows=[PolicyRow("role^old_editor", "act^courses.manage_tags", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "+ p, role^course_editor, act^courses.view_course, course-v1^*, allow" in output + assert "- p, role^old_editor, act^courses.manage_tags, course-v1^*, allow" in output + # The old Python-list repr form must not leak into the report. + assert "['role^course_editor'" not in output + + def test_dry_run_reports_compact_definitions_section_without_casbin(self): + """A definitions-only change prints 'Definitions (N)' with the Casbin section omitted.""" + plan = ChangePlan( + unchanged=False, + roles=DefinitionDiff(added=["course_editor"]), + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Definitions (1)" in output + assert "+ role: course_editor" in output + assert "Casbin policies" not in output + + def test_dry_run_reports_blocking_assignments(self): + """A dry run flags assignments that would require ``--force`` to remove.""" + plan = ChangePlan( + added_rows=[], + removed_rows=[], + unchanged=False, + blocking_assignments=[("role^course_editor", "user^alice")], + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "requires --force" in output + assert "role^course_editor assigned to user^alice" in output + + +class TestDefinitionReport: + """The dry-run report covers definition changes too (ADR 0018 §6). + + Apply syncs the definition tables even when no ``p`` row changes, so a + metadata-only edit has to appear in the report. + """ + + def test_metadata_only_change_is_reported_without_any_rows(self): + """A metadata-only edit is reported even though no policy row changes.""" + plan = ChangePlan( + added_rows=[], + removed_rows=[], + unchanged=False, + roles=DefinitionDiff(updated=["course_editor"]), + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Definitions (1)" in output + assert "~ role: course_editor" in output + + def test_added_and_removed_definitions_are_reported_per_kind(self): + """Definition changes are grouped under one header and labeled inline per kind.""" + plan = ChangePlan( + unchanged=False, + categories=DefinitionDiff(added=["course_content"]), + permissions=DefinitionDiff(removed=["courses.manage_tags"]), + grants=DefinitionDiff(added=["course_editor -> courses.view_course @ course-v1"]), + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Definitions (3)" in output + assert "+ category: course_content" in output + assert "- permission: courses.manage_tags" in output + assert "+ role-permission: course_editor -> courses.view_course @ course-v1" in output + + def test_untouched_kinds_are_omitted(self): + """Kinds with no changes are left out of the report.""" + plan = ChangePlan(unchanged=False, roles=DefinitionDiff(added=["course_editor"])) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Definitions (1)" in output + assert "category" not in output + assert "permission" not in output + + def test_row_only_change_says_definitions_unchanged(self): + """A row-only change states explicitly that definitions are unchanged.""" + plan = ChangePlan( + added_rows=[PolicyRow("role^r", "act^courses.view_course", "course-v1^*", "allow")], + unchanged=False, + ) + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = plan + output = _run("--dry-run") + + assert "Role/permission/category definitions unchanged." in output + + +class TestDirectoryOption: + """Cover the --dir option wiring into SchemaDiscovery.""" + + def test_dir_builds_discovery_with_passed_in_directories(self): + """Repeated ``--dir`` options are passed to ``SchemaDiscovery`` as directories.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls, mock.patch(DISCOVERY_PATH) as discovery_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(unchanged=True) + _run("--dir", "pkg_a/authz/schema", "--dir", "pkg_b/authz/schema") + + discovery_cls.assert_called_once_with(passed_in_directories=["pkg_a/authz/schema", "pkg_b/authz/schema"]) + # The pipeline is built with that discovery instance. + pipeline_cls.assert_called_once_with(discovery=discovery_cls.return_value) + + def test_no_dir_uses_default_discovery(self): + """Without ``--dir`` the command builds a default ``SchemaDiscovery``.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls, mock.patch(DISCOVERY_PATH) as discovery_cls: + pipeline_cls.return_value.apply.return_value = ApplyResult(unchanged=True) + _run() + + # Default discovery (no explicit directories) is constructed. + discovery_cls.assert_called_once_with() + + +class TestErrorHandling: + """Cover translation of pipeline errors into CommandError. + + Deployment must stop with a readable message rather than a traceback, and + ``SchemaDiscoveryError`` needs handling separately because it does not + inherit from ``SchemaError``. + """ + + def test_schema_error_becomes_command_error(self): + """A validation error during apply is surfaced as ``CommandError``.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.side_effect = SchemaValidationError([]) + with pytest.raises(CommandError): + _run() + + def test_dry_run_error_becomes_command_error(self): + """A validation error during a dry run is surfaced as ``CommandError``.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.side_effect = SchemaValidationError([]) + with pytest.raises(CommandError): + _run("--dry-run") + + def test_discovery_error_becomes_command_error(self): + """ADR 0019 §1: a failing provider stops deployment, naming the app.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.side_effect = SchemaDiscoveryError( + "authz.schema provider 'broken_app' failed during discovery: boom" + ) + with pytest.raises(CommandError, match="broken_app"): + _run() + + def test_discovery_error_in_dry_run_becomes_command_error(self): + """A discovery error during a dry run is surfaced as ``CommandError``.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.side_effect = SchemaDiscoveryError("bad directory") + with pytest.raises(CommandError, match="bad directory"): + _run("--dry-run") + + def test_compile_error_becomes_command_error(self): + """A compile error is surfaced as ``CommandError`` with its message.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.side_effect = SchemaCompileError("equal priority conflict") + with pytest.raises(CommandError, match="equal priority conflict"): + _run() + + def test_apply_error_becomes_command_error(self): + """The force gate surfaces as a message, not a traceback.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.apply.side_effect = SchemaApplyError("Refusing to proceed") + with pytest.raises(CommandError, match="Refusing to proceed"): + _run() + + +class TestOptionCombinations: + """Options compose: a dry run can also take explicit directories.""" + + def test_dry_run_with_dir_plans_against_that_directory(self): + """``--dry-run`` composes with ``--dir``: it plans against the given directory.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls, mock.patch(DISCOVERY_PATH) as discovery_cls: + pipeline_cls.return_value.plan.return_value = ChangePlan(unchanged=True) + _run("--dry-run", "--dir", "pkg_a/authz/schema") + + discovery_cls.assert_called_once_with(passed_in_directories=["pkg_a/authz/schema"]) + pipeline_cls.assert_called_once_with(discovery=discovery_cls.return_value) + pipeline_cls.return_value.plan.assert_called_once_with() + pipeline_cls.return_value.apply.assert_not_called() + + def test_dry_run_ignores_force(self): + """A dry run writes nothing, so force has nothing to authorize.""" + with mock.patch(PIPELINE_PATH) as pipeline_cls: + pipeline_cls.return_value.plan.return_value = ChangePlan(unchanged=True) + _run("--dry-run", "--force") + + pipeline_cls.return_value.plan.assert_called_once_with() + pipeline_cls.return_value.apply.assert_not_called() diff --git a/src/openedx_authz/tests/schema/test_loading.py b/src/openedx_authz/tests/schema/test_loading.py index 40435758..8dc1bf7d 100644 --- a/src/openedx_authz/tests/schema/test_loading.py +++ b/src/openedx_authz/tests/schema/test_loading.py @@ -1,10 +1,11 @@ """Unit tests for the schema loading step.""" +import logging from importlib import metadata import pytest -from openedx_authz.engine.schema.discovery import SchemaDiscovery +from openedx_authz.engine.schema.discovery import DiscoveredResource, Origin, SchemaDiscovery from openedx_authz.engine.schema.exceptions import SchemaLoadError from openedx_authz.engine.schema.loading import SchemaLoader @@ -244,6 +245,26 @@ def test_ambiguous_ownership_falls_back_to_a_deterministic_choice(self, monkeypa assert docs[0].source.distribution == "alpha-dist" + def test_same_distribution_listed_twice_is_not_ambiguous(self, monkeypatch, caplog): + """A distribution listed more than once for a package is one owner, not an ambiguity. + + ``packages_distributions`` can report the same distribution twice (seen + with editable installs / overlapping metadata), e.g. + ``["openedx-authz", "openedx-authz"]``. That is a single real owner, so + it must resolve to that distribution with no "multiple distributions" + fallback warning — collapsing the duplicates is what prevents the log + from repeating on every schema resource. + """ + monkeypatch.setattr(metadata, "packages_distributions", lambda: {"pkg": ["dup-dist", "dup-dist"]}) + monkeypatch.setattr(metadata, "version", lambda _name: "1.0") + + with caplog.at_level(logging.INFO, logger="openedx_authz.engine.schema.loading"): + docs = _load(b"schema_version: '1.0'\npriority: 1\n") + + assert docs[0].source.distribution == "dup-dist" + assert docs[0].source.distribution_version == "1.0" + assert "multiple distributions" not in caplog.text + def test_digest_reflects_the_file_contents(self): """Different file contents produce different content digests.""" first = _load(b"schema_version: '1.0'\npriority: 1\n")[0] @@ -338,3 +359,267 @@ def test_error_names_the_source(self): """A malformed entry error names the contributing source.""" with pytest.raises(SchemaLoadError, match="pkg.mod"): _load(b"schema_version: '1.0'\npriority: 1\nroles:\n - 5\n") + + +class TestDistributionAmbiguityLogging: + """Test distribution resolution logging and deduplication. + + When multiple distributions claim the same top-level package, the loader + must select one deterministically and log the ambiguity exactly once per + distinct (package, resource_path, selected) combination. + """ + + @pytest.fixture(autouse=True) + def _clear_warned_distributions(self) -> None: + """Clear the class-level distribution ambiguity tracker before and after each test. + + Ensures test isolation and order-independence by resetting the dedupe set + that prevents duplicate logging of the same ambiguity. + """ + SchemaLoader._distribution_ambiguity_warned.clear() # pylint: disable=protected-access + yield + SchemaLoader._distribution_ambiguity_warned.clear() # pylint: disable=protected-access + + def test_single_candidate_no_log( + self, + caplog: pytest.LogCaptureFixture, + ) -> None: + """When there is a single candidate, no ambiguity log is emitted.""" + caplog.set_level(logging.INFO) + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + + result = SchemaLoader._select_owning_distribution(["dist1"], resource) # pylint: disable=protected-access + + assert result == "dist1" + assert len(caplog.records) == 0 + + def test_unique_owner_no_log( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """When ownership can be uniquely resolved, no ambiguity log is emitted.""" + caplog.set_level(logging.INFO) + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + + # Mock _distribution_ships so only dist1 claims the resource + def mock_ships(distribution: str, _installed_path: str) -> bool: + return distribution == "dist1" + + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(mock_ships)) + + result = SchemaLoader._select_owning_distribution(["dist1", "dist2"], resource) # pylint: disable=protected-access + + assert result == "dist1" + assert len(caplog.records) == 0 + + def test_ambiguous_fallback_logs_once( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """When ownership is ambiguous, a single info-level log is emitted.""" + caplog.set_level(logging.INFO) + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + + # Mock _distribution_ships so no distribution claims the resource + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource) # pylint: disable=protected-access + + # Should emit exactly one log record + assert len(caplog.records) == 1 + assert caplog.records[0].levelno == logging.INFO + assert caplog.records[0].levelname == "INFO" + + def test_log_message_includes_distribution( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Log message includes the selected distribution name and package.""" + caplog.set_level(logging.INFO) + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + result = SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource) # pylint: disable=protected-access + + assert result == "dist1" + log_msg = caplog.records[0].message + assert "dist1" in log_msg + assert "test_package" in log_msg + + def test_log_extra_dict_preserved( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Log record preserves all structured context in the 'extra' dict.""" + caplog.set_level(logging.INFO) + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource) # pylint: disable=protected-access + + record = caplog.records[0] + assert hasattr(record, "package") + assert record.package == "test_package" + assert hasattr(record, "resource_path") + assert record.resource_path == "schema/test.yaml" + assert hasattr(record, "candidates") + assert set(record.candidates) == {"dist1", "dist2"} + assert hasattr(record, "matched_owners") + assert record.matched_owners == [] + assert hasattr(record, "selected") + assert record.selected == "dist1" + + def test_same_ambiguity_logged_only_once( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Identical (package, resource_path, selected) combinations log only once. + + This tests the deduplication behavior across multiple calls within + the same process. + """ + caplog.set_level(logging.INFO) + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + + # Call multiple times with the same resource + for _ in range(3): + SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource) # pylint: disable=protected-access + + # Should emit exactly one log record despite three calls + assert len(caplog.records) == 1 + + def test_different_resources_each_logged( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Different resources (even with same candidates) each log once.""" + caplog.set_level(logging.INFO) + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + resource1 = DiscoveredResource( + package="test_package", + resource_path="schema/test1.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + resource2 = DiscoveredResource( + package="test_package", + resource_path="schema/test2.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + + SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource1) # pylint: disable=protected-access + SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource2) # pylint: disable=protected-access + + # Should emit two log records (one per unique resource_path) + assert len(caplog.records) == 2 + + def test_selection_logic_unchanged( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Fallback selection is still deterministic (sorted first candidate).""" + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + # Test with various orderings; should always return sorted[0] + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + candidates = ["zebra", "apple", "banana"] + + result = SchemaLoader._select_owning_distribution(candidates, resource) # pylint: disable=protected-access + + assert result == "apple" # sorted(candidates)[0] + + def test_log_level_is_info_not_warning( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Log level is INFO, not WARNING, for the ambiguity.""" + caplog.set_level(logging.DEBUG) + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource) # pylint: disable=protected-access + + assert len(caplog.records) == 1 + record = caplog.records[0] + assert record.levelno == logging.INFO + assert record.levelname == "INFO" + + def test_dedupe_key_includes_selected_distribution( + self, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Deduplication key depends on the selected distribution. + + Since sorting is deterministic, the same candidates always yield + the same fallback, so we verify that identical candidates produce + identical fallbacks and thus deduplicate. + """ + caplog.set_level(logging.INFO) + monkeypatch.setattr(SchemaLoader, "_distribution_ships", staticmethod(lambda *args: False)) + + resource = DiscoveredResource( + package="test_package", + resource_path="schema/test.yaml", + module="test_package.schema", + origin=Origin.ENTRY_POINT, + ) + + # Both orderings should select the same deterministic fallback + result1 = SchemaLoader._select_owning_distribution(["dist2", "dist1"], resource) # pylint: disable=protected-access + result2 = SchemaLoader._select_owning_distribution(["dist1", "dist2"], resource) # pylint: disable=protected-access + + assert result1 == result2 == "dist1" + # Only one log record because the warn_key is identical + assert len(caplog.records) == 1 diff --git a/src/openedx_authz/tests/schema/test_pipeline.py b/src/openedx_authz/tests/schema/test_pipeline.py new file mode 100644 index 00000000..948a7712 --- /dev/null +++ b/src/openedx_authz/tests/schema/test_pipeline.py @@ -0,0 +1,200 @@ +"""Unit tests for the SchemaPipeline orchestrator. + +The pipeline is pure wiring: it sequences discovery -> load -> validate -> +compile -> render -> plan/apply. These tests inject mocked components so the +orchestration (ordering, error propagation, delegation) is verified without a +database, Casbin, or real schema files. +""" + +from unittest import mock + +import pytest + +from openedx_authz.engine.schema.exceptions import SchemaValidationError +from openedx_authz.engine.schema.pipeline import SchemaPipeline +from openedx_authz.engine.schema.validation import ValidationIssue + + +def _pipeline(*, issues=None, compiled_issues=None): + """Build a SchemaPipeline with every component mocked. + + ``issues`` seeds the document-level validator result and ``compiled_issues`` + the post-compile one (both default to none). + """ + discovery = mock.Mock(name="discovery") + discovery.discover.return_value = ["resource"] + + loader = mock.Mock(name="loader") + loader.load.return_value = ["document"] + + validator = mock.Mock(name="validator") + validator.validate.return_value = issues or [] + validator.validate_compiled.return_value = compiled_issues or [] + validator.has_errors.side_effect = lambda found: any(i.is_error for i in found) + + compiler = mock.Mock(name="compiler") + compiler.compile.return_value = "compiled-schema" + + renderer = mock.Mock(name="renderer") + renderer.render.return_value = "rendered-policy" + + applier = mock.Mock(name="applier") + + pipeline = SchemaPipeline( + discovery=discovery, + loader=loader, + validator=validator, + compiler=compiler, + renderer=renderer, + applier=applier, + ) + return pipeline, { + "discovery": discovery, + "loader": loader, + "validator": validator, + "compiler": compiler, + "renderer": renderer, + "applier": applier, + } + + +class TestCompile: + """Cover SchemaPipeline.compile step ordering and validation gating.""" + + def test_runs_steps_in_order_and_returns_compiled_schema(self): + """Compile runs discover -> load -> validate -> compile -> validate_compiled in order.""" + pipeline, m = _pipeline() + + result = pipeline.compile() + + assert result == "compiled-schema" + m["discovery"].discover.assert_called_once_with() + m["loader"].load.assert_called_once_with(["resource"]) + m["validator"].validate.assert_called_once_with(["document"]) + m["compiler"].compile.assert_called_once_with(["document"]) + m["validator"].validate_compiled.assert_called_once_with("compiled-schema") + + def test_raises_when_compiled_schema_has_errors(self): + """The second gate runs on the compiled schema (ADR 0017 §4). + + Extensions and priority resolution can only be checked after they are + applied, so validation runs again post-compile. + """ + error = ValidationIssue("error", "scope not supported", "src") + pipeline, m = _pipeline(compiled_issues=[error]) + + with pytest.raises(SchemaValidationError) as exc_info: + pipeline.compile() + + assert exc_info.value.issues == [error] + m["compiler"].compile.assert_called_once() + + def test_compiled_errors_stop_before_render_and_apply(self): + """A post-compile error aborts apply before rendering or applying.""" + error = ValidationIssue("error", "scope not supported", "src") + pipeline, m = _pipeline(compiled_issues=[error]) + + with pytest.raises(SchemaValidationError): + pipeline.apply() + + m["renderer"].render.assert_not_called() + m["applier"].apply.assert_not_called() + + def test_compiled_warnings_do_not_stop_compilation(self): + """A post-compile warning is non-fatal; compilation still returns the schema.""" + warning = ValidationIssue("warning", "heads up", "src") + pipeline, _ = _pipeline(compiled_issues=[warning]) + + assert pipeline.compile() == "compiled-schema" + + def test_document_errors_skip_the_compiled_check(self): + """A failed first gate must not reach the second one.""" + error = ValidationIssue("error", "boom", "src") + pipeline, m = _pipeline(issues=[error]) + + with pytest.raises(SchemaValidationError): + pipeline.compile() + + m["validator"].validate_compiled.assert_not_called() + + def test_raises_when_validation_has_errors(self): + """A document-level validation error stops before compilation runs.""" + error = ValidationIssue("error", "boom", "src") + pipeline, m = _pipeline(issues=[error]) + + with pytest.raises(SchemaValidationError) as exc_info: + pipeline.compile() + + # Only error-level issues are carried on the exception. + assert exc_info.value.issues == [error] + # Compilation must not run once validation fails. + m["compiler"].compile.assert_not_called() + + def test_warning_only_issues_do_not_stop_compilation(self): + """A document-level warning is non-fatal; compilation still proceeds.""" + warning = ValidationIssue("warning", "heads up", "src") + pipeline, m = _pipeline(issues=[warning]) + + result = pipeline.compile() + + assert result == "compiled-schema" + m["compiler"].compile.assert_called_once() + + +class TestPlan: + """Cover SchemaPipeline.plan delegation to render + applier.plan.""" + + def test_delegates_to_renderer_and_applier_plan(self): + """Plan renders the compiled schema and delegates to ``applier.plan``.""" + pipeline, m = _pipeline() + + result = pipeline.plan() + + m["renderer"].render.assert_called_once_with("compiled-schema") + # The schema goes along with the rendered rows so the report can cover + # definition changes, not just policy rows (ADR 0018 §6). + m["applier"].plan.assert_called_once_with("rendered-policy", "compiled-schema") + assert result is m["applier"].plan.return_value + + def test_plan_does_not_apply(self): + """Plan is read-only: it never calls ``applier.apply``.""" + pipeline, m = _pipeline() + pipeline.plan() + m["applier"].apply.assert_not_called() + + +class TestApply: + """Cover SchemaPipeline.apply delegation and force forwarding.""" + + def test_delegates_to_applier_apply_without_force(self): + """Apply renders the schema and delegates to ``applier.apply`` with ``force=False``.""" + pipeline, m = _pipeline() + + result = pipeline.apply() + + m["renderer"].render.assert_called_once_with("compiled-schema") + m["applier"].apply.assert_called_once_with("rendered-policy", "compiled-schema", force=False) + assert result is m["applier"].apply.return_value + + def test_forwards_force_flag(self): + """The ``force`` flag is forwarded to ``applier.apply``.""" + pipeline, m = _pipeline() + pipeline.apply(force=True) + m["applier"].apply.assert_called_once_with("rendered-policy", "compiled-schema", force=True) + + +class TestDefaultComponents: + """Constructing a pipeline without injected components wires real defaults.""" + + def test_default_components_are_constructed_when_not_injected(self): + """A bare SchemaPipeline wires real default components (smoke test).""" + pipeline = SchemaPipeline() + # Internal defaults exist; we don't run them here (that needs real data), + # only assert the orchestrator is fully constructed. + # pylint: disable=protected-access + assert pipeline._discovery is not None + assert pipeline._loader is not None + assert pipeline._validator is not None + assert pipeline._compiler is not None + assert pipeline._renderer is not None + assert pipeline._applier is not None From 1e19c312c4838a51541b6909d581aeef8e04e446 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rodrigo=20M=C3=A9ndez?= Date: Tue, 13 Oct 2026 11:51:37 -0600 Subject: [PATCH 2/2] squash!: Ensure concurrent schema applies dont conflict with each other --- .../0018-authorization-schema-lifecycle.rst | 21 ++ src/openedx_authz/engine/policy.py | 12 + src/openedx_authz/engine/renderer.py | 127 ++++++-- src/openedx_authz/models/engine.py | 54 +++- .../tests/integration/test_schema_apply.py | 200 ++++++++++++- src/openedx_authz/tests/schema/test_apply.py | 272 +++++++++++++++++- src/openedx_authz/tests/test_models.py | 48 +++- 7 files changed, 699 insertions(+), 35 deletions(-) diff --git a/docs/decisions/0018-authorization-schema-lifecycle.rst b/docs/decisions/0018-authorization-schema-lifecycle.rst index 82f82148..ff88dc03 100644 --- a/docs/decisions/0018-authorization-schema-lifecycle.rst +++ b/docs/decisions/0018-authorization-schema-lifecycle.rst @@ -87,6 +87,26 @@ An explicit force option allows an operator to remove the role and its remaining Application and plugin code cannot edit generated static rows directly. During an incident, however, an administrator may use the Django admin, which records who changed what and why, and explains whether the next schema deployment will replace the change. +8. Serialize concurrent applies +=============================== + +Section 5 keeps the writes of a single apply atomic. This section covers what happens when two applies run at the same time — for example two deployments racing, or a deployment overlapping an administrative re-run. + +Apply is a read-modify-write cycle: it reads the current policy and definitions, computes the difference against the compiled schema, then writes the generated policy and the definition records. Two applies that run concurrently can each read the same starting state, each compute a difference against it, and then both write. The result is a lost update in which the generated policy and the definition records disagree — for instance the definitions retain a permission that the policy no longer grants — even though each apply individually did an atomic write. + +An apply therefore runs under a lock that lets only one apply proceed at a time. A second apply waits until the first finishes and then reads the state the first committed, so its difference is computed against current data rather than a stale snapshot. The lock is released automatically when the apply's transaction ends — on success, failure, or a lost connection — so a failed or interrupted apply cannot leave the lock held. + +Taking the lock is not sufficient on its own; the apply must also read current data once it holds the lock. Two things make that true. First, the apply reads committed database state under the lock rather than a snapshot captured before the lock was granted, because some databases fix a transaction's read view at its first read. Second, the apply refreshes any in-memory copy of the policy it keeps, so the difference is computed from what the database actually holds and not from a cached view a previous apply left behind. + +The signal that tells running services their cached policy is stale is updated inside the same transaction as the policy writes. If that signal were updated after the transaction, the new policy would be committed and visible while the signal still indicated the old version, and a service could serve the superseded policy for that window. Updating the signal together with the writes closes that window: a service either sees both the new policy and the new signal, or neither. + +This ADR records the decision to serialize applies and to read current state under the lock. It does not prescribe the specific locking primitive, isolation level, or signalling mechanism; those are implementation choices that depend on the deployment's database. + +Two limitations are accepted: + +* **Single-database assumption.** Serialization holds only when the generated policy and the lock live in the same database. The engine already assumes one database for its policy storage, so this is consistent with the rest of the system. +* **Blocking wait without observability.** A second apply waits for the first with no timeout and no distinct signal, so a blocked apply is operationally indistinguishable from a slow one. This is acceptable because apply runs at deployment time, not on the request path. + Consequences ************ @@ -97,6 +117,7 @@ Consequences * Applying a schema requires a database transaction or another mechanism that updates all generated rows together. * Operators receive the change report before the database is updated. * An assignment to a role being removed blocks deployment unless an operator explicitly removes the role and its assignments together. +* Concurrent applies are serialized so they cannot interleave into a lost update, and each apply reads current committed state under the lock and updates the cache-staleness signal together with its writes. References ********** diff --git a/src/openedx_authz/engine/policy.py b/src/openedx_authz/engine/policy.py index 3281e336..059bc49d 100644 --- a/src/openedx_authz/engine/policy.py +++ b/src/openedx_authz/engine/policy.py @@ -199,3 +199,15 @@ def remove_grouping_row(self, row: GroupingRow) -> bool: than a reconstructed one. """ return self._resolved_enforcer.remove_grouping_policy(*row.as_policy()) + + def reload(self) -> None: + """Reload the enforcer's in-memory model from the database (ADR 0018 §8). + + The schema applier calls this right after taking the serialization lock + so the subsequent ``get_policy_rows`` read reflects the committed ``p`` + rows the lock just made visible, rather than a prior run's version-gated + in-memory snapshot. Delegating through the store (instead of reaching for + the global enforcer) keeps the applier's "never hold the enforcer handle" + contract and lets injected fakes model the reload. + """ + self._resolved_enforcer.load_policy() diff --git a/src/openedx_authz/engine/renderer.py b/src/openedx_authz/engine/renderer.py index c2d1fbca..c580914f 100644 --- a/src/openedx_authz/engine/renderer.py +++ b/src/openedx_authz/engine/renderer.py @@ -1,4 +1,4 @@ -"""Render compiled definitions to Casbin rows and apply them (ADR 0018 §1, §5). +"""Render compiled definitions to Casbin rows and apply them (ADR 0018 §1, §5, §8). This is the only Casbin/Django-aware part of the schema pipeline. It implements the ``render`` and ``apply`` lifecycle steps: @@ -43,6 +43,7 @@ from __future__ import annotations import logging +from contextlib import contextmanager from dataclasses import dataclass, field from crum import get_current_user @@ -56,6 +57,7 @@ from openedx_authz.engine.schema.types import CompiledSchema, RoleDefinition from openedx_authz.models import schema as schema_models from openedx_authz.models.core import RoleAssignmentAudit +from openedx_authz.models.engine import PolicyCacheControl logger = logging.getLogger(__name__) @@ -233,32 +235,44 @@ def apply( invalidated so the enforcer reloads the last committed state rather than keeping the uncommitted in-memory rows (ADR 0018 §5). + Concurrency (ADR 0018 §8): the whole read-modify-write cycle runs inside + :meth:`_serialized_section`, which takes an exclusive row lock as the + transaction's first statement, reloads the enforcer, and — on a change — + bumps the policy cache version *inside* the same transaction. Two + concurrent ``load_authz_schema`` runs would otherwise each diff against + the same starting snapshot and both write, leaving the Casbin ``p`` rows + and the ``Authz*`` definition tables disagreeing (a lost update); the lock + serializes them so the second reads the already-applied state. Bumping the + version in-transaction also closes the reader stale window: without it the + policy writes commit before the version changes, so a concurrent enforcer + could serve stale in-memory policy against already-updated rows. + Raises: SchemaApplyError: If the plan has blocking assignments and ``force`` is False. """ - plan = self.plan(rendered, schema) - - if plan.blocking_assignments and not force: - details = ", ".join(f"{role} (assigned to {subject})" for role, subject in plan.blocking_assignments) - raise SchemaApplyError( - "Refusing to proceed: static roles with existing assignments would be removed: " - f"{details}. Re-run with force to remove them together with their assignments." - ) - policy_store = self._policy_store - # Reconcile p rows and sync definition/source records atomically. - # Definitions are synced even when p rows are unchanged so metadata-only - # edits land and pre-existing p rows get adopted on first run. - # - # add_policy_row/remove_policy_row mutate the enforcer's in-memory model - # as well as the database, so a rollback would otherwise leave this - # process enforcing rows the database no longer has. Bumping the policy - # cache version on the failure path forces a reload from the committed - # state, keeping Casbin on the last working version (ADR 0018 §5). try: - with transaction.atomic(): + with self._serialized_section(): + plan = self.plan(rendered, schema) + + # Invariant: this gate must stay *before* any add_policy_row/ + # remove_policy_row below. SchemaApplyError is caught separately + # and re-raised without invalidating the policy cache, which is + # only safe while nothing has mutated the in-memory enforcer. If a + # future change needs to raise SchemaApplyError after a write, it + # must invalidate the cache (like the generic except below) so the + # process does not keep enforcing rolled-back rows. + if plan.blocking_assignments and not force: + details = ", ".join( + f"{role} (assigned to {subject})" for role, subject in plan.blocking_assignments + ) + raise SchemaApplyError( + "Refusing to proceed: static roles with existing assignments would be removed: " + f"{details}. Re-run with force to remove them together with their assignments." + ) + for row in plan.added_rows: policy_store.add_policy_row(row) for row in plan.removed_rows: @@ -267,20 +281,39 @@ def apply( if force and plan.blocking_assignments: removed_assignments = self._remove_assignments(policy_store, plan.blocking_assignments) self._store_sources(schema) + + changed = bool(plan.added_rows or plan.removed_rows) + if changed: + # Bump the cache version *inside* the transaction so it commits + # atomically with the p-row writes. If it ran after the block + # (as it once did), the rows would be committed-and-visible + # while the version still read old, and a concurrent enforcer + # would skip the reload and serve stale policy for that window. + AuthzEnforcer.invalidate_policy_cache() + # Emit the audit events only if the transaction commits, mirroring # unassign_role_from_subject_in_scope, so no audit row is written for # an assignment removal that gets rolled back. if removed_assignments: transaction.on_commit(lambda: self._emit_assignment_deleted(removed_assignments)) + except SchemaApplyError: + # The force gate fired before any write (see the invariant above): + # nothing was mutated and the cache does not need bumping. Re-raise + # without the failure handling below, which exists for partial-write + # rollbacks. + raise except Exception: - # Runs outside the rolled-back block, so the new version commits. + # The transaction rolled back the database, but add_policy_row/ + # remove_policy_row already mutated the enforcer's in-memory model, so + # this process would keep enforcing rows the database no longer has. + # The in-transaction bump above was rolled back too, so bump again + # here — outside the rolled-back block, where it commits — to force a + # reload from the committed (pre-apply) state (ADR 0018 §5). AuthzEnforcer.invalidate_policy_cache() logger.exception("Authz schema apply failed; policy cache invalidated to force a reload.") raise - changed = bool(plan.added_rows or plan.removed_rows) - if changed: - AuthzEnforcer.invalidate_policy_cache() + if plan.added_rows or plan.removed_rows: logger.info( "Authz schema apply: added %d p row(s), removed %d p row(s).", len(plan.added_rows), @@ -296,6 +329,52 @@ def apply( plan=plan, ) + @contextmanager + def _serialized_section(self): + """Open the serialized critical section for an apply (ADR 0018 §8). + + Owns the concurrency contract so :meth:`apply` reads as plain + read-diff-write. On entry it, in order: + + 1. ``ensure_exists`` the lock row **before** the transaction opens — a + creating ``SELECT``/INSERT inside the transaction would be its first + read and, under InnoDB Repeatable Read, pin the consistent-read + snapshot before the lock is taken, leaving every post-lock read stale. + 2. opens ``transaction.atomic()``; + 3. takes the exclusive row lock as the transaction's first read, so a + second runner blocks until the first commits and this runner's reads + see the committed state; + 4. reloads the enforcer, because the lock only makes the *database* + current — the version-gated in-memory model would otherwise still hold + a prior run's rows, so the ``plan`` read of ``get_policy`` could be + stale even with the lock held. + + The caller does its read-diff-write inside the ``with`` block. The lock + (a row lock) releases automatically when the transaction commits, rolls + back, or the connection dies, so no stale lock can outlive a crashed + apply. Keeping the lock, reload, and the cache-version bump the caller + performs all within this one transaction is what makes the critical + section cohesive and leaves no committed-but-unversioned reader window. + + Two accepted limitations, deferred (ADR 0018 §8): + + * ``PolicyCacheControl`` (the lock) and the ``transaction.atomic()`` + default to the ``default`` database, while the Casbin ``p`` rows live + in ``CASBIN_DB_ALIAS``. Serialization is only meaningful while those + coincide, which matches the engine's existing single-database policy + storage assumption. If ``CASBIN_DB_ALIAS`` is ever pointed at a + separate database, the lock must move there too. + * The lock wait is unbounded and silent: a second apply blocks with no + ``nowait``/timeout and no log line, so a blocked apply looks like a slow + one. Acceptable because apply runs at deploy time, not on the request + path. + """ + PolicyCacheControl.ensure_exists() + with transaction.atomic(): + PolicyCacheControl.lock() + self._policy_store.reload() + yield + @staticmethod def _remove_assignments( policy_store: PolicyStore, blocking_assignments: list[tuple[str, str]] diff --git a/src/openedx_authz/models/engine.py b/src/openedx_authz/models/engine.py index cc6f8bc4..978ffd15 100644 --- a/src/openedx_authz/models/engine.py +++ b/src/openedx_authz/models/engine.py @@ -6,11 +6,19 @@ class PolicyCacheControl(models.Model): - """Model to control policy cache invalidation. + """Singleton row that both versions and guards the authorization policy. - This model can be used to trigger cache invalidation for authorization policies - by changing the version. Whenever this model is updated, the authorization - engine should invalidate its cached policies. + Its ``version`` is the cache-invalidation signal: whenever the stored policy + changes, the version is bumped and every enforcer reloads on its next check + (see :meth:`AuthzEnforcer.load_policy_if_needed`). + + Because this one row is also what every reader polls, it doubles as the + natural serialization point for writers: :meth:`lock` takes an exclusive row + lock on it so concurrent schema applies run one at a time. The lock lives + here only as a thin primitive over this row; the *orchestration* of the + critical section (ordering, enforcer reload, bumping the version inside the + writer's transaction) belongs to the applier that owns the write, not to this + model (see ``SchemaApplier._serialized_section``). .. no_pii: """ @@ -28,6 +36,44 @@ def get(cls): obj, _ = cls.objects.get_or_create(pk=1) return obj + @classmethod + def ensure_exists(cls): + """Create the singleton row if missing. Call *outside* the apply transaction. + + ``select_for_update`` cannot lock a row that does not exist, so the row + has to be there before :meth:`lock` runs. This must be called before the + apply ``transaction.atomic()`` block opens, not inside it: a plain + ``SELECT``/INSERT inside that transaction would be its first read and, + under InnoDB Repeatable Read, would pin the consistent-read snapshot + *before* :meth:`lock` blocks — defeating the lock, because the post-lock + reads would still see the pre-lock snapshot. Keeping row creation out of + the apply transaction leaves the locking read in :meth:`lock` as the + genuinely first read of that transaction. + """ + cls.objects.get_or_create(pk=1) + + @classmethod + def lock(cls): + """Take an exclusive row lock on the singleton (``SELECT ... FOR UPDATE``). + + A thin serialization primitive: it blocks until any other transaction + holding the lock commits or rolls back, then returns the row. It does not + manage its own transaction or ordering — the caller owns both. For the + lock to actually serialize a read-modify-write cycle it must be taken + inside a ``transaction.atomic()`` block, as that transaction's first read + (so the Repeatable Read snapshot forms after the lock is granted), and + held across the reads and writes. ``SchemaApplier._serialized_section`` + is the orchestrator that satisfies those requirements; this method only + provides the lock itself. + + :meth:`ensure_exists` must have run and committed first, since + ``select_for_update`` cannot lock a row that does not exist yet. + + Returns: + PolicyCacheControl: The locked singleton instance. + """ + return cls.objects.select_for_update().get(pk=1) + @classmethod def get_version(cls): """Get the version for policy cache control. diff --git a/src/openedx_authz/tests/integration/test_schema_apply.py b/src/openedx_authz/tests/integration/test_schema_apply.py index b06c607e..5f99c06a 100644 --- a/src/openedx_authz/tests/integration/test_schema_apply.py +++ b/src/openedx_authz/tests/integration/test_schema_apply.py @@ -1,4 +1,4 @@ -"""End-to-end integration tests for schema apply pruning (ADR 0018 §2, §5, §6). +"""End-to-end integration tests for schema apply pruning (ADR 0018 §2, §5, §6, §8). Unlike the unit tests in ``tests/schema/test_apply.py`` (which use a fake enforcer and stub persistence), these exercise the *real* stack: @@ -38,11 +38,13 @@ from __future__ import annotations +import threading +import time from unittest import mock import pytest -from django.db import IntegrityError -from django.test import TestCase +from django.db import IntegrityError, connection, connections +from django.test import TestCase, TransactionTestCase from openedx_authz.engine.enforcer import AuthzEnforcer from openedx_authz.engine.renderer import PolicyRenderer, SchemaApplier @@ -444,3 +446,195 @@ def test_failed_apply_leaves_enforcement_on_the_committed_state(self): reloaded = AuthzEnforcer.get_enforcer() self.assertEqual([row for row in reloaded.get_policy() if row[0] == ROLE_SUBJECT], []) self.assertFalse(reloaded.enforce(USER_SUBJECT, VIEW_ACTION, COURSE_SCOPE)) + + +class TestSchemaApplyConcurrencyIntegration(TransactionTestCase): + """Concurrent applies serialize so the two stores never diverge (ADR 0018 §8, lost-update fix). + + Regression for a lost update between two overlapping applies. When an + old-schema apply and an updated-schema apply ran at the same time, each + computed its diff against the same pre-change snapshot and then both wrote, + leaving the two stores inconsistent: the definition tables kept *both* + permissions while the Casbin policy kept only the updated one. Apply now + takes an exclusive row lock (``SELECT ... FOR UPDATE`` on the + ``PolicyCacheControl`` singleton) as the first statement in its transaction, + so a second runner blocks until the first commits and then reconciles against + the already-applied state. + + This runs on a real stack across two database connections, so it needs + ``TransactionTestCase``: each apply must truly commit for the other + connection to see it, which the per-test transaction of ``TestCase`` would + prevent. ``select_for_update`` is a no-op on SQLite (and SQLite serializes + writers at the file level anyway), so the test is skipped there — it is + meaningful only on a backend with real row locking (the MySQL environment + these integration tests target). + + Scope: this proves serialization and the consistent end state of a real + overlapping race. The individual mechanisms the fix relies on — the lock row + being created before the transaction opens (fresh Repeatable Read snapshot) + and the enforcer being reloaded under the lock (fresh in-memory p rows) — are + asserted deterministically in ``TestApplySerialization`` in + ``tests/schema/test_apply.py``, which does not depend on backend-specific MVCC + timing. + """ + + def setUp(self): + """Start from a clean, committed policy with one permission applied.""" + if not connection.features.has_select_for_update: + self.skipTest("Backend does not support SELECT ... FOR UPDATE row locking.") + + self.enforcer = AuthzEnforcer.get_enforcer() + self.enforcer.clear_policy() + # Seed the starting state both overlapping applies will diff against: + # the role granting only view_course. + self._apply_sync(_document(roles=[_editor_role(("courses.view_course",))])) + + def tearDown(self): + """Leave no policy behind for other integration tests.""" + self.enforcer.clear_policy() + + # -- helpers ------------------------------------------------------------ + + @staticmethod + def _apply_sync(*documents, force=False): + """Compile, render, and apply ``documents`` synchronously (no pause).""" + schema = SchemaCompiler().compile(list(documents)) + rendered = PolicyRenderer().render(schema) + return SchemaApplier().apply(rendered, schema, force=force) + + def _p_actions_for_role(self): + """Return the set of actions the test role grants in the stored policy.""" + self.enforcer.load_policy() + return {row[1] for row in self.enforcer.get_policy() if row[0] == ROLE_SUBJECT} + + def _definition_actions_for_role(self): + """Return the set of permission names the role's definition grants.""" + return { + f"act^{perm.permission.namespace}.{perm.permission.name}" + for perm in AuthzRolePermission.objects.select_related("permission").filter( + role__role_id="schema_apply_editor" + ) + } + + def test_overlapping_applies_serialize_and_leave_consistent_stores(self): + """A stale apply overlapping an updated apply ends with both stores agreeing. + + This drives the exact window that produced the lost update. The seed + grants ``view_course``. Thread A applies a schema that *drops* + ``view_course`` and grants ``manage_tags`` — a real change — and pauses + with its write done but its transaction *not yet committed*, still + holding the lock. Thread B then applies a schema granting *both* + ``view_course`` and ``manage_tags`` while A holds the lock, so B contends + for the lock before A commits. + + The point is what B must observe once it proceeds: A has removed + ``view_course``, so B's reconciliation has to re-add it on top of A's + committed state. B reads the ``p`` rows (via the enforcer) and the + definition tables (via plain ``SELECT``). Under InnoDB Repeatable Read, a + read taken before the lock blocks would pin B's snapshot to the pre-A + state (``view_course`` present, ``manage_tags`` absent); its diff would + then be computed against stale data and the two stores would diverge. + Taking the locking read first and reloading the enforcer under the lock + makes B see A's commit, so both stores end up granting exactly + ``{view_course, manage_tags}`` and agree with each other. + """ + a_schema = _document(roles=[_editor_role(("courses.manage_tags",))]) + b_schema = _document(roles=[_editor_role(("courses.view_course", "courses.manage_tags"))]) + + order, errors, release_fired = self._run_overlapping_applies(a_schema, b_schema) + + self.assertEqual(errors, {}, f"A worker raised: {errors}") + self.assertTrue(release_fired, "Thread A never reached its release point") + + # Serialization: B committed only after A committed. + self.assertIn("a_committed", order) + self.assertIn("b_committed", order) + self.assertLess( + order.index("a_committed"), + order.index("b_committed"), + f"B must commit after A; observed order: {order}", + ) + + # Snapshot freshness + cross-store consistency: B reconciled against A's + # committed state (view_course removed), so B re-added it on top. Both + # the Casbin policy and the definition tables end up granting exactly + # {view_course, manage_tags} and agree with each other. A stale pre-A + # snapshot in B would leave the stores disagreeing. + expected = {VIEW_ACTION, TAGS_ACTION} + self.assertEqual(self._p_actions_for_role(), expected) + self.assertEqual(self._definition_actions_for_role(), expected) + self.assertEqual(self._p_actions_for_role(), self._definition_actions_for_role()) + + def _run_overlapping_applies(self, a_schema, b_schema): + """Run two applies that overlap on the lock and return their observed timeline. + + Thread A applies ``a_schema`` but pauses after its writes and before + commit, still inside the locked transaction (``_store_sources`` is the + last write in ``apply``, so holding there holds the lock with A's change + written-but-uncommitted). Thread B applies ``b_schema`` and is released + only once A is parked mid-write, so B is forced to contend for the lock + while A's change is still uncommitted — the exact window the fix guards. + + Returns: + tuple: ``(order, errors, release_fired)`` where ``order`` is the list + of timeline markers, ``errors`` maps worker name to any raised + exception, and ``release_fired`` is whether A reached its release + point (a guard against a silently stuck A). + """ + a_holds_lock_mid_write = threading.Event() + release_a = threading.Event() + b_contending = threading.Event() + order: list[str] = [] + order_lock = threading.Lock() + errors: dict[str, BaseException] = {} + + def mark(event: str) -> None: + with order_lock: + order.append(event) + + applier_a = SchemaApplier() + real_store_sources = applier_a._store_sources # pylint: disable=protected-access + + def paused_store_sources(schema): + real_store_sources(schema) + mark("a_wrote") + a_holds_lock_mid_write.set() + # Wait until B is actively contending for the lock, then hold a touch + # longer so B is parked on the lock (not merely about to ask for it) + # before A commits. Bounded so a bug cannot hang the suite. + b_contending.wait(timeout=10) + time.sleep(1) + release_a.set() + + def run_a(): + try: + with mock.patch.object(applier_a, "_store_sources", side_effect=paused_store_sources): + schema = SchemaCompiler().compile([a_schema]) + applier_a.apply(PolicyRenderer().render(schema), schema) + mark("a_committed") + except BaseException as exc: + errors["a"] = exc + finally: + connections.close_all() + + def run_b(): + try: + a_holds_lock_mid_write.wait(timeout=10) + b_contending.set() + self._apply_sync(b_schema) + mark("b_committed") + except BaseException as exc: + errors["b"] = exc + finally: + connections.close_all() + + thread_a = threading.Thread(name="apply-A", target=run_a) + thread_b = threading.Thread(name="apply-B", target=run_b) + thread_a.start() + thread_b.start() + thread_a.join(timeout=25) + thread_b.join(timeout=25) + + self.assertFalse(thread_a.is_alive(), "Thread A did not finish") + self.assertFalse(thread_b.is_alive(), "Thread B did not finish") + return order, errors, release_a.is_set() diff --git a/src/openedx_authz/tests/schema/test_apply.py b/src/openedx_authz/tests/schema/test_apply.py index 04411a4b..423d9023 100644 --- a/src/openedx_authz/tests/schema/test_apply.py +++ b/src/openedx_authz/tests/schema/test_apply.py @@ -1,4 +1,4 @@ -"""Tests for the apply/plan reconciliation path (ADR 0018 §2, §5, §6). +"""Tests for the apply/plan reconciliation path (ADR 0018 §2, §5, §6, §8). Three layers are exercised: @@ -18,13 +18,14 @@ from unittest import mock import pytest -from django.db import IntegrityError +from django.db import IntegrityError, transaction from django.test import TestCase from openedx_authz.engine.policy import PolicyStore from openedx_authz.engine.renderer import PolicyRenderer, SchemaApplier from openedx_authz.engine.schema.compilation import SchemaCompiler from openedx_authz.engine.schema.exceptions import SchemaApplyError +from openedx_authz.models.engine import PolicyCacheControl from openedx_authz.models.schema import ( AuthzPermissionCategory, AuthzPermissionDefinition, @@ -63,6 +64,9 @@ def __init__(self, policies=None, grouping=None): self._policies = [list(row) for row in (policies or [])] self._grouping = [list(row) for row in (grouping or [])] + def load_policy(self): + """No-op reload hook: the fake is already its own in-memory source of truth.""" + def get_policy(self): """Return a copy of the stored ``p`` rows.""" return [list(row) for row in self._policies] @@ -304,15 +308,23 @@ def _assigned_enforcer(self): enforcer.add_grouping_policy("user^alice", "role^course_editor", "course-v1:OpenedX+DemoX+Demo") return enforcer - def test_blocking_assignment_aborts_without_force(self): + def test_blocking_assignment_aborts_without_force(self, cache_invalidation): """Removing a role with a live assignment aborts without ``force``.""" enforcer = self._assigned_enforcer() + # Seeding the role in _assigned_enforcer already bumped the cache; only + # the aborted apply below should be checked, so start from a clean mock. + cache_invalidation.reset_mock() with pytest.raises(SchemaApplyError): _apply(enforcer, force=False) # No write happened: the p row is still there. assert len(enforcer.get_policy()) == 1 + # The force gate raises SchemaApplyError before any mutation, and that + # branch deliberately skips the cache bump — nothing changed, so forcing + # a reload would be pointless churn. Pin that so a future change that + # starts invalidating on the gate path is caught. + cache_invalidation.assert_not_called() def test_force_removes_role_rows_and_assignments(self): """With ``force``, the removed role's policy and assignment rows are pruned.""" @@ -581,3 +593,257 @@ def test_plan_does_not_write(self): assert enforcer.get_policy() == [] assert AuthzRoleDefinition.objects.count() == 0 + + +@pytest.mark.django_db +@pytest.mark.usefixtures("cache_invalidation") +class TestApplySerialization: + """Apply serializes concurrent runs with an exclusive lock (ADR 0018 §8, lost-update fix). + + Two concurrent ``load_authz_schema`` runs used to each diff against the same + starting snapshot and both write, leaving the Casbin ``p`` rows and the + definition tables disagreeing. Apply now takes ``PolicyCacheControl.lock`` + (a ``SELECT ... FOR UPDATE`` on the singleton row) as the first statement + inside its transaction, so a second runner blocks until the first commits + and then reads the already-applied state. + + These are deterministic, single-process assertions on the *mechanism* the + fix depends on — complementing the real two-connection race in + ``tests/integration/test_schema_apply.py``. They assert the lock is taken, + that it is taken before any state is read and held inside the transaction, + that the lock row is created before the transaction opens (so the locking + read is the transaction's first read, which is what keeps the Repeatable + Read snapshot fresh), and that the enforcer is reloaded under the lock before + the p-row read (so the in-memory model cannot leave the read stale). + """ + + def test_apply_takes_the_lock(self): + """A normal apply acquires the serialization lock exactly once.""" + enforcer = FakeEnforcer() + + with mock.patch( + "openedx_authz.engine.renderer.PolicyCacheControl.lock", + wraps=PolicyCacheControl.lock, + ) as lock: + _apply(enforcer, _editor(("courses.view_course",))) + + lock.assert_called_once_with() + + def test_lock_is_acquired_before_state_is_read(self): + """The lock is taken before plan() reads the stored policy. + + Reading first and locking second would reintroduce the race: the diff + would be computed against a snapshot another apply could change before + this one writes. The order is the whole point of the fix. + """ + enforcer = FakeEnforcer() + calls: list[str] = [] + real_lock = PolicyCacheControl.lock + + def record_lock(): + calls.append("lock") + return real_lock() + + original_get_policy = enforcer.get_policy + + def record_read(): + calls.append("read") + return original_get_policy() + + with ( + mock.patch( + "openedx_authz.engine.renderer.PolicyCacheControl.lock", + side_effect=record_lock, + ), + mock.patch.object(enforcer, "get_policy", side_effect=record_read), + ): + _apply(enforcer, _editor(("courses.view_course",))) + + assert calls[0] == "lock" + assert "read" in calls + + def test_lock_is_acquired_inside_a_transaction(self): + """The lock is taken while a transaction is open, so it serializes writers. + + A row lock only holds for the life of its transaction; if apply took it + outside ``transaction.atomic()`` it would release immediately and serialize + nothing. + """ + enforcer = FakeEnforcer() + in_atomic_block: list[bool] = [] + real_lock = PolicyCacheControl.lock + + def record_lock(): + in_atomic_block.append(transaction.get_connection().in_atomic_block) + return real_lock() + + with mock.patch( + "openedx_authz.engine.renderer.PolicyCacheControl.lock", + side_effect=record_lock, + ): + _apply(enforcer, _editor(("courses.view_course",))) + + assert in_atomic_block == [True] + + def test_lock_row_is_created_before_the_transaction_opens(self): + """ensure_exists() runs outside the apply transaction, before the lock. + + Creating the lock row inside the apply transaction would make its + SELECT/INSERT the transaction's first read and, under InnoDB Repeatable + Read, pin the consistent-read snapshot before the lock is taken — the + post-lock reads would then see stale data. ensure_exists() must therefore + run (and commit) *before* ``transaction.atomic()`` opens, so the locking + read in lock() is the transaction's genuinely first read. + """ + enforcer = FakeEnforcer() + events: list[tuple[str, int]] = [] + real_lock = PolicyCacheControl.lock + real_ensure = PolicyCacheControl.ensure_exists + + def atomic_depth() -> int: + # Number of open atomic blocks on this connection. The test harness + # wraps the test in one, so apply() opening its own block increments + # this; comparing the depth at ensure_exists vs lock proves apply's + # transaction opened *between* them. + return len(transaction.get_connection().atomic_blocks) + + def record_ensure(): + events.append(("ensure_exists", atomic_depth())) + return real_ensure() + + def record_lock(): + events.append(("lock", atomic_depth())) + return real_lock() + + with ( + mock.patch( + "openedx_authz.engine.renderer.PolicyCacheControl.ensure_exists", + side_effect=record_ensure, + ), + mock.patch( + "openedx_authz.engine.renderer.PolicyCacheControl.lock", + side_effect=record_lock, + ), + ): + _apply(enforcer, _editor(("courses.view_course",))) + + # ensure_exists runs first; apply then opens its own atomic block before + # taking the lock, so lock sees a strictly deeper nesting. That ordering + # is what keeps the lock creation out of the snapshot-pinning transaction. + assert [name for name, _ in events] == ["ensure_exists", "lock"] + ensure_depth = events[0][1] + lock_depth = events[1][1] + assert lock_depth > ensure_depth + + def test_enforcer_is_reloaded_under_the_lock_before_state_is_read(self): + """apply() reloads the enforcer after the lock and before plan() reads p rows. + + The lock makes the *database* current, but the enforcer's in-memory model + is version-gated and the version bump happens after the transaction, so + without an explicit reload plan() could diff against a prior run's + in-memory rows. The reload must land after the lock (so it reads the + committed state the lock exposed) and before the p-row read in plan(). + """ + enforcer = FakeEnforcer() + calls: list[str] = [] + real_lock = PolicyCacheControl.lock + + def record_lock(): + calls.append("lock") + return real_lock() + + original_load = enforcer.load_policy + original_get_policy = enforcer.get_policy + + def record_reload(): + calls.append("reload") + return original_load() + + def record_get_policy(): + calls.append("get_policy") + return original_get_policy() + + with ( + mock.patch( + "openedx_authz.engine.renderer.PolicyCacheControl.lock", + side_effect=record_lock, + ), + mock.patch.object(enforcer, "load_policy", side_effect=record_reload), + mock.patch.object(enforcer, "get_policy", side_effect=record_get_policy), + ): + _apply(enforcer, _editor(("courses.view_course",))) + + # Order: take the lock, reload the enforcer, then read p rows in plan(). + assert calls.index("lock") < calls.index("reload") + assert calls.index("reload") < calls.index("get_policy") + + @staticmethod + def _atomic_depth() -> int: + """Open atomic-block count on the default connection. + + The test harness wraps each test in one atomic block, so apply() opening + its own increments this. Comparing the depth at a point of interest + against the baseline tells whether that point ran inside apply's + transaction (deeper) or after it closed (back at baseline). + """ + return len(transaction.get_connection().atomic_blocks) + + def test_cache_version_bump_commits_inside_the_transaction(self): + """On a change, the version bump runs inside the apply transaction. + + If it ran after the transaction (as it once did), the policy rows would + be committed and visible while the version still read old, so a + concurrent enforcer would skip its reload and serve stale policy for that + window. Bumping inside the transaction makes the version change commit + atomically with the rows, closing that reader window. + """ + enforcer = FakeEnforcer() + baseline = self._atomic_depth() + depth_at_bump: list[int] = [] + + def record_invalidate(): + depth_at_bump.append(self._atomic_depth()) + + with mock.patch( + "openedx_authz.engine.renderer.AuthzEnforcer.invalidate_policy_cache", + side_effect=record_invalidate, + ): + _apply(enforcer, _editor(("courses.view_course",))) + + # Exactly one bump on the success path, at a deeper nesting than the + # baseline — i.e. inside apply's own still-open transaction. + assert len(depth_at_bump) == 1 + assert depth_at_bump[0] > baseline + + def test_failed_apply_bumps_cache_version_after_rollback(self): + """On failure the bump runs *outside* the rolled-back transaction. + + A bump inside the transaction would roll back with it, leaving the + process enforcing the in-memory rows the database discarded. The failure + path must bump after rollback so the new version actually commits and the + next enforcer access reloads the committed (pre-apply) state. + """ + enforcer = FakeEnforcer() + baseline = self._atomic_depth() + depth_at_bump: list[int] = [] + + def failing_store(_self, _schema): + raise IntegrityError("simulated write failure") + + def record_invalidate(): + depth_at_bump.append(self._atomic_depth()) + + with ( + mock.patch.object(SchemaApplier, "_store_sources", failing_store), + mock.patch( + "openedx_authz.engine.renderer.AuthzEnforcer.invalidate_policy_cache", + side_effect=record_invalidate, + ), + pytest.raises(IntegrityError), + ): + _apply(enforcer, _editor(("courses.view_course",))) + + # One bump, back at the baseline nesting — apply's transaction had already + # closed (rolled back), so the bump commits rather than rolling back too. + assert len(depth_at_bump) == 1 + assert depth_at_bump[0] == baseline diff --git a/src/openedx_authz/tests/test_models.py b/src/openedx_authz/tests/test_models.py index ad0621e0..f8aefdd3 100644 --- a/src/openedx_authz/tests/test_models.py +++ b/src/openedx_authz/tests/test_models.py @@ -19,7 +19,7 @@ from casbin_adapter.models import CasbinRule from django.contrib.auth import get_user_model -from django.db import IntegrityError +from django.db import IntegrityError, transaction from django.test import TestCase from opaque_keys.edx.keys import CourseKey from opaque_keys.edx.locator import LibraryLocatorV2 @@ -374,6 +374,52 @@ def test_singleton_behavior(self): all_instances = PolicyCacheControl.objects.all() self.assertEqual(all_instances.count(), 1) + def test_lock_returns_the_singleton_inside_a_transaction(self): + """lock() returns the singleton row so an apply can serialize on it. + + Expected Result: + - lock() returns the pk=1 singleton instance. + - Calling it does not create a second row. + """ + PolicyCacheControl.ensure_exists() + + with transaction.atomic(): + locked = PolicyCacheControl.lock() + + self.assertEqual(locked.pk, 1) + self.assertEqual(PolicyCacheControl.objects.count(), 1) + + def test_ensure_exists_creates_the_singleton_when_missing(self): + """ensure_exists() creates the lock row so lock() has a row to lock. + + Row creation is separated from lock() because, inside the apply + transaction, a creating SELECT/INSERT would pin the Repeatable Read + snapshot before the lock is taken. ensure_exists() runs beforehand. + + Expected Result: + - With no row present, ensure_exists() creates exactly one pk=1 row. + - A subsequent lock() returns that singleton. + """ + PolicyCacheControl.objects.all().delete() + + PolicyCacheControl.ensure_exists() + with transaction.atomic(): + locked = PolicyCacheControl.lock() + + self.assertEqual(locked.pk, 1) + self.assertEqual(PolicyCacheControl.objects.count(), 1) + + def test_ensure_exists_is_idempotent(self): + """ensure_exists() does not create duplicates when the row already exists. + + Expected Result: + - Calling ensure_exists() twice leaves exactly one row. + """ + PolicyCacheControl.ensure_exists() + PolicyCacheControl.ensure_exists() + + self.assertEqual(PolicyCacheControl.objects.count(), 1) + class TestAuthzCourseAuthoringMigrationRun(TestCase): """Tests for ``AuthzCourseAuthoringMigrationRun`` lifecycle and uniqueness rules."""