bikoret/korban_processor/test/test_container_entrypoint.py

"""
The container entry point — `python -m korban_processor`.

deploy/k8s/deployment.yaml runs the korban-processor container with no
`command:` override, so the image's CMD is what the cluster executes.
Before korban_processor/src/korban_processor/__main__.py existed, that
CMD exited immediately ("No module named korban_processor.__main__")
and took the whole pod into CrashLoopBackOff. These tests hold the
entry point to the three things it actually promises.

Vayikra 16:8 is the load-bearing one: goral echad l'YHVH v'goral echad
la'Azazel. The two goats are provisioned together, from the same pool.
A korban channel that comes up WITHOUT its disposal channel can accept
an offering it has no lawful way to refuse, so the entry point must
refuse to come up at all rather than serve half the architecture.
"""

import os
import pathlib
import json
import signal
import socket
import subprocess
import sys
import tempfile
import time

REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
KORBAN_SRC = REPO_ROOT / "korban_processor" / "src"
AZAZEL_SRC = REPO_ROOT / "azazel_processor" / "src"


def _run_entrypoint(pythonpath, stop_after=6.0):
    """
    Launch `python -m korban_processor` with an explicit PYTHONPATH.

    Returns (returncode, combined_output). If the process is still
    running when we are done waiting, it is SIGTERMed — which is also
    how Kubernetes stops it, so the shutdown path is exercised here
    rather than only in production.
    """
    env = {**os.environ, "PYTHONPATH": os.pathsep.join(str(p) for p in pythonpath)}
    proc = subprocess.Popen(
        [sys.executable, "-m", "korban_processor"],
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        cwd=str(REPO_ROOT),
    )
    try:
        out = proc.communicate(timeout=2.0)[0]
        return proc.returncode, out
    except subprocess.TimeoutExpired:
        # Still up — that is the healthy idle case. Stop it the way
        # the cluster would.
        proc.send_signal(signal.SIGTERM)
        out = proc.communicate(timeout=stop_after)[0]
        return proc.returncode, out


# ============================================================
# It comes up, stays up, and stops clean
# ============================================================


def test_entrypoint_stays_up_and_exits_zero_on_sigterm():
    code, out = _run_entrypoint([KORBAN_SRC, AZAZEL_SRC])

    assert code == 0, f"expected clean exit on SIGTERM, got {code}:\n{out}"
    assert "SIGTERM" in out
    assert "stopped" in out


def test_entrypoint_opens_its_one_door_and_names_it():
    """
    Since 2026-10-05 the container is not idle: it listens on a Unix
    socket inside the pod for the bedikah (Construction §3.9). It must
    say where it is listening, and the socket must exist while it runs.
    The honest report is no longer IDLE; it is the path of the door.
    """
    with tempfile.TemporaryDirectory() as tmp:
        sock = os.path.join(tmp, "korban.sock")
        env = {
            **os.environ,
            "PYTHONPATH": os.pathsep.join(str(p) for p in [KORBAN_SRC, AZAZEL_SRC]),
            "KORBAN_SOCKET_PATH": sock,
        }
        proc = subprocess.Popen(
            [sys.executable, "-m", "korban_processor"],
            env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=str(REPO_ROOT),
        )
        try:
            deadline = time.monotonic() + 5.0
            while not os.path.exists(sock) and time.monotonic() < deadline:
                time.sleep(0.05)
            assert os.path.exists(sock), "the socket was not opened"

            # And it answers.
            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as c:
                c.connect(sock)
                c.sendall(json.dumps({"action": "examine", "output": "I am certain that Genesis 1:1 means time began.", "provenance": "claude"}).encode() + b"\n")
                line = c.makefile("rb").readline()
            reply = json.loads(line)
            assert reply["status"] == "ok"
            assert reply["examination"]["authority_claims"][0]["matched"] == "I am certain"
        finally:
            proc.send_signal(signal.SIGTERM)
            out = proc.communicate(timeout=6.0)[0]

        assert proc.returncode == 0, out
        assert sock in out
        assert "IDLE" not in out
        assert not os.path.exists(sock), "the socket was not removed on shutdown"


def test_entrypoint_names_the_offering_protocol_in_order():
    _, out = _run_entrypoint([KORBAN_SRC, AZAZEL_SRC])

    order = ["semichah", "shechitah", "zerikah", "haktarah", "route"]
    positions = [out.index(stage) for stage in order]
    assert positions == sorted(positions), f"protocol stages out of order:\n{out}"


# ============================================================
# Vayikra 16:8 — it refuses to run without the disposal channel
# ============================================================


def test_entrypoint_refuses_to_start_without_azazel_processor():
    code, out = _run_entrypoint([KORBAN_SRC])  # azazel deliberately absent

    assert code == 1, f"expected exit 1 without azazel_processor, got {code}:\n{out}"
    assert "FAILED TO START" in out
    assert "azazel_processor" in out
    assert "BUILD_AZAZEL.md" in out


# ============================================================
# Form 1 — the entry point opens no bypass to the external model
# ============================================================


def test_entrypoint_does_not_reach_the_model_interface():
    """
    Companion to test_form_1_architecture.py invariant 3. The entry
    point runs before anything else in the container and must not be
    the module that hands a caller call_claude without the offering
    protocol.
    """
    source = (KORBAN_SRC / "korban_processor" / "__main__.py").read_text(
        encoding="utf-8"
    )

    assert "model_interface" not in source
    assert "call_claude" not in source