logos-logoscore-py/tests/conftest.py

"""Shared pytest fixtures.

Integration tests require a real `logoscore` binary and a modules directory.
They are skipped when the required env vars are not set:

    LOGOSCORE_BIN             — absolute path to the logoscore binary
    LOGOSCORE_TEST_MODULES_DIR — directory with built test module plugins

The Nix flake's `integration` check sets both. Running `pytest tests/unit`
needs neither.
"""
from __future__ import annotations

import os
import shutil
import subprocess
from pathlib import Path

import pytest


def pytest_addoption(parser: pytest.Parser) -> None:
    parser.addoption(
        "--transport",
        action="store",
        default="local",
        help=(
            "Transport to exercise in integration tests. `local` uses "
            "QLocalSocket (default). `tcp` / `tcp_ssl` re-run the same "
            "suites against a network transport. `tcp_ssl` uses a "
            "throwaway self-signed cert generated by the "
            "`self_signed_cert` fixture (requires openssl on PATH)."
        ),
    )
    parser.addoption(
        "--docker-flavor",
        action="store",
        default="portable",
        help=(
            "Which logoscore:smoke-<flavor> docker image the docker "
            "smoke tests target: `portable` (default, self-contained "
            "cli-bundle-dir — matches how released binaries ship) or "
            "`dev` (nix-store-linked, faster to build when the nix "
            "cache is warm but requires /nix/store in the image). "
            "Use `both` to replay the matrix against each in turn."
        ),
    )


@pytest.fixture(scope="session")
def logoscore_bin() -> str:
    binary = os.environ.get("LOGOSCORE_BIN") or shutil.which("logoscore")
    if not binary:
        pytest.skip("LOGOSCORE_BIN not set and `logoscore` not on PATH")
    return binary


@pytest.fixture(scope="session")
def test_modules_dir() -> str:
    path = os.environ.get("LOGOSCORE_TEST_MODULES_DIR")
    if not path:
        pytest.skip("LOGOSCORE_TEST_MODULES_DIR not set")
    return path


@pytest.fixture(scope="session")
def transport(request: pytest.FixtureRequest) -> str:
    t = request.config.getoption("--transport")
    if t not in ("local", "tcp", "tcp_ssl"):
        # Misconfiguration of the test invocation should fail loudly,
        # not silently report a clean pass — pytest.skip would mask
        # typos in --transport on CI.
        raise pytest.UsageError(f"unsupported --transport value: {t}")
    return t


@pytest.fixture(scope="session")
def self_signed_cert(tmp_path_factory) -> tuple[Path, Path]:
    """Generate a throwaway self-signed cert+key for CN=localhost.

    Session-scoped so the same cert is reused across every test that
    needs `tcp_ssl` — generation takes a noticeable fraction of a
    second via openssl and CN=localhost is always the same, so
    there's no per-test isolation reason to regenerate.

    Skips the requesting test if `openssl` isn't on PATH (common on
    minimal CI images that don't bundle it). The Nix flake's
    integration shells include it.
    """
    if not shutil.which("openssl"):
        pytest.skip("openssl not on PATH; can't generate self-signed cert")

    d = tmp_path_factory.mktemp("tls")
    cert = d / "cert.pem"
    key = d / "key.pem"
    subprocess.run(
        [
            "openssl", "req", "-x509",
            "-newkey", "rsa:2048",
            "-keyout", str(key),
            "-out", str(cert),
            "-days", "1",
            "-nodes",
            "-subj", "/CN=localhost",
        ],
        check=True, capture_output=True,
    )
    return cert, key


def _pick_free_port() -> int:
    """Pick an ephemeral port by binding + closing. Race-prone in theory,
    fine in practice at our test concurrency levels."""
    import socket
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


@pytest.fixture
def tcp_port() -> int:
    """Ephemeral port for the daemon's TCP listener. Tests pick this
    upfront so they have the host-correct port to dial; the daemon's
    state.json carries the actually-bound port (resolved post-bind),
    but tests that round-trip a known port avoid the introspection
    step."""
    return _pick_free_port()


@pytest.fixture
def tcp_ssl_port() -> int:
    """Ephemeral port for the daemon's tcp_ssl listener. Separate
    from `tcp_port` so a test that exercises both transports can
    bind two distinct ephemeral ports without one fighting the other."""
    return _pick_free_port()