vent_agent_2.0/tests/test_algorithms.py

"""真实图片计算、HTTP 契约及计算隔离回归。"""
import io
import json
from pathlib import Path
import subprocess
import zipfile
import threading

import cv2
import numpy as np
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from algorithms.characteristic_curve_recognition import service
from api.algorithm_routes import router


@pytest.fixture
def client(tmp_path, monkeypatch):
    monkeypatch.setattr("config.ALGORITHM_JOB_DIR", str(tmp_path / "jobs"))
    app = FastAPI()
    app.include_router(router, prefix="/api")
    with TestClient(app) as value:
        yield value


@pytest.fixture
def chart():
    image = np.full((320, 420, 3), 255, np.uint8)
    cv2.line(image, (45, 25), (45, 275), (0, 0, 0), 2)
    cv2.line(image, (45, 275), (390, 275), (0, 0, 0), 2)
    for offset in (0, 30):
        points = np.array([[x, int(45 + offset + .0012 * (x - 50) ** 2)]
                           for x in range(60, 380)], np.int32)
        cv2.polylines(image, [points], False, (0, 0, 0), 2)
    return cv2.imencode(".png", image)[1].tobytes()


def test_real_json_and_zip(client, chart):
    response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("../../图.png", chart)})
    assert response.status_code == 200, response.text
    result = response.json()
    assert result["algorithm"] == "characteristic_curve_recognition"
    assert result["model"]["execution_status"] == "completed"
    assert result["model"]["quality_status"] == "needs_review"
    assert result["model"]["physical_coordinate_status"] == "not_calibrated"
    curves = result["model"]["curves"]
    assert len(curves) == 2
    assert all(1 <= curve["segment_count"] <= 3 for curve in curves)
    assert all(not curve["model_samples_used_as_observations"] for curve in curves)
    assert result["geometry"]["grid"]["cells"]
    assert "vent-characteristic-curve-" not in response.text

    response = client.post("/api/algorithms/characteristic-curve-recognition?output=zip&detailed=true",
                           files={"file": ("图.png", chart)})
    assert response.status_code == 200
    assert response.headers["content-type"] == "application/zip"
    with zipfile.ZipFile(io.BytesIO(response.content)) as archive:
        names = archive.namelist()
        assert {"index.html", "result.json", "final_overlay.png", "final_redraw.png",
                "review/pipeline_overview.png"} <= set(names)
        assert all(not name.startswith("/") and ".." not in Path(name).parts for name in names)
        assert json.loads(archive.read("final_model.json")) == result["model"]


@pytest.mark.parametrize("content,code", [(b"", "invalid_image"),
    (b"hello", "invalid_image"), (b"\x89PNG\r\n\x1a\ninvalid", "invalid_image")])
def test_invalid_image(client, content, code):
    response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("a.png", content)})
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == code


def test_blank_chart(client):
    content = cv2.imencode(".png", np.full((200, 200, 3), 255, np.uint8))[1].tobytes()
    response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("a.png", content)})
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "detection_failed"


@pytest.mark.parametrize("shape,extension", [((16, 48, 3), ".jpg"), ((4001, 4000), ".png")])
def test_image_dimensions(client, shape, extension):
    content = cv2.imencode(extension, np.full(shape, 255, np.uint8))[1].tobytes()
    response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("a" + extension, content)})
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "invalid_image"


def test_limits_and_parameters(client, chart, monkeypatch):
    monkeypatch.setattr("api.algorithm_routes.MAX_UPLOAD_BYTES", 8)
    monkeypatch.setattr(service, "MAX_UPLOAD_BYTES", 8)
    response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("a.png", chart)})
    assert response.status_code == 413
    assert client.post("/api/algorithms/characteristic-curve-recognition").status_code == 422
    assert client.post("/api/algorithms/characteristic-curve-recognition?output=exe", files={"file": ("a.png", chart)}).status_code == 422


def test_busy(client, chart, monkeypatch):
    monkeypatch.setattr(service, "_slots", threading.BoundedSemaphore(1))
    assert service._slots.acquire(blocking=False)
    try:
        response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("a.png", chart)})
        assert response.status_code == 429
        assert response.headers["retry-after"] == "5"
    finally:
        service._slots.release()


def test_timeout_cleans_temp_and_releases_slot(client, chart, monkeypatch):
    paths = []
    def timeout(command, **kwargs):
        paths.append(Path(command[4]).parent)
        assert Path(command[4]).is_file()
        raise subprocess.TimeoutExpired(command, kwargs["timeout"])
    monkeypatch.setattr(service.subprocess, "run", timeout)
    for _ in range(2):
        response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("a.png", chart)})
        assert response.status_code == 504
    assert all(not path.exists() for path in paths)


def test_missing_dependencies(client, chart, monkeypatch):
    monkeypatch.setattr(service.importlib.util, "find_spec", lambda name: None)
    response = client.post("/api/algorithms/characteristic-curve-recognition", files={"file": ("a.png", chart)})
    assert response.status_code == 503


def test_registered_in_main_app():
    from main import app
    assert "/api/algorithms/characteristic-curve-recognition" in app.openapi()["paths"]