Gan, Jimmy
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""Tests for agent_worker.py — mocks Qwen API."""
|
|
import pytest
|
|
import os
|
|
import json
|
|
import aiosqlite
|
|
import httpx
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_worker_polls_and_processes(test_db, monkeypatch):
|
|
"""Worker picks up a pending item, calls Qwen, marks it done."""
|
|
from agent_worker import _process_item
|
|
|
|
# Seed a pending queue item + video
|
|
await test_db.execute("""
|
|
INSERT INTO videos (video_id, title, channel_name, url, summary, status)
|
|
VALUES ('ag_vid', 'Test Video', 'Test Channel', 'https://youtube.com/watch?v=ag_vid',
|
|
'**Summary**\n- Point 1\n- Point 2\nVerdict.', 'pending')
|
|
""")
|
|
await test_db.execute("""
|
|
INSERT INTO agent_queue (video_id, user_prompt, status)
|
|
VALUES ('ag_vid', 'Save this as a skill', 'pending')
|
|
""")
|
|
await test_db.commit()
|
|
|
|
# Get the item
|
|
cursor = await test_db.execute(
|
|
"SELECT id, video_id, user_prompt FROM agent_queue WHERE status = 'pending'"
|
|
)
|
|
item = dict(await cursor.fetchone())
|
|
|
|
# Mock Qwen API
|
|
mock_response = MagicMock(spec=httpx.Response)
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"choices": [{"message": {"content": "I've saved the skill. Done."}}]
|
|
}
|
|
|
|
with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response):
|
|
await _process_item(item)
|
|
|
|
# Verify
|
|
cursor = await test_db.execute(
|
|
"SELECT status, result FROM agent_queue WHERE id = ?", (item["id"],)
|
|
)
|
|
row = await cursor.fetchone()
|
|
assert row["status"] == "done"
|
|
assert "I've saved the skill" in row["result"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_worker_handles_missing_video(test_db):
|
|
"""When video doesn't exist, marks as failed gracefully."""
|
|
from agent_worker import _process_item
|
|
|
|
await test_db.execute("""
|
|
INSERT INTO agent_queue (video_id, user_prompt, status)
|
|
VALUES ('missing_vid', 'Do something', 'pending')
|
|
""")
|
|
await test_db.commit()
|
|
|
|
cursor = await test_db.execute(
|
|
"SELECT id, video_id, user_prompt FROM agent_queue WHERE status = 'pending'"
|
|
)
|
|
item = dict(await cursor.fetchone())
|
|
|
|
await _process_item(item)
|
|
|
|
cursor = await test_db.execute(
|
|
"SELECT status, result FROM agent_queue WHERE id = ?", (item["id"],)
|
|
)
|
|
row = await cursor.fetchone()
|
|
assert row["status"] == "failed"
|
|
assert "not found" in row["result"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_worker_qwen_error_marks_failed(test_db):
|
|
"""Qwen API failure should mark as failed, not crash."""
|
|
from agent_worker import _process_item
|
|
|
|
await test_db.execute("""
|
|
INSERT INTO videos (video_id, title, channel_name, url, status)
|
|
VALUES ('err_vid', 'Error Video', 'Test', 'https://youtube.com/watch?v=err_vid', 'pending')
|
|
""")
|
|
await test_db.execute("""
|
|
INSERT INTO agent_queue (video_id, user_prompt, status)
|
|
VALUES ('err_vid', 'Do it', 'pending')
|
|
""")
|
|
await test_db.commit()
|
|
|
|
cursor = await test_db.execute(
|
|
"SELECT id, video_id, user_prompt FROM agent_queue WHERE status = 'pending'"
|
|
)
|
|
item = dict(await cursor.fetchone())
|
|
|
|
# Mock HTTP error
|
|
mock_response = MagicMock(spec=httpx.Response)
|
|
mock_response.status_code = 503
|
|
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
|
|
"503 Service Unavailable", request=MagicMock(), response=mock_response
|
|
)
|
|
|
|
with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response):
|
|
await _process_item(item)
|
|
|
|
cursor = await test_db.execute(
|
|
"SELECT status FROM agent_queue WHERE id = ?", (item["id"],)
|
|
)
|
|
row = await cursor.fetchone()
|
|
assert row["status"] == "failed"
|