MCP_CyberArk/tests/conftest.py

"""
Shared pytest fixtures for the MCP Privileged Access test suite.

──────────────────────────────────────────────────────────────────────────────
HOW MCP TOOLS WORK  (read this to understand what the tests are testing)
──────────────────────────────────────────────────────────────────────────────

An MCP tool is just an async Python function decorated with @mcp.tool().
The decorator registers the function in FastMCP's tool registry — it does NOT
change how the function itself is called.  This means tests can call tool
functions directly as plain async functions:

    result = await ssh_execute(host="...", command="...", ...)

The MCP framework wraps tool calls in a JSON-RPC envelope when running for
real, but for unit tests we skip the envelope entirely.

FastMCP injects a Context object as the `ctx` parameter.  The Context carries:
  • ctx.info(msg)  — progress notification sent back to the caller
  • ctx.error(msg) — error notification
  • ctx.request_context.request — the raw HTTP request (for IP extraction etc.)

In tests we pass a MagicMock for Context so we can assert what was logged
without making any real network calls.

SECRET HANDLE LIFECYCLE
  1. CyberArk MCP calls secret_store.store(username, password) → "secret://abc…"
  2. Handle is returned to Claude (only the handle token, never the password).
  3. SSH / PowerShell / DB tool calls secret_store.resolve(handle) → (user, pass).
  4. If handle_single_use=True (default), the handle is deleted after step 3.
  5. The password is used for the connection and then deleted from local scope.

This means:
  • Each test that needs to resolve a credential must create its OWN fresh handle.
  • Attempting to resolve the same handle twice raises KeyError.
"""

from __future__ import annotations

import os

# Must be set before any mcp_privileged import triggers Settings() at module level.
os.environ.setdefault("MCP_API_KEYS", "test-key-for-pytest")

from unittest.mock import AsyncMock, MagicMock

import pytest

from mcp_privileged.secret_store import secret_store


# ── Context mock ──────────────────────────────────────────────────────────────

@pytest.fixture
def mock_ctx() -> MagicMock:
    """
    Minimal mock of the FastMCP Context object.

    ctx.info() and ctx.error() are AsyncMocks so tests can await them and
    also assert what messages were emitted:

        ctx.error.assert_awaited_once()
        assert "expired" in str(ctx.error.call_args)
    """
    ctx = MagicMock()
    ctx.info = AsyncMock()
    ctx.error = AsyncMock()
    # _extract_client_ip reads these — plain dict works fine
    ctx.request_context.request.headers = {}
    ctx.request_context.request.client = None
    return ctx


# ── Credential handle factory ─────────────────────────────────────────────────

@pytest.fixture
async def credential_handle() -> str:
    """
    Store a test credential and return a fresh secret handle.

    Because handle_single_use=True (default), each test fixture invocation
    creates a NEW handle so tests don't step on each other.

    Usage:
        async def test_something(credential_handle, mock_ctx):
            result = await ssh_execute(..., secret_handle=credential_handle, ctx=mock_ctx)
    """
    return await secret_store.store("svc_user", "P@ssw0rd!")


@pytest.fixture
async def credential_handle_with_details() -> tuple[str, str, str]:
    """
    Return (handle, username, password) so tests can assert on the values.
    The password is exposed here ONLY for test assertions — never in prod code.
    """
    username = "admin_user"
    password = "S3cr3tP@ss123"
    handle = await secret_store.store(username, password)
    return handle, username, password


# ── asyncssh mock helpers ─────────────────────────────────────────────────────

def make_ssh_cm(
    stdout: str = "",
    stderr: str = "",
    exit_status: int = 0,
) -> tuple[AsyncMock, AsyncMock]:
    """
    Build a mock for asyncssh.connect used as an async context manager.

    asyncssh.connect() is called as:
        async with asyncssh.connect(host, port=..., ...) as conn:
            result = await conn.run(command, timeout=...)

    The mock chain:
        asyncssh.connect(...)         → returns mock_cm
        async with mock_cm as conn:   → calls mock_cm.__aenter__() → mock_conn
        await conn.run(...)           → returns MagicMock(stdout, stderr, exit_status)

    Returns (mock_cm, mock_conn) so tests can inspect call_args on mock_conn.run.
    """
    mock_conn = AsyncMock()
    mock_conn.run = AsyncMock(
        return_value=MagicMock(stdout=stdout, stderr=stderr, exit_status=exit_status)
    )
    mock_cm = AsyncMock()
    mock_cm.__aenter__ = AsyncMock(return_value=mock_conn)
    mock_cm.__aexit__ = AsyncMock(return_value=False)
    return mock_cm, mock_conn


# ── pypsrp mock helpers ────────────────────────────────────────────────────────

def make_ps_result(
    output: list[str] | None = None,
    had_errors: bool = False,
    errors: list[str] | None = None,
) -> tuple[list[str], bool, list[str]]:
    """
    Build the tuple returned by _run_ps_sync so tests can patch it directly.

    Usage:
        with patch(
            "mcp_privileged.powershell.server._run_ps_sync",
            return_value=make_ps_result(output=["Hello"]),
        ):
            ...
    """
    return (output or [], had_errors, errors or [])


# ── asyncpg / aiomysql mock helpers ───────────────────────────────────────────

def make_db_result(
    columns: list[str],
    rows: list[list],
) -> tuple[list[str], list[list]]:
    """
    Build the (columns, rows) tuple returned by _dispatch_query.

    Usage:
        with patch(
            "mcp_privileged.database.server._dispatch_query",
            new=AsyncMock(return_value=make_db_result(["id", "name"], [[1, "Alice"]])),
        ):
            ...
    """
    return columns, rows