""" Tests for the Portainer backup executor. The point of this executor is producing an archive that will still open on the day it is needed, so most of these cover the failure paths: a truncated body behind a 200, a partial file left on disk, and retention deleting the wrong thing. """ import gzip import io import tarfile from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest from src.config import Settings from src.executors import portainer_backup_executor as pbe def _tar_gz_bytes(names=("compose/1/docker-compose.yml", "certs/cert.pem")) -> bytes: buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: for n in names: data = b"x" info = tarfile.TarInfo(name=n) info.size = len(data) tar.addfile(info, io.BytesIO(data)) return buf.getvalue() def _mock_post(status=200, content=None): response = MagicMock() response.status_code = status response.content = content if content is not None else _tar_gz_bytes() response.text = "error body" client = MagicMock() client.post = AsyncMock(return_value=response) ctx = MagicMock() ctx.__aenter__ = AsyncMock(return_value=client) ctx.__aexit__ = AsyncMock(return_value=False) return ctx, client @pytest.mark.executor @pytest.mark.unit class TestConfigValidation: @pytest.mark.asyncio async def test_missing_url_rejected(self, test_settings: Settings, tmp_path): with pytest.raises(ValueError, match="url"): await pbe.execute({"api_key": "k", "output_dir": str(tmp_path)}, test_settings) @pytest.mark.asyncio async def test_missing_api_key_rejected(self, test_settings: Settings, tmp_path): with pytest.raises(ValueError, match="api_key"): await pbe.execute({"url": "http://x", "output_dir": str(tmp_path)}, test_settings) @pytest.mark.asyncio async def test_unresolved_env_var_rejected(self, test_settings: Settings, tmp_path, monkeypatch): """${VAR} that expands to nothing must fail, not send an empty key.""" monkeypatch.delenv("NOPE_MISSING", raising=False) with pytest.raises(ValueError, match="api_key"): await pbe.execute( {"url": "http://x", "api_key": "${NOPE_MISSING}", "output_dir": str(tmp_path)}, test_settings, ) @pytest.mark.asyncio @pytest.mark.parametrize("bad", [0, -1, "30", None, True]) async def test_bad_retention_rejected(self, bad, test_settings: Settings, tmp_path): with pytest.raises(ValueError, match="retention_days"): await pbe.execute( {"url": "http://x", "api_key": "k", "output_dir": str(tmp_path), "retention_days": bad}, test_settings, ) @pytest.mark.executor @pytest.mark.unit class TestBackupBehaviour: def _config(self, tmp_path, **over): cfg = {"url": "http://portainer:9000", "api_key": "k", "output_dir": str(tmp_path), "retention_days": 30} cfg.update(over) return cfg @pytest.mark.asyncio async def test_writes_verified_archive(self, test_settings: Settings, tmp_path): ctx, _ = _mock_post() with patch("httpx.AsyncClient", return_value=ctx): result = await pbe.execute(self._config(tmp_path), test_settings) files = list(tmp_path.glob("portainer-*.tar.gz")) assert len(files) == 1 assert "2 entries" in result with tarfile.open(files[0], "r:gz") as tar: # opens = usable backup assert "certs/cert.pem" in tar.getnames() @pytest.mark.asyncio async def test_archive_is_not_world_readable(self, test_settings: Settings, tmp_path): """It contains TLS private keys.""" ctx, _ = _mock_post() with patch("httpx.AsyncClient", return_value=ctx): await pbe.execute(self._config(tmp_path), test_settings) f = next(tmp_path.glob("portainer-*.tar.gz")) assert oct(f.stat().st_mode)[-3:] == "600" @pytest.mark.asyncio async def test_http_error_raises_and_leaves_nothing(self, test_settings: Settings, tmp_path): ctx, _ = _mock_post(status=401) with patch("httpx.AsyncClient", return_value=ctx): with pytest.raises(Exception, match="HTTP 401"): await pbe.execute(self._config(tmp_path), test_settings) assert list(tmp_path.iterdir()) == [] @pytest.mark.asyncio async def test_truncated_body_behind_200_is_rejected(self, test_settings: Settings, tmp_path): """The dangerous case: a 200 whose body is not a usable archive.""" broken = _tar_gz_bytes()[:40] ctx, _ = _mock_post(content=broken) with patch("httpx.AsyncClient", return_value=ctx): with pytest.raises(Exception, match="not a readable archive"): await pbe.execute(self._config(tmp_path), test_settings) # no .partial and no final file left behind assert list(tmp_path.iterdir()) == [] @pytest.mark.asyncio async def test_gzip_that_is_not_a_tar_is_rejected(self, test_settings: Settings, tmp_path): ctx, _ = _mock_post(content=gzip.compress(b"not a tar")) with patch("httpx.AsyncClient", return_value=ctx): with pytest.raises(Exception, match="not a readable archive"): await pbe.execute(self._config(tmp_path), test_settings) assert list(tmp_path.iterdir()) == [] @pytest.mark.asyncio async def test_api_key_resolved_from_env(self, test_settings: Settings, tmp_path, monkeypatch): monkeypatch.setenv("PT_KEY", "secret-value") ctx, client = _mock_post() with patch("httpx.AsyncClient", return_value=ctx): await pbe.execute(self._config(tmp_path, api_key="${PT_KEY}"), test_settings) assert client.post.call_args.kwargs["headers"]["X-API-Key"] == "secret-value" @pytest.mark.executor @pytest.mark.unit class TestRetention: def _age(self, path, days): import os old = (datetime.now(timezone.utc) - timedelta(days=days)).timestamp() os.utime(path, (old, old)) def test_prunes_only_past_the_window(self, tmp_path): fresh = tmp_path / "portainer-20260808T120000Z.tar.gz" stale = tmp_path / "portainer-20260101T120000Z.tar.gz" for f in (fresh, stale): f.write_bytes(b"x") self._age(stale, 45) assert pbe._prune(tmp_path, 30) == 1 assert fresh.exists() and not stale.exists() def test_leaves_unrelated_files_alone(self, tmp_path): """Retention must not touch anything it did not write.""" other = tmp_path / "important-database-dump.tar.gz" named_alike = tmp_path / "portainer-backup-manual.tar.gz" for f in (other, named_alike): f.write_bytes(b"x") self._age(f, 400) assert pbe._prune(tmp_path, 30) == 0 assert other.exists() and named_alike.exists()