"""隔离数据库、文件和上游的接口回归测试；不使用真实 API Key。"""

import base64
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

_runtime = tempfile.TemporaryDirectory(prefix="ai-image-tests-")
os.environ["AI_IMAGE_DATA_DIR"] = str(Path(_runtime.name) / "data")
os.environ["AI_IMAGE_OUTPUT_DIR"] = str(Path(_runtime.name) / "output")
os.environ["AI_IMAGE_DB"] = str(Path(_runtime.name) / "data" / "test.db")
os.environ["AI_IMAGE_SECRET_KEY"] = "isolated-test-key"

import httpx
from fastapi.testclient import TestClient

from main import dispatcher
from app import config, db
from app.providers import ImageResult, ProviderAdapter, ProviderError
from app.services import generation, providers

PNG = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+a4l8AAAAASUVORK5CYII=")


class AppTests(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(dispatcher)
        with db.session() as conn:
            for table in ("generations", "models", "providers"):
                conn.execute(f"DELETE FROM {table}")
        self.provider = dict(id="test", name="测试供应商", adapter="openai_compat",
                             base_url="https://upstream.invalid/v1", api_key="test-secret-key")
        self.client.post("/api/providers", json=self.provider).raise_for_status()
        providers.upsert_models("test", ["test-image"])
        self.payload = dict(provider_id="test", model="test-image", prompt="测试图片")

    def upstream(self, method, path, **kwargs):
        if path == "/models":
            return httpx.Response(200, json={"data": [{"id": "test-image"}, {"id": "chat"}]})
        return httpx.Response(200, json={"data": [{"b64_json": base64.b64encode(PNG).decode()}]})

    def test_pages_and_read_endpoints(self):
        for url in ("/", "/create.html", "/providers.html", "/models.html", "/records.html",
                    "/skills.html", "/api.js", "/components.js", "/styles.css", "/health",
                    "/api/providers", "/api/models", "/api/generations", "/api/galleries",
                    "/api/skills", "/api/skills/raw"):
            with self.subTest(url=url):
                self.assertEqual(self.client.get(url).status_code, 200)

    def test_secret_is_encrypted_and_empty_update_keeps_key(self):
        result = self.client.post("/api/providers", json={**self.provider, "api_key": ""})
        self.assertNotIn("test-secret-key", result.text)
        self.assertNotEqual(providers.get_provider_row("test")["api_key"], "test-secret-key")
        self.assertEqual(providers.provider_config("test").api_key, "test-secret-key")

    def test_discover_preserves_disabled_model(self):
        providers.set_model_enabled("test:test-image", False)
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream):
            result = self.client.post("/api/providers/test/discover")
        self.assertEqual(result.json()["matched"], ["test-image"])
        self.assertFalse(providers.get_model("test:test-image")["enabled"])

    def test_generation_round_trip(self):
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream):
            result = self.client.post("/api/generate", json=self.payload)
        self.assertEqual(result.status_code, 200, result.text)
        image = result.json()["images"][0]
        self.assertEqual(self.client.get(image["url"]).content, PNG)
        self.assertEqual(self.client.get("/api/generations").json()[0]["status"], "success")
        self.assertTrue(self.client.get("/api/galleries/user").json())
        self.assertEqual(self.client.delete("/api/galleries/user/" + image["filename"]).status_code, 200)
        self.assertEqual(self.client.get(image["url"]).status_code, 404)

    def test_gallery_prompt_and_legacy_image(self):
        prompt = "第一行提示词\n<script>保持原样显示</script>"
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream):
            result = self.client.post("/api/generate", json={**self.payload, "prompt": prompt,
                                                           "profile_name": "prompt-preview"})
        result.raise_for_status()
        filename = result.json()["images"][0]["filename"]
        (config.OUTPUT_DIR / "prompt-preview" / "legacy.png").write_bytes(PNG)
        items = {item["filename"]: item for item in self.client.get("/api/galleries/prompt-preview").json()}
        self.assertEqual(items[filename]["prompt"], prompt)
        self.assertEqual(items["legacy.png"]["prompt"], "")

    def test_disabled_or_unknown_model_never_calls_upstream(self):
        for model, enabled, status in (("missing-image", True, 404), ("test-image", False, 400)):
            providers.set_model_enabled("test:test-image", enabled)
            with patch.object(ProviderAdapter, "_request", side_effect=self.upstream) as request:
                response = self.client.post("/api/generate", json={**self.payload, "model": model})
            self.assertEqual(response.status_code, status)
            request.assert_not_called()

    def test_disabled_provider_never_calls_upstream(self):
        self.client.post("/api/providers", json={**self.provider, "enabled": False})
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream) as request:
            self.assertEqual(self.client.post("/api/generate", json=self.payload).status_code, 400)
        request.assert_not_called()

    def test_unsupported_reference_image_is_rejected(self):
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream) as request:
            result = self.client.post("/api/generate", json={**self.payload, "images": ["https://example.invalid/ref.png"]})
        self.assertEqual(result.status_code, 400)
        request.assert_not_called()

    def test_failed_upstream_is_recorded(self):
        with patch.object(ProviderAdapter, "_request", side_effect=ProviderError("上游失败")):
            self.assertEqual(self.client.post("/api/generate", json=self.payload).status_code, 502)
        self.assertEqual(generation.list_generations()[0]["status"], "failed")

    def test_empty_and_malformed_image_responses_are_recorded_as_failures(self):
        for data in ([], [{"b64_json": "%%%"}]):
            with patch.object(ProviderAdapter, "_request", return_value=httpx.Response(200, json={"data": data})):
                result = self.client.post("/api/generate", json=self.payload)
            self.assertEqual(result.status_code, 502)
            self.assertEqual(generation.list_generations()[0]["status"], "failed")

    def test_download_failure_is_recorded(self):
        response = httpx.Response(200, json={"data": [{"url": "https://example.invalid/image.png"}]})
        with patch.object(ProviderAdapter, "_request", return_value=response), patch.object(
            ProviderAdapter, "download", side_effect=ProviderError("下载失败")
        ):
            self.assertEqual(self.client.post("/api/generate", json=self.payload).status_code, 502)
        self.assertEqual(generation.list_generations()[0]["error"], "下载失败")

    def test_gallery_delete_only_changes_its_profile(self):
        for profile in ("first", "second"):
            folder = config.OUTPUT_DIR / profile
            folder.mkdir(exist_ok=True)
            (folder / "same.png").write_bytes(PNG)
            generation.record_generation(provider_id="test", provider_name="测试", model="test-image",
                profile_name=profile, prompt="test", size="1024x1024", ratio="", n=1,
                status="success", error="", files=["same.png"], duration_ms=1)
        self.client.delete("/api/galleries/first/same.png").raise_for_status()
        other = next(g for g in generation.list_generations() if g["profile_name"] == "second")
        self.assertEqual(other["files"], ["same.png"])

    def test_missing_provider_models_returns_404(self):
        result = self.client.post("/api/providers/missing/models", json={"model_ids": ["test-image"]})
        self.assertEqual(result.status_code, 404)

    def test_model_ids_with_slashes(self):
        providers.upsert_models("test", ["vendor/test-image"])
        result = self.client.patch("/api/models/test%3Avendor%2Ftest-image", json={"enabled": False})
        self.assertEqual(result.status_code, 200)

    def test_invalid_generation_inputs(self):
        for extra in ({"n": 0}, {"n": 5}, {"prompt": " "}, {"profile_name": "../data"},
                      {"size": "0x1024"}, {"size": "unknown"}):
            self.assertEqual(self.client.post("/api/generate", json={**self.payload, **extra}).status_code, 422)

    def test_gallery_rejects_invalid_paths(self):
        for profile in ("..", "bad\\path"):
            self.assertEqual(self.client.get("/api/galleries/" + profile.replace(".", "%2E")).status_code, 400)

    def test_delete_provider_cascades_models(self):
        self.client.delete("/api/providers/test").raise_for_status()
        self.assertEqual(providers.list_models(), [])

    def test_agnes_reference_image_and_size_translation(self):
        self.client.post("/api/providers", json={**self.provider, "adapter": "agnes"}).raise_for_status()
        providers.upsert_models("test", ["test-image"])
        self.assertTrue(providers.get_model("test:test-image")["enabled"])
        self.assertTrue(providers.get_model("test:test-image")["supports_image_input"])
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream) as request:
            result = self.client.post("/api/generate", json={**self.payload, "size": "2048x1152",
                                                            "images": ["data:image/png;base64," + base64.b64encode(PNG).decode()]})
        self.assertEqual(result.status_code, 200, result.text)
        body = request.call_args.kwargs["json_body"]
        self.assertEqual((body["size"], body["ratio"]), ("2K", "16:9"))
        self.assertTrue(body["extra_body"]["image"])
        self.assertEqual(result.json()["ratio"], "16:9")

    def test_tier_is_translated_for_openai(self):
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream) as request:
            result = self.client.post("/api/generate", json={**self.payload, "size": "2k"})
        self.assertEqual(result.status_code, 200)
        self.assertEqual(request.call_args.kwargs["json_body"]["size"], "2048x2048")

    def test_discovery_credentials_are_validated(self):
        for extra in ({"adapter": "unknown"}, {"base_url": "file:///tmp"}):
            response = self.client.post("/api/providers/discover", json={**self.provider, **extra})
            self.assertEqual(response.status_code, 422)

    def test_agent_service_validates_count(self):
        with self.assertRaises(ProviderError):
            generation.generate(**self.payload, n=100)

    def test_partial_save_failure_rolls_back_images_and_records_failure(self):
        with patch("app.services.generation.get_adapter") as get_adapter:
            adapter = get_adapter.return_value
            adapter.generate.return_value = [ImageResult(data=PNG), ImageResult()]
            result = self.client.post("/api/generate", json=self.payload)
        self.assertEqual(result.status_code, 502)
        record = generation.list_generations()[0]
        self.assertEqual((record["status"], record["files"]), ("failed", []))

    def test_agent_discovery_only_returns_enabled_models(self):
        providers.set_model_enabled("test:test-image", False)
        with patch.object(ProviderAdapter, "_request", side_effect=self.upstream):
            self.assertEqual(providers.discover("test", require_enabled=True)["matched"], [])

    def test_malformed_upstream_payloads_return_controlled_errors(self):
        for payload in ([], {"data": {"unexpected": "object"}}):
            with patch.object(ProviderAdapter, "_request", return_value=httpx.Response(200, json=payload)):
                result = self.client.post("/api/generate", json=self.payload)
                self.assertEqual(result.status_code, 502)
                self.assertEqual(generation.list_generations()[0]["status"], "failed")
                self.assertEqual(self.client.post("/api/providers/test/discover").status_code, 502)


if __name__ == "__main__":
    unittest.main()