import os import sys import tempfile _fd, _TMP = tempfile.mkstemp(suffix=".db") os.close(_fd) os.unlink(_TMP) os.environ["AGENT_OFFICE_DB_PATH"] = _TMP sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pytest from unittest.mock import AsyncMock from app.agents.insta import InstaAgent @pytest.fixture(autouse=True) def _init_db(): import gc gc.collect() if os.path.exists(_TMP): os.remove(_TMP) from app.db import init_db init_db() yield gc.collect() @pytest.mark.asyncio async def test_autonomous_issue_previews_eligible(monkeypatch): agent = InstaAgent() agent.state = "idle" monkeypatch.setattr("app.agents.insta.get_agent_config", lambda aid: {"custom_config": {"autonomous_issue": True, "select_threshold": 0.5, "max_per_day": 2}}) monkeypatch.setattr(agent, "transition", AsyncMock()) monkeypatch.setattr(agent, "_run_collect_and_extract", AsyncMock()) monkeypatch.setattr("app.agents.insta.service_proxy.insta_ranked", AsyncMock(return_value=[ {"id": 1, "keyword": "금리", "category": "economy", "eligible": True, "final_score": 0.8, "breakdown": {}}, {"id": 2, "keyword": "x", "category": "economy", "eligible": False, "final_score": 0.1, "breakdown": {}}, ])) preview = AsyncMock() monkeypatch.setattr(agent, "_generate_and_preview", preview) monkeypatch.setattr("app.agents.insta.create_task", lambda *a, **k: "t1") monkeypatch.setattr("app.agents.insta.update_task_status", lambda *a, **k: None) monkeypatch.setattr("app.agents.insta.add_log", lambda *a, **k: None) await agent.on_schedule() assert preview.await_count == 1 assert preview.await_args.args[0]["id"] == 1 @pytest.mark.asyncio async def test_callback_approve_publishes_and_delivers(monkeypatch): agent = InstaAgent() monkeypatch.setattr("app.agents.insta.service_proxy.insta_decision", AsyncMock(return_value={"status": "published"})) monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_slate", AsyncMock(return_value={ "assets": [{"page_index": i} for i in range(1, 11)], "suggested_caption": "cap", "hashtags": ["#a"]})) monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_asset_bytes", AsyncMock(return_value=b"png")) monkeypatch.setattr("app.agents.insta._send_media_group", AsyncMock(return_value={"ok": True})) monkeypatch.setattr("app.agents.insta.messaging.send_raw", AsyncMock()) res = await agent.on_callback("issue_approve", {"slate_id": 8}) assert res["ok"] is True @pytest.mark.asyncio async def test_callback_reject_marks_rejected(monkeypatch): agent = InstaAgent() dec = AsyncMock(return_value={"status": "rejected"}) monkeypatch.setattr("app.agents.insta.service_proxy.insta_decision", dec) monkeypatch.setattr("app.agents.insta.messaging.send_raw", AsyncMock()) res = await agent.on_callback("issue_reject", {"slate_id": 8}) assert res["ok"] is True dec.assert_awaited_once_with(8, "rejected") @pytest.mark.asyncio async def test_handle_insta_issue_dispatch(monkeypatch): """_handle_insta_issue: issue_approve_8 → on_callback('issue_approve', {slate_id:8}).""" import sys # stub api_call so answerCallbackQuery doesn't hit real Telegram import app.telegram.webhook as wh monkeypatch.setattr(wh, "api_call", AsyncMock(return_value={"ok": True})) agent = InstaAgent() on_cb = AsyncMock(return_value={"ok": True}) monkeypatch.setattr(agent, "on_callback", on_cb) from app.agents import AGENT_REGISTRY old = AGENT_REGISTRY.get("insta") AGENT_REGISTRY["insta"] = agent try: result = await wh._handle_insta_issue( {"id": "cq1", "data": "issue_approve_8"}, "issue_approve_8", ) finally: if old is None: AGENT_REGISTRY.pop("insta", None) else: AGENT_REGISTRY["insta"] = old on_cb.assert_awaited_once_with("issue_approve", {"slate_id": 8}) assert result["ok"] is True @pytest.mark.asyncio async def test_handle_insta_issue_invalid_data(monkeypatch): """_handle_insta_issue: 잘못된 callback_data → ok=False, error=invalid_callback_data.""" import app.telegram.webhook as wh monkeypatch.setattr(wh, "api_call", AsyncMock(return_value={"ok": True})) monkeypatch.setattr("app.telegram.messaging.send_raw", AsyncMock()) result = await wh._handle_insta_issue( {"id": "cq2", "data": "issue_bad"}, "issue_bad", ) assert result["ok"] is False assert result["error"] == "invalid_callback_data" @pytest.mark.asyncio async def test_backward_compat_non_autonomous_uses_legacy_path(monkeypatch): """autonomous_issue=False, auto_select=False → insta_ranked 미호출, _push_keyword_candidates 호출.""" agent = InstaAgent() agent.state = "idle" monkeypatch.setattr("app.agents.insta.get_agent_config", lambda aid: {"custom_config": {"autonomous_issue": False, "auto_select": False}}) monkeypatch.setattr(agent, "transition", AsyncMock()) monkeypatch.setattr(agent, "_run_collect_and_extract", AsyncMock()) # insta_get_preferences는 try/except 안에 있으므로 예외를 던져도 안전하지만 깔끔하게 mock monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_preferences", AsyncMock(return_value={})) # 비자율 경로에서 insta_ranked는 호출되면 안 된다 ranked = AsyncMock() monkeypatch.setattr("app.agents.insta.service_proxy.insta_ranked", ranked) # insta_list_keywords: 비자율 경로에서 반드시 호출 monkeypatch.setattr("app.agents.insta.service_proxy.insta_list_keywords", AsyncMock(return_value=[])) # auto_select=False → _push_keyword_candidates 경로 push = AsyncMock() monkeypatch.setattr(agent, "_push_keyword_candidates", push) gen = AsyncMock() monkeypatch.setattr(agent, "_generate_and_preview", gen) monkeypatch.setattr("app.agents.insta.create_task", lambda *a, **k: "t1") monkeypatch.setattr("app.agents.insta.update_task_status", lambda *a, **k: None) monkeypatch.setattr("app.agents.insta.add_log", lambda *a, **k: None) await agent.on_schedule() ranked.assert_not_awaited() # 자율 경로(insta_ranked) 미진입 확인 gen.assert_not_awaited() # _generate_and_preview 미호출 확인 push.assert_awaited_once() # 기존 candidate-push 경로 진입 확인 @pytest.mark.asyncio async def test_callback_regen_rejects_old_and_regenerates(monkeypatch): """issue_regen: 기존 슬레이트 rejected 처리 후 같은 키워드로 _generate_and_preview 재호출.""" agent = InstaAgent() monkeypatch.setattr("app.agents.insta.service_proxy.insta_get_slate", AsyncMock(return_value={"keyword": "금리", "category": "economy"})) dec = AsyncMock(return_value={"status": "rejected"}) monkeypatch.setattr("app.agents.insta.service_proxy.insta_decision", dec) gen = AsyncMock() monkeypatch.setattr(agent, "_generate_and_preview", gen) res = await agent.on_callback("issue_regen", {"slate_id": 8}) assert res["ok"] is True dec.assert_awaited_once_with(8, "rejected") # 이전 슬레이트 폐기 gen.assert_awaited_once() # 같은 키워드로 재생성 assert gen.await_args.args[0]["keyword"] == "금리"