Skip to content

Pytest plugin

pytest_plugin

Pytest plugin that hands your tests a fake VLT on demand.

Registered through the pytest11 entry point, so installing the package is enough. No conftest gymnastics, no imports, it just shows up. Like a cold.

SimHandle dataclass 

SimHandle(
    transport: str, endpoint: str, stop: Callable[[], None]
)

What a test gets back: the transport, where to connect, and how to shut it up early.

endpoint is "host:port" for TCP or a filesystem path for a PTY. Either way it's ready to use the moment you get the handle.

morpheus_vlt_server 

morpheus_vlt_server(
    tmp_path_factory: TempPathFactory,
) -> Callable[..., SimHandle]

Factory fixture: call it to spin up a simulator in a background thread.

Call it as many times as you like. It returns once the server is actually listening, so connect right away. Every server gets stopped and joined at teardown, so you don't have to remember. Port 0 picks a free port, and a PTY with no path lands in a pytest temp dir named ttyMORPHEUS.

The request handling is the CLI's: Unidesa frames get cut out of the byte stream, and a protocol error gets logged instead of killing the server. DEX always gets the built-in EVA-DTS file here, there's no knob for it.

Source code in src/morpheus_vlt/pytest_plugin.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@pytest.fixture
def morpheus_vlt_server(tmp_path_factory: pytest.TempPathFactory) -> Callable[..., SimHandle]:
    """Factory fixture: call it to spin up a simulator in a background thread.

    Call it as many times as you like. It returns once the server is actually listening, so
    connect right away. Every server gets stopped and joined at teardown, so you don't have to
    remember. Port 0 picks a free port, and a PTY with no path lands in a pytest temp dir named
    ttyMORPHEUS.

    The request handling is the CLI's: Unidesa frames get cut out of the byte stream, and a
    protocol error gets logged instead of killing the server. DEX always gets the built-in
    EVA-DTS file here, there's no knob for it.
    """
    threads: list[threading.Thread] = []
    servers: list[PtyByteServer | TCPByteServer] = []

    def _find_free_port() -> int:
        # Ask the kernel for a free port, then hang up. Somebody could grab it in between.
        # Hasn't happened yet. Famous last words.
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
        sock.close()
        return port

    def _launch(srv: PtyByteServer | TCPByteServer, transport: str) -> None:
        thread = threading.Thread(target=srv.start, daemon=True)
        thread.start()
        threads.append(thread)
        servers.append(srv)
        if not srv.ready.wait(timeout=READY_TIMEOUT_S):
            srv.stop()
            raise RuntimeError(
                f"{transport} server didn't come up within {READY_TIMEOUT_S:.0f} s. "
                "The thread's traceback usually says why."
            )

    def _start(
        *,
        protocol: str = "unidesa",
        transport: str = "tcp",
        host: str = "127.0.0.1",
        port: int = 0,
        tty_name: str | None = None,
        tty_path: str | None = None,
        vdai_type: str | None = None,
        sas_addr: int = 0x01,
        chaos: ChaosProfile | None = None,
        state: MachineState | None = None,
    ) -> SimHandle:
        if state is None:
            state = MachineState()
        proto = build_protocol(protocol, state, vdai_type, sas_addr)
        handler = build_handler(Simulator(protocol=proto, chaos=chaos), protocol)

        if transport == "pty":
            if not tty_path:
                if not tty_name:
                    tty_name = "ttyMORPHEUS"
                tty_path = str(tmp_path_factory.mktemp("pty") / tty_name)
            pty_srv = PtyByteServer(PtyConfig(link_path=tty_path), handler)
            _launch(pty_srv, "pty")
            return SimHandle(transport="pty", endpoint=tty_path, stop=pty_srv.stop)

        if port == 0:
            port = _find_free_port()
        tcp_srv = TCPByteServer(ServerConfig(host=host, port=port), handler)
        _launch(tcp_srv, "tcp")
        return SimHandle(transport="tcp", endpoint=f"{host}:{tcp_srv.cfg.port}", stop=tcp_srv.stop)

    yield _start

    # Teardown. Stop everything we started, give the threads a couple of seconds to notice.
    for srv in servers:
        srv.stop()
    for thread in threads:
        thread.join(timeout=2.0)