diff --git a/cloud_pipelines_backend/access_control/checks.py b/cloud_pipelines_backend/access_control/checks.py new file mode 100644 index 0000000..ed5c1e9 --- /dev/null +++ b/cloud_pipelines_backend/access_control/checks.py @@ -0,0 +1,79 @@ +import dataclasses +from typing import Any + +from .. import errors +from . import keys +from . import scopes + + +class NotAuthorizedError(errors.PermissionError): + """A failed scope check; `owner` is carried for the host to show or withhold.""" + + def __init__(self, message: str, *, owner: str | None = None) -> None: + super().__init__(message) + self.owner = owner + + +@dataclasses.dataclass(frozen=True, kw_only=True) +class Caller: + """Who is asking; is_admin holds every scope, for hosts that pass it rather than skip the check.""" + + email: str | None + is_admin: bool = False + + +def normalize_email(*, email: str) -> str: + """Lowercases and trims an email so the same address always compares equal.""" + return email.strip().lower() + + +def is_owner(*, owner: str | None, caller: Caller) -> bool: + """True when the caller's email matches the owner's, ignoring case and whitespace.""" + return bool( + owner + and caller.email + and normalize_email(email=owner) == normalize_email(email=caller.email) + ) + + +def effective_scopes( + *, + resource_type: scopes.ResourceType, + owner: str | None, + access_control: dict[str, Any] | None, + caller: Caller, +) -> frozenset[str]: + """Fine scopes the caller holds: all for the owner and admins, else what their grant expands to.""" + if caller.is_admin or is_owner(owner=owner, caller=caller): + return resource_type.fine_scopes + if not caller.email: + return frozenset() + users = (access_control or {}).get(keys.Key.USERS) or {} + entry = users.get(normalize_email(email=caller.email)) or {} + granted: set[str] = set() + for scope in entry.get(keys.Key.PERMISSIONS) or []: + granted |= resource_type.expand(scope=scope) + return frozenset(granted) + + +def require( + *, + scope: str, + resource_type: scopes.ResourceType, + resource_id: str, + owner: str | None, + access_control: dict[str, Any] | None, + caller: Caller, +) -> None: + """Raises NotAuthorizedError unless the caller holds the scope on the resource.""" + granted = effective_scopes( + resource_type=resource_type, + owner=owner, + access_control=access_control, + caller=caller, + ) + if scope not in granted: + raise NotAuthorizedError( + f"{caller.email} does not have {scope} on {resource_type.name} {resource_id}.", + owner=owner, + ) diff --git a/cloud_pipelines_backend/access_control/edits.py b/cloud_pipelines_backend/access_control/edits.py new file mode 100644 index 0000000..87f7beb --- /dev/null +++ b/cloud_pipelines_backend/access_control/edits.py @@ -0,0 +1,98 @@ +import datetime +from typing import Any, Final + +from .. import errors +from . import checks +from . import keys +from . import scopes + +MAX_HISTORY_ENTRIES: Final[int] = 100 + + +class StaleRevisionError(Exception): + pass + + +def latest_revision(*, access_control: dict[str, Any] | None) -> int: + """Revision of the newest history entry, which is always last, or 0 when access has never been changed.""" + history = (access_control or {}).get(keys.Key.HISTORY) or [] + return history[-1][keys.Key.REVISION] if history else 0 + + +def _validated_users( + *, + resource_type: scopes.ResourceType, + owner: str | None, + users: dict[str, dict[str, list[str]]], +) -> dict[str, dict[str, list[str]]]: + """Normalizes a requested user list and drops empty grants. + + Rejects blank emails, the owner, duplicates and unknown scopes. + """ + validated: dict[str, dict[str, list[str]]] = {} + for raw_email, entry in users.items(): + email = checks.normalize_email(email=raw_email) + if not email: + raise errors.ApiValidationError("User email must not be empty.") + if owner and email == checks.normalize_email(email=owner): + raise errors.ApiValidationError( + f"{email} owns this {resource_type.name} and cannot be listed." + ) + if email in validated: + raise errors.ApiValidationError(f"{email} is listed more than once.") + permissions = sorted(set(entry.get(keys.Key.PERMISSIONS) or [])) + unknown = [ + scope for scope in permissions if not resource_type.is_known(scope=scope) + ] + if unknown: + raise errors.ApiValidationError( + f"Unknown permissions for {resource_type.name}: {', '.join(unknown)}." + ) + if permissions: + validated[email] = {keys.Key.PERMISSIONS: permissions} + return validated + + +def apply_users_edit( + *, + resource_type: scopes.ResourceType, + owner: str | None, + access_control: dict[str, Any] | None, + users: dict[str, dict[str, list[str]]], + changed_by: str, + changed_at: datetime.datetime, + expected_revision: int, +) -> dict[str, Any]: + """Replaces the user list at expected_revision and logs each user's from/to as one history entry. + + The revision check only sees this snapshot, so the caller must hold a row lock across the read + that produced access_control and the write of the result. Keeps the newest MAX_HISTORY_ENTRIES. + """ + current = access_control or {} + revision = latest_revision(access_control=current) + if expected_revision != revision: + raise StaleRevisionError( + f"Access was changed since revision {expected_revision}; the latest is {revision}." + ) + old_users = current.get(keys.Key.USERS) or {} + new_users = _validated_users(resource_type=resource_type, owner=owner, users=users) + changes = {} + for email in sorted(old_users.keys() | new_users.keys()): + before = (old_users.get(email) or {}).get(keys.Key.PERMISSIONS) or [] + after = (new_users.get(email) or {}).get(keys.Key.PERMISSIONS) or [] + if before != after: + changes[email] = {keys.Key.FROM: before, keys.Key.TO: after} + history = list(current.get(keys.Key.HISTORY) or []) + if changes: + history.append( + { + keys.Key.REVISION: revision + 1, + keys.Key.CHANGED_BY: checks.normalize_email(email=changed_by), + keys.Key.CHANGED_AT: changed_at.isoformat(), + keys.Key.CHANGES: changes, + } + ) + return { + keys.Key.USERS: new_users, + keys.Key.HISTORY: history[-MAX_HISTORY_ENTRIES:], + } diff --git a/cloud_pipelines_backend/access_control/keys.py b/cloud_pipelines_backend/access_control/keys.py new file mode 100644 index 0000000..17e5932 --- /dev/null +++ b/cloud_pipelines_backend/access_control/keys.py @@ -0,0 +1,16 @@ +import enum + + +class Key(str, enum.Enum): + __hash__ = str.__hash__ + __str__ = str.__str__ + + USERS = "users" + PERMISSIONS = "permissions" + HISTORY = "history" + REVISION = "revision" + CHANGED_BY = "changed_by" + CHANGED_AT = "changed_at" + CHANGES = "changes" + FROM = "from" + TO = "to" diff --git a/cloud_pipelines_backend/access_control/runs.py b/cloud_pipelines_backend/access_control/runs.py new file mode 100644 index 0000000..21d53fd --- /dev/null +++ b/cloud_pipelines_backend/access_control/runs.py @@ -0,0 +1,120 @@ +import dataclasses +import enum +from collections import abc +from typing import Any, Protocol + +from sqlalchemy import orm + +from .. import backend_types_sql as bts +from . import checks +from . import scopes + + +class RunScope(str, enum.Enum): + __hash__ = str.__hash__ + __str__ = str.__str__ + + CANCEL = "run:cancel" + ANNOTATE = "run:annotate" + + +@dataclasses.dataclass(frozen=True, kw_only=True) +class RunParent: + resource_type: scopes.ResourceType + owner: str | None + access_control: dict[str, Any] | None + + +class RunParentResolver(Protocol): + def __call__( + self, *, session: orm.Session, pipeline_run: bts.PipelineRun + ) -> abc.Iterable[RunParent]: ... + + +_parent_resolvers: list[RunParentResolver] = [] +_parent_annotation_keys: set[str] = set() + + +def register_run_parent_resolver( + *, resolver: RunParentResolver, annotation_keys: abc.Iterable[str] = () +) -> None: + """Adds a host hook that finds the resources a run inherits run:* scopes from, via the given annotations.""" + if resolver not in _parent_resolvers: + _parent_resolvers.append(resolver) + _parent_annotation_keys.update(annotation_keys) + + +def _is_starter_or_admin( + *, pipeline_run: bts.PipelineRun, caller: checks.Caller +) -> bool: + """True for admins and whoever started the run.""" + return ( + caller.is_admin + or caller.email == pipeline_run.created_by + or checks.is_owner(owner=pipeline_run.created_by, caller=caller) + ) + + +def run_scopes( + *, + session: orm.Session, + pipeline_run: bts.PipelineRun, + caller: checks.Caller, +) -> frozenset[str]: + """Run scopes the caller holds: all for the starter and admins, else those inherited from parents.""" + all_scopes = frozenset(scope.value for scope in RunScope) + if _is_starter_or_admin(pipeline_run=pipeline_run, caller=caller): + return all_scopes + granted: set[str] = set() + for resolver in _parent_resolvers: + for parent in resolver(session=session, pipeline_run=pipeline_run): + granted |= checks.effective_scopes( + resource_type=parent.resource_type, + owner=parent.owner, + access_control=parent.access_control, + caller=caller, + ) + return frozenset(granted) & all_scopes + + +def require_run_scope( + *, + session: orm.Session, + pipeline_run: bts.PipelineRun, + caller: checks.Caller, + scope: RunScope, +) -> None: + """Raises NotAuthorizedError unless the caller holds the scope on the run.""" + if scope not in run_scopes( + session=session, pipeline_run=pipeline_run, caller=caller + ): + raise checks.NotAuthorizedError( + f"The pipeline run {pipeline_run.id} was started by {pipeline_run.created_by} and {caller.email} does not have {scope} on it.", + owner=pipeline_run.created_by, + ) + + +def require_annotation_change( + *, + session: orm.Session, + pipeline_run: bts.PipelineRun, + caller: checks.Caller, + key: str, +) -> None: + """Raises NotAuthorizedError unless the caller may set or delete the annotation. + + Annotations a parent resolver reads are kept to the starter and admins, so a grantee can't re-parent the run. + """ + require_run_scope( + session=session, + pipeline_run=pipeline_run, + caller=caller, + scope=RunScope.ANNOTATE, + ) + if key in _parent_annotation_keys and not _is_starter_or_admin( + pipeline_run=pipeline_run, caller=caller + ): + raise checks.NotAuthorizedError( + f"Only {pipeline_run.created_by} can change the {key} annotation on pipeline run {pipeline_run.id}.", + owner=pipeline_run.created_by, + ) diff --git a/cloud_pipelines_backend/access_control/scopes.py b/cloud_pipelines_backend/access_control/scopes.py new file mode 100644 index 0000000..e5d7bdd --- /dev/null +++ b/cloud_pipelines_backend/access_control/scopes.py @@ -0,0 +1,69 @@ +import dataclasses +import enum + +from .. import errors + + +class BroadScope(str, enum.Enum): + __hash__ = str.__hash__ + __str__ = str.__str__ + + OPERATE = "operate" + EDIT = "edit" + MANAGE = "manage" + + +@dataclasses.dataclass(frozen=True, kw_only=True) +class ResourceType: + name: str + operate: frozenset[str] + edit: frozenset[str] + manage: frozenset[str] + + @property + def fine_scopes(self) -> frozenset[str]: + """Every fine scope this resource type defines.""" + return self.operate | self.edit | self.manage + + def expand(self, *, scope: str) -> frozenset[str]: + """Fine scopes a grant stands for; a broad scope includes lower tiers, an unknown one nothing.""" + match scope: + case BroadScope.OPERATE: + return self.operate + case BroadScope.EDIT: + return self.operate | self.edit + case BroadScope.MANAGE: + return self.fine_scopes + return frozenset({scope}) & self.fine_scopes + + def is_known(self, *, scope: str) -> bool: + """True for a broad scope or one of this type's fine scopes.""" + return scope in set(BroadScope) or scope in self.fine_scopes + + +_resource_types: dict[str, ResourceType] = {} + + +def register_resource_type( + *, + name: str, + operate: frozenset[str], + edit: frozenset[str], + manage: frozenset[str], +) -> ResourceType: + """Declares a resource type's fine scopes per tier; re-declaring it with different scopes raises.""" + resource_type = ResourceType(name=name, operate=operate, edit=edit, manage=manage) + existing = _resource_types.get(name) + if existing is not None and existing != resource_type: + raise errors.ApiValidationError( + f"Resource type {name} is already registered with different scopes." + ) + _resource_types[name] = resource_type + return resource_type + + +def get_resource_type(*, name: str) -> ResourceType: + """Looks up a registered resource type, raising ItemNotFoundError if none has that name.""" + if name not in _resource_types: + raise errors.ItemNotFoundError(f"Unknown resource type {name}.") + return _resource_types[name] diff --git a/cloud_pipelines_backend/api_server_sql.py b/cloud_pipelines_backend/api_server_sql.py index 607a177..c667a7b 100644 --- a/cloud_pipelines_backend/api_server_sql.py +++ b/cloud_pipelines_backend/api_server_sql.py @@ -10,6 +10,8 @@ from . import backend_types_sql as bts from . import component_structures as structures from . import errors +from .access_control import checks as access_control_checks +from .access_control import runs as access_control_runs from . import filter_query_sql if typing.TYPE_CHECKING: @@ -209,9 +211,12 @@ def terminate( pipeline_run = session.get(bts.PipelineRun, id) if not pipeline_run: raise errors.ItemNotFoundError(f"Pipeline run {id} not found.") - if not skip_user_check and (terminated_by != pipeline_run.created_by): - raise errors.PermissionError( - f"The pipeline run {id} was started by {pipeline_run.created_by} and cannot be terminated by {terminated_by}" + if not skip_user_check: + access_control_runs.require_run_scope( + session=session, + pipeline_run=pipeline_run, + caller=access_control_checks.Caller(email=terminated_by), + scope=access_control_runs.RunScope.CANCEL, ) _logger.info( f"{pipeline_run.id=} The pipeline run is being cancelled by {terminated_by}." @@ -408,9 +413,12 @@ def set_annotation( pipeline_run = session.get(bts.PipelineRun, id) if not pipeline_run: raise errors.ItemNotFoundError(f"Pipeline run {id} not found.") - if not skip_user_check and (user_name != pipeline_run.created_by): - raise errors.PermissionError( - f"The pipeline run {id} was started by {pipeline_run.created_by} and cannot be changed by {user_name}" + if not skip_user_check: + access_control_runs.require_annotation_change( + session=session, + pipeline_run=pipeline_run, + caller=access_control_checks.Caller(email=user_name), + key=key, ) _mirror_single_pipeline_run_annotation( session=session, @@ -433,9 +441,12 @@ def delete_annotation( pipeline_run = session.get(bts.PipelineRun, id) if not pipeline_run: raise errors.ItemNotFoundError(f"Pipeline run {id} not found.") - if not skip_user_check and (user_name != pipeline_run.created_by): - raise errors.PermissionError( - f"The pipeline run {id} was started by {pipeline_run.created_by} and cannot be changed by {user_name}" + if not skip_user_check: + access_control_runs.require_annotation_change( + session=session, + pipeline_run=pipeline_run, + caller=access_control_checks.Caller(email=user_name), + key=key, ) existing_annotation = session.get(bts.PipelineRunAnnotation, (id, key)) diff --git a/tests/test_access_control.py b/tests/test_access_control.py new file mode 100644 index 0000000..01d7d6d --- /dev/null +++ b/tests/test_access_control.py @@ -0,0 +1,344 @@ +import datetime + +import pytest +from sqlalchemy import orm + +from cloud_pipelines_backend import api_server_sql +from cloud_pipelines_backend import backend_types_sql as bts +from cloud_pipelines_backend import component_structures as structures +from cloud_pipelines_backend import database_ops +from cloud_pipelines_backend import errors +from cloud_pipelines_backend.access_control import checks +from cloud_pipelines_backend.access_control import edits +from cloud_pipelines_backend.access_control import runs +from cloud_pipelines_backend.access_control import scopes + +_NOW = datetime.datetime(2026, 10, 7, tzinfo=datetime.timezone.utc) + +_PIPELINE = scopes.ResourceType( + name="pipeline", + operate=frozenset(runs.RunScope), + edit=frozenset({"pipeline:update"}), + manage=frozenset({"pipeline:delete", "pipeline:share"}), +) + + +def _granted(*permissions: str) -> dict: + return {"users": {"sam@example.com": {"permissions": list(permissions)}}} + + +class TestEffectiveScopes: + @pytest.mark.parametrize( + ("permissions", "expected"), + [ + (["operate"], {"run:cancel", "run:annotate"}), + (["edit"], {"run:cancel", "run:annotate", "pipeline:update"}), + (["manage"], _PIPELINE.fine_scopes), + (["run:cancel"], {"run:cancel"}), + (["made:up"], set()), + ], + ) + def test_expands_permissions(self, permissions, expected) -> None: + granted = checks.effective_scopes( + resource_type=_PIPELINE, + owner="ada@example.com", + access_control=_granted(*permissions), + caller=checks.Caller(email=" Sam@Example.com "), + ) + assert granted == expected + + @pytest.mark.parametrize( + "caller", + [ + checks.Caller(email="ADA@example.com"), + checks.Caller(email="other@example.com", is_admin=True), + ], + ) + def test_owner_and_admin_hold_manage(self, caller) -> None: + granted = checks.effective_scopes( + resource_type=_PIPELINE, + owner="ada@example.com", + access_control=None, + caller=caller, + ) + assert granted == _PIPELINE.fine_scopes + + def test_require_raises_permission_error(self) -> None: + with pytest.raises(errors.PermissionError) as raised: + checks.require( + scope="pipeline:update", + resource_type=_PIPELINE, + resource_id="p1", + owner="ada@example.com", + access_control=_granted("operate"), + caller=checks.Caller(email="sam@example.com"), + ) + assert raised.value.owner == "ada@example.com" + assert "ada@example.com" not in str(raised.value) + + +class TestRegisterResourceType: + def test_same_scopes_are_idempotent(self) -> None: + for _ in range(2): + scopes.register_resource_type( + name="test_idempotent", + operate=frozenset(), + edit=frozenset({"thing:update"}), + manage=frozenset(), + ) + + def test_different_scopes_are_refused(self) -> None: + scopes.register_resource_type( + name="test_conflict", + operate=frozenset(), + edit=frozenset({"thing:update"}), + manage=frozenset(), + ) + with pytest.raises(errors.ApiValidationError): + scopes.register_resource_type( + name="test_conflict", + operate=frozenset(), + edit=frozenset(), + manage=frozenset({"thing:update"}), + ) + + +class TestApplyUsersEdit: + def _edit(self, access_control, users, *, expected_revision=None) -> dict: + return edits.apply_users_edit( + resource_type=_PIPELINE, + owner="ada@example.com", + access_control=access_control, + users=users, + changed_by="Ada@example.com", + changed_at=_NOW, + expected_revision=( + edits.latest_revision(access_control=access_control) + if expected_revision is None + else expected_revision + ), + ) + + def test_history_records_each_change(self) -> None: + access_control = self._edit( + None, {"Sam@example.com": {"permissions": ["operate"]}} + ) + access_control = self._edit( + access_control, + { + "sam@example.com": {"permissions": ["edit"]}, + "kai@example.com": {"permissions": ["run:cancel"]}, + }, + ) + access_control = self._edit( + access_control, {"kai@example.com": {"permissions": ["run:cancel"]}} + ) + history = access_control["history"] + assert [entry["revision"] for entry in history] == [1, 2, 3] + assert history[0]["changes"] == { + "sam@example.com": {"from": [], "to": ["operate"]} + } + assert history[2]["changes"] == { + "sam@example.com": {"from": ["edit"], "to": []} + } + assert history[0]["changed_by"] == "ada@example.com" + assert access_control["users"] == { + "kai@example.com": {"permissions": ["run:cancel"]} + } + + def test_history_keeps_the_newest_entries(self) -> None: + access_control = None + for index in range(edits.MAX_HISTORY_ENTRIES + 5): + access_control = self._edit( + access_control, + {f"user{index}@example.com": {"permissions": ["operate"]}}, + ) + history = access_control["history"] + assert len(history) == edits.MAX_HISTORY_ENTRIES + assert history[0]["revision"] == 6 + assert edits.latest_revision(access_control=access_control) == ( + edits.MAX_HISTORY_ENTRIES + 5 + ) + + def test_concurrent_writers_need_the_callers_lock(self) -> None: + access_control = self._edit( + None, {"sam@example.com": {"permissions": ["edit"]}} + ) + first = self._edit( + access_control, + {"kai@example.com": {"permissions": ["operate"]}}, + expected_revision=1, + ) + second = self._edit( + access_control, + {"lee@example.com": {"permissions": ["operate"]}}, + expected_revision=1, + ) + assert edits.latest_revision(access_control=first) == 2 + assert edits.latest_revision(access_control=second) == 2 + + def test_no_change_appends_nothing(self) -> None: + access_control = self._edit( + None, {"sam@example.com": {"permissions": ["edit"]}} + ) + again = self._edit( + access_control, {"sam@example.com": {"permissions": ["edit"]}} + ) + assert again["history"] == access_control["history"] + + def test_stale_revision_is_rejected(self) -> None: + access_control = self._edit( + None, {"sam@example.com": {"permissions": ["edit"]}} + ) + with pytest.raises(edits.StaleRevisionError): + self._edit(access_control, {}, expected_revision=0) + + @pytest.mark.parametrize( + "users", + [ + {"sam@example.com": {"permissions": ["made:up"]}}, + {"ada@example.com": {"permissions": ["edit"]}}, + {"Sam@example.com": {"permissions": ["edit"]}, "sam@example.com ": {}}, + {" ": {"permissions": ["edit"]}}, + ], + ) + def test_invalid_users_are_rejected(self, users) -> None: + with pytest.raises(errors.ApiValidationError): + self._edit(None, users) + + +@pytest.fixture() +def session_factory(): + engine = database_ops.create_db_engine(database_uri="sqlite://") + bts._TableBase.metadata.create_all(engine) + return orm.sessionmaker(engine) + + +@pytest.fixture() +def shared_with_sam(monkeypatch): + def _resolver(*, session, pipeline_run): + yield runs.RunParent( + resource_type=_PIPELINE, + owner="ada@example.com", + access_control=_granted("run:cancel"), + ) + + monkeypatch.setattr(runs, "_parent_resolvers", [_resolver]) + + +@pytest.fixture() +def annotate_shared_with_sam(monkeypatch): + def _resolver(*, session, pipeline_run): + yield runs.RunParent( + resource_type=_PIPELINE, + owner="ada@example.com", + access_control=_granted("run:annotate"), + ) + + monkeypatch.setattr(runs, "_parent_resolvers", []) + monkeypatch.setattr(runs, "_parent_annotation_keys", set()) + runs.register_run_parent_resolver(resolver=_resolver, annotation_keys={"parent"}) + + +class TestRunScopes: + def _run(self, session_factory) -> api_server_sql.PipelineRunResponse: + root_task = structures.TaskSpec( + component_ref=structures.ComponentReference( + spec=structures.ComponentSpec( + name="p", + implementation=structures.GraphImplementation( + graph=structures.GraphSpec(tasks={}) + ), + ) + ) + ) + with session_factory() as session: + return api_server_sql.PipelineRunsApiService_Sql().create( + session, root_task=root_task, created_by="ada@example.com" + ) + + def test_inherited_cancel(self, session_factory, shared_with_sam) -> None: + run = self._run(session_factory) + service = api_server_sql.PipelineRunsApiService_Sql() + with session_factory() as session: + service.terminate(session, run.id, terminated_by="sam@example.com") + with session_factory() as session: + assert ( + session.get(bts.PipelineRun, run.id).extra_data["desired_state"] + == "TERMINATED" + ) + + def test_cancel_does_not_grant_annotate( + self, session_factory, shared_with_sam + ) -> None: + run = self._run(session_factory) + with session_factory() as session, pytest.raises(errors.PermissionError): + api_server_sql.PipelineRunsApiService_Sql().set_annotation( + session=session, + id=run.id, + key="k", + value="v", + user_name="sam@example.com", + ) + + def test_scopes_are_plain_strings(self, session_factory, shared_with_sam) -> None: + run = self._run(session_factory) + with session_factory() as session: + pipeline_run = session.get(bts.PipelineRun, run.id) + for email in ("ada@example.com", "sam@example.com"): + granted = runs.run_scopes( + session=session, + pipeline_run=pipeline_run, + caller=checks.Caller(email=email), + ) + assert {type(scope) for scope in granted} == {str} + + def test_stranger_cannot_cancel(self, session_factory) -> None: + run = self._run(session_factory) + with session_factory() as session, pytest.raises(errors.PermissionError): + api_server_sql.PipelineRunsApiService_Sql().terminate( + session, run.id, terminated_by="kai@example.com" + ) + + @pytest.mark.parametrize(("key", "allowed"), [("note", True), ("parent", False)]) + def test_grantee_cannot_change_parent_annotations( + self, session_factory, annotate_shared_with_sam, key, allowed + ) -> None: + run = self._run(session_factory) + service = api_server_sql.PipelineRunsApiService_Sql() + with session_factory() as session: + if allowed: + service.set_annotation( + session=session, + id=run.id, + key=key, + value="v", + user_name="sam@example.com", + ) + else: + with pytest.raises(errors.PermissionError): + service.set_annotation( + session=session, + id=run.id, + key=key, + value="v", + user_name="sam@example.com", + ) + + def test_starter_changes_parent_annotations( + self, session_factory, annotate_shared_with_sam + ) -> None: + run = self._run(session_factory) + service = api_server_sql.PipelineRunsApiService_Sql() + with session_factory() as session: + service.set_annotation( + session=session, + id=run.id, + key="parent", + value="v", + user_name="ada@example.com", + ) + with session_factory() as session: + service.delete_annotation( + session=session, id=run.id, key="parent", user_name="ada@example.com" + )