azazel_processor: commit 1 — skeleton + stages (properties 11 & 12)

Pass 4 / Finding F, commit 1 of 10 from BUILD_AZAZEL.md. Lays the
load-bearing foundation: classification discipline (property 11) at
the type system level, and unmanaged-destination opacity (property
12) via a Python sentinel. Every later commit in the Azazel build
inherits these defenses for free.

CREATED:

  azazel_processor/
    pyproject.toml                      — Poetry config matching
                                          korban_processor; pythonpath
                                          includes ../korban_processor/src
                                          so the cross-package
                                          classification-discipline test
                                          can import both packages
    src/azazel_processor/
      __init__.py                       — public surface for commit 1:
                                          AzazelStage, AzazelStaged,
                                          AzazelStageError, ERETZ_GEZERAH,
                                          is_eretz_gezerah
      eretz_gezerah.py                  — the destination sentinel; module
                                          __getattr__ refuses any
                                          attribute beyond the two public
                                          names so introspection cannot
                                          discover further handles
      stages.py                          — AzazelStage enum (six
                                          checkpoints from the build doc:
                                          STAGED → CONFESSED → ROUTED →
                                          DESTROYED → ACKNOWLEDGED →
                                          RECORDED), AzazelStaged frozen
                                          dataclass, AzazelStageError
    test/
      test_eretz_gezerah.py             — 11 tests covering: sentinel
                                          identity stable across imports,
                                          not-a-string, no public attrs,
                                          repr leaks no destination info,
                                          identity (is) not equality (==),
                                          impostor with __eq__ rejected,
                                          module __getattr__ refuses
                                          unknown access, CI invariant
                                          forbidding destination-handle
                                          ASSIGNMENTS in source (catches
                                          regressions; mention in
                                          docstrings/__all__/getattr
                                          allowlist is fine)
      test_stages.py                    — 10 tests: intra-Azazel gating
                                          (happy path, mismatch raises
                                          naming both stages, frozen
                                          dataclass, six distinct
                                          checkpoints) and CROSS-PACKAGE
                                          classification discipline:
                                          KorbanStaged and AzazelStaged
                                          are distinct types; enum value
                                          spaces are disjoint; an Azazel
                                          function refuses a KorbanStaged
                                          input and vice versa; no
                                          conversion function exists in
                                          the public surface

PROPERTY COVERAGE THIS COMMIT:

  Property 11 (classification discipline):
    Defended by the type-distinctness of AzazelStaged vs.
    korban_processor.stages.Staged. Cross-package tests verify the
    two cannot be substituted at runtime; the absence of any
    korban_to_azazel/to_korban/from_korban/convert function in the
    public surface is verified by name-shape grep.

  Property 12 (unmanaged destination):
    Defended by the sentinel design: only object identity is exposed;
    no string handle exists; module __getattr__ refuses introspection;
    CI invariant catches assignment-shaped regressions across all of
    azazel_processor/src/.

PENDING (commits 2-10 per BUILD_AZAZEL.md):

  vidui (commit 2 — enumerative commit, property 3)
  shaliach + tevilah (commit 3 — role-separated operator, property 4)
  goral (commit 4 — blind routing, property 2)
  azazel orchestrator + state machine (commit 5 — properties 5, 7, 10)
  protocol envelope (commit 6 — protocol-restricted access, property 8)
  L3 azazel_records (commit 7 — operator-internal half of property 9)
  /peah/azazel_log endpoint (commit 8 — public half of property 9)
  Korban→Azazel routing on empty-vital + chatzotzrot wiring
    (commit 9 — properties 1, 6)
  Closure pass (commit 10)

Tests: azazel_processor 21 passed. korban_processor unchanged at
203 passed, 1 skipped (live-API). No regressions across the boundary.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Scott KarpandClaude Opus 4.7 committed 2026-04-27 16:07:44 -04:00
1 parent 1fb7d33124
commit d06ecd244d
7 files changed
+666

No files matched your search

