ShadowMountPlus/tools/api_test.py

#!/usr/bin/env python3
"""Validate a running ShadowMount HTTP/JSON API endpoint."""

import argparse
import concurrent.futures
import errno
import http.client
import json
import socket
import struct
import sys
from typing import Any, Dict, Optional, Tuple


API_VERSION = 1
MAX_RESPONSE_SIZE = 16 * 1024 * 1024

ROUTE_INDEX = "/"
ROUTE_VERSION = "/api/v1/version"
ROUTE_STORAGE = "/api/v1/storage"
ROUTE_IMAGES = "/api/v1/images"
ROUTE_SCAN = "/api/v1/scan"
ROUTE_MANUAL_LIST = "/api/v1/manual/list"
ROUTE_MANUAL_ADD = "/api/v1/manual/add"
ROUTE_MANUAL_REMOVE = "/api/v1/manual/remove"
ROUTE_GAMES = "/api/v1/games"
ROUTE_GAME_INFO = "/api/v1/games/info"
ROUTE_MOUNT = "/api/v1/games/mount"
ROUTE_UNMOUNT = "/api/v1/games/unmount"
ROUTE_UNINSTALL = "/api/v1/games/uninstall"
ROUTE_MOVE = "/api/v1/games/move"
ROUTE_COPY = "/api/v1/games/copy"
ROUTE_DELETE = "/api/v1/games/delete"
ROUTE_UNPACK = "/api/v1/games/unpack"
ROUTE_STORAGE_JOB_STATUS = "/api/v1/games/storage/status"
ROUTE_STORAGE_JOB_CANCEL = "/api/v1/games/storage/cancel"
ROUTE_SETTINGS = "/api/v1/settings"
ROUTE_SETTINGS_UPDATE = "/api/v1/settings/update"
ROUTE_DEBUG_LOG = "/api/v1/debug-log"
ROUTE_KERNEL_LOG = "/api/v1/kernel-log"


class ApiTestError(RuntimeError):
    pass


