warp-tech/warpgate
1
1
Fork 0
mirror of https://github.com/warp-tech/warpgate.git synced 2024-09-20 06:46:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

timeouts

Browse source
This commit is contained in:
Eugene Pankov 2022-09-02 11:30:27 +02:00
parent f000e6fe6b
commit eaed2d2b48
No known key found for this signature in database
GPG key ID: 5896FCBBDD1CF4F4
10 changed files with 38 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/test.yml vendored
View file

@ -43,7 +43,7 @@ jobs:
- name: Run
working-directory: tests
run: |
poetry run ./run.sh
TIMEOUT=120 poetry run ./run.sh
cargo llvm-cov --no-run --hide-instantiations --lcov > coverage.lcov
- name: SonarCloud Scan

6
tests/conftest.py
View file

@ -221,6 +221,12 @@ class ProcessManager:
return p
@pytest.fixture(scope='session')
def timeout():
t = os.getenv('TIMEOUT', '10')
return int(t)
@pytest.fixture(scope='session')
def ctx():
with tempfile.TemporaryDirectory() as tmpdir:

5
tests/test_http_conntest.py
View file

@ -6,6 +6,7 @@ class Test:
self,
processes,
echo_server_port,
timeout,
):
proc, _ = processes.start_wg(
config=dedent(
@ -20,7 +21,7 @@ class Test:
),
args=['test-target', 'target'],
)
proc.wait(timeout=5)
proc.wait(timeout=timeout)
assert proc.returncode == 0
def test_fail_no_connection(self, processes):
@ -37,5 +38,5 @@ class Test:
),
args=['test-target', 'target'],
)
proc.wait(timeout=5)
proc.wait(timeout=timeout)
assert proc.returncode != 0

6
tests/test_mysql_user_auth_password.py
View file

@ -6,7 +6,7 @@ from .util import wait_port, wait_mysql_port, mysql_client_ssl_opt, mysql_client
class Test:
def test(self, processes: ProcessManager, password_123_hash):
def test(self, processes: ProcessManager, password_123_hash, timeout):
db_port = processes.start_mysql_server()
_, wg_ports = processes.start_wg(
@ -50,7 +50,7 @@ class Test:
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
)
assert b'\ndb\n' in client.communicate(b'show schemas;', timeout=30)[0]
assert b'\ndb\n' in client.communicate(b'show schemas;', timeout=timeout)[0]
assert client.returncode == 0
client = processes.start(
@ -70,5 +70,5 @@ class Test:
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
)
client.communicate(b'show schemas;', timeout=30)
client.communicate(b'show schemas;', timeout=timeout)
assert client.returncode != 0

8
tests/test_ssh_conntest.py
View file

@ -7,7 +7,7 @@ from .util import alloc_port, wait_port
class Test:
def test_success(
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path, timeout
):
ssh_port = processes.start_ssh_server(
trusted_keys=[wg_c_ed25519_pubkey.read_text()]
@ -27,10 +27,10 @@ class Test:
),
args=['test-target', 'ssh'],
)
proc.wait(timeout=5)
proc.wait(timeout=timeout)
assert proc.returncode == 0
def test_fail(self, processes: ProcessManager):
def test_fail(self, processes: ProcessManager, timeout):
ssh_port = alloc_port()
proc, _ = processes.start_wg(
config=dedent(
@ -46,5 +46,5 @@ class Test:
),
args=['test-target', 'ssh'],
)
proc.wait(timeout=5)
proc.wait(timeout=timeout)
assert proc.returncode != 0

11
tests/test_ssh_proto.py
View file

