OpenShell/e2e/python/conftest.py

# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

import fcntl
import os
import time
from typing import TYPE_CHECKING

import grpc
import pytest

from openshell import Sandbox, SandboxClient, WorkspaceClient

if TYPE_CHECKING:
    from collections.abc import Callable, Iterator


def pytest_configure(config: pytest.Config) -> None:
    config.addinivalue_line(
        "markers",
        "exclusive_gateway_config: hold an exclusive lock on gateway-global "
        "config so no other xdist worker observes a transient global setting",
    )


@pytest.fixture(autouse=True)
def _gateway_config_guard(
    request: pytest.FixtureRequest,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[None]:
    """Readers-writer guard over gateway-global config mutations.

    The e2e gateway is shared across xdist workers, so a test that flips a
    gateway-global setting can leak that transient value into another worker's
    sandbox creation. Every test holds a shared lock by default; a test marked
    ``exclusive_gateway_config`` holds an exclusive lock, so while it mutates and
    restores the setting no other worker is mid-test and none can observe the
    transient value. The lock file lives in the run's shared base temp dir
    (``getbasetemp().parent``), which is common to all xdist workers.
    """
    lock_path = tmp_path_factory.getbasetemp().parent / "gateway-config.lock"
    exclusive = (
        request.node.get_closest_marker("exclusive_gateway_config") is not None
    )
    with lock_path.open("w") as lock_file:
        fcntl.flock(lock_file, fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH)
        yield


@pytest.fixture(scope="session")
def cluster_name() -> str | None:
    return os.environ.get("OPENSHELL_GATEWAY")


@pytest.fixture(scope="session")
def sandbox_client(cluster_name: str | None) -> Iterator[SandboxClient]:
    with SandboxClient.from_active_cluster(cluster=cluster_name) as client:
        yield client


@pytest.fixture(scope="session", autouse=True)
def ensure_sandbox_persistence_ready(sandbox_client: SandboxClient) -> None:
    for _ in range(60):
        try:
            sandbox_client.list_ids(workspace="default", page_size=1)
            return
        except grpc.RpcError as exc:
            details = exc.details() or ""
            if exc.code() == grpc.StatusCode.UNAVAILABLE:
                time.sleep(2)
                continue
            if (
                exc.code() == grpc.StatusCode.INTERNAL
                and "no such table: objects" in details
            ):
                time.sleep(1)
                continue
            raise

    pytest.fail(
        "openshell-server persistence is not initialized (missing sqlite objects table); "
        "redeploy the active cluster and rerun e2e sandbox tests"
    )


@pytest.fixture
def sandbox(cluster_name: str | None) -> Callable[..., Sandbox]:
    def _create(*, spec: object | None = None, delete_on_exit: bool = True) -> Sandbox:
        return Sandbox(
            workspace="default",
            cluster=cluster_name,
            spec=spec,
            delete_on_exit=delete_on_exit,
            # Allow time to pull an explicitly supplied workload fixture.
            ready_timeout_seconds=300.0,
        )

    return _create


@pytest.fixture(scope="session")
def workspace_client(sandbox_client: SandboxClient) -> WorkspaceClient:
    return WorkspaceClient.from_sandbox_client(sandbox_client)


@pytest.fixture(scope="session")
def _worker_suffix(worker_id: str) -> str:
    """Return a suffix for worker-unique resource names.

    Uses the built-in ``worker_id`` fixture from pytest-xdist which returns
    ``"gw0"``, ``"gw1"``, etc. for workers, or ``"master"`` for non-xdist runs.
    """
    if worker_id == "master":
        return ""
    return f"-{worker_id}"


@pytest.fixture
def run_python() -> Callable[[Sandbox, str], tuple[int, str, str]]:
    def _run(sandbox: Sandbox, code: str) -> tuple[int, str, str]:
        result = sandbox.exec(["python", "-c", code], timeout_seconds=20)
        return result.exit_code, result.stdout, result.stderr

    return _run