Type Validation Issues

# marker.py:244-248
if isinstance(input_data, BulkSceneMarkerUpdateInput):
    input_dict = input_data.to_graphql()
else:
    input_dict = input_data  # ❌ No validation of dict structure!

if isinstance(input_data, BulkSceneMarkerUpdateInput):
    input_dict = input_data.to_graphql()
else:
    # Type check first
    if not isinstance(input_data, dict):
        raise TypeError(
            f"input_data must be BulkSceneMarkerUpdateInput or dict, "
            f"got {type(input_data).__name__}"
        )
    # Validate dict structure through Pydantic
    validated = BulkSceneMarkerUpdateInput(**input_data)
    input_dict = validated.to_graphql()

# Convert path to Path object with error handling
try:
    path = Path(image_path)
    if not path.exists():
        raise FileNotFoundError(f"Image file not found: {path}")
    if not path.is_file():
        raise ValueError(f"Path is not a file: {path}")
except (OSError, TypeError) as e:
    raise ValueError(f"Invalid image path: {image_path!r}") from e

async def directory(
    self, path: str | Path | None = None, locale: str = "en"
) -> Directory:
    """Browse filesystem directory.

    Args:
        path: The directory path to list (string or Path object). If None, returns root directories.
              Note: Path is converted to string and sent to Stash server.
        locale: Desired collation locale (e.g., 'en-US', 'pt-BR'). Default is 'en'
    """
    # Convert Path to string if provided
    path_str = str(path) if path is not None else None
    variables = {"path": path_str, "locale": locale}
    result = await self.execute(fragments.DIRECTORY_QUERY, variables)
    return self._decode_result(Directory, result["directory"])

@model_validator(mode="after")
def _reject_path_modifications(self):
    """Reject attempts to modify server filesystem paths."""
    protected_paths = {
        "database_path", "backup_directory_path", "delete_trash_path",
        "generated_path", "metadata_path", "scrapers_path",
        "plugins_path", "cache_path", "blobs_path",
        "ffmpeg_path", "ffprobe_path", "python_path",
    }

    set_paths = [field for field in protected_paths if is_set(getattr(self, field))]

    if set_paths:
        raise StashConfigurationError(
            f"Modifying server filesystem paths is not allowed: {', '.join(set_paths)}. "
            "These critical paths should only be changed through the Stash web interface..."
        )

@staticmethod
def _normalize_conn_keys(conn: dict[str, Any]) -> dict[str, Any]:
    """Normalize connection dict keys to canonical case."""
    if not conn:
        return {}

    # Use CIMultiDict for case-insensitive lookup during normalization
    ci_conn = CIMultiDict(conn)

    # Canonical key mappings (lowercase -> canonical)
    canonical_keys = {
        "logger": "Logger",
        "scheme": "Scheme",
        "host": "Host",
        "port": "Port",
        "apikey": "ApiKey",
    }

    normalized = {}
    for lower_key, canonical_key in canonical_keys.items():
        if value := ci_conn.get(lower_key):
            normalized[canonical_key] = value

    return normalized

def destroy_tag(self, id: str) -> bool:
    result = self._graphql_call(...)
    return result.get("tagDestroy") is True  # ✅ Explicit identity check

def _parse_obj_for_ID(self, param: Any, str_key: str = "name") -> Any:
    if isinstance(param, str):
        try:
            return int(param)  # ❌ No validation: negative IDs, zero OK?
        except ValueError:
            return {str_key: param.strip()}
    elif isinstance(param, dict):
        if param.get("stored_id"):
            return int(param["stored_id"])  # ❌ Could be None or non-numeric
        if param.get("id"):
            return int(param["id"])  # ❌ Same issue
    return param

