diff --git a/src/openedx_authz/engine/schema/validation.py b/src/openedx_authz/engine/schema/validation.py new file mode 100644 index 00000000..daf4d222 --- /dev/null +++ b/src/openedx_authz/engine/schema/validation.py @@ -0,0 +1,418 @@ +"""Validate schema documents individually and as a whole (the ``validate`` step). + +Rules come from ADR 0017 §4 and the field reference: + +Per-document checks: + * ``schema_version`` is a supported, quoted ``major.minor`` value. + * ``namespace``, ``name``, category ``id``, role ``id`` match + :data:`IDENTIFIER_RE` (lowercase snake_case, begins with a letter). + * Casbin forms (``act^...``, ``role^...``) are rejected as identifiers. + * Required fields are present. + * ``scopes`` are non-empty and look like scope namespaces (hyphens allowed, + e.g. ``course-v1``); they are exempt from the identifier regex. + +Whole-set checks (after all documents load): + * Every permission ``category`` references an existing category. + * Every role/extension permission references an existing permission. + * A role's ``scopes`` are supported by each of its permissions. + * ``role_extensions`` target an existing role (ADR 0023). + * Conflicting duplicate base definitions fail; identical duplicates warn. + +Post-compile checks (after extensions and priority are resolved): + * A role's ``scopes`` are supported by every permission it *ends up* with, + including permissions contributed by ``role_extensions``. + * The resolved permissions and categories still exist. + +Validation collects issues rather than raising on the first problem, so the +deployment report can list every error and warning at once. No Casbin/Django +imports. +""" + +from __future__ import annotations + +import re +from collections.abc import Container +from dataclasses import dataclass +from enum import StrEnum + +from openedx_authz.constants import SchemaOriginKind +from openedx_authz.engine.schema.types import ( + CompiledSchema, + SchemaDocument, +) + +_IDENTIFIER = r"[a-z][a-z0-9_]*" +IDENTIFIER_RE = re.compile(rf"^{_IDENTIFIER}$") +# A complete permission ID joins ``namespace`` and ``name`` with a single +# period, each part being a bare identifier. Mirrors the ``permission_id`` +# ``$def`` in ``schema/authz-schema-v1.json``. +PERMISSION_ID_RE = re.compile(rf"^{_IDENTIFIER}\.{_IDENTIFIER}$") +# Scope namespaces follow their registered spelling and may contain hyphens. +SCOPE_RE = re.compile(r"^[a-z][a-z0-9_-]*$") + +# Casbin-internal prefixes that must never appear in a schema identifier. +CASBIN_INTERNAL_PREFIXES = ("act^", "role^", "sub^", "scope^", "g^", "p^") + + +class IssueLevel(StrEnum): + """Severity of a :class:`ValidationIssue`. + + Members: + ERROR: Blocks deployment. + WARNING: Reported only, does not block deployment. + """ + + ERROR = "error" + WARNING = "warning" + + +@dataclass(frozen=True) +class ValidationIssue: + """A single validation finding. + + Attributes: + level: :attr:`IssueLevel.ERROR` (blocks deployment) or + :attr:`IssueLevel.WARNING` (reported only). + message: Human-readable description. + source_id: The contributing source, when the issue is file-specific. + """ + + level: IssueLevel + message: str + source_id: str | None = None + + @property + def is_error(self) -> bool: + return self.level == IssueLevel.ERROR + + +class SchemaValidator: + """Runs per-document and whole-set validation.""" + + SUPPORTED_SCHEMA_VERSIONS = frozenset({"1.0"}) + + def validate(self, documents: list[SchemaDocument]) -> list[ValidationIssue]: + """Run per-document then whole-set validation, returning all issues.""" + issues: list[ValidationIssue] = [] + for document in documents: + issues.extend(self.validate_document(document)) + issues.extend(self.validate_set(documents)) + return issues + + def validate_compiled(self, schema: CompiledSchema) -> list[ValidationIssue]: + """Re-check the resolved schema after extensions and priority are applied. + + Document-level validation only sees base declarations, so a + ``role_extension`` that adds a permission is checked for existence but + never for scope compatibility. Without this pass an extension can grant + a permission in a scope the permission does not support, and the + renderer still emits an enforceable ``p`` row for it (ADR 0017 §4, + ADR 0023 §3). + + The compiled role's permission set is the one that becomes Casbin ``p`` + rows, so it is the set that has to satisfy the scope rule. + """ + issues: list[ValidationIssue] = [] + + for role_id, compiled_role in schema.roles.items(): + role = compiled_role.definition + for perm_id in role.permissions: + compiled_permission = schema.permissions.get(perm_id) + permission_scopes = compiled_permission.definition.scopes if compiled_permission else None + issues.extend( + self._check_role_permission_scope( + role_id, + perm_id, + role.scopes, + permission_scopes, + self._role_permission_source_id(schema, role_id, perm_id), + missing_verb="resolves to", + ) + ) + + for perm_id, compiled_permission in schema.permissions.items(): + sources = compiled_permission.sources + issues.extend( + self._check_permission_category( + perm_id, + compiled_permission.definition.category_id, + schema.categories, + sources[0].source_id if sources else None, + ) + ) + + return issues + + @staticmethod + def has_errors(issues: list[ValidationIssue]) -> bool: + """True if any issue is error-level.""" + return any(issue.is_error for issue in issues) + + def validate_document(self, document: SchemaDocument) -> list[ValidationIssue]: + """Per-file checks that need no cross-file context.""" + issues: list[ValidationIssue] = [] + sid = document.source.source_id + + if document.source.schema_version not in self.SUPPORTED_SCHEMA_VERSIONS: + issues.append( + ValidationIssue( + IssueLevel.ERROR, + f"Unsupported schema_version {document.source.schema_version!r}; " + f"supported: {sorted(self.SUPPORTED_SCHEMA_VERSIONS)}.", + sid, + ) + ) + + for category in document.categories: + issues.extend(self._check_identifier(category.id, "category id", sid)) + issues.extend(self._require(category.id, "category id", sid)) + + for permission in document.permissions: + issues.extend(self._check_identifier(permission.namespace, "permission namespace", sid)) + issues.extend(self._check_identifier(permission.name, "permission name", sid)) + issues.extend(self._require(permission.category_id, f"category for {permission.identifier}", sid)) + issues.extend(self._check_scopes(permission.scopes, f"permission {permission.identifier}", sid)) + + for role in document.roles: + issues.extend(self._check_identifier(role.id, "role id", sid)) + issues.extend(self._check_scopes(role.scopes, f"role {role.id}", sid)) + for perm_id in role.permissions: + issues.extend(self._check_permission_id(perm_id, f"role {role.id}", sid)) + + for extension in document.role_extensions: + issues.extend(self._check_identifier(extension.role_id, "role_extension target", sid)) + for perm_id in (*extension.add_permissions, *extension.remove_permissions): + issues.extend(self._check_permission_id(perm_id, f"role_extension {extension.role_id}", sid)) + + return issues + + def validate_set(self, documents: list[SchemaDocument]) -> list[ValidationIssue]: + """Whole-set checks across all loaded documents.""" + issues: list[ValidationIssue] = [] + + category_ids: set[str] = set() + permission_index: dict[str, tuple[str, ...]] = {} # id -> scopes + role_ids: set[str] = set() + + issues.extend(self._collect_and_check_duplicates(documents, category_ids, permission_index, role_ids)) + + # Reference integrity: permission categories exist. + for document in documents: + sid = document.source.source_id + for permission in document.permissions: + issues.extend( + self._check_permission_category(permission.identifier, permission.category_id, category_ids, sid) + ) + + # Role permissions exist, and role scopes are supported by each permission. + for role in document.roles: + for perm_id in role.permissions: + issues.extend( + self._check_role_permission_scope( + role.id, + perm_id, + role.scopes, + permission_index.get(perm_id), + sid, + missing_verb="references", + ) + ) + + # Extensions target existing roles and reference existing permissions. + for extension in document.role_extensions: + if extension.role_id not in role_ids: + issues.append( + ValidationIssue( + IssueLevel.ERROR, + f"role_extension targets unknown role {extension.role_id!r}.", + sid, + ) + ) + for perm_id in (*extension.add_permissions, *extension.remove_permissions): + if perm_id not in permission_index: + issues.append( + ValidationIssue( + IssueLevel.ERROR, + f"role_extension {extension.role_id} references unknown permission {perm_id!r}.", + sid, + ) + ) + + return issues + + def _collect_and_check_duplicates( + self, + documents: list[SchemaDocument], + category_ids: set[str], + permission_index: dict[str, tuple[str, ...]], + role_ids: set[str], + ) -> list[ValidationIssue]: + """Populate the id indexes and flag conflicting/identical duplicates.""" + issues: list[ValidationIssue] = [] + categories: dict[str, object] = {} + permissions: dict[str, object] = {} + roles: dict[str, object] = {} + + for document in documents: + sid = document.source.source_id + for category in document.categories: + issues.extend(self._register(categories, category.id, category, "category", sid)) + category_ids.add(category.id) + for permission in document.permissions: + issues.extend(self._register(permissions, permission.identifier, permission, "permission", sid)) + permission_index[permission.identifier] = permission.scopes + for role in document.roles: + issues.extend(self._register(roles, role.id, role, "role", sid)) + role_ids.add(role.id) + return issues + + @staticmethod + def _register(index: dict, key: str, value, kind: str, sid: str) -> list[ValidationIssue]: + """Record a base definition, flagging duplicates. + + Identical duplicate → warning; conflicting duplicate → error. + """ + if key not in index: + index[key] = value + return [] + if index[key] == value: + return [ValidationIssue(IssueLevel.WARNING, f"Duplicate identical {kind} {key!r}.", sid)] + return [ValidationIssue(IssueLevel.ERROR, f"Conflicting {kind} definition for {key!r}.", sid)] + + def _check_identifier(self, value: str, label: str, sid: str) -> list[ValidationIssue]: + """Validate a single identifier is lowercase snake_case and not a Casbin form.""" + if not value: + return [] # emptiness handled by _require where relevant + if any(value.startswith(prefix) for prefix in CASBIN_INTERNAL_PREFIXES): + return [ValidationIssue(IssueLevel.ERROR, f"{label} {value!r} uses an internal Casbin form.", sid)] + if not IDENTIFIER_RE.match(value): + return [ + ValidationIssue( + IssueLevel.ERROR, + f"{label} {value!r} must match {IDENTIFIER_RE.pattern} (lowercase snake_case).", + sid, + ) + ] + return [] + + def _check_permission_id(self, value: str, context: str, sid: str) -> list[ValidationIssue]: + """A complete permission id is ``namespace.name`` with both parts valid. + + The shape is checked against :data:`PERMISSION_ID_RE`, which mirrors the + ``permission_id`` definition in ``schema/authz-schema-v1.json``: a single + period joining two lowercase snake_case identifiers, each beginning with + a letter. Internal Casbin forms are rejected with a clearer message. + """ + if any(value.startswith(prefix) for prefix in CASBIN_INTERNAL_PREFIXES): + return [ + ValidationIssue( + IssueLevel.ERROR, + f"{context}: permission id {value!r} uses an internal Casbin form.", + sid, + ) + ] + if not PERMISSION_ID_RE.match(value): + return [ + ValidationIssue( + IssueLevel.ERROR, + f"{context}: permission id {value!r} must match {PERMISSION_ID_RE.pattern} " + "(two lowercase snake_case identifiers joined by a period, e.g. 'courses.view_course').", + sid, + ) + ] + return [] + + @staticmethod + def _check_role_permission_scope( + role_id: str, + perm_id: str, + role_scopes: tuple[str, ...], + permission_scopes: tuple[str, ...] | None, + sid: str | None, + *, + missing_verb: str, + ) -> list[ValidationIssue]: + """Check one role-permission pairing: the permission exists and supports the role's scopes. + + ``permission_scopes`` is the permission's supported scopes, or ``None`` + when the permission does not exist. ``missing_verb`` is the phrase used + for the unknown-permission message, so the pre-compile pass can say the + role *references* a permission while the post-compile pass says it + *resolves to* one. + """ + if permission_scopes is None: + return [ + ValidationIssue( + IssueLevel.ERROR, + f"Role {role_id} {missing_verb} unknown permission {perm_id!r}.", + sid, + ) + ] + unsupported = set(role_scopes) - set(permission_scopes) + if unsupported: + return [ + ValidationIssue( + IssueLevel.ERROR, + f"Role {role_id} is defined for scope(s) {sorted(unsupported)} " + f"that permission {perm_id!r} does not support.", + sid, + ) + ] + return [] + + @staticmethod + def _check_permission_category( + perm_id: str, + category_id: str, + known_category_ids: Container[str], + sid: str | None, + ) -> list[ValidationIssue]: + """A permission's category, when set, must reference an existing category.""" + if category_id and category_id not in known_category_ids: + return [ + ValidationIssue( + IssueLevel.ERROR, + f"Permission {perm_id} references unknown category {category_id!r}.", + sid, + ) + ] + return [] + + def _check_scopes(self, scopes: tuple[str, ...], context: str, sid: str) -> list[ValidationIssue]: + """Validate that at least one scope is declared and each scope namespace is well-formed.""" + if not scopes: + return [ValidationIssue(IssueLevel.ERROR, f"{context} must declare at least one scope.", sid)] + issues: list[ValidationIssue] = [] + for scope in scopes: + if not SCOPE_RE.match(scope): + issues.append(ValidationIssue(IssueLevel.ERROR, f"{context}: invalid scope namespace {scope!r}.", sid)) + return issues + + @staticmethod + def _require(value: str, label: str, sid: str) -> list[ValidationIssue]: + if not value: + return [ValidationIssue(IssueLevel.ERROR, f"Missing required field: {label}.", sid)] + return [] + + @staticmethod + def _role_permission_source_id(schema: CompiledSchema, role_id: str, perm_id: str) -> str | None: + """Name the source(s) responsible for one role-permission grant. + + A role-permission grant is the link that assigns a permission to a role: + a single ``(role_id, permission_id)`` pairing in the compiled schema. ADR + 0025 defines it as the atomic unit of attribution (one grant renders to + one Casbin ``p`` row). It is distinct from a *role assignment*, which + links a subject to a role (the ``g`` policies of ADR 0012). The grant's + provenance is tracked on ``schema.role_permission_sources``; this returns + the source id(s) to attribute a validation issue to. + + Prefers extension contributions: when an extension introduces the + offending permission, the operator needs the extending file's id, not + the file that declared the role. + """ + sources = schema.role_permission_sources.get((role_id, perm_id), []) + extensions = [src for src in sources if src.origin_kind == SchemaOriginKind.EXTENSION] + chosen = extensions or sources + if not chosen: + return None + return ", ".join(sorted({src.source.source_id for src in chosen})) diff --git a/src/openedx_authz/tests/schema/test_validation.py b/src/openedx_authz/tests/schema/test_validation.py new file mode 100644 index 00000000..20f8b8b6 --- /dev/null +++ b/src/openedx_authz/tests/schema/test_validation.py @@ -0,0 +1,424 @@ +"""Unit tests for the schema validation step. + +Grouped by concern: per-document validation, duplicate severity, identifier and +scope rules, required fields, extension references, post-compile validation, and +the ``has_errors`` gate the pipeline uses to decide whether to stop. +""" + +import pytest + +from openedx_authz.engine.schema.compilation import SchemaCompiler +from openedx_authz.engine.schema.validation import ( + IssueLevel, + SchemaValidator, + ValidationIssue, +) + +from .factories import category, extension, make_document, make_source, permission, role + + +def _errors(issues): + return [i for i in issues if i.is_error] + + +def _warnings(issues): + return [i for i in issues if not i.is_error] + + +class TestDocumentValidation: + """Per-document ``validate`` checks (ADR 0017 §4). + + These run before compilation and see each document's own declarations, so + they cover identifier shape, Casbin-form rejection, intra-document + references (category, permission, role, scope), and duplicate conflicts. + """ + + def test_valid_document_has_no_errors(self): + """A well-formed document produces no error issues.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat")], + roles=[role(permissions=("courses.view_course",))], + ) + assert not _errors(SchemaValidator().validate([doc])) + + def test_unsupported_schema_version_is_error(self): + """A ``schema_version`` the validator cannot read is rejected.""" + doc = make_document(schema_version="9.9", categories=[category()]) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("Unsupported schema_version" in m for m in messages) + + def test_non_snakecase_identifier_is_error(self): + """Identifiers must be lowercase snake_case.""" + doc = make_document(permissions=[permission(namespace="Courses")]) + assert _errors(SchemaValidator().validate([doc])) + + def test_casbin_internal_form_rejected(self): + """An identifier using an internal Casbin form (``act^``) is rejected.""" + doc = make_document(categories=[category("act^foo")]) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("internal Casbin form" in m for m in messages) + + def test_unknown_category_reference_is_error(self): + """A permission referencing a category not defined in the schema fails.""" + doc = make_document(permissions=[permission(cat="missing")]) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("unknown category" in m for m in messages) + + def test_unknown_permission_in_role_is_error(self): + """A role listing a permission that does not exist fails.""" + doc = make_document(roles=[role(permissions=("courses.nope",))]) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("unknown permission" in m for m in messages) + + def test_role_scope_not_supported_by_permission_is_error(self): + """A role cannot grant a permission outside the permission's scopes.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat", scopes=("course-v1",))], + roles=[role(rid="r", scopes=("lib",), permissions=("courses.view_course",))], + ) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("does not support" in m for m in messages) + + def test_extension_targeting_unknown_role_is_error(self): + """An extension for a role absent from the same document fails here.""" + doc = make_document(role_extensions=[extension("ghost", add_permissions=("courses.view_course",))]) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("unknown role" in m for m in messages) + + def test_missing_scope_is_error(self): + """A permission must declare at least one scope.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat", scopes=())], + ) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("at least one scope" in m for m in messages) + + def test_conflicting_duplicate_definition_is_error(self): + """Two definitions with the same id but different content conflict.""" + doc = make_document( + categories=[category("cat")], + permissions=[ + permission(cat="cat", display_name="One"), + permission(cat="cat", display_name="Two"), # same id, different content + ], + ) + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + assert any("Conflicting permission" in m for m in messages) + + +class TestDuplicateSeverity: + """ADR 0017 §4 splits duplicates by severity: identical warns, differing fails.""" + + def test_identical_duplicate_warns_instead_of_failing(self): + """Two packages shipping the same definition is legal, not an error.""" + first = make_document("first", categories=[category("cat")]) + second = make_document("second", categories=[category("cat")]) + + issues = SchemaValidator().validate([first, second]) + + assert not _errors(issues) + assert any("Duplicate identical category" in i.message for i in _warnings(issues)) + + def test_identical_duplicate_role_warns(self): + """Two identical role definitions warn rather than error.""" + first = make_document("first", roles=[role()]) + second = make_document("second", roles=[role()]) + + issues = SchemaValidator().validate([first, second]) + + assert any("Duplicate identical role" in i.message for i in _warnings(issues)) + + def test_conflicting_duplicate_role_is_error(self): + """Two role definitions with the same id but different content conflict.""" + first = make_document("first", roles=[role(display_name="Editor")]) + second = make_document("second", roles=[role(display_name="Author")]) + + messages = [i.message for i in _errors(SchemaValidator().validate([first, second]))] + + assert any("Conflicting role definition" in m for m in messages) + + def test_warning_names_the_second_source(self): + """The duplicate warning is attributed to the later contributing source.""" + first = make_document("first", categories=[category("cat")]) + second = make_document("second", categories=[category("cat")]) + + issues = _warnings(SchemaValidator().validate([first, second])) + + assert [i.source_id for i in issues] == [make_source("second").source_id] + + +class TestIdentifierAndScopeRules: + """Field-shape rules from ADR 0017 §4.""" + + @pytest.mark.parametrize("value", ["courses", "courses.view.course", "courses.", ".view", "courses..view", ""]) + def test_permission_id_must_be_namespace_dot_name(self, value): + """A permission id must be exactly two identifiers joined by one period.""" + doc = make_document(roles=[role(permissions=(value,))]) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("permission id" in m and "must match" in m for m in messages) + + def test_permission_id_halves_are_validated(self): + """Both halves of a permission id must be lowercase snake_case.""" + doc = make_document(roles=[role(permissions=("Courses.View_Course",))]) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("lowercase snake_case" in m for m in messages) + + @pytest.mark.parametrize("prefix", ["act^", "role^", "sub^", "scope^", "g^", "p^"]) + def test_every_casbin_prefix_is_rejected(self, prefix): + """Every reserved Casbin prefix (act^, role^, ...) is rejected in an id.""" + doc = make_document(roles=[role(rid=f"{prefix}thing")]) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("internal Casbin form" in m for m in messages) + + @pytest.mark.parametrize("scope", ["Course-V1", "1course", "course v1", "course.v1"]) + def test_invalid_scope_namespace_is_error(self, scope): + """A malformed scope namespace (caps, leading digit, spaces, dots) is rejected.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat", scopes=(scope,))], + ) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("invalid scope namespace" in m for m in messages) + + def test_hyphenated_scope_is_allowed(self): + """Scope namespaces keep their registered spelling, e.g. ``course-v1``.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat", scopes=("course-v1",))], + ) + + assert not _errors(SchemaValidator().validate([doc])) + + def test_role_without_scopes_is_error(self): + """A role must declare at least one scope.""" + doc = make_document(roles=[role(scopes=())]) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("at least one scope" in m for m in messages) + + +class TestRequiredFields: + """The subset of required fields the validator enforces today. + + Fuller required-field coverage (display fields, unknown keys, sizes) arrives + with JSON Schema validation; these pin the rules already in place. + """ + + def test_empty_category_id_is_error(self): + """A category with an empty id is a missing-required-field error.""" + doc = make_document(categories=[category("")]) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("Missing required field: category id" in m for m in messages) + + def test_permission_without_a_category_is_error(self): + """A permission missing its required category is an error.""" + doc = make_document(permissions=[permission(cat="")]) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("Missing required field: category for courses.view_course" in m for m in messages) + + +class TestExtensionReferences: + """ADR 0023 §3: an extension must target a real role and real permissions.""" + + def test_added_permission_must_exist(self): + """An extension that adds a nonexistent permission is an error.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat")], + roles=[role()], + role_extensions=[extension("course_editor", add_permissions=("courses.ghost",))], + ) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("references unknown permission 'courses.ghost'" in m for m in messages) + + def test_removed_permission_must_exist(self): + """An extension that removes a nonexistent permission is an error.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat")], + roles=[role()], + role_extensions=[extension("course_editor", remove_permissions=("courses.ghost",))], + ) + + messages = [i.message for i in _errors(SchemaValidator().validate([doc]))] + + assert any("references unknown permission 'courses.ghost'" in m for m in messages) + + def test_extension_may_target_a_role_from_another_document(self): + """An extension may target a role defined in a different document.""" + base = make_document( + "base", + categories=[category("cat")], + permissions=[permission(cat="cat")], + roles=[role()], + ) + ext = make_document("ext", role_extensions=[extension("course_editor", display_name="Author")]) + + assert not _errors(SchemaValidator().validate([base, ext])) + + +class TestValidateCompiled: + """Post-compile rules that only the resolved schema can answer. + + Document-level validation sees base declarations only, so these cases pass + ``validate`` and must be caught by ``validate_compiled``. + """ + + @staticmethod + def _compile(*documents): + return SchemaCompiler().compile(list(documents)) + + def test_extension_added_permission_must_support_role_scopes(self): + """An extension cannot grant a permission outside the role's scopes. + + Regression test: ``course_editor`` is a ``course-v1`` role, the added + permission only applies to ``lib``, yet document validation is clean + because it never inspects the extension's effect (ADR 0017 §4). + """ + base = make_document( + "base", + priority=100, + categories=[category("cat")], + permissions=[ + permission(cat="cat", scopes=("course-v1",)), + permission("libraries", "edit_library", cat="cat", scopes=("lib",)), + ], + roles=[role(rid="course_editor", scopes=("course-v1",), permissions=("courses.view_course",))], + ) + ext = make_document( + "ext", + priority=200, + role_extensions=[extension("course_editor", add_permissions=("libraries.edit_library",))], + ) + validator = SchemaValidator() + assert not _errors(validator.validate([base, ext])) + + messages = [i.message for i in _errors(validator.validate_compiled(self._compile(base, ext)))] + + assert any("libraries.edit_library" in m and "does not support" in m for m in messages) + + def test_error_names_the_extending_source(self): + """The operator needs the extending file's id, not the role's file.""" + base = make_document( + "base", + priority=100, + categories=[category("cat")], + permissions=[ + permission(cat="cat", scopes=("course-v1",)), + permission("libraries", "edit_library", cat="cat", scopes=("lib",)), + ], + roles=[role(rid="course_editor", scopes=("course-v1",), permissions=("courses.view_course",))], + ) + ext = make_document( + "ext", + priority=200, + role_extensions=[extension("course_editor", add_permissions=("libraries.edit_library",))], + ) + + issues = _errors(SchemaValidator().validate_compiled(self._compile(base, ext))) + + assert [i.source_id for i in issues] == [make_source("ext").source_id] + + def test_no_error_when_extension_permission_shares_the_role_scope(self): + """Negative control: the rule must not fire on a compatible extension.""" + base = make_document( + "base", + priority=100, + categories=[category("cat")], + permissions=[ + permission(cat="cat", scopes=("course-v1",)), + permission("courses", "export_course", cat="cat", scopes=("course-v1",)), + ], + roles=[role(rid="course_editor", scopes=("course-v1",), permissions=("courses.view_course",))], + ) + ext = make_document( + "ext", + priority=200, + role_extensions=[extension("course_editor", add_permissions=("courses.export_course",))], + ) + + assert not _errors(SchemaValidator().validate_compiled(self._compile(base, ext))) + + def test_multi_scope_permission_supports_a_narrower_role(self): + """A permission may support more scopes than the role uses.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat", scopes=("course-v1", "lib"))], + roles=[role(rid="course_editor", scopes=("course-v1",), permissions=("courses.view_course",))], + ) + + assert not _errors(SchemaValidator().validate_compiled(self._compile(doc))) + + def test_compiled_permission_with_unknown_category_is_error(self): + """Defensive: a category that vanished during resolution is reported.""" + doc = make_document( + permissions=[permission(cat="missing")], + ) + + messages = [i.message for i in _errors(SchemaValidator().validate_compiled(self._compile(doc)))] + + assert any("unknown category" in m for m in messages) + + def test_valid_schema_has_no_compiled_errors(self): + """A well-formed compiled schema produces no post-compile errors.""" + doc = make_document( + categories=[category("cat")], + permissions=[permission(cat="cat")], + roles=[role(permissions=("courses.view_course",))], + ) + + assert not _errors(SchemaValidator().validate_compiled(self._compile(doc))) + + def test_compiled_role_referencing_a_missing_permission_is_error(self): + """Defensive: a permission definition that never made it into the schema.""" + doc = make_document(roles=[role(permissions=("courses.ghost",))]) + + messages = [i.message for i in _errors(SchemaValidator().validate_compiled(self._compile(doc)))] + + assert any("resolves to unknown permission 'courses.ghost'" in m for m in messages) + + def test_source_id_is_omitted_when_provenance_is_missing(self): + """A grant with no recorded provenance still produces a usable issue.""" + doc = make_document(roles=[role(permissions=("courses.ghost",))]) + schema = self._compile(doc) + schema.role_permission_sources.clear() + + issues = _errors(SchemaValidator().validate_compiled(schema)) + + assert [i.source_id for i in issues] == [None] + + +class TestHasErrors: + """The gate the pipeline uses to decide whether to stop.""" + + def test_true_when_any_issue_is_an_error(self): + """``has_errors`` is True when any issue is error-level.""" + issues = [ValidationIssue(IssueLevel.WARNING, "heads up"), ValidationIssue(IssueLevel.ERROR, "boom")] + + assert SchemaValidator.has_errors(issues) is True + + def test_false_for_warnings_only(self): + """``has_errors`` is False when every issue is a warning.""" + assert SchemaValidator.has_errors([ValidationIssue(IssueLevel.WARNING, "heads up")]) is False + + def test_false_for_no_issues(self): + """``has_errors`` is False for an empty issue list.""" + assert SchemaValidator.has_errors([]) is False