Bruno Charest
ecefbc8611
Some checks failed
Tests / Backend Tests (Python) (3.10) (push) Has been cancelled
Tests / Backend Tests (Python) (3.11) (push) Has been cancelled
Tests / Backend Tests (Python) (3.12) (push) Has been cancelled
Tests / Frontend Tests (JS) (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / All Tests Passed (push) Has been cancelled
312 lines
11 KiB
Python
312 lines
11 KiB
Python
"""
|
|
Tests pour le service Ansible.
|
|
|
|
Couvre:
|
|
- Parsing de l'inventaire
|
|
- Listing des playbooks
|
|
- Gestion du cache
|
|
- Exécution de playbooks (mocked subprocess)
|
|
- Manipulation de l'inventaire (add/remove hosts)
|
|
"""
|
|
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
from datetime import datetime, timezone
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class TestAnsibleServiceInventory:
|
|
"""Tests pour le parsing de l'inventaire."""
|
|
|
|
def test_load_inventory_file_not_exists(self, tmp_path):
|
|
"""Retourne dict vide si fichier n'existe pas."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_path)
|
|
result = service.load_inventory()
|
|
|
|
assert result == {}
|
|
|
|
def test_load_inventory_parses_yaml(self, tmp_ansible_dir):
|
|
"""Parse correctement un fichier YAML."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
result = service.load_inventory()
|
|
|
|
assert "all" in result
|
|
assert "children" in result["all"]
|
|
|
|
def test_get_hosts_from_inventory(self, tmp_ansible_dir):
|
|
"""Extrait les hôtes de l'inventaire."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
hosts = service.get_hosts_from_inventory()
|
|
|
|
assert len(hosts) == 1
|
|
assert hosts[0].name == "test-host-1"
|
|
assert hosts[0].ansible_host == "192.168.1.10"
|
|
assert "env_test" in hosts[0].groups
|
|
|
|
def test_get_hosts_with_group_filter(self, tmp_ansible_dir):
|
|
"""Filtre les hôtes par groupe."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
# With matching filter
|
|
hosts = service.get_hosts_from_inventory(group_filter="env_test")
|
|
assert len(hosts) == 1
|
|
|
|
# With non-matching filter
|
|
hosts = service.get_hosts_from_inventory(group_filter="env_prod")
|
|
assert len(hosts) == 0
|
|
|
|
def test_get_groups(self, tmp_ansible_dir):
|
|
"""Liste tous les groupes."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
groups = service.get_groups()
|
|
|
|
assert "env_test" in groups
|
|
|
|
def test_get_env_groups(self, tmp_ansible_dir):
|
|
"""Liste les groupes env_*."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
env_groups = service.get_env_groups()
|
|
|
|
assert all(g.startswith("env_") for g in env_groups)
|
|
|
|
def test_host_exists(self, tmp_ansible_dir):
|
|
"""Vérifie l'existence d'un hôte."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
assert service.host_exists("test-host-1") is True
|
|
assert service.host_exists("nonexistent") is False
|
|
|
|
|
|
class TestAnsibleServicePlaybooks:
|
|
"""Tests pour le listing des playbooks."""
|
|
|
|
def test_get_playbooks_empty_dir(self, tmp_path):
|
|
"""Retourne liste vide si pas de playbooks."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_path)
|
|
playbooks = service.get_playbooks()
|
|
|
|
assert playbooks == []
|
|
|
|
def test_get_playbooks_parses_files(self, tmp_ansible_dir):
|
|
"""Parse les fichiers playbook."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
playbooks = service.get_playbooks()
|
|
|
|
assert len(playbooks) == 1
|
|
pb = playbooks[0]
|
|
assert pb["name"] == "health-check"
|
|
assert pb["filename"] == "health-check.yml"
|
|
assert pb["hosts"] == "all"
|
|
|
|
def test_get_playbooks_caches_result(self, tmp_ansible_dir):
|
|
"""Les playbooks sont cachés."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
# First call
|
|
playbooks1 = service.get_playbooks()
|
|
# Second call should use cache
|
|
playbooks2 = service.get_playbooks()
|
|
|
|
assert playbooks1 == playbooks2
|
|
assert service._playbooks_cache is not None
|
|
|
|
def test_invalidate_cache(self, tmp_ansible_dir):
|
|
"""Invalider le cache fonctionne."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
service.get_playbooks()
|
|
assert service._playbooks_cache is not None
|
|
|
|
service.invalidate_cache()
|
|
assert service._playbooks_cache is None
|
|
|
|
def test_is_target_compatible_with_playbook(self, tmp_ansible_dir):
|
|
"""Vérifie la compatibilité target/playbook."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
# 'all' is always compatible
|
|
assert service.is_target_compatible_with_playbook("any_target", "all") is True
|
|
assert service.is_target_compatible_with_playbook("all", "specific") is True
|
|
|
|
# Exact match
|
|
assert service.is_target_compatible_with_playbook("webservers", "webservers") is True
|
|
|
|
|
|
class TestAnsibleServiceExecution:
|
|
"""Tests pour l'exécution des playbooks."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_playbook_success(self, tmp_ansible_dir):
|
|
"""Exécution réussie d'un playbook."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
with patch("asyncio.create_subprocess_exec") as mock_exec:
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(
|
|
b"PLAY RECAP\nok=1 changed=0 failed=0",
|
|
b""
|
|
))
|
|
mock_exec.return_value = mock_process
|
|
|
|
result = await service.execute_playbook(
|
|
playbook="health-check.yml",
|
|
target="all"
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["return_code"] == 0
|
|
assert "PLAY RECAP" in result["stdout"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_playbook_failure(self, tmp_ansible_dir):
|
|
"""Exécution échouée d'un playbook."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
with patch("asyncio.create_subprocess_exec") as mock_exec:
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 2
|
|
mock_process.communicate = AsyncMock(return_value=(
|
|
b"",
|
|
b"fatal: [host]: UNREACHABLE!"
|
|
))
|
|
mock_exec.return_value = mock_process
|
|
|
|
result = await service.execute_playbook(
|
|
playbook="health-check.yml",
|
|
target="all"
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["return_code"] == 2
|
|
assert "UNREACHABLE" in result["stderr"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_playbook_not_found(self, tmp_ansible_dir):
|
|
"""Erreur si playbook non trouvé."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
await service.execute_playbook(
|
|
playbook="nonexistent.yml",
|
|
target="all"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_playbook_with_extra_vars(self, tmp_ansible_dir):
|
|
"""Exécution avec extra_vars."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
with patch("asyncio.create_subprocess_exec") as mock_exec:
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(b"ok", b""))
|
|
mock_exec.return_value = mock_process
|
|
|
|
await service.execute_playbook(
|
|
playbook="health-check.yml",
|
|
target="all",
|
|
extra_vars={"key": "value"}
|
|
)
|
|
|
|
# Verify --extra-vars was passed
|
|
call_args = mock_exec.call_args
|
|
cmd = call_args[0]
|
|
assert "--extra-vars" in cmd
|
|
|
|
|
|
class TestAnsibleServiceInventoryManipulation:
|
|
"""Tests pour la manipulation de l'inventaire."""
|
|
|
|
def test_add_host_to_inventory(self, tmp_ansible_dir):
|
|
"""Ajoute un hôte à l'inventaire."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
service.add_host_to_inventory(
|
|
hostname="new-host",
|
|
env_group="env_test",
|
|
role_groups=["role_web"],
|
|
ansible_host="10.0.0.99"
|
|
)
|
|
|
|
# Verify host was added
|
|
hosts = service.get_hosts_from_inventory()
|
|
names = [h.name for h in hosts]
|
|
assert "new-host" in names
|
|
|
|
def test_remove_host_from_inventory(self, tmp_ansible_dir):
|
|
"""Supprime un hôte de l'inventaire."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
# Verify host exists first
|
|
assert service.host_exists("test-host-1") is True
|
|
|
|
service.remove_host_from_inventory("test-host-1")
|
|
service.invalidate_cache()
|
|
|
|
assert service.host_exists("test-host-1") is False
|
|
|
|
def test_add_group(self, tmp_ansible_dir):
|
|
"""Ajoute un groupe à l'inventaire."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
service.add_group("role_new_group")
|
|
service.invalidate_cache()
|
|
|
|
assert "role_new_group" in service.get_groups()
|
|
|
|
def test_delete_group(self, tmp_ansible_dir):
|
|
"""Supprime un groupe de l'inventaire."""
|
|
from app.services.ansible_service import AnsibleService
|
|
|
|
service = AnsibleService(ansible_dir=tmp_ansible_dir)
|
|
|
|
# Add then delete
|
|
service.add_group("temp_group")
|
|
service.invalidate_cache()
|
|
assert "temp_group" in service.get_groups()
|
|
|
|
service.delete_group("temp_group")
|
|
service.invalidate_cache()
|
|
assert "temp_group" not in service.get_groups()
|