Public Access
emil
ef3b882e5b
Post-merge / detect-type (push) Successful in 13s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 28s
Post-merge / vikunja (push) Successful in 44s
Post-merge / sync-wiki (push) Successful in 58s
Post-merge / release (push) Successful in 1m7s
Post-merge / publish (push) Successful in 44s
Post-merge / badges (push) Successful in 1m6s
72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
"""Unit tests for devx.utils.network."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from devx.utils.network import check_http_connectivity, wait_for_ssh
|
|
|
|
_no_sleep = MagicMock()
|
|
|
|
|
|
class TestCheckHttpConnectivity:
|
|
@patch("devx.utils.network.requests.get")
|
|
def test_success(self, mock_get: MagicMock) -> None:
|
|
mock_get.return_value = MagicMock(status_code=200)
|
|
check_http_connectivity("https://example.com", max_attempts=3)
|
|
mock_get.assert_called_once()
|
|
|
|
@patch("devx.utils.network.requests.get")
|
|
def test_retries_on_connection_error(self, mock_get: MagicMock) -> None:
|
|
mock_get.side_effect = [
|
|
requests.exceptions.ConnectionError("refused"),
|
|
requests.exceptions.ConnectionError("refused"),
|
|
MagicMock(status_code=200),
|
|
]
|
|
check_http_connectivity("https://example.com", max_attempts=5, sleep=_no_sleep)
|
|
assert mock_get.call_count == 3
|
|
|
|
@patch("devx.utils.network.requests.get")
|
|
def test_raises_after_max_attempts(self, mock_get: MagicMock) -> None:
|
|
mock_get.side_effect = requests.exceptions.ConnectionError("refused")
|
|
with pytest.raises(requests.exceptions.ConnectionError):
|
|
check_http_connectivity("https://example.com", max_attempts=2, sleep=_no_sleep)
|
|
assert mock_get.call_count == 2
|
|
|
|
@patch("devx.utils.network.requests.get")
|
|
def test_verify_false(self, mock_get: MagicMock) -> None:
|
|
mock_get.return_value = MagicMock(status_code=200)
|
|
check_http_connectivity("https://example.com", verify=False)
|
|
mock_get.assert_called_once_with("https://example.com", timeout=10, verify=False)
|
|
|
|
|
|
class TestWaitForSsh:
|
|
@patch("devx.utils.network.socket.create_connection")
|
|
def test_immediate_success(self, mock_conn: MagicMock) -> None:
|
|
mock_conn.return_value.__enter__ = MagicMock()
|
|
mock_conn.return_value.__exit__ = MagicMock(return_value=False)
|
|
wait_for_ssh("10.0.0.1")
|
|
mock_conn.assert_called_once()
|
|
|
|
@patch("devx.utils.network.socket.create_connection")
|
|
def test_retries_until_success(self, mock_conn: MagicMock) -> None:
|
|
mock_conn.side_effect = [
|
|
OSError("refused"),
|
|
OSError("refused"),
|
|
MagicMock(),
|
|
]
|
|
mock_conn.return_value.__enter__ = MagicMock()
|
|
mock_conn.return_value.__exit__ = MagicMock(return_value=False)
|
|
wait_for_ssh("10.0.0.1", max_attempts=5, sleep=_no_sleep)
|
|
assert mock_conn.call_count == 3
|
|
|
|
@patch("devx.utils.network.socket.create_connection")
|
|
def test_timeout_after_max_attempts(self, mock_conn: MagicMock) -> None:
|
|
mock_conn.side_effect = OSError("refused")
|
|
with pytest.raises(RuntimeError, match="SSH not available"):
|
|
wait_for_ssh("10.0.0.1", max_attempts=3, sleep=_no_sleep)
|
|
assert mock_conn.call_count == 3
|