+24
View File
@@ -0,0 +1,24 @@
[project]
name = "azazel-processor"
version = "0.1.0"
description = "Azazel Processor — disposal channel parallel to korban_processor (the two-goat architecture, Vayikra 16:8)"
authors = [
{name = "Bezalel"}
]
requires-python = "==3.12.2"
dependencies = [
"pydantic==2.6.1",
]
[dependency-groups]
dev = [
"pytest>=8.0,<9.0",
]
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.pytest.ini_options]
testpaths = ["test"]
pythonpath = ["src", "../korban_processor/src"]
@@ -0,0 +1,41 @@
"""
azazel_processor — the disposal channel parallel to korban_processor.
Vayikra 16 — the two-goat architecture (shavim b'mareh uv'komah
uv'd'mim, Yoma 62a). One goat to HaShem (processing); one goat to
Azazel (disposal). The two channels are provisioned at parity, from
the same pool, with assignment blind to the operator.
Build state: commit 1 — Skeleton + stages. See BUILD_AZAZEL.md for
the full 10-commit staging plan and the 12-property → code mapping
from azazel_engineering_reading.md §14.
This commit covers:
- Property 11 (classification discipline) — AzazelStage and
AzazelStaged are non-interconvertible with korban_processor.
Staged.
- Property 12 (unmanaged destination) — ERETZ_GEZERAH sentinel
is the only legitimate way to refer to the disposal destination.
Public surface so far:
AzazelStage, AzazelStaged, AzazelStageError — stage-gating types
ERETZ_GEZERAH, is_eretz_gezerah — the destination
Pending in subsequent commits:
vidui (commit 2 — enumerative commit)
shaliach + tevilah (commit 3 — role-separated operator)
goral (commit 4 — blind routing)
azazel orchestrator (commit 5 — full state machine)
protocol envelope (commit 6 — protocol-restricted access)
"""
from azazel_processor.eretz_gezerah import ERETZ_GEZERAH, is_eretz_gezerah
from azazel_processor.stages import AzazelStage, AzazelStaged, AzazelStageError
__all__ = [
"ERETZ_GEZERAH",
"is_eretz_gezerah",
"AzazelStage",
"AzazelStaged",
"AzazelStageError",
]
@@ -0,0 +1,93 @@
"""
Eretz Gezerah — the unmanaged destination.
Vayikra 16:22 — וְשִׁלַּח אֶת־הַשָּׂעִיר בַּמִּדְבָּר … אֶל־אֶרֶץ גְּזֵרָה.
And he shall send the goat into the wilderness … to a cut-off land.
The word `gezerah` is the same word Torah uses for binding judgment
(`gezerah shavah`, the second of R. Yishmael's 13 middot, is the
hermeneutic-decree principle). The destination is *cut off* — severed,
decreed, not-coming-back.
This module defines the only legitimate way the codebase refers to the
disposal destination: a Python sentinel object with no attributes, no
identifier, no name that encodes anything about where state actually
goes after disposal. The opacity is the specification (property 13 of
azazel_engineering_reading.md).
WHY A SENTINEL AND NOT A STRING:
If the destination had a string identifier — "void", "/dev/null",
"oblivion", a URL, a queue name, a hostname — that string would become
a handle. Operators would reach for it, verify arrival with it, build
relationships with it, optimize against it. That is exactly what the
Mishkan two-goat architecture is designed to prevent: coupling between
the system and its disposal sink. A coupled sink is a managed sink; a
managed sink ceases to be a sink and becomes a queue. The disposal
channel that quietly accumulates state is the architectural failure
mode.
A sentinel `object()` has no attributes, no `__repr__` that leaks
location info, no string conversion that helps an operator track it.
You can compare to it with `is` (identity check); you cannot do
anything else with it. That is the maximal opacity Python permits.
WHY NOT A STRING-CONSTANT WITH AN OPAQUE NAME:
A string is still tracked by string identity across the codebase. Two
files importing the same string constant share that string in the
intern table; tooling can grep for it; logs can record it. A sentinel
is created once at import time and identified only by its memory
address — which changes per process, cannot be serialized meaningfully,
and resists tooling.
The CI invariant that defends property 12: the codebase MUST NOT
contain any string identifier, URL pattern, queue name, or hostname
representing the destination. The only path to "the destination" is
`from azazel_processor.eretz_gezerah import ERETZ_GEZERAH`. A future
contributor who tries to add `DESTINATION_URL = "..."` to any module
in `azazel_processor/` is reintroducing the failure mode the sentinel
was created to prevent.
USE:
from azazel_processor.eretz_gezerah import ERETZ_GEZERAH, is_eretz_gezerah
record.routing = ERETZ_GEZERAH
assert is_eretz_gezerah(record.routing)
Comparison MUST use identity (`is`), not equality (`==`). Identity
guarantees you are pointing at the canonical sentinel; equality could
be satisfied by any object with the right `__eq__`, defeating the
opacity guarantee.
"""
# The sentinel itself. A plain object() — no class, no attributes,
# no identifier. Created once at module import time.
ERETZ_GEZERAH = object()
def is_eretz_gezerah(value) -> bool:
"""Identity check against the canonical sentinel.
Always uses `is`, never `==`. The identity check is what
guarantees opacity — equality would let any object impersonate
the destination, defeating property 12.
"""
return value is ERETZ_GEZERAH
def __getattr__(name):
"""Refuse attribute access to anything other than the two
public names. The sentinel has no further structure; module
introspection should not be able to discover any.
"""
if name in {"ERETZ_GEZERAH", "is_eretz_gezerah"}:
return globals()[name]
raise AttributeError(
f"eretz_gezerah has no attribute {name!r}. The destination "
f"is opaque by specification — there are no further handles "
f"to discover. (Vayikra 16:22, property 12 of "
f"azazel_engineering_reading.md §14.)"
)
@@ -0,0 +1,117 @@
"""
Azazel stage gating — role-attribution for the disposal protocol.
Parallel to korban_processor/stages.py for the processing protocol,
but intentionally a DIFFERENT type. Property 11 of
azazel_engineering_reading.md §14: processing and disposal state are
not interconvertible. The type system is what enforces this — a
`Staged[KorbanStage.X]` cannot be passed to an Azazel function and an
`AzazelStaged[AzazelStage.X]` cannot be passed to a Korban function.
The classification discipline lives in the type system, not in
runtime checks the operator could disable.
THE SIX CHECKPOINTS (from BUILD_AZAZEL.md commit 5):
STAGED — input received, classification known, disposal protocol
entered (the goat-pair has been provisioned and the
lot has fallen on this one for la'Azazel)
CONFESSED — vidui constructed; what is being disposed has been
named (Vayikra 16:21, v'hitvadah alav)
ROUTED — destination decision committed (always ERETZ_GEZERAH);
routing record attached
DESTROYED — payload reference cleared from memory; the disposed
thing no longer exists (eretz gezerah, no return)
ACKNOWLEDGED — out-of-band acknowledgment fired (chatzotzrot
:moadim signal sent); the public fact of disposal
is now visible (property 9, Rambam's
observability requirement)
RECORDED — L3 azazel_record persisted; the disposal is now
part of permanent oral tradition
Each checkpoint is a recoverable point — if the orchestrator halts
mid-route, restart resumes from the last checkpoint reached.
WHY NOT REUSE korban_processor.stages.Staged?
A reused type would let `Staged[KorbanStage.SHECHITAH]` be passed
into an Azazel function unchecked. The compression would re-introduce
the calf-shape: state classified for processing being silently
re-routed into disposal, or vice versa. The two protocols MUST be
expressible as non-interconvertible types. Future contributors who
want to "deduplicate" the two staging modules are deleting the
defense; the duplication is load-bearing.
"""
from dataclasses import dataclass
from enum import Enum
from typing import Generic, TypeVar
class AzazelStage(Enum):
"""The checkpoints of the disposal protocol.
NOTE: this enum is DELIBERATELY distinct from korban_processor.
KorbanStage. They are not interchangeable; a function that accepts
one cannot accept the other. The type system enforces classification
discipline (property 11).
"""
STAGED = "staged"
CONFESSED = "confessed"
ROUTED = "routed"
DESTROYED = "destroyed"
ACKNOWLEDGED = "acknowledged"
RECORDED = "recorded"
T = TypeVar("T")
class AzazelStageError(RuntimeError):
"""Raised when an Azazel checkpoint is invoked with input from the
wrong upstream checkpoint, or when a value bearing a Korban-side
stage marker is passed to an Azazel function.
The Mishkan analog: the shaliach received the goat from the wrong
place — not from the kohen at the prior checkpoint, but from
somewhere else entirely. The error names BOTH the marker actually
present and the marker the function expected so the operator can
diagnose call-site vs. tag-site.
"""
@dataclass(frozen=True)
class AzazelStaged(Generic[T]):
"""A payload tagged with the disposal-protocol checkpoint that
produced it.
The marker is set at construction and cannot be mutated afterward
(`frozen=True`). There is no path to relabeling a payload as
belonging to a checkpoint it did not actually pass through.
DISTINCT TYPE FROM korban_processor.stages.Staged. The two are
structurally identical Python dataclasses, but the type system
treats them as unrelated — by intent, not by accident. See module
docstring.
"""
stage: AzazelStage
payload: T
def expect_stage(self, expected: AzazelStage) -> T:
"""Return the payload iff the stage marker matches `expected`.
Raises AzazelStageError otherwise. The error message names
BOTH stages so the operator can see exactly which transition
was violated.
"""
if self.stage != expected:
raise AzazelStageError(
f"Azazel stage mismatch: payload bears stage "
f"{self.stage.value!r} but the next disposal-protocol "
f"checkpoint requires stage {expected.value!r}. The "
f"shaliach receives the goat from the prior "
f"checkpoint's hand; an offering that skipped a "
f"checkpoint cannot be received."
)
return self.payload
View File
Whitespace-only changes.
+209
View File
@@ -0,0 +1,209 @@
"""
Tests for the eretz_gezerah sentinel — property 12 (unmanaged
destination opacity) from azazel_engineering_reading.md §14.
The destination MUST be opaque. There is no name, no URL, no queue
identifier, no string handle that encodes anything about where state
goes after disposal. The only legitimate reference is the sentinel
ERETZ_GEZERAH; identity comparison (`is`) is the only legitimate
predicate.
"""
import pathlib
import pytest
from azazel_processor.eretz_gezerah import ERETZ_GEZERAH, is_eretz_gezerah
# ============================================================
# The sentinel itself
# ============================================================
def test_sentinel_exists():
"""ERETZ_GEZERAH is importable and is the same object across imports."""
from azazel_processor.eretz_gezerah import ERETZ_GEZERAH as second_import
assert ERETZ_GEZERAH is second_import
def test_sentinel_is_a_plain_object_not_a_string():
"""A string would be searchable, interned, serializable. The sentinel
must NOT be a string. Property 12: opacity of name."""
assert not isinstance(ERETZ_GEZERAH, str)
assert not isinstance(ERETZ_GEZERAH, bytes)
def test_sentinel_has_no_meaningful_attributes():
"""A bare object() has no useful attributes for tooling to
discover. We don't grant the sentinel a name, a destination
string, or any other tracking handle."""
# Plain object() exposes only the dunder methods. No public attrs.
public = [a for a in dir(ERETZ_GEZERAH) if not a.startswith("_")]
assert public == []
def test_sentinel_repr_does_not_leak_destination_info():
"""The repr is just object()'s default — no operator-readable name
encoding 'where it goes'. Default Python repr is `<object object at
0x...>` which leaks only the address (changes per process)."""
r = repr(ERETZ_GEZERAH)
assert "<object object at 0x" in r
assert "azazel" not in r.lower()
assert "void" not in r.lower()
assert "null" not in r.lower()
# ============================================================
# is_eretz_gezerah/1
# ============================================================
def test_is_eretz_gezerah_recognizes_canonical_sentinel():
assert is_eretz_gezerah(ERETZ_GEZERAH)
def test_is_eretz_gezerah_rejects_other_objects():
assert not is_eretz_gezerah(None)
assert not is_eretz_gezerah(object())
assert not is_eretz_gezerah("eretz_gezerah")
assert not is_eretz_gezerah("ERETZ_GEZERAH")
assert not is_eretz_gezerah(0)
assert not is_eretz_gezerah([])
def test_is_eretz_gezerah_uses_identity_not_equality():
"""A class with __eq__ that returns True for anything must not
be confused with the sentinel. Identity is the gate; equality
would let any impostor in."""
class Impostor:
def __eq__(self, other):
return True
def __hash__(self):
return 0
impostor = Impostor()
assert impostor == ERETZ_GEZERAH # equality is True (impostor lies)
assert not is_eretz_gezerah(impostor) # but identity is not
# ============================================================
# Module-level __getattr__ refuses introspection
# ============================================================
def test_module_refuses_unknown_attribute_access():
"""The module exposes only ERETZ_GEZERAH and is_eretz_gezerah.
Any other attribute access must raise — there are no further
handles to discover."""
import azazel_processor.eretz_gezerah as mod
with pytest.raises(AttributeError) as exc:
mod.DESTINATION_URL # noqa: B018 — deliberate access for the test
assert "no further handles" in str(exc.value).lower() or \
"opaque" in str(exc.value).lower()
def test_module_refuses_invented_metadata_lookup():
import azazel_processor.eretz_gezerah as mod
for invented in ("destination", "endpoint", "queue_name", "host", "address"):
with pytest.raises(AttributeError):
getattr(mod, invented)
# ============================================================
# CI-style invariant: no destination string in the codebase
# ============================================================
#
# This is the grep-the-codebase check from BUILD_AZAZEL.md §12.
# It runs as a normal test so a regression at any layer of the
# build (a future contributor adding `DESTINATION_URL = "..."` to
# any module) fires immediately.
# Forbidden ASSIGNMENT-SHAPED patterns. The failure mode is someone
# adding a module-level constant that reintroduces a destination
# handle (e.g. `DESTINATION_URL = "..."`). Mere mention in a
# docstring, an `__all__` re-export, or a `__getattr__` allowlist is
# NOT the failure mode; what we defend against is binding a forbidden
# name to a value at module scope.
import re
_FORBIDDEN_ASSIGNMENT_PATTERNS = [
re.compile(rf"^\s*{name}\s*=", re.MULTILINE)
for name in (
"DESTINATION_URL",
"DESTINATION_HOST",
"AZAZEL_QUEUE",
"AZAZEL_ENDPOINT",
"DISPOSAL_HOST",
"DISPOSAL_URL",
"DISPOSAL_QUEUE",
"AZAZEL_RECIPIENT",
"DISPOSAL_RECIPIENT",
)
]
def _walk_source_files():
"""Yield every .py file under src/ — the source the operator has
to read to understand the channel. Tests are excluded (test
files legitimately reference forbidden patterns in assertions)."""
src_root = pathlib.Path(__file__).resolve().parent.parent / "src"
yield from src_root.rglob("*.py")
def test_no_destination_handle_assigned_in_source():
"""Property 12 enforced as a CI invariant. If a future contributor
adds `DESTINATION_URL = "..."` (or any of the other forbidden
handle-shaped names) to ANY module in azazel_processor/, this
test fails — they reintroduced the failure mode the sentinel
exists to prevent.
Catches assignment, not mention. Documenting "do not add
DESTINATION_URL" in a docstring is fine; `DESTINATION_URL = "x"`
is the regression.
"""
offenders = []
for path in _walk_source_files():
text = path.read_text()
for pattern in _FORBIDDEN_ASSIGNMENT_PATTERNS:
for match in pattern.finditer(text):
offenders.append((path.name, match.group().strip()))
assert offenders == [], (
f"Forbidden destination-handle assignment(s) in azazel_processor "
f"source: {offenders}. The destination is opaque by specification "
f"(property 12, azazel_engineering_reading.md §14). Use "
f"ERETZ_GEZERAH (the Python sentinel), not a string-valued "
f"module constant."
)
def test_sentinel_name_not_assigned_to_string_in_source():
"""A module-level `X = 'ERETZ_GEZERAH'` would treat the sentinel
name as if it were a string handle — re-introducing the very
string-tracking the sentinel exists to prevent. Mere mention in
`__all__` or `__getattr__`'s allowlist is fine; binding a name
to the string is not.
"""
string_assignment = re.compile(
r"""^\s*\w+\s*=\s*["']ERETZ_GEZERAH["']""", re.MULTILINE
)
offenders = []
for path in _walk_source_files():
text = path.read_text()
if string_assignment.search(text):
offenders.append(path.name)
assert offenders == [], (
f"Module-level assignment binding the string 'ERETZ_GEZERAH' "
f"to a name in: {offenders}. The sentinel is a Python object, "
f"not a string. Refer to it via the imported name, not via a "
f"string handle."
)
+182
View File
@@ -0,0 +1,182 @@
"""
Tests for the Azazel stage gating — properties 11 (classification
discipline) and the basic gating semantics that mirror
korban_processor/test/test_stages.py.
Two layers of test:
1. INTRA-AZAZEL: stage-mismatch within the disposal protocol
(the kohen receives the goat from the wrong checkpoint).
2. CROSS-PACKAGE: a Korban-side staged value cannot be used as
an Azazel-side staged value, and vice versa. Property 11:
processing and disposal are non-interconvertible at the type
level. The two structurally identical dataclasses are the same
SHAPE but the type system treats them as unrelated.
The cross-package test is the load-bearing one. Every later commit
in the Azazel build inherits its protection: if AzazelStaged and
korban_processor.stages.Staged were the same type (or interconvertible
via subtype), every downstream function would have to add runtime
checks to defend the boundary. Because they are distinct types, the
defense is automatic.
"""
import pytest
from azazel_processor.stages import (
AzazelStage,
AzazelStaged,
AzazelStageError,
)
# ============================================================
# Intra-Azazel: stage gating within the disposal protocol
# ============================================================
def test_expect_stage_returns_payload_when_marker_matches():
s = AzazelStaged(AzazelStage.STAGED, "some-input")
assert s.expect_stage(AzazelStage.STAGED) == "some-input"
def test_expect_stage_raises_when_marker_does_not_match():
s = AzazelStaged(AzazelStage.STAGED, "some-input")
with pytest.raises(AzazelStageError) as exc:
s.expect_stage(AzazelStage.CONFESSED)
msg = str(exc.value)
# Both stages named — the operator can diagnose call-site vs. tag-site.
assert "staged" in msg
assert "confessed" in msg
def test_azazel_staged_is_frozen():
"""The marker cannot be re-labeled after construction. There is
no path to taking a RECORDED-tagged value and re-tagging it as
STAGED to bypass the gate."""
s = AzazelStaged(AzazelStage.STAGED, "some-input")
with pytest.raises((AttributeError, Exception)):
s.stage = AzazelStage.RECORDED # type: ignore[misc]
def test_all_six_checkpoints_are_distinct():
"""STAGED → CONFESSED → ROUTED → DESTROYED → ACKNOWLEDGED →
RECORDED. No collapse, no aliases."""
stages = [
AzazelStage.STAGED,
AzazelStage.CONFESSED,
AzazelStage.ROUTED,
AzazelStage.DESTROYED,
AzazelStage.ACKNOWLEDGED,
AzazelStage.RECORDED,
]
assert len(set(stages)) == 6
def test_stage_error_names_both_stages():
s = AzazelStaged(AzazelStage.ROUTED, "p")
with pytest.raises(AzazelStageError) as exc:
s.expect_stage(AzazelStage.STAGED)
msg = str(exc.value)
assert "routed" in msg
assert "staged" in msg
# ============================================================
# Cross-package: classification discipline (property 11)
# ============================================================
#
# These tests import from BOTH korban_processor and azazel_processor
# to verify the two staging types are non-interconvertible. The test
# is feasible because pyproject.toml's pythonpath includes both
# packages' src directories.
def test_korban_staged_is_a_distinct_type_from_azazel_staged():
"""Even though both are dataclasses with the same field shape,
they are distinct Python types. isinstance/type checks see them
as unrelated."""
from korban_processor.stages import KorbanStage, Staged as KorbanStaged
k = KorbanStaged(KorbanStage.SEMICHAH, "a-korban")
a = AzazelStaged(AzazelStage.STAGED, "a-disposal-input")
assert type(k) is not type(a)
assert not isinstance(k, AzazelStaged)
assert not isinstance(a, KorbanStaged)
def test_korban_stage_enum_distinct_from_azazel_stage_enum():
from korban_processor.stages import KorbanStage
# The two enums share NO members — the names that DO overlap
# by coincidence (none currently) would still be distinct
# enum identities.
korban_values = {s.value for s in KorbanStage}
azazel_values = {s.value for s in AzazelStage}
# No accidental overlap in stage names. If a future contributor
# adds a stage that collides, this fires — and they should pick
# a different name to keep the two channels lexically distinct.
assert korban_values.isdisjoint(azazel_values), (
f"KorbanStage and AzazelStage share value(s): "
f"{korban_values & azazel_values}. The two protocols should "
f"have lexically distinct stage names so logs and call sites "
f"cannot accidentally treat one as the other."
)
def test_azazel_function_refuses_korban_staged_value():
"""The structural test of property 11: an Azazel function cannot
receive a KorbanStaged value. This is enforced via expect_stage's
AttributeError when called on the wrong dataclass — KorbanStaged
has expect_stage too (it's the same method name) but its stage
field carries a KorbanStage value, which `==` against an
AzazelStage member is always False, raising AzazelStageError."""
from korban_processor.stages import KorbanStage, Staged as KorbanStaged
korban_value = KorbanStaged(KorbanStage.SHECHITAH, "wrong-channel-payload")
# An Azazel function would call .expect_stage(AzazelStage.STAGED)
# on this value. KorbanStage.SHECHITAH != AzazelStage.STAGED
# (different enums, never equal), so expect_stage raises.
with pytest.raises(Exception):
# We deliberately cast through Any-ish to mimic the misuse;
# at runtime this fires.
korban_value.expect_stage(AzazelStage.STAGED) # type: ignore[arg-type]
def test_korban_function_refuses_azazel_staged_value():
"""The reverse direction: a Korban function cannot receive an
AzazelStaged value."""
from korban_processor.stages import KorbanStage
azazel_value = AzazelStaged(AzazelStage.STAGED, "wrong-channel-payload")
with pytest.raises(Exception):
azazel_value.expect_stage(KorbanStage.SEMICHAH) # type: ignore[arg-type]
def test_no_korban_to_azazel_conversion_function_exists():
"""There must be no `korban_to_azazel` or similar conversion
function in azazel_processor. If one is added, this test fails —
classification discipline is being deliberately broken."""
import azazel_processor
public = [name for name in dir(azazel_processor) if not name.startswith("_")]
forbidden_substrings = ("korban_to", "to_korban", "from_korban", "convert")
offenders = [
name
for name in public
for sub in forbidden_substrings
if sub in name.lower()
]
assert offenders == [], (
f"azazel_processor exposes conversion-shaped names: {offenders}. "
f"Property 11 (classification discipline): processing and disposal "
f"state are not interconvertible. There must be no path from one "
f"channel to the other."
)