responder/tests/test_cli.py

"""
Test module for Responder CLI functionality.

This module tests the following CLI commands:
- responder --version: Version display
- responder build: Build command execution
- responder run: Server execution

Requirements:
- The `docopt-ng` package must be installed
- Example application must be present at `examples/helloworld.py`
- This file should implement a basic HTTP server with a "/hello" endpoint
  that returns "hello, world!" as response
"""

import json
import os
import subprocess
import time
import typing as t
from pathlib import Path
from urllib.request import urlopen

import pytest
from _pytest.capture import CaptureFixture

from responder.__version__ import __version__
from responder.util.cmd import ResponderProgram, ResponderServer
from tests.util import random_port, wait_server_tcp

# Skip test if optional CLI dependency is not installed.
pytest.importorskip("docopt", reason="docopt-ng package not installed")


# Pseudo-wait for server idleness
SERVER_IDLE_WAIT = float(os.getenv("RESPONDER_SERVER_IDLE_WAIT", "0.25"))

# Maximum time to wait for server startup or teardown (adjust for slower systems)
SERVER_TIMEOUT = float(os.getenv("RESPONDER_SERVER_TIMEOUT", "5"))

# Maximum time to wait for HTTP requests (adjust for slower networks)
REQUEST_TIMEOUT = float(os.getenv("RESPONDER_REQUEST_TIMEOUT", "5"))

# Endpoint to use for `responder run`.
HELLO_ENDPOINT = "/hello"


def test_cli_version(capfd):
    """
    Verify that `responder --version` works as expected.
    """
    try:
        # Suppress security checks for subprocess calls in tests.
        # S603: subprocess call - safe as we use fixed command
        # S607: start process with partial path - safe as we use installed package
        subprocess.check_call(["responder", "--version"])  # noqa: S603, S607
    except subprocess.CalledProcessError as ex:
        pytest.fail(
            f"responder --version failed with exit code {ex.returncode}. Error: {ex}"
        )

    stdout = capfd.readouterr().out.strip()

    # TODO: Accommodate PyPy as installed by `uv`, it emits spurious output on stdout before the version number.
    # AssertionError: assert '(_common_types_metatype, 9088, 128, 128)\n(cython_function_or_method, 157568, 128, 128)\n3.6.0' == '3.6.0'
    lines = [line.strip() for line in stdout.splitlines() if line.strip()]
    assert lines, "Expected version output on stdout"
    assert lines[-1] == __version__


def responder_build(path: Path, capfd: CaptureFixture) -> t.Tuple[str, str]:
    """
    Execute responder build command and capture its output.

    Args:
        path: Directory containing package.json
        capfd: Pytest fixture for capturing output

    Returns:
        tuple: (stdout, stderr) containing the captured output
    """

    ResponderProgram.build(path=path)
    output = capfd.readouterr()

    stdout = output.out.strip()
    stderr = output.err.strip()

    return stdout, stderr


def test_cli_build_success(capfd, tmp_path):
    """
    Verify that `responder build` works as expected.
    """

    # Temporary surrogate `package.json` file.
    package_json = {"scripts": {"build": "echo Hotzenplotz"}}
    package_json_file = tmp_path / "package.json"
    package_json_file.write_text(json.dumps(package_json))

    # Invoke `responder build`.
    stdout, stderr = responder_build(tmp_path, capfd)
    assert "Hotzenplotz" in stdout


def test_cli_build_missing_package_json(capfd, tmp_path):
    """
    Verify `responder build`, while `package.json` file is missing.
    """

    # Invoke `responder build`.
    stdout, stderr = responder_build(tmp_path, capfd)
    assert "Invalid target directory or missing package.json" in stderr


@pytest.mark.parametrize(
    "invalid_content,npm_error,expected_error",
    [
        (
            "foobar",
            "code EJSONPARSE",
            ["is not valid JSON", "Failed to parse JSON data", "EJSONPARSE"],
        ),
        ("{", "code EJSONPARSE", ["Unexpected end of JSON", "EJSONPARSE"]),
        ('{"scripts": }', "code EJSONPARSE", ["Unexpected token", "EJSONPARSE"]),
        (
            '{"scripts": null}',
            "error",
            [
                "Cannot convert undefined or null",
                "scripts.build",
                "Missing script",
                "null",
            ],
        ),
        (
            '{"scripts": {"build": null}}',
            "Missing script",
            ['"build"', "missing script", "build"],
        ),
        (
            '{"scripts": {"build": 123}}',
            "Missing script",
            ['"build"', "missing script", "build"],
        ),
    ],
    ids=[
        "invalid_json_content",
        "incomplete_json",
        "syntax_error",
        "null_scripts",
        "missing_script_null",
        "missing_script_number",
    ],
)
def test_cli_build_invalid_package_json(
    capfd, tmp_path, invalid_content, npm_error, expected_error
):
    """
    Verify `responder build` using an invalid `package.json` file.
    """

    # Temporary surrogate `package.json` file.
    package_json_file = tmp_path / "package.json"
    package_json_file.write_text(invalid_content)

    # Invoke `responder build`.
    stdout, stderr = responder_build(tmp_path, capfd)
    assert npm_error.lower() in stderr.lower()
    if isinstance(expected_error, str):
        expected_error = [expected_error]
    assert any(item.lower() in stderr.lower() for item in expected_error)


