copykar/tests/test_prompts.py

"""The rewrite prompt embeds a literal JSON example, which str.format() must not choke on."""
import sys

sys.path.insert(0, "/app")

from services.ai_processor import build_rewrite_system_prompt, build_verify_system_prompt, AIProcessor
from core.llm import LLMClient
from db.models import TargetChannel
import asyncio


def test_prompt_builds_with_literal_json_braces():
    prompt = build_rewrite_system_prompt(
        channel_title="کانال تست",
        personality="لحن دوستانه",
        custom_footer="@test_channel",
        language="fa",
        has_media=False,
    )
    assert "کانال تست" in prompt
    assert "لحن دوستانه" in prompt
    assert "@test_channel" in prompt
    # The JSON schema example must survive verbatim.
    assert '"rewritten_text"' in prompt
    assert "{" in prompt and "}" in prompt
    assert "4000 characters" in prompt


def test_prompt_languages():
    for lang, expected in [
        ("en", "English"),
        ("es", "Spanish"),
        ("ar", "Arabic"),
        ("tr", "Turkish"),
        ("fa", "Persian"),
    ]:
        prompt = build_rewrite_system_prompt("Test", "Tone", "@tag", language=lang)
        assert expected in prompt, f"Expected {expected} in prompt for lang {lang}"


def test_prompt_media_length_limits():
    media_prompt = build_rewrite_system_prompt("Test", "Tone", "@tag", has_media=True)
    assert "1000 characters" in media_prompt
    assert "1024-character caption limit" in media_prompt

    text_prompt = build_rewrite_system_prompt("Test", "Tone", "@tag", has_media=False)
    assert "4000 characters" in text_prompt
    assert "4096-character message limit" in text_prompt


def test_verify_prompt():
    v_prompt = build_verify_system_prompt("English Channel", "Humorous Tone", "@eng_footer", language="en", has_media=True)
    assert "English Channel" in v_prompt
    assert "English" in v_prompt
    assert "Humorous Tone" in v_prompt
    assert "@eng_footer" in v_prompt
    assert "1000 characters" in v_prompt
    assert '"final_text"' in v_prompt



def test_llm_client_agy_provider():
    client = LLMClient(provider="agy", base_url="http://127.0.0.1:8000/v1")
    assert client.provider == "agy"
    assert client.model == "antigravity"
    assert client.base_url == "http://127.0.0.1:8000/v1"


def test_llm_client_reasoning_effort():
    client = LLMClient(reasoning_effort="high")
    assert client.reasoning_effort == "high"



class FakeLLM:
    def __init__(self):
        self.calls = []

    async def generate_json(self, prompt: str, system_prompt: str = None, action_name: str = "general", *args, **kwargs):
        self.calls.append((prompt, system_prompt, action_name))
        if action_name == "rewrite_target_post":
            return {"rewritten_text": "Draft with Spanish and English mixed"}
        elif action_name == "verify_target_post":
            return {"final_text": "Clean English only verified post"}
        return {}


async def async_test_ai_processor_double_check():
    fake_llm = FakeLLM()
    processor = AIProcessor(repo=None, llm=fake_llm, double_check=True)
    target = TargetChannel(id=1, channel_id=123, title="Target", username="tgt", language="en")
    
    result = await processor.rewrite_for_target("Original Raw Post", target, has_media=False)
    assert result.rewritten_text == "Clean English only verified post"
    assert str(result) == "Clean English only verified post"
    assert result.is_rejected is False
    assert len(fake_llm.calls) == 2
    assert fake_llm.calls[0][2] == "rewrite_target_post"
    assert fake_llm.calls[1][2] == "verify_target_post"



class FakeRepo:
    def __init__(self):
        self.settings = {"ai_provider": "gemini", "ai_model": "gemini-2.5-flash", "ai_double_check": "true"}
        self.logs = []
        self.active_provider = None

    async def get_setting(self, key, default=None):
        return self.settings.get(key, default)

    async def set_setting(self, key, value, description=None):
        self.settings[key] = value

    async def record_ai_log(self, action_name, provider, model, prompt, system_prompt=None, response_text=None, duration_sec=0.0, status="success", error_message=None):
        self.logs.append({
            "action_name": action_name,
            "provider": provider,
            "model": model,
            "prompt": prompt,
            "status": status,
        })
        return len(self.logs)

    async def get_active_provider_profile(self):
        return self.active_provider


async def async_test_dynamic_config_and_logging():
    fake_repo = FakeRepo()
    client = LLMClient(provider="openai", model="orcarouter/auto", repo=fake_repo)
    await client.sync_config_from_repo()
    assert client.provider == "gemini"
    assert client.model == "gemini-2.5-flash"

    processor = AIProcessor(repo=fake_repo)
    assert await processor._is_double_check_enabled() is True


async def async_test_provider_profile_sync():
    from db.models import AIProviderProfile
    fake_repo = FakeRepo()
    fake_repo.active_provider = AIProviderProfile(
        id=1,
        name="Local AGY",
        provider_type="agy",
        model="antigravity-turbo",
        base_url="http://localhost:8000/v1",
        api_key="",
        is_active=True
    )
    client = LLMClient(provider="openai", model="orcarouter/auto", repo=fake_repo)
    await client.sync_config_from_repo()
    assert client.provider == "agy"
    assert client.model == "antigravity-turbo"
    assert client.base_url == "http://localhost:8000/v1"


async def async_test_ai_processor_rejection():
    class RejectLLM:
        async def generate_json(self, prompt: str, system_prompt: str = None, action_name: str = "general", *args, **kwargs):
            return {
                "decision": "reject",
                "rejection_reason": "تبلیغات نامرتبط و غیرمجاز",
                "rewritten_text": ""
            }

    processor = AIProcessor(repo=None, llm=RejectLLM())
    target = TargetChannel(id=1, channel_id=123, title="News Channel", username="news", custom_prompt="فقط خبرهای فناوری")
    res = await processor.rewrite_for_target("تبلیغ خرید پکیج آموزشی", target)
    assert res.is_rejected is True
    assert res.rejection_reason == "تبلیغات نامرتبط و غیرمجاز"
    assert str(res) == ""


def test_custom_target_prompt_in_system_prompt():
    prompt = build_rewrite_system_prompt(
        channel_title="Tech News",
        personality="رسمی",
        custom_footer="@tech_news",
        custom_prompt="همیشه تیتر را با ایموجی 🔴 شروع کن و خلاصه‌ای در ۳ خط بنویس",
        language="fa"
    )
    assert "TARGET CHANNEL SPECIFIC COMMANDS & FORMATTING RULES:" in prompt
    assert "همیشه تیتر را با ایموجی 🔴 شروع کن" in prompt
    assert "decision" in prompt
    assert "rejection_reason" in prompt


if __name__ == "__main__":
    test_prompt_builds_with_literal_json_braces()
    test_prompt_languages()
    test_prompt_media_length_limits()
    test_verify_prompt()
    test_custom_target_prompt_in_system_prompt()
    test_llm_client_agy_provider()
    test_llm_client_reasoning_effort()
    asyncio.run(async_test_ai_processor_double_check())
    asyncio.run(async_test_ai_processor_rejection())
    asyncio.run(async_test_dynamic_config_and_logging())
    asyncio.run(async_test_provider_profile_sync())
    print("All prompt, custom instructions, rejection, and multi-provider profile unit tests passed!")