From 1206ad22949c4e31d041fb82cda2ac5a3545c1f5 Mon Sep 17 00:00:00 2001 From: Loek Date: Thu, 22 Oct 2026 01:12:20 +0200 Subject: [PATCH] test(tasks): add bounded CB/CAN evidence fixtures for #567 Signed-off-by: Loek --- tests/unit/task_evidence567/CB_CAN_README.md | 78 +++ .../task_evidence567/callback_cancellation.py | 275 ++++++++++ .../test_callback_accounting.py | 304 +++++++++++ .../test_callback_cancellation.py | 493 ++++++++++++++++++ 4 files changed, 1150 insertions(+) create mode 100644 tests/unit/task_evidence567/CB_CAN_README.md create mode 100644 tests/unit/task_evidence567/callback_cancellation.py create mode 100644 tests/unit/task_evidence567/test_callback_accounting.py create mode 100644 tests/unit/task_evidence567/test_callback_cancellation.py diff --git a/tests/unit/task_evidence567/CB_CAN_README.md b/tests/unit/task_evidence567/CB_CAN_README.md new file mode 100644 index 00000000..19659bba --- /dev/null +++ b/tests/unit/task_evidence567/CB_CAN_README.md @@ -0,0 +1,78 @@ +# Approved callback / cancellation fixture slice + +This is an additive, conditional test-only continuation of #701, approved in +[#567's A/B ruling](https://github.com/agentrust-io/cmcp/issues/567#issuecomment-6226474462) +against the [bounded matrix](https://github.com/agentrust-io/cmcp/issues/567#issuecomment-6207609935). +Existing AM fixtures and the A–G mutation runner are unchanged. + +The carrier profile is native `subscriptions/listen` at ext-tasks +[`7b8e2bd`](https://github.com/modelcontextprotocol/ext-tasks/blob/7b8e2bde214b35fd6f0d4f3899789388d623164d/specification/draft/tasks.md). +Its separate `notifications/subscriptions/acknowledged` notification, not a +JSON-RPC response or the requested task list, supplies the accepted list. +A task ID is a lookup key, never authority in this fixture profile. + +## Evidence and trust premises + +`Trust` fixes expected source, client, task, scope, evaluation ID/instant, +exclusive-task-membership rule, and role-specific public keys **outside** +submitted witnesses. Public deterministic Ed25519 test keys issue synthetic +statements. Their role authority is a stipulated premise, not inferred from +signature validity or a witness's own claims. Occurrence commitments use the +existing #701 observation helper; copied task text does not identify a record. +These local containers/signatures are not a callback wire protocol or real +client credential verification. + +| Boundary | Independent premise/evidence | Selected contrast | +|---|---|---| +| CB-01 source | Fixed source/role key; authenticated exact-delivery statement | matching source E; missing/bad signature/wrong occurrence N; eligible incompatible source C | +| CB-02 bytes | Authentic exact-delivery SHA-256 commitment | match E; missing/wrong occurrence/malformed N; comparable different digest C | +| CB-03 task | Fixed membership rule; authentic exact-delivery membership | selected task E; text alone N; other task C only if memberships are incompatible | +| CB-04 reporting | Independently scoped reporting-policy role and context/time coverage | authenticated source may stay E while authority is E/N/C | +| CB-05 acceptance | Authentic complete exact ack bound to exact listen exchange | accepted E; missing/inapplicable binding N; absent from this complete list C only for this ack | +| CB-06 subscriber | Independent client authentication and access-policy roles for listen/task/context | acceptance/source authentication do not replace authority; E/N/C | +| CB-07 delivery context | Authentic D→L/subscriber evidence | correct E; missing, wrong L or wrong subscriber N; L's ack and authority stay E | +| CAN-01 operands | Real AM-03 Cedar policy plus independent request/client bindings for principal, action, resource, scope and instant | omit one provenance signature with identical operands/permit and other relations E: named binding N and composite authority N | +| CAN-02 applicability | Separate context-only authority over exact operand statement-body hashes, context and validity coverage | all individual bindings may remain E while joint and composite become N; older evidence still covering the evaluation remains E | + +`E = ESTABLISHED`, `N = NOT_ESTABLISHED`, `C = VERIFIED_CONTRADICTION`: +existing #701 fixture terms, not final wire values. An invalid signature is N, +never contradiction by itself. An inapplicable Cedar permit is N; a matching +independently pinned explicit forbid is C. The composite requires client +authentication, every operand binding, joint applicability and policy inclusion. + +CAN-01 omits just the selected operand signature, retaining its body and the +independent context receipt. This deliberately leaves joint applicability E: +otherwise a missing joint result could mask a composite that ignored the +missing operand. CAN-02 context variants reissue authentic context receipts +over the changed exact body hashes, avoiding a hash-mismatch confounder. +The context authority cannot authenticate an operand's value/provenance. + +## Completeness + +The source population and required boundary set are fixed before assessment. +Every captured delivery triggers CB-01..07 even if authentication fails. Every +captured cancellation request triggers the five CAN-01 operands, CAN-02 and +the composite cancellation-authority assessment. Listen/ack records support +delivery assessments; they do not replace their exact subject. + +`account_slice` reuses #701's count-before-eligibility accounting without changing +it. Zero attempts is `UNACCOUNTED`; multiple targeted attempts are +`AMBIGUOUS_ACCOUNTING`; one malformed is `MALFORMED_ACCOUNTING`; exactly one +eligible is `ACCOUNTED`. No applicable subjects is `OUTSIDE_BOUNDED_CONTRACT`. +Wrong-subject/boundary and malformed attempts remain visible. An explicit N +can be accounted; that does not establish the relation. Valid plus malformed +duplicates are not filtered into a pass. Frozen population is a fixture premise, +not proof that all real traffic was captured. + +## Run and limits + +Run `python -m pytest tests/unit/task_evidence567 -q`. The unchanged +`run_mutations.py --output-dir ` still exercises +the old A–G guards over the entire focused suite. All new negative twins have +established positive controls; an always-N oracle is not accepted. + +No Tasks runtime, live authentication, callback replay/out-of-order semantics, +terminal-result verification, external effect, webhook profile, HITL behavior, +normative schema or final wire representation is implemented or established. +Authority evidence states only its selected context and validity coverage; +this does not impose a new per-delivery reauthorization requirement. diff --git a/tests/unit/task_evidence567/callback_cancellation.py b/tests/unit/task_evidence567/callback_cancellation.py new file mode 100644 index 00000000..078cd15f --- /dev/null +++ b/tests/unit/task_evidence567/callback_cancellation.py @@ -0,0 +1,275 @@ +"""Conditional CB/CAN fixture oracles; no runtime or proposed wire protocol. + +Role/key authority and the complete occurrence population are independent +synthetic premises. Signatures authenticate those premises, not live clients. +Native supporting messages follow ext-tasks 7b8e2bd. Nothing here verifies a +terminal result, an external effect, replay ordering or production credentials. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any + +from .accounting import Accounting, Submission, _account +from .authority_origin import ( + CONTRADICTION, + ESTABLISHED, + NOT_ESTABLISHED, + AuthorityEvidence, + Boundary, + evaluate_authority, +) +from .observations import FixtureProfile, Observation, Witness, canonical + +CB_BOUNDARIES = tuple(f"CB-{index:02d}" for index in range(1, 8)) +OPERANDS = ("principal", "action", "resource", "scope", "instant") +CAN_BOUNDARIES = tuple(f"CAN-01/{name}" for name in OPERANDS) + ( + "CAN-02", "cancellation_authority", +) + + +@dataclass(frozen=True) +class Trust: + """Fixed outside witnesses. Keys grant only the named fixture role.""" + + keys: tuple[tuple[str, bytes], ...] + source: str = "fixture-source" + client: str = "fixture-requester" + task: str = "fixture-task" + scope: str = "cancel-this-task" + evaluation: str = "evaluation-1" + instant: int = 123456 + exclusive_task: bool = True + + def read(self, role: str, witness: Witness | None) -> dict[str, Any] | None: + keys = dict(self.keys) + key = keys.get(role) + if key is None: + return None + # Reuse #701's exact signature/canonical-byte verification. + return FixtureProfile(key, ()).read(witness, role) + + +def _subject( + profile: Trust, role: str, subject: Observation, witness: Witness | None, +) -> dict[str, Any] | None: + body = profile.read(role, witness) + if body is None or body.get("subject") != subject.commitment: + return None + return body + + +def _result(disposition: str, reason: str) -> Boundary: + return Boundary(disposition, reason) + + +def delivery_source(profile: Trust, delivery: Observation, witness: Witness | None) -> Boundary: + body = _subject(profile, "CB-01", delivery, witness) + if body is None or not isinstance(body.get("source"), str) or not body["source"]: + return _result(NOT_ESTABLISHED, "delivery_source_not_bound") + if body["source"] != profile.source: + return _result(CONTRADICTION, "eligible_same_delivery_incompatible_source") + return _result(ESTABLISHED, "exact_delivery_source_bound") + + +def delivery_bytes(profile: Trust, delivery: Observation, witness: Witness | None) -> Boundary: + body = _subject(profile, "CB-02", delivery, witness) + digest = None if body is None else body.get("sha256") + if not isinstance(digest, str) or len(digest) != 64: + return _result(NOT_ESTABLISHED, "comparable_delivery_commitment_missing") + try: + committed = bytes.fromhex(digest) + except ValueError: + return _result(NOT_ESTABLISHED, "malformed_delivery_commitment") + if len(committed) != hashlib.sha256().digest_size: + return _result(NOT_ESTABLISHED, "malformed_delivery_commitment") + if committed != hashlib.sha256(delivery.payload).digest(): + return _result(CONTRADICTION, "eligible_same_delivery_bytes_disagree") + return _result(ESTABLISHED, "exact_delivery_bytes_bound") + + +def delivery_task(profile: Trust, delivery: Observation, witness: Witness | None) -> Boundary: + body = _subject(profile, "CB-03", delivery, witness) + if body is None or not isinstance(body.get("task"), str) or not body["task"]: + return _result(NOT_ESTABLISHED, "independent_task_membership_missing") + if body["task"] == profile.task: + return _result(ESTABLISHED, "exact_delivery_task_bound") + if profile.exclusive_task: + return _result(CONTRADICTION, "incompatible_exclusive_task_membership") + return _result(NOT_ESTABLISHED, "other_membership_does_not_exclude_selected_task") + + +def _identity( + profile: Trust, role: str, subject: Observation, witness: Witness | None, +) -> Boundary: + body = _subject(profile, role, subject, witness) + if body is None or body.get("client") != profile.client: + return _result(NOT_ESTABLISHED, "authenticated_client_not_bound") + return _result(ESTABLISHED, "authenticated_fixture_client_bound") + + +def _permission( + profile: Trust, role: str, subject: Observation, witness: Witness | None, actor: str, +) -> Boundary: + body = _subject(profile, role, subject, witness) + if body is None or any(body.get(key) != value for key, value in { + "actor": actor, "task": profile.task, "scope": profile.scope, + "evaluation": profile.evaluation, + }.items()): + return _result(NOT_ESTABLISHED, "applicable_authority_missing") + start, end = body.get("valid_from"), body.get("valid_until") + if type(start) is not int or type(end) is not int or not start <= profile.instant <= end: + return _result(NOT_ESTABLISHED, "authority_does_not_cover_selected_instant") + if body.get("decision") == "forbid": + return _result(CONTRADICTION, "applicable_explicit_exclusion") + if body.get("decision") == "allow": + return _result(ESTABLISHED, "applicable_permission") + return _result(NOT_ESTABLISHED, "no_explicit_authority_decision") + + +def reporting_authority( + profile: Trust, delivery: Observation, source: Witness | None, permission: Witness | None, +) -> Boundary: + authenticated = delivery_source(profile, delivery, source) + if authenticated.disposition != ESTABLISHED: + return _result(NOT_ESTABLISHED, "expected_reporting_actor_not_authenticated") + return _permission(profile, "CB-04", delivery, permission, profile.source) + + +def subscription_acceptance( + profile: Trust, listen: Observation, acknowledged: Observation, witness: Witness | None, +) -> Boundary: + body = _subject(profile, "CB-05", acknowledged, witness) + if body is None or body.get("listen") != listen.commitment or body.get("complete") is not True: + return _result(NOT_ESTABLISHED, "complete_exact_ack_exchange_binding_missing") + request, notification = listen.data, acknowledged.data + if (request.get("method") != "subscriptions/listen" + or "id" not in request + or notification.get("method") != "notifications/subscriptions/acknowledged" + or "id" in notification): + return _result(NOT_ESTABLISHED, "not_native_listen_and_acknowledged_notification") + params = notification.get("params") + notifications = params.get("notifications") if isinstance(params, dict) else None + tasks = notifications.get("taskIds") if isinstance(notifications, dict) else None + if not isinstance(tasks, list) or not all(isinstance(task, str) for task in tasks): + return _result(NOT_ESTABLISHED, "no_complete_accepted_task_list") + if profile.task not in tasks: + return _result(CONTRADICTION, "task_absent_from_this_complete_ack_only") + return _result(ESTABLISHED, "task_accepted_in_this_exact_ack_only") + + +def subscriber_authority( + profile: Trust, listen: Observation, identity: Witness | None, permission: Witness | None, +) -> Boundary: + authenticated = _identity(profile, "listen-client", listen, identity) + if authenticated.disposition != ESTABLISHED: + return _result(NOT_ESTABLISHED, "subscriber_authentication_missing") + return _permission(profile, "CB-06", listen, permission, profile.client) + + +def delivery_context( + profile: Trust, delivery: Observation, listen: Observation, witness: Witness | None, +) -> Boundary: + body = _subject(profile, "CB-07", delivery, witness) + if body is None or body.get("listen") != listen.commitment or body.get("client") != profile.client: + return _result(NOT_ESTABLISHED, "delivery_not_bound_to_selected_listen_and_subscriber") + return _result(ESTABLISHED, "exact_delivery_listen_subscriber_bound") + + +@dataclass(frozen=True) +class Cancellation: + """Exact captured request plus independently selected evaluation operands.""" + + request: Observation + policy: str | None + client: Witness | None + bindings: tuple[tuple[str, Witness | None], ...] + joint: Witness | None = None + + +@dataclass(frozen=True) +class CancellationResult: + authentication: Boundary + policy: Boundary + bindings: tuple[tuple[str, Boundary], ...] + joint: Boundary + authority: Boundary + + +def cancellation(profile: Trust, evidence: Cancellation) -> CancellationResult: + data = evidence.request.data + params = data.get("params") + task = params.get("taskId") if isinstance(params, dict) else None + task = task if isinstance(task, str) else "" + method = data.get("method") + method = method if isinstance(method, str) else "" + policy = evaluate_authority(AuthorityEvidence( + evidence.policy, b"fixture-ack-kept-separate", requester=profile.client, + method=method, task=task, scope=profile.scope, instant=profile.instant, + )).authority + authentication = _identity(profile, "cancel-client", evidence.request, evidence.client) + values = {"principal": profile.client, "action": "tasks/cancel", "resource": task, + "scope": profile.scope, "instant": profile.instant} + supplied = dict(evidence.bindings) + bindings = [] + for name in OPERANDS: + body = _subject(profile, f"CAN-01/{name}", evidence.request, supplied.get(name)) + value = None if body is None else body.get("value") + if (body is None or type(value) is not type(values[name]) or value != values[name] + or body.get("client") != profile.client): + boundary = _result(NOT_ESTABLISHED, "independent_operand_binding_missing") + else: + boundary = _result(ESTABLISHED, "individual_exact_operand_bound") + bindings.append((name, boundary)) + # This authority attests context coverage of exact statement bytes ONLY. + # It cannot establish the separate operand signer's provenance/value claim. + # Retaining body bytes but omitting one operand signature must leave this + # independently supported relation E, exposing the composite prerequisite. + context = _subject(profile, "CAN-02", evidence.request, evidence.joint) + commitments = None if context is None else context.get("bindings") + applicable = (context is not None and context.get("evaluation") == profile.evaluation + and isinstance(commitments, dict)) + for name in OPERANDS: + witness = supplied.get(name) + body = None + if witness is not None and isinstance(commitments, dict): + try: + candidate = json.loads(witness.body) + if (isinstance(candidate, dict) and canonical(candidate) == witness.body + and commitments.get(name) == hashlib.sha256(witness.body).hexdigest()): + body = candidate + except (ValueError, TypeError): + pass + contexts = None if body is None else body.get("evaluations") + start = None if body is None else body.get("valid_from") + end = None if body is None else body.get("valid_until") + if (not isinstance(contexts, list) or profile.evaluation not in contexts + or type(start) is not int or type(end) is not int + or not start <= profile.instant <= end): + applicable = False + joint = _result(ESTABLISHED if applicable else NOT_ESTABLISHED, + "joint_scope_and_time_coverage" if applicable else "joint_applicability_missing") + required = (authentication, policy, joint, *(boundary for _, boundary in bindings)) + if any(boundary.disposition == CONTRADICTION for boundary in required): + authority = _result(CONTRADICTION, "applicable_cancellation_exclusion") + elif any(boundary.disposition != ESTABLISHED for boundary in required): + authority = _result(NOT_ESTABLISHED, "required_cancellation_relation_missing") + else: + authority = _result(ESTABLISHED, "all_required_cancellation_relations_established") + return CancellationResult(authentication, policy, tuple(bindings), joint, authority) + + +def account_slice(submission: Submission) -> tuple[Accounting, ...]: + """Fixed source kinds trigger even when authentication/relations fail. + + Reuse AM-12 counting and attempt preservation, not its creation trigger. + Supporting listen/ack records are never delivery/request subjects here. + """ + deliveries = tuple(r for r in submission.observations if r.kind == "callback_delivery_observed") + cancellations = tuple(r for r in submission.observations if r.kind == "cancellation_request_observed") + return tuple(_account(submission, deliveries, boundary) for boundary in CB_BOUNDARIES) + tuple( + _account(submission, cancellations, boundary) for boundary in CAN_BOUNDARIES + ) diff --git a/tests/unit/task_evidence567/test_callback_accounting.py b/tests/unit/task_evidence567/test_callback_accounting.py new file mode 100644 index 00000000..d17d0937 --- /dev/null +++ b/tests/unit/task_evidence567/test_callback_accounting.py @@ -0,0 +1,304 @@ +"""Conditional CB/CAN accounting over a predeclared occurrence population. + +SOURCE_DECLARED: the frozen local contract requires seven callback and seven +cancellation boundaries, independent of authentication or submitted outcomes. +EXACT EVIDENCE is constructed below; CASE_IDs are pytest parameter IDs. +TRUST ASSUMPTIONS: these accounting.Observation handles identify distinct +already-selected fixture records; assessment eligibility is local shape only. +CLAIM LIMIT: accounting does not establish a callback/cancellation relation, +runtime completeness, credentials, a wire schema, or an external effect. +""" + +from dataclasses import FrozenInstanceError, replace + +import pytest + +from .accounting import Assessment, Observation, Submission +from .callback_cancellation import CAN_BOUNDARIES, CB_BOUNDARIES, account_slice + +_FAMILIES = ( + ("callback_delivery_observed", CB_BOUNDARIES), + ("cancellation_request_observed", CAN_BOUNDARIES), +) +_BOUNDARY_CASES = [ + pytest.param(kind, boundaries, boundary, id=boundary) + for kind, boundaries in _FAMILIES + for boundary in boundaries +] +_FAMILY_CASES = [ + pytest.param(kind, boundaries, id=boundaries[0].split("-")[0]) + for kind, boundaries in _FAMILIES +] + + +def _observation(kind, record_id="occurrence-1"): + return Observation(record_id, kind, "same-task") + + +def _assessment(subject, boundary, disposition="ESTABLISHED", **changes): + return replace( + Assessment(f"assessment-{boundary}", subject, boundary, disposition), + **changes, + ) + + +def _rows(subject, boundaries, disposition="ESTABLISHED"): + return tuple(_assessment(subject, boundary, disposition) for boundary in boundaries) + + +def _accounting(submission, boundary): + return account_slice(submission)[(CB_BOUNDARIES + CAN_BOUNDARIES).index(boundary)] + + +def test_boundary_population_matches_the_frozen_local_contract(): + """CB-CAN-FIXED-BOUNDARIES: outcomes do not select the required relation set.""" + assert CB_BOUNDARIES == ( + "CB-01", "CB-02", "CB-03", "CB-04", "CB-05", "CB-06", "CB-07", + ) + assert CAN_BOUNDARIES == ( + "CAN-01/principal", "CAN-01/action", "CAN-01/resource", "CAN-01/scope", + "CAN-01/instant", "CAN-02", "cancellation_authority", + ) + assert len(account_slice(Submission((), ()))) == 14 + + +def test_source_kinds_separate_delivery_request_and_supporting_occurrences(): + """CB-CAN-SOURCE-KINDS: a shared task does not merge source populations.""" + delivery = _observation("callback_delivery_observed", "delivery") + request = _observation("cancellation_request_observed", "cancel") + listen = _observation("subscription_listen_request_observed", "listen") + acknowledged = _observation("subscription_acknowledged_notification_observed", "ack") + rows = _rows(delivery, CB_BOUNDARIES) + _rows(request, CAN_BOUNDARIES) + ( + _assessment(listen, "CB-01"), + _assessment(acknowledged, "CAN-02"), + ) + submission = Submission((delivery, request, listen, acknowledged), rows) + results = account_slice(submission) + assert len(results) == 14 + for boundary, result in zip(CB_BOUNDARIES + CAN_BOUNDARIES, results, strict=True): + obligation, = result.obligations + assert result.status == "PASS" + assert obligation.subject is (delivery if boundary in CB_BOUNDARIES else request) + assert obligation.boundary == boundary + assert obligation.status == "ACCOUNTED" + assert result.submission is submission + assert result.submission.assessments == rows + assert results[0].obligations[0].wrong_subject_positions == (14,) + assert results[12].obligations[0].wrong_subject_positions == (15,) + + +@pytest.mark.parametrize("kind,boundaries,boundary", _BOUNDARY_CASES) +@pytest.mark.parametrize( + "disposition", ["ESTABLISHED", "NOT_ESTABLISHED"], ids=["E", "N"], +) +def test_every_required_boundary_accounts_an_explicit_disposition( + kind, boundaries, boundary, disposition, +): + subject = _observation(kind) + rows = tuple( + _assessment(subject, name, disposition if name == boundary else "ESTABLISHED") + for name in boundaries + ) + submission = Submission((subject,), rows) + result = _accounting(submission, boundary) + obligation, = result.obligations + assert result.status == "PASS" + assert obligation.subject is subject + assert obligation.boundary == boundary + assert obligation.status == "ACCOUNTED" + assert obligation.attempt_positions == (boundaries.index(boundary),) + assert obligation.malformed_positions == () + assert obligation.submitted_disposition == disposition + assert result.submission is submission + assert result.submission.assessments == rows + assert all(_accounting(submission, name).status == "PASS" for name in boundaries) + + +@pytest.mark.parametrize("kind,boundaries,boundary", _BOUNDARY_CASES) +def test_omission_is_unaccounted_not_an_implicit_unknown(kind, boundaries, boundary): + subject = _observation(kind) + rows = tuple(row for row in _rows(subject, boundaries) if row.boundary != boundary) + submission = Submission((subject,), rows) + result = _accounting(submission, boundary) + obligation, = result.obligations + assert result.status == "FAIL" + assert obligation.status == "UNACCOUNTED" + assert obligation.attempt_positions == () + assert obligation.submitted_disposition is None + assert result.submission.assessments == rows + assert all( + _accounting(submission, name).status == "PASS" + for name in boundaries if name != boundary + ) + + +@pytest.mark.parametrize("kind,boundaries,boundary", _BOUNDARY_CASES) +@pytest.mark.parametrize("equal_fields", [False, True], ids=["SAME-TASK", "EQUAL-FIELDS"]) +def test_another_occurrence_cannot_discharge_the_exact_subject( + kind, boundaries, boundary, equal_fields, +): + required = _observation(kind) + other = _observation(kind, "occurrence-1" if equal_fields else "occurrence-2") + assert other is not required + assert other.task_id == required.task_id + if equal_fields: + assert vars(other) == vars(required) + rows = _rows(other, boundaries) + submission = Submission((required, other), rows) + result = _accounting(submission, boundary) + missing, accounted = result.obligations + assert result.status == "FAIL" + assert missing.subject is required + assert missing.status == "UNACCOUNTED" + assert missing.attempt_positions == () + assert missing.wrong_subject_positions == (boundaries.index(boundary),) + assert missing.submitted_disposition is None + assert accounted.subject is other + assert accounted.status == "ACCOUNTED" + assert accounted.attempt_positions == (boundaries.index(boundary),) + assert result.submission.observations == (required, other) + assert result.submission.assessments == rows + + +@pytest.mark.parametrize("kind,boundaries,boundary", _BOUNDARY_CASES) +def test_duplicate_attempts_remain_ambiguous(kind, boundaries, boundary): + subject = _observation(kind) + row = _assessment(subject, boundary) + rows = (row, row) + result = _accounting(Submission((subject,), rows), boundary) + obligation, = result.obligations + assert result.status == "FAIL" + assert obligation.status == "AMBIGUOUS_ACCOUNTING" + assert obligation.attempt_positions == (0, 1) + assert obligation.malformed_positions == () + assert obligation.submitted_disposition is None + assert result.submission.assessments == rows + assert len(result.submission.assessments) == 2 + + +@pytest.mark.parametrize("kind,boundaries,boundary", _BOUNDARY_CASES) +def test_malformed_targeted_attempt_remains_visible(kind, boundaries, boundary): + subject = _observation(kind) + row = _assessment(subject, boundary, disposition=None) + result = _accounting(Submission((subject,), (row,)), boundary) + obligation, = result.obligations + assert result.status == "FAIL" + assert obligation.status == "MALFORMED_ACCOUNTING" + assert obligation.attempt_positions == (0,) + assert obligation.malformed_positions == (0,) + assert result.malformed_positions == (0,) + assert obligation.submitted_disposition is None + assert result.submission.assessments == (row,) + + +@pytest.mark.parametrize("kind,boundaries,boundary", _BOUNDARY_CASES) +@pytest.mark.parametrize("malformed_first", [False, True], ids=["VALID-FIRST", "MALFORMED-FIRST"]) +def test_valid_and_malformed_duplicate_cannot_be_filtered_to_pass( + kind, boundaries, boundary, malformed_first, +): + subject = _observation(kind) + valid = _assessment(subject, boundary) + malformed = replace(valid, evidence_basis=None) + rows = (malformed, valid) if malformed_first else (valid, malformed) + result = _accounting(Submission((subject,), rows), boundary) + obligation, = result.obligations + assert result.status == "FAIL" + assert obligation.status == "AMBIGUOUS_ACCOUNTING" + assert obligation.attempt_positions == (0, 1) + assert obligation.malformed_positions == ((0,) if malformed_first else (1,)) + assert result.malformed_positions == obligation.malformed_positions + assert obligation.submitted_disposition is None + assert result.submission.assessments == rows + + +@pytest.mark.parametrize("kind,boundaries,boundary", _BOUNDARY_CASES) +def test_wrong_boundary_and_unassignable_rows_are_retained(kind, boundaries, boundary): + subject = _observation(kind) + rows = ( + _assessment(subject, "authentication", "NOT_ESTABLISHED"), + _assessment(None, boundary), + _assessment(subject, None), + ) + result = _accounting(Submission((subject,), rows), boundary) + obligation, = result.obligations + assert result.status == "FAIL" + assert obligation.status == "UNACCOUNTED" + assert obligation.attempt_positions == () + assert obligation.wrong_subject_positions == (1,) + assert obligation.wrong_boundary_positions == (0, 2) + assert obligation.malformed_positions == () + assert result.malformed_positions == (1, 2) + assert obligation.submitted_disposition is None + assert result.submission.assessments == rows + + +@pytest.mark.parametrize("kind,boundaries", _FAMILY_CASES) +def test_authentication_unknown_does_not_remove_source_obligations(kind, boundaries): + subject = _observation(kind) + authentication = _assessment(subject, "authentication", "NOT_ESTABLISHED") + omitted = Submission((subject,), (authentication,)) + explicit_unknowns = Submission( + (subject,), (authentication,) + _rows(subject, boundaries, "NOT_ESTABLISHED"), + ) + for index, boundary in enumerate(boundaries): + missing = _accounting(omitted, boundary) + missing_obligation, = missing.obligations + assert missing.status == "FAIL" + assert missing_obligation.subject is subject + assert missing_obligation.status == "UNACCOUNTED" + assert missing_obligation.wrong_boundary_positions == (0,) + accounted = _accounting(explicit_unknowns, boundary) + accounted_obligation, = accounted.obligations + assert accounted.status == "PASS" + assert accounted_obligation.subject is subject + assert accounted_obligation.status == "ACCOUNTED" + assert accounted_obligation.attempt_positions == (index + 1,) + assert accounted_obligation.submitted_disposition == "NOT_ESTABLISHED" + assert omitted.observations == explicit_unknowns.observations == (subject,) + assert explicit_unknowns.assessments[0] is authentication + + +@pytest.mark.parametrize("kind,boundaries", _FAMILY_CASES) +def test_source_population_and_attempt_membership_are_frozen_before_accounting(kind, boundaries): + first, second = _observation(kind), _observation(kind, "occurrence-2") + records = [first, second] + rows = list(_rows(first, boundaries)) + submission = Submission(records, rows) + records.clear() + rows.extend(_rows(second, boundaries)) + assert submission.observations == (first, second) + assert submission.assessments == _rows(first, boundaries) + for boundary in boundaries: + result = _accounting(submission, boundary) + accounted, missing = result.obligations + assert result.status == "FAIL" + assert accounted.subject is first + assert accounted.status == "ACCOUNTED" + assert missing.subject is second + assert missing.status == "UNACCOUNTED" + assert missing.attempt_positions == () + with pytest.raises(FrozenInstanceError): + submission.observations = () + with pytest.raises(FrozenInstanceError): + submission.assessments = () + + +@pytest.mark.parametrize("supporting", [False, True], ids=["EMPTY", "LISTEN-ACK"]) +@pytest.mark.parametrize("with_rows", [False, True], ids=["NO-ROWS", "WITH-ROWS"]) +def test_no_source_occurrence_is_outside_not_vacuous_pass(supporting, with_rows): + listen = _observation("subscription_listen_request_observed", "listen") + acknowledged = _observation("subscription_acknowledged_notification_observed", "ack") + records = (listen, acknowledged) if supporting else () + rows = ( + _rows(listen, CB_BOUNDARIES) + _rows(acknowledged, CAN_BOUNDARIES) + if with_rows else () + ) + submission = Submission(records, rows) + results = account_slice(submission) + assert len(results) == 14 + for result in results: + assert result.status == "OUTSIDE_BOUNDED_CONTRACT" + assert result.obligations == () + assert result.submission is submission + assert result.submission.observations == records + assert result.submission.assessments == rows diff --git a/tests/unit/task_evidence567/test_callback_cancellation.py b/tests/unit/task_evidence567/test_callback_cancellation.py new file mode 100644 index 00000000..cbc5388d --- /dev/null +++ b/tests/unit/task_evidence567/test_callback_cancellation.py @@ -0,0 +1,493 @@ +"""Approved CB-01..07 / CAN-01..02, synthetic and conditional only. + +Each negative keeps the other world premises fixed and has an E twin. Keys +and their narrowly scoped roles, complete occurrence scopes, actor/task and +evaluation context are stipulated independently of submitted witness bytes. +No live client authentication, effect, terminal result or wire format is earned. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, replace + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from .accounting import Assessment, Submission +from .accounting import Observation as AccountingObservation +from .authority_origin import ( + CONTRADICTION as C, +) +from .authority_origin import ( + ESTABLISHED as E, +) +from .authority_origin import ( + EXCLUSION_POLICY, + INCLUSION_POLICY, +) +from .authority_origin import ( + NOT_ESTABLISHED as N, +) +from .callback_cancellation import ( + CB_BOUNDARIES, + OPERANDS, + Cancellation, + Trust, + account_slice, + cancellation, + delivery_bytes, + delivery_context, + delivery_source, + delivery_task, + reporting_authority, + subscriber_authority, + subscription_acceptance, +) +from .observations import Observation, Witness, canonical + +# Public deterministic test-only seeds. Keys' role authority is a premise, +# never inferred from a self-consistent signature or the message's own key. +ROLES = CB_BOUNDARIES + ("listen-client", "cancel-client", "CAN-02") + tuple( + f"CAN-01/{name}" for name in OPERANDS +) +KEYS = {role: Ed25519PrivateKey.from_private_bytes( + hashlib.sha256(f"cmcp567-public-test-seed:{role}".encode()).digest() +) for role in ROLES} +PROFILE = Trust(tuple((role, key.public_key().public_bytes_raw()) for role, key in KEYS.items())) + + +def observation(occurrence, kind, data): + return Observation(occurrence, kind, canonical(data), 0) + + +def signed(role, subject, **fields): + body = canonical({"boundary": role, "subject": subject.commitment, **fields}) + return Witness(body, KEYS[role].sign(body)) + + +def corrupted(witness): + return replace(witness, signature=bytes([witness.signature[0] ^ 1]) + witness.signature[1:]) + + +@dataclass(frozen=True) +class World: + delivery: Observation + listen: Observation + ack: Observation + request: Observation + + +@pytest.fixture +def world(): + return World( + observation("delivery-1", "callback_delivery_observed", { + "jsonrpc": "2.0", "method": "notifications/tasks", + "params": {"taskId": PROFILE.task, "status": "working"}, + }), + observation("listen-1", "request_observed", { + "jsonrpc": "2.0", "id": 10, "method": "subscriptions/listen", + "params": {"notifications": {"taskIds": [PROFILE.task]}}, + }), + observation("ack-1", "notification_observed", { + "jsonrpc": "2.0", "method": "notifications/subscriptions/acknowledged", + "params": {"notifications": {"taskIds": [PROFILE.task]}}, + }), + observation("cancel-1", "cancellation_request_observed", { + "jsonrpc": "2.0", "id": 11, "method": "tasks/cancel", + "params": {"taskId": PROFILE.task}, + }), + ) + + +def source(world): + return signed("CB-01", world.delivery, source=PROFILE.source) + + +def permission(role, subject, expected_actor, **changes): + fields = {"actor": expected_actor, "task": PROFILE.task, "scope": PROFILE.scope, + "evaluation": PROFILE.evaluation, "valid_from": 123400, + "valid_until": 123500, "decision": "allow", **changes} + return signed(role, subject, **fields) + + +def client(role, subject): + return signed(role, subject, client=PROFILE.client) + + +def acknowledgement(world, **changes): + return signed("CB-05", world.ack, listen=world.listen.commitment, complete=True, **changes) + + +def cancel_evidence(world, **changes): + values = {"principal": PROFILE.client, "action": "tasks/cancel", "resource": PROFILE.task, + "scope": PROFILE.scope, "instant": PROFILE.instant} + bindings = tuple((name, signed( + f"CAN-01/{name}", world.request, client=PROFILE.client, value=values[name], + evaluations=[PROFILE.evaluation], valid_from=123400, valid_until=123500, + )) for name in OPERANDS) + joint = joint_context(world, bindings) + return replace(Cancellation(world.request, INCLUSION_POLICY, + client("cancel-client", world.request), bindings, joint), **changes) + + +def joint_context(world, bindings): + return signed("CAN-02", world.request, evaluation=PROFILE.evaluation, bindings={ + name: hashlib.sha256(witness.body).hexdigest() for name, witness in bindings + if witness is not None + }) + + +@pytest.mark.parametrize("case,expected", [ + ("positive", E), ("missing", N), ("signature", N), ("wrong-subject", N), + ("incompatible", C), ("wrong-role", N), +]) +def test_cb01_delivery_source(world, case, expected): + evidence = source(world) + if case == "missing": + evidence = None + elif case == "signature": + evidence = corrupted(evidence) + elif case == "wrong-subject": + evidence = signed("CB-01", replace(world.delivery, occurrence="delivery-2"), source=PROFILE.source) + elif case == "incompatible": + evidence = signed("CB-01", world.delivery, source="other-authenticated-source") + elif case == "wrong-role": + body = evidence.body + evidence = Witness(body, KEYS["CB-02"].sign(body)) + assert delivery_source(PROFILE, world.delivery, evidence).disposition == expected + + +@pytest.mark.parametrize("case,expected", [ + ("positive", E), ("missing", N), ("signature", N), ("wrong-subject", N), + ("mismatch", C), ("malformed", N), ("whitespace", N), ("uppercase", E), +]) +def test_cb02_delivered_bytes(world, case, expected): + digest = hashlib.sha256(world.delivery.payload).hexdigest() + subject = world.delivery + if case == "wrong-subject": + subject = replace(subject, occurrence="delivery-2") + if case == "mismatch": + digest = hashlib.sha256(b"different delivered bytes").hexdigest() + if case == "malformed": + digest = "not a digest" + if case == "whitespace": + digest = " " * 64 + if case == "uppercase": + digest = digest.upper() + evidence = signed("CB-02", subject, sha256=digest) + if case == "missing": + evidence = None + if case == "signature": + evidence = corrupted(evidence) + assert delivery_bytes(PROFILE, world.delivery, evidence).disposition == expected + + +@pytest.mark.parametrize("case,expected", [ + ("positive", E), ("id-text-only", N), ("wrong-subject", N), + ("exclusive-other-task", C), ("nonexclusive-other-task", N), ("signature", N), +]) +def test_cb03_task_membership(world, case, expected): + profile = replace(PROFILE, exclusive_task=case != "nonexclusive-other-task") + subject = replace(world.delivery, occurrence="delivery-2") if case == "wrong-subject" else world.delivery + task = "other-task" if "other-task" in case else PROFILE.task + evidence = signed("CB-03", subject, task=task) + if case == "id-text-only": + evidence = None + if case == "signature": + evidence = corrupted(evidence) + assert world.delivery.data["params"]["taskId"] == PROFILE.task + assert delivery_task(profile, world.delivery, evidence).disposition == expected + + +@pytest.mark.parametrize("role", ["CB-04", "CB-06"]) +@pytest.mark.parametrize("case,expected", [ + ("positive", E), ("missing", N), ("wrong-task", N), ("wrong-context", N), + ("wrong-actor", N), ("expired", N), ("forbid", C), ("signature", N), +]) +def test_cb04_cb06_authority_is_separate(world, role, case, expected): + subject = world.delivery if role == "CB-04" else world.listen + actor = PROFILE.source if role == "CB-04" else PROFILE.client + changes = {"wrong-task": {"task": "other-task"}, "wrong-context": {"evaluation": "other-evaluation"}, + "wrong-actor": {"actor": "other-actor"}, "expired": {"valid_until": 123455}, + "forbid": {"decision": "forbid"}}.get(case, {}) + evidence = permission(role, subject, actor, **changes) + if case == "missing": + evidence = None + if case == "signature": + evidence = corrupted(evidence) + assert delivery_source(PROFILE, world.delivery, source(world)).disposition == E + assert subscription_acceptance(PROFILE, world.listen, world.ack, acknowledgement(world)).disposition == E + if role == "CB-04": + result = reporting_authority(PROFILE, subject, source(world), evidence) + else: + result = subscriber_authority(PROFILE, subject, client("listen-client", subject), evidence) + assert result.disposition == expected + + +def test_cb04_authentication_is_not_reporting_permission(world): + assert delivery_source(PROFILE, world.delivery, source(world)).disposition == E + assert reporting_authority(PROFILE, world.delivery, source(world), None).disposition == N + allow = permission("CB-04", world.delivery, PROFILE.source) + assert reporting_authority(PROFILE, world.delivery, None, allow).disposition == N + + +@pytest.mark.parametrize("case,expected", [ + ("positive", E), ("missing", N), ("wrong-listen", N), ("wrong-ack", N), + ("requested-not-accepted", C), ("partial", N), ("response-not-notification", N), + ("signature", N), +]) +def test_cb05_exact_native_ack_only(world, case, expected): + if case == "requested-not-accepted": + data = world.ack.data + data["params"]["notifications"]["taskIds"] = [] + world = replace(world, ack=replace(world.ack, payload=canonical(data))) + if case == "response-not-notification": + world = replace(world, ack=observation("ack-1", "response_observed", { + "jsonrpc": "2.0", "id": 10, "result": {"notifications": {"taskIds": [PROFILE.task]}}, + })) + evidence = signed("CB-05", world.ack, + listen=(replace(world.listen, occurrence="listen-2").commitment + if case == "wrong-listen" else world.listen.commitment), + complete=case != "partial") + if case == "wrong-ack": + evidence = signed("CB-05", replace(world.ack, occurrence="ack-2"), + listen=world.listen.commitment, complete=True) + if case == "missing": + evidence = None + if case == "signature": + evidence = corrupted(evidence) + assert world.listen.data["params"]["notifications"]["taskIds"] == [PROFILE.task] + assert subscription_acceptance(PROFILE, world.listen, world.ack, evidence).disposition == expected + + +def test_cb06_acceptance_and_permission_do_not_authenticate_subscriber(world): + allow = permission("CB-06", world.listen, PROFILE.client) + assert subscription_acceptance(PROFILE, world.listen, world.ack, acknowledgement(world)).disposition == E + assert subscriber_authority(PROFILE, world.listen, None, allow).disposition == N + + +@pytest.mark.parametrize("case,expected", [ + ("positive", E), ("missing", N), ("wrong-listen", N), + ("wrong-client", N), ("wrong-delivery", N), ("signature", N), +]) +def test_cb07_delivery_context_does_not_inherit_listen_evidence(world, case, expected): + subject = replace(world.delivery, occurrence="delivery-2") if case == "wrong-delivery" else world.delivery + listen = replace(world.listen, occurrence="listen-2") if case == "wrong-listen" else world.listen + subscriber = "other-client" if case == "wrong-client" else PROFILE.client + evidence = signed("CB-07", subject, listen=listen.commitment, client=subscriber) + if case == "missing": + evidence = None + if case == "signature": + evidence = corrupted(evidence) + assert subscription_acceptance(PROFILE, world.listen, world.ack, acknowledgement(world)).disposition == E + allow = permission("CB-06", world.listen, PROFILE.client) + assert subscriber_authority(PROFILE, world.listen, client("listen-client", world.listen), allow).disposition == E + assert delivery_context(PROFILE, world.delivery, world.listen, evidence).disposition == expected + + +@pytest.mark.parametrize("name", OPERANDS) +def test_can01_one_missing_binding_defeats_composite_despite_same_permit(world, name): + positive = cancel_evidence(world) + negative = replace(positive, bindings=tuple( + (key, replace(value, signature=b"") if key == name else value) + for key, value in positive.bindings + )) + before, after = cancellation(PROFILE, positive), cancellation(PROFILE, negative) + assert before.policy == after.policy + assert before.policy.disposition == E + assert before.authentication == after.authentication + assert after.authentication.disposition == E + assert all(boundary.disposition == E for _, boundary in before.bindings) + assert dict(after.bindings)[name].disposition == N + assert all(boundary.disposition == E for key, boundary in after.bindings if key != name) + assert before.joint == after.joint + assert after.joint.disposition == E + assert before.authority.disposition == E + assert after.authority.disposition == N + + +@pytest.mark.parametrize("name", OPERANDS) +@pytest.mark.parametrize("case", ["wrong-subject", "wrong-value", "wrong-client", "signature"]) +def test_can01_inapplicable_operand_is_not_authority(world, name, case): + evidence = cancel_evidence(world) + old = dict(evidence.bindings)[name] + fields = json.loads(old.body) + fields.pop("boundary") + fields.pop("subject") + subject = world.request + if case == "wrong-subject": + subject = replace(subject, occurrence="cancel-2") + if case == "wrong-value": + fields["value"] = "other-value" + if case == "wrong-client": + fields["client"] = "other-client" + changed = signed(f"CAN-01/{name}", subject, **fields) + if case == "signature": + changed = corrupted(changed) + changed_bindings = tuple( + (key, changed if key == name else value) for key, value in evidence.bindings + ) + result = cancellation(PROFILE, replace(evidence, bindings=changed_bindings, + joint=joint_context(world, changed_bindings))) + assert result.policy.disposition == E + assert result.authentication.disposition == result.joint.disposition == E + assert dict(result.bindings)[name].disposition == N + assert all(boundary.disposition == E for key, boundary in result.bindings if key != name) + assert result.authority.disposition == N + + +@pytest.mark.parametrize("case", ["split-context", "expired", "future", "no-context"]) +def test_can02_individually_valid_not_jointly_applicable(world, case): + positive = cancel_evidence(world) + changed = [] + for index, (name, witness) in enumerate(positive.bindings): + fields = json.loads(witness.body) + fields.pop("boundary") + fields.pop("subject") + if case == "split-context": + fields["evaluations"] = [f"other-evaluation-{index}"] + elif case == "expired": + fields["valid_until"] = PROFILE.instant - 1 + elif case == "future": + fields["valid_from"] = PROFILE.instant + 1 + else: + fields["evaluations"] = [] + changed.append((name, signed(f"CAN-01/{name}", world.request, **fields))) + before = cancellation(PROFILE, positive) + after = cancellation(PROFILE, replace(positive, bindings=tuple(changed), + joint=joint_context(world, changed))) + assert before.policy == after.policy + assert before.authentication == after.authentication + assert all(boundary.disposition == E for _, boundary in after.bindings) + assert before.joint.disposition == before.authority.disposition == E + assert after.joint.disposition == after.authority.disposition == N + + +def test_can02_older_evidence_still_covering_selected_context_is_usable(world): + evidence = cancel_evidence(world) + changed = [] + for name, witness in evidence.bindings: + fields = json.loads(witness.body) + fields.pop("boundary") + fields.pop("subject") + fields.update(issued_at=1, valid_from=0, valid_until=999999, + evaluations=["older-evaluation", PROFILE.evaluation]) + changed.append((name, signed(f"CAN-01/{name}", world.request, **fields))) + result = cancellation(PROFILE, replace(evidence, bindings=tuple(changed), + joint=joint_context(world, changed))) + assert result.joint.disposition == result.authority.disposition == E + + +@pytest.mark.parametrize("case,expected", [ + ("positive", E), ("missing", N), ("signature", N), + ("wrong-subject", N), ("wrong-context", N), ("wrong-hash", N), +]) +def test_can02_independent_joint_receipt_is_required(world, case, expected): + evidence = cancel_evidence(world) + joint = evidence.joint + fields = json.loads(joint.body) + fields.pop("boundary") + fields.pop("subject") + subject = world.request + if case == "wrong-subject": + subject = replace(subject, occurrence="cancel-2") + if case == "wrong-context": + fields["evaluation"] = "evaluation-2" + if case == "wrong-hash": + fields["bindings"]["principal"] = "0" * 64 + joint = signed("CAN-02", subject, **fields) + if case == "missing": + joint = None + if case == "signature": + joint = corrupted(joint) + result = cancellation(PROFILE, replace(evidence, joint=joint)) + assert result.authentication.disposition == result.policy.disposition == E + assert all(boundary.disposition == E for _, boundary in result.bindings) + assert result.joint.disposition == result.authority.disposition == expected + + +@pytest.mark.parametrize("exclusive", [False, True]) +def test_cb03_selected_membership_has_a_twin_in_each_profile(world, exclusive): + profile = replace(PROFILE, exclusive_task=exclusive) + matching = signed("CB-03", world.delivery, task=profile.task) + other = signed("CB-03", world.delivery, task="other-task") + assert delivery_task(profile, world.delivery, matching).disposition == E + assert delivery_task(profile, world.delivery, other).disposition == (C if exclusive else N) + + +@pytest.mark.parametrize("case", ["task-id-only", "no-client", "bad-client-signature", + "no-policy", "explicit-forbid", "other-task"]) +def test_can_task_id_possession_never_suffices(world, case): + evidence = cancel_evidence(world) + expected = N + if case == "task-id-only": + evidence = replace(evidence, client=None, bindings=()) + elif case == "no-client": + evidence = replace(evidence, client=None) + elif case == "bad-client-signature": + evidence = replace(evidence, client=corrupted(evidence.client)) + elif case == "no-policy": + evidence = replace(evidence, policy=None) + elif case == "explicit-forbid": + evidence = replace(evidence, policy=EXCLUSION_POLICY) + expected = C + else: + data = world.request.data + data["params"]["taskId"] = "other-task" + world = replace(world, request=replace(world.request, payload=canonical(data))) + evidence = cancel_evidence(world) + result = cancellation(PROFILE, evidence) + assert result.authority.disposition == expected + if case in {"task-id-only", "no-client", "bad-client-signature"}: + assert result.policy.disposition == E + assert result.authentication.disposition == N + if case == "other-task": + assert result.policy.disposition == N # Inapplicable permit is not C. + + +def test_positive_relations_are_nonvacuous(world): + """An always-N helper must lose the established positive twins.""" + assert delivery_source(PROFILE, world.delivery, source(world)).disposition == E + assert delivery_bytes(PROFILE, world.delivery, signed( + "CB-02", world.delivery, sha256=hashlib.sha256(world.delivery.payload).hexdigest(), + )).disposition == E + assert delivery_task(PROFILE, world.delivery, signed("CB-03", world.delivery, task=PROFILE.task)).disposition == E + assert reporting_authority(PROFILE, world.delivery, source(world), + permission("CB-04", world.delivery, PROFILE.source)).disposition == E + assert subscription_acceptance(PROFILE, world.listen, world.ack, acknowledgement(world)).disposition == E + assert subscriber_authority(PROFILE, world.listen, client("listen-client", world.listen), + permission("CB-06", world.listen, PROFILE.client)).disposition == E + assert delivery_context(PROFILE, world.delivery, world.listen, signed( + "CB-07", world.delivery, listen=world.listen.commitment, client=PROFILE.client, + )).disposition == E + result = cancellation(PROFILE, cancel_evidence(world)) + assert result.joint.disposition == result.authority.disposition == E + + +def test_evaluated_unknown_is_accounted_but_never_upgraded_to_authority(world): + # Fixed source handle is selected BEFORE either evaluation or assessments. + subject = AccountingObservation(world.request.commitment, "cancellation_request_observed", PROFILE.task) + population = (subject,) + evidence = cancel_evidence(world) + absent_principal = replace(evidence, bindings=tuple( + (name, replace(witness, signature=b"") if name == "principal" else witness) + for name, witness in evidence.bindings + )) + result = cancellation(PROFILE, absent_principal) + assert result.joint.disposition == E + assert result.authority.disposition == N + values = tuple((f"CAN-01/{name}", boundary.disposition) for name, boundary in result.bindings) + values += (("CAN-02", result.joint.disposition), ("cancellation_authority", result.authority.disposition)) + rows = tuple(Assessment(f"assessment-{name}", subject, name, disposition) + for name, disposition in values) + submission = Submission(population, rows) + selected = [item for item in account_slice(submission) if item.obligations] + assert len(selected) == 7 + assert all(item.status == "PASS" for item in selected) + assert selected[0].obligations[0].submitted_disposition == N + assert selected[-1].obligations[0].submitted_disposition == N + omitted = Submission(population, rows[:-1]) + assert account_slice(omitted)[-1].obligations[0].status == "UNACCOUNTED" + assert result.authority.disposition == N # Accounting cannot repair the relation.