tests/test_utils.py

"""
Юнит-тесты для утилитарных функций colab_tunnel._utils.
"""
import sys
import subprocess

import pytest

from colab_tunnel._utils import is_ipv4, run, read_until_pattern


# is_ipv4

@pytest.mark.parametrize('addr, expected', [
    ('192.168.1.1',     True),
    ('10.0.0.1',        True),
    ('255.255.255.255',  True),
    ('0.0.0.0',         True),
    ('192.168.1.256',   False),
    ('192.168.1',       False),
    ('not-an-ip',       False),
    ('',                False),
    ('999.0.0.1',       False),
    ('1.2.3.4.5',       False),
])
def test_is_ipv4(addr: str, expected: bool) -> None:
    assert is_ipv4(addr) is expected


# run()

def test_run_captures_output() -> None:
    result = run('echo hello_world')
    assert result['status_code'] == 0
    assert 'hello_world' in result['output']


def test_run_nonzero_exit() -> None:
    result = run('sh -c "exit 42"')
    assert result['status_code'] == 42


def test_run_captures_stderr() -> None:
    result = run('sh -c "echo err_msg >&2"')
    assert 'err_msg' in result['output']


def test_run_timeout_returns_minus_one() -> None:
    result = run('sleep 60', timeout=0.3)
    assert result['status_code'] == -1
    assert 'Таймаут' in result['output']


# read_until_pattern()

@pytest.mark.skipif(sys.platform == 'win32', reason='select() с пайпами не поддерживается на Windows')
def test_read_until_pattern_found_in_stdout() -> None:
    process = subprocess.Popen(
        [sys.executable, '-c',
         'import sys, time; print("https://example.com"); sys.stdout.flush(); time.sleep(60)'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        url, output = read_until_pattern(process, r'https://\S+', timeout=5.0)
        assert url == 'https://example.com'
        assert 'https://example.com' in output
    finally:
        process.kill()
        process.wait()


@pytest.mark.skipif(sys.platform == 'win32', reason='select() с пайпами не поддерживается на Windows')
def test_read_until_pattern_found_in_stderr() -> None:
    process = subprocess.Popen(
        [sys.executable, '-c',
         'import sys, time; sys.stderr.write("https://stderr.example.com\\n"); sys.stderr.flush(); time.sleep(60)'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        url, _ = read_until_pattern(
            process, r'https://\S+', timeout=5.0, read_from_stderr=True
        )
        assert 'https://stderr.example.com' in url
    finally:
        process.kill()
        process.wait()


@pytest.mark.skipif(sys.platform == 'win32', reason='select() с пайпами не поддерживается на Windows')
def test_read_until_pattern_raises_on_timeout() -> None:
    process = subprocess.Popen(
        [sys.executable, '-c', 'import time; time.sleep(60)'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        with pytest.raises(RuntimeError, match='URL не найден'):
            read_until_pattern(process, r'https://\S+', timeout=0.5)
    finally:
        process.kill()
        process.wait()


@pytest.mark.skipif(sys.platform == 'win32', reason='select() с пайпами не поддерживается на Windows')
def test_read_until_pattern_raises_when_process_exits_without_url() -> None:
    process = subprocess.Popen(
        [sys.executable, '-c', 'print("no url here at all")'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    with pytest.raises(RuntimeError):
        read_until_pattern(process, r'https://\S+', timeout=5.0)


@pytest.mark.skipif(sys.platform == 'win32', reason='select() с пайпами не поддерживается на Windows')
def test_read_both_streams_finds_url_in_either() -> None:
    """URL в stderr должен находиться при read_both_streams=True."""
    process = subprocess.Popen(
        [sys.executable, '-c',
         'import sys, time; sys.stderr.write("https://both.example.com\\n"); sys.stderr.flush(); time.sleep(60)'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        url, _ = read_until_pattern(
            process, r'https://\S+', timeout=5.0, read_both_streams=True
        )
        assert 'https://both.example.com' in url
    finally:
        process.kill()
        process.wait()