# Copyright (c) Microsoft Corporation. # Licensed under the MIT license. import pathlib import tempfile from unittest import mock import pytest from pyrit.registry import InitializerRegistry from pyrit.setup.configuration_loader import ( ConfigurationLoader, InitializerConfig, ServerConfig, initialize_from_config_async, ) from pyrit.setup.pyrit_initializer import PyRITInitializer class TestInitializerConfig: """Tests for InitializerConfig dataclass.""" def test_initializer_config_with_name_only(self): """Test creating InitializerConfig with just a name.""" config = InitializerConfig(name="simple") assert config.name == "simple" assert config.args is None def test_initializer_config_with_args(self): """Test creating InitializerConfig with name and args.""" config = InitializerConfig(name="custom", args={"param1": "value1"}) assert config.name == "custom" assert config.args == {"param1": "value1"} class TestConfigurationLoader: """Tests for ConfigurationLoader class.""" def test_default_values(self): """Test default configuration values.""" config = ConfigurationLoader() assert config.memory_db_type == "sqlite" assert config.initializers == [] assert config.initialization_scripts is None # None means "use defaults" assert config.env_files is None # None means "use defaults" assert config.env_akv_ref is None assert config.env_akv_strict is True assert config.custom_initializers_source is None assert config.silent is False def test_custom_initializers_source_loads_from_yaml(self, tmp_path: pathlib.Path) -> None: """Test loading an Azure Blob container URI for custom initializer scripts.""" config_path = tmp_path / "config.yaml" config_path.write_text( "custom_initializers_source: https://account.blob.core.windows.net/initializers\n", encoding="utf-8", ) config = ConfigurationLoader.from_yaml_file(config_path) assert config.custom_initializers_source == "https://account.blob.core.windows.net/initializers" @pytest.mark.parametrize("invalid_value", ["", " ", 42]) def test_custom_initializers_source_rejects_invalid_value(self, invalid_value: object) -> None: """Test rejecting empty or non-string custom initializer sources.""" with pytest.raises(ValueError, match="custom_initializers_source"): ConfigurationLoader(custom_initializers_source=invalid_value) # type: ignore[arg-type] @pytest.mark.parametrize("invalid_value", ["false", "true", 0, 1, None, [], {}]) def test_rejects_non_boolean_env_akv_strict(self, invalid_value): with pytest.raises(TypeError, match=r"env_akv_strict must be a bool"): ConfigurationLoader(env_akv_strict=invalid_value) # type: ignore[arg-type] @pytest.mark.parametrize("invalid_value", ["false", "true", 0, 1, None, [], {}]) def test_rejects_non_boolean_allow_custom_initializers(self, invalid_value: object) -> None: with pytest.raises(TypeError, match=r"allow_custom_initializers must be a bool"): ConfigurationLoader(allow_custom_initializers=invalid_value) # type: ignore[arg-type] def test_valid_memory_db_types_snake_case(self): """Test all valid memory database types in snake_case.""" for db_type in ["in_memory", "sqlite", "azure_sql"]: config = ConfigurationLoader(memory_db_type=db_type) assert config.memory_db_type == db_type def test_memory_db_type_normalization_from_pascal_case(self): """Test that PascalCase memory_db_type is normalized to snake_case.""" config = ConfigurationLoader(memory_db_type="InMemory") assert config.memory_db_type == "in_memory" config = ConfigurationLoader(memory_db_type="SQLite") assert config.memory_db_type == "sqlite" config = ConfigurationLoader(memory_db_type="AzureSQL") assert config.memory_db_type == "azure_sql" def test_memory_db_type_normalization_case_insensitive(self): """Test that memory_db_type normalization is case-insensitive.""" config = ConfigurationLoader(memory_db_type="SQLITE") assert config.memory_db_type == "sqlite" config = ConfigurationLoader(memory_db_type="In_Memory") assert config.memory_db_type == "in_memory" def test_invalid_memory_db_type_raises_error(self): """Test that invalid memory_db_type raises ValueError.""" with pytest.raises(ValueError, match="Invalid memory_db_type"): ConfigurationLoader(memory_db_type="InvalidType") def test_initializer_as_string(self): """Test initializers specified as simple strings.""" config = ConfigurationLoader(initializers=["simple", "airt"]) assert len(config._initializer_configs) == 2 assert config._initializer_configs[0].name == "simple" assert config._initializer_configs[0].args is None assert config._initializer_configs[1].name == "airt" def test_initializer_as_dict_with_name_only(self): """Test initializers specified as dicts with only name.""" config = ConfigurationLoader(initializers=[{"name": "simple"}]) assert len(config._initializer_configs) == 1 assert config._initializer_configs[0].name == "simple" assert config._initializer_configs[0].args is None def test_initializer_as_dict_with_args(self): """Test initializers specified as dicts with name and args.""" config = ConfigurationLoader(initializers=[{"name": "custom", "args": {"param1": "value1", "param2": 42}}]) assert len(config._initializer_configs) == 1 assert config._initializer_configs[0].name == "custom" assert config._initializer_configs[0].args == {"param1": "value1", "param2": 42} def test_mixed_initializer_formats(self): """Test initializers with mixed string and dict formats.""" config = ConfigurationLoader( initializers=[ "simple", {"name": "airt"}, {"name": "custom", "args": {"key": "value"}}, ] ) assert len(config._initializer_configs) == 3 assert config._initializer_configs[0].name == "simple" assert config._initializer_configs[1].name == "airt" assert config._initializer_configs[2].name == "custom" assert config._initializer_configs[2].args == {"key": "value"} def test_initializer_name_normalization_from_pascal_case(self): """Test that PascalCase initializer names are normalized to snake_case.""" config = ConfigurationLoader(initializers=["SimpleInitializer", "AIRTInitializer"]) assert config._initializer_configs[0].name == "simple_initializer" assert config._initializer_configs[1].name == "airt_initializer" def test_initializer_name_normalization_preserves_snake_case(self): """Test that snake_case names are preserved.""" config = ConfigurationLoader(initializers=["simple_initializer", "airt_init"]) assert config._initializer_configs[0].name == "simple_initializer" assert config._initializer_configs[1].name == "airt_init" def test_initializer_name_already_snake_case(self): """Test that snake_case names remain unchanged.""" config = ConfigurationLoader(initializers=["load_default_datasets", "scorer"]) assert config._initializer_configs[0].name == "load_default_datasets" assert config._initializer_configs[1].name == "scorer" def test_initializer_dict_without_name_raises_error(self): """Test that dict initializer without 'name' raises ValueError.""" with pytest.raises(ValueError, match="must have a 'name' field"): ConfigurationLoader(initializers=[{"args": {"key": "value"}}]) def test_initializer_invalid_type_raises_error(self): """Test that invalid initializer type raises ValueError.""" with pytest.raises(ValueError, match="must be a string or dict"): ConfigurationLoader(initializers=[123]) # type: ignore[arg-type] def test_from_dict_with_all_fields(self): """Test from_dict with all configuration fields.""" data = { "memory_db_type": "InMemory", "initializers": ["simple"], "initialization_scripts": ["/path/to/script.py"], "env_files": ["/path/to/.env"], "env_akv_ref": ["https://vault.vault.azure.net/secrets/one"], "env_akv_strict": False, "silent": True, } config = ConfigurationLoader.from_dict(data) assert config.memory_db_type == "in_memory" # Normalized to snake_case assert config.initializers == ["simple"] assert config.initialization_scripts == ["/path/to/script.py"] assert config.env_files == ["/path/to/.env"] assert config.env_akv_ref == ["https://vault.vault.azure.net/secrets/one"] assert config.env_akv_strict is False assert config.silent is True def test_from_dict_filters_none_values(self): """Test that from_dict filters out None values.""" data = { "memory_db_type": "SQLite", "initializers": None, "env_files": [], } config = ConfigurationLoader.from_dict(data) assert config.memory_db_type == "sqlite" # Normalized to snake_case assert config.initializers == [] # Uses default, not None def test_from_dict_preserves_unknown_top_level_keys_in_extensions(self): """Unknown top-level keys should be preserved for downstream tooling.""" data = { "memory_db_type": "in_memory", "targets": [{"name": "x"}], "scan_modes": {"default": {"threshold": 0.8}}, } config = ConfigurationLoader.from_dict(data) assert config.memory_db_type == "in_memory" assert config.extensions == { "targets": [{"name": "x"}], "scan_modes": {"default": {"threshold": 0.8}}, } def test_from_dict_merges_explicit_extensions_with_unknown_keys(self): data = { "memory_db_type": "sqlite", "extensions": {"team": "red"}, "targets": [{"name": "x"}], } config = ConfigurationLoader.from_dict(data) assert config.extensions == {"team": "red", "targets": [{"name": "x"}]} def test_from_dict_explicit_extensions_wins_on_collision(self): """When a key appears both as unknown top-level and inside extensions, extensions wins.""" data = { "memory_db_type": "sqlite", "extensions": {"targets": "from_extensions"}, "targets": "top_level_value", } config = ConfigurationLoader.from_dict(data) assert config.extensions["targets"] == "from_extensions" def test_from_dict_rejects_non_dict_extensions(self): with pytest.raises(ValueError, match="extensions must be a dict"): ConfigurationLoader.from_dict({"extensions": ["not", "a", "dict"]}) def test_from_yaml_file(self): """Test loading configuration from a YAML file.""" yaml_content = """ memory_db_type: in_memory initializers: - simple - name: airt args: key: value silent: true """ with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: f.write(yaml_content) yaml_path = f.name try: config = ConfigurationLoader.from_yaml_file(yaml_path) assert config.memory_db_type == "in_memory" assert len(config._initializer_configs) == 2 assert config._initializer_configs[0].name == "simple" assert config._initializer_configs[1].name == "airt" assert config._initializer_configs[1].args == {"key": "value"} assert config.silent is True finally: pathlib.Path(yaml_path).unlink() def test_example_uses_on_demand_dataset_fetching(self) -> None: example_path = pathlib.Path(__file__).parents[3] / ".pyrit_conf_example" config = ConfigurationLoader.from_yaml_file(example_path) initializer_names = [initializer.name for initializer in config._initializer_configs] example_text = example_path.read_text(encoding="utf-8") assert initializer_names == ["target", "scorer", "technique"] assert "# - name: load_default_datasets" in example_text assert "several minutes" in example_text assert "provider credentials" in example_text assert "dataset licenses" in example_text assert "server.startup_timeout" in example_text def test_from_yaml_rejects_quoted_env_akv_strict(self, tmp_path): yaml_path = tmp_path / "quoted-boolean.yaml" yaml_path.write_text('env_akv_strict: "false"\n', encoding="utf-8") with pytest.raises(TypeError, match=r"env_akv_strict must be a bool"): ConfigurationLoader.from_yaml_file(yaml_path) def test_from_yaml_accepts_native_env_akv_strict(self, tmp_path): yaml_path = tmp_path / "native-booleans.yaml" yaml_path.write_text("env_akv_strict: false\n", encoding="utf-8") config = ConfigurationLoader.from_yaml_file(yaml_path) assert config.env_akv_strict is False def test_from_empty_yaml_file_raises_value_error(self, tmp_path): """Test that an empty YAML file raises a clear ValueError.""" yaml_path = tmp_path / "empty.yaml" yaml_path.write_text("", encoding="utf-8") with pytest.raises(ValueError, match="is empty"): ConfigurationLoader.from_yaml_file(yaml_path) def test_get_default_config_path(self): """Test get_default_config_path returns expected path.""" default_path = ConfigurationLoader.get_default_config_path() assert default_path.name == ".pyrit_conf" assert ".pyrit" in str(default_path) class TestConfigurationLoaderResolvers: """Tests for ConfigurationLoader path resolution methods.""" def test_resolve_initializers_uses_singleton_registry(self) -> None: """Test resolving a custom initializer registered during backend startup.""" class CustomInitializer(PyRITInitializer): async def initialize_async(self) -> None: pass registry = InitializerRegistry(lazy_discovery=True) registry.register_class(CustomInitializer, name="custom") config = ConfigurationLoader(initializers=["custom"]) with mock.patch.object(InitializerRegistry, "get_registry_singleton", return_value=registry): resolved = config.resolve_initializers() assert len(resolved) == 1 assert isinstance(resolved[0], CustomInitializer) def testresolve_initialization_scripts_none_returns_none(self): """Test that None (default) returns None to signal 'use defaults'.""" config = ConfigurationLoader() assert config.resolve_initialization_scripts() is None def testresolve_initialization_scripts_empty_list_returns_empty_list(self): """Test that explicit empty list [] returns empty list to signal 'load nothing'.""" config = ConfigurationLoader(initialization_scripts=[]) resolved = config.resolve_initialization_scripts() assert resolved is not None assert resolved == [] def testresolve_initialization_scripts_absolute_path(self): """Test resolving absolute script paths.""" config = ConfigurationLoader(initialization_scripts=["/absolute/path/script.py"]) resolved = config.resolve_initialization_scripts() assert resolved is not None assert len(resolved) == 1 # Check path ends with expected components (Windows adds drive letter to Unix-style paths) assert resolved[0].parts[-3:] == ("absolute", "path", "script.py") def testresolve_initialization_scripts_relative_path(self): """Test resolving relative script paths (converted to absolute).""" config = ConfigurationLoader(initialization_scripts=["relative/script.py"]) resolved = config.resolve_initialization_scripts() assert resolved is not None assert len(resolved) == 1 assert resolved[0].is_absolute() # Check path ends with expected components (works on both Unix and Windows) assert resolved[0].parts[-2:] == ("relative", "script.py") def testresolve_env_files_none_returns_none(self): """Test that None (default) returns None to signal 'use defaults'.""" config = ConfigurationLoader() assert config.resolve_env_files() is None def testresolve_env_files_empty_list_returns_empty_list(self): """Test that explicit empty list [] returns empty list to signal 'load nothing'.""" config = ConfigurationLoader(env_files=[]) resolved = config.resolve_env_files() assert resolved is not None assert resolved == [] def testresolve_env_files_absolute_path(self): """Test resolving absolute env file paths.""" config = ConfigurationLoader(env_files=["/path/to/.env"]) resolved = config.resolve_env_files() assert resolved is not None assert len(resolved) == 1 # Check path ends with expected components (Windows adds drive letter to Unix-style paths) assert resolved[0].parts[-3:] == ("path", "to", ".env") def testresolve_env_akv_ref_none_returns_none(self): """Test that None is returned when env_akv_ref is not configured.""" config = ConfigurationLoader() assert config.resolve_env_akv_ref() is None def testresolve_env_akv_ref_returns_configured_values(self): """Test that the configured AKV references are returned unchanged.""" refs = ["https://vault.vault.azure.net/secrets/bootstrap"] config = ConfigurationLoader(env_akv_ref=refs) assert config.resolve_env_akv_ref() == refs def test_env_akv_ref_allows_empty_list(self): assert ConfigurationLoader(env_akv_ref=[]).env_akv_ref == [] @pytest.mark.parametrize("env_akv_ref", ["", "https://vault.vault.azure.net/secrets/one", [""], [None]]) def test_env_akv_ref_rejects_scalar_or_invalid_entries(self, env_akv_ref): with pytest.raises(ValueError, match="env_akv_ref must"): ConfigurationLoader(env_akv_ref=env_akv_ref) # type: ignore[arg-type] def test_env_akv_ref_rejects_multiple_bootstrap_urls(self): with pytest.raises(ValueError, match="at most one"): ConfigurationLoader( env_akv_ref=[ "https://vault.vault.azure.net/secrets/first", "https://vault.vault.azure.net/secrets/second", ] ) @pytest.mark.usefixtures("patch_central_database") class TestConfigurationLoaderInitialization: """Tests for ConfigurationLoader.initialize_pyrit_async method.""" @mock.patch("pyrit.setup.configuration_loader.initialize_pyrit_async") async def test_initialize_pyrit_async_basic(self, mock_init): """Test basic initialization with minimal configuration.""" config = ConfigurationLoader(memory_db_type="in_memory") await config.initialize_pyrit_async() mock_init.assert_called_once() call_kwargs = mock_init.call_args.kwargs # Should map snake_case to internal constant assert call_kwargs["memory_db_type"] == "InMemory" assert call_kwargs["initialization_scripts"] is None assert call_kwargs["initializers"] is None assert call_kwargs["env_files"] is None assert call_kwargs["env_akv_ref"] is None assert call_kwargs["env_akv_strict"] is True assert call_kwargs["silent"] is False @mock.patch("pyrit.setup.configuration_loader.initialize_pyrit_async") async def test_initialize_pyrit_async_with_env_akv_ref(self, mock_init): """Test initialization forwards env_akv_ref to initialize_pyrit_async.""" refs = ["https://vault.vault.azure.net/secrets/bootstrap"] config = ConfigurationLoader( memory_db_type="in_memory", env_akv_ref=refs, env_akv_strict=False, ) await config.initialize_pyrit_async() mock_init.assert_called_once() call_kwargs = mock_init.call_args.kwargs assert call_kwargs["env_akv_ref"] == refs assert call_kwargs["env_akv_strict"] is False @mock.patch("pyrit.setup.configuration_loader.initialize_pyrit_async") @mock.patch("pyrit.registry.InitializerRegistry") async def test_initialize_pyrit_async_with_initializers(self, mock_registry_cls, mock_init): """Test initialization with initializers resolved from registry.""" # Setup mock registry mock_registry = mock.MagicMock() mock_registry_cls.get_registry_singleton.return_value = mock_registry # Mock the configured initializer instance produced by the registry mock_initializer_instance = mock.MagicMock() mock_registry.create_and_configure.return_value = mock_initializer_instance config = ConfigurationLoader( memory_db_type="in_memory", initializers=["simple"], ) await config.initialize_pyrit_async() # Verify registry was used to resolve and configure the initializer mock_registry.create_and_configure.assert_called_once_with("simple", initializer_params=None) # Verify initialize was called with resolved initializers mock_init.assert_called_once() call_kwargs = mock_init.call_args.kwargs assert call_kwargs["initializers"] == [mock_initializer_instance] @mock.patch("pyrit.registry.InitializerRegistry") async def test_initialize_pyrit_async_unknown_initializer_raises_error(self, mock_registry_cls): """Test that unknown initializer name raises ValueError.""" mock_registry = mock.MagicMock() mock_registry_cls.get_registry_singleton.return_value = mock_registry mock_registry.create_and_configure.side_effect = KeyError("unknown_initializer") mock_registry.get_class_names.return_value = ["simple", "airt"] config = ConfigurationLoader( memory_db_type="in_memory", initializers=["unknown_initializer"], ) with pytest.raises(ValueError, match="not found in registry"): await config.initialize_pyrit_async() @mock.patch("pyrit.setup.configuration_loader.initialize_pyrit_async") @mock.patch("pyrit.registry.InitializerRegistry") async def test_initialize_pyrit_async_can_skip_unresolved_initializer(self, mock_registry_cls, mock_init): mock_registry = mock.MagicMock() mock_registry_cls.get_registry_singleton.return_value = mock_registry healthy = mock.MagicMock() mock_registry.create_and_configure.side_effect = [KeyError("bad"), healthy] mock_registry.get_class_names.return_value = ["good"] config = ConfigurationLoader(memory_db_type="in_memory", initializers=["bad", "good"]) await config.initialize_pyrit_async(raise_on_initializer_error=False) assert mock_registry.create_and_configure.call_count == 2 assert mock_init.call_args.kwargs["initializers"] == [healthy] assert mock_init.call_args.kwargs["raise_on_initializer_error"] is False @pytest.mark.usefixtures("patch_central_database") class TestInitializeFromConfigAsync: """Tests for initialize_from_config_async function.""" @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.from_yaml_file") @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.initialize_pyrit_async") async def test_initialize_from_config_with_path(self, mock_init, mock_from_yaml): """Test initialize_from_config_async with explicit path.""" mock_config = ConfigurationLoader() mock_from_yaml.return_value = mock_config result = await initialize_from_config_async("/path/to/config.yaml") mock_from_yaml.assert_called_once_with(pathlib.Path("/path/to/config.yaml")) mock_init.assert_called_once() assert result is mock_config @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.from_yaml_file") @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.initialize_pyrit_async") async def test_initialize_from_config_with_string_path(self, mock_init, mock_from_yaml): """Test initialize_from_config_async with string path.""" mock_config = ConfigurationLoader() mock_from_yaml.return_value = mock_config result = await initialize_from_config_async("/path/to/config.yaml") # Should convert string to Path call_args = mock_from_yaml.call_args[0][0] assert isinstance(call_args, pathlib.Path) @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.get_default_config_path") @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.from_yaml_file") @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.initialize_pyrit_async") async def test_initialize_from_config_default_path(self, mock_init, mock_from_yaml, mock_default_path): """Test initialize_from_config_async uses default path when none specified.""" mock_config = ConfigurationLoader() mock_from_yaml.return_value = mock_config mock_default_path.return_value = pathlib.Path("/default/path/.pyrit_conf") await initialize_from_config_async() mock_default_path.assert_called_once() mock_from_yaml.assert_called_once_with(pathlib.Path("/default/path/.pyrit_conf")) class TestLoadWithOverrides: """Tests for ConfigurationLoader.load_with_overrides static method.""" @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_defaults_when_no_config_file(self, mock_default_path): """Test default values when no config file exists.""" mock_default_path.exists.return_value = False config = ConfigurationLoader.load_with_overrides() assert config.memory_db_type == "sqlite" assert config.initializers == [] assert config.initialization_scripts is None assert config.env_files is None @mock.patch("pyrit.setup.configuration_loader.ConfigurationLoader.from_yaml_file") @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_reads_env_akv_ref_from_default_config(self, mock_default_path, mock_from_yaml): """Test env_akv_ref is loaded when default config file exists.""" mock_default_path.exists.return_value = True mock_from_yaml.return_value = ConfigurationLoader( memory_db_type="sqlite", env_akv_ref=["https://default.vault.azure.net/secrets/from-default"], ) config = ConfigurationLoader.load_with_overrides() mock_from_yaml.assert_called_once_with(mock_default_path) assert config.env_akv_ref == ["https://default.vault.azure.net/secrets/from-default"] @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_memory_db_type_override(self, mock_default_path): """Test memory_db_type override takes precedence.""" mock_default_path.exists.return_value = False config = ConfigurationLoader.load_with_overrides(memory_db_type="in_memory") assert config.memory_db_type == "in_memory" @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_initializers_override(self, mock_default_path): """Test initializers override takes precedence.""" mock_default_path.exists.return_value = False config = ConfigurationLoader.load_with_overrides(initializers=["custom_init"]) assert len(config._initializer_configs) == 1 assert config._initializer_configs[0].name == "custom_init" @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_initialization_scripts_override(self, mock_default_path): """Test initialization_scripts override takes precedence.""" mock_default_path.exists.return_value = False config = ConfigurationLoader.load_with_overrides(initialization_scripts=["/path/to/script.py"]) assert config.initialization_scripts == ["/path/to/script.py"] @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_env_files_override(self, mock_default_path): """Test env_files override takes precedence.""" mock_default_path.exists.return_value = False config = ConfigurationLoader.load_with_overrides(env_files=["/path/to/.env"]) assert config.env_files == ["/path/to/.env"] @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_env_akv_ref_override(self, mock_default_path): """Test env_akv_ref override takes precedence.""" mock_default_path.exists.return_value = False config = ConfigurationLoader.load_with_overrides( env_akv_ref=["https://vault.vault.azure.net/secrets/one"], ) assert config.env_akv_ref == ["https://vault.vault.azure.net/secrets/one"] @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_converts_sequence_to_list(self, mock_default_path): """Test that Sequence inputs are converted to list for dataclass compatibility.""" mock_default_path.exists.return_value = False # Pass tuples (which are Sequences but not Lists) config = ConfigurationLoader.load_with_overrides( initializers=("init1", "init2"), initialization_scripts=("/path/script1.py", "/path/script2.py"), env_files=("/path/.env",), env_akv_ref=("https://vault.vault.azure.net/secrets/one",), ) # Verify they are stored as lists assert isinstance(config.initializers, list) assert isinstance(config.initialization_scripts, list) assert isinstance(config.env_files, list) assert isinstance(config.env_akv_ref, list) assert config.initializers == ["init1", "init2"] assert config.initialization_scripts == ["/path/script1.py", "/path/script2.py"] assert config.env_files == ["/path/.env"] assert config.env_akv_ref == ["https://vault.vault.azure.net/secrets/one"] def test_load_with_overrides_explicit_config_file_not_found(self): """Test FileNotFoundError when explicit config file doesn't exist.""" non_existent_path = pathlib.Path("/non/existent/config.yaml") with pytest.raises(FileNotFoundError, match="Configuration file not found"): ConfigurationLoader.load_with_overrides(config_file=non_existent_path) @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_explicit_config_file_overrides_default(self, mock_default_path): """Test explicit config file values override default config file.""" mock_default_path.exists.return_value = False # Create a temp config file yaml_content = """ memory_db_type: azure_sql initializers: - explicit_init initialization_scripts: - /explicit/script.py env_files: - /explicit/.env env_akv_ref: - https://vault.vault.azure.net/secrets/explicit """ with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: f.write(yaml_content) config_path = pathlib.Path(f.name) try: config = ConfigurationLoader.load_with_overrides(config_file=config_path) assert config.memory_db_type == "azure_sql" assert config._initializer_configs[0].name == "explicit_init" assert config.initialization_scripts == ["/explicit/script.py"] assert config.env_files == ["/explicit/.env"] assert config.env_akv_ref == ["https://vault.vault.azure.net/secrets/explicit"] finally: config_path.unlink() @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_cli_overrides_config_file(self, mock_default_path): """Test CLI arguments override config file values.""" mock_default_path.exists.return_value = False # Create a temp config file with some values yaml_content = """ memory_db_type: azure_sql initializers: - file_init """ with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: f.write(yaml_content) config_path = pathlib.Path(f.name) try: # CLI overrides should take precedence config = ConfigurationLoader.load_with_overrides( config_file=config_path, memory_db_type="in_memory", initializers=["cli_init"], ) assert config.memory_db_type == "in_memory" assert config._initializer_configs[0].name == "cli_init" finally: config_path.unlink() @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_preserves_silent_from_config_file(self, mock_default_path): """Test that load_with_overrides preserves the silent flag from config files.""" mock_default_path.exists.return_value = False yaml_content = """ memory_db_type: sqlite silent: true """ with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: f.write(yaml_content) config_path = pathlib.Path(f.name) try: config = ConfigurationLoader.load_with_overrides(config_file=config_path) assert config.silent is True finally: config_path.unlink() @mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH") def test_load_with_overrides_preserves_all_explicit_config_fields(self, mock_default_path, tmp_path): mock_default_path.exists.return_value = False config_path = tmp_path / "explicit.yaml" config_path.write_text( """ max_concurrent_scenario_runs: 7 allow_custom_initializers: true server: url: http://localhost:8765/ startup_timeout: 45 extensions: feature_flag: enabled """, encoding="utf-8", ) config = ConfigurationLoader.load_with_overrides(config_file=config_path) assert config.max_concurrent_scenario_runs == 7 assert config.allow_custom_initializers is True assert config.server == {"url": "http://localhost:8765/", "startup_timeout": 45} assert config.server_config == ServerConfig(url="http://localhost:8765", startup_timeout=45.0) assert config.extensions == {"feature_flag": "enabled"} def test_load_with_overrides_uses_key_presence_when_merging_layers(self, tmp_path): default_path = tmp_path / ".pyrit_conf" default_path.write_text( """ initializers: - target initialization_scripts: - default.py silent: true operator: default-operator operation: default-operation allow_custom_initializers: true server: url: http://localhost:9000 startup_timeout: 180 extensions: default_extension: enabled """, encoding="utf-8", ) overlay_path = tmp_path / "overlay.yaml" overlay_path.write_text( """ initialization_scripts: [] silent: false operator: "" operation: null allow_custom_initializers: false extensions: {} """, encoding="utf-8", ) with mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", default_path): config = ConfigurationLoader.load_with_overrides(config_file=overlay_path) assert config.initializers == ["target"] assert config.initialization_scripts == [] assert config.silent is False assert config.operator == "" assert config.operation is None assert config.allow_custom_initializers is False assert config.server_config == ServerConfig(url="http://localhost:9000", startup_timeout=180.0) assert config.extensions == {} def test_load_with_overrides_merges_explicit_extension_keys(self, tmp_path): default_path = tmp_path / ".pyrit_conf" default_path.write_text("alpha: 1\nbeta: 2\n", encoding="utf-8") overlay_path = tmp_path / "overlay.yaml" overlay_path.write_text("alpha: 3\n", encoding="utf-8") with mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", default_path): config = ConfigurationLoader.load_with_overrides(config_file=overlay_path) assert config.extensions == {"alpha": 3, "beta": 2} def test_load_with_overrides_merges_server_fields(self, tmp_path): default_path = tmp_path / ".pyrit_conf" default_path.write_text( "server:\n url: http://localhost:9000\n startup_timeout: 180\n", encoding="utf-8", ) overlay_path = tmp_path / "overlay.yaml" overlay_path.write_text("server:\n startup_timeout: 45\n", encoding="utf-8") with mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", default_path): config = ConfigurationLoader.load_with_overrides(config_file=overlay_path) assert config.server == {"url": "http://localhost:9000", "startup_timeout": 45} assert config.server_config == ServerConfig(url="http://localhost:9000", startup_timeout=45.0) def test_load_with_overrides_null_server_resets_inherited_block(self, tmp_path): default_path = tmp_path / ".pyrit_conf" default_path.write_text( "server:\n url: http://localhost:9000\n startup_timeout: 180\n", encoding="utf-8", ) overlay_path = tmp_path / "overlay.yaml" overlay_path.write_text("server: null\n", encoding="utf-8") with mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", default_path): config = ConfigurationLoader.load_with_overrides(config_file=overlay_path) assert config.server is None assert config.server_config is None def test_load_with_overrides_removes_null_top_level_extension(self, tmp_path): default_path = tmp_path / ".pyrit_conf" default_path.write_text("alpha: 1\nbeta: 2\n", encoding="utf-8") overlay_path = tmp_path / "overlay.yaml" overlay_path.write_text("alpha: null\n", encoding="utf-8") with mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", default_path): config = ConfigurationLoader.load_with_overrides(config_file=overlay_path) assert config.extensions == {"beta": 2} def test_load_with_overrides_preserves_extension_provenance(self, tmp_path): default_path = tmp_path / ".pyrit_conf" default_path.write_text("alpha: 1\nbeta: 2\n", encoding="utf-8") overlay_path = tmp_path / "overlay.yaml" overlay_path.write_text( "alpha: top-level\nextensions:\n alpha: nested\n", encoding="utf-8", ) with mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", default_path): config = ConfigurationLoader.load_with_overrides(config_file=overlay_path) assert config.extensions == {"alpha": "nested", "beta": 2} def test_extensions_reject_non_string_keys(self): with pytest.raises(ValueError, match="extensions keys must be strings"): ConfigurationLoader.from_dict({"extensions": {1: "value"}}) @pytest.mark.parametrize( ("yaml_content", "error"), [ ("", "is empty"), ("invalid: [", "Invalid YAML"), ], ) def test_load_with_overrides_rejects_invalid_explicit_yaml(self, tmp_path, yaml_content, error): default_path = tmp_path / "missing-default.yaml" config_path = tmp_path / "explicit.yaml" config_path.write_text(yaml_content, encoding="utf-8") with ( mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", default_path), pytest.raises(ValueError, match=error), ): ConfigurationLoader.load_with_overrides(config_file=config_path) @pytest.mark.parametrize("scenario_config", [None, {"name": "scam"}]) def test_scenario_config_block_is_rejected(scenario_config): with pytest.raises(ValueError, match="'scenario' configuration block is no longer supported"): ConfigurationLoader.from_dict({"scenario": scenario_config}) def test_load_with_overrides_rejects_scenario_config_block(): yaml_content = "scenario:\n name: scam\n args:\n max_turns: 10\n" with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: f.write(yaml_content) config_path = pathlib.Path(f.name) try: with pytest.raises(ValueError, match="'scenario' configuration block is no longer supported"): ConfigurationLoader.load_with_overrides(config_file=config_path) finally: config_path.unlink() def test_load_with_overrides_rejects_scenario_block_in_default_config(tmp_path): config_path = tmp_path / ".pyrit_conf" config_path.write_text("scenario:\n name: scam\n", encoding="utf-8") with mock.patch("pyrit.setup.configuration_loader.DEFAULT_CONFIG_PATH", config_path): with pytest.raises(ValueError, match="'scenario' configuration block is no longer supported"): ConfigurationLoader.load_with_overrides() class TestNormalizeServer: """Tests for ConfigurationLoader._normalize_server.""" def test_server_none_yields_no_server_config(self): config = ConfigurationLoader(server=None) assert config.server_config is None def test_server_dict_with_url_normalizes(self): config = ConfigurationLoader(server={"url": "http://remote:9000/"}) assert config.server_config == ServerConfig(url="http://remote:9000") def test_server_dict_without_url_uses_default(self): config = ConfigurationLoader(server={}) assert config.server_config == ServerConfig(url="http://localhost:8000") def test_server_startup_timeout_normalizes(self): config = ConfigurationLoader(server={"startup_timeout": 45}) assert config.server_config == ServerConfig(url="http://localhost:8000", startup_timeout=45.0) @pytest.mark.parametrize("startup_timeout", [True, 0, -1, float("inf"), "slow"]) def test_server_invalid_startup_timeout_raises(self, startup_timeout): with pytest.raises(ValueError, match="'startup_timeout' must be a finite number greater than 0"): ConfigurationLoader(server={"startup_timeout": startup_timeout}) def test_server_url_non_string_raises(self): with pytest.raises(ValueError, match="Server 'url' must be a string"): ConfigurationLoader(server={"url": 12345}) def test_server_non_dict_raises(self): with pytest.raises(ValueError, match="Server entry must be a dict"): ConfigurationLoader(server="http://oops:8000") # type: ignore[arg-type]