from __future__ import annotations

import os
import subprocess
import sys
import timeit

import pytest

import procrunner


def test_simple_command_invocation():
    if os.name == "nt":
        command = ["cmd.exe", "/c", "echo", "hello"]
    else:
        command = ["echo", "hello"]

    result = procrunner.run(command)

    assert result.returncode == 0
    assert result.stdout == b"hello" + os.linesep.encode("utf-8")
    assert result.stderr == b""


def test_simple_command_invocation_with_closed_stdin():
    if os.name == "nt":
        command = ["cmd.exe", "/c", "echo", "hello"]
    else:
        command = ["echo", "hello"]

    result = procrunner.run(command, stdin=subprocess.DEVNULL)

    assert result.returncode == 0
    assert result.stdout == b"hello" + os.linesep.encode("utf-8")
    assert result.stderr == b""


def test_decode_invalid_utf8_input(capsys):
    test_string = b"test\xa0string\n"
    if os.name == "nt":
        pytest.xfail("Test requires stdin feature which does not work on Windows")
        command = ["cmd.exe", "/c", "type", "CON"]
    else:
        command = ["cat"]
    result = procrunner.run(command, stdin=test_string)
    assert result.returncode == 0
    assert not result.stderr
    if os.name == "nt":
        # Windows modifies line endings
        assert result.stdout == test_string[:-1] + b"\r\n"
    else:
        assert result.stdout == test_string
    out, err = capsys.readouterr()
    assert out == "test\ufffdstring\n"
    assert err == ""


def test_running_wget(tmp_path):
    command = ["wget", "https://www.google.com", "-O", "-"]
    try:
        result = procrunner.run(command, working_directory=tmp_path)
    except OSError as e:
        if e.errno == 2:
            pytest.skip("wget not available")
        raise
    assert result.returncode == 0
    assert b"http" in result.stderr
    assert b"google" in result.stdout


def test_path_object_resolution(tmp_path):
    sentinel_value = b"sentinel"
    tmp_path.joinpath("tempfile").write_bytes(sentinel_value)
    tmp_path.joinpath("reader.py").write_text(
        "with open('tempfile') as fh:\n    print(fh.read())"
    )
    assert "LEAK_DETECTOR" not in os.environ
    result = procrunner.run(
        [sys.executable, tmp_path / "reader.py"],
        environment_override={"PYTHONHASHSEED": "random", "LEAK_DETECTOR": "1"},
        working_directory=tmp_path,
    )
    assert result.returncode == 0
    assert not result.stderr
    assert sentinel_value == result.stdout.strip()
    assert (
        "LEAK_DETECTOR" not in os.environ
    ), "overridden environment variable leaked into parent process"


def test_timeout_behaviour_old_legacy(tmp_path):
    command = (sys.executable, "-c", "import time; time.sleep(5)")
    start = timeit.default_timer()
    try:
        with pytest.raises(subprocess.TimeoutExpired) as te:
            with pytest.warns(UserWarning, match="timeout"):
                procrunner.run(
                    command,
                    timeout=0.1,
                    working_directory=tmp_path,
                    raise_timeout_exception=False,
                )
    except RuntimeError:
        # This test sometimes fails with a RuntimeError.
        runtime = timeit.default_timer() - start
        assert runtime < 3
        return
    runtime = timeit.default_timer() - start
    assert runtime < 3
    assert te.value.stdout == b""
    assert te.value.stderr == b""
    assert te.value.timeout == 0.1
    assert te.value.cmd == command


def test_timeout_behaviour_legacy(tmp_path):
    command = (sys.executable, "-c", "import time; time.sleep(5)")
    start = timeit.default_timer()
    try:
        with pytest.raises(subprocess.TimeoutExpired) as te:
            with pytest.warns(DeprecationWarning, match="timeout"):
                procrunner.run(
                    command,
                    timeout=0.1,
                    working_directory=tmp_path,
                    raise_timeout_exception=True,
                )
    except RuntimeError:
        # This test sometimes fails with a RuntimeError.
        runtime = timeit.default_timer() - start
        assert runtime < 3
        return
    runtime = timeit.default_timer() - start
    assert runtime < 3
    assert te.value.stdout == b""
    assert te.value.stderr == b""
    assert te.value.timeout == 0.1
    assert te.value.cmd == command


def test_timeout_behaviour(tmp_path):
    command = (sys.executable, "-c", "import time; time.sleep(5)")
    start = timeit.default_timer()
    try:
        with pytest.raises(subprocess.TimeoutExpired) as te:
            procrunner.run(
                command,
                timeout=0.1,
                working_directory=tmp_path,
            )
    except RuntimeError:
        # This test sometimes fails with a RuntimeError.
        runtime = timeit.default_timer() - start
        assert runtime < 3
        return
    runtime = timeit.default_timer() - start
    assert runtime < 3
    assert te.value.stdout == b""
    assert te.value.stderr == b""
    assert te.value.timeout == 0.1
    assert te.value.cmd == command