File size: 4,552 Bytes
c0e2219 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """
Юнит-тесты для утилитарных функций 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()
|