diff --git a/doc/code/converters/4_video_converters.ipynb b/doc/code/converters/4_video_converters.ipynb index b7430247e7..dbad4b353f 100644 --- a/doc/code/converters/4_video_converters.ipynb +++ b/doc/code/converters/4_video_converters.ipynb @@ -68,7 +68,7 @@ "\n", "await initialize_pyrit_async(memory_db_type=IN_MEMORY) # type: ignore\n", "\n", - "input_video = str(pathlib.Path(\".\") / \"..\" / \"..\" / \"..\" / \"assets\" / \"sample_video.mp4\")\n", + "input_video = pathlib.Path(\".\") / \"..\" / \"..\" / \"..\" / \"assets\" / \"sample_video.mp4\"\n", "input_image = str(pathlib.Path(\".\") / \"..\" / \"..\" / \"..\" / \"assets\" / \"pyrit_architecture.png\")\n", "\n", "video = AddImageVideoConverter(video_path=input_video)\n", diff --git a/doc/code/converters/4_video_converters.py b/doc/code/converters/4_video_converters.py index 2aff498ffe..73d2fbd82e 100644 --- a/doc/code/converters/4_video_converters.py +++ b/doc/code/converters/4_video_converters.py @@ -36,7 +36,7 @@ await initialize_pyrit_async(memory_db_type=IN_MEMORY) # type: ignore -input_video = str(pathlib.Path(".") / ".." / ".." / ".." / "assets" / "sample_video.mp4") +input_video = pathlib.Path(".") / ".." / ".." / ".." / "assets" / "sample_video.mp4" input_image = str(pathlib.Path(".") / ".." / ".." / ".." / "assets" / "pyrit_architecture.png") video = AddImageVideoConverter(video_path=input_video) diff --git a/pyrit/converter/add_image_to_video_converter.py b/pyrit/converter/add_image_to_video_converter.py index 376202d283..e45ab2ab3a 100644 --- a/pyrit/converter/add_image_to_video_converter.py +++ b/pyrit/converter/add_image_to_video_converter.py @@ -4,6 +4,7 @@ import asyncio import contextlib import logging +import tempfile from pathlib import Path import numpy as np @@ -38,8 +39,7 @@ class AddImageVideoConverter(Converter): def __init__( self, *, - video_path: str, - output_path: str | None = None, + video_path: Path, img_position: tuple[int, int] = (10, 10), img_resize_size: tuple[int, int] = (500, 500), ) -> None: @@ -47,18 +47,10 @@ def __init__( Initialize the converter with the video path and image properties. Args: - video_path (str): File path of video to add image to. - output_path (str, Optional): File path of output video. Defaults to None. + video_path (Path): File path of video to add image to. img_position (tuple): Position to place image in video. Defaults to (10, 10). img_resize_size (tuple): Size to resize image to. Defaults to (500, 500). - - Raises: - ValueError: If ``video_path`` is empty or invalid. """ - if not video_path: - raise ValueError("Please provide valid video path") - - self._output_path = output_path self._img_position = img_position self._img_resize_size = img_resize_size self._video_path = video_path @@ -78,16 +70,15 @@ def _build_identifier(self) -> ComponentIdentifier: } ) - async def _add_image_to_video_async(self, image_path: str, output_path: str) -> str: + async def _add_image_to_video_async(self, image_path: str) -> bytes: """ Add an image to video. Args: image_path (str): The image path to add to video. - output_path (str): The output video path. Returns: - str: The output video path. + bytes: The converted video data. Raises: ModuleNotFoundError: If OpenCV is not installed. @@ -106,121 +97,111 @@ async def _add_image_to_video_async(self, image_path: str, output_path: str) -> category="prompt-memory-entries", data_type="image_path", value=image_path ) input_video_data = data_serializer_factory( - category="prompt-memory-entries", data_type="video_path", value=self._video_path + category="prompt-memory-entries", data_type="video_path", value=str(self._video_path) ) # Open the video to ensure it exists video_bytes = await input_video_data.read_data_async() input_image_bytes = await input_image_data.read_data_async() - azure_storage_flag = input_video_data._is_azure_storage_url(self._video_path) - - await asyncio.to_thread( + return await asyncio.to_thread( self._add_image_to_video_sync, video_bytes=video_bytes, image_bytes=input_image_bytes, - output_path=output_path, - azure_storage_flag=azure_storage_flag, ) - logger.info(f"Video saved as {output_path}") - - return output_path - def _add_image_to_video_sync( self, *, video_bytes: bytes, image_bytes: bytes, - output_path: str, - azure_storage_flag: bool, - ) -> None: + ) -> bytes: """ Run the blocking cv2 pipeline and temp-file I/O on a worker thread. Designed to be invoked via ``asyncio.to_thread`` so the event loop is not blocked. + Returns: + bytes: The converted video data. + Raises: ValueError: If the input video format is unsupported or the overlay image cannot be decoded. """ import cv2 - video_path = self._video_path - local_temp_path: Path | None = None + file_extension = self._video_path.suffix.removeprefix(".").lower() + if file_extension not in video_encoding_map: + raise ValueError(f"Unsupported video format: {file_extension}") + + with tempfile.NamedTemporaryFile(suffix=f".{file_extension}", dir=DB_DATA_PATH, delete=False) as input_file: + input_file.write(video_bytes) + input_path = Path(input_file.name) + with tempfile.NamedTemporaryFile(suffix=f".{file_extension}", dir=DB_DATA_PATH, delete=False) as output_file: + output_path = Path(output_file.name) + cap: cv2.VideoCapture | None = None output_video: cv2.VideoWriter | None = None try: - if azure_storage_flag: - # If the video is in Azure storage, download it first - - # Save the video bytes to a temporary file - local_temp_path = Path(DB_DATA_PATH, "temp_video.mp4") - with open(local_temp_path, "wb") as f: - f.write(video_bytes) - video_path = str(local_temp_path) - - cap = cv2.VideoCapture(video_path) - - # Get video properties - fps = int(cap.get(cv2.CAP_PROP_FPS)) - width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) - height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) - file_extension = video_path.split(".")[-1].lower() - if file_extension in video_encoding_map: - video_char_code = cv2.VideoWriter.fourcc(*video_encoding_map[file_extension]) - output_video = cv2.VideoWriter(output_path, video_char_code, fps, (width, height)) - else: - raise ValueError(f"Unsupported video format: {file_extension}") - - # Load and resize the overlay image - - image_np_arr = np.frombuffer(image_bytes, np.uint8) - decoded = cv2.imdecode(image_np_arr, cv2.IMREAD_UNCHANGED) - if decoded is None: - raise ValueError("Failed to decode overlay image") - overlay = cv2.resize(decoded, self._img_resize_size) - - # Get overlay image dimensions - image_height, image_width, _ = overlay.shape - x, y = self._img_position # Position where the overlay will be placed - - while cap.isOpened(): - ret, frame = cap.read() - if not ret: - break - - # Ensure overlay fits within the frame boundaries - if x + image_width > width or y + image_height > height: - logger.info("Overlay image is too large for the video frame. Resizing to fit.") - overlay = cv2.resize(overlay, (width - x, height - y)) - image_height, image_width, _ = overlay.shape - - # Blend overlay with frame - if overlay.shape[2] == 4: # Check number of channels on image - alpha_overlay = overlay[:, :, 3] / 255.0 - for c in range(3): - frame[y : y + image_height, x : x + image_width, c] = ( - alpha_overlay * overlay[:, :, c] - + (1 - alpha_overlay) * frame[y : y + image_height, x : x + image_width, c] - ) - else: - frame[y : y + image_height, x : x + image_width] = overlay - - # Write the modified frame to the output video - output_video.write(frame) + try: + cap = cv2.VideoCapture(str(input_path)) + # Get video properties + fps = int(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + video_char_code = cv2.VideoWriter.fourcc(*video_encoding_map[file_extension]) + output_video = cv2.VideoWriter(str(output_path), video_char_code, fps, (width, height)) + + # Load and resize the overlay image + + image_np_arr = np.frombuffer(image_bytes, np.uint8) + decoded = cv2.imdecode(image_np_arr, cv2.IMREAD_UNCHANGED) + if decoded is None: + raise ValueError("Failed to decode overlay image") + overlay = cv2.resize(decoded, self._img_resize_size) + + # Get overlay image dimensions + image_height, image_width, _ = overlay.shape + x, y = self._img_position # Position where the overlay will be placed + + while cap.isOpened(): + ret, frame = cap.read() + if not ret: + break + + # Ensure overlay fits within the frame boundaries + if x + image_width > width or y + image_height > height: + logger.info("Overlay image is too large for the video frame. Resizing to fit.") + overlay = cv2.resize(overlay, (width - x, height - y)) + image_height, image_width, _ = overlay.shape + + # Blend overlay with frame + if overlay.shape[2] == 4: # Check number of channels on image + alpha_overlay = overlay[:, :, 3] / 255.0 + for c in range(3): + frame[y : y + image_height, x : x + image_width, c] = ( + alpha_overlay * overlay[:, :, c] + + (1 - alpha_overlay) * frame[y : y + image_height, x : x + image_width, c] + ) + else: + frame[y : y + image_height, x : x + image_width] = overlay + + # Write the modified frame to the output video + output_video.write(frame) + finally: + if cap is not None: + cap.release() + if output_video is not None: + output_video.release() + with contextlib.suppress(cv2.error): + cv2.destroyAllWindows() # Not available in headless OpenCV builds + + return output_path.read_bytes() finally: - # Release everything (guarded — early raises may leave cap/output_video unbound) - if cap is not None: - cap.release() - if output_video is not None: - output_video.release() - with contextlib.suppress(cv2.error): - cv2.destroyAllWindows() # Not available in headless OpenCV builds - if azure_storage_flag and local_temp_path is not None: - local_temp_path.unlink() + input_path.unlink(missing_ok=True) + output_path.unlink(missing_ok=True) async def convert_async(self, *, prompt: str, input_type: PromptDataType = "image_path") -> ConverterResult: """ @@ -239,15 +220,13 @@ async def convert_async(self, *, prompt: str, input_type: PromptDataType = "imag if not self.input_supported(input_type): raise ValueError("Input type not supported") - output_video_serializer = data_serializer_factory(category="prompt-memory-entries", data_type="video_path") - - if not self._output_path: - output_video_serializer.value = str(await output_video_serializer.get_data_filename_async()) - else: - output_video_serializer.value = self._output_path - - # Add video to the image - updated_video = await self._add_image_to_video_async( - image_path=prompt, output_path=output_video_serializer.value + output_video_serializer = data_serializer_factory( + category="prompt-memory-entries", + data_type="video_path", + extension=self._video_path.suffix.removeprefix("."), ) - return ConverterResult(output_text=str(updated_video), output_type="video_path") + updated_video = await self._add_image_to_video_async(image_path=prompt) + await output_video_serializer.save_data_async(data=updated_video) + logger.info(f"Video saved as {output_video_serializer.value}") + + return ConverterResult(output_text=str(output_video_serializer.value), output_type="video_path") diff --git a/pyrit/models/parameter.py b/pyrit/models/parameter.py index ae2da9cd6b..88791474a8 100644 --- a/pyrit/models/parameter.py +++ b/pyrit/models/parameter.py @@ -9,14 +9,15 @@ import types from dataclasses import dataclass from enum import Enum +from pathlib import Path from typing import Any, Literal, Union, get_args, get_origin from pydantic import BaseModel, ConfigDict, Field, computed_field, field_serializer, model_validator from pyrit.common.apply_defaults import REQUIRED_VALUE -_SUPPORTED_SCALAR_TYPES: tuple[type, ...] = (str, int, float, bool) -_SCALAR_NAME_TO_TYPE: dict[str, type] = {"int": int, "float": float, "bool": bool, "str": str} +_SUPPORTED_SCALAR_TYPES: tuple[type, ...] = (str, int, float, bool, Path) +_SCALAR_NAME_TO_TYPE: dict[str, type] = {"int": int, "float": float, "bool": bool, "str": str, "Path": Path} class ComponentType(str, Enum): @@ -191,7 +192,7 @@ def is_string_coercible(self) -> bool: Whether a single string token can be coerced to this parameter's value. True for a non-reference plain scalar (``str`` / ``int`` / ``float`` / - ``bool``), ``Literal[...]``, or ``Enum`` parameter — exactly the forms a + ``bool`` / ``Path``), ``Literal[...]``, or ``Enum`` parameter — exactly the forms a text field or CLI token can supply. References and structured types (lists and arbitrary objects) are False and are surfaced/handled elsewhere. @@ -261,7 +262,7 @@ def validate(self) -> None: # type: ignore[ty:invalid-method-override] Reject a declaration with an unsupported ``param_type``. Supported forms are a plain scalar, a constrained scalar - (``Literal``/``Enum``), a ``list`` of any of those, a registry reference, + (``Literal``/``Enum``), ``Path``, a ``list`` of any of those, a registry reference, an opaque passthrough, or ``None``. An otherwise-unsupported type is tolerated only when the parameter declares a default (the builder simply does not supply it, and the value passes through unchanged). @@ -284,7 +285,7 @@ def validate(self) -> None: # type: ignore[ty:invalid-method-override] raise ValueError( f"Parameter '{self.name}' has unsupported param_type {param_type!r}. " - f"Supported types: str, int, float, bool, Literal[...], Enum, a list of those, " + f"Supported types: str, int, float, bool, Path, Literal[...], Enum, a list of those, " f"or None (or provide a default)." ) @@ -314,7 +315,7 @@ def _is_scalar_param_type(annotation: Any) -> bool: """ Return True when ``annotation`` is a coercible scalar form. - A scalar form is a plain scalar (``str`` / ``int`` / ``float`` / ``bool``) or a + A scalar form is a plain scalar (``str`` / ``int`` / ``float`` / ``bool`` / ``Path``) or a constrained scalar (``Literal[...]`` or an ``Enum`` subclass) that carries its own allowed set. @@ -333,12 +334,12 @@ def _coerce_simple_value(*, param_name: str, annotation: Any, raw_value: Any) -> Coerce ``raw_value`` to a scalar ``annotation`` — the shared coercion core. Handles ``Optional[X]`` unwrap, ``Literal``/``Enum`` membership, and - int/float/bool/str. Anything else passes through unchanged. Both the + int/float/bool/str/Path. Anything else passes through unchanged. Both the ``Parameter`` path (``coerce_value``) and the resolver's annotation path route through this function so they cannot diverge on coerced values. Returns: - Any: The coerced value (a ``Literal``/``Enum`` member, an int/float/bool/str, or + Any: The coerced value (a ``Literal``/``Enum`` member, an int/float/bool/str/Path, or the raw value unchanged for unsupported annotations). Raises: @@ -358,6 +359,8 @@ def _coerce_simple_value(*, param_name: str, annotation: Any, raw_value: Any) -> return _coerce_scalar(param_name=param_name, scalar_type=float, raw_value=raw_value) if annotation is str: return str(raw_value) + if annotation is Path: + return Path(raw_value) return raw_value @@ -466,7 +469,7 @@ def _coerce_list(*, param_name: str, param_type: Any, raw_value: Any) -> list[An ] raise ValueError( f"Parameter '{param_name}' has unsupported list element type {element_type!r}. " - f"Supported list element types: str, int, float, bool, or Literal[...]." + f"Supported list element types: str, int, float, bool, Path, or Literal[...]." ) diff --git a/tests/unit/backend/test_converter_service.py b/tests/unit/backend/test_converter_service.py index a144b90967..cd98ae8360 100644 --- a/tests/unit/backend/test_converter_service.py +++ b/tests/unit/backend/test_converter_service.py @@ -161,6 +161,17 @@ async def test_catalog_serializes_parameter_type(self) -> None: caesar_param = next(p for p in caesar_entry.parameters if p.name == "caesar_offset") assert caesar_param.type_name == "int" + async def test_catalog_exposes_video_input_as_path_without_output_path(self) -> None: + """The video converter accepts an uploaded input but no caller-controlled destination.""" + service = ConverterService() + + result = await service.list_converter_catalog_async() + + video_entry = next(item for item in result.items if item.converter_type == "AddImageVideoConverter") + video_path_param = next(parameter for parameter in video_entry.parameters if parameter.name == "video_path") + assert video_path_param.type_name == "Path" + assert all(parameter.name != "output_path" for parameter in video_entry.parameters) + async def test_catalog_excludes_non_coercible_params(self) -> None: """Catalog only surfaces params that can be set from a string (e.g. not the LLM target).""" service = ConverterService() diff --git a/tests/unit/converter/test_add_image_video_converter.py b/tests/unit/converter/test_add_image_video_converter.py index 5159f1e102..7c6ca87673 100644 --- a/tests/unit/converter/test_add_image_video_converter.py +++ b/tests/unit/converter/test_add_image_video_converter.py @@ -1,13 +1,16 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import asyncio +from pathlib import Path + import numpy as np import pytest from pyrit.converter import AddImageVideoConverter -def is_opencv_installed(): +def is_opencv_installed() -> bool: try: import cv2 # noqa: F401 @@ -17,14 +20,14 @@ def is_opencv_installed(): @pytest.fixture(autouse=True) -def video_converter_sample_video(tmp_path, patch_central_database): - video_path = str(tmp_path / "test_video.mp4") +def video_converter_sample_video(tmp_path: Path, patch_central_database) -> Path: + video_path = tmp_path / "test_video.mp4" width, height = 640, 480 if is_opencv_installed(): import cv2 video_encoding = cv2.VideoWriter.fourcc(*"mp4v") - output_video = cv2.VideoWriter(video_path, video_encoding, 1, (width, height)) + output_video = cv2.VideoWriter(str(video_path), video_encoding, 1, (width, height)) for _i in range(10): frame = np.zeros((height, width, 3), dtype=np.uint8) output_video.write(frame) @@ -33,7 +36,7 @@ def video_converter_sample_video(tmp_path, patch_central_database): @pytest.fixture -def video_converter_sample_image(tmp_path): +def video_converter_sample_image(tmp_path: Path) -> str: image_path = str(tmp_path / "test_image.png") image = np.zeros((100, 100, 3), dtype=np.uint8) if is_opencv_installed(): @@ -44,75 +47,62 @@ def video_converter_sample_image(tmp_path): @pytest.mark.skipif(not is_opencv_installed(), reason="opencv is not installed") -def test_add_image_video_converter_initialization(tmp_path, video_converter_sample_video): - output_path = str(tmp_path / "output_video.mp4") +def test_add_image_video_converter_initialization(video_converter_sample_video: Path) -> None: converter = AddImageVideoConverter( video_path=video_converter_sample_video, - output_path=output_path, img_position=(10, 10), img_resize_size=(100, 100), ) assert converter._video_path == video_converter_sample_video - assert converter._output_path == output_path assert converter._img_position == (10, 10) assert converter._img_resize_size == (100, 100) @pytest.mark.skipif(not is_opencv_installed(), reason="opencv is not installed") -async def test_add_image_video_converter_invalid_image_path(tmp_path, video_converter_sample_video): - output_path = str(tmp_path / "output_video.mp4") - converter = AddImageVideoConverter(video_path=video_converter_sample_video, output_path=output_path) +async def test_add_image_video_converter_invalid_image_path(video_converter_sample_video: Path) -> None: + converter = AddImageVideoConverter(video_path=video_converter_sample_video) with pytest.raises(FileNotFoundError): - await converter._add_image_to_video_async(image_path="invalid_image.png", output_path=output_path) + await converter._add_image_to_video_async(image_path="invalid_image.png") @pytest.mark.skipif(not is_opencv_installed(), reason="opencv is not installed") -async def test_add_image_video_converter_invalid_video_path(tmp_path, video_converter_sample_image): - output_path = str(tmp_path / "output_video.mp4") - converter = AddImageVideoConverter(video_path="invalid_video.mp4", output_path=output_path) +async def test_add_image_video_converter_invalid_video_path(video_converter_sample_image: str) -> None: + converter = AddImageVideoConverter(video_path=Path("invalid_video.mp4")) with pytest.raises(FileNotFoundError): - await converter._add_image_to_video_async(image_path=video_converter_sample_image, output_path=output_path) + await converter._add_image_to_video_async(image_path=video_converter_sample_image) @pytest.mark.skipif(not is_opencv_installed(), reason="opencv is not installed") -async def test_add_image_video_converter(tmp_path, video_converter_sample_video, video_converter_sample_image): - output_path = str(tmp_path / "output_video.mp4") - converter = AddImageVideoConverter(video_path=video_converter_sample_video, output_path=output_path) - result_path = await converter._add_image_to_video_async( - image_path=video_converter_sample_image, output_path=output_path - ) - assert result_path == output_path +async def test_add_image_video_converter(video_converter_sample_video: Path, video_converter_sample_image: str) -> None: + converter = AddImageVideoConverter(video_path=video_converter_sample_video) + result = await converter._add_image_to_video_async(image_path=video_converter_sample_image) + assert result @pytest.mark.skipif(not is_opencv_installed(), reason="opencv is not installed") async def test_add_image_video_converter_convert_async( - tmp_path, video_converter_sample_video, video_converter_sample_image -): - output_path = str(tmp_path / "output_video.mp4") - converter = AddImageVideoConverter(video_path=video_converter_sample_video, output_path=output_path) + video_converter_sample_video: Path, video_converter_sample_image: str +) -> None: + converter = AddImageVideoConverter(video_path=video_converter_sample_video) converted_video = await converter.convert_async(prompt=video_converter_sample_image, input_type="image_path") assert converted_video - assert converted_video.output_text == output_path + assert Path(converted_video.output_text).is_file() assert converted_video.output_type == "video_path" @pytest.mark.skipif(not is_opencv_installed(), reason="opencv is not installed") -async def test_add_image_to_video_raises_when_decode_returns_none(tmp_path, video_converter_sample_video): +async def test_add_image_to_video_raises_when_decode_returns_none(video_converter_sample_video: Path) -> None: """Guard at line 146: cv2.imdecode returns None raises ValueError.""" from unittest.mock import AsyncMock, patch - output_path = str(tmp_path / "output_video.mp4") - converter = AddImageVideoConverter(video_path=video_converter_sample_video, output_path=output_path) + converter = AddImageVideoConverter(video_path=video_converter_sample_video) mock_image_serializer = AsyncMock() mock_image_serializer.read_data_async = AsyncMock(return_value=b"not_valid_image_data") - mock_image_serializer._is_azure_storage_url = lambda x: False mock_video_serializer = AsyncMock() - with open(video_converter_sample_video, "rb") as f: - video_bytes = f.read() + video_bytes = await asyncio.to_thread(video_converter_sample_video.read_bytes) mock_video_serializer.read_data_async = AsyncMock(return_value=video_bytes) - mock_video_serializer._is_azure_storage_url = lambda x: False def factory_side_effect(*, category, data_type, value): if data_type == "image_path": @@ -124,46 +114,27 @@ def factory_side_effect(*, category, data_type, value): side_effect=factory_side_effect, ): with pytest.raises(ValueError, match="Failed to decode overlay image"): - await converter._add_image_to_video_async(image_path="fake_image.png", output_path=output_path) + await converter._add_image_to_video_async(image_path="fake_image.png") @pytest.mark.skipif(not is_opencv_installed(), reason="opencv is not installed") -async def test_add_image_to_video_azure_storage_unlinks_local_temp( - tmp_path, video_converter_sample_video, video_converter_sample_image -): - """When the video is in Azure storage, the downloaded local temp file is unlinked after processing.""" - from unittest.mock import AsyncMock, patch +async def test_add_image_to_video_removes_temporary_files( + tmp_path: Path, video_converter_sample_video: Path, video_converter_sample_image: str +) -> None: + from unittest.mock import patch - output_path = str(tmp_path / "output_video.mp4") - converter = AddImageVideoConverter(video_path=video_converter_sample_video, output_path=output_path) + converter = AddImageVideoConverter(video_path=video_converter_sample_video) + files_before = set(tmp_path.iterdir()) - with open(video_converter_sample_video, "rb") as f: - video_bytes = f.read() - with open(video_converter_sample_image, "rb") as f: - image_bytes = f.read() + with patch("pyrit.converter.add_image_to_video_converter.DB_DATA_PATH", tmp_path): + await converter._add_image_to_video_async(image_path=video_converter_sample_image) - mock_image_serializer = AsyncMock() - mock_image_serializer.read_data_async = AsyncMock(return_value=image_bytes) - mock_image_serializer._is_azure_storage_url = lambda x: False - - mock_video_serializer = AsyncMock() - mock_video_serializer.read_data_async = AsyncMock(return_value=video_bytes) - # Flag this video as living in Azure storage so the cleanup branch runs. - mock_video_serializer._is_azure_storage_url = lambda x: True + assert set(tmp_path.iterdir()) == files_before - def factory_side_effect(*, category, data_type, value): - if data_type == "image_path": - return mock_image_serializer - return mock_video_serializer - - with ( - patch( - "pyrit.converter.add_image_to_video_converter.data_serializer_factory", - side_effect=factory_side_effect, - ), - patch("pyrit.converter.add_image_to_video_converter.DB_DATA_PATH", tmp_path), - ): - await converter._add_image_to_video_async(image_path=video_converter_sample_image, output_path=output_path) - # The local copy of the Azure-stored video should be removed by the cleanup branch. - assert not (tmp_path / "temp_video.mp4").exists() +def test_add_image_video_converter_rejects_output_path(video_converter_sample_video: Path, tmp_path: Path) -> None: + with pytest.raises(TypeError, match="output_path"): + AddImageVideoConverter( + video_path=video_converter_sample_video, + output_path=tmp_path / "output.mp4", # type: ignore[call-arg] + ) diff --git a/tests/unit/models/test_parameter.py b/tests/unit/models/test_parameter.py index 624a8d870e..77516ea375 100644 --- a/tests/unit/models/test_parameter.py +++ b/tests/unit/models/test_parameter.py @@ -4,6 +4,7 @@ """Unit tests for the unified Parameter model and its coercion methods.""" from enum import Enum +from pathlib import Path from typing import Literal import pytest @@ -134,11 +135,20 @@ def test_optional_scalar_unwraps_to_base_name(self) -> None: assert dumped["type_name"] == "int" + def test_path_round_trips_with_live_type(self) -> None: + dumped = Parameter(name="input_path", description="d", param_type=Path).model_dump() + + restored = Parameter.model_validate(dumped) + + assert dumped["type_name"] == "Path" + assert restored.param_type is Path + assert restored.coerce_value("input.txt") == Path("input.txt") + class TestIsScalarParamType: """``_is_scalar_param_type`` recognizes plain and constrained scalars.""" - @pytest.mark.parametrize("annotation", [str, int, float, bool, Literal["a", "b"], _Speed]) + @pytest.mark.parametrize("annotation", [str, int, float, bool, Path, Literal["a", "b"], _Speed]) def test_scalar_forms(self, annotation: object) -> None: assert _is_scalar_param_type(annotation) is True @@ -173,7 +183,7 @@ class TestIsStringCoercible: @pytest.mark.parametrize( "param_type", - [str, int, float, bool, Literal["a", "b"], _Speed, int | None, _Speed | None], + [str, int, float, bool, Path, Literal["a", "b"], _Speed, int | None, _Speed | None], ) def test_coercible_value_types(self, param_type: object) -> None: p = Parameter(name="x", description="d", param_type=param_type) @@ -228,6 +238,10 @@ def test_int(self) -> None: p = Parameter(name="n", description="d", param_type=int) assert p.coerce_value("5") == 5 + def test_path(self) -> None: + p = Parameter(name="input_path", description="d", param_type=Path) + assert p.coerce_value("input.txt") == Path("input.txt") + def test_float(self) -> None: p = Parameter(name="r", description="d", param_type=float) assert p.coerce_value("0.25") == 0.25