orchestrata_ultimate.promotion.identity_skips

Reference › orchestrata_ultimate › promotion › identity_skips

The rules that exclude a discovered unit from the identity check with a stated reason, and never silently. canvas_skip_reason excludes the canvas service by name (ruling R8: it is not a deploy-root unit); the name check runs before any systemctl show. not_runnable_skip_reason excludes a unit whose Requires= or BindsTo= names a unit that systemctl cat cannot load, using requires_bindsto_names from unit_deps, the same rule the restart stage applies. resolve_skip combines them for one unit and returns its properties, a skip reason or a show error, with exactly one of the last two set. Skips are listed in the report.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/promotion/identity_skips.py

"""Unit-scope skip rules for ``identity.check_identity`` (U-VERIFY1).

Some discovered units are unit-SCOPE exclusions, not identity failures, and
the verifier's scope has to match the installer it verifies:

* ``orchestrata-canvas.service`` -- ADR-0062 (R1+R5) read together with
  ruling R8 (arch-canvas stays in ``~/.local/share``;
  ``deploy/systemd/install.sh`` already skips this unit on that ruling) means
  it is not an ``/opt/orchestrata`` unit at all. Decided by NAME alone,
  before any ``systemctl show``/``/proc`` inspection of the unit.
* A unit whose ``Requires=``/``BindsTo=`` names a unit ``systemctl [--user]
  cat`` cannot load is "not runnable here" -- the same rule
  ``promotion/stages_units.py::daemon_units`` (U-UNIT2) already applies
  before restarting a unit, reusing ``unit_deps.requires_bindsto_names`` for
  the parsing and the caller's own injected ``RunFn`` seam for the probe
  (never a fake ``systemctl``).

Neither rule ever turns a genuine failure into a pass: a unit that is simply
inactive for its own reasons is still judged, and still FAILS, exactly as
before -- these functions only decide whether a unit is judged at all.
``resolve_skip`` is ``identity.check_identity``'s single entry point: it
calls ``identity_scope.show`` itself (the one ``systemctl show`` per unit,
never a second round-trip) so the caller sees one three-way result.
"""

from __future__ import annotations

from collections.abc import Sequence
from pathlib import Path

from orchestrata_ultimate.promotion.identity_scope import RunFn, show
from orchestrata_ultimate.promotion.unit_deps import requires_bindsto_names

__all__ = ["canvas_skip_reason", "not_runnable_skip_reason", "resolve_skip"]

CANVAS_UNIT_NAME = "orchestrata-canvas.service"


def canvas_skip_reason(name: str) -> str | None:
    """R8: ``orchestrata-canvas.service`` is excluded from the
    ``/opt/orchestrata`` identity scope outright -- checked by unit NAME
    alone, so a caller can apply this before calling ``systemctl show`` (or
    touching ``/proc``) for the unit at all."""
    if name != CANVAS_UNIT_NAME:
        return None
    return "ruling R8: not an /opt/orchestrata unit"


def not_runnable_skip_reason(
    properties: dict[str, str], run: RunFn, scope: str
) -> str | None:
    """U-UNIT2's own rule, reused: a unit whose ``Requires=``/``BindsTo=``
    names a unit ``systemctl [--user] cat`` cannot load is not runnable
    here. ``properties["FragmentPath"]`` is the real unit file already
    fetched by ``systemctl show`` (never a module-level subprocess of its
    own); each dependency name is probed through the caller's injected
    ``run`` seam, exactly like ``daemon_units``'s own ``_run([*prefix,
    "cat", dep])`` check."""
    fragment_path = properties.get("FragmentPath", "")
    if not fragment_path:
        return None
    flags = ["--user"] if scope == "user" else []
    for dep in requires_bindsto_names(Path(fragment_path)):
        result = run(["systemctl", *flags, "cat", dep])
        if result.returncode != 0:
            return f"not runnable here: required unit {dep} is not loadable"
    return None


def resolve_skip(
    name: str, scope: str, run: RunFn, unit_properties: Sequence[str]
) -> tuple[dict[str, str], str | None, str | None]:
    """One discovered unit -> ``(properties, skip_reason, show_error)``,
    exactly one of the last two ever set. The canvas rule never costs a
    ``systemctl show`` round-trip; the not-runnable-here rule needs
    ``properties`` (for ``FragmentPath``), so it runs right after ``show``
    succeeds and before the caller does anything else with them."""
    canvas_reason = canvas_skip_reason(name)
    if canvas_reason is not None:
        return {}, canvas_reason, None
    properties, error = show(run, scope, name, unit_properties)
    if error is not None:
        return {}, None, error
    not_runnable_reason = not_runnable_skip_reason(properties, run, scope)
    if not_runnable_reason is not None:
        return {}, not_runnable_reason, None
    return properties, None, None