325 lines
10 KiB
Python
325 lines
10 KiB
Python
"""Unit tests for ProfileLoader - YAML scan profile loading with env var expansion."""
|
|
|
|
import os
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.cli.profile_loader import ProfileLoader, ProfileLoaderError
|
|
from iac_reverse.models import ProviderType
|
|
|
|
|
|
@pytest.fixture
|
|
def loader():
|
|
"""Create a ProfileLoader instance."""
|
|
return ProfileLoader()
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_profile(tmp_path):
|
|
"""Helper to write a YAML profile to a temp file and return its path."""
|
|
|
|
def _write(content: str) -> str:
|
|
profile_file = tmp_path / "profile.yaml"
|
|
profile_file.write_text(textwrap.dedent(content), encoding="utf-8")
|
|
return str(profile_file)
|
|
|
|
return _write
|
|
|
|
|
|
class TestSingleProfileLoading:
|
|
"""Tests for loading a single profile from YAML."""
|
|
|
|
def test_loads_single_profile(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
provider: kubernetes
|
|
credentials:
|
|
kubeconfig_path: /home/user/.kube/config
|
|
context: pi-cluster
|
|
endpoints:
|
|
- https://k8s-api.internal.lab:6443
|
|
resource_type_filters:
|
|
- kubernetes_deployment
|
|
- kubernetes_service
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert len(profiles) == 1
|
|
profile = profiles[0]
|
|
assert profile.provider == ProviderType.KUBERNETES
|
|
assert profile.credentials == {
|
|
"kubeconfig_path": "/home/user/.kube/config",
|
|
"context": "pi-cluster",
|
|
}
|
|
assert profile.endpoints == ["https://k8s-api.internal.lab:6443"]
|
|
assert profile.resource_type_filters == [
|
|
"kubernetes_deployment",
|
|
"kubernetes_service",
|
|
]
|
|
|
|
def test_loads_profile_without_optional_fields(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
provider: docker_swarm
|
|
credentials:
|
|
host: tcp://swarm-manager:2376
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert len(profiles) == 1
|
|
profile = profiles[0]
|
|
assert profile.provider == ProviderType.DOCKER_SWARM
|
|
assert profile.endpoints is None
|
|
assert profile.resource_type_filters is None
|
|
assert profile.authentik_token is None
|
|
|
|
|
|
class TestMultiProfileLoading:
|
|
"""Tests for loading multiple profiles from a YAML list."""
|
|
|
|
def test_loads_multi_profile_yaml(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
- provider: kubernetes
|
|
credentials:
|
|
kubeconfig_path: /home/user/.kube/config
|
|
context: pi-cluster
|
|
endpoints:
|
|
- https://k8s-api.internal.lab:6443
|
|
|
|
- provider: synology
|
|
credentials:
|
|
host: nas01.internal.lab
|
|
port: "5001"
|
|
username: admin
|
|
password: secret
|
|
endpoints:
|
|
- nas01.internal.lab:5001
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert len(profiles) == 2
|
|
assert profiles[0].provider == ProviderType.KUBERNETES
|
|
assert profiles[1].provider == ProviderType.SYNOLOGY
|
|
assert profiles[1].credentials["host"] == "nas01.internal.lab"
|
|
|
|
def test_loads_three_profiles(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
- provider: kubernetes
|
|
credentials:
|
|
context: cluster-1
|
|
|
|
- provider: docker_swarm
|
|
credentials:
|
|
host: tcp://swarm:2376
|
|
|
|
- provider: windows
|
|
credentials:
|
|
host: win-server-01
|
|
username: admin
|
|
password: pass
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert len(profiles) == 3
|
|
assert profiles[0].provider == ProviderType.KUBERNETES
|
|
assert profiles[1].provider == ProviderType.DOCKER_SWARM
|
|
assert profiles[2].provider == ProviderType.WINDOWS
|
|
|
|
|
|
class TestEnvVarExpansion:
|
|
"""Tests for ${ENV_VAR} and ${ENV_VAR:-default} expansion."""
|
|
|
|
def test_expands_env_var(self, loader, monkeypatch):
|
|
monkeypatch.setenv("MY_SECRET", "super-secret-value")
|
|
|
|
result = loader.expand_env_vars("${MY_SECRET}")
|
|
|
|
assert result == "super-secret-value"
|
|
|
|
def test_expands_env_var_with_surrounding_text(self, loader, monkeypatch):
|
|
monkeypatch.setenv("HOST", "myserver.local")
|
|
|
|
result = loader.expand_env_vars("https://${HOST}:8443")
|
|
|
|
assert result == "https://myserver.local:8443"
|
|
|
|
def test_expands_multiple_env_vars(self, loader, monkeypatch):
|
|
monkeypatch.setenv("USER", "admin")
|
|
monkeypatch.setenv("PASS", "secret123")
|
|
|
|
result = loader.expand_env_vars("${USER}:${PASS}")
|
|
|
|
assert result == "admin:secret123"
|
|
|
|
def test_expands_env_var_with_default(self, loader, monkeypatch):
|
|
monkeypatch.delenv("MISSING_VAR", raising=False)
|
|
|
|
result = loader.expand_env_vars("${MISSING_VAR:-fallback_value}")
|
|
|
|
assert result == "fallback_value"
|
|
|
|
def test_env_var_set_overrides_default(self, loader, monkeypatch):
|
|
monkeypatch.setenv("MY_VAR", "actual_value")
|
|
|
|
result = loader.expand_env_vars("${MY_VAR:-default_value}")
|
|
|
|
assert result == "actual_value"
|
|
|
|
def test_empty_default_is_valid(self, loader, monkeypatch):
|
|
monkeypatch.delenv("UNSET_VAR", raising=False)
|
|
|
|
result = loader.expand_env_vars("prefix_${UNSET_VAR:-}_suffix")
|
|
|
|
assert result == "prefix__suffix"
|
|
|
|
def test_missing_env_var_without_default_raises_error(self, loader, monkeypatch):
|
|
monkeypatch.delenv("NONEXISTENT_VAR", raising=False)
|
|
|
|
with pytest.raises(ProfileLoaderError, match="NONEXISTENT_VAR"):
|
|
loader.expand_env_vars("${NONEXISTENT_VAR}")
|
|
|
|
def test_no_env_vars_returns_unchanged(self, loader):
|
|
result = loader.expand_env_vars("plain text without vars")
|
|
|
|
assert result == "plain text without vars"
|
|
|
|
|
|
class TestCredentialExpansion:
|
|
"""Tests for env var expansion applied to credential fields in profiles."""
|
|
|
|
def test_expands_credentials_in_profile(self, loader, tmp_profile, monkeypatch):
|
|
monkeypatch.setenv("SYNOLOGY_USER", "admin")
|
|
monkeypatch.setenv("SYNOLOGY_PASSWORD", "my_password")
|
|
|
|
path = tmp_profile("""
|
|
provider: synology
|
|
credentials:
|
|
host: nas01.internal.lab
|
|
username: "${SYNOLOGY_USER}"
|
|
password: "${SYNOLOGY_PASSWORD}"
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert profiles[0].credentials["username"] == "admin"
|
|
assert profiles[0].credentials["password"] == "my_password"
|
|
# Non-env-var values remain unchanged
|
|
assert profiles[0].credentials["host"] == "nas01.internal.lab"
|
|
|
|
def test_expands_nested_credential_values(self, loader, tmp_profile, monkeypatch):
|
|
monkeypatch.setenv("INNER_SECRET", "nested_value")
|
|
|
|
path = tmp_profile("""
|
|
provider: windows
|
|
credentials:
|
|
host: win-server-01
|
|
auth:
|
|
token: "${INNER_SECRET}"
|
|
type: bearer
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert profiles[0].credentials["auth"]["token"] == "nested_value"
|
|
assert profiles[0].credentials["auth"]["type"] == "bearer"
|
|
|
|
def test_expands_authentik_token(self, loader, tmp_profile, monkeypatch):
|
|
monkeypatch.setenv("AUTH_TOKEN", "my-sso-token")
|
|
|
|
path = tmp_profile("""
|
|
provider: kubernetes
|
|
credentials:
|
|
context: cluster-1
|
|
authentik_token: "${AUTH_TOKEN}"
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert profiles[0].authentik_token == "my-sso-token"
|
|
|
|
def test_credential_with_default_value(self, loader, tmp_profile, monkeypatch):
|
|
monkeypatch.delenv("OPTIONAL_PORT", raising=False)
|
|
|
|
path = tmp_profile("""
|
|
provider: synology
|
|
credentials:
|
|
host: nas01
|
|
port: "${OPTIONAL_PORT:-5001}"
|
|
""")
|
|
|
|
profiles = loader.load(path)
|
|
|
|
assert profiles[0].credentials["port"] == "5001"
|
|
|
|
|
|
class TestErrorHandling:
|
|
"""Tests for error cases in profile loading."""
|
|
|
|
def test_file_not_found_raises_error(self, loader):
|
|
with pytest.raises(ProfileLoaderError, match="Profile not found"):
|
|
loader.load("/nonexistent/path/profile.yaml")
|
|
|
|
def test_invalid_yaml_raises_error(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
provider: kubernetes
|
|
credentials: [invalid: yaml: content
|
|
""")
|
|
|
|
with pytest.raises(ProfileLoaderError, match="Invalid YAML"):
|
|
loader.load(path)
|
|
|
|
def test_empty_file_raises_error(self, loader, tmp_path):
|
|
profile_file = tmp_path / "empty.yaml"
|
|
profile_file.write_text("", encoding="utf-8")
|
|
|
|
with pytest.raises(ProfileLoaderError, match="empty"):
|
|
loader.load(str(profile_file))
|
|
|
|
def test_unknown_provider_raises_error(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
provider: unknown_provider
|
|
credentials:
|
|
key: value
|
|
""")
|
|
|
|
with pytest.raises(ProfileLoaderError, match="Unknown provider"):
|
|
loader.load(path)
|
|
|
|
def test_missing_provider_raises_error(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
credentials:
|
|
key: value
|
|
""")
|
|
|
|
with pytest.raises(ProfileLoaderError, match="Missing 'provider'"):
|
|
loader.load(path)
|
|
|
|
def test_missing_env_var_in_credentials_raises_error(
|
|
self, loader, tmp_profile, monkeypatch
|
|
):
|
|
monkeypatch.delenv("MISSING_CRED", raising=False)
|
|
|
|
path = tmp_profile("""
|
|
provider: kubernetes
|
|
credentials:
|
|
token: "${MISSING_CRED}"
|
|
""")
|
|
|
|
with pytest.raises(ProfileLoaderError, match="MISSING_CRED"):
|
|
loader.load(path)
|
|
|
|
def test_non_dict_in_multi_profile_raises_error(self, loader, tmp_profile):
|
|
path = tmp_profile("""
|
|
- provider: kubernetes
|
|
credentials:
|
|
context: cluster-1
|
|
- just a string
|
|
""")
|
|
|
|
with pytest.raises(ProfileLoaderError, match="index 1.*mapping"):
|
|
loader.load(path)
|