feat: add stack management and container delete endpoints
Build and Push / build (release) Successful in 29s
Build and Push / build (release) Successful in 29s
- Add stack compose YAML get/update endpoints - Add stack environment variable get/update endpoints - Add stack deploy and rebuild endpoints - Add container delete endpoint with force option - Add Portainer client methods for new operations - Add comprehensive test coverage (30 new tests) - Remove REQUESTED_SERVICES.md (now implemented) 🤖 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
@@ -561,6 +561,216 @@ class TestPortainerClientWrapperMethods:
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestPortainerClientStackFile:
|
||||
"""Test stack file operations."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_stack_file_returns_content(self):
|
||||
"""get_stack_file should return compose YAML content."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
compose_content = "version: '3'\nservices:\n web:\n image: nginx"
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"StackFileContent": compose_content}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.get.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await client.get_stack_file(1)
|
||||
|
||||
assert result == compose_content
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_stack_file_returns_empty_string_if_missing(self):
|
||||
"""get_stack_file should return empty string if content missing."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.get.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await client.get_stack_file(1)
|
||||
|
||||
assert result == ""
|
||||
|
||||
|
||||
class TestPortainerClientRedeployStack:
|
||||
"""Test stack redeploy operations."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_redeploy_stack_without_pull(self):
|
||||
"""redeploy_stack should redeploy without pulling images by default."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_stack_file", new_callable=AsyncMock) as mock_file:
|
||||
mock_file.return_value = "version: '3'"
|
||||
|
||||
with patch.object(client, "get_stack", new_callable=AsyncMock) as mock_stack:
|
||||
mock_stack.return_value = {"Id": 1, "Env": [{"name": "KEY", "value": "val"}]}
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"Id": 1}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.put.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await client.redeploy_stack(1, 1, pull_image=False)
|
||||
|
||||
call_args = mock_client.put.call_args
|
||||
assert call_args[1]["json"]["pullImage"] is False
|
||||
assert result == {"Id": 1}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_redeploy_stack_with_pull(self):
|
||||
"""redeploy_stack should pull images when requested."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch.object(client, "get_stack_file", new_callable=AsyncMock) as mock_file:
|
||||
mock_file.return_value = "version: '3'"
|
||||
|
||||
with patch.object(client, "get_stack", new_callable=AsyncMock) as mock_stack:
|
||||
mock_stack.return_value = {"Id": 1, "Env": []}
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"Id": 1}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.put.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
await client.redeploy_stack(1, 1, pull_image=True)
|
||||
|
||||
call_args = mock_client.put.call_args
|
||||
assert call_args[1]["json"]["pullImage"] is True
|
||||
|
||||
|
||||
class TestPortainerClientUpdateStackEnv:
|
||||
"""Test stack environment variable updates."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_stack_env_sends_env_vars(self):
|
||||
"""update_stack_env should update environment variables."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
env_vars = [{"name": "DB_HOST", "value": "localhost"}]
|
||||
|
||||
with patch.object(client, "get_stack_file", new_callable=AsyncMock) as mock_file:
|
||||
mock_file.return_value = "version: '3'"
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"Id": 1}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.put.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
await client.update_stack_env(1, 1, env_vars)
|
||||
|
||||
call_args = mock_client.put.call_args
|
||||
assert call_args[1]["json"]["env"] == env_vars
|
||||
|
||||
|
||||
class TestPortainerClientDeleteContainer:
|
||||
"""Test container deletion."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_container_returns_true(self):
|
||||
"""delete_container should return True on success."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 204
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.delete.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await client.delete_container(1, "abc123")
|
||||
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_container_with_force(self):
|
||||
"""delete_container should pass force parameter."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 204
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.delete.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
await client.delete_container(1, "abc123", force=True)
|
||||
|
||||
call_args = mock_client.delete.call_args
|
||||
assert call_args[1]["params"]["force"] == "true"
|
||||
|
||||
|
||||
class TestPortainerClientRestartContainer:
|
||||
"""Test container restart."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restart_container_returns_true(self):
|
||||
"""restart_container should return True on success."""
|
||||
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 204
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await client.restart_container(1, "abc123")
|
||||
|
||||
assert result is True
|
||||
mock_client.post.assert_called_once()
|
||||
|
||||
|
||||
class TestPortainerClientSingleton:
|
||||
"""Test singleton pattern."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user