feat(load-testing): integrate Locust for on-demand load testing and add related scripts

- Added Locust as a dependency in `pyproject.toml` for load testing capabilities.
- Introduced `run-load-test` script to facilitate running load tests with customizable parameters.
- Created `load/semantic_search.py` for defining the load test scenario targeting the semantic search endpoint.
- Implemented GitHub Actions workflow for automated load testing.
- Added end-to-end smoke test for semantic search in `tests/e2e/test_semantic_search.py` to validate deployment functionality.

Author: pmcfadin

.github/workflows/load_test.yml

@@ -0,0 +1,30 @@
+name: on-demand-load-test
+
+on:
+  workflow_dispatch:
+
+jobs:
+  load-test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.10'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install poetry
+          poetry install --no-interaction --no-root
+          pip install locust
+
+      - name: Run Locust load test (headless)
+        env:
+          STAGING_BASE_URL: ${{ secrets.STAGING_BASE_URL }}
+        run: |
+          locust -f load/semantic_search.py --headless -u 200 -r 20 -t 5m --host "$STAGING_BASE_URL" | cat 

README.md

@@ -89,6 +89,36 @@ poetry run pytest -v           # verbose
 poetry run pytest --cov=app    # with coverage (needs pytest-cov)
 ```
 
+### End-to-End Smoke Test (staging)
+Provide the base URL of a running deployment via the `STAGING_BASE_URL` env var and run the *e2e* marked tests:
+
+```bash
+STAGING_BASE_URL=https://staging.killrvideo.com \
+poetry run pytest tests/e2e -m e2e -q
+```
+
+The test performs a single semantic search request and validates the JSON schema.
+
+### On-Demand Load Testing
+A lightweight Locust scenario ships with the repo. Use the `run-load-test` helper (registered as a Poetry script) to drive a burst of semantic searches against any environment:
+
+```bash
+# 200 users, ramping at 20/s for 5 minutes
+poetry run run-load-test https://staging.killrvideo.com \
+  --users 200 --spawn-rate 20 --duration 5m
+```
+
+Flags:
+* `URL` (positional) – base URL to test
+* `--users` – concurrent users (default 200)
+* `--spawn-rate` – users spawned per second (default 20)
+* `--duration` – test length (Locust time string, default `5m`)
+
+Behind the scenes this wraps:
+```bash
+locust -f load/semantic_search.py --headless -u <users> -r <spawn> -t <duration> --host <URL>
+```
+
 ---
 
 ## Project Structure

load/semantic_search.py

@@ -0,0 +1,28 @@
+"""Locust load-test file exercising the semantic search endpoint.
+
+Run standalone, e.g.:
+
+    locust -f load/semantic_search.py --headless -u 200 -r 20 -t 5m \
+           --host https://staging.killrvideo.com
+
+The parameters above spin up 200 concurrent users with a hatch rate of 20
+users/second, roughly mapping to ~20 requests per second steady-state given
+our simple user scenario.  Adjust figures to match your capacity planning.
+
+The *host* URL is provided at runtime via the ``--host`` flag or the
+``STAGING_BASE_URL`` environment variable.
+"""
+
+from locust import HttpUser, task, between
+
+
+class SemanticSearchUser(HttpUser):  # noqa: D401 – Locust user class
+    # Short random wait to reach ~20 RPS with 200 users
+    wait_time = between(0.1, 0.3)
+
+    @task
+    def search(self):  # noqa: D401
+        # Static query keeps the test deterministic; real test could randomise.
+        self.client.get(
+            "/api/v1/search/videos", params={"query": "cats", "mode": "semantic"}
+        ) 

pyproject.toml

@@ -26,6 +26,7 @@ ruff = "^0.11.12"
 email-validator = "^2.2.0"
 pyyaml = {version = "^6.0", optional = false}
 python-dotenv = "^1.1.0"
+locust = "^2.26.1"
 
 [build-system]
 requires = ["poetry-core"]
@@ -40,3 +41,4 @@ lint.extend-ignore = ["E402", "E702"]
 
 [tool.poetry.scripts]
 gen-openapi = "scripts.generate_openapi:main"
+run-load-test = "scripts.run_load_test:main"

scripts/enable_vector_flag.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+"""Toggle the *VECTOR_SEARCH_ENABLED* flag and optionally trigger DB migration.
+
+Usage (module mode):
+
+    python -m scripts.enable_vector_flag           # Enable flag, run migration
+    python -m scripts.enable_vector_flag --disable # Disable flag
+
+The script works by rewriting the project's ``.env`` file in place. If the
+flag line is missing it is appended.  A best-effort attempt is then made to
+run the migrations via ``scripts.migrate`` (no-op if the module is absent).
+"""
+
+import argparse
+import logging
+import pathlib
+import subprocess
+import sys
+import importlib
+
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+# Project root is two levels up from this file (scripts/enable_vector_flag.py)
+_PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent
+_ENV_FILE = _PROJECT_ROOT / ".env"
+
+
+def _rewrite_env_file(enable: bool) -> None:  # noqa: D401
+    """Update the .env file in-place to reflect *enable* state."""
+
+    if not _ENV_FILE.exists():
+        raise FileNotFoundError(
+            ".env file not found. Aborting toggle of VECTOR_SEARCH_ENABLED."
+        )
+
+    lines = _ENV_FILE.read_text().splitlines(keepends=False)
+    flag_written = False
+    for i, line in enumerate(lines):
+        if line.strip().startswith("VECTOR_SEARCH_ENABLED"):
+            lines[i] = f"VECTOR_SEARCH_ENABLED={'true' if enable else 'false'}"
+            flag_written = True
+            break
+
+    if not flag_written:
+        # Append newline if file doesn't end with one
+        if lines and lines[-1] and not lines[-1].endswith("\n"):
+            lines[-1] += "\n"
+        lines.append(f"VECTOR_SEARCH_ENABLED={'true' if enable else 'false'}")
+
+    _ENV_FILE.write_text("\n".join(lines) + "\n")
+    logger.info("VECTOR_SEARCH_ENABLED=%s written to %s", enable, _ENV_FILE)
+
+
+def _run_migrations() -> None:  # noqa: D401
+    """Attempt to execute migrations via scripts.migrate if present."""
+
+    try:
+        migrate_mod = importlib.import_module("scripts.migrate")
+    except ModuleNotFoundError:
+        logger.warning("scripts.migrate not found – skipping migration step.")
+        return
+
+    if hasattr(migrate_mod, "main"):
+        logger.info("Running DB migrations via scripts.migrate.main() …")
+        migrate_mod.main()  # type: ignore[attr-defined]
+    elif hasattr(migrate_mod, "run"):
+        logger.info("Running DB migrations via scripts.migrate.run() …")
+        migrate_mod.run()  # type: ignore[attr-defined]
+    else:
+        # Fallback to module execution in subprocess to preserve CLI semantics
+        logger.info("Running 'python -m scripts.migrate' as subprocess …")
+        subprocess.run([sys.executable, "-m", "scripts.migrate"], check=False)
+
+
+def main() -> None:  # noqa: D401 – entry point
+    parser = argparse.ArgumentParser(description="Toggle VECTOR_SEARCH_ENABLED flag")
+    parser.add_argument(
+        "--disable",
+        action="store_true",
+        help="Disable vector search instead of enabling it",
+    )
+    parser.add_argument(
+        "--skip-migration",
+        action="store_true",
+        help="Do not attempt to run DB migrations after toggling",
+    )
+
+    args = parser.parse_args()
+    enable = not args.disable
+
+    _rewrite_env_file(enable)
+
+    if not args.skip_migration and enable:
+        _run_migrations()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main() 

scripts/run_load_test.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+"""Convenience wrapper to launch the semantic-search Locust test headless.
+
+Example usage::
+
+- Positional URL (recommended):
+
+    python -m scripts.run_load_test https://staging.killrvideo.com \
+        --users 200 --spawn-rate 20 --duration 5m
+
+- Legacy flag form (still supported):
+
+    python -m scripts.run_load_test --host https://staging.killrvideo.com
+"""
+
+import argparse
+import subprocess
+import sys
+from pathlib import Path
+
+
+DEFAULT_USERS = 200
+DEFAULT_SPAWN_RATE = 20
+DEFAULT_DURATION = "5m"
+
+
+def main() -> None:  # noqa: D401 – entry point
+    parser = argparse.ArgumentParser(description="Run Locust load test (headless)")
+    parser.add_argument(
+        "url",
+        metavar="URL",
+        help="Base URL of the KillrVideo deployment (e.g. https://staging.killrvideo.com)",
+    )
+    # Backwards-compat optional flag (not shown in usage)
+    parser.add_argument("--host", dest="_host_legacy", help=argparse.SUPPRESS)
+    parser.add_argument("--users", type=int, default=DEFAULT_USERS, help="Number of concurrent users (default: 200)")
+    parser.add_argument("--spawn-rate", type=int, default=DEFAULT_SPAWN_RATE, help="User hatch rate per second (default: 20)")
+    parser.add_argument("--duration", default=DEFAULT_DURATION, help="Test duration (Locust time format, default: 5m)")
+    parser.add_argument(
+        "--locust-file",
+        default=str(Path("load/semantic_search.py")),
+        help="Path to the Locust test file (default: load/semantic_search.py)",
+    )
+
+    args = parser.parse_args()
+
+    cmd = [
+        "locust",
+        "-f",
+        args.locust_file,
+        "--headless",
+        "-u",
+        str(args.users),
+        "-r",
+        str(args.spawn_rate),
+        "-t",
+        args.duration,
+        "--host",
+        args._host_legacy if args._host_legacy else args.url,
+    ]
+
+    print("Running:", " ".join(cmd))
+    try:
+        subprocess.run(cmd, check=True)
+    except subprocess.CalledProcessError as exc:
+        sys.exit(exc.returncode)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main() 

tests/e2e/test_semantic_search.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+"""End-to-end smoke test hitting the *staging* KillrVideo deployment.
+
+The test is **skipped automatically** unless the environment variable
+``STAGING_BASE_URL`` is provided.  This allows the regular CI unit-test
+matrix (which spins up an isolated in-process FastAPI app) to pass
+without requiring external connectivity.
+
+When executed with the variable set, the test performs a single semantic
+search request and asserts a successful HTTP 200 response plus the
+presence of the expected JSON keys.
+"""
+
+import os
+
+import pytest
+import httpx
+
+
+STAGING_BASE_URL = os.getenv("STAGING_BASE_URL")
+
+pytestmark = [
+    pytest.mark.e2e,
+    pytest.mark.skipif(
+        not STAGING_BASE_URL, reason="STAGING_BASE_URL env var not configured"
+    ),
+]
+
+
+@pytest.mark.asyncio
+async def test_semantic_search_smoke():  # noqa: D401 – simple smoke test
+    """Perform a single semantic-mode search and validate basic schema."""
+
+    async with httpx.AsyncClient(base_url=STAGING_BASE_URL, timeout=10) as client:
+        resp = await client.get(
+            "/api/v1/search/videos", params={"query": "cats", "mode": "semantic"}
+        )
+
+    # --- Assertions -------------------------------------------------------
+    assert resp.status_code == 200, resp.text
+
+    payload = resp.json()
+
+    # Basic structural checks – we don't lock-step on exact schema here to
+    # remain resilient to future extensions, but we do want the core keys.
+    assert isinstance(payload, dict)
+    assert "data" in payload, "Missing 'data' key in response"
+    assert "pagination" in payload, "Missing 'pagination' key in response"
+
+    # Ensure we actually received *some* results in staging (sanity guard).
+    assert isinstance(payload["data"], list)
+    # We don't enforce a minimum count (could be empty) but the type must hold.