diff --git a/doc/code/framework.md b/doc/code/framework.md
index 2080ccf670..dd5e31f923 100644
--- a/doc/code/framework.md
+++ b/doc/code/framework.md
@@ -396,6 +396,7 @@ See [message normalizers](./targets/11_message_normalizer) for capability behavi
- If you are creating a component with user input (e.g. via config, REST, or automatically) it should always use the registry
- If you are storing an instance of a component, it should always use the registry
+- The registry accepts only explicitly supported external inputs, permits opaque Python objects only for in-process callers, and leaves component validation to constructors.
## [Setup](./setup/0_setup)
diff --git a/doc/gui/0_gui.md b/doc/gui/0_gui.md
index 4afd1e8d35..d77bd99e16 100644
--- a/doc/gui/0_gui.md
+++ b/doc/gui/0_gui.md
@@ -588,9 +588,11 @@ at `GET /api/runtime`.
## Registry API Migration Notes
Use `/api/converters/types` and `/api/targets/types` for registry build metadata.
-These endpoints return all constructor parameters from the registry, including
-lists, unions, and component references. The temporary `/catalog` routes retain
-their scalar-only filtering for the current UI.
+These endpoints return the constructor parameters external callers can set, each
+described in the form callers send it: a flat collection as a list, a union as its
+first alternative callers can send (`font_size: int | tuple[int, int]` as `int`),
+and a component reference as a name. They leave out types external callers can't
+create. Registry metadata keeps every parameter with its full annotation.
Create requests should supply an explicit registry `name`. Converter creation
returns the complete `ConverterInstance`; read its type from
`identifier.class_name`, not the old top-level `converter_type` field. Treat
@@ -607,10 +609,8 @@ allowlisted image, audio, and video extensions inline. Other files, including PD
SVG, HTML, text, and executables, download as `application/octet-stream` attachments.
**Temporary compatibility, scheduled for removal with the chat migration:**
-the `/api/converters/catalog` and `/api/targets/catalog` routes project the same
-registry metadata for the current UI. Create requests without a name receive a
-generated `compat_...` name. New clients should not depend on these routes or
-unnamed creation.
+target create requests without a name receive a generated `compat_...` name.
+New clients should supply an explicit name.
## Connection Health
diff --git a/frontend/src/components/Config/CreateTargetDialog.test.tsx b/frontend/src/components/Config/CreateTargetDialog.test.tsx
index dbc6ecb256..818c72e746 100644
--- a/frontend/src/components/Config/CreateTargetDialog.test.tsx
+++ b/frontend/src/components/Config/CreateTargetDialog.test.tsx
@@ -1140,6 +1140,66 @@ describe("CreateTargetDialog", () => {
});
});
+ it("should submit the catalog video duration as a number", async () => {
+ const user = userEvent.setup();
+ const videoParameters: Parameter[] = [
+ {
+ name: "resolution_dimensions",
+ type_name: "str",
+ is_list: false,
+ choices: ["720x1280", "1280x720", "1024x1792", "1792x1024"],
+ default: "1280x720",
+ required: false,
+ },
+ { name: "n_seconds", type_name: "int", is_list: false, choices: null, default: "4", required: false },
+ { name: "model_name", type_name: "str", is_list: false, choices: null, default: null, required: true },
+ { name: "endpoint", type_name: "str", is_list: false, choices: null, default: null, required: true },
+ { name: "api_key", type_name: "str", is_list: false, choices: null, default: null, required: false },
+ { name: "underlying_model", type_name: "str", is_list: false, choices: null, default: null, required: false },
+ ];
+ mockedTargetsApi.listTargetTypes.mockResolvedValue({
+ items: TARGET_TYPES.items.map((item) => (
+ item.target_type === "OpenAIVideoTarget" ? { ...item, parameters: videoParameters } : item
+ )),
+ });
+ mockedTargetsApi.createTarget.mockResolvedValue(makeTarget({
+ target_registry_name: "openai_video_custom",
+ target_type: "OpenAIVideoTarget",
+ }));
+
+ render(
+
+
+
+ );
+
+ await selectTargetType("OpenAIVideoTarget");
+ await user.click(screen.getByText("Advanced settings"));
+
+ fireEvent.change(screen.getByPlaceholderText("https://your-resource.openai.azure.com/"), {
+ target: { value: "https://api.openai.com" },
+ });
+ fireEvent.change(screen.getByPlaceholderText("e.g. gpt-4o, my-deployment"), {
+ target: { value: "sora-2" },
+ });
+ fireEvent.change(screen.getByLabelText("N Seconds"), {
+ target: { value: "8" },
+ });
+
+ await user.click(screen.getByText("Create Target"));
+
+ await waitFor(() => {
+ expect(mockedTargetsApi.createTarget).toHaveBeenCalledWith({
+ type: "OpenAIVideoTarget",
+ params: {
+ endpoint: "https://api.openai.com",
+ model_name: "sora-2",
+ n_seconds: 8,
+ },
+ });
+ });
+ });
+
it("should render and submit metadata choices and booleans", async () => {
const user = userEvent.setup();
mockedTargetsApi.createTarget.mockResolvedValue(makeTarget({
diff --git a/frontend/src/components/Registry/CreateConverterDialog.test.tsx b/frontend/src/components/Registry/CreateConverterDialog.test.tsx
index e8b9783268..cb1852f591 100644
--- a/frontend/src/components/Registry/CreateConverterDialog.test.tsx
+++ b/frontend/src/components/Registry/CreateConverterDialog.test.tsx
@@ -282,7 +282,7 @@ describe('CreateConverterDialog', () => {
expect(mockedConvertersApi.createConverter).toHaveBeenCalledWith({
name: 'caesar-custom',
type: 'CaesarConverter',
- params: { caesar_offset: '5' },
+ params: { caesar_offset: 5 },
})
expect(onCreated).toHaveBeenCalledWith('caesar-custom')
})
@@ -332,11 +332,85 @@ describe('CreateConverterDialog', () => {
expect(mockedConvertersApi.createConverter).toHaveBeenCalledWith({
name: 'SearchReplaceConverter',
type: 'SearchReplaceConverter',
- params: { pattern: 'hello', replace: 'world', regex_flags: '0' },
+ params: { pattern: 'hello', replace: 'world', regex_flags: 0 },
})
},
)
+ it('should submit the catalog font size as a number', async () => {
+ const user = userEvent.setup()
+ mockConverterParameters([
+ { name: 'img_to_add', type_name: 'Path', is_list: false, choices: null, default: null, required: true },
+ { name: 'font_name', type_name: 'Path', is_list: false, choices: null, default: null, required: false },
+ { name: 'font_size', type_name: 'int', is_list: false, choices: null, default: '15', required: false },
+ { name: 'rotation', type_name: 'float', is_list: false, choices: null, default: '0.0', required: false },
+ { name: 'center_text', type_name: 'bool', is_list: false, choices: null, default: 'False', required: false },
+ ], 'AddImageTextConverter')
+ renderDialog()
+ await selectConverterType('AddImageTextConverter')
+
+ await user.type(screen.getByRole('textbox', { name: 'img_to_add *' }), 'data:image/png;base64,iVBORw0K')
+ const fontSize = screen.getByRole('textbox', { name: 'font_size' })
+ expect(fontSize).toHaveValue('15')
+ await user.clear(fontSize)
+ await user.type(fontSize, 'large')
+ await user.click(screen.getByRole('button', { name: 'Add Converter' }))
+ expect(screen.getByText('font_size must be a number.')).toBeInTheDocument()
+ expect(mockedConvertersApi.createConverter).not.toHaveBeenCalled()
+
+ await user.clear(fontSize)
+ await user.type(fontSize, '24')
+ await user.click(screen.getByRole('button', { name: 'Add Converter' }))
+ expect(mockedConvertersApi.createConverter).toHaveBeenCalledWith({
+ name: 'AddImageTextConverter',
+ type: 'AddImageTextConverter',
+ params: { img_to_add: 'data:image/png;base64,iVBORw0K', font_size: 24, rotation: 0 },
+ })
+ })
+
+ it('should submit the catalog SATA word lists as arrays', async () => {
+ const user = userEvent.setup()
+ mockConverterParameters([
+ { name: 'mask_token', type_name: 'str', is_list: false, choices: null, default: '[MASK]', required: false },
+ { ...wordSelectionParameter, name: 'selection_strategy' },
+ { name: 'num_masks', type_name: 'int', is_list: false, choices: null, default: null, required: false },
+ { name: 'stopwords', type_name: 'list[str]', is_list: true, choices: null, default: null, required: false },
+ {
+ name: 'candidate_words', type_name: 'list[str]', is_list: true, choices: null, default: null, required: false,
+ },
+ ], 'SATAMaskingConverter')
+ renderDialog()
+ await selectConverterType('SATAMaskingConverter')
+
+ await user.type(screen.getByRole('textbox', { name: 'stopwords' }), 'the, a')
+ await user.type(screen.getByRole('textbox', { name: 'candidate_words' }), 'bomb')
+ await user.click(screen.getByRole('button', { name: 'Add Converter' }))
+ expect(mockedConvertersApi.createConverter).toHaveBeenCalledWith({
+ name: 'SATAMaskingConverter',
+ type: 'SATAMaskingConverter',
+ params: { mask_token: '[MASK]', stopwords: ['the', 'a'], candidate_words: ['bomb'] },
+ })
+ })
+
+ it('should send text values exactly as entered', async () => {
+ const user = userEvent.setup()
+ mockConverterParameters([
+ { name: 'pattern', type_name: 'str', required: true },
+ { name: 'replace', type_name: 'str', required: true },
+ ], 'SearchReplaceConverter')
+ renderDialog()
+ await selectConverterType('SearchReplaceConverter')
+
+ await user.type(screen.getByRole('textbox', { name: 'pattern *' }), 'cat')
+ await user.type(screen.getByRole('textbox', { name: 'replace *' }), ' dog ')
+ await user.click(screen.getByRole('button', { name: 'Add Converter' }))
+ expect(mockedConvertersApi.createConverter).toHaveBeenCalledWith({
+ name: 'SearchReplaceConverter',
+ type: 'SearchReplaceConverter',
+ params: { pattern: 'cat', replace: ' dog ' },
+ })
+ })
+
it('selects a registered target for a target reference parameter', async () => {
mockedConvertersApi.listConverterTypes.mockResolvedValue({
items: [
diff --git a/frontend/src/components/Registry/CreateConverterDialog.tsx b/frontend/src/components/Registry/CreateConverterDialog.tsx
index a32c0c0a0d..9d6c56ff23 100644
--- a/frontend/src/components/Registry/CreateConverterDialog.tsx
+++ b/frontend/src/components/Registry/CreateConverterDialog.tsx
@@ -28,7 +28,9 @@ import ParameterField from '@/components/Parameters/ParameterField'
import {
buildParametersFromForm,
getInitialFormValues,
+ getParameterControlKind,
isStructuredParameterFormValue,
+ type ParameterControlKind,
type ParameterFormValue,
} from '@/components/Parameters/parameterForm'
@@ -117,6 +119,15 @@ function canConfigureConverterType(converterType: ConverterTypeEntry): boolean {
)
}
+// Text, JSON, and multiselect values go to the API exactly as entered: whitespace can be the
+// value (a replacement, for one), and this dialog shows those parameters as text or a single
+// select. The other controls are converted to the JSON type their parameter describes.
+const TYPED_CONTROL_KINDS = new Set(['structured', 'boolean', 'select', 'list', 'number'])
+
+function sendsTypedValue(parameter: Parameter): boolean {
+ return TYPED_CONTROL_KINDS.has(getParameterControlKind(parameter))
+}
+
function parameterDefaultValue(parameter: Parameter): string {
if (Array.isArray(parameter.default)) {
return parameter.default.join(', ')
@@ -408,21 +419,23 @@ export default function CreateConverterDialog({
}
const parameters = selectedConverterType?.parameters ?? []
- const params = Object.fromEntries(
- Object.entries(parameterValues).filter(([, value]) => !isStructuredParameterFormValue(value)),
- )
- const structured = buildParametersFromForm(
- parameters.filter((parameter) => parameter.variants),
- parameterValues,
+ const params: Record = Object.fromEntries(
+ parameters
+ .filter((parameter) => !sendsTypedValue(parameter))
+ .flatMap((parameter) => {
+ const value = parameterValues[parameter.name]
+ return typeof value === 'string' ? [[parameter.name, value]] : []
+ }),
)
- if (!structured.ok) {
+ const typed = buildParametersFromForm(parameters.filter(sendsTypedValue), parameterValues)
+ if (!typed.ok) {
// Not tagged as a submission failure: nothing was disabled, so the keyboard
// is still on the primary action and has nothing to be restored from.
- setError({ message: structured.error, fromSubmit: false })
+ setError({ message: typed.error, fromSubmit: false })
return
}
- if (structured.parameters) {
- Object.assign(params, structured.parameters)
+ if (typed.parameters) {
+ Object.assign(params, typed.parameters)
}
const epoch = openEpochRef.current
diff --git a/pyrit/backend/services/converter_service.py b/pyrit/backend/services/converter_service.py
index 5ba393fc9d..2f31db3a1f 100644
--- a/pyrit/backend/services/converter_service.py
+++ b/pyrit/backend/services/converter_service.py
@@ -112,9 +112,11 @@ async def list_converter_types_async(self) -> ConverterTypeResponse:
"""
List all available converter types from the converter class registry.
- Returns every constructible converter. Deciding which entries to surface
- to a user is a presentation concern owned by the caller (e.g. the
- frontend), not this service.
+ Returns every converter that external callers can build, with only the
+ parameters they may supply, each described in the form callers send it;
+ converters that need a Python object for a required parameter are left out.
+ Deciding which entries to surface to a user is a presentation concern owned
+ by the caller (e.g. the frontend), not this service.
Returns:
ConverterTypeResponse containing all available converter classes.
@@ -124,11 +126,14 @@ async def list_converter_types_async(self) -> ConverterTypeResponse:
converter_type=metadata.class_name,
supported_input_types=list(metadata.supported_input_types),
supported_output_types=list(metadata.supported_output_types),
- parameters=list(metadata.parameters),
+ parameters=[
+ parameter.for_external_catalog() for parameter in metadata.parameters if parameter.is_external_input
+ ],
is_llm_based=metadata.is_llm_based,
description=metadata.class_description or None,
)
for metadata in self._registry.get_all_registered_class_metadata()
+ if all(parameter.is_external_input for parameter in metadata.parameters if parameter.required)
]
return ConverterTypeResponse(items=items)
@@ -197,7 +202,7 @@ async def create_converter_async(self, *, request: CreateConverterRequest) -> Co
try:
# Uploads may have yielded to another request that took the name.
self._registry.instances.validate_name_available(request.name)
- converter_obj = self._registry.create_instance(request.type, **params)
+ converter_obj = self._registry.create_instance_from_external_input(request.type, params=params)
converter = self._build_instance_from_object(converter_id=request.name, converter_obj=converter_obj)
self._registry.instances.register(
converter_obj,
diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py
index aa0339e32d..2f6928efb0 100644
--- a/pyrit/backend/services/scenario_run_service.py
+++ b/pyrit/backend/services/scenario_run_service.py
@@ -76,7 +76,7 @@
)
from pyrit.prompt_target import PromptTarget
from pyrit.registry import InitializerRegistry, ScenarioRegistry
-from pyrit.registry.resolution import resolve_declared_params
+from pyrit.registry.resolution import reject_non_external_params, resolve_declared_params
from pyrit.scenario import Scenario
from pyrit.scenario.core import override_default_adversarial_target
@@ -626,6 +626,12 @@ async def _prepare_run_async(self, *, request: RunScenarioRequest) -> _PreparedR
ValueError: If scenario, target, initializer, or technique cannot be found.
"""
scenario_class = self._configuration_resolver.resolve_scenario_class(scenario_name=request.scenario_name)
+ if request.scenario_params:
+ reject_non_external_params(
+ params=request.scenario_params,
+ declared=scenario_class.supported_parameters(),
+ owner=request.scenario_name,
+ )
await self._run_initializers_async(request=request)
objective_target = self._configuration_resolver.resolve_target(target_name=request.target_name)
adversarial_target = self._configuration_resolver.resolve_adversarial_target(
diff --git a/pyrit/backend/services/scenario_service.py b/pyrit/backend/services/scenario_service.py
index 1596e2c59b..1c24f04e2d 100644
--- a/pyrit/backend/services/scenario_service.py
+++ b/pyrit/backend/services/scenario_service.py
@@ -21,6 +21,7 @@
ScenarioRunSizeEstimateRequest,
)
from pyrit.registry import ScenarioMetadata, ScenarioRegistry
+from pyrit.registry.resolution import reject_non_external_params
from pyrit.scenario.core import Scenario, override_default_adversarial_target
from pyrit.scenario.core.dataset_configuration import read_only_dataset_resolution
@@ -68,7 +69,11 @@ def _metadata_to_registered_scenario(
all_techniques=list(metadata.all_techniques),
technique_summaries=list(metadata.technique_summaries),
default_datasets=list(metadata.default_datasets),
- supported_parameters=list(metadata.supported_parameters),
+ supported_parameters=[
+ parameter.for_external_catalog()
+ for parameter in metadata.supported_parameters
+ if parameter.is_external_input
+ ],
baseline_policy=metadata.baseline_policy,
include_baseline_by_default=metadata.include_baseline_by_default,
uses_default_adversarial_target=metadata.uses_default_adversarial_target,
@@ -193,6 +198,12 @@ async def estimate_scenario_run_size_async(
scenario_class = self._registry.get_class(scenario_name)
except KeyError:
return None
+ if request.scenario_params:
+ reject_non_external_params(
+ params=request.scenario_params,
+ declared=scenario_class.supported_parameters(),
+ owner=scenario_name,
+ )
estimate_key = self._build_configured_estimate_key(
scenario_name=scenario_name, scenario_class=scenario_class, request=request
diff --git a/pyrit/backend/services/scorer_service.py b/pyrit/backend/services/scorer_service.py
index ba8b711e44..98282610bf 100644
--- a/pyrit/backend/services/scorer_service.py
+++ b/pyrit/backend/services/scorer_service.py
@@ -36,21 +36,30 @@ def _build_instance(self, *, name: str, scorer: Any) -> ScorerInstance:
async def list_scorer_types_async(self) -> ScorerTypeResponse:
"""
- List registered scorer class metadata without constructing scorers.
+ List the scorer types external callers can build, without constructing scorers.
+
+ Each entry lists only the parameters external callers may supply, each
+ described in the form callers send it; types that need a Python object for a
+ required parameter are left out.
Returns:
- ScorerTypeResponse: All registered scorer type metadata.
+ ScorerTypeResponse: Scorer type metadata for external callers.
"""
def list_types() -> ScorerTypeResponse:
items = [
ScorerTypeEntry(
scorer_type=metadata.class_name,
- parameters=list(metadata.parameters),
+ parameters=[
+ parameter.for_external_catalog()
+ for parameter in metadata.parameters
+ if parameter.is_external_input
+ ],
is_llm_based=metadata.is_llm_based,
description=metadata.class_description or None,
)
for metadata in self._registry.get_all_registered_class_metadata()
+ if all(parameter.is_external_input for parameter in metadata.parameters if parameter.required)
]
return ScorerTypeResponse(items=items)
@@ -110,6 +119,7 @@ def create() -> ScorerInstance:
name=request.name,
type_name=request.type,
params=request.params,
+ external_input=True,
)
return self._build_instance(name=request.name, scorer=scorer)
diff --git a/pyrit/backend/services/target_service.py b/pyrit/backend/services/target_service.py
index 8d1176ed07..eb56dbce62 100644
--- a/pyrit/backend/services/target_service.py
+++ b/pyrit/backend/services/target_service.py
@@ -188,9 +188,11 @@ async def list_target_types_async(self) -> TargetTypeResponse:
"""
List all available target types from the target class registry.
- Returns every constructible target with its derived constructor
- parameters and the auth modes it supports, all projected from the
- registry's ``TargetMetadata``. Deciding which entries to surface to a
+ Returns every target that external callers can build, with the
+ constructor parameters they may supply, each described in the form callers
+ send it, and the auth modes it supports, all projected from the registry's
+ ``TargetMetadata``; targets that need a Python object for a required
+ parameter are left out. Deciding which entries to surface to a
user is a presentation concern owned by the caller (e.g. the frontend),
not this service.
@@ -201,14 +203,19 @@ async def list_target_types_async(self) -> TargetTypeResponse:
items: list[TargetTypeEntry] = [
TargetTypeEntry(
target_type=metadata.class_name,
- parameters=self._project_target_parameters(
- target_type=metadata.class_name,
- parameters=metadata.parameters,
- ),
+ parameters=[
+ parameter.for_external_catalog()
+ for parameter in self._project_target_parameters(
+ target_type=metadata.class_name,
+ parameters=metadata.parameters,
+ )
+ if parameter.is_external_input
+ ],
supported_auth_modes=self._get_supported_auth_modes(metadata.supported_auth_modes),
description=metadata.class_description or None,
)
for metadata in metadata_items
+ if all(parameter.is_external_input for parameter in metadata.parameters if parameter.required)
]
return TargetTypeResponse(items=items)
@@ -265,7 +272,7 @@ async def create_target_async(self, *, request: CreateTargetRequest) -> TargetIn
# Remove this generated fallback after that UI sends an explicit name.
target_registry_name = request.name or f"compat_{uuid.uuid4().hex}"
self._registry.instances.validate_name_available(target_registry_name)
- target_obj = self._registry.create_instance(request.type, **params)
+ target_obj = self._registry.create_instance_from_external_input(request.type, params=params)
target = self._build_instance_from_object(target_registry_name=target_registry_name, target_obj=target_obj)
self._registry.instances.register(target_obj, name=target_registry_name)
return target
diff --git a/pyrit/converter/word_doc_converter.py b/pyrit/converter/word_doc_converter.py
index aad439182e..a87c0cac39 100644
--- a/pyrit/converter/word_doc_converter.py
+++ b/pyrit/converter/word_doc_converter.py
@@ -7,6 +7,7 @@
import hashlib
from dataclasses import dataclass
from io import BytesIO
+from pathlib import Path # noqa: TC003 - registry annotation resolution
from typing import TYPE_CHECKING, Any
from docx import Document
@@ -16,8 +17,6 @@
from pyrit.memory import data_serializer_factory
if TYPE_CHECKING:
- from pathlib import Path
-
from pyrit.memory import DataTypeSerializer
from pyrit.models import ComponentIdentifier, PromptDataType, SeedPrompt
diff --git a/pyrit/models/parameter.py b/pyrit/models/parameter.py
index 21eeb8c32c..d7f05c1f27 100644
--- a/pyrit/models/parameter.py
+++ b/pyrit/models/parameter.py
@@ -8,6 +8,7 @@
import copy
import types
from abc import ABC, abstractmethod
+from collections.abc import Collection, Sequence
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
@@ -277,6 +278,65 @@ def is_string_coercible(self) -> bool:
return False
return _is_scalar_param_type(_unwrap_optional(self.param_type))
+ @property
+ def is_external_input(self) -> bool:
+ """
+ Whether REST, CLI, and GUI callers may supply this parameter.
+
+ True for registry references, declared structured inputs, scalars (``Path`` and
+ ``Path | str`` included), flat ``list`` / ``Collection`` / ``Sequence`` of non-path
+ scalars, and other unions with one of those as an alternative and no path
+ alternative. External callers supply that
+ alternative, as for ``api_key: str | Callable[...]`` or
+ ``font_size: int | tuple[int, int]``; the other alternatives are for in-process
+ callers. Other parameters take Python objects from in-process callers only.
+
+ Returns:
+ bool: True when external callers may supply this parameter.
+ """
+ if self.reference is not None or self.variants is not None:
+ return True
+ if self.opaque:
+ return False
+ param_type = _unwrap_optional(self.param_type)
+ if _is_scalar_param_type(param_type):
+ return True
+ if get_origin(param_type) in (Union, types.UnionType):
+ members = [member for member in get_args(param_type) if member is not type(None)]
+ return not any(_mentions_path(member) for member in members) and any(
+ _is_non_path_json_type(member) for member in members
+ )
+ return _is_non_path_json_type(param_type)
+
+ def for_external_catalog(self) -> Parameter:
+ """
+ Describe this parameter in the form external callers send it.
+
+ A flat ``Collection`` or ``Sequence`` is described as a ``list``, and a union as its first
+ alternative in declaration order that external callers can send, so
+ ``font_size: int | tuple[int, int]`` is described as ``int``; ``Path | str`` keeps its own
+ form. Other alternatives are still accepted and coercion is unchanged; only the
+ catalog description changes, and registry metadata keeps the full annotation. A default the
+ described form cannot hold is left out, so callers omit the value and the constructor
+ default applies.
+
+ Returns:
+ Parameter: This parameter, or a copy whose ``param_type`` is the external form.
+ """
+ if self.reference is not None or self.variants is not None or self.opaque:
+ return self
+ external_type = _external_input_type(self.param_type)
+ if external_type == self.param_type:
+ return self
+ described = self.model_copy(update={"param_type": external_type})
+ if self.default is None or self.default is REQUIRED_VALUE:
+ return described
+ try:
+ described.coerce_value(self.default)
+ except ValueError:
+ return described.model_copy(update={"default": None})
+ return described
+
def is_reference_to(self, component_type: ComponentType) -> bool:
"""
Whether this parameter is a registry reference to the given component family.
@@ -411,6 +471,58 @@ def _is_scalar_param_type(annotation: Any) -> bool:
return _is_enum_type(annotation)
+def _is_non_path_json_type(annotation: Any) -> bool:
+ """
+ Return whether the annotation is a non-path scalar or a flat collection of one.
+
+ A flat collection is a ``list``, ``Collection``, or ``Sequence`` of a single non-path
+ scalar; external callers send it as a JSON array, which reaches the constructor as a list.
+
+ Returns:
+ bool: True for ``str``/``int``/``float``/``bool``/``Literal``/``Enum`` or a flat collection of them.
+ """
+ if get_origin(annotation) in (list, Collection, Sequence):
+ type_args = get_args(annotation)
+ annotation = type_args[0] if len(type_args) == 1 else None
+ return _is_scalar_param_type(annotation) and annotation is not Path and not _is_path_or_str(annotation)
+
+
+def _mentions_path(annotation: Any) -> bool:
+ """
+ Return whether the annotation is ``Path`` or has ``Path`` among its type arguments.
+
+ Returns:
+ bool: True when a value of this type may be a local file path.
+ """
+ return annotation is Path or any(_mentions_path(argument) for argument in get_args(annotation))
+
+
+def _external_input_type(annotation: Any) -> Any:
+ """
+ Return the form external callers send for an annotation.
+
+ A flat ``Collection`` or ``Sequence`` becomes a ``list``, and a union becomes its first
+ alternative external callers can send, keeping ``None`` when the union allows it.
+ ``Path | str`` and every other annotation are returned unchanged.
+
+ Returns:
+ Any: The external form of the annotation.
+ """
+ if _is_path_or_str(annotation):
+ return annotation
+ origin = get_origin(annotation)
+ if origin in (Union, types.UnionType):
+ members = get_args(annotation)
+ supported = next((member for member in members if _is_non_path_json_type(member)), None)
+ if supported is None:
+ return annotation
+ external = _external_input_type(supported)
+ return external | None if type(None) in members else external
+ if origin in (Collection, Sequence) and len(get_args(annotation)) == 1:
+ return list[get_args(annotation)[0]] # ty: ignore[invalid-type-form]
+ return annotation
+
+
def _coerce_simple_value(*, param_name: str, annotation: Any, raw_value: Any) -> Any:
"""
Coerce ``raw_value`` to a scalar ``annotation`` — the shared coercion core.
diff --git a/pyrit/prompt_target/a2a_target.py b/pyrit/prompt_target/a2a_target.py
index a434e9e9bb..b607f701a3 100644
--- a/pyrit/prompt_target/a2a_target.py
+++ b/pyrit/prompt_target/a2a_target.py
@@ -8,7 +8,7 @@
import math
import time
import uuid
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Awaitable, Callable # noqa: TC003 - registry annotation resolution
from dataclasses import dataclass
from email.utils import parsedate_to_datetime
from typing import TYPE_CHECKING, Any, Literal, cast
@@ -31,8 +31,6 @@
from pyrit.prompt_target.common.utils import limit_requests_per_minute
if TYPE_CHECKING:
- from collections.abc import Awaitable, Callable
-
from a2a.client import Client
from a2a.types import a2a_pb2
diff --git a/pyrit/registry/registry.py b/pyrit/registry/registry.py
index 4328f97484..2376a5fb5e 100644
--- a/pyrit/registry/registry.py
+++ b/pyrit/registry/registry.py
@@ -720,6 +720,37 @@ def create_instance(self, name: str, **kwargs: object) -> T:
)
return cls(**resolved)
+ def create_instance_from_external_input(self, name: str, *, params: Mapping[str, object]) -> T:
+ """
+ Build a configured instance from external input such as a REST or CLI request.
+
+ Unlike ``create_instance``, which serves in-process callers that may pass any
+ Python object, this accepts only parameters with ``Parameter.is_external_input``
+ and registry references given by name. Anything else is rejected before
+ construction; the constructor still validates the values it receives.
+
+ Args:
+ name (str): The catalog name to build.
+ params (Mapping[str, object]): Constructor arguments from the external caller.
+
+ Returns:
+ T: The constructed instance.
+
+ Raises:
+ KeyError: If the name is not registered.
+ ValueError: If an argument is not a valid constructor parameter or not an
+ external input, a registry reference cannot be resolved, or a value
+ cannot be coerced.
+ """
+ cls = self.get_class(name)
+ resolved = resolve_constructor_args(
+ cls=cls,
+ raw_args=dict(params),
+ identifier_type=self._identifier_type(),
+ external_input=True,
+ )
+ return cls(**resolved)
+
def __contains__(self, name: str) -> bool:
"""
Check if a name is registered.
@@ -794,6 +825,7 @@ def create_named_instance(
type_name: str,
params: Mapping[str, object] | None = None,
registry_metadata: dict[str, Any] | None = None,
+ external_input: bool = False,
) -> InstanceT:
"""
Build and store a configured instance under an explicit name.
@@ -804,12 +836,20 @@ def create_named_instance(
params (Mapping[str, object] | None): Constructor arguments.
registry_metadata (dict[str, Any] | None): Per-entry metadata to store
with the instance.
+ external_input (bool): Whether ``params`` come from an external caller, in which
+ case the instance is built with ``create_instance_from_external_input``.
+ Defaults to False.
Returns:
InstanceT: The constructed and registered instance.
"""
self.instances.validate_name_available(name)
- instance = self.create_instance(type_name, **dict(params) if params is not None else {})
+ args = dict(params) if params is not None else {}
+ instance = (
+ self.create_instance_from_external_input(type_name, params=args)
+ if external_input
+ else self.create_instance(type_name, **args)
+ )
self.instances.register(instance, name=name, metadata=registry_metadata)
return instance
diff --git a/pyrit/registry/resolution.py b/pyrit/registry/resolution.py
index 265bd00e53..17e16dd498 100644
--- a/pyrit/registry/resolution.py
+++ b/pyrit/registry/resolution.py
@@ -21,6 +21,8 @@
constructor-ready keyword arguments — coercing simple string values via
``Parameter.coerce_value`` and resolving registry-reference parameters by name
from the owning domain's registry. Defaults are left to the constructor.
+ Callers building from external input (REST, CLI) select the external path,
+ which accepts only parameters with ``Parameter.is_external_input``.
- **Resolve from a declared list** (``resolve_declared_params``): the sibling for
a component that declares an explicit ``list[Parameter]`` (e.g. a scenario's
``supported_parameters()``). It has no references, coerces every supplied
@@ -57,7 +59,7 @@
from pyrit.models.parameter import display_choices as display_choices
if TYPE_CHECKING:
- from collections.abc import Callable
+ from collections.abc import Callable, Mapping
from pyrit.models.identifiers.component_identifier import ComponentIdentifier
@@ -559,6 +561,7 @@ def resolve_constructor_args(
cls: type,
raw_args: dict[str, Any],
identifier_type: type[ComponentIdentifier] | None = None,
+ external_input: bool = False,
) -> dict[str, Any]:
"""
Resolve a flat argument dict into constructor-ready keyword arguments.
@@ -574,15 +577,22 @@ def resolve_constructor_args(
identifier_type (type[ComponentIdentifier] | None): The domain identifier
whose ``Param.*`` markers declare which parameters are registry
references. When None, no parameter is treated as a reference.
+ external_input (bool): Whether ``raw_args`` come from an external caller (REST, CLI).
+ External callers may set only parameters with ``Parameter.is_external_input``
+ and must name registry references. Defaults to False (in-process callers, which
+ may pass any Python object).
Returns:
dict[str, Any]: Arguments ready to pass to ``cls(**resolved)``.
Raises:
ValueError: If an argument is not a declared parameter, a registry
- reference cannot be resolved, or a simple value cannot be coerced.
+ reference cannot be resolved, a simple value cannot be coerced, or
+ external input sets a parameter that is not an external input.
"""
by_name = {param.name: param for param in derive_parameters(cls=cls, identifier_type=identifier_type)}
+ if external_input:
+ reject_non_external_params(params=raw_args, declared=list(by_name.values()), owner=cls.__name__)
resolved: dict[str, Any] = {}
for name, value in raw_args.items():
@@ -622,6 +632,50 @@ def resolve_constructor_args(
return resolved
+def reject_non_external_params(*, params: Mapping[str, Any], declared: Sequence[Parameter], owner: str) -> None:
+ """
+ Reject external input that is not an explicitly supported external input.
+
+ Every name must be declared and have ``Parameter.is_external_input``, and registry
+ references must be given by name. Values are otherwise left to the component.
+
+ Args:
+ params (Mapping[str, Any]): The parameter values supplied by an external caller.
+ declared (Sequence[Parameter]): The parameters the component declares.
+ owner (str): The owning class or scenario name, for error messages.
+
+ Raises:
+ ValueError: If ``params`` sets an undeclared parameter or one that is not an external
+ input, or gives a registry reference as anything but a name.
+ """
+ declared_by_name = {parameter.name: parameter for parameter in declared}
+ for name, value in params.items():
+ parameter = declared_by_name.get(name)
+ if parameter is None:
+ external_names = sorted(
+ known for known, candidate in declared_by_name.items() if candidate.is_external_input
+ )
+ raise ValueError(f"Unknown parameter '{name}' for '{owner}'. Valid parameters: {external_names}")
+ if not parameter.is_external_input:
+ raise ValueError(
+ f"Parameter '{name}' of '{owner}' cannot be set through the API; pass it from Python instead."
+ )
+ if parameter.reference is not None:
+ _require_reference_names(value=value, owner=owner, name=name)
+
+
+def _require_reference_names(*, value: Any, owner: str, name: str) -> None:
+ """
+ Require external input for a registry reference to be a registry name or a list of names.
+
+ Raises:
+ ValueError: If the value is neither null, a name, nor a list of names.
+ """
+ names = value if isinstance(value, list) else [value]
+ if value is not None and not all(isinstance(item, str) for item in names):
+ raise ValueError(f"{owner}.{name}: expected a registry name, but got {type(value).__name__}.")
+
+
def _resolve_structured_input(*, parameter: Parameter, value: Any) -> Any:
"""
Build a declared structured-input variant from its JSON representation.
diff --git a/tests/unit/backend/test_converter_service.py b/tests/unit/backend/test_converter_service.py
index ed39ed57d7..da025112fc 100644
--- a/tests/unit/backend/test_converter_service.py
+++ b/tests/unit/backend/test_converter_service.py
@@ -9,11 +9,13 @@
import base64
import codecs
from collections.abc import AsyncGenerator
+from io import BytesIO
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, call, patch
import pytest
from fastapi import HTTPException
+from PIL import Image
from pydantic import ValidationError
from pyrit import converter
@@ -219,19 +221,17 @@ async def test_list_converter_types_includes_supported_types(self) -> None:
assert "text" in base64_entry.supported_input_types
assert "text" in base64_entry.supported_output_types
- async def test_types_include_all_constructible_converters(self) -> None:
- """The projection surfaces every constructible converter, including base/helper classes.
-
- Whether to display a given converter is left to the caller (e.g. the frontend),
- so the service no longer hides anything.
- """
+ async def test_types_omit_converters_that_need_python_objects(self) -> None:
+ """Converters whose required parameters take Python objects cannot be built from the API."""
service = ConverterService()
result = await service.list_converter_types_async()
converter_types = [item.converter_type for item in result.items]
assert "Base64Converter" in converter_types
- assert "SelectiveTextConverter" in converter_types
+ assert "SearchReplaceConverter" in converter_types
+ assert "SelectiveTextConverter" not in converter_types
+ assert "TextJailbreakConverter" not in converter_types
async def test_types_serialize_parameter_type(self) -> None:
"""Type entries render the raw annotation into a human-readable type_name."""
@@ -265,20 +265,71 @@ async def test_types_include_registry_reference_params(self) -> None:
target_param = next(param for param in persuasion_entry.parameters if param.name == "converter_target")
assert target_param.reference_type == "target"
- async def test_types_preserve_all_registry_parameters(self, upload_service: ConverterService) -> None:
+ async def test_types_expose_component_external_inputs(self, upload_service: ConverterService) -> None:
+ result = await upload_service.list_converter_types_async()
+ parameters = {
+ entry.converter_type: {parameter.name for parameter in entry.parameters} for entry in result.items
+ }
+
+ assert {
+ "GridCompositeConverter",
+ "SelectiveTextConverter",
+ "TextJailbreakConverter",
+ "TokenBijectionConverter",
+ }.isdisjoint(parameters)
+ assert "font_size" in parameters["AddImageTextConverter"]
+ assert {"stopwords", "candidate_words"} <= parameters["SATAMaskingConverter"]
+ assert {"existing_docx", "placeholder"} <= parameters["WordDocConverter"]
+ assert "existing_pdf" in parameters["PDFConverter"]
+ assert "font_color" not in parameters["PDFConverter"]
+
+ async def test_registry_metadata_keeps_parameters_the_api_cannot_set(
+ self, upload_service: ConverterService
+ ) -> None:
+ metadata = upload_service._registry.get_registered_class_metadata("PDFConverter")
+ assert metadata is not None
+ registry_parameters = {parameter.name: parameter for parameter in metadata.parameters}
+
+ assert registry_parameters["font_color"].type_name == "tuple[int, int, int]"
+ assert not registry_parameters["font_color"].is_external_input
+
+ async def test_types_describe_parameters_in_the_form_callers_send(self, upload_service: ConverterService) -> None:
result = await upload_service.list_converter_types_async()
- metadata_by_name = {
- metadata.class_name: metadata for metadata in upload_service._registry.get_all_registered_class_metadata()
+ serialized = {
+ (entry.converter_type, parameter["name"]): {
+ key: parameter[key] for key in ("type_name", "is_list", "choices", "default", "required")
+ }
+ for entry in result.items
+ for parameter in entry.model_dump(mode="json")["parameters"]
}
- assert {entry.converter_type for entry in result.items} == set(metadata_by_name)
- for entry in result.items:
- assert entry.parameters == list(metadata_by_name[entry.converter_type].parameters)
+ assert serialized[("AddImageTextConverter", "font_size")] == {
+ "type_name": "int",
+ "is_list": False,
+ "choices": None,
+ "default": "15",
+ "required": False,
+ }
+ for name in ("stopwords", "candidate_words"):
+ assert serialized[("SATAMaskingConverter", name)] == {
+ "type_name": "list[str]",
+ "is_list": True,
+ "choices": None,
+ "default": None,
+ "required": False,
+ }
+ registry = {
+ (metadata.class_name, parameter.name): parameter.type_name
+ for metadata in upload_service._registry.get_all_registered_class_metadata()
+ for parameter in metadata.parameters
+ }
+ assert registry[("AddImageTextConverter", "font_size")] == "int | tuple[int, int]"
+ assert registry[("SATAMaskingConverter", "stopwords")] == "collections.abc.Collection[str]"
@pytest.mark.parametrize(
("converter_type", "parameter_name", "type_name", "required", "is_list"),
[
- ("SearchReplaceConverter", "replace", "str | list[str]", True, False),
+ ("SearchReplaceConverter", "replace", "str", True, False),
("DenylistConverter", "denylist", "list[str]", False, True),
],
)
@@ -419,6 +470,27 @@ async def test_create_converter_raises_for_invalid_type(self) -> None:
with pytest.raises(ValueError, match="not found"):
await service.create_converter_async(request=request)
+ async def test_create_converter_rejects_parameters_that_take_python_objects(self) -> None:
+ service = ConverterService()
+ request = CreateConverterRequest(
+ name="jailbreak", type="TextJailbreakConverter", params={"jailbreak_template": {"name": "x"}}
+ )
+
+ with pytest.raises(ValueError, match="'jailbreak_template' of 'TextJailbreakConverter' cannot be set"):
+ await service.create_converter_async(request=request)
+
+ assert service.get_converter_object(converter_id="jailbreak") is None
+
+ async def test_create_converter_accepts_string_for_string_union_parameter(self) -> None:
+ service = ConverterService()
+ request = CreateConverterRequest(
+ name="replace", type="SearchReplaceConverter", params={"pattern": "a", "replace": "b"}
+ )
+
+ result = await service.create_converter_async(request=request)
+
+ assert result.converter_id == "replace"
+
async def test_create_converter_success(self) -> None:
"""Test successful converter creation."""
service = ConverterService()
@@ -773,9 +845,7 @@ async def test_persist_data_uri_rejects_non_base64_data_uri(self, upload_service
assert list(service._upload_path.iterdir()) == []
- async def test_create_converter_cleans_upload_when_construction_fails(
- self, upload_service: ConverterService
- ) -> None:
+ async def test_create_converter_cleans_upload_when_creation_fails(self, upload_service: ConverterService) -> None:
service = upload_service
params = {
"existing_pdf": _make_data_uri(mime_type="application/pdf", content=b"%PDF-1.4\n"),
@@ -783,12 +853,76 @@ async def test_create_converter_cleans_upload_when_construction_fails(
}
request = CreateConverterRequest(name="invalid-pdf", type="PDFConverter", params=params)
- with pytest.raises(ValueError, match="Invalid font_color"):
+ with pytest.raises(ValueError, match="'font_color' of 'PDFConverter' cannot be set through the API"):
await service.create_converter_async(request=request)
assert service._registry.instances.get("invalid-pdf") is None
assert list(service._upload_path.iterdir()) == []
+ async def test_create_converter_cleans_upload_when_constructor_rejects_value(
+ self, upload_service: ConverterService
+ ) -> None:
+ docx_mime_type = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
+ request = CreateConverterRequest(
+ name="invalid-word-doc",
+ type="WordDocConverter",
+ params={
+ "existing_docx": _make_data_uri(mime_type=docx_mime_type, content=b"PK\x03\x04"),
+ "placeholder": "",
+ },
+ )
+
+ with pytest.raises(ValueError, match="Placeholder must be a non-empty string"):
+ await upload_service.create_converter_async(request=request)
+
+ assert upload_service._registry.instances.get("invalid-word-doc") is None
+ assert list(upload_service._upload_path.iterdir()) == []
+
+ async def test_create_converter_accepts_scalar_alternative_of_union(self, upload_service: ConverterService) -> None:
+ image = BytesIO()
+ Image.new("RGB", (8, 8)).save(image, format="PNG")
+ request = CreateConverterRequest(
+ name="sized-text",
+ type="AddImageTextConverter",
+ params={"img_to_add": _make_data_uri(mime_type="image/png", content=image.getvalue()), "font_size": 24},
+ )
+
+ response = await upload_service.create_converter_async(request=request)
+
+ converter = upload_service.get_converter_object(converter_id=response.converter_id)
+ assert converter._font_size == 24
+
+ async def test_create_converter_accepts_word_lists(self, upload_service: ConverterService) -> None:
+ request = CreateConverterRequest(
+ name="masked",
+ type="SATAMaskingConverter",
+ params={"stopwords": ["the", "a"], "candidate_words": ["bomb"]},
+ )
+
+ response = await upload_service.create_converter_async(request=request)
+
+ strategy_params = upload_service.get_converter_object(
+ converter_id=response.converter_id
+ )._selection_strategy.get_identifier_params()
+ assert strategy_params["stopwords"] == ["a", "the"]
+ assert strategy_params["candidate_words"] == ["bomb"]
+
+ @pytest.mark.parametrize("innocuous_image", ["/etc/hosts", "https://example.com/cat.png"])
+ async def test_create_converter_rejects_file_collections(
+ self, upload_service: ConverterService, innocuous_image: str
+ ) -> None:
+ request = CreateConverterRequest(
+ name="grid",
+ type="GridCompositeConverter",
+ params={"innocuous_images": [innocuous_image]},
+ )
+
+ with pytest.raises(ValueError, match="'innocuous_images' of 'GridCompositeConverter' cannot be set"):
+ await upload_service.create_converter_async(request=request)
+
+ assert upload_service._registry.instances.get("grid") is None
+ assert list(upload_service._upload_path.iterdir()) == []
+
async def test_create_converter_registers_nothing_when_response_mapping_fails(
self, upload_service: ConverterService
) -> None:
diff --git a/tests/unit/backend/test_scenario_run_service.py b/tests/unit/backend/test_scenario_run_service.py
index 35c05dcf3e..9f687b7808 100644
--- a/tests/unit/backend/test_scenario_run_service.py
+++ b/tests/unit/backend/test_scenario_run_service.py
@@ -65,6 +65,8 @@
get_default_adversarial_target,
)
from pyrit.scenario.core.scenario_technique import ScenarioTechnique
+from pyrit.scenario.scenarios.airt.jailbreak import Jailbreak
+from pyrit.scenario.scenarios.airt.scam import Scam
from pyrit.score.scorer_evaluation.scorer_metrics import ObjectiveScorerMetrics
from unit.mocks import MockPromptTarget, get_mock_target_identifier, make_scenario_result
@@ -160,6 +162,50 @@ def _make_request(
)
+@pytest.mark.parametrize(
+ ("scenario_params", "rejected"),
+ [
+ ({"dataset_config": "x"}, "'dataset_config' of 'airt.scam' cannot be set through the API"),
+ ({"objective_target": {"name": "x"}}, "airt.scam.objective_target: expected a registry name"),
+ ({"unknown": None}, "Unknown parameter 'unknown' for 'airt.scam'"),
+ ({"max_concurrency": 2}, None),
+ ],
+)
+async def test_start_run_checks_scenario_params_in_the_preparation_worker(
+ patch_central_database: MagicMock, scenario_params: dict[str, Any], rejected: str | None
+) -> None:
+ loop_thread = threading.current_thread()
+ lookup_threads: list[threading.Thread] = []
+
+ def get_class(name: str) -> type[Scenario]:
+ lookup_threads.append(threading.current_thread())
+ return Scam
+
+ registry = MagicMock(spec=ScenarioRegistry)
+ registry.get_class.side_effect = get_class
+ service = ScenarioRunService()
+ request = _make_request(scenario_name="airt.scam", scenario_params=scenario_params)
+
+ with (
+ patch(f"{_REGISTRY_PATCH_BASE}.ScenarioRegistry.get_registry_singleton", return_value=registry),
+ patch.object(
+ service,
+ "_run_initializers_async",
+ new_callable=AsyncMock,
+ side_effect=RuntimeError("initializers reached"),
+ ) as run_initializers,
+ pytest.raises(ValueError if rejected else RuntimeError, match=rejected or "initializers reached"),
+ ):
+ await service.start_run_async(request=request)
+
+ assert lookup_threads
+ assert loop_thread not in lookup_threads
+ assert run_initializers.await_count == (0 if rejected else 1)
+ # The finished preparation's done callback can still be queued when its error reaches the caller.
+ await asyncio.sleep(0)
+ await service.close_async()
+
+
def _make_db_scenario_result(
*,
result_id: str = "sr-uuid-1",
@@ -621,6 +667,8 @@ def get_aggregate_tags(cls) -> set[str]:
scenario_instance = mock_all_registries["scenario_instance"]
scenario_instance._technique_class = _JailbreakTechnique
+ mock_sr = mock_all_registries["scenario_registry"]
+ mock_sr.get_class.return_value.supported_parameters.return_value = Jailbreak.supported_parameters()
objective_target = mock_all_registries["target_registry"].instances.get.return_value
scenario_params = {"num_jailbreaks": 2, "num_jailbreak_attempts": 1}
@@ -663,6 +711,7 @@ def get_aggregate_tags(cls) -> set[str]:
service = ScenarioRunService()
mock_sr = mock_all_registries["scenario_registry"]
+ mock_sr.get_class.return_value.supported_parameters.return_value = Jailbreak.supported_parameters()
mock_memory = mock_all_registries["memory"]
mock_all_registries["scenario_instance"]._technique_class = _JailbreakTechnique
records: dict[str, MagicMock] = {}
diff --git a/tests/unit/backend/test_scenario_service.py b/tests/unit/backend/test_scenario_service.py
index ee6bb2f21b..4949df6752 100644
--- a/tests/unit/backend/test_scenario_service.py
+++ b/tests/unit/backend/test_scenario_service.py
@@ -8,6 +8,8 @@
import asyncio
import threading
from collections import OrderedDict
+from collections.abc import Collection
+from dataclasses import replace
from typing import TYPE_CHECKING, Literal
from unittest.mock import AsyncMock, MagicMock, patch
@@ -133,6 +135,66 @@ def _make_scenario_metadata(
)
+def test_catalog_lists_only_external_scenario_parameters() -> None:
+ metadata = replace(
+ _make_scenario_metadata(),
+ supported_parameters=(
+ Parameter(name="dataset_config", description="Dataset source configuration.", opaque=True),
+ Parameter(name="max_concurrency", description="Maximum concurrency.", param_type=int, default=4),
+ ),
+ )
+
+ summary = _metadata_to_registered_scenario(metadata=metadata)
+
+ assert [parameter.name for parameter in summary.supported_parameters] == ["max_concurrency"]
+
+
+def test_catalog_describes_scenario_parameters_in_the_form_callers_send() -> None:
+ metadata = replace(
+ _make_scenario_metadata(),
+ supported_parameters=(
+ Parameter(name="words", description="Words to skip.", param_type=Collection[str] | None, default=None),
+ Parameter(name="size", description="Size or bounds.", param_type=int | tuple[int, int], default=(8, 20)),
+ ),
+ )
+
+ summary = _metadata_to_registered_scenario(metadata=metadata)
+
+ described = {
+ parameter["name"]: {key: parameter[key] for key in ("type_name", "is_list", "default")}
+ for parameter in summary.model_dump(mode="json")["supported_parameters"]
+ }
+ assert described == {
+ "words": {"type_name": "list[str]", "is_list": True, "default": None},
+ "size": {"type_name": "int", "is_list": False, "default": None},
+ }
+ assert metadata.supported_parameters[0].param_type == Collection[str] | None
+
+
+@pytest.mark.parametrize(
+ ("scenario_params", "message"),
+ [
+ ({"dataset_config": "x"}, "'dataset_config' of 'airt.scam' cannot be set through the API"),
+ ({"objective_target": {"name": "x"}}, "airt.scam.objective_target: expected a registry name"),
+ ({"unknown": None}, "Unknown parameter 'unknown' for 'airt.scam'"),
+ ],
+)
+async def test_configured_estimate_rejects_unsupported_scenario_params_async(
+ scenario_params: dict[str, object], message: str
+) -> None:
+ registry = MagicMock(spec=ScenarioRegistry)
+ registry.get_class.return_value = Scam
+ with patch.object(ScenarioRegistry, "get_registry_singleton", return_value=registry):
+ service = ScenarioService()
+ with pytest.raises(ValueError, match=message):
+ await service.estimate_scenario_run_size_async(
+ scenario_name="airt.scam",
+ request=ScenarioRunSizeEstimateRequest(scenario_params=scenario_params),
+ )
+
+ registry.create_and_estimate_async.assert_not_called()
+
+
@pytest.mark.parametrize("uses_default", [False, True])
def test_catalog_preserves_adversarial_default_usage(uses_default: bool) -> None:
"""The public catalog exposes usage from real registry metadata."""
@@ -1095,6 +1157,9 @@ async def estimate_async(
service = ScenarioService()
service._registry = MagicMock()
service._registry.get_registered_class_metadata.return_value = metadata
+ service._registry.get_class.return_value.supported_parameters.return_value = [
+ Parameter(name=name, description="", param_type=int) for name in ("first", "second")
+ ]
service._estimate_configured_run_size_async = AsyncMock(side_effect=estimate_async)
first = asyncio.create_task(
service.estimate_scenario_run_size_async(
@@ -1252,6 +1317,9 @@ async def estimate_async(
service = ScenarioService()
service._registry = MagicMock()
service._registry.get_registered_class_metadata.return_value = metadata
+ service._registry.get_class.return_value.supported_parameters.return_value = [
+ Parameter(name="request_index", description="", param_type=int)
+ ]
service._configured_estimate_semaphore = asyncio.Semaphore(2)
service._estimate_configured_run_size_async = AsyncMock(side_effect=estimate_async)
@@ -1457,6 +1525,10 @@ async def test_configured_estimate_uses_shared_launch_resolution(self) -> None:
introspection_instance._technique_class = _EstimateTechnique
introspection_instance._default_dataset_config = DatasetAttackConfiguration(dataset_names=["harmbench"])
scenario_class = MagicMock(return_value=introspection_instance)
+ scenario_class.supported_parameters.return_value = [
+ Parameter(name=name, description="", param_type=int)
+ for name in ("num_jailbreaks", "num_jailbreak_attempts")
+ ]
objective_target = MagicMock()
with (
diff --git a/tests/unit/backend/test_scorer_service.py b/tests/unit/backend/test_scorer_service.py
index b3f6d9dff2..a1aed98abc 100644
--- a/tests/unit/backend/test_scorer_service.py
+++ b/tests/unit/backend/test_scorer_service.py
@@ -5,7 +5,7 @@
import asyncio
import threading
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
from unittest.mock import MagicMock, patch
import pytest
@@ -42,6 +42,14 @@ def get_scorer_metrics(self):
return None
+class _ObjectConfiguredScorer(_ServiceScorer):
+ """A scorer with one external parameter and one that takes a Python object."""
+
+ def __init__(self, *, label: str = "service", tokenizer: Callable[[str], list[str]] | None = None) -> None:
+ super().__init__(label=label)
+ self.tokenizer = tokenizer
+
+
@pytest.fixture(autouse=True)
def reset_scorer_registry(patch_central_database: MagicMock) -> Iterator[None]:
get_scorer_service.cache_clear()
@@ -159,3 +167,50 @@ async def test_close_services_clears_scorer_cache_before_registry_replacement()
await new_service.create_scorer_async(request=CreateScorerRequest(name="created", type="_ServiceScorer"))
assert new_registry.instances.get("created") is not None
assert old_registry.instances.get("created") is None
+
+
+async def test_create_scorer_accepts_only_external_inputs() -> None:
+ ScorerRegistry.get_registry_singleton().register_class(_ObjectConfiguredScorer)
+ service = ScorerService()
+
+ with pytest.raises(ValueError, match="'tokenizer' of '_ObjectConfiguredScorer' cannot be set through the API"):
+ await service.create_scorer_async(
+ request=CreateScorerRequest(name="tokenized", type="_ObjectConfiguredScorer", params={"tokenizer": "x"})
+ )
+ created = await service.create_scorer_async(
+ request=CreateScorerRequest(name="labeled", type="_ObjectConfiguredScorer", params={"label": "custom"})
+ )
+
+ assert created.scorer_registry_name == "labeled"
+ assert ScorerRegistry.get_registry_singleton().instances.get("tokenized") is None
+
+
+async def test_types_list_only_external_inputs() -> None:
+ registry = ScorerRegistry.get_registry_singleton()
+ registry.register_class(_ObjectConfiguredScorer)
+
+ types = await ScorerService().list_scorer_types_async()
+
+ entry = next(item for item in types.items if item.scorer_type == "_ObjectConfiguredScorer")
+ assert [parameter.name for parameter in entry.parameters] == ["label"]
+ metadata = registry.get_registered_class_metadata("_ObjectConfiguredScorer")
+ assert metadata is not None
+ assert {parameter.name for parameter in metadata.parameters} == {"label", "tokenizer"}
+
+
+async def test_types_describe_sequences_as_lists() -> None:
+ types = await ScorerService().list_scorer_types_async()
+
+ serialized = {
+ (entry.scorer_type, parameter["name"]): (parameter["type_name"], parameter["is_list"], parameter["required"])
+ for entry in types.items
+ for parameter in entry.model_dump(mode="json")["parameters"]
+ }
+ assert serialized[("AgentThreatRulesScorer", "fields")] == ("list[str]", True, False)
+ assert serialized[("InsecureCodeScorer", "harm_categories")] == ("list[str]", True, True)
+ assert serialized[("InsecureCodeScorer", "system_prompt")] == ("str", False, True)
+ metadata = ScorerRegistry.get_registry_singleton().get_registered_class_metadata("AgentThreatRulesScorer")
+ assert metadata is not None
+ assert next(parameter for parameter in metadata.parameters if parameter.name == "fields").type_name == (
+ "collections.abc.Sequence[str]"
+ )
diff --git a/tests/unit/backend/test_target_service.py b/tests/unit/backend/test_target_service.py
index e6803cc181..5b618db901 100644
--- a/tests/unit/backend/test_target_service.py
+++ b/tests/unit/backend/test_target_service.py
@@ -297,24 +297,59 @@ async def test_types_include_structured_parameters(self) -> None:
assert weights_parameter.is_list is True
assert weights_parameter.required is False
- async def test_types_preserve_registry_parameter_order_without_mutating_metadata(self) -> None:
+ async def test_types_keep_registry_parameter_order_without_mutating_metadata(self) -> None:
service = TargetService()
result = await service.list_target_types_async()
metadata_by_name = {
metadata.class_name: metadata for metadata in service._registry.get_all_registered_class_metadata()
}
- assert {entry.target_type for entry in result.items} == set(metadata_by_name)
for entry in result.items:
- registry_parameters = metadata_by_name[entry.target_type].parameters
- assert [parameter.name for parameter in entry.parameters] == [
- parameter.name for parameter in registry_parameters
- ]
+ entry_names = [parameter.name for parameter in entry.parameters]
+ registry_names = [parameter.name for parameter in metadata_by_name[entry.target_type].parameters]
+ assert entry_names == [name for name in registry_names if name in entry_names]
registry_openai = {parameter.name: parameter for parameter in metadata_by_name["OpenAIChatTarget"].parameters}
assert registry_openai["endpoint"].required is False
assert registry_openai["model_name"].required is False
+ async def test_types_expose_component_external_inputs(self) -> None:
+ service = TargetService()
+ result = await service.list_target_types_async()
+
+ parameters = {entry.target_type: {parameter.name for parameter in entry.parameters} for entry in result.items}
+ assert {"PlaywrightTarget", "PlaywrightCopilotTarget", "WebsocketTarget"}.isdisjoint(parameters)
+ assert "api_key" in parameters["OpenAIChatTarget"]
+ assert "custom_configuration" not in parameters["OpenAIChatTarget"]
+ assert "n_seconds" in parameters["OpenAIVideoTarget"]
+
+ async def test_registry_metadata_keeps_parameters_the_api_cannot_set(self) -> None:
+ metadata = TargetService()._registry.get_registered_class_metadata("OpenAIChatTarget")
+ assert metadata is not None
+ registry_parameters = {parameter.name: parameter for parameter in metadata.parameters}
+
+ assert not registry_parameters["custom_configuration"].is_external_input
+
+ async def test_types_describe_the_video_duration_as_an_int(self) -> None:
+ service = TargetService()
+ result = await service.list_target_types_async()
+
+ entry = next(item for item in result.items if item.target_type == "OpenAIVideoTarget")
+ n_seconds = next(
+ parameter for parameter in entry.model_dump(mode="json")["parameters"] if parameter["name"] == "n_seconds"
+ )
+ assert {key: n_seconds[key] for key in ("type_name", "is_list", "choices", "default", "required")} == {
+ "type_name": "int",
+ "is_list": False,
+ "choices": None,
+ "default": "4",
+ "required": False,
+ }
+ metadata = service._registry.get_registered_class_metadata("OpenAIVideoTarget")
+ assert metadata is not None
+ registry_n_seconds = next(parameter for parameter in metadata.parameters if parameter.name == "n_seconds")
+ assert "Literal['4', '8', '12']" in registry_n_seconds.type_name
+
async def test_types_cold_and_warm_results_are_equal(self) -> None:
service = TargetService()
@@ -361,13 +396,6 @@ async def test_types_refresh_after_runtime_class_registration(self) -> None:
False,
["text/plain", "text/html"],
),
- (
- "PlaywrightCopilotTarget",
- "copilot_type",
- "CopilotType",
- False,
- ["consumer", "m365"],
- ),
],
)
async def test_types_include_enum_parameters(
@@ -393,6 +421,32 @@ async def test_types_include_enum_parameters(
class TestCreateTarget:
"""Tests for TargetService.create_target method."""
+ async def test_create_target_rejects_parameters_that_take_python_objects(self, sqlite_instance) -> None:
+ service = TargetService()
+ request = CreateTargetRequest(name="text", type="TextTarget", params={"custom_configuration": {}})
+
+ with pytest.raises(ValueError, match="'custom_configuration' of 'TextTarget' cannot be set through the API"):
+ await service.create_target_async(request=request)
+
+ assert service.get_target_object(target_registry_name="text") is None
+
+ async def test_create_target_accepts_scalar_alternative_of_union(self, sqlite_instance) -> None:
+ service = TargetService()
+ request = CreateTargetRequest(
+ name="video",
+ type="OpenAIVideoTarget",
+ params={
+ "endpoint": "https://example.openai.azure.com/openai/v1",
+ "api_key": "test-key",
+ "model_name": "sora-2",
+ "n_seconds": 8,
+ },
+ )
+
+ await service.create_target_async(request=request)
+
+ assert service.get_target_object(target_registry_name="video")._n_seconds == "8"
+
async def test_create_target_raises_for_invalid_type(self) -> None:
"""Test that create_target raises for invalid target type."""
service = TargetService()
@@ -464,12 +518,16 @@ async def test_create_target_registers_nothing_when_response_mapping_fails(self,
assert service.get_target_object(target_registry_name="unmapped") is None
async def test_create_target_delegates_construction_to_registry(self, sqlite_instance) -> None:
- """Every target construction path is owned by the registry."""
+ """Every target construction path is owned by the registry, through its external-input path."""
service = TargetService()
- with patch.object(service._registry, "create_instance", wraps=service._registry.create_instance) as create:
+ with patch.object(
+ service._registry,
+ "create_instance_from_external_input",
+ wraps=service._registry.create_instance_from_external_input,
+ ) as create:
await service.create_target_async(request=CreateTargetRequest(type="TextTarget", params={}))
- create.assert_called_once()
+ create.assert_called_once_with("TextTarget", params={})
async def test_create_gandalf_target_coerces_level_string(self, sqlite_instance) -> None:
"""A Gandalf level from the JSON request is coerced to its enum before construction."""
diff --git a/tests/unit/converter/test_word_doc_converter.py b/tests/unit/converter/test_word_doc_converter.py
index 1ca2d6b891..971a4c1b8e 100644
--- a/tests/unit/converter/test_word_doc_converter.py
+++ b/tests/unit/converter/test_word_doc_converter.py
@@ -10,6 +10,7 @@
from pyrit.converter import ConverterResult, WordDocConverter
from pyrit.models import SeedPrompt
+from pyrit.registry.resolution import derive_parameters
@pytest.fixture
@@ -165,6 +166,12 @@ def test_constructor_custom_placeholder() -> None:
assert converter._injection_config.placeholder == "<>"
+def test_existing_docx_is_declared_as_path_parameter() -> None:
+ """The registry must see ``existing_docx`` as a ``Path`` so API callers upload the file."""
+ parameter = next(param for param in derive_parameters(cls=WordDocConverter) if param.name == "existing_docx")
+ assert parameter.is_path
+
+
def test_build_identifier_without_template() -> None:
"""_build_identifier should return correct params when no template is set."""
converter = WordDocConverter()
diff --git a/tests/unit/models/test_parameter.py b/tests/unit/models/test_parameter.py
index bcb0489315..4103d185b7 100644
--- a/tests/unit/models/test_parameter.py
+++ b/tests/unit/models/test_parameter.py
@@ -3,9 +3,10 @@
"""Unit tests for the unified Parameter model and its coercion methods."""
+from collections.abc import Callable, Collection, Sequence
from enum import Enum
from pathlib import Path
-from typing import Literal, Union
+from typing import Any, Literal, Union
import pytest
from pydantic import ValidationError
@@ -540,3 +541,145 @@ def __init__(self, *, value=None) -> None:
param = next(p for p in derive_parameters(cls=_Holder) if p.name == "value")
assert param.coerce_value(raw) == expected
+
+
+class TestIsExternalInput:
+ """``is_external_input`` marks the parameters REST, CLI, and GUI callers may supply."""
+
+ @pytest.mark.parametrize(
+ "param_type",
+ [
+ str,
+ int | None,
+ Path,
+ Path | str,
+ Literal["a", "b"],
+ _Speed,
+ list[str],
+ list[_Speed] | None,
+ Sequence[str],
+ Collection[str] | None,
+ Sequence[_Speed],
+ str | list[str],
+ str | Callable[[], str] | None,
+ _Speed | str,
+ int | tuple[int, int],
+ int | Literal["4", "8"],
+ str | dict[str, str],
+ str | _Unsupported,
+ Collection[str] | _Unsupported,
+ ],
+ )
+ def test_supported_external_types(self, param_type: object) -> None:
+ assert Parameter(name="p", description="d", param_type=param_type).is_external_input
+
+ @pytest.mark.parametrize(
+ "param_type",
+ [
+ None,
+ Any,
+ "SeedPrompt | None",
+ _Unsupported,
+ Callable[[], str],
+ tuple[int, int],
+ dict[str, str],
+ set[str],
+ list[Path],
+ list[Path | str],
+ Collection[Path],
+ Sequence[Path | str] | None,
+ _Speed | Path,
+ str | Path | int,
+ str | list[Path],
+ _Unsupported | Path,
+ list[list[str]],
+ Sequence[list[str]],
+ tuple[int, int] | _Unsupported,
+ ],
+ )
+ def test_other_types_take_python_objects_only(self, param_type: object) -> None:
+ assert not Parameter(name="p", description="d", param_type=param_type).is_external_input
+
+ def test_references_and_structured_inputs_are_external(self) -> None:
+ reference = Parameter(
+ name="t", description="d", reference=RegistryReference(component_type=ComponentType.TARGET)
+ )
+ structured = Parameter(name="s", description="d", param_type=_Unsupported, variants={"one": []})
+
+ assert reference.is_external_input
+ assert structured.is_external_input
+
+ def test_opaque_parameter_is_not_external(self) -> None:
+ assert not Parameter(name="o", description="d", param_type=str, opaque=True).is_external_input
+
+
+class TestForExternalCatalog:
+ """``for_external_catalog`` describes a parameter in the form external callers send it."""
+
+ @pytest.mark.parametrize(
+ ("param_type", "type_name", "is_list", "choices"),
+ [
+ (Collection[str], "list[str]", True, None),
+ (Sequence[int] | None, "list[int]", True, None),
+ (Collection[_Speed], "list[str]", True, ["fast", "slow"]),
+ (int | tuple[int, int], "int", False, None),
+ (int | Literal["4", "8", "12"], "int", False, None),
+ (str | list[str], "str", False, None),
+ (Sequence[str] | str, "list[str]", True, None),
+ (_Speed | str, "_Speed", False, ["fast", "slow"]),
+ (str | Callable[[], str] | None, "str", False, None),
+ ],
+ )
+ def test_describes_the_external_form(
+ self, param_type: object, type_name: str, is_list: bool, choices: list[str] | None
+ ) -> None:
+ parameter = Parameter(name="p", description="d", param_type=param_type, default=None)
+
+ described = parameter.for_external_catalog().model_dump(mode="json")
+
+ assert (described["type_name"], described["is_list"], described["choices"]) == (type_name, is_list, choices)
+ assert parameter.param_type == param_type
+
+ def test_flat_enum_collection_matches_the_enum_list_contract(self) -> None:
+ listed = Parameter(name="p", description="d", param_type=list[_Speed]).model_dump(mode="json")
+ collected = Parameter(name="p", description="d", param_type=Collection[_Speed]).for_external_catalog()
+
+ assert collected.model_dump(mode="json") == listed
+
+ @pytest.mark.parametrize("param_type", [str, int | None, Path | str, list[str], Literal["a", "b"], _Speed])
+ def test_external_forms_are_returned_unchanged(self, param_type: object) -> None:
+ parameter = Parameter(name="p", description="d", param_type=param_type)
+
+ assert parameter.for_external_catalog() is parameter
+
+ def test_references_and_structured_inputs_are_returned_unchanged(self) -> None:
+ reference = Parameter(
+ name="t",
+ description="d",
+ param_type=Collection[str],
+ reference=RegistryReference(component_type=ComponentType.TARGET),
+ )
+ structured = Parameter(name="s", description="d", param_type=_Unsupported, variants={"one": []})
+
+ assert reference.for_external_catalog() is reference
+ assert structured.for_external_catalog() is structured
+
+ def test_keeps_defaults_the_external_form_can_hold(self) -> None:
+ parameter = Parameter(name="font_size", description="d", param_type=int | tuple[int, int], default=15)
+
+ assert parameter.for_external_catalog().model_dump(mode="json")["default"] == "15"
+
+ def test_leaves_out_defaults_the_external_form_cannot_hold(self) -> None:
+ parameter = Parameter(name="font_size", description="d", param_type=int | tuple[int, int], default=(8, 20))
+
+ described = parameter.for_external_catalog()
+
+ assert described.model_dump(mode="json")["default"] is None
+ assert not described.required
+ assert parameter.default == (8, 20)
+
+ def test_coercion_keeps_every_alternative(self) -> None:
+ parameter = Parameter(name="replace", description="d", param_type=str | list[str], default=REQUIRED_VALUE)
+
+ assert parameter.for_external_catalog().required
+ assert parameter.coerce_value(["a", "b"]) == ["a", "b"]
diff --git a/tests/unit/registry/test_converter_registry.py b/tests/unit/registry/test_converter_registry.py
index 4253c16f7a..da92bce70e 100644
--- a/tests/unit/registry/test_converter_registry.py
+++ b/tests/unit/registry/test_converter_registry.py
@@ -206,6 +206,24 @@ def test_create_named_instance_stores_registry_metadata(self, registry: Converte
assert entry.instance is converter
assert entry.metadata == {"owned_artifact_paths": ["managed.dat"]}
+ def test_create_instance_from_external_input_rejects_object_parameters(self, registry: ConverterRegistry):
+ with pytest.raises(ValueError, match="'jailbreak_template' of 'TextJailbreakConverter' cannot be set"):
+ registry.create_instance_from_external_input(
+ "TextJailbreakConverter", params={"jailbreak_template": {"template": "x"}}
+ )
+
+ def test_create_named_instance_selects_external_input_explicitly(self, registry: ConverterRegistry):
+ with pytest.raises(ValueError, match="cannot be set through the API"):
+ registry.create_named_instance(
+ name="math", type_name="MathObfuscationConverter", params={"rng": None}, external_input=True
+ )
+ assert registry.instances.get("math") is None
+
+ converter = registry.create_named_instance(
+ name="caesar", type_name="CaesarConverter", params={"caesar_offset": "3"}, external_input=True
+ )
+ assert registry.instances.get("caesar") is converter
+
@pytest.mark.parametrize("name", ["preview", "types"])
def test_create_named_instance_rejects_reserved_name(self, registry: ConverterRegistry, name: str):
with pytest.raises(ValueError, match="reserved"):
diff --git a/tests/unit/registry/test_resolution.py b/tests/unit/registry/test_resolution.py
index 5cf32d4e85..2b079793bc 100644
--- a/tests/unit/registry/test_resolution.py
+++ b/tests/unit/registry/test_resolution.py
@@ -5,8 +5,9 @@
Tests for the shared registry constructor-argument resolution primitive.
"""
+from collections.abc import Callable
from enum import Enum
-from typing import Literal
+from typing import Any, Literal
import pytest
@@ -21,6 +22,7 @@
_registry_getter_for_component_type,
derive_parameters,
display_choices,
+ reject_non_external_params,
resolve_constructor_args,
)
@@ -260,6 +262,149 @@ def test_unknown_registry_reference_empty_registry_hint(self, empty_target_regis
_resolve(_NeedsTarget, {"converter_target": "missing"}, identifier_type=ConverterIdentifier)
+class _Handle:
+ """A Python object no external caller can supply."""
+
+
+class _Mixed:
+ """Helper whose constructor mixes external inputs with Python-object parameters."""
+
+ def __init__(
+ self,
+ *,
+ count: int = 1,
+ words: list[str] | None = None,
+ key: str | Callable[[], str] | None = None,
+ handle: _Handle | None = None,
+ options: dict[str, Any] | None = None,
+ ) -> None:
+ self.count = count
+ self.words = words
+ self.key = key
+ self.handle = handle
+ self.options = options
+
+
+@pytest.mark.usefixtures("patch_central_database")
+class TestExternalInput:
+ """The external path accepts only supported inputs; the in-process path is unchanged."""
+
+ def test_external_accepts_supported_inputs(self) -> None:
+ raw_args: dict[str, object] = {"count": "3", "words": ["a", "b"], "key": "secret"}
+
+ resolved = resolve_constructor_args(cls=_Mixed, raw_args=raw_args, external_input=True)
+
+ assert resolved == {"count": 3, "words": ["a", "b"], "key": "secret"}
+
+ @pytest.mark.parametrize("raw_args", [{"handle": {}}, {"handle": None}, {"options": {"a": 1}}])
+ def test_external_rejects_parameters_that_take_python_objects(self, raw_args: dict[str, object]) -> None:
+ with pytest.raises(ValueError, match="cannot be set through the API"):
+ resolve_constructor_args(cls=_Mixed, raw_args=raw_args, external_input=True)
+
+ def test_external_still_rejects_unknown_parameters(self) -> None:
+ with pytest.raises(ValueError, match="Unknown parameter 'nope'"):
+ resolve_constructor_args(cls=_Mixed, raw_args={"nope": 1}, external_input=True)
+
+ def test_in_process_callers_keep_passing_python_objects(self) -> None:
+ handle = _Handle()
+ options = {"dtype": object()}
+
+ resolved = _resolve(_Mixed, {"handle": handle, "options": options, "key": print})
+
+ assert resolved["handle"] is handle
+ assert resolved["options"] is options
+ assert resolved["key"] is print
+
+ def test_external_values_of_supported_parameters_reach_the_constructor(self) -> None:
+ assert resolve_constructor_args(cls=_Mixed, raw_args={"words": "a, b"}, external_input=True) == {
+ "words": "a, b"
+ }
+
+ @pytest.mark.parametrize("value", [{"name": "my_target"}, 5, True])
+ def test_external_reference_must_be_a_name(self, target_registry: TargetRegistry, value: object) -> None:
+ with pytest.raises(ValueError, match="expected a registry name"):
+ resolve_constructor_args(
+ cls=_NeedsTarget,
+ raw_args={"converter_target": value},
+ identifier_type=ConverterIdentifier,
+ external_input=True,
+ )
+
+ def test_external_list_reference_must_be_names(self, target_registry: TargetRegistry) -> None:
+ with pytest.raises(ValueError, match="expected a registry name"):
+ resolve_constructor_args(
+ cls=_NeedsTargets,
+ raw_args={"targets": ["my_target", {"name": "x"}]},
+ identifier_type=TargetIdentifier,
+ external_input=True,
+ )
+
+ resolved = resolve_constructor_args(
+ cls=_NeedsTargets,
+ raw_args={"targets": ["my_target"]},
+ identifier_type=TargetIdentifier,
+ external_input=True,
+ )
+ assert resolved["targets"] == [target_registry.instances.get("my_target")]
+
+ def test_external_reference_resolves_name(self, target_registry: TargetRegistry) -> None:
+ resolved = resolve_constructor_args(
+ cls=_NeedsTarget,
+ raw_args={"converter_target": "my_target"},
+ identifier_type=ConverterIdentifier,
+ external_input=True,
+ )
+
+ assert resolved["converter_target"] is target_registry.instances.get("my_target")
+
+ def test_reject_non_external_params_checks_names_and_references(self) -> None:
+ declared = derive_parameters(cls=_Mixed)
+ reference = derive_parameters(cls=_NeedsTarget, identifier_type=ConverterIdentifier)
+
+ reject_non_external_params(params={"count": 1, "key": "k"}, declared=declared, owner="demo")
+ with pytest.raises(ValueError, match="'handle' of 'demo' cannot be set through the API"):
+ reject_non_external_params(params={"handle": None}, declared=declared, owner="demo")
+ with pytest.raises(ValueError, match="Unknown parameter 'unknown' for 'demo'"):
+ reject_non_external_params(params={"unknown": None}, declared=declared, owner="demo")
+ with pytest.raises(ValueError, match="demo.converter_target: expected a registry name"):
+ reject_non_external_params(params={"converter_target": {"name": "x"}}, declared=reference, owner="demo")
+
+ @pytest.mark.parametrize(
+ ("registry_type", "identifier_type", "type_name", "raw_args"),
+ [
+ (TargetRegistry, TargetIdentifier, "TextTarget", {"custom_configuration": {}}),
+ (TargetRegistry, TargetIdentifier, "OpenAIChatTarget", {"httpx_client_kwargs": {"timeout": 5}}),
+ (ConverterRegistry, ConverterIdentifier, "TextJailbreakConverter", {"jailbreak_template": "x"}),
+ (ConverterRegistry, ConverterIdentifier, "PDFConverter", {"font_color": [1, 2, 3]}),
+ ],
+ )
+ def test_registered_components_reject_object_parameters_from_external_input(
+ self, registry_type: type, identifier_type: type, type_name: str, raw_args: dict[str, object]
+ ) -> None:
+ cls = registry_type.get_registry_singleton().get_class(type_name)
+
+ with pytest.raises(ValueError, match="cannot be set through the API"):
+ resolve_constructor_args(cls=cls, raw_args=raw_args, identifier_type=identifier_type, external_input=True)
+
+ @pytest.mark.parametrize(
+ ("registry_type", "identifier_type", "type_name", "raw_args"),
+ [
+ (TargetRegistry, TargetIdentifier, "OpenAIChatTarget", {"api_key": "key", "temperature": 0.5}),
+ (ConverterRegistry, ConverterIdentifier, "SearchReplaceConverter", {"pattern": "a", "replace": "b"}),
+ ],
+ )
+ def test_registered_components_accept_string_inputs_from_external_input(
+ self, registry_type: type, identifier_type: type, type_name: str, raw_args: dict[str, object]
+ ) -> None:
+ cls = registry_type.get_registry_singleton().get_class(type_name)
+
+ resolved = resolve_constructor_args(
+ cls=cls, raw_args=raw_args, identifier_type=identifier_type, external_input=True
+ )
+
+ assert resolved == raw_args
+
+
class TestDeriveParameters:
"""Tests for deriving the Parameter contract from a constructor signature."""