xiaoxia-saas/tests/unit/test_2035_coverage.py

"""Additional unit tests to hit uncovered lines for diff-coverage >=60%."""
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from packages.shared.ai_client import DoubaoClient


class _FakeSettings:
    doubao_api_key = "test-key"
    doubao_model = "test-model"
    doubao_fast_model = "test-fast-model"
    doubao_base_url = "https://ark.cn-beijing.volces.com/api/v3"
    doubao_timeout = 10
    doubao_max_retries = 0
    doubao_vision_model = "test-vision"
    doubao_vision_lite_model = "test-vision-lite"
    doubao_vision_use_lite = False
    doubao_embedding_model = "test-embedding"


def _make_client(api_key: str = "test-key") -> DoubaoClient:
    with patch("packages.shared.ai_client.get_shared_settings", return_value=_FakeSettings()):
        c = DoubaoClient()
    c.api_key = api_key
    c.max_retries = 0
    return c


class TestDoubaoClientEmbedText:
    def test_no_api_key_returns_none(self):
        c = _make_client(api_key="")
        assert c.embed_text("hello") is None

    def test_empty_text_returns_none(self):
        c = _make_client()
        assert c.embed_text("") is None
        assert c.embed_text("   ") is None

    def test_none_text_returns_none(self):
        c = _make_client()
        assert c.embed_text(None) is None

    @patch("packages.shared.ai_client.httpx.post")
    def test_successful_embedding(self, mock_post):
        mock_resp = MagicMock()
        mock_resp.json.return_value = {"data": [{"embedding": [0.1, 0.2, 0.3]}]}
        mock_resp.raise_for_status = MagicMock()
        mock_post.return_value = mock_resp
        c = _make_client()
        result = c.embed_text("hello world")
        assert result == [0.1, 0.2, 0.3]
        mock_post.assert_called_once()

    @patch("packages.shared.ai_client.httpx.post")
    def test_malformed_response_returns_none(self, mock_post):
        mock_resp = MagicMock()
        mock_resp.json.return_value = {"data": []}
        mock_resp.raise_for_status = MagicMock()
        mock_post.return_value = mock_resp
        c = _make_client()
        assert c.embed_text("hello") is None

    @patch("packages.shared.ai_client.httpx.post", side_effect=Exception("network error"))
    def test_network_error_returns_none(self, mock_post):
        c = _make_client()
        assert c.embed_text("hello") is None

    def test_is_available_with_key(self):
        c = _make_client(api_key="sk-xxx")
        assert c.is_available is True

    def test_is_available_without_key(self):
        c = _make_client(api_key="")
        assert c.is_available is False


# --- 2. _infer_expected_categories ---
_GEN_TASKS_PATH = Path(__file__).resolve().parents[2] / "apps/api/app/api/routes/generation_tasks.py"


def _load_infer_func():
    src = _GEN_TASKS_PATH.read_text()
    start = src.index("# #2035：文案关键词")
    end = src.index("from packages.middleware")
    code = src[start:end]
    ns: dict = {}
    exec(code, ns)
    return ns["_infer_expected_categories"]


_infer_expected_categories = _load_infer_func()


class TestInferExpectedCategories:
    def test_none_returns_none(self):
        assert _infer_expected_categories(None) is None
        assert _infer_expected_categories(set()) is None

    def test_product_keyword_matches(self):
        cats = _infer_expected_categories({"产品展示"})
        assert cats is not None
        assert "product" in cats

    def test_scenic_keyword_matches(self):
        cats = _infer_expected_categories({"户外风景"})
        assert cats is not None
        assert "scenic" in cats

    def test_food_keyword_matches(self):
        cats = _infer_expected_categories({"美食制作"})
        assert cats is not None
        assert "food" in cats

    def test_no_match_returns_none(self):
        assert _infer_expected_categories({"抽象概念xyz"}) is None


# --- 3. parse_vision_response edge cases ---
from packages.domain.atom_clip_tagger import parse_vision_response


