76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
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")
|