@ -57,6 +57,7 @@ class Test:
self,
processes: ProcessManager,
wg_port,
timeout,
):
ssh_client = processes.start_ssh_client(
'-p',
@ -69,7 +70,7 @@ class Test:
stderr=subprocess.PIPE,
)
stdout, stderr = ssh_client.communicate(timeout=30)
stdout, stderr = ssh_client.communicate(timeout=timeout)
assert b'stdout' == stdout
assert stderr.endswith(b'stderr')
@ -77,6 +78,7 @@ class Test:
self,
processes: ProcessManager,
wg_port,
timeout,
):
ssh_client = processes.start_ssh_client(
'-p',
@ -88,7 +90,7 @@ class Test:
password='123',
)
output = ssh_client.communicate(timeout=30)[0]
output = ssh_client.communicate(timeout=timeout)[0]
assert b'Warpgate' in output
assert b'Selected target:' in output
assert b'hello\r\n' in output
@ -134,10 +136,11 @@ class Test:
self,
processes: ProcessManager,
wg_port,
timeout,
):
script = dedent(
f'''
set timeout 10
set timeout {timeout - 5}
spawn ssh -tt user:ssh@localhost -p {wg_port} -o StrictHostKeychecking=no -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=password
@ -163,7 +166,7 @@ class Test:
['expect', '-d'], stdin=subprocess.PIPE, stdout=subprocess.PIPE
)
output = ssh_client.communicate(script.encode(), timeout=30)[0]
output = ssh_client.communicate(script.encode(), timeout=timeout)[0]
assert ssh_client.returncode == 0, output
def test_connection_error(

9
tests/test_ssh_user_auth_otp.py
View file

@ -14,6 +14,7 @@ class Test:
wg_c_ed25519_pubkey: Path,
otp_key_base32: str,
otp_key_base64: str,
timeout,
):
ssh_port = processes.start_ssh_server(
trusted_keys=[wg_c_ed25519_pubkey.read_text()]
@ -49,7 +50,7 @@ class Test:
script = dedent(
f'''
set timeout 25
set timeout {timeout - 5}
spawn ssh user:ssh@localhost -p {wg_ports['ssh']} -o StrictHostKeychecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -o IdentityFile=ssh-keys/id_ed25519 -o PreferredAuthentications=publickey,keyboard-interactive ls /bin/sh
@ -68,12 +69,12 @@ class Test:
['expect'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
output, stderr = ssh_client.communicate(script.encode(), timeout=30)
output, stderr = ssh_client.communicate(script.encode(), timeout=timeout)
assert ssh_client.returncode == 0, output + stderr
script = dedent(
f'''
set timeout 10
set timeout {timeout - 5}
spawn ssh user:ssh@localhost -p {[wg_ports['ssh']]} -o StrictHostKeychecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -o IdentityFile=ssh-keys/id_ed25519 -o PreferredAuthentications=publickey,keyboard-interactive ls /bin/sh
@ -93,5 +94,5 @@ class Test:
['expect'], stdin=subprocess.PIPE, stdout=subprocess.PIPE
)
output = ssh_client.communicate(script.encode(), timeout=30)[0]
output = ssh_client.communicate(script.encode(), timeout=timeout)[0]
assert ssh_client.returncode != 0, output

1
tests/test_ssh_user_auth_password.py
View file

@ -37,6 +37,7 @@ class Test:
ssh_client = processes.start_ssh_client(
'user:ssh@localhost',
'-v',
'-p',
str(wg_ports['ssh']),
'-i',

12
tests/test_ssh_user_auth_pubkey.py
View file

@ -7,7 +7,7 @@ from .util import wait_port
class Test:
def test_ed25519(
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path, timeout
):
ssh_port = processes.start_ssh_server(
trusted_keys=[wg_c_ed25519_pubkey.read_text()]
@ -47,7 +47,7 @@ class Test:
'ls',
'/bin/sh',
)
assert ssh_client.communicate(timeout=30)[0] == b'/bin/sh\n'
assert ssh_client.communicate(timeout=timeout)[0] == b'/bin/sh\n'
assert ssh_client.returncode == 0
ssh_client = processes.start_ssh_client(
@ -61,11 +61,11 @@ class Test:
'ls',
'/bin/sh',
)
assert ssh_client.communicate(timeout=30)[0] == b''
assert ssh_client.communicate(timeout=timeout)[0] == b''
assert ssh_client.returncode != 0
def test_rsa(
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path, timeout
):
ssh_port = processes.start_ssh_server(
trusted_keys=[wg_c_ed25519_pubkey.read_text()]
@ -106,7 +106,7 @@ class Test:
'ls',
'/bin/sh',
)
assert ssh_client.communicate(timeout=30)[0] == b'/bin/sh\n'
assert ssh_client.communicate(timeout=timeout)[0] == b'/bin/sh\n'
assert ssh_client.returncode == 0
ssh_client = processes.start_ssh_client(
@ -121,5 +121,5 @@ class Test:
'ls',
'/bin/sh',
)
assert ssh_client.communicate(timeout=30)[0] == b''
assert ssh_client.communicate(timeout=timeout)[0] == b''
assert ssh_client.returncode != 0

4
tests/test_ssh_user_auth_ticket.py
View file

@ -7,7 +7,7 @@ from .util import create_ticket, wait_port
class Test:
def test(
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path, password_123_hash
self, processes: ProcessManager, wg_c_ed25519_pubkey: Path, password_123_hash, timeout
):
ssh_port = processes.start_ssh_server(
trusted_keys=[wg_c_ed25519_pubkey.read_text()]
@ -59,5 +59,5 @@ class Test:
'/bin/sh',
password='123',
)
assert ssh_client.communicate(timeout=30)[0] == b'/bin/sh\n'
assert ssh_client.communicate(timeout=timeout)[0] == b'/bin/sh\n'
assert ssh_client.returncode == 0