class TestParseVisionResponseEdgeCases:
    def test_person_count_type_error_defaults_zero(self):
        text = json.dumps({
            "scene": [], "objects": [], "action": [], "shot": "", "has_text": False,
            "person_count": "not-an-int", "text_content": "", "caption": "x",
        })
        r = parse_vision_response(text)
        assert r["person_count"] == 0

    def test_person_count_out_of_range_clamped(self):
        text = json.dumps({
            "scene": [], "objects": [], "action": [], "shot": "", "has_text": False,
            "person_count": 10, "text_content": "", "caption": "x",
        })
        r = parse_vision_response(text)
        assert r["person_count"] == 3

    def test_person_count_negative_clamped(self):
        text = json.dumps({
            "scene": [], "objects": [], "action": [], "shot": "", "has_text": False,
            "person_count": -5, "text_content": "", "caption": "x",
        })
        r = parse_vision_response(text)
        assert r["person_count"] == 0

    def test_text_content_non_string_defaults_empty(self):
        text = '{"scene":[],"objects":[],"action":[],"shot":"","has_text":true,"person_count":0,"text_content":123,"caption":"x"}'
        r = parse_vision_response(text)
        assert r["text_content"] == ""

    def test_caption_truncation_at_80(self):
        long_caption = "描" * 100
        text = json.dumps({
            "scene": [], "objects": [], "action": [], "shot": "", "has_text": False,
            "person_count": 0, "text_content": "", "caption": long_caption,
        })
        r = parse_vision_response(text)
        assert len(r["caption"]) == 80


# --- 4. smart_match normalize_tag ---
from packages.domain.smart_match import normalize_tag


class TestNormalizeTagEdge:
    def test_none_returns_empty(self):
        assert normalize_tag(None) == ""

    def test_non_string_converted(self):
        assert normalize_tag(123) == "123"

    def test_strip_and_lower(self):
        assert normalize_tag("  FOO Bar  ") == "foo bar"


# --- 5. narrative_match non-dict clip_tags skip ---
from packages.domain.narrative_match import match_assets_by_script_tags


@dataclass
class _FA:
    id: str
    tags: list


class TestNarrativeMatchNonDictClipTags:
    def test_non_dict_clip_tags_are_skipped(self):
        a1 = _FA("a1", tags=[])
        clip_map = {"a1": [None, "bad", {"scene": ["工厂"], "objects": [], "action": []}, 123]}
        matched, unmatched = match_assets_by_script_tags(
            [a1], script_tags=["工厂"], clip_ai_tags_by_asset=clip_map
        )
        assert [a.id for a in matched] == ["a1"]


# --- 6. update_caption_embedding ---
class _FakeSession:
    def __init__(self, rows_found: int = 1):
        self.rows_found = rows_found
        self.commits = 0
        self.updates = []

    def query(self, model):
        return _FQuery(self)

    def commit(self):
        self.commits += 1


class _FQuery:
    def __init__(self, session):
        self.session = session

    def filter(self, *a, **kw):
        return self

    def update(self, upd):
        self.session.updates.append(upd)
        return self.session.rows_found


class TestUpdateCaptionEmbedding:
    def _make_repo(self, session):
        from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import SQLAlchemyAssetAtomClipRepository
        repo = SQLAlchemyAssetAtomClipRepository.__new__(SQLAlchemyAssetAtomClipRepository)
        repo.session = session
        return repo

    def test_updates_both_caption_and_embedding(self):
        s = _FakeSession(rows_found=1)
        repo = self._make_repo(s)
        ok = repo.update_caption_embedding("c1", "new caption", [0.1, 0.2])
        assert ok is True
        assert s.commits == 1
        assert s.updates[0]["caption"] == "new caption"
        assert s.updates[0]["embedding"] == [0.1, 0.2]

    def test_only_caption_update(self):
        s = _FakeSession(rows_found=1)
        repo = self._make_repo(s)
        ok = repo.update_caption_embedding("c1", "cap", None)
        assert ok is True
        assert "embedding" not in s.updates[0]
        assert s.updates[0]["caption"] == "cap"

    def test_no_update_when_both_none(self):
        s = _FakeSession()
        repo = self._make_repo(s)
        ok = repo.update_caption_embedding("c1", None, None)
        assert ok is False
        assert s.commits == 0
        assert s.updates == []

    def test_returns_false_when_row_not_found(self):
        s = _FakeSession(rows_found=0)
        repo = self._make_repo(s)
        ok = repo.update_caption_embedding("c1", "x", [0.1])
        assert ok is False