test: update tests for Portainer client without socket fallback
- Remove TestPortainerClientDockerSocketFallback class - Update wrapper method tests to expect RuntimeError on missing endpoints - Remove all socket fallback related test cases 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.5
+24
-113
@@ -391,40 +391,12 @@ class TestPortainerClientContainers:
|
||||
assert result is True
|
||||
|
||||
|
||||
class TestPortainerClientDockerSocketFallback:
|
||||
"""Test Docker socket fallback methods."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_containers_via_socket_returns_empty_on_error(self):
|
||||
"""_list_containers_via_socket should return empty list on error."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch("httpx.AsyncHTTPTransport") as mock_transport:
|
||||
mock_transport.side_effect = Exception("Socket not available")
|
||||
|
||||
result = await client._list_containers_via_socket()
|
||||
|
||||
assert result == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inspect_container_via_socket_returns_none_on_error(self):
|
||||
"""_inspect_container_via_socket should return None on error."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch("httpx.AsyncHTTPTransport") as mock_transport:
|
||||
mock_transport.side_effect = Exception("Socket not available")
|
||||
|
||||
result = await client._inspect_container_via_socket("container_name")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestPortainerClientWrapperMethods:
|
||||
"""Test convenience wrapper methods."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_containers_uses_portainer_first(self):
|
||||
"""list_containers should try Portainer first."""
|
||||
async def test_list_containers_uses_portainer(self):
|
||||
"""list_containers should use Portainer API."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
containers = [{"Id": "abc123", "Names": ["/test"]}]
|
||||
@@ -442,57 +414,19 @@ class TestPortainerClientWrapperMethods:
|
||||
mock_containers.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_containers_falls_back_to_socket(self):
|
||||
"""list_containers should fallback to Docker socket if Portainer returns empty."""
|
||||
async def test_list_containers_raises_when_no_endpoints(self):
|
||||
"""list_containers should raise RuntimeError when no endpoints available."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
|
||||
mock_endpoints.return_value = [{"Id": 1}]
|
||||
mock_endpoints.return_value = []
|
||||
|
||||
with patch.object(client, "get_containers", new_callable=AsyncMock) as mock_containers:
|
||||
mock_containers.return_value = []
|
||||
|
||||
with patch.object(client, "_list_containers_via_socket", new_callable=AsyncMock) as mock_socket:
|
||||
mock_socket.return_value = [{"Id": "from_socket"}]
|
||||
|
||||
result = await client.list_containers()
|
||||
|
||||
assert result == [{"Id": "from_socket"}]
|
||||
mock_socket.assert_called_once()
|
||||
with pytest.raises(RuntimeError, match="No Portainer endpoints available"):
|
||||
await client.list_containers()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_containers_handles_exception_with_fallback(self):
|
||||
"""list_containers should try fallback even on exception."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
|
||||
mock_endpoints.side_effect = Exception("API error")
|
||||
|
||||
with patch.object(client, "_list_containers_via_socket", new_callable=AsyncMock) as mock_socket:
|
||||
mock_socket.return_value = [{"Id": "fallback"}]
|
||||
|
||||
result = await client.list_containers()
|
||||
|
||||
assert result == [{"Id": "fallback"}]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_containers_returns_empty_when_all_fails(self):
|
||||
"""list_containers should return empty list when everything fails."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
|
||||
mock_endpoints.side_effect = Exception("API error")
|
||||
|
||||
with patch.object(client, "_list_containers_via_socket", new_callable=AsyncMock) as mock_socket:
|
||||
mock_socket.side_effect = Exception("Socket error")
|
||||
|
||||
result = await client.list_containers()
|
||||
|
||||
assert result == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inspect_container_uses_portainer_first(self):
|
||||
"""inspect_container should try Portainer first."""
|
||||
async def test_inspect_container_uses_portainer(self):
|
||||
"""inspect_container should use Portainer API."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
container_list = [{"Id": "abc123", "Names": ["/mycontainer"]}]
|
||||
@@ -512,54 +446,31 @@ class TestPortainerClientWrapperMethods:
|
||||
assert result == container_detail
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inspect_container_falls_back_to_socket(self):
|
||||
"""inspect_container should fallback if not found in Portainer."""
|
||||
async def test_inspect_container_returns_none_when_not_found(self):
|
||||
"""inspect_container should return None if container not found."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
|
||||
mock_endpoints.return_value = [{"Id": 1}]
|
||||
|
||||
with patch.object(client, "get_containers", new_callable=AsyncMock) as mock_list:
|
||||
mock_list.return_value = [] # Container not found
|
||||
mock_list.return_value = [] # No containers
|
||||
|
||||
with patch.object(client, "_inspect_container_via_socket", new_callable=AsyncMock) as mock_socket:
|
||||
mock_socket.return_value = {"Id": "from_socket"}
|
||||
|
||||
result = await client.inspect_container("missing_container")
|
||||
|
||||
assert result == {"Id": "from_socket"}
|
||||
mock_socket.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inspect_container_handles_exception_with_fallback(self):
|
||||
"""inspect_container should try fallback even on exception."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
|
||||
mock_endpoints.side_effect = Exception("API error")
|
||||
|
||||
with patch.object(client, "_inspect_container_via_socket", new_callable=AsyncMock) as mock_socket:
|
||||
mock_socket.return_value = {"Id": "fallback"}
|
||||
|
||||
result = await client.inspect_container("container")
|
||||
|
||||
assert result == {"Id": "fallback"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inspect_container_returns_none_when_all_fails(self):
|
||||
"""inspect_container should return None when everything fails."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
|
||||
mock_endpoints.side_effect = Exception("API error")
|
||||
|
||||
with patch.object(client, "_inspect_container_via_socket", new_callable=AsyncMock) as mock_socket:
|
||||
mock_socket.side_effect = Exception("Socket error")
|
||||
|
||||
result = await client.inspect_container("container")
|
||||
result = await client.inspect_container("missing_container")
|
||||
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inspect_container_raises_when_no_endpoints(self):
|
||||
"""inspect_container should raise RuntimeError when no endpoints available."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
|
||||
mock_endpoints.return_value = []
|
||||
|
||||
with pytest.raises(RuntimeError, match="No Portainer endpoints available"):
|
||||
await client.inspect_container("container")
|
||||
|
||||
|
||||
class TestPortainerClientStackFile:
|
||||
"""Test stack file operations."""
|
||||
|
||||
Reference in New Issue
Block a user