orisys-app/shared/tests/test_module_boundaries.py

"""The module boundaries the responsibility split is supposed to create.

A pure move is only worth anything if the dependency directions actually change:
the finger worker must not need the box worker to describe a packet, the session
file format must not need the sampling state machine, and the models must not
import either. These assertions are cheap and they fail loudly if someone
re-introduces the cycle.
"""

from __future__ import annotations

import ast
from pathlib import Path

SHARED = Path(__file__).resolve().parents[1] / "src"
ACQUISITION = SHARED / "orisys_app_shared" / "acquisition"
FORCE = SHARED / "orisys_app_shared" / "force"
RECORD = SHARED / "orisys_app_shared" / "record"


def _imports(path: Path) -> set[str]:
    """Every module name imported by ``path`` (``from x import y`` counts as x)."""
    tree = ast.parse(path.read_text(encoding="utf-8"))
    found: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                found.add(alias.name)
        elif isinstance(node, ast.ImportFrom):
            if node.module:
                found.add(node.module)
            # ``from . import models`` also names the submodule.
            for alias in node.names:
                found.add(f"{node.module}.{alias.name}" if node.module else alias.name)
    return found


def _imports_any(path: Path, needles) -> bool:
    found = _imports(path)
    for needle in needles:
        if any(entry == needle or entry.endswith(f".{needle}") for entry in found):
            return True
    return False


def test_the_packet_model_does_not_import_a_worker():
    assert not _imports_any(
        ACQUISITION / "models.py", {"sdk_worker", "finger_worker"}
    ), "acquisition.models depends on a worker"


def test_the_finger_worker_takes_the_packet_from_models():
    """It used to import the box worker just to describe a frame."""
    source = (ACQUISITION / "finger_worker.py").read_text(encoding="utf-8")
    tree = ast.parse(source)
    from_sdk_worker: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and (node.module or "").endswith("sdk_worker"):
            from_sdk_worker.update(alias.name for alias in node.names)
    assert "SdkPacket" not in from_sdk_worker, "the finger worker still borrows the packet"
    assert "FixedCameraMode" not in from_sdk_worker
    assert "load_fixed_mode" not in from_sdk_worker
    # ...and the helpers it genuinely shares still come from there.
    assert "process_chain" in from_sdk_worker


def test_the_calibration_models_do_not_import_the_sampler_or_the_session_file():
    for path in (FORCE / "models.py",):
        assert not _imports_any(
            path, {"calibration_recorder", "calibration_session", "session"}
        ), f"{path.name} depends on a higher layer"


def test_the_session_format_does_not_import_the_sampling_state_machine():
    assert not _imports_any(
        FORCE / "calibration_session.py", {"calibration_recorder"}
    ), "the session file format still depends on the sampler"


def test_the_moved_models_are_the_same_objects_everywhere():
    from orisys_app_shared.acquisition import finger_worker, models, sdk_worker
    from orisys_app_shared.force import calibration_recorder, calibration_session, models as force_models

    assert sdk_worker.SdkPacket is models.SdkPacket is finger_worker.SdkPacket
    assert sdk_worker.FixedCameraMode is models.FixedCameraMode
    assert calibration_recorder.CalibrationPoint is force_models.CalibrationPoint
    assert calibration_recorder.AxisStats is force_models.AxisStats
    assert calibration_session.SAMPLE_FIELDS is force_models.SAMPLE_FIELDS
    # The recorder still offers the trace helpers it used to own.
    assert callable(calibration_recorder.trace_from)
    assert callable(calibration_recorder.trace_dt)


def test_the_recording_pieces_are_separate_modules():
    from orisys_app_shared.record import depth, experiment
    from orisys_app_shared.record.depth_backends import BoxReplay, FingerReplay
    from orisys_app_shared.record.video_writer import VideoSegments

    # The depth replay no longer defines its backends inline...
    assert not hasattr(depth, "BoxReplay")
    assert not hasattr(depth, "_BoxReplay")
    # ...and the writer delegates video encoding to its own component.
    assert not hasattr(experiment, "_open_segment")
    assert callable(VideoSegments.write) and callable(VideoSegments.close)
    assert BoxReplay.__name__ == "BoxReplay" and FingerReplay.__name__ == "FingerReplay"