class ApiClient:
    def __init__(self, address: str, port: int, timeout: float) -> None:
        self.address = address
        self.port = port
        self.timeout = timeout

    def request(
        self,
        method: str,
        route: str,
        body: bytes = b"",
        headers: Optional[Dict[str, str]] = None,
    ) -> Tuple[int, Dict[str, str], bytes]:
        connection = http.client.HTTPConnection(
            self.address, self.port, timeout=self.timeout
        )
        try:
            connection.request(method, route, body=body, headers=headers or {})
            response = connection.getresponse()
            response_body = response.read(MAX_RESPONSE_SIZE + 1)
            response_headers = {
                name.lower(): value.strip() for name, value in response.getheaders()
            }
            response_status = response.status
        finally:
            connection.close()

        if len(response_body) > MAX_RESPONSE_SIZE:
            raise ApiTestError("response body exceeds 16 MiB")
        if response_headers.get("access-control-allow-origin") != "*":
            raise ApiTestError(f"{method} {route}: missing CORS allow-origin header")

        content_length = response_headers.get("content-length")
        if response_status == 204:
            if response_body:
                raise ApiTestError(f"{method} {route}: HTTP 204 response has a body")
            return response_status, response_headers, response_body
        if content_length is None:
            raise ApiTestError(f"{method} {route}: missing Content-Length header")
        try:
            declared_size = int(content_length, 10)
        except ValueError as error:
            raise ApiTestError(
                f"{method} {route}: invalid Content-Length header"
            ) from error
        if declared_size != len(response_body):
            raise ApiTestError(
                f"{method} {route}: Content-Length is {declared_size}, "
                f"received {len(response_body)} bytes"
            )
        return response_status, response_headers, response_body

    def post(self, route: str, payload: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]:
        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        status, headers, response_body = self.request(
            "POST",
            route,
            body,
            {
                "Content-Type": "application/json",
                "Content-Length": str(len(body)),
                "Connection": "close",
            },
        )
        if not headers.get("content-type", "").lower().startswith(
            "application/json"
        ):
            raise ApiTestError(f"POST {route}: response is not JSON")
        try:
            response = json.loads(response_body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as error:
            raise ApiTestError(f"POST {route}: invalid JSON response") from error
        if not isinstance(response, dict):
            raise ApiTestError(f"POST {route}: response must be a JSON object")
        return status, response


def require(condition: bool, message: str) -> None:
    if not condition:
        raise ApiTestError(message)


def integer_field(response: Dict[str, Any], key: str) -> int:
    value = response.get(key)
    if type(value) is not int:
        raise ApiTestError(f"response field {key!r} must be an integer")
    return value


def test_preflight(client: ApiClient) -> None:
    status, headers, body = client.request(
        "OPTIONS",
        ROUTE_VERSION,
        headers={
            "Origin": "http://example.test",
            "Access-Control-Request-Method": "POST",
            "Access-Control-Request-Headers": "Content-Type",
            "Access-Control-Request-Private-Network": "true",
            "Connection": "close",
        },
    )
    require(status == 204, f"CORS preflight returned HTTP {status}")
    require(not body, "CORS preflight returned a response body")
    require(
        headers.get("access-control-allow-methods") == "GET, POST, OPTIONS",
        "CORS preflight returned invalid allowed methods",
    )
    require(
        headers.get("access-control-allow-headers") == "Content-Type",
        "CORS preflight returned invalid allowed headers",
    )
    require(
        headers.get("access-control-allow-private-network") == "true",
        "CORS private-network access is not enabled",
    )
    print("[API-TEST] CORS preflight=PASS")


def test_index_page(client: ApiClient) -> str:
    status, headers, body = client.request("GET", ROUTE_INDEX)
    content_type = headers.get("content-type", "").lower()
    if status == 404:
        require(
            content_type.startswith("application/json"),
            "missing web UI response is not JSON",
        )
        try:
            error = json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exception:
            raise ApiTestError("missing web UI returned invalid JSON") from exception
        require(
            integer_field(error, "status") == errno.ENOENT,
            "missing web UI did not return ENOENT",
        )
        print("[API-TEST] web UI=SKIP (index.html is not installed)")
        return "SKIP"
    require(status == 200, f"web UI returned HTTP {status}")
    require(content_type.startswith("text/html"), "web UI response is not HTML")
    require(bool(body.strip()), "web UI response is empty")
    print(f"[API-TEST] web UI=PASS ({len(body)} bytes)")
    return "PASS"


def test_version(client: ApiClient) -> None:
    http_status, response = client.post(ROUTE_VERSION, {})
    require(http_status == 200, f"version request returned HTTP {http_status}")
    require(integer_field(response, "status") == 0, "version request failed")
    require(
        integer_field(response, "api_version") == API_VERSION,
        "unsupported API version",
    )
    version = response.get("shadowmount_version")
    require(isinstance(version, str), "missing ShadowMount version")
    capabilities = response.get("capabilities")
    require(isinstance(capabilities, list), "missing API capabilities")
    required_capabilities = {
        "web_ui",
        "storage_space",
        "rescan",
        "uninstall_game",
        "game_info",
        "game_icon",
        "move_game_source",
        "copy_game_source",
        "delete_game_source",
        "unpack_game_image",
        "storage_job_status",
        "storage_job_cancel",
        "list_manual_sources",
        "add_manual_source",
        "remove_manual_source",
        "manage_settings",
        "read_debug_log",
        "read_kernel_log",
    }
    require(
        required_capabilities.issubset(capabilities),
        "API does not advertise all mutation capabilities",
    )
    print(f"[API-TEST] ShadowMount={version} API={API_VERSION}")


def test_storage(client: ApiClient) -> int:
    http_status, response = client.post(ROUTE_STORAGE, {})
    require(http_status == 200, f"storage request returned HTTP {http_status}")
    require(integer_field(response, "status") == 0, "storage request failed")
    count = integer_field(response, "count")
    mounts = response.get("mounts")
    require(count >= 0, "storage mount count is negative")
    require(isinstance(mounts, list), "storage mounts field must be an array")
    require(len(mounts) == count, "storage mount count does not match array length")

    required_fields = {
        "source",
        "mount_point",
        "filesystem",
        "total_bytes",
        "free_bytes",
        "available_bytes",
        "used_bytes",
        "read_only",
    }
    for mount in mounts:
        require(isinstance(mount, dict), "storage mounts contains a non-object item")
        require(required_fields.issubset(mount), "storage mount lacks fields")
        require(
            isinstance(mount["source"], str)
            and isinstance(mount["mount_point"], str)
            and mount["mount_point"].startswith("/")
            and isinstance(mount["filesystem"], str),
            "storage mount contains invalid identity fields",
        )
        total = integer_field(mount, "total_bytes")
        free = integer_field(mount, "free_bytes")
        available = integer_field(mount, "available_bytes")
        used = integer_field(mount, "used_bytes")
        require(
            0 <= available <= free <= total and used == total - available,
            "storage mount byte counters are inconsistent",
        )
        require(type(mount["read_only"]) is bool, "storage read_only must be boolean")
    print(f"[API-TEST] storage mounts={count}")
    return count


def validate_storage_job(response: Dict[str, Any]) -> None:
    required_fields = {
        "job_id",
        "operation",
        "state",
        "active",
        "cancellable",
        "cancel_requested",
        "title_id",
        "source_type",
        "source",
        "runtime_source",
        "destination",
        "delete_source",
        "total_bytes",
        "processed_bytes",
        "total_files",
        "processed_files",
        "progress_percent",
        "speed_bytes_per_second",
        "elapsed_ms",
        "affected_titles",
        "result_status",
        "result_error",
        "scan_queued",
    }
    require(required_fields.issubset(response), "storage job response lacks fields")
    job_id = integer_field(response, "job_id")
    require(job_id >= 0, "storage job ID is invalid")
    operation = response["operation"]
    valid_operation = operation == "" if job_id == 0 else operation in {
        "copy",
        "move",
        "delete",
        "unpack",
    }
    require(
        valid_operation,
        "storage job operation is invalid",
    )
    if operation == "delete":
        require(response["destination"] == "", "delete job has a destination")
    require(
        response["state"]
        in {
            "idle",
            "preparing",
            "measuring",
            "transferring",
            "deleting",
            "finalizing",
            "completed",
            "failed",
            "cancelled",
        },
        "storage job state is invalid",
    )
    for key in (
        "active",
        "cancellable",
        "cancel_requested",
        "delete_source",
        "scan_queued",
    ):
        require(type(response[key]) is bool, f"storage job {key} must be boolean")
    for key in (
        "total_bytes",
        "processed_bytes",
        "total_files",
        "processed_files",
        "speed_bytes_per_second",
        "elapsed_ms",
        "affected_titles",
        "result_status",
    ):
        require(integer_field(response, key) >= 0, f"storage job {key} is invalid")
    progress = response["progress_percent"]
    require(
        type(progress) in (int, float) and 0 <= progress <= 100,
        "storage job progress_percent is invalid",
    )


def test_storage_job_status(client: ApiClient) -> None:
    http_status, response = client.post(ROUTE_STORAGE_JOB_STATUS, {})
    require(http_status == 200, f"storage job status returned HTTP {http_status}")
    require(integer_field(response, "status") == 0, "storage job status failed")
    validate_storage_job(response)
    print(
        f"[API-TEST] storage job={response['state']} "
        f"(id={response['job_id']}, active={response['active']})"
    )


def test_settings_and_log(client: ApiClient) -> None:
    http_status, settings = client.post(ROUTE_SETTINGS, {})
    require(http_status == 200, f"settings returned HTTP {http_status}")
    require(integer_field(settings, "status") == 0, "settings request failed")
    for key in (
        "api_enabled",
        "allow_lan_access",
        "debug",
        "quiet_mode",
        "update_emulators",
        "auto_update_ampr",
        "auto_remove_missing_games",
    ):
        require(type(settings.get(key)) is bool, f"settings {key} must be boolean")
    require(settings["api_enabled"], "API reports itself disabled")
    fan_target = integer_field(settings, "fan_target_temperature")
    require(fan_target == 0 or 50 <= fan_target <= 91, "fan target is invalid")
    remove_delay = integer_field(settings, "auto_remove_missing_delay_seconds")
    require(1 <= remove_delay <= 86400, "auto-remove delay is invalid")
    scan_paths = settings.get("scan_paths")
    require(isinstance(scan_paths, list), "scan_paths must be an array")
    require(
        len(scan_paths) == integer_field(settings, "scan_path_count"),
        "scan path count does not match",
    )
    require(
        all(isinstance(path, str) and path.startswith("/") for path in scan_paths),
        "scan_paths contains an invalid path",
    )

    for route, name, total_key in (
        (ROUTE_DEBUG_LOG, "debug log", "file_size"),
        (ROUTE_KERNEL_LOG, "kernel log", "total_bytes"),
    ):
        http_status, log = client.post(route, {"max_bytes": 4096})
        require(http_status == 200, f"{name} returned HTTP {http_status}")
        require(integer_field(log, "status") == 0, f"{name} request failed")
        require(isinstance(log.get("content"), str), f"{name} content must be text")
        require(type(log.get("truncated")) is bool, f"{name} truncated must be boolean")
        require(integer_field(log, total_key) >= 0, f"{name} total size is invalid")
        require(
            0 <= integer_field(log, "returned_bytes") <= 4096,
            f"{name} response is too large",
        )
    print(f"[API-TEST] settings={len(scan_paths)} scan paths, logs=PASS")


def test_list(client: ApiClient, route: str, key: str) -> list:
    http_status, response = client.post(route, {})
    require(http_status == 200, f"{key} request returned HTTP {http_status}")
    require(integer_field(response, "status") == 0, f"{key} request failed")
    count = integer_field(response, "count")
    items = response.get(key)
    require(count >= 0, f"{key} count is negative")
    require(isinstance(items, list), f"{key} field must be an array")
    require(len(items) == count, f"{key} count does not match array length")
    if key == "games":
        require(response.get("size_included") is False, "game size was enabled")
    require(
        all(isinstance(item, dict) for item in items),
        f"{key} array contains a non-object item",
    )
    print(f"[API-TEST] {key}={count}")
    return items


def test_game_details(client: ApiClient, games: list) -> None:
    if not games:
        print("[API-TEST] game details=SKIP (no games)")
        return

    required_fields = {
        "path",
        "runtime_path",
        "source_type",
        "image_type",
        "platform",
        "title_id",
        "content_id",
        "title_name",
        "last_access_time",
        "install_time",
        "icon_url",
        "app_db_size_bytes",
    }
    for game in games:
        require(required_fields.issubset(game), "game list item lacks metadata")
        require("size_bytes" not in game, "game list calculated size by default")

    thumbnail_url = next(
        (
            game.get("icon_url")
            for game in games
            if isinstance(game.get("icon_url"), str) and game.get("icon_url")
        ),
        "",
    )
    if thumbnail_url:
        require("size=thumb" in thumbnail_url, "game list icon is not a thumbnail")
        first_status, headers, first_thumbnail = client.request("GET", thumbnail_url)
        second_status, _, second_thumbnail = client.request("GET", thumbnail_url)
        require(first_status == 200, f"game thumbnail returned HTTP {first_status}")
        require(second_status == 200, "cached game thumbnail request failed")
        require(
            headers.get("content-type", "").lower().startswith("image/png"),
            "game thumbnail response is not PNG",
        )
        require(
            "max-age=" in headers.get("cache-control", "").lower(),
            "game thumbnail response is not cacheable",
        )
        require(
            len(first_thumbnail) >= 24 and first_thumbnail[:8] == b"\x89PNG\r\n\x1a\n",
            "game thumbnail has an invalid PNG header",
        )
        require(
            struct.unpack(">II", first_thumbnail[16:24]) == (128, 128),
            "game thumbnail is not 128x128",
        )
        require(first_thumbnail == second_thumbnail, "cached thumbnail changed")

    title_id = games[0].get("title_id")
    require(isinstance(title_id, str) and title_id, "game has invalid title_id")
    http_status, detail = client.post(ROUTE_GAME_INFO, {"title_id": title_id})
    require(http_status == 200, f"game info returned HTTP {http_status}")
    require(integer_field(detail, "status") == 0, "game info failed")
    size_status = integer_field(detail, "size_status")
    require(size_status >= 0, "game size status is invalid")
    if size_status == 0:
        require(integer_field(detail, "size_bytes") >= 0, "game size is invalid")

    icon_url = detail.get("icon_url")
    if isinstance(icon_url, str) and icon_url:
        icon_status, headers, icon = client.request("GET", icon_url)
        require(icon_status == 200, f"game icon returned HTTP {icon_status}")
        require(
            headers.get("content-type", "").lower().startswith("image/png"),
            "game icon response is not PNG",
        )
        require(bool(icon), "game icon is empty")
    print(f"[API-TEST] game details=PASS ({title_id}, size_status={size_status})")


def find_mount_candidate(games: list, mode: str) -> str:
    for game in games:
        title_id = game.get("title_id")
        source_path = game.get("path")
        if (
            game.get("managed") is True
            and game.get("source_available") is True
            and game.get("mounted") is False
            and (not mode or game.get("image_backed") is True)
            and (
                mode != "rw"
                or not isinstance(source_path, str)
                or not source_path.lower().endswith(".ffpfsc")
            )
            and isinstance(title_id, str)
            and title_id
        ):
            return title_id
    return ""


def title_operation(
    client: ApiClient, route: str, title_id: str, mode: str = ""
) -> Tuple[int, int]:
    payload = {"title_id": title_id}
    if mode:
        payload["mode"] = mode
    http_status, response = client.post(route, payload)
    return http_status, integer_field(response, "status")


def test_mount_cycle(client: ApiClient, title_id: str, mode: str) -> str:
    if not title_id:
        print("[API-TEST] mount cycle=SKIP (no candidate)")
        return "SKIP"

    http_status, status = title_operation(client, ROUTE_MOUNT, title_id, mode)
    if http_status == 409 and status == errno.EBUSY:
        print(f"[API-TEST] mount cycle=SKIP ({title_id} is busy)")
        return "SKIP"
    if http_status == 501 and status == errno.ENOTSUP:
        print(f"[API-TEST] mount cycle=SKIP ({title_id} rejects mode={mode})")
        return "SKIP"
    require(
        http_status == 200 and status == 0,
        f"mount {title_id} failed: HTTP {http_status}, status {status}",
    )

    http_status, status = title_operation(client, ROUTE_UNMOUNT, title_id)
    require(
        http_status == 200 and status == 0,
        f"unmount {title_id} failed: HTTP {http_status}, status {status}",
    )
    mode_text = mode or "default"
    print(f"[API-TEST] mount cycle=PASS ({title_id}, mode={mode_text})")
    return "PASS"


def test_rescan(client: ApiClient) -> None:
    http_status, response = client.post(ROUTE_SCAN, {})
    require(http_status == 200, f"rescan request returned HTTP {http_status}")
    require(integer_field(response, "status") == 0, "rescan request failed")
    require(response.get("queued") is True, "rescan was not queued")
    require(
        response.get("reset_attempts") is False,
        "rescan reset attempts without an explicit option",
    )
    print("[API-TEST] full rescan=QUEUED")


def test_manual_list(client: ApiClient) -> int:
    http_status, response = client.post(ROUTE_MANUAL_LIST, {})
    require(http_status == 200, f"manual list returned HTTP {http_status}")
    require(integer_field(response, "status") == 0, "manual list failed")
    count = integer_field(response, "count")
    paths = response.get("paths")
    require(count >= 0, "manual path count is negative")
    require(isinstance(paths, list), "manual paths field must be an array")
    require(len(paths) == count, "manual path count does not match array length")
    require(
        all(isinstance(path, str) and path.startswith("/") for path in paths),
        "manual paths contains an invalid path",
    )
    print(f"[API-TEST] manual paths={count}")
    return count


def test_parallel_requests(client: ApiClient) -> None:
    stalled = socket.create_connection(
        (client.address, client.port), timeout=client.timeout
    )
    try:
        stalled.sendall(
            (
                f"POST {ROUTE_VERSION} HTTP/1.1\r\n"
                f"Host: {client.address}:{client.port}\r\n"
                "Content-Type: application/json\r\n"
                "Content-Length: 2\r\n"
                "Connection: close\r\n\r\n"
                "{"
            ).encode("ascii")
        )

        def get_version(_: int) -> None:
            http_status, response = client.post(ROUTE_VERSION, {})
            require(
                http_status == 200 and integer_field(response, "status") == 0,
                f"parallel version request returned HTTP {http_status}",
            )

        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
            list(executor.map(get_version, range(16)))
    finally:
        stalled.close()
    print("[API-TEST] parallel requests=PASS (16 requests + stalled client)")


def test_mutation_validation(client: ApiClient) -> None:
    invalid_requests = (
        (ROUTE_SCAN, {"reset_attempts": "yes"}),
        (ROUTE_MANUAL_ADD, {"path": "relative/path"}),
        (ROUTE_MANUAL_REMOVE, {"path": "/"}),
        (ROUTE_UNINSTALL, {"title_id": "INVALID"}),
        (ROUTE_MOVE, {"title_id": "INVALID", "destination_dir": "/mnt"}),
        (ROUTE_COPY, {"title_id": "INVALID", "destination_dir": "/mnt"}),
        (ROUTE_UNPACK, {"title_id": "INVALID", "destination_dir": "/mnt"}),
        (ROUTE_DELETE, {"title_id": "INVALID", "confirm": True}),
        (ROUTE_DELETE, {"title_id": "PPSA00000"}),
        (ROUTE_STORAGE_JOB_STATUS, {"job_id": "invalid"}),
        (ROUTE_STORAGE_JOB_CANCEL, {"job_id": 0}),
        (ROUTE_SETTINGS_UPDATE, {}),
        (
            ROUTE_SETTINGS_UPDATE,
            {
                "debug": True,
                "quiet_mode": False,
                "update_emulators": False,
                "auto_remove_missing_delay_seconds": 0,
                "allow_lan_access": False,
                "fan_target_temperature": 0,
                "scan_paths": [],
            },
        ),
        (
            ROUTE_SETTINGS_UPDATE,
            {
                "debug": True,
                "quiet_mode": False,
                "update_emulators": False,
                "allow_lan_access": False,
                "fan_target_temperature": 0,
                "scan_paths": ["/"],
            },
        ),
        (
            ROUTE_SETTINGS_UPDATE,
            {
                "debug": True,
                "quiet_mode": False,
                "update_emulators": False,
                "allow_lan_access": False,
                "fan_target_temperature": 0,
                "scan_paths": ["/mnt/usb0\ndebug=0"],
            },
        ),
        (ROUTE_DEBUG_LOG, {"max_bytes": 1}),
        (ROUTE_KERNEL_LOG, {"max_bytes": 1}),
    )
    for route, payload in invalid_requests:
        http_status, response = client.post(route, payload)
        require(http_status == 400, f"{route} validation returned HTTP {http_status}")
        require(
            integer_field(response, "status") == errno.EINVAL,
            f"{route} validation did not return EINVAL",
        )
    print("[API-TEST] mutation validation=PASS")


def valid_port(value: str) -> int:
    try:
        port = int(value, 10)
    except ValueError as error:
        raise argparse.ArgumentTypeError("port must be an integer") from error
    if not 1 <= port <= 65535:
        raise argparse.ArgumentTypeError("port must be in range 1..65535")
    return port


def positive_timeout(value: str) -> float:
    try:
        timeout = float(value)
    except ValueError as error:
        raise argparse.ArgumentTypeError("timeout must be a number") from error
    if timeout <= 0:
        raise argparse.ArgumentTypeError("timeout must be positive")
    return timeout


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("address", help="PS5 IPv4 address or hostname")
    parser.add_argument("port", type=valid_port, help="ShadowMount HTTP API port")
    parser.add_argument(
        "--timeout", type=positive_timeout, default=60.0, help="request timeout"
    )
    parser.add_argument(
        "--mount-mode",
        choices=("default", "ro", "rw"),
        default="default",
        help="request-scoped mode for the optional mount cycle",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    client = ApiClient(args.address, args.port, args.timeout)
    try:
        test_preflight(client)
        web_result = test_index_page(client)
        test_version(client)
        storage_count = test_storage(client)
        test_storage_job_status(client)
        test_settings_and_log(client)
        images = test_list(client, ROUTE_IMAGES, "images")
        games = test_list(client, ROUTE_GAMES, "games")
        test_game_details(client, games)
        manual_count = test_manual_list(client)
        test_parallel_requests(client)
        mount_mode = "" if args.mount_mode == "default" else args.mount_mode
        mount_result = test_mount_cycle(
            client, find_mount_candidate(games, mount_mode), mount_mode
        )
        test_mutation_validation(client)
        test_rescan(client)
    except (ApiTestError, http.client.HTTPException, OSError) as error:
        print(f"[API-TEST] FAIL: {error}", file=sys.stderr)
        return 1

    print(
        f"[API-TEST] PASS: images={len(images)} games={len(games)} "
        f"storage={storage_count} manual={manual_count} web={web_result} "
        f"mount={mount_result}"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())