"""
Comprehensive error path testing for unimus_client.py.
Tests network errors, HTTP status codes, JSON parsing failures, and
complex error scenarios following Gemini's Phase 2 guidance.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
import json
import re
import requests
import base64
from unimus_client import (
UnimusRestClient,
UnimusError,
UnimusNotFoundError,
UnimusValidationError,
UnimusAuthenticationError
)
from tests.fixtures.mock_data import MockDataFactory
class MockResponse:
"""Mock response class for detailed HTTP testing."""
def __init__(self, json_data=None, status_code=200, raise_for_status_exc=None, text=None):
self._json_data = json_data
self.status_code = status_code
self._raise_for_status_exc = raise_for_status_exc
self.text = text or json.dumps(json_data) if json_data else ""
self.content = self.text.encode('utf-8')
def json(self):
if self._json_data is None:
raise json.JSONDecodeError("No JSON object could be decoded", "", 0)
return self._json_data
def raise_for_status(self):
if self._raise_for_status_exc:
raise self._raise_for_status_exc
if self.status_code >= 400:
raise requests.exceptions.HTTPError(response=self)
class MockNonJsonResponse:
"""Mock response that cannot be parsed as JSON."""
def __init__(self, status_code=200, content="Not JSON"):
self.status_code = status_code
self.content = content.encode('utf-8')
self.text = content
def json(self):
raise json.JSONDecodeError("Expecting value", self.text, 0)
def raise_for_status(self):
if self.status_code >= 400:
raise requests.exceptions.HTTPError(response=self)
class TestHTTPStatusCodeErrors:
"""Test comprehensive HTTP status code error handling."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
@pytest.mark.parametrize("status_code,error_message,expected_exception", [
(400, "Bad Request", UnimusValidationError),
(401, "Unauthorized", UnimusAuthenticationError),
(403, "Forbidden", UnimusError),
(404, "Not Found", UnimusNotFoundError),
(405, "Method Not Allowed", UnimusError),
(408, "Request Timeout", UnimusError),
(409, "Conflict", UnimusError),
(422, "Unprocessable Entity", UnimusValidationError),
(429, "Too Many Requests", UnimusError),
(500, "Internal Server Error", UnimusError),
(502, "Bad Gateway", UnimusError),
(503, "Service Unavailable", UnimusError),
(504, "Gateway Timeout", UnimusError),
(507, "Insufficient Storage", UnimusError),
])
def test_http_status_code_errors(self, client, status_code, error_message, expected_exception):
"""Test handling of various HTTP status codes."""
with patch.object(client.session, 'get') as mock_get:
mock_get.return_value = MockResponse(
json_data={"message": error_message},
status_code=status_code
)
with pytest.raises(expected_exception):
client.get_health()
def test_http_error_with_detailed_message(self, client):
"""Test HTTP error with detailed error message."""
with patch.object(client.session, 'get') as mock_get:
mock_get.return_value = MockResponse(
json_data={
"message": "Device ID must be a positive integer"
},
status_code=400
)
with pytest.raises(UnimusValidationError) as exc_info:
client.get_device_by_id(1)
assert "Validation error" in str(exc_info.value)
def test_http_error_without_json_body(self, client):
"""Test HTTP error without JSON error body."""
with patch.object(client.session, 'get') as mock_get:
mock_get.return_value = MockNonJsonResponse(
status_code=500,
content="Internal Server Error"
)
# Non-JSON responses with error status codes raise HTTPError via raise_for_status()
with pytest.raises(requests.exceptions.HTTPError):
client.get_health()
class TestNetworkAndConnectionErrors:
"""Test network and connection error scenarios."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
def test_connection_error(self, client):
"""Test connection error handling."""
with patch.object(client.session, 'get') as mock_get:
mock_get.side_effect = requests.exceptions.ConnectionError("Connection refused")
# The actual client will let the ConnectionError through - it's not wrapped
with pytest.raises(requests.exceptions.ConnectionError):
client.get_health()
def test_timeout_error(self, client):
"""Test timeout error handling."""
with patch.object(client.session, 'get') as mock_get:
mock_get.side_effect = requests.exceptions.Timeout("Request timed out")
with pytest.raises(requests.exceptions.Timeout):
client.get_health()
def test_dns_resolution_error(self, client):
"""Test DNS resolution error handling."""
with patch.object(client.session, 'get') as mock_get:
mock_get.side_effect = requests.exceptions.ConnectionError("Name or service not known")
with pytest.raises(requests.exceptions.ConnectionError):
client.get_health()
def test_ssl_certificate_error(self, client):
"""Test SSL certificate error handling."""
with patch.object(client.session, 'get') as mock_get:
mock_get.side_effect = requests.exceptions.SSLError("SSL certificate verify failed")
with pytest.raises(requests.exceptions.SSLError):
client.get_health()
def test_read_timeout_error(self, client):
"""Test read timeout error handling."""
with patch.object(client.session, 'get') as mock_get:
mock_get.side_effect = requests.exceptions.ReadTimeout("Read timed out")
with pytest.raises(requests.exceptions.ReadTimeout):
client.get_health()
def test_connection_pool_error(self, client):
"""Test connection pool error handling."""
with patch.object(client.session, 'get') as mock_get:
mock_get.side_effect = requests.exceptions.ConnectionError("Connection pool is full")
with pytest.raises(requests.exceptions.ConnectionError):
client.get_health()
class TestJSONParsingErrors:
"""Test JSON parsing and response format errors."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
def test_invalid_json_response(self, client):
"""Test handling of invalid JSON response."""
with patch.object(client.session, 'get') as mock_get:
mock_response = MockNonJsonResponse(
status_code=200,
content="This is not valid JSON: {invalid"
)
mock_get.return_value = mock_response
# According to _handle_response, non-JSON responses call raise_for_status()
# For 200 status, it should return empty dict
result = client.get_health()
assert result == {}
def test_empty_response_body(self, client):
"""Test handling of empty response body."""
with patch.object(client.session, 'get') as mock_get:
mock_response = MockNonJsonResponse(
status_code=200,
content=""
)
mock_get.return_value = mock_response
# Empty response with 200 status should return empty dict
result = client.get_health()
assert result == {}
def test_malformed_json_response(self, client):
"""Test handling of malformed JSON response."""
with patch.object(client.session, 'get') as mock_get:
mock_response = MockNonJsonResponse(
status_code=200,
content='{"incomplete": "json"'
)
mock_get.return_value = mock_response
# Malformed JSON with 200 status should return empty dict
result = client.get_health()
assert result == {}
def test_json_with_unexpected_structure(self, client):
"""Test handling of JSON with unexpected structure."""
with patch.object(client.session, 'get') as mock_get:
mock_get.return_value = MockResponse(
json_data="unexpected string instead of object",
status_code=200
)
# Should not raise exception, but return the unexpected data
result = client.get_health() # Use actual method instead of non-existent _make_request
assert result == "unexpected string instead of object"
class TestBackupContentSearchErrors:
"""Test backup content search error paths and edge cases."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
def test_invalid_regex_pattern(self, client):
"""Test search with invalid regex pattern."""
# Use an actually invalid regex pattern
with pytest.raises(UnimusValidationError, match="Invalid regex pattern"):
client.search_backup_content(device_filters={"name": "test_device"}, pattern="[invalid_regex")
def test_negative_context_lines(self, client):
"""Test search with negative context lines."""
with pytest.raises(UnimusValidationError, match="context_lines must be non-negative"):
client.search_backup_content(device_filters={"name": "test_device"}, pattern="pattern", context_lines=-1)
def test_invalid_limit_parameter(self, client):
"""Test search with invalid limit parameter."""
with pytest.raises(UnimusValidationError, match="limit must be a positive integer"):
client.search_backup_content(device_filters={"name": "test_device"}, pattern="pattern", limit=0)
with pytest.raises(UnimusValidationError, match="limit must be a positive integer"):
client.search_backup_content(device_filters={"name": "test_device"}, pattern="pattern", limit=-5)
def test_device_filters_no_devices_found(self, client):
"""Test search when device filters result in no devices."""
with patch.object(client, 'get_devices') as mock_get_devices:
mock_get_devices.return_value = []
result = client.search_backup_content(device_filters={"name": "nonexistent"}, pattern="pattern")
assert result == []
def test_device_with_no_backups(self, client):
"""Test search on device with no backups."""
device = MockDataFactory.create_device(device_id=123)
with patch.object(client, 'get_devices') as mock_get_devices, \
patch.object(client, 'get_device_latest_backup') as mock_get_backup:
mock_get_devices.return_value = [device]
mock_get_backup.return_value = None
result = client.search_backup_content(device_filters={"id": 123}, pattern="pattern")
assert result == []
def test_binary_backup_content_skipped(self, client):
"""Test that binary backup content is skipped."""
device = MockDataFactory.create_device(device_id=123)
# Create a BINARY type backup - these should be skipped in search
backup = {
"id": 1,
"type": "BINARY",
"bytes": "YmluYXJ5X2NvbnRlbnQ=", # base64 encoded binary content
"timestamp": "2023-01-01T00:00:00Z"
}
with patch.object(client, 'get_devices') as mock_get_devices, \
patch.object(client, 'get_device_latest_backup') as mock_get_backup:
mock_get_devices.return_value = [device]
mock_get_backup.return_value = backup
result = client.search_backup_content(device_filters={"id": 123}, pattern="pattern")
assert result == []
def test_backup_content_decode_error(self, client):
"""Test backup content that cannot be decoded."""
device = MockDataFactory.create_device(device_id=123)
# Create a TEXT backup with invalid base64
backup = {
"id": 1,
"type": "TEXT",
"bytes": "invalid_base64_content!!!", # Invalid base64
"timestamp": "2023-01-01T00:00:00Z"
}
with patch.object(client, 'get_devices') as mock_get_devices, \
patch.object(client, 'get_device_latest_backup') as mock_get_backup:
mock_get_devices.return_value = [device]
mock_get_backup.return_value = backup
# Should gracefully handle decode errors and continue without results
result = client.search_backup_content(device_filters={"id": 123}, pattern="pattern")
assert result == []
def test_backup_content_utf8_decode_error(self, client):
"""Test backup content that cannot be decoded as UTF-8."""
device = MockDataFactory.create_device(device_id=123)
# Create a TEXT backup with valid base64 but invalid UTF-8
backup = {
"id": 1,
"type": "TEXT",
"bytes": base64.b64encode(b'\xff\xfe\xfd').decode('ascii'), # Invalid UTF-8 bytes
"timestamp": "2023-01-01T00:00:00Z"
}
with patch.object(client, 'get_devices') as mock_get_devices, \
patch.object(client, 'get_device_latest_backup') as mock_get_backup:
mock_get_devices.return_value = [device]
mock_get_backup.return_value = backup
# Should gracefully handle UTF-8 decode errors and continue without results
result = client.search_backup_content(device_filters={"id": 123}, pattern="pattern")
assert result == []
class TestMetadataEnrichmentErrors:
"""Test metadata enrichment error paths."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
def test_enrich_metadata_missing_device_id(self, client):
"""Test metadata enrichment with missing device ID."""
device_without_id = {"address": "192.168.1.1", "description": "Test Device"}
with patch.object(client, 'get_devices') as mock_get_devices:
mock_get_devices.return_value = [device_without_id]
result = client.get_devices(enrich_metadata=True)
# Should handle gracefully and return device without enrichment
assert len(result) == 1
assert 'metadata' not in result[0]
def test_calculate_backup_metadata_error(self, client):
"""Test error in backup metadata calculation."""
device = MockDataFactory.create_device(device_id=123)
with patch.object(client, 'get_devices') as mock_get_devices, \
patch.object(client, '_calculate_backup_metadata') as mock_calc_backup:
mock_get_devices.return_value = [device]
mock_calc_backup.side_effect = Exception("Backup calculation failed")
result = client.get_devices(enrich_metadata=True)
# Should handle error gracefully
assert len(result) == 1
# Metadata should be empty or partial
def test_calculate_connectivity_metadata_error(self, client):
"""Test error in connectivity metadata calculation."""
device = MockDataFactory.create_device(device_id=123)
with patch.object(client, 'get_devices') as mock_get_devices, \
patch.object(client, '_calculate_connectivity_metadata') as mock_calc_conn:
mock_get_devices.return_value = [device]
mock_calc_conn.side_effect = Exception("Connectivity calculation failed")
result = client.get_devices(enrich_metadata=True)
# Should handle error gracefully
assert len(result) == 1
def test_calculate_configuration_metadata_error(self, client):
"""Test error in configuration metadata calculation."""
device = MockDataFactory.create_device(device_id=123)
with patch.object(client, 'get_devices') as mock_get_devices, \
patch.object(client, '_calculate_configuration_metadata') as mock_calc_config:
mock_get_devices.return_value = [device]
mock_calc_config.side_effect = Exception("Configuration calculation failed")
result = client.get_devices(enrich_metadata=True)
# Should handle error gracefully
assert len(result) == 1
class TestNetworkRelationshipErrors:
"""Test network relationship discovery error paths."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
def test_discover_network_neighbors_missing_address(self, client):
"""Test network neighbor discovery with missing IP address."""
device_without_address = MockDataFactory.create_device(device_id=123)
device_without_address.pop('address', None)
with patch.object(client, 'get_device_by_id') as mock_get_device, \
patch.object(client, 'get_devices') as mock_get_devices:
mock_get_device.return_value = device_without_address
mock_get_devices.return_value = [device_without_address]
result = client.get_device_relationships(123)
# Should handle gracefully
assert 'networkNeighbors' in result
assert len(result['networkNeighbors']) == 0
def test_discover_network_neighbors_invalid_ip(self, client):
"""Test network neighbor discovery with invalid IP address."""
device_with_invalid_ip = MockDataFactory.create_device(device_id=123)
device_with_invalid_ip['address'] = "invalid.ip.address"
with patch.object(client, 'get_device_by_id') as mock_get_device, \
patch.object(client, 'get_devices') as mock_get_devices:
mock_get_device.return_value = device_with_invalid_ip
mock_get_devices.return_value = [device_with_invalid_ip]
result = client.get_device_relationships(123)
# Should handle gracefully
assert 'networkNeighbors' in result
def test_analyze_network_relationship_unrelated_ips(self, client):
"""Test network relationship analysis with completely unrelated IPs."""
device1 = MockDataFactory.create_device(device_id=1, address="192.168.1.1")
device2 = MockDataFactory.create_device(device_id=2, address="10.0.0.1")
with patch.object(client, 'get_device_by_id') as mock_get_device, \
patch.object(client, 'get_devices') as mock_get_devices:
mock_get_device.return_value = device1
mock_get_devices.return_value = [device1, device2]
result = client.get_device_relationships(1)
# Should find no close relationships
network_neighbors = result.get('networkNeighbors', [])
close_neighbors = [n for n in network_neighbors if n.get('distance', 999) < 100]
assert len(close_neighbors) == 0
def test_discover_zone_peers_missing_zone_id(self, client):
"""Test zone peer discovery with missing zone ID."""
device_without_zone = MockDataFactory.create_device(device_id=123)
device_without_zone.pop('zoneId', None)
with patch.object(client, 'get_device_by_id') as mock_get_device, \
patch.object(client, 'get_devices') as mock_get_devices:
mock_get_device.return_value = device_without_zone
mock_get_devices.return_value = [device_without_zone]
result = client.get_device_relationships(123)
# Should handle gracefully
assert 'zonePeers' in result
assert len(result['zonePeers']) == 0
class TestTopologyAnalysisErrors:
"""Test topology analysis boundary conditions and errors."""
@pytest.fixture
def client(self):
return UnimusRestClient(
url="https://test.unimus.com",
token="test-token"
)
def test_identify_device_clusters_no_clusters(self, client):
"""Test device clustering when no clusters can be formed."""
# Devices in completely different subnets
devices = [
MockDataFactory.create_device(device_id=1, address="192.168.1.1"),
MockDataFactory.create_device(device_id=2, address="10.0.0.1"),
MockDataFactory.create_device(device_id=3, address="172.16.0.1"),
]
with patch.object(client, 'get_devices') as mock_get_devices:
mock_get_devices.return_value = devices
result = client.get_network_topology_analysis()
# Should handle gracefully with minimal clustering
assert 'deviceClusters' in result
def test_analyze_network_segments_insufficient_devices(self, client):
"""Test network segment analysis with insufficient devices."""
single_device = [MockDataFactory.create_device(device_id=1, address="192.168.1.1")]
with patch.object(client, 'get_devices') as mock_get_devices:
mock_get_devices.return_value = single_device
result = client.get_network_topology_analysis()
# Should handle single device scenario
assert 'networkSegments' in result
def test_analyze_zone_topology_empty_devices(self, client):
"""Test zone topology analysis with empty device list."""
with patch.object(client, 'get_devices') as mock_get_devices:
mock_get_devices.return_value = []
result = client.get_network_topology_analysis()
# Should handle empty list gracefully - zoneTopology is a dict with metadata
assert 'zoneTopology' in result
zone_topology = result['zoneTopology']
# The zoneTopology should be a dict with zones, total_zones, total_devices keys
assert isinstance(zone_topology, dict)
assert zone_topology.get('total_devices', 0) == 0
def test_analyze_zone_topology_devices_without_zone_id(self, client):
"""Test zone topology analysis with devices missing zone IDs."""
devices_without_zones = [
MockDataFactory.create_device(device_id=1),
MockDataFactory.create_device(device_id=2),
]
# Remove zone IDs
for device in devices_without_zones:
device.pop('zoneId', None)
with patch.object(client, 'get_devices') as mock_get_devices:
mock_get_devices.return_value = devices_without_zones
result = client.get_network_topology_analysis()
# Should handle devices without zones
assert 'zoneTopology' in result
def test_device_clustering_boundary_conditions(self, client):
"""Test device clustering with boundary conditions."""
# Create devices at the edge of clustering thresholds
devices = [
MockDataFactory.create_device(device_id=1, address="192.168.1.1", vendor="Cisco"),
MockDataFactory.create_device(device_id=2, address="192.168.1.2", vendor="Cisco"),
# Just below vendor clustering threshold
]
with patch.object(client, 'get_devices') as mock_get_devices:
mock_get_devices.return_value = devices
result = client.get_network_topology_analysis()
# Should handle threshold boundaries correctly
assert 'deviceClusters' in result
# Verify the structure - deviceClusters is actually a list of clusters
device_clusters = result['deviceClusters']
assert isinstance(device_clusters, list)
# Each cluster should be a valid cluster object
for cluster in device_clusters:
assert isinstance(cluster, dict)
assert 'cluster_type' in cluster
assert 'devices' in cluster
# Verify minimum cluster size for meaningful clusters
if 'device_count' in cluster:
assert cluster['device_count'] >= 1
