mirror of
https://github.com/AUTOMATIC1111/stable-diffusion-webui.git
synced 2026-03-22 14:20:39 -07:00
961297f09d
- Add test_api_runner_contract.py: proves API txt2img path invokes ProcessingRunner - No routing changes: API continues to call process_images (orchestration boundary) - Monkeypatch CI env + runner.run; call API method directly - M14_plan.md, M14_toolcalls.md: governance docs Phase III — Runner & Service Boundary Made-with: Cursor
64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
"""M14 contract test: API txt2img path uses ProcessingRunner.
|
|
|
|
Verifies that the API execution path flows through process_images → runner,
|
|
not direct process_images_inner calls. No routing changes; verification only.
|
|
"""
|
|
from threading import Lock
|
|
|
|
import pytest
|
|
|
|
|
|
def test_api_txt2img_uses_runner(monkeypatch, initialize):
|
|
"""API txt2img path invokes ProcessingRunner when process_images is called."""
|
|
from fastapi import FastAPI
|
|
|
|
from modules.api.api import Api
|
|
from modules.api import models
|
|
from modules.processing import Processed
|
|
from modules.runtime.runner import ProcessingRunner
|
|
|
|
called = {"run": False}
|
|
|
|
# Force real execution path (bypass CI early return)
|
|
monkeypatch.setenv("CI", "false")
|
|
|
|
# Patch runner to track invocation
|
|
original_run = ProcessingRunner.run
|
|
|
|
def tracking_run(self, request):
|
|
called["run"] = True
|
|
return original_run(self, request)
|
|
|
|
monkeypatch.setattr(ProcessingRunner, "run", tracking_run)
|
|
|
|
# Mock process_images_inner to avoid full pipeline
|
|
def fake_inner(proc):
|
|
return Processed(proc, [], seed=-1, info="", comments="")
|
|
|
|
import modules.processing as proc_mod
|
|
|
|
monkeypatch.setattr(proc_mod, "process_images_inner", fake_inner)
|
|
|
|
# Mock model reload and token merging to avoid model/device ops
|
|
import modules.sd_models as sd_models_mod
|
|
|
|
monkeypatch.setattr(sd_models_mod, "reload_model_weights", lambda: None)
|
|
monkeypatch.setattr(sd_models_mod, "apply_token_merging", lambda m, r: None)
|
|
|
|
# Call API method directly (no HTTP)
|
|
app = FastAPI()
|
|
api = Api(app, Lock())
|
|
|
|
req = models.StableDiffusionTxt2ImgProcessingAPI(
|
|
prompt="test",
|
|
steps=1,
|
|
width=64,
|
|
height=64,
|
|
)
|
|
|
|
result = api.text2imgapi(req)
|
|
|
|
assert called["run"] is True
|
|
assert result is not None
|
|
assert hasattr(result, "images")
|
|
assert hasattr(result, "info")
|