patx/skirr

"""Exercise the CLI with the working directory absent from Python's import path."""
import http.client
import os
from pathlib import Path
import subprocess
import sys
import time

import pytest
import skirr


def cli_env():
    # Console scripts start with their bin directory on sys.path. -P gives the
    # subprocess the same constraint without relying on an installed launcher.
    return dict(os.environ, PYTHONPATH=str(Path(skirr.__file__).resolve().parent.parent))


@pytest.mark.parametrize('directory', ['cwd', 'explicit'])
def test_cli_serves_application_from_selected_directory(tmp_path, unused_tcp_port, directory):
    project = tmp_path / 'project with spaces'
    project.mkdir()
    (project / 'app.py').write_text(
        "async def app(scope, receive, send):\n"
        "    await send({'type': 'http.response.start', 'status': 200})\n"
        "    await send({'type': 'http.response.body', 'body': b'local-app'})\n"
    )
    cwd = project if directory == 'cwd' else tmp_path
    arguments = ['app:app', '--port', str(unused_tcp_port), '--lifespan', 'off']
    if directory == 'explicit':
        (cwd / 'app.py').write_text("raise RuntimeError('Imported the wrong app directory')\n")
        arguments += ['--app-dir', str(project.relative_to(cwd))]
    process = subprocess.Popen(
        [sys.executable, '-P', '-c', 'from skirr import main; main()', *arguments],
        cwd=cwd, env=cli_env(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
        text=True,
    )
    try:
        deadline = time.monotonic() + 5
        while True:
            assert process.poll() is None, process.communicate()[0]
            client = http.client.HTTPConnection('127.0.0.1', unused_tcp_port, timeout=.5)
            try:
                client.request('GET', '/')
                response = client.getresponse()
                assert response.status == 200
                assert response.read() == b'local-app'
                break
            except (OSError, http.client.HTTPException):
                if time.monotonic() >= deadline:
                    raise
                time.sleep(.02)
            finally:
                client.close()
    finally:
        if process.poll() is None:
            process.terminate()
        try:
            output, _ = process.communicate(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
            process.communicate()
            raise
    assert process.returncode == 0, output


@pytest.mark.parametrize('reference', ['app', 'app:', ':app'])
def test_cli_reports_malformed_application_reference(tmp_path, reference):
    result = subprocess.run(
        [sys.executable, '-P', '-c', 'from skirr import main; main()', reference],
        cwd=tmp_path, env=cli_env(), capture_output=True, text=True, timeout=5,
    )
    assert result.returncode == 2
    assert 'module:attribute' in result.stderr
    assert 'Traceback' not in result.stderr