cmshoppe/tests/test_appconfig.py

import json
import os
import sys
import unittest
from unittest import mock

sys.path.insert(0, os.path.dirname(__file__))

from _helpers import TempDirMixin

from app import appconfig


class AppConfigTests(TempDirMixin, unittest.TestCase):
    def test_config_load_update_and_response_timeout(self):
        with self.make_temp_dir() as temp_dir:
            config_path = os.path.join(temp_dir, "config.json")

            config = appconfig.load_config(config_path)
            self.assertTrue(os.path.exists(config_path))
            self.assertEqual("images", appconfig.image_dir(config))
            self.assertEqual(240, appconfig.response_timeout(config))
            self.assertFalse(appconfig.ai_config(config)["generate_cover"])

            updated = appconfig.update_config(
                {"ai": {"resolution": "2k"}},
                path=config_path,
            )
            self.assertEqual(360, appconfig.response_timeout(updated))
            self.assertEqual((9222, 9260), appconfig.debug_port_range(updated))

        self.assert_removed(temp_dir)

    def test_cmhub_defaults_old_config_and_key_helper(self):
        with self.make_temp_dir() as temp_dir:
            config_path = os.path.join(temp_dir, "config.json")
            cmhub_path = os.path.join(temp_dir, "config", "cmhub.json")

            config = appconfig.load_config(config_path)
            ai = appconfig.ai_config(config)
            self.assertEqual("direct", ai["backend"])
            self.assertEqual("direct", appconfig.ai_backend(config))
            self.assertEqual("", appconfig.cmhub_config(config)["base_url"])
            self.assertFalse(os.path.exists(cmhub_path))
            self.assertEqual({"api_key": ""}, appconfig.load_cmhub_config(cmhub_path))

            saved = appconfig.save_cmhub_config(
                {"api_key": "sk-cmhub-123456"},
                path=cmhub_path,
            )
            self.assertEqual("sk-cmhub-123456", saved["api_key"])
            self.assertEqual("sk-cmhub-123456", appconfig.get_cmhub_api_key(cmhub_path))
            self.assertEqual("sk-c***3456", appconfig.get_cmhub_api_key(cmhub_path, masked=True))

            with open(config_path, "w", encoding="utf-8") as fh:
                json.dump({"ai": {"resolution": "512"}}, fh)
            migrated = appconfig.load_config(config_path)
            self.assertEqual("direct", appconfig.ai_config(migrated)["backend"])
            self.assertEqual(180, appconfig.response_timeout(migrated))

        self.assert_removed(temp_dir)

    def test_config_rejects_sensitive_fields(self):
        with self.make_temp_dir() as temp_dir:
            config_path = os.path.join(temp_dir, "config.json")

            with self.assertRaises(appconfig.ConfigError):
                appconfig.save_config({"api_key": "secret"}, path=config_path)

            with self.assertRaises(appconfig.ConfigError):
                appconfig.save_config(
                    {"ai": {"provider_token": "secret"}},
                    path=config_path,
                )

        self.assert_removed(temp_dir)

    def test_ai_models_crud_filter_mask_and_get_model(self):
        with self.make_temp_dir() as temp_dir:
            models_path = os.path.join(temp_dir, "ai_models.json")

            models = appconfig.list_ai_models(path=models_path)
            self.assertEqual({"text", "image"}, {model["category"] for model in models})
            self.assertTrue(all("api_key_set" in model for model in models))

            appconfig.add_ai_model(
                {
                    "name": "Text 2",
                    "category": "text",
                    "enabled": True,
                    "url": "https://example.invalid/v1/chat/completions",
                    "model": "demo-model",
                    "api_key": "sk-1234567890",
                    "api_type": "chat",
                    "connect_timeout_seconds": 1,
                    "extra_body": {"temperature": 0},
                },
                path=models_path,
            )

            text_models = appconfig.list_ai_models("text", path=models_path)
            self.assertEqual(2, len(text_models))
            self.assertEqual("sk-1***7890", text_models[-1]["api_key"])
            self.assertTrue(text_models[-1]["api_key_set"])

            private_model = appconfig.get_model("Text 2", path=models_path)
            self.assertEqual("sk-1234567890", private_model["api_key"])
            self.assertEqual({"temperature": 0}, private_model["extra_body"])

            appconfig.update_ai_model(
                "Text 2",
                path=models_path,
                name="Text 3",
                enabled=False,
            )
            self.assertFalse(appconfig.get_model("Text 3", path=models_path)["enabled"])

        self.assert_removed(temp_dir)

    def test_ai_model_constraints_and_connection_validation(self):
        with self.make_temp_dir() as temp_dir:
            models_path = os.path.join(temp_dir, "ai_models.json")
            appconfig.list_ai_models(path=models_path)

            with self.assertRaises(appconfig.ConfigError):
                appconfig.add_ai_model(
                    {
                        "name": "GPT-5.5 文本",
                        "category": "text",
                        "enabled": True,
                        "api_type": "chat",
                        "connect_timeout_seconds": 30,
                    },
                    path=models_path,
                )

            with self.assertRaises(appconfig.ConfigError):
                appconfig.delete_ai_model("Nano Banana 2", path=models_path)

            result = appconfig.test_ai_model("GPT-5.5 文本", path=models_path)
            self.assertFalse(result["ok"])
            self.assertIn("url", result["error"])
            self.assertIn("model", result["error"])
            self.assertIn("api_key", result["error"])

        self.assert_removed(temp_dir)


    def test_model_request_url_accepts_base_and_full_endpoint(self):
        self.assertEqual(
            "https://api.example.com/v1/chat/completions",
            appconfig.model_request_url(
                {"url": "https://api.example.com/v1", "api_type": "chat"}
            ),
        )
        self.assertEqual(
            "https://openrouter.ai/api/v1/chat/completions",
            appconfig.model_request_url(
                {"url": "https://openrouter.ai/api/v1/", "api_type": "auto"}
            ),
        )
        self.assertEqual(
            "https://api.example.com/v1/chat/completions?region=tw",
            appconfig.model_request_url(
                {"url": "https://api.example.com/v1?region=tw", "api_type": "chat"}
            ),
        )
        self.assertEqual(
            "https://api.example.com/v1/chat/completions",
            appconfig.model_request_url(
                {
                    "url": "https://api.example.com/v1/chat/completions",
                    "api_type": "chat",
                }
            ),
        )
        self.assertEqual(
            "https://api.example.com/v1/images/edits",
            appconfig.model_request_url(
                {"url": "https://api.example.com/v1", "api_type": "images_edits"}
            ),
        )
        self.assertEqual(
            "https://api.example.com/custom/generate",
            appconfig.model_request_url(
                {"url": "https://api.example.com/custom/generate", "api_type": "chat"}
            ),
        )

    def test_ai_model_test_uses_resolved_base_url(self):
        with self.make_temp_dir() as temp_dir:
            models_path = os.path.join(temp_dir, "ai_models.json")
            appconfig.save_ai_models_config(
                {
                    "models": [
                        {
                            "name": "Text",
                            "category": "text",
                            "enabled": True,
                            "url": "https://api.example.com/v1",
                            "model": "text-model",
                            "api_key": "sk-text-secret",
                            "api_type": "chat",
                            "connect_timeout_seconds": 1,
                            "extra_body": {},
                        },
                        {
                            "name": "Image",
                            "category": "image",
                            "enabled": True,
                            "url": "https://api.example.com/v1/chat/completions",
                            "model": "image-model",
                            "api_key": "sk-image-secret",
                            "api_type": "auto",
                            "connect_timeout_seconds": 1,
                            "extra_body": {},
                        },
                    ]
                },
                path=models_path,
            )
            calls = []

            class Response:
                status = 200

                def __enter__(self):
                    return self

                def __exit__(self, exc_type, exc, tb):
                    return False

                def read(self, size=-1):
                    return b"{}"

            def fake_urlopen(request, timeout=None):
                calls.append((request, timeout))
                return Response()

            with mock.patch("app.appconfig.urllib.request.urlopen", side_effect=fake_urlopen):
                result = appconfig.test_ai_model("Text", path=models_path)

            self.assertTrue(result["ok"])
            self.assertEqual(200, result["status"])
            self.assertEqual(
                "https://api.example.com/v1/chat/completions",
                calls[0][0].full_url,
            )
            self.assertEqual(1, calls[0][1])

        self.assert_removed(temp_dir)
    def test_sanitize_for_log_masks_secret_fields(self):
        payload = {
            "name": "demo",
            "api_key": "sk-1234567890",
            "nested": {
                "password": "account-secret",
                "items": [
                    {"provider_token": "token-secret"},
                    {"value": "safe"},
                    {"api_key": {"value": "nested-secret"}},
                ],
            },
        }

        sanitized = appconfig.sanitize_for_log(payload)

        self.assertEqual("demo", sanitized["name"])
        self.assertEqual("sk-1***7890", sanitized["api_key"])
        self.assertEqual("acco***cret", sanitized["nested"]["password"])
        self.assertEqual("toke***cret", sanitized["nested"]["items"][0]["provider_token"])
        self.assertEqual("safe", sanitized["nested"]["items"][1]["value"])
        self.assertEqual("***", sanitized["nested"]["items"][2]["api_key"])
        self.assertEqual("sk-1234567890", payload["api_key"])


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