def _parse_obj_for_ID(self, param: Any, str_key: str = "name") -> Any:
    if isinstance(param, str):
        stripped = param.strip()
        try:
            id_val = int(stripped)
        except ValueError:
            return {str_key: stripped}
        if id_val <= 0:  # ✅ ADDED: Positive validation
            raise ValueError("ID must be positive")
        return id_val
    if isinstance(param, dict):
        for key in ("stored_id", "id"):
            if key in param:
                try:
                    id_val = int(param[key])
                except (TypeError, ValueError) as e:  # ✅ ADDED: Error handling
                    raise ValueError(f"Invalid {key}: {param[key]!r}") from e
                if id_val <= 0:  # ✅ ADDED: Positive validation
                    raise ValueError(f"{key} must be positive")
                return id_val
    return param

# Validate Scheme
scheme = conn.get("Scheme", "http")
if scheme not in ("http", "https"):
    raise ValueError(f"Scheme must be 'http' or 'https', got {scheme!r}")
self.scheme = scheme

async def find_markers(
    self,
    filter_: dict[str, Any] | None = None,
    marker_filter: dict[str, Any] | None = None,
    ...
) -> FindSceneMarkersResultType:
    # marker_filter is passed to GraphQL, which handles None correctly
    result = await self.execute(
        fragments.FIND_MARKERS_QUERY,
        {"filter": filter_, "marker_filter": marker_filter},  # ✅ None is OK
    )

with open(path, "rb") as f:
    image_data = f.read()
image_b64 = base64.b64encode(image_data).decode("utf-8")
mime = mimetypes.types_map.get(path.suffix, "image/jpeg")  # ❌ Default might be wrong

mime = mimetypes.types_map.get(path.suffix.lower())
if not mime or not mime.startswith("image/"):
    raise ValueError(f"File does not appear to be an image: {path.suffix}")

def _normalize_sort_direction(self, filter_: dict[str, Any]) -> dict[str, Any]:
    if "direction" not in filter_:
        return filter_

    direction = filter_.get("direction")
    if direction is None:
        return filter_

    if isinstance(direction, SortDirectionEnum):  # ✅ Handles enum
        updated = dict(filter_)
        updated["direction"] = direction.value
        return updated

    if isinstance(direction, str):
        if direction in ("ASC", "DESC"):  # ✅ Validates strings
            return filter_
        raise ValueError(f"direction must be 'ASC' or 'DESC', got {direction!r}")

if status.status == SystemStatusEnum.NEEDS_MIGRATION:  # ❌ Assumes enum, might be string

async def scene_add_o(
    self,
    id: str,
    times: list[str] | None = None,  # ❌ No validation of string format
    ...

async def scene_add_o(
    self,
    id: str,
    times: list[Timestamp] | None = None,  # ✅ Uses validated Timestamp type
) -> HistoryMutationResult:

Timestamp = Annotated[
    datetime,
    BeforeValidator(_parse_timestamp_value),
    PlainSerializer(_serialize_timestamp, return_type=str),
]

async def gallery_chapter_create(
    self,
    gallery_id: str,
    title: str,
    image_index: int,  # ❌ No validation: negative? zero? too large?
    ...
) -> GalleryChapter:
    if image_index is not None:
        input_data["image_index"] = str(image_index)

if image_index < 0:  # ✅ ADDED: Range validation
    raise ValueError(f"image_index must be non-negative, got {image_index}")

# Validate and convert verify_ssl to bool
if isinstance(verify_ssl, str):
    verify_ssl = verify_ssl.lower() in ("true", "1", "yes")
elif not isinstance(verify_ssl, bool):
    raise TypeError(
        f"verify_ssl must be bool or string ('true'/'false'), got {type(verify_ssl).__name__}"
    )

def test_batch_size_positive():
    """filter_and_populate should accept positive batch_size"""
    # Should work

def test_batch_size_zero_raises_valueerror():
    """filter_and_populate should reject batch_size=0"""
    # Should raise ValueError

def test_batch_size_negative_raises_valueerror():
    """filter_and_populate should reject negative batch_size"""
    # Should raise ValueError