"""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)