import ast
from pathlib import Path
from types import SimpleNamespace
import tempfile
import unittest
from unittest.mock import Mock, patch
import sys

sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
from qwen_prompt import make_candidate, apply_candidate
from qwen_prompt_runtime import QwenRuntime, ResourceUnavailable, available_ram, require_disk, GIB


class TestQwenPrompt(unittest.TestCase):
    def test_preview_and_explicit_apply(self):
        with patch("qwen_prompt.runtime.complete",return_value="1girl, (red hair:1.2), garden, soft sunlight") as run:
            candidate,state,_ = make_candidate("1girl, (red hair:1.2), garden","柔らかな日差しだけ追加","original")
        self.assertEqual(state,{"original":"original"})
        self.assertIn("追加指示: 柔らかな日差しだけ追加",run.call_args.args[0][1]["content"])
        self.assertEqual(apply_candidate(candidate,state,"original")[0],candidate)
        self.assertEqual(apply_candidate(candidate,state,"edited")[0],{"__type__":"update"})

    def test_failure_and_weight_damage_clear_state(self):
        for outcome in (RuntimeError(),ResourceUnavailable("メモリ不足"),"1girl, red hair"):
            with patch("qwen_prompt.runtime.complete",**({"side_effect":outcome} if isinstance(outcome,Exception) else {"return_value":outcome})):
                candidate,state,status = make_candidate("1girl, (red hair:1.2)","","original")
            self.assertEqual(candidate,""); self.assertIsNone(state); self.assertTrue(status)

    def test_empty_and_long_input_do_not_load(self):
        with patch("qwen_prompt.runtime.complete") as run:
            for source,instruction in (("",""),("x"*4001,""),("x"*3000,"x"*1001)):
                self.assertEqual(make_candidate(source,instruction,"current")[:2],("",None))
            run.assert_not_called()

    def test_clinical_vocabulary_not_filtered(self):
        with patch("qwen_prompt.runtime.complete",return_value="breast, nipple, areola") as run:
            candidate,_,_ = make_candidate("医学用語：乳房、乳首、乳輪","","current")
        self.assertIn("乳首",run.call_args.args[0][1]["content"])
        self.assertEqual(candidate,"breast, nipple, areola")

    def test_idle_releases_process_without_touching_files(self):
        now = [699]
        runtime = QwenRuntime(clock=lambda:now[0])
        process = Mock(); process.poll.return_value=None
        runtime.process=process; runtime.last_used=100
        with patch("qwen_prompt_runtime.available_ram",return_value=10*GIB):
            self.assertFalse(runtime.expire_idle())
            now[0]=700
            self.assertTrue(runtime.expire_idle())
        process.terminate.assert_called_once()
        self.assertIsNone(runtime.process)
        runtime.close()

    def test_active_request_not_terminated(self):
        runtime=QwenRuntime(clock=lambda:1000)
        process=Mock(); process.poll.return_value=None
        runtime.process=process
        runtime.lock.acquire()
        try:
            self.assertFalse(runtime.expire_idle())
            self.assertFalse(runtime.unload())
            process.terminate.assert_not_called()
        finally:
            runtime.lock.release(); runtime.close()

    def test_memory_pressure_releases_idle_model_early(self):
        runtime=QwenRuntime(clock=lambda:100)
        process=Mock(); process.poll.return_value=None
        runtime.process=process; runtime.last_used=99
        with patch("qwen_prompt_runtime.available_ram",return_value=GIB//2):
            self.assertTrue(runtime.expire_idle())
        process.terminate.assert_called_once(); runtime.close()

    def test_insufficient_ram_does_not_download_or_start(self):
        provider=Mock()
        runtime=QwenRuntime(asset_provider=provider)
        try:
            with patch("qwen_prompt_runtime.available_ram",return_value=5*GIB):
                with self.assertRaises(ResourceUnavailable):
                    runtime.complete([])
            provider.assert_not_called()
            self.assertIsNone(runtime.process)
        finally:
            runtime.close()

    def test_cgroup_ram_and_disk_limits(self):
        with tempfile.TemporaryDirectory() as directory:
            root=Path(directory)
            (root/"memory.max").write_text(str(8*GIB))
            (root/"memory.current").write_text(str(5*GIB))
            with patch("qwen_prompt_runtime.psutil.virtual_memory",return_value=SimpleNamespace(available=30*GIB)):
                self.assertEqual(available_ram(root),3*GIB)
        with patch("qwen_prompt_runtime.shutil.disk_usage",return_value=SimpleNamespace(free=1)):
            with self.assertRaises(ResourceUnavailable):
                require_disk(Path("."),1000)

    def test_gpu_quota_size_and_cpu_isolation(self):
        root=Path(__file__).resolve().parents[1]
        source=(root/"app.py").read_text(encoding="utf-8")
        tree=ast.parse(source)
        core=next(n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name=="_generate_image_core")
        decorator=next(d for d in core.decorator_list if isinstance(d,ast.Call) and isinstance(d.func,ast.Attribute) and d.func.attr=="GPU")
        self.assertEqual(next(ast.literal_eval(k.value) for k in decorator.keywords if k.arg=="size"),"large")
        helper=(root/"qwen_prompt_runtime.py").read_text(encoding="utf-8")
        self.assertNotIn("import torch",helper)
        self.assertNotIn("import spaces",helper)
        self.assertIn('"-ngl", "0"',helper)
        self.assertIn('CUDA_VISIBLE_DEVICES=""',helper)
        self.assertIn('if family == "anima":\n                build_qwen_prompt(prompt_input)',source)


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