sfa_services_valid = [
    str(Path("examples") / "helloworld.py"),
    "https://github.com/kennethreitz/responder/raw/refs/heads/main/examples/helloworld.py",
]


# The test is marked as flaky due to potential race conditions in server startup
# and port availability. Known error codes by platform:
# - macOS:   [Errno 61] Connection refused (Failed to establish a new connection)
# - Linux:   [Errno 111] Connection refused (Failed to establish a new connection)
# - Windows: [WinError 10061] No connection could be made because target machine
#            actively refused it
@pytest.mark.flaky(reruns=3, reruns_delay=2, only_rerun=["TimeoutError"])
@pytest.mark.parametrize("target", sfa_services_valid, ids=sfa_services_valid)
def test_cli_run(capfd, target):
    """
    Verify that `responder run` works as expected.
    """

    # Start a Responder service instance in the background, using its CLI.
    # Make it terminate itself after serving one HTTP request.
    server = ResponderServer(target=str(target), port=random_port(), limit_max_requests=1)
    try:
        # Start server and wait until it responds on TCP.
        server.start()
        wait_server_tcp(server.port)

        # Submit a single probing HTTP request that also will terminate the server.
        with urlopen(  # noqa: S310
            f"http://127.0.0.1:{server.port}{HELLO_ENDPOINT}",
            timeout=REQUEST_TIMEOUT,
        ) as response:
            assert "hello, world!" == response.read().decode()
    finally:
        # stop() restores the process-wide SIGTERM/SIGINT handlers start()
        # installed; without it they leak into every later test.
        server.stop()
        server.join(timeout=SERVER_TIMEOUT)

    # Capture process output.
    time.sleep(SERVER_IDLE_WAIT)
    output = capfd.readouterr()

    stdout = output.out.strip()
    assert f'"GET {HELLO_ENDPOINT} HTTP/1.1" 200 OK' in stdout

    stderr = output.err.strip()

    # Define expected lifecycle messages in order.
    lifecycle_messages = [
        # Startup phase
        "Started server process",
        "Waiting for application startup",
        "Application startup complete",
        "Uvicorn running",
        # Shutdown phase
        "Shutting down",
        "Waiting for application shutdown",
        "Application shutdown complete",
        "Finished server process",
    ]

    # Verify messages appear in expected order.
    last_pos = -1
    for msg in lifecycle_messages:
        pos = stderr.find(msg)
        assert pos > last_pos, f"Expected '{msg}' to appear after previous message"
        last_pos = pos


def _client_app(tmp_path):
    app = tmp_path / "clientapp.py"
    app.write_text(
        'import responder\n'
        'api = responder.API(title="T", version="1", openapi="3.1.0",\n'
        '                    secret_key="k" * 16, allowed_hosts=[";"],\n'
        '                    session_https_only=False)\n'
        '@api.get("/ping")\n'
        'def ping(req, resp):\n'
        '    resp.text = "pong"\n'
    )
    return app


def test_cli_client_to_stdout(monkeypatch, capsys, tmp_path):
    """`responder client <target>` prints a Python client."""
    app = _client_app(tmp_path)
    monkeypatch.setattr("sys.argv", ["responder", "client", f"{app}:api"])
    from responder.ext.cli import cli

    cli()
    out = capsys.readouterr().out
    assert "class APIClient" in out
    assert "def get_ping" in out


def test_cli_client_can_target_atelier_example(monkeypatch, capsys):
    """The public golden example can be used as a CLI client target."""
    app = Path(__file__).resolve().parents[1] / "examples" / "atelier.py"
    monkeypatch.setattr(
        "sys.argv",
        [
            "responder", "client",
            "--class-name", "AtelierClient",
            f"{app}:api",
        ],
    )
    from responder.ext.cli import cli

    cli()
    out = capsys.readouterr().out
    assert "class AtelierClient" in out
    assert "class ProjectIn(TypedDict" in out
    assert "def create_project" in out
    assert "def publish_project" in out


def test_cli_client_output_file_and_lang(monkeypatch, tmp_path):
    """`--output` writes the file; `--lang` and `--class-name` are honored."""
    app = _client_app(tmp_path)
    out_file = tmp_path / "client.ts"
    monkeypatch.setattr(
        "sys.argv",
        [
            "responder", "client",
            "--lang", "typescript",
            "--class-name", "PingClient",
            "-o", str(out_file),
            f"{app}:api",
        ],
    )
    from responder.ext.cli import cli

    cli()
    text = out_file.read_text()
    assert "class PingClient" in text


def test_cli_client_invalid_language_exits(monkeypatch, tmp_path):
    app = _client_app(tmp_path)
    monkeypatch.setattr(
        "sys.argv", ["responder", "client", "--lang", "cobol", f"{app}:api"]
    )
    from responder.ext.cli import cli

    with pytest.raises(SystemExit) as exc:
        cli()
    assert exc.value.code == 1