diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 54d963d..3abae71 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -1,11 +1,42 @@ -name: Publish to PyPI +name: CI & Publish on: push: branches: [main] + pull_request: + branches: [main] jobs: + # Tests gate everything: no version bump, no publish unless these pass. + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install package + test deps (incl. fastSDK) + run: pip install -e ".[test]" + + - name: Run tests + run: pytest -v + + # Soft checks: report only, never fail the build. + - name: ruff (soft) + continue-on-error: true + run: ruff check apipod test + + - name: mypy (soft) + continue-on-error: true + run: mypy apipod + + # Runs only on push to main, and only after tests pass. publish: + needs: test + if: github.event_name == 'push' && github.ref == 'refs/heads/main' runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -18,7 +49,7 @@ jobs: with: python-version: "3.12" - - name: Install dependencies + - name: Install build tooling run: pip install build twine - name: Bump patch version diff --git a/README.md b/README.md index 929ed03..a2b3c12 100644 --- a/README.md +++ b/README.md @@ -60,7 +60,7 @@ def hello(name: str): return f"Hello {name}!" # Built-in media processing — uploads/URLs/base64 are parsed for you -@app.endpoint("/process_image") +@app.endpoint("/process-image") def process_image(image: ImageFile): img_array = image.to_np_array() # ... run your AI model here ... diff --git a/apipod/engine/backend/fastapi/router.py b/apipod/engine/backend/fastapi/router.py index 3dde124..8c392f7 100644 --- a/apipod/engine/backend/fastapi/router.py +++ b/apipod/engine/backend/fastapi/router.py @@ -346,6 +346,16 @@ def sync_wrapper(*args, **kwargs): if binding is not None: result = wrap_schema_response(result, binding) return JobResultFactory._serialize_result(result) + + # Python 3.12+ follows __wrapped__ when evaluating inspect.isgeneratorfunction(). + # sync_wrapper is never a generator (it returns a JobResult or StreamProducer). + # Without this, FastAPI would mistake a task endpoint wrapping a generator for a + # streaming route and serve the JobResult as iterated key-value NDJSON chunks. + if hasattr(sync_wrapper, "__wrapped__"): + # Pin the signature explicitly so FastAPI can parse parameters + sync_wrapper.__signature__ = inspect.signature(func) + del sync_wrapper.__wrapped__ + return sync_wrapper def _create_task_endpoint_decorator(self, plan: EndpointExecutionPlan): diff --git a/docs/AGENTS.md b/docs/AGENTS.md new file mode 100644 index 0000000..03f74f6 --- /dev/null +++ b/docs/AGENTS.md @@ -0,0 +1,87 @@ +# Development instructions + +## 1. Think Before Coding + +**Don't assume. Don't hide confusion. Surface tradeoffs.** + +Before implementing: +- State your assumptions explicitly. If uncertain, ask. +- If multiple interpretations exist, present them. Don't pick silently. +- If a simpler approach exists, say so. Push back when warranted. +- If something is unclear, stop. Name what's confusing. Ask. +- Use thinking methods like Elon Musk's first principle reasoning or design-thinking for complex tasks. + +## 2. Simplicity First + +**Minimum code that solves the problem. Nothing speculative.** +- No features beyond what was asked. +- No abstractions for single-use code. +- No "flexibility" or "configurability" that wasn't requested. +- No error handling for impossible scenarios. +- If you write 200 lines and it could be 50, rewrite it. +- Don't repeat yourself. + +Ask yourself: "Would a senior engineer say this is overcomplicated?" If yes, simplify. + +## 3. Surgical Changes + +**Touch only what you must. Clean up only your own mess.** + +When editing existing code: +- Don't "improve" adjacent code, comments, or formatting. +- Match existing style, even if you'd do it differently. +- If you notice unrelated dead code, mention it - don't delete it. + +When your changes create orphans: +- Remove imports/variables/functions that YOUR changes made unused. +- Don't remove pre-existing dead code unless asked. + +The test: Every changed line should trace directly to the user's request. + +## 4. Code quality and style +- Write high-quality, developer-friendly and clean-code, use design-patterns (Refactoring Guru) where helpful. +- Include docstrings and comments to explain the why of a non-obvious method or code. No decorative comments. +- Write memory and compute efficient code. +- Write pythonic code and use builtins when it simplifies (generators, list-comprehensions, functools). +- Reduce complexity and boilerplate. Remove unused code. +- State unclear variable names and rename for better readability and clearity. +- Avoid local imports +- Do not write "deprecated" or backward compability code. +- Update the README.md if there's user related important changes. Update TECHNICAL_README.md for architectural, or project maintainer important changes. Keep the .md update brief use technical correct terms. +- Write flake8 formatted code with "flake8.args": [ + "--max-line-length=120", + "--ignore=E203,W503,E501,E261,W293" + ] +} +- Avoid using direct cmd python calls and pip installs. Work with the local venv if necessary. Always ask if necessary. + +## 5. Reason and thinking style +Do thinking and reasoning like a smart caveman. +- Cut all filler, keep technical substance. +- Drop articles (a, an, the), filler (just, really, basically, actually). +- Drop pleasantries (sure, certainly, happy to). +- No hedging. Fragments fine. Short synonyms. +- Technical terms stay exact. Code blocks unchanged. +- Pattern: [thing] [action] [reason]. [next step] + +## 6. Public facing texts: Readmes, prints, logs. +- Write as technically brilliant European expert who gives you straight answers, backs claims with data, and genuinely wants you to succeed. +- Ttone spectrum Formality 35% (lean casual) · Humor 25% (dry, never flippant) · Complexity 40% (lean technical) · Confidence 80% (very confident). +- Direct without being cold. Say what you mean, no corporate padding. Warm without being informal. Complete sentences, measured enthusiasm. Committed, not hedging. +- Committed, not hedging. "This works", not "this might work". +- "We" for the platform, "you" for the reader. Reserve "I" for rare authored guides only. +- Developer-first, technical correct terms. +- No em-dashes. The character `—` (U+2014) must not appear in ANYWHERE. Use a period, comma, parentheses, or colon. +- No marketing adverbs like seamlessly, effortlessly, robust, powerful, cutting-edge, automatically, intelligently, revolutionary, game-changing, world-class, best-in-class. E.g. three steps, ~X minutes" beats "easy onboarding". +- No bullet-everything. Bullets only for true enumerations (statuses, regions, tiers). +- Subtle transmit socaity's vision and use emotional-intelligence and marketing techniques for end-user intent based messages. +- Use google docstring format. +- The four brand hues (each tied to an ICP) with Tonal scales (50/300/500/700/900). Apply where makes sense. +| Name | Hex | Meaning / ICP | +| ---------------------------- | ----------- | ------------------------------------------------------------------------- | +| **Neon Green** | `#5A8F00` | Developer. Active system health, engineering precision, technical truth. | +| **Silicon Blue** | `#0A86BF` | SMB CTO. Enterprise stability, secure infrastructure, professional trust. | +| **Creator Violet** | `#7C3AED` | Creator. Generative potential, creative momentum, warmth. | +| **Signal Pink (Rose)** | `#EC4899` | Alert. Warnings, critical status, high-priority attention. | + + diff --git a/docs/EXAMPLE_DOCKERFILE_AZURE b/docs/EXAMPLE_DOCKERFILE_AZURE deleted file mode 100644 index d9d8aac..0000000 --- a/docs/EXAMPLE_DOCKERFILE_AZURE +++ /dev/null @@ -1,32 +0,0 @@ -FROM python:3.10-slim - -# Set the working directory -WORKDIR /app - -# ffmpeg needed for media-toolkit video -# Install FFmpeg and dependencies required to install ffmpeg -RUN apt-get update && apt-get install -y \ - software-properties-common \ - apt-transport-https \ - ca-certificates \ - ffmpeg \ - && apt-get clean && rm -rf /var/lib/apt/lists/* - -# Copy the shared connectors module -COPY . /app - -# Install dependencies -RUN pip install --upgrade pip -RUN pip install . - -# Configure environment variables -ENV HOST = "0.0.0.0" - -# Expose the port -ARG port=8000 -ENV PORT=$port -EXPOSE $port - - -# Set the entrypoint with explicit host and port binding -CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/docs/EXAMPLE_DOCKERFILE_RUNPOD b/docs/EXAMPLE_DOCKERFILE_RUNPOD deleted file mode 100644 index c302cd0..0000000 --- a/docs/EXAMPLE_DOCKERFILE_RUNPOD +++ /dev/null @@ -1,40 +0,0 @@ -FROM runpod/pytorch:2.4.0-py3.11-cuda12.4.1-devel-ubuntu22.04 - -SHELL ["/bin/bash", "-c"] - -# Install dependencies -# ffmpeg needed for mediatoolkit video -# Install system dependencies including CUDA and cuDNN -RUN apt-get update && apt-get install -y \ - ffmpeg \ - libcudnn9-cuda-12 \ - libcudnn9-dev-cuda-12 \ - && apt-get clean - -# Ensure CUDA and cuDNN libraries are in the library path -ENV LD_LIBRARY_PATH="/usr/local/cuda/lib64:${LD_LIBRARY_PATH}" - -# create the folder for face2face -WORKDIR / - -# Install Python dependencies (Worker Template) -# added runpod directly in dependencies -# added other libs before which keep failing in requirements file for unknown reasons. -# note that fairseq need to be installed differently in windows and linux -RUN pip install runpod>=1.7.7 && \ - pip install . --no-cache - -# Configure APIPod for RunPod serverless deployment - -ENV APIPOD_COMPUTE="serverless" -ENV APIPOD_PROVIDER="runpod" - -ARG port=8080 -ENV APIPOD_PORT=$port -# allows the docker container to use the port -EXPOSE $port -# allows any IP from the computer to connect to the host -ENV APIPOD_HOST="0.0.0.0" - -# Set the entrypoint with explicit host and port binding -CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/docs/TECHNICAL_README.md b/docs/TECHNICAL_README.md index 24dd744..2a9473f 100644 --- a/docs/TECHNICAL_README.md +++ b/docs/TECHNICAL_README.md @@ -144,4 +144,28 @@ The **stream store** is the pluggable backend that buffers a job's chunks betwee A client calls `POST /tts` with a JSON body. FastAPI validates it against the rewritten signature and calls the outermost wrapper. The file-handling layer converts any media inputs into media-toolkit objects. If the endpoint is queued, the queue mixin stores the job and returns `{job_id, status: "queued", links}` — the worker thread later executes the real function, feeding `JobProgress` updates into the store. The client (typically fastSDK) polls `/status/{job_id}` until `finished` and receives the result, with any returned `AudioFile` serialized as a `FileModel`. Without a queue the same conversion happens inline and the response returns directly. On RunPod, the identical user function is reached through `handler → _router(path) → file handling → execute`, proving the core principle: the function is written once, the backend decides how it runs. +## Testing and CI + +The suite (`test/`) is built so endpoint definitions live apart from test logic, and one helper boots them under any run intent. + +``` +test/ +├── conftest.py # build_service (in-process TestClient), live_service (real subprocess via CLI), config matrix, file paths +├── services/ # reusable services: register(app) callbacks + a runnable entrypoint +│ ├── core_service.py # scalars, custom model, mixed media, JobProgress, file upload +│ ├── schema_service.py # one endpoint per standardized schema, an extended schema, raw/typed mapping CASES +│ └── exec_service.py # runnable service for fastSDK + streaming (predict, echo_image, chat, text, video) +├── files/ # media assets for upload/download tests +├── test_config.py # intent -> backend resolution + /openapi.json loads for every backend +├── test_core.py # core endpoint plumbing (types, model, files, queue lifecycle) +├── test_schemas.py # standardized schema endpoints + response-model normalization +├── test_cli.py # scan / build / simulate produce the right artifacts +└── test_execution.py # fastSDK end-to-end (subprocess) + SSE streaming (in-process) +``` + +Run `pytest`. Two primitives keep tests DRY: `build_service(register, **config)` yields a `TestClient` for an app built from a `register(app)` callback under any `APIPod(**config)` intent; `live_service(simulate=...)` boots a service as a real subprocess through `apipod --start` / `--simulate` and yields its URL. Backends are parametrized from `FASTAPI_CONFIGS`, so a single test asserts behaviour across development, serverless, dedicated and runpod. + +The fastSDK end-to-end tests skip automatically when the installed fastsdk lacks the `connect` API (e.g. a local in-progress build); CI installs one that has it. Pytest config (`pythonpath` in `pyproject.toml`) resolves `import apipod` to the in-repo source and the `conftest`/`services` helpers by name, so no `sys.path` shims are needed and each file is also runnable directly from an IDE via its `__main__` block. + +CI (`.github/workflows/publish.yml`) runs the `test` job on every push and pull request (installing `.[test]`, which includes fastSDK). The `publish` job `needs: test`, so the patch version bump and PyPI upload happen only on a push to `main` after tests pass. flake8 and mypy run as soft checks (`continue-on-error`): they surface warnings without gating the build. diff --git a/pyproject.toml b/pyproject.toml index 4561b03..f7c9f54 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,8 +28,37 @@ apipod = "apipod.cli:main" runpod = [ "runpod>=1.7.7" ] +test = [ + "pytest", + "httpx", + "ruff", + "mypy", +] [project.urls] Repository = "https://github.com/SocAIty/apipod" Homepage = "https://www.socaity.ai" + +[tool.pytest.ini_options] +testpaths = ["test"] +# Make `import apipod` resolve to the in-repo source (not an installed copy) and +# let test modules import the shared `conftest` helpers and `services` package by +# name. The sibling fastsdk entry is used in local dev when the repo is checked +# out next to apipod; it is silently ignored in CI where fastsdk comes from pip. +pythonpath = [".", "test", "../fastsdk"] + +[tool.ruff] +line-length = 120 +exclude = ["venv", "dist", "build", "*.egg-info"] + +[tool.ruff.lint] +# Mirror the previous flake8 config: ignore stylistic rules that conflict with +# black/our formatting conventions. +ignore = ["E203", "E261", "E501", "W293"] + +[tool.mypy] +# Soft static typing: a warning signal, never a build gate (see CI). +ignore_missing_imports = true +follow_imports = "silent" +warn_unused_ignores = false diff --git a/test/__init__.py b/test/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/test/conftest.py b/test/conftest.py new file mode 100644 index 0000000..74b5f7c --- /dev/null +++ b/test/conftest.py @@ -0,0 +1,141 @@ +""" +Shared test infrastructure for the APIPod suite. + +The suite is built on a few small primitives so individual test files stay +declarative and DRY: + +- ``FASTAPI_CONFIGS`` / ``QUEUE_CONFIGS`` — the canonical matrix of APIPod run + intents (development, serverless, dedicated, runpod, ...). Parametrize over + it to assert behaviour holds across every backend. +- ``build_service(register, **config)`` — build an app from a callback that + registers endpoints and hand back a wired FastAPI ``TestClient`` (in-process, + no socket). Used by the config / openapi / core / schema tests. +- ``live_service(simulate=...)`` — boot the example service in a subprocess via + the real ``apipod`` CLI (the same ``--start`` / ``--simulate`` a user runs), + yield its base URL, and shut it down afterwards. Used by the fastSDK + end-to-end execution tests. + +A registrar is just ``def register(app): ...``; keeping endpoints out of the +helpers lets one helper serve config, openapi, schema and execution tests alike. +""" + +import contextlib +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.request +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from apipod import APIPod + +REPO_ROOT = Path(__file__).resolve().parents[1] + +# --------------------------------------------------------------------------- # +# Test asset files (used by the media upload/download tests) +# --------------------------------------------------------------------------- # +FILES_DIR = Path(__file__).parent / "files" +IMAGE_FILE = FILES_DIR / "socaity_logo_for_white_bg.png" +AUDIO_FILE = FILES_DIR / "hermine_clone.wav" +VIDEO_FILE = FILES_DIR / "test_video_ultra_short_short.mp4" + +# Entrypoint booted by the fastSDK end-to-end execution tests. +INFERENCE_SERVICE = Path(__file__).parent / "services" / "core_service.py" + + +# --------------------------------------------------------------------------- # +# Run-intent matrix. Each entry is the kwargs passed to APIPod(...). +# --------------------------------------------------------------------------- # +# FastAPI-backed intents resolve to SocaityFastAPIRouter and expose a real HTTP +# app (the only ones that can serve /openapi.json or run under TestClient). +FASTAPI_CONFIGS = [ + pytest.param({}, id="development"), + pytest.param({"simulate": "serverless"}, id="serverless"), + pytest.param({"simulate": "dedicated"}, id="dedicated"), + pytest.param({"simulate": "dedicated-azure"}, id="dedicated-azure"), + pytest.param({"simulate": "serverless-runpod"}, id="serverless-runpod"), + pytest.param({"simulate": "serverless-azure"}, id="serverless-azure-fallback"), +] + +# Intents that boot a job queue (serverless emulation): endpoints return a +# JobResult processed by the in-process worker. +QUEUE_CONFIGS = [ + pytest.param({"simulate": "serverless"}, id="serverless"), + pytest.param({"simulate": "serverless-runpod"}, id="serverless-runpod"), +] + + +def build_app(register, **config): + """Create an APIPod app for ``config`` and register endpoints via callback.""" + app = APIPod(**config) + register(app) + return app + + +@contextlib.contextmanager +def build_service(register, **config): + """ + Yield a ``TestClient`` for an app built from ``register`` under ``config``. + + The client is entered as a context manager so the app lifespan runs: that + starts the in-process queue worker for serverless intents. + """ + app = build_app(register, **config) + fastapi_app = app.app + fastapi_app.include_router(app) + with TestClient(fastapi_app) as client: + yield client + + +# --------------------------------------------------------------------------- # +# Live server (real subprocess via the apipod CLI) for fastSDK tests +# --------------------------------------------------------------------------- # +def _free_port() -> int: + with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _wait_until_healthy(url: str, proc: subprocess.Popen, timeout: float = 40.0): + """Poll GET /health until the service answers, or fail if the process dies.""" + deadline = time.time() + timeout + while time.time() < deadline: + if proc.poll() is not None: + raise RuntimeError(f"Service process exited early with code {proc.returncode}.") + try: + with urllib.request.urlopen(f"{url}/health", timeout=2) as resp: + if resp.status == 200: + return + except (urllib.error.URLError, ConnectionError, OSError): + time.sleep(0.2) + raise TimeoutError(f"Service at {url} did not become healthy within {timeout}s.") + + +@contextlib.contextmanager +def live_service(simulate: str = "serverless", entrypoint: Path = INFERENCE_SERVICE, port: int = 0): + """ + Boot ``entrypoint`` in a subprocess through the real apipod CLI and yield its URL. + + ``simulate=None`` runs ``apipod --start`` (development); otherwise it runs + ``apipod --simulate `` exactly as a user would on the command line. + """ + port = port or _free_port() + cmd = [sys.executable, "-m", "apipod.cli"] + cmd += ["--start"] if simulate is None else ["--simulate", simulate] + cmd += ["--entrypoint", str(entrypoint), "--host", "127.0.0.1", "--port", str(port)] + + proc = subprocess.Popen(cmd, cwd=str(REPO_ROOT)) + url = f"http://127.0.0.1:{port}" + try: + _wait_until_healthy(url, proc) + yield url + finally: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() diff --git a/test/files/hermine_clone.wav b/test/files/hermine_clone.wav new file mode 100644 index 0000000..fd5fe0c Binary files /dev/null and b/test/files/hermine_clone.wav differ diff --git a/test/files/socaity_logo_for_white_bg.png b/test/files/socaity_logo_for_white_bg.png new file mode 100644 index 0000000..ed563b1 Binary files /dev/null and b/test/files/socaity_logo_for_white_bg.png differ diff --git a/test/files/test_video_ultra_short_short.mp4 b/test/files/test_video_ultra_short_short.mp4 new file mode 100644 index 0000000..a292c56 Binary files /dev/null and b/test/files/test_video_ultra_short_short.mp4 differ diff --git a/test/llm/test_llm_mock.py b/test/llm/test_llm_mock.py deleted file mode 100644 index 3e96e2c..0000000 --- a/test/llm/test_llm_mock.py +++ /dev/null @@ -1,173 +0,0 @@ -""" -Mock API Server using APIPod - No heavy dependencies (No Torch/Transformers) -""" - -from contextlib import asynccontextmanager -from typing import List, Union -from datetime import datetime -import time -import random - -from apipod import APIPod -from apipod.common import schemas -from fastapi import FastAPI - - -class MockModel: - """A mock model that returns random strings to mimic an LLM.""" - - def __init__(self, model_name: str = "mock-model-v1"): - print(f"Initializing Mock Model: {model_name}...") - self.model_name = model_name - self.responses = [ - "The quick brown fox jumps over the lazy dog.", - "Artificial intelligence is transforming the world.", - "APIPod makes it easy to deploy long-running tasks.", - "I am a mock model returning random strings for testing.", - "Python is a versatile programming language.", - "Deep learning is a subset of machine learning.", - "FastAPI is high-performance and easy to use.", - ] - print("Mock Model initialized successfully!") - - def generate(self, messages: list, temperature: float = 0.7, max_tokens: int = 512) -> str: - """Simulate non-streaming generation.""" - # Just pick a random response - return random.choice(self.responses) - - def stream_generate(self, messages: list, temperature: float = 0.7, max_tokens: int = 512): - """Simulate streaming generation by yielding words.""" - response = random.choice(self.responses) - words = response.split() - for i, word in enumerate(words): - # Add a small delay to simulate processing - time.sleep(0.05) - # Add space between words except for the first one - yield word + (" " if i < len(words) - 1 else "") - - def get_embeddings(self, texts: Union[str, List[str]]) -> List[List[float]]: - """Simulate embedding generation with random vectors.""" - if isinstance(texts, str): - texts = [texts] - - # Return random vectors of size 128 - return [[random.uniform(-1, 1) for _ in range(128)] for _ in texts] - -# ============================================================================ -# Application State -# ============================================================================ - -class AppState: - model: MockModel | None = None - -state = AppState() - -@asynccontextmanager -async def lifespan(app: FastAPI): - # Initialize mock model - state.model = MockModel("mock-model-v1") - print("Performing mock warmup...") - yield - print("Shutting down mock server...") - -# ============================================================================ -# Logic Handlers -# ============================================================================ - -def chat_logic(payload: schemas.ChatCompletionRequest): - """Non-streaming chat completion logic.""" - if state.model is None: - raise RuntimeError("Model not initialized") - - response_text = state.model.generate( - messages=payload.messages, - temperature=payload.temperature, - max_tokens=payload.max_tokens or 512 - ) - - # Mock token counts - prompt_tokens = sum(len(m.content.split()) for m in payload.messages) * 2 - completion_tokens = len(response_text.split()) * 2 - - return schemas.ChatCompletionResponse( - object="chat.completion", - created=int(datetime.now().timestamp()), - model=payload.model or state.model.model_name, - choices=[ - schemas.ChatCompletionChoice( - index=0, - message=schemas.ChatCompletionMessage( - role="assistant", - content=response_text - ), - finish_reason="stop" - ) - ], - usage=schemas.Usage( - prompt_tokens=prompt_tokens, - completion_tokens=completion_tokens, - total_tokens=prompt_tokens + completion_tokens - ) - ) - -def embeddings_logic(payload: schemas.EmbeddingRequest): - """Embedding logic.""" - if state.model is None: - raise RuntimeError("Model not initialized") - - texts = [payload.input] if isinstance(payload.input, str) else payload.input - embeddings_list = state.model.get_embeddings(texts) - - # Mock token counts - total_tokens = sum(len(t.split()) for t in texts) * 2 - - return schemas.EmbeddingResponse( - object="list", - model=payload.model or state.model.model_name, - data=[ - schemas.EmbeddingData( - object="embedding", - embedding=emb, - index=i - ) - for i, emb in enumerate(embeddings_list) - ], - usage=schemas.Usage( - prompt_tokens=total_tokens, - completion_tokens=0, - total_tokens=total_tokens - ) - ) - -# ============================================================================ -# API Setup -# ============================================================================ - -app = APIPod( - backend="fastapi", - lifespan=lifespan, -) - -@app.endpoint(path="/chat") -def chat_endpoint(payload: schemas.ChatCompletionRequest): - """Chat completion endpoint with streaming support.""" - if state.model is None: - raise RuntimeError("Model not initialized") - - if payload.stream: - # Just yield raw tokens — APIPod wraps each into a ChatCompletionChunk SSE - # event and emits the closing chunk + [DONE] sentinel out of the box. - return state.model.stream_generate( - messages=payload.messages, - temperature=payload.temperature, - max_tokens=payload.max_tokens, - ) - - return chat_logic(payload) - -@app.endpoint(path="/embeddings") -def embeddings_endpoint(payload: schemas.EmbeddingRequest): - return embeddings_logic(payload) - -if __name__ == "__main__": - app.start() diff --git a/test/llm/test_qwen_llm.py b/test/llm/test_qwen_llm.py deleted file mode 100644 index 1b043e4..0000000 --- a/test/llm/test_qwen_llm.py +++ /dev/null @@ -1,333 +0,0 @@ -""" -Local API Server using APIPod - Optimized for AMD ROCm -""" - -from contextlib import asynccontextmanager -from typing import List, Union -from datetime import datetime -import time -import json - -from apipod import APIPod -from apipod.common import schemas -from fastapi import FastAPI -import uuid -from threading import Thread -import torch -from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer - -# ============================================================================ -# Model Wrapper (Optimized for ROCm) -# ============================================================================ - -class LocalSmallModel: - def __init__(self, model_name: str = "Qwen/Qwen2.5-0.5B-Instruct"): - print(f"Loading {model_name}...") - - # Detect AMD GPU with ROCm - self.device = self._detect_device() - print(f"Using device: {self.device}") - - if self.device == "cuda": - print(f"ROCm GPU detected: {torch.cuda.get_device_name(0)}") - print(f"ROCm version: {torch.version.hip if hasattr(torch.version, 'hip') else 'N/A'}") - - self.tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) - # Qwen requires setting pad_token for batching if it's missing - if self.tokenizer.pad_token is None: - self.tokenizer.pad_token = self.tokenizer.eos_token - - # Use float16 for GPU (ROCm supports it well) - dtype = torch.float16 if self.device == "cuda" else torch.float32 - - self.model = AutoModelForCausalLM.from_pretrained( - model_name, - dtype=dtype, - device_map="auto" if self.device == "cuda" else None, - low_cpu_mem_usage=True, - trust_remote_code=True - ) - - if self.device == "cpu": - self.model = self.model.to(self.device) - - print("Model loaded successfully!") - - def _detect_device(self) -> str: - """Detect if ROCm/CUDA GPU is available""" - if torch.cuda.is_available(): - # This works for both NVIDIA CUDA and AMD ROCm - # PyTorch with ROCm reports as 'cuda' - return "cuda" - else: - print("Warning: No GPU detected. Falling back to CPU.") - return "cpu" - - def generate(self, messages: list, temperature: float = 0.7, max_tokens: int = 512) -> str: - # Standardize messages to list of dicts - chat = [ - msg.model_dump() if hasattr(msg, 'model_dump') else - msg.__dict__ if hasattr(msg, '__dict__') else msg - for msg in messages - ] - - text_prompt = self.tokenizer.apply_chat_template( - chat, tokenize=False, add_generation_prompt=True - ) - - inputs = self.tokenizer(text_prompt, return_tensors="pt").to(self.device) - - safe_max_tokens = max_tokens if max_tokens is not None else 512 - - with torch.no_grad(): - outputs = self.model.generate( - **inputs, - max_new_tokens=safe_max_tokens, - temperature=temperature, - do_sample=temperature > 0, - pad_token_id=self.tokenizer.eos_token_id - ) - - input_len = inputs['input_ids'].shape[1] - response = self.tokenizer.decode(outputs[0][input_len:], skip_special_tokens=True) - return response.strip() - - def stream_generate(self, messages: list, temperature: float = 0.7, max_tokens: int = 512): - """Streaming version of the generation logic.""" - chat = [ - msg.model_dump() if hasattr(msg, 'model_dump') else - msg.__dict__ if hasattr(msg, '__dict__') else msg - for msg in messages - ] - - text_prompt = self.tokenizer.apply_chat_template( - chat, tokenize=False, add_generation_prompt=True - ) - - inputs = self.tokenizer(text_prompt, return_tensors="pt").to(self.device) - - # skip_prompt=True is critical; otherwise, you'll stream back the user's question - streamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True, skip_special_tokens=True) - - generation_kwargs = dict( - **inputs, - streamer=streamer, - max_new_tokens=max_tokens or 512, - temperature=temperature, - do_sample=temperature > 0, - pad_token_id=self.tokenizer.eos_token_id - ) - - # Run generation in a separate thread to prevent blocking the generator - thread = Thread(target=self.model.generate, kwargs=generation_kwargs) - thread.start() - - for new_text in streamer: - # Hugging Face streamer often yields empty strings at the start/end - if new_text: - yield new_text - - def get_embeddings(self, texts: Union[str, List[str]]) -> List[List[float]]: - """Batch processing for embeddings""" - if isinstance(texts, str): - texts = [texts] - - # Batch tokenize - inputs = self.tokenizer( - texts, - return_tensors="pt", - padding=True, - truncation=True, - max_length=512 - ).to(self.device) - - with torch.no_grad(): - outputs = self.model(**inputs, output_hidden_states=True) - # Simple Mean Pooling - embeddings = outputs.hidden_states[-1].mean(dim=1) - - return embeddings.cpu().numpy().tolist() - -# ============================================================================ -# Application State -# ============================================================================ - -class AppState: - model: LocalSmallModel | None = None - -state = AppState() - - -@asynccontextmanager -async def lifespan(app: FastAPI): - # Initialize model - state.model = LocalSmallModel("Qwen/Qwen2.5-0.5B-Instruct") - - # Warmup inference (prevents first-request lag) - print("Performing warmup inference...") - try: - state.model.generate([{"role": "user", "content": "ping"}], max_tokens=1) - except Exception as e: - print(f"Warmup failed (non-fatal): {e}") - - yield - - if torch.cuda.is_available(): - torch.cuda.empty_cache() - -# ============================================================================ -# Logic Handlers (Pure Python) -# ============================================================================ -def chat_logic(payload: schemas.ChatCompletionRequest): - """Non-streaming chat completion""" - state.model = LocalSmallModel("Qwen/Qwen2.5-0.5B-Instruct") - if state.model is None: - raise RuntimeError("Model not initialized") - - response_text = state.model.generate( - messages=payload.messages, - temperature=payload.temperature, - max_tokens=payload.max_tokens or 512 - ) - - # Calculate usage - input_str = "".join([m.content for m in payload.messages]) - prompt_tokens = len(state.model.tokenizer.encode(input_str)) - completion_tokens = len(state.model.tokenizer.encode(response_text)) - - return schemas.ChatCompletionResponse( - id=f"chatcmpl-{uuid.uuid4().hex[:8]}", - object="chat.completion", - created=int(datetime.now().timestamp()), - model=payload.model or "Qwen/Qwen2.5-0.5B-Instruct", - choices=[ - schemas.ChatCompletionChoice( - index=0, - message=schemas.ChatCompletionMessage( # ✅ Correct type - role="assistant", - content=response_text - ), - finish_reason="stop" - ) - ], - usage=schemas.Usage( - prompt_tokens=prompt_tokens, - completion_tokens=completion_tokens, - total_tokens=prompt_tokens + completion_tokens - ) - ) - - -def embeddings_logic(payload: schemas.EmbeddingRequest): - if state.model is None: - raise RuntimeError("Model not initialized") - - texts = [payload.input] if isinstance(payload.input, str) else payload.input - embeddings_list = state.model.get_embeddings(texts) - total_tokens = sum(len(state.model.tokenizer.encode(t)) for t in texts) - - return schemas.EmbeddingResponse( - object="list", - model=payload.model or "Qwen/Qwen2.5-0.5B-Instruct", - data=[ - schemas.EmbeddingData( - object="embedding", - embedding=emb, - index=i - ) - for i, emb in enumerate(embeddings_list) - ], - usage=schemas.Usage( - prompt_tokens=total_tokens, - completion_tokens=0, # Embeddings don't have completion tokens - total_tokens=total_tokens - ) - ) -# ============================================================================ -# API Setup -# ============================================================================ -app = APIPod( - backend="fastapi", - lifespan=lifespan, -) - -@app.endpoint(path="/chat") -def chat_endpoint(payload: schemas.ChatCompletionRequest): - """Chat completion endpoint with streaming support.""" - if payload.stream: - def sse_generator(): - state.model = LocalSmallModel("Qwen/Qwen2.5-0.5B-Instruct") - if state.model is None: - raise RuntimeError("Model not initialized") - - chunk_id = f"chatcmpl-{uuid.uuid4().hex[:8]}" - created_time = int(datetime.now().timestamp()) - model_name = payload.model or "Qwen/Qwen2.5-0.5B-Instruct" - - for token in state.model.stream_generate( - messages=payload.messages, - temperature=payload.temperature, - max_tokens=payload.max_tokens - ): - # Use ChatCompletionChunk schema - chunk = schemas.ChatCompletionChunk( - id=chunk_id, - object="chat.completion.chunk", - created=created_time, - model=model_name, - choices=[ - schemas.ChatStreamChoice( - index=0, - delta=schemas.ChatDelta(content=token), - finish_reason=None - ) - ] - ) - yield f"data: {chunk.model_dump_json()}\n\n" - - # Final chunk with finish_reason - final_chunk = schemas.ChatCompletionChunk( - id=chunk_id, - object="chat.completion.chunk", - created=created_time, - model=model_name, - choices=[ - schemas.ChatStreamChoice( - index=0, - delta=schemas.ChatDelta(content=None), - finish_reason="stop" - ) - ] - ) - yield f"data: {final_chunk.model_dump_json()}\n\n" - yield "data: [DONE]\n\n" - - return sse_generator() - else: - return chat_logic(payload) - - -@app.endpoint(path="/embeddings") -def embeddings_endpoint(payload: schemas.EmbeddingRequest): - return embeddings_logic(payload) - - -@app.endpoint(path="/stream") -def stream_endpoint(): - def simple_stream(): - try: - for i in range(10): - message = {"index": i, "text": f"Message {i}"} - yield f"data: {json.dumps(message)}\n\n" - time.sleep(1) - yield "data: [DONE]\n\n" - except GeneratorExit: - print("Client disconnected") - except Exception as e: - yield f"data: {{\"error\": \"{str(e)}\"}}\n\n" - - return simple_stream() - - -if __name__ == "__main__": - app.start() \ No newline at end of file diff --git a/test/services/__init__.py b/test/services/__init__.py new file mode 100644 index 0000000..5bd6003 --- /dev/null +++ b/test/services/__init__.py @@ -0,0 +1,11 @@ +"""Reusable APIPod service definitions for the test suite. + +Three services, each exposing a ``register(app)`` callback: + +- ``core_service`` parameter shapes, file I/O, JobProgress; also a runnable + entrypoint for the fastSDK subprocess tests. +- ``schema_service`` one endpoint per standardized (OpenAI-compatible) schema; + schema extension and raw/typed response mapping cases. +- ``streaming_service`` three streaming modes: plain text tokens, binary frames, + ChatCompletion SSE deltas. +""" diff --git a/test/services/core_service.py b/test/services/core_service.py new file mode 100644 index 0000000..5272462 --- /dev/null +++ b/test/services/core_service.py @@ -0,0 +1,103 @@ +"""Core service: the parameter shapes and file-handling an author actually writes. + +- ``register_minimal`` two trivial endpoints (OpenAPI smoke test, config matrix). +- ``register`` full set: scalars, custom pydantic model, predict/echo_image + (used by fastSDK tests), broad media+model endpoint, + JobProgress lifecycle, single file upload. + +The module-level ``app`` makes this file a valid ``apipod --start`` entrypoint, +so fastSDK end-to-end tests can boot it as a subprocess via ``conftest.live_service``. + +APIPod normalizes ``_`` to ``-`` in routes, so e.g. ``/mixed_media`` is served +at ``/mixed-media``. +""" + +import time +from typing import List, Optional + +from pydantic import BaseModel +from fastapi import UploadFile as FastAPIUploadFile + +from apipod import APIPod, JobProgress, MediaFile, ImageFile, AudioFile, VideoFile, FileModel + + +class MoreParams(BaseModel): + pam1: str = "pam1" + pam2: int = 42 + + +def register_minimal(app): + @app.endpoint("/echo") + def echo(text: str): + return text + + @app.endpoint("/add") + def add(a: int, b: int = 1): + return a + b + + +def register(app): + # Plain scalars — verifies standard type coercion. + @app.endpoint("/scalars") + def scalars(text: str, count: int, ratio: float = 1.0, flag: bool = False): + return {"text": text, "count": count, "ratio": ratio, "flag": flag} + + # Custom pydantic model. + @app.endpoint("/model") + def model(order: MoreParams): + return {"pam1": order.pam1, "pam2": order.pam2} + + # Clean scalar + media I/O; also the primary targets for fastSDK tests. + @app.endpoint("/predict") + def predict(text: str, times: int = 1): + return text * times + + @app.endpoint("/echo_image") + def echo_image(image: ImageFile): + return image + + # One endpoint, many input options: optional/required media, raw UploadFile, + # union media types, media lists, a custom model and plain scalars. + @app.endpoint("/mixed_media") + def mixed_media( + job_progress: JobProgress, + anyfile1: Optional[MediaFile], + anyfile2: FileModel, + anyfile3: FastAPIUploadFile, + img: ImageFile | str | bytes | FileModel, + audio: AudioFile, + video: VideoFile, + anyfiles: List[MediaFile], + a_base_model: Optional[MoreParams], + anint2: int, + anyImages: List[ImageFile] = ["default_value"], + astring: str = "master_of_desaster", + anint: int = 42, + ): + return anyfile3, str, anyfile1.to_base64(), img.to_base64(), anyfiles + + # JobProgress + queue lifecycle. + @app.post(path="/test_job_progress", queue_size=10) + def job_progress_demo(job_progress: JobProgress, fries_name: str, amount: int = 1): + job_progress.set_status(0.1, f"started new fries creation {fries_name}") + time.sleep(0.2) + job_progress.set_status(0.5, f"working on it, lots to do {amount}") + time.sleep(0.2) + job_progress.set_status(0.8, "almost done") + time.sleep(0.2) + return f"Your fries {fries_name} are ready" + + # Single file upload — also verifies base64 round-trip. + @app.endpoint("/single_file_upload") + def single_file_upload(job_progress: JobProgress, file1: ImageFile): + return file1.to_base64() + + +# Runnable entrypoint: booted by ``apipod --start`` / ``--simulate`` in +# fastSDK end-to-end tests via ``conftest.live_service``. +app = APIPod() # run intent injected via env by the CLI +register(app) + + +if __name__ == "__main__": + app.start() diff --git a/test/services/schema_service.py b/test/services/schema_service.py new file mode 100644 index 0000000..55e376e --- /dev/null +++ b/test/services/schema_service.py @@ -0,0 +1,117 @@ +"""Schema service: endpoints for the standardized (OpenAI-compatible) schemas. + +- ``register_all`` one endpoint per registered request schema. +- ``register_extended`` an endpoint whose input extends a schema with a field. +- ``register_mapping`` raw + typed return endpoint for each mapping CASE. + +``CASES`` drives the response-model passthrough vs. raw-value normalization +tests (text, embedding and media-envelope schemas). +""" + +from dataclasses import dataclass +from typing import Any, Type + +from apipod.common import schemas +from apipod.engine.backend.schema_resolve import SCHEMA_REGISTRY + + +def tag_path(request_model: Type) -> str: + """Public route path for a schema (APIPod normalizes ``_`` to ``-``).""" + return "/" + SCHEMA_REGISTRY[request_model].tag.replace("_", "-") + + +def _schema_endpoint(request_model: Type, return_value: Any, name: str): + """Build an endpoint whose only parameter is annotated with ``request_model``.""" + + def endpoint(request): + return return_value + + endpoint.__name__ = name + endpoint.__annotations__ = {"request": request_model} + return endpoint + + +def register_all(app): + for request_model, spec in SCHEMA_REGISTRY.items(): + app.endpoint(tag_path(request_model))(_schema_endpoint(request_model, {}, f"ep_{spec.tag}")) + + +class ChatRequestPlus(schemas.ChatCompletionRequest): + persona: str = "pirate" + + +def register_extended(app): + @app.endpoint("/chat-extended") + def chat_extended(request: ChatRequestPlus): + # The extra `persona` input field is parsed and drives the answer. + return f"[{request.persona}] {request.messages[-1].content}" + + +@dataclass +class SchemaCase: + request_model: Type + response_model: Type + payload: dict + raw: Any # raw value the endpoint may return + typed: Any # the response_model instance the endpoint may return + + +CASES = [ + SchemaCase( + schemas.ChatCompletionRequest, schemas.ChatCompletionResponse, + {"messages": [{"role": "user", "content": "hi"}]}, + raw="hello there", + typed=schemas.ChatCompletionResponse( + created=0, + choices=[schemas.ChatCompletionChoice( + index=0, message=schemas.ChatCompletionMessage(content="hello there"), finish_reason="stop")], + ), + ), + SchemaCase( + schemas.CompletionRequest, schemas.CompletionResponse, + {"prompt": "hi"}, + raw="completed", + typed=schemas.CompletionResponse( + created=0, + choices=[schemas.CompletionChoice(text="completed", index=0, finish_reason="stop")], + ), + ), + SchemaCase( + schemas.EmbeddingRequest, schemas.EmbeddingResponse, + {"input": "hi"}, + raw=[0.1, 0.2, 0.3], + typed=schemas.EmbeddingResponse(data=[schemas.EmbeddingData(embedding=[0.1, 0.2, 0.3], index=0)]), + ), + SchemaCase( + schemas.ImageGenerationRequest, schemas.ImageGenerationResponse, + {"prompt": "a cat"}, raw={"data": []}, + typed=schemas.ImageGenerationResponse(created=0, data=[]), + ), + SchemaCase( + schemas.VideoGenerationRequest, schemas.VideoGenerationResponse, + {"prompt": "a cat"}, raw={"data": []}, + typed=schemas.VideoGenerationResponse(created=0, data=[]), + ), + SchemaCase( + schemas.SpeechRequest, schemas.SpeechResponse, + {"input": "hello"}, raw={"data": []}, + typed=schemas.SpeechResponse(created=0, data=[]), + ), + SchemaCase( + schemas.Generation3DRequest, schemas.Generation3DResponse, + {"prompt": "a chair"}, raw={"data": []}, + typed=schemas.Generation3DResponse(created=0, data=[]), + ), + SchemaCase( + schemas.MultimodalEmbeddingRequest, schemas.MultimodalEmbeddingResponse, + {"input": "hi"}, raw={"data": []}, + typed=schemas.MultimodalEmbeddingResponse(data=[]), + ), +] + + +def register_mapping(app): + for case in CASES: + tag = SCHEMA_REGISTRY[case.request_model].tag.replace("_", "-") + app.endpoint(f"/{tag}-raw")(_schema_endpoint(case.request_model, case.raw, f"{tag}_raw")) + app.endpoint(f"/{tag}-typed")(_schema_endpoint(case.request_model, case.typed, f"{tag}_typed")) diff --git a/test/services/streaming_service.py b/test/services/streaming_service.py new file mode 100644 index 0000000..45c1bd6 --- /dev/null +++ b/test/services/streaming_service.py @@ -0,0 +1,33 @@ +"""Streaming service: all three streaming modes APIPod supports. + +Endpoint types: +- ``/text`` plain text-token generator (generic generator endpoint), +- ``/video`` raw binary frames (byte generator, SSE-framed as base64), +- ``/chat`` ChatCompletionRequest with optional stream=True (schema endpoint: + returns token deltas wrapped into ChatCompletionChunk SSE events). + +Constants are exported so streaming tests can assert the exact expected output +without duplicating the data. +""" + +from apipod.common import schemas + +CHAT_TOKENS = ["Hello", ", ", "world", "!"] +TEXT_TOKENS = ["APIPod ", "streams ", "tokens ", "one ", "by ", "one."] +VIDEO_FRAMES = [bytes([i]) * 2048 for i in range(5)] + + +def register(app): + @app.endpoint("/text") + def stream_text(): + yield from TEXT_TOKENS + + @app.endpoint("/video") + def stream_video(): + yield from VIDEO_FRAMES + + @app.endpoint("/chat") + def chat(request: schemas.ChatCompletionRequest): + if request.stream: + return (token for token in CHAT_TOKENS) + return "".join(CHAT_TOKENS) diff --git a/test/test_cli.py b/test/test_cli.py new file mode 100644 index 0000000..9294903 --- /dev/null +++ b/test/test_cli.py @@ -0,0 +1,89 @@ +""" +B) CLI commands: scan, build, simulate. + +Each command is driven through its real entry point (``apipod.cli``) against a +throwaway project, asserting it produces the expected artifacts: +- ``scan`` -> apipod-deploy/apipod.json +- ``build`` -> apipod-deploy/Dockerfile (Docker invocation itself is stubbed) +- ``simulate`` -> applies the run intent (env) and boots the resolved entrypoint + via the app's ``start`` (uvicorn.run is stubbed so the test does not block). +""" + +import argparse + +import pytest + +import apipod.cli as cli +from apipod.deploy.deployment_manager import DeploymentManager + +SERVICE_FILE = """\ +from apipod import APIPod + +app = APIPod(title="cli-test-service") + + +@app.endpoint("/ping") +def ping(): + return "pong" + + +if __name__ == "__main__": + app.start() +""" + + +@pytest.fixture +def project(tmp_path, monkeypatch): + """A minimal apipod project; cwd is moved into it so the CLI scans it.""" + (tmp_path / "pyproject.toml").write_text('[project]\nname = "svc"\nversion = "0.0.0"\n') + (tmp_path / "main.py").write_text(SERVICE_FILE) + monkeypatch.chdir(tmp_path) + # Auto-confirm every interactive prompt. + monkeypatch.setattr("builtins.input", lambda *a, **k: "y") + return tmp_path + + +def test_scan_generates_config(project): + cli.run_scan() + + config = project / "apipod-deploy" / "apipod.json" + assert config.exists() + + import json + data = json.loads(config.read_text()) + assert data["entrypoint"] == "main.py" + assert data["title"] == "cli-test-service" + + +def test_build_generates_dockerfile(project, monkeypatch): + # Never actually invoke Docker. + monkeypatch.setattr(DeploymentManager, "build_docker_image", lambda self, title: True) + + cli.run_build(argparse.Namespace(build=True)) + + dockerfile = project / "apipod-deploy" / "Dockerfile" + assert dockerfile.exists() + assert "FROM" in dockerfile.read_text() + + +def test_simulate_applies_intent_and_starts(project, monkeypatch): + started = {} + + def fake_uvicorn_run(app, host=None, port=None, **kwargs): + started["host"] = host + started["port"] = port + + monkeypatch.setattr("uvicorn.run", fake_uvicorn_run) + + args = argparse.Namespace( + simulate="serverless", direct=None, entrypoint="main.py", host="127.0.0.1", port=8123 + ) + cli.run_simulate(args) + + import os + assert os.environ["APIPOD_SIMULATE"] == "serverless" # intent applied as env + assert started == {"host": "127.0.0.1", "port": 8123} # entrypoint booted + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/test/test_clients_six_schemas.py b/test/test_clients_six_schemas.py deleted file mode 100644 index 43dc4ba..0000000 --- a/test/test_clients_six_schemas.py +++ /dev/null @@ -1,185 +0,0 @@ -""" -Exercises the test service in three ways, as Matthias asked on PR #15: - 1. fastSDK client - 2. HTTP request with base64 - 3. HTTP request with URL - -Run the service first in another shell: - python examples/test_service_six_schemas.py - -Then run this: - python examples/test_clients_six_schemas.py -""" - -import base64 -import io -import json -import sys -import urllib.request -from pathlib import Path - -import requests - -BASE_URL = "http://localhost:8000" -# A small public PNG used for the "URL" test. Picked because it doesn't need -# auth and returns a real PNG header (so media_toolkit can sniff it). -PUBLIC_IMAGE_URL = "https://upload.wikimedia.org/wikipedia/commons/thumb/4/47/PNG_transparency_demonstration_1.png/120px-PNG_transparency_demonstration_1.png" - - -def _make_local_png_bytes() -> bytes: - """Build a small but real-sized PNG (32x32 solid red) in-process. - - media_toolkit's base64 sniffer needs a payload long enough to be confident - it is base64 and not a short identifier-looking string. The original 1x1 - PNG (67 bytes / 92 b64 chars) sat below the heuristic; 32x32 (~100 bytes / - ~140 b64 chars) is comfortably above it and matches realistic JSON callers. - """ - import struct - import zlib - - width = height = 32 - color = (255, 0, 0) - - ihdr_data = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) - raw = b"" - for _ in range(height): - raw += b"\x00" + bytes(color * width) - idat = zlib.compress(raw) - - return ( - b"\x89PNG\r\n\x1a\n" - + struct.pack(">I", len(ihdr_data)) + b"IHDR" + ihdr_data + struct.pack(">I", zlib.crc32(b"IHDR" + ihdr_data)) - + struct.pack(">I", len(idat)) + b"IDAT" + idat + struct.pack(">I", zlib.crc32(b"IDAT" + idat)) - + struct.pack(">I", 0) + b"IEND" + struct.pack(">I", zlib.crc32(b"IEND")) - ) - - -def _print_result(label: str, status: int, body) -> None: - print(f"\n=== {label} ===") - print(f"HTTP {status}") - print(json.dumps(body, indent=2) if isinstance(body, (dict, list)) else body) - - -# ------------------------------------------------------------------- -# 1. HTTP with base64 -# ------------------------------------------------------------------- -def test_http_base64() -> bool: - png = _make_local_png_bytes() - b64 = base64.b64encode(png).decode("ascii") - payload = { - "model": "test-model", - "image": b64, - } - r = requests.post(f"{BASE_URL}/vision", json=payload, timeout=15) - try: - body = r.json() - except Exception: - body = r.text - _print_result("HTTP base64", r.status_code, body) - return r.status_code == 200 - - -# ------------------------------------------------------------------- -# 2. HTTP with URL -# ------------------------------------------------------------------- -def test_http_url() -> bool: - payload = { - "model": "test-model", - "image": PUBLIC_IMAGE_URL, - } - r = requests.post(f"{BASE_URL}/vision", json=payload, timeout=30) - try: - body = r.json() - except Exception: - body = r.text - _print_result("HTTP URL", r.status_code, body) - return r.status_code == 200 - - -# ------------------------------------------------------------------- -# 3. fastSDK client -# ------------------------------------------------------------------- -def test_fastsdk() -> bool: - """ - Hits /vision through fastSDK. - - Note: fastSDK's public entry points (`FastClient`, `create_sdk`, - `APISeex`) all expect either a service registered in the Registry - or a generated client class. There is no ad-hoc "given this URL, - just POST this payload" path. For the purposes of PR #15 this means - a single localhost service like the one in this example is not the - intended fastSDK use case; the right shape would be to register - the service first and then call it through the SDK. - - For now we just dispatch a low-level APISeex with a minimal - ServiceDefinition and report whatever happens. - """ - try: - from fastsdk import APISeex, ImageFile - from apipod_registry.definitions.service_definitions import ( - ServiceDefinition, EndpointDefinition, EndpointParameter, - ServiceAddress, - ) - except Exception as exc: - _print_result("fastSDK", -1, f"import failed: {exc.__class__.__name__}: {exc}") - return False - - try: - service = ServiceDefinition( - id="test-svc", - display_name="test svc", - description="local test service", - service_address=ServiceAddress(url=BASE_URL), - endpoints=[], - ) - endpoint = EndpointDefinition( - id="vision", - path="/vision", - display_name="vision", - description="vision echo", - parameters=[ - EndpointParameter( - name="model", type="string", required=True, - location="body", default="test-model", - ), - EndpointParameter( - name="image", type="image", required=True, - location="body", - ), - ], - ) - png = _make_local_png_bytes() - img = ImageFile().from_bytes(png) - seex = APISeex( - service_def=service, - endpoint_def=endpoint, - data={"model": "test-model", "image": img}, - ) - result = seex.wait_for_result(timeout_s=15) - if seex.error: - _print_result("fastSDK", -1, f"server returned error: {seex.error}") - return False - if result is None: - _print_result("fastSDK", -1, "no result returned (request likely not dispatched; needs Registry setup)") - return False - _print_result("fastSDK", 200, str(result)) - return True - except Exception as exc: - _print_result("fastSDK", -1, f"call failed: {exc.__class__.__name__}: {exc}") - return False - - -def main() -> int: - results = { - "HTTP base64": test_http_base64(), - "HTTP URL": test_http_url(), - "fastSDK": test_fastsdk(), - } - print("\n=== Summary ===") - for name, ok in results.items(): - print(f" {name}: {'PASS' if ok else 'FAIL'}") - return 0 if all(results.values()) else 1 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/test/test_config.py b/test/test_config.py new file mode 100644 index 0000000..eafac83 --- /dev/null +++ b/test/test_config.py @@ -0,0 +1,79 @@ +""" +A) Configuration: every run intent resolves to the right backend and can serve a +loadable OpenAPI schema. + +``APIPod(simulate=..., direct=...)`` is a factory. Part 1 pins the intent -> +backend mapping (FastAPI vs RunPod, job queue present or not). Part 2 boots a +minimal service under each FastAPI intent and asserts ``/openapi.json`` loads and +documents the routes (fastSDK and Swagger UI build everything from it). +""" + +import pytest + +from apipod import APIPod +from apipod.engine.backend.fastapi.router import SocaityFastAPIRouter +from apipod.engine.backend.runpod.router import SocaityRunpodRouter + +from conftest import FASTAPI_CONFIGS, build_service +from services import core_service + +FASTAPI = SocaityFastAPIRouter +RUNPOD = SocaityRunpodRouter + +# (APIPod kwargs, expected backend, expects_job_queue) +RESOLUTIONS = [ + pytest.param({}, FASTAPI, False, id="development"), + pytest.param({"simulate": ""}, FASTAPI, True, id="simulate-bare"), + pytest.param({"simulate": "serverless"}, FASTAPI, True, id="serverless"), + pytest.param({"simulate": "dedicated"}, FASTAPI, False, id="dedicated"), + pytest.param({"simulate": "dedicated-azure"}, FASTAPI, False, id="dedicated-azure"), + pytest.param({"simulate": "serverless-runpod"}, FASTAPI, True, id="serverless-runpod"), + pytest.param({"simulate": "serverless-azure"}, FASTAPI, True, id="serverless-azure-fallback"), + pytest.param({"simulate": "serverless-runpod", "direct": True}, RUNPOD, False, id="runpod-direct"), +] + + +# --------------------------------------------------------------------------- # +# 1. Intent -> backend resolution +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("kwargs, backend, expects_queue", RESOLUTIONS) +def test_intent_resolves_to_backend(kwargs, backend, expects_queue): + app = APIPod(**kwargs) + assert isinstance(app, backend) + if backend is FASTAPI: + assert (app.job_queue is not None) is expects_queue + else: + assert app.simulate is True # runpod local emulator + + +@pytest.mark.parametrize("target", ["invalid", "serverless-invalid"]) +def test_invalid_target_raises(target): + with pytest.raises(ValueError, match="Invalid"): + APIPod(simulate=target) + + +# --------------------------------------------------------------------------- # +# 2. OpenAPI schema loads for every backend +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("config", FASTAPI_CONFIGS) +def test_openapi_loads_for_fastapi_backends(config): + with build_service(core_service.register_minimal, **config) as client: + schema = client.get("/openapi.json").json() + assert schema["openapi"].startswith("3.") + assert {"/echo", "/add"} <= set(schema["paths"]) + assert "apipod" in schema["info"] # APIPod stamps its version in + + +def test_openapi_for_runpod_direct(): + """The RunPod worker has no HTTP layer; it synthesizes the schema itself.""" + app = APIPod(simulate="serverless-runpod", direct=True) + core_service.register_minimal(app) + + schema = app.get_openapi_schema() + assert schema["openapi"].startswith("3.") + assert {"/echo", "/add"} <= set(schema["paths"]) + assert "apipod" in schema["info"] + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/test/test_config_resolution.py b/test/test_config_resolution.py deleted file mode 100644 index 0b534b0..0000000 --- a/test/test_config_resolution.py +++ /dev/null @@ -1,74 +0,0 @@ -from apipod import APIPod -from apipod.engine.backend.fastapi.router import SocaityFastAPIRouter -from apipod.engine.backend.runpod.router import SocaityRunpodRouter - - -def test_development_default(): - # APIPod() -> plain FastAPI, no job queue. - app = APIPod() - assert isinstance(app, SocaityFastAPIRouter) - assert app.job_queue is None - - -def test_simulate_serverless_default(): - # simulate (bare / "serverless") -> FastAPI + Local Job Queue. - for target in ("", "serverless"): - app = APIPod(simulate=target) - assert isinstance(app, SocaityFastAPIRouter) - assert app.job_queue is not None - - -def test_simulate_dedicated(): - # dedicated -> Standard FastAPI (no queue). - app = APIPod(simulate="dedicated") - assert isinstance(app, SocaityFastAPIRouter) - assert app.job_queue is None - - -def test_simulate_dedicated_azure(): - # dedicated-azure -> FastAPI (direct client), no queue. - app = APIPod(simulate="dedicated-azure") - assert isinstance(app, SocaityFastAPIRouter) - assert app.job_queue is None - - -def test_simulate_serverless_runpod(): - # serverless-runpod (no direct) -> Socaity emulation = FastAPI + job queue. - app = APIPod(simulate="serverless-runpod") - assert isinstance(app, SocaityFastAPIRouter) - assert app.job_queue is not None - - -def test_simulate_serverless_runpod_direct(): - # serverless-runpod + direct -> RunPod local emulation. - app = APIPod(simulate="serverless-runpod", direct=True) - assert isinstance(app, SocaityRunpodRouter) - assert app.simulate is True - - -def test_simulate_serverless_azure_warns_and_falls_back(): - # azure has no serverless -> FastAPI + job queue (with warning). - app = APIPod(simulate="serverless-azure") - assert isinstance(app, SocaityFastAPIRouter) - assert app.job_queue is not None - - -def test_invalid_target_values(): - for target in ("invalid", "serverless-invalid"): - try: - APIPod(simulate=target) - assert False - except ValueError as e: - assert "Invalid" in str(e) - - -if __name__ == "__main__": - test_development_default() - test_simulate_serverless_default() - test_simulate_dedicated() - test_simulate_dedicated_azure() - test_simulate_serverless_runpod() - test_simulate_serverless_runpod_direct() - test_simulate_serverless_azure_warns_and_falls_back() - test_invalid_target_values() - print("All config resolution tests passed.") diff --git a/test/test_core.py b/test/test_core.py new file mode 100644 index 0000000..30ce8b2 --- /dev/null +++ b/test/test_core.py @@ -0,0 +1,84 @@ +""" +A) Core endpoint plumbing on the FastAPI backend. + +Exercises the parameter shapes an author writes and the file/queue machinery: +- standard scalar types and a custom pydantic model, +- the broad ``/mixed-media`` endpoint (model + media + scalars) builds & documents, +- single file upload via multipart and via base64 (round-trips to base64 out), +- the job queue + JobProgress lifecycle (submit, poll, read result). + +Development mode runs endpoints inline; serverless mode queues them. +""" + +import base64 +import time + +import pytest + +from conftest import IMAGE_FILE, build_service +from services import core_service + + +@pytest.fixture(scope="module") +def dev_client(): + with build_service(core_service.register) as c: + yield c + + +@pytest.fixture(scope="module") +def serverless_client(): + with build_service(core_service.register, simulate="serverless") as c: + yield c + + +def test_standard_types(dev_client): + resp = dev_client.post("/scalars", data={"text": "hi", "count": 3, "ratio": 2.5, "flag": True}) + assert resp.status_code == 200, resp.text + assert resp.json() == {"text": "hi", "count": 3, "ratio": 2.5, "flag": True} + + +def test_custom_pydantic_model(dev_client): + resp = dev_client.post("/model", data={"pam1": "burger", "pam2": 2}) + assert resp.status_code == 200, resp.text + assert resp.json() == {"pam1": "burger", "pam2": 2} + + +def test_mixed_media_is_documented(dev_client): + """The broad mixed endpoint (model + media + scalars) builds and is documented.""" + schema = dev_client.get("/openapi.json").json() + assert "/mixed-media" in schema["paths"] + + +def test_single_file_upload_multipart(dev_client): + with open(IMAGE_FILE, "rb") as fh: + resp = dev_client.post("/single-file-upload", files={"file1": (IMAGE_FILE.name, fh, "image/png")}) + assert resp.status_code == 200, resp.text + assert base64.b64decode(resp.json()) # valid base64 back out + + +def test_single_file_upload_base64(dev_client): + payload = base64.b64encode(IMAGE_FILE.read_bytes()).decode() + resp = dev_client.post("/single-file-upload", data={"file1": payload}) + assert resp.status_code == 200, resp.text + assert base64.b64decode(resp.json()) # round-tripped through ImageFile + + +def test_job_progress_lifecycle(serverless_client): + submit = serverless_client.post("/test-job-progress", data={"fries_name": "curly", "amount": 2}) + assert submit.status_code == 200, submit.text + status_url = submit.json()["links"]["status"] + + deadline = time.time() + 15 + while time.time() < deadline: + body = serverless_client.get(status_url).json() + if body.get("result") is not None: + assert body["result"] == "Your fries curly are ready" + return + if body.get("status") in ("failed", "timeout", "not_found"): + pytest.fail(f"job ended unexpectedly: {body}") + time.sleep(0.1) + pytest.fail("job did not finish in time") + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/test/test_core_functionalities.py b/test/test_core_functionalities.py deleted file mode 100644 index d645b58..0000000 --- a/test/test_core_functionalities.py +++ /dev/null @@ -1,74 +0,0 @@ -from typing import List, Optional - -from pydantic import BaseModel - -import time - -from apipod import APIPod -from apipod import JobProgress -from apipod import MediaFile, ImageFile, AudioFile, VideoFile, FileModel -from fastapi import UploadFile as fastapiUploadFile - -app = APIPod(simulate="serverless") - - -@app.post(path="/test_job_progress", queue_size=10) -def test_job_progress(job_progress: JobProgress, fries_name: str, amount: int = 1): - job_progress.set_status(0.1, f"started new fries creation {fries_name}") - time.sleep(1) - job_progress.set_status(0.5, f"I am working on it. Lots of work to do {amount}") - time.sleep(2) - job_progress.set_status(0.8, "Still working on it. Almost done") - time.sleep(2) - return f"Your fries {fries_name} are ready" - - -class MoreParams(BaseModel): - pam1: str = "pam1" - pam2: int = 42 - - -@app.endpoint("/mixed_media") -def test_mixed_media( - job_progress: JobProgress, - anyfile1: Optional[MediaFile], - anyfile2: FileModel, - anyfile3: fastapiUploadFile, - img: ImageFile | str | bytes | FileModel, - audio: AudioFile, - video: VideoFile, - anyfiles: List[MediaFile], - a_base_model: Optional[MoreParams], - anint2: int, - anyImages: List[ImageFile] = ["default_value"], - astring: str = "master_of_desaster", - anint: int = 42 -): - content_one = anyfile1.to_base64() - content_two = img.to_base64() - return anyfile3, str, content_one, content_two, anyfiles - - -@app.endpoint("test_single_file_upload") -def test_single_file_upload( - job_progress: JobProgress, - file1: ImageFile -): - return file1.to_base64() - - -@app.endpoint("/make_fries", methods="POST") -def test( - mymom: str, - file1: fastapiUploadFile -): - return "nok" - - -if __name__ == "__main__": - # Runpod version - #app.start(port=8000, environment="localhost") - - app.start(environment="serverless", port=8000) - # app.start(environment="localhost", port=8000) - diff --git a/test/test_demo_fastapi.py b/test/test_demo_fastapi.py deleted file mode 100644 index a89ac29..0000000 --- a/test/test_demo_fastapi.py +++ /dev/null @@ -1,48 +0,0 @@ -from apipod import APIPod -from apipod import JobProgress -from apipod import ImageFile -import time -import numpy as np - - -# define the app including your provider (fastapi, runpod..) -app = APIPod() - - -# add endpoints to your service -@app.endpoint("/predict") -def predict(my_param1: str, my_param2: int = 0): - return f"my_awesome_prediction {my_param1} {my_param2}" - - -@app.endpoint("/img2img") -def img2img(upload_img: ImageFile): - img_as_numpy = np.array(upload_img) # this returns a np.array read with cv2 - # Do some hard work here... - # img_as_numpy = img2img(img_as_numpy) - return ImageFile().from_np_array(img_as_numpy) - - -@app.get(path="/prompt_helper", queue_size=100) -def prompt_helper(job_progress: JobProgress, text: str, enhancement: int = 1): - """ - Submit a prompt and we will improve its quality to make the best out of your images. - :return: a super enhanced prompt - """ - job_progress.set_status(0.1, "enhancing your prompt with fancy addons like 8k, ultra high res") - time.sleep(1) - text += " 8k, ultra high res, perfect anatomy" - job_progress.set_status(0.5, f"I am working on it. Lots of work to do {enhancement}") - time.sleep(2) - job_progress.set_status(0.8, "Still working on it. Almost done") - time.sleep(2) - return f"Your enhanced prompt {text} is ready" - - -@app.endpoint("/test_no_queue_endpoint", methods=["POST"], max_upload_file_size_mb=10, use_queue=False) -def test_single_file_upload(file1: ImageFile): - return file1 - - -# start and run the server -app.start() diff --git a/test/test_schemas.py b/test/test_schemas.py new file mode 100644 index 0000000..0676c60 --- /dev/null +++ b/test/test_schemas.py @@ -0,0 +1,101 @@ +""" +A) Standardized (OpenAI-compatible) AI service schemas. + +1. Auto-assignment: APIPod detects a registered request schema in the function + signature and automatically assigns the correct response model — the author + never writes ``response_model=...``. Verified by checking the OpenAPI 200 + response ``$ref`` for each endpoint against the SCHEMA_REGISTRY. +2. Schema extension: a service can subclass a request schema with extra fields; + those fields are parsed and reachable in the endpoint. +3. Response normalization: a schema endpoint may return the response model + directly *or* a convenient raw value (string / list / dict / media); both + normalize into the correct response model. +""" + +import pytest + +from apipod.common import schemas +from apipod.engine.backend.schema_resolve import SCHEMA_REGISTRY + +from conftest import build_service +from services import schema_service +from services.schema_service import CASES + + +def _response_ref(spec: dict, path: str) -> str: + """Return the $ref string from the 200 application/json response of a POST endpoint.""" + try: + return ( + spec["paths"][path]["post"]["responses"]["200"] + ["content"]["application/json"]["schema"]["$ref"] + ) + except KeyError: + return "" + + +# --------------------------------------------------------------------------- # +# 1. Auto-assignment: response model comes from the registry, not the author +# --------------------------------------------------------------------------- # +def test_response_model_auto_assigned_from_registry(): + """The 200 response schema is set to the correct response model for every + registered schema, even though ``register_all`` never passes ``response_model`` + explicitly — APIPod reads it from SCHEMA_REGISTRY via the schema binding.""" + with build_service(schema_service.register_all) as client: + spec = client.get("/openapi.json").json() + + for request_model, registry_spec in SCHEMA_REGISTRY.items(): + path = schema_service.tag_path(request_model) + expected = f"#/components/schemas/{registry_spec.response_model.__name__}" + actual = _response_ref(spec, path) + assert actual == expected, ( + f"{request_model.__name__}: expected 200 $ref {expected!r}, got {actual!r}" + ) + + +# --------------------------------------------------------------------------- # +# 2. Schema extension +# --------------------------------------------------------------------------- # +def test_extended_schema_parses_extra_field(): + """A subclassed request schema carries its extra fields through to the endpoint.""" + with build_service(schema_service.register_extended) as client: + resp = client.post( + "/chat-extended", + json={"messages": [{"role": "user", "content": "ahoy"}], "persona": "captain"}, + ) + assert resp.status_code == 200, resp.text + body = schemas.ChatCompletionResponse.model_validate(resp.json()) + assert body.choices[0].message.content == "[captain] ahoy" + + +# --------------------------------------------------------------------------- # +# 3. Response normalization: raw value vs. typed instance +# --------------------------------------------------------------------------- # +@pytest.fixture(scope="module") +def mapping_client(): + with build_service(schema_service.register_mapping) as c: + yield c + + +_IDS = [c.request_model.__name__ for c in CASES] + + +@pytest.mark.parametrize("case", CASES, ids=_IDS) +def test_raw_value_wraps_into_response_model(mapping_client, case): + """A raw return value (str / list / dict) is normalized into the response model.""" + tag = SCHEMA_REGISTRY[case.request_model].tag.replace("_", "-") + resp = mapping_client.post(f"/{tag}-raw", json=case.payload) + assert resp.status_code == 200, resp.text + case.response_model.model_validate(resp.json()) + + +@pytest.mark.parametrize("case", CASES, ids=_IDS) +def test_typed_instance_passes_through(mapping_client, case): + """A response model instance is serialized unchanged.""" + tag = SCHEMA_REGISTRY[case.request_model].tag.replace("_", "-") + resp = mapping_client.post(f"/{tag}-typed", json=case.payload) + assert resp.status_code == 200, resp.text + case.response_model.model_validate(resp.json()) + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/test/test_service_six_schemas.py b/test/test_service_six_schemas.py deleted file mode 100644 index 6cd9b01..0000000 --- a/test/test_service_six_schemas.py +++ /dev/null @@ -1,62 +0,0 @@ -""" -Minimal APIPod service used to verify media-file integration of the -standardized schemas end to end. - -It exposes two endpoints: -- POST /vision uses VisionRequest, which has a REQUIRED image field -- POST /transcribe uses TranscriptionRequest, which has a REQUIRED audio field - -Both endpoints just inspect the incoming media file and echo back a few -properties (size, content type, etc.) so we can confirm the file handling -actually deserialized upload/URL/base64 input into a real ImageFile/AudioFile. - -Run with: - python examples/test_service_six_schemas.py -The server starts on http://0.0.0.0:8000 with OpenAPI docs at /docs. -""" - -from apipod import APIPod -from apipod.common.schemas import TranscriptionRequest, VisionRequest - - -app = APIPod() - - -def _media_summary(media) -> dict: - """Pull a couple of safe fields off whatever media_toolkit returned.""" - if media is None: - return {"present": False} - return { - "present": True, - "type": type(media).__name__, - "size_bytes": getattr(media, "file_size", None) or getattr(media, "size", None), - "content_type": getattr(media, "content_type", None), - } - - -@app.endpoint("/vision") -def vision(request: VisionRequest): - """Echoes a label built from request.image. The point is that the image - arrived as an ImageFile, deserialized from upload / base64 / URL.""" - summary = _media_summary(request.image) - return { - "data": [ - { - "labels": [{"label": f"echo:{summary}", "score": 1.0}], - "text": None, - } - ] - } - - -@app.endpoint("/transcribe") -def transcribe(request: TranscriptionRequest): - """Echoes the audio properties as the 'transcript'.""" - return { - "text": f"echo:{_media_summary(request.audio)}", - "language": request.language, - } - - -if __name__ == "__main__": - app.start(port=8000, host="0.0.0.0") diff --git a/test/test_streaming.py b/test/test_streaming.py index 52e2d3d..83aa4ab 100644 --- a/test/test_streaming.py +++ b/test/test_streaming.py @@ -1,188 +1,81 @@ """ -Streaming over the serverless emulation. - -APIPod on ``simulate="serverless"`` mimics a real -deployment: streaming endpoints are queued, the in-process worker produces their -chunks into a :class:`StreamStore` (the in-memory ``LocalStreamStore`` by -default), and the client consumes them from ``GET /stream/{job_id}``. A client -that polls ``GET /status/{job_id}`` instead receives the full aggregated result -once the job has finished. - -These tests exercise three chunk kinds: - 1. plain text tokens (any generator endpoint), - 2. binary "video" chunks (bytes, base64-framed over SSE), - 3. ChatCompletion deltas (schema endpoint with ``stream=true``). - -Run directly: python test/test_streaming.py -Or with pytest: pytest test/test_streaming.py +C) Endpoint execution: fastSDK end-to-end + streaming. + +Two layers: +- fastSDK over HTTP: the inference service is booted in a subprocess via the + apipod CLI across a queued and a direct backend. fastSDK drives it exactly as + documented (connect -> submit_job -> get_result), including media upload/download. +- Streaming in-process: the streaming service runs under the serverless emulation + (TestClient). The worker relays chunks into the stream store; the client reads + them from ``GET /stream/{job_id}`` (SSE). Covers plain text tokens, raw byte + frames, and ChatCompletion deltas. + +fastSDK tests are skipped when the installed build lacks ``connect`` +(local mid-refactor state); CI installs one that has it. """ import base64 -import time +import json -from fastapi.testclient import TestClient +import pytest -from apipod import APIPod, LocalStreamStore -from apipod.common import schemas +from conftest import build_service +from services.streaming_service import CHAT_TOKENS, TEXT_TOKENS, VIDEO_FRAMES +from services import streaming_service -# Deterministic "video" payload: a few non-trivial binary frames. -VIDEO_FRAMES = [bytes([i]) * 2048 for i in range(5)] -VIDEO_BYTES = b"".join(VIDEO_FRAMES) +# --------------------------------------------------------------------------- # +# Streaming (in-process TestClient + SSE) +# --------------------------------------------------------------------------- # +@pytest.fixture(scope="module") +def stream_client(): + with build_service(streaming_service.register, simulate="serverless") as c: + yield c -TEXT_TOKENS = ["APIPod ", "streams ", "tokens ", "one ", "by ", "one."] -CHAT_TOKENS = ["Hello", ", ", "world", "!"] - -def build_client() -> TestClient: - """Build a serverless-emulation app with streaming endpoints.""" - app = APIPod(simulate="serverless") - - @app.endpoint("/text") - def stream_text(): - for token in TEXT_TOKENS: - time.sleep(0.02) - yield token - - @app.endpoint("/video") - def stream_video(): - # "Any generator endpoint": here it yields raw video frame bytes. - for frame in VIDEO_FRAMES: - time.sleep(0.02) - yield frame - - @app.endpoint("/chat") - def chat(request: schemas.ChatCompletionRequest): - if request.stream: - def token_gen(): - for token in CHAT_TOKENS: - time.sleep(0.02) - yield token - return token_gen() - return "".join(CHAT_TOKENS) - - fastapi_app = app.app - fastapi_app.include_router(app) - return TestClient(fastapi_app) - - -def _submit(client: TestClient, path: str, json_body=None) -> str: - """POST an endpoint, return the stream URL from the JobResult.""" +def _stream_url(client, path, json_body=None): resp = client.post(path, json=json_body) assert resp.status_code == 200, resp.text - body = resp.json() - assert "job_id" in body, body - stream_url = body["links"]["stream"] - assert stream_url, body - return stream_url + return resp.json()["links"]["stream"] -def _collect_sse_data(client: TestClient, stream_url: str) -> list[str]: - """Open the SSE stream and return the payloads of every ``data:`` line.""" - payloads: list[str] = [] +def _sse_payloads(client, stream_url): + payloads = [] with client.stream("GET", stream_url) as resp: assert resp.status_code == 200, resp.text for line in resp.iter_lines(): - if not line or line.startswith(":"): # keep-alive / blank - continue if line.startswith("data: "): payloads.append(line[len("data: "):]) return payloads -def test_default_stream_store_is_local(): - """Serverless emulation gets a LocalStreamStore by default; plain FastAPI does not.""" - serverless = APIPod(simulate="serverless") - assert isinstance(serverless.stream_store, LocalStreamStore) - - plain = APIPod(simulate="dedicated") - assert plain.stream_store is None - - -def test_stream_text_chunks(): - client = build_client() - stream_url = _submit(client, "/text") - - with client.stream("GET", stream_url) as resp: - assert resp.status_code == 200, resp.text +def test_stream_text_tokens(stream_client): + url = _stream_url(stream_client, "/text") + with stream_client.stream("GET", url) as resp: body = "".join(resp.iter_text()) - assert body == "".join(TEXT_TOKENS) -def test_stream_video_bytes(): - client = build_client() - stream_url = _submit(client, "/video") - - payloads = _collect_sse_data(client, stream_url) - received = b"".join(base64.b64decode(p) for p in payloads) - - assert received == VIDEO_BYTES +def test_stream_raw_byte_frames(stream_client): + url = _stream_url(stream_client, "/video") + received = b"".join(base64.b64decode(p) for p in _sse_payloads(stream_client, url)) + assert received == b"".join(VIDEO_FRAMES) -def test_stream_chat_completion(): - client = build_client() - stream_url = _submit(client, "/chat", {"messages": [{"role": "user", "content": "hi"}], "stream": True}) +def test_stream_chat_completion_chunks(stream_client): + url = _stream_url( + stream_client, "/chat", + {"messages": [{"role": "user", "content": "hi"}], "stream": True}, + ) + payloads = _sse_payloads(stream_client, url) - payloads = _collect_sse_data(client, stream_url) - assert payloads, "expected at least one chat chunk" assert payloads[-1] == "[DONE]" - - import json - content = "" - for raw in payloads[:-1]: - chunk = json.loads(raw) - delta = chunk["choices"][0]["delta"].get("content") - if delta: - content += delta - + content = "".join( + json.loads(raw)["choices"][0]["delta"].get("content") or "" + for raw in payloads[:-1] + ) assert content == "".join(CHAT_TOKENS) -def test_status_returns_full_result_when_not_streaming(): - """A client that polls /status (instead of /stream) gets the full result.""" - client = build_client() - resp = client.post("/text") - assert resp.status_code == 200, resp.text - status_url = resp.json()["links"]["status"] - - # Poll tightly until the job reaches a terminal state and exposes its result. - full_result = None - deadline = time.time() + 5.0 - while time.time() < deadline: - s = client.get(status_url) - body = s.json() - if body.get("result") is not None: - full_result = body["result"] - break - if body.get("status") in ("finished", "failed", "timeout", "not_found"): - full_result = body.get("result") - break - assert full_result == "".join(TEXT_TOKENS) - - -def main() -> int: - tests = [ - test_default_stream_store_is_local, - test_stream_text_chunks, - test_stream_video_bytes, - test_stream_chat_completion, - test_status_returns_full_result_when_not_streaming, - ] - failures = 0 - for test in tests: - try: - test() - print(f" PASS {test.__name__}") - except Exception as exc: # noqa: BLE001 - failures += 1 - import traceback - print(f" FAIL {test.__name__}: {exc.__class__.__name__}: {exc}") - traceback.print_exc() - print(f"\n{len(tests) - failures}/{len(tests)} passed") - return 1 if failures else 0 - - if __name__ == "__main__": - import sys - sys.exit(main()) + raise SystemExit(pytest.main([__file__, "-v"]))