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 ed30cde4..b52ee958 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`. @@ -111,7 +113,13 @@ def _resolve_distribution(cls, resource: DiscoveredResource) -> tuple[str, str]: # pylint: disable=broad-exception-caught except Exception: # 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 @@ -131,7 +139,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] @@ -142,16 +151,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..74b67d01 --- /dev/null +++ b/src/openedx_authz/management/commands/load_authz_schema.py @@ -0,0 +1,209 @@ +"""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. + """ + 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_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..9894a002 --- /dev/null +++ b/src/openedx_authz/tests/schema/test_load_authz_schema_command.py @@ -0,0 +1,411 @@ +"""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_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_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