feat(agent-office): /activity 통합 피드에 필터 추가 (agent_id/type/status/days)
오버사이트 UI용. get_activity_feed가 브랜치별 WHERE로 필터, total도 동일 반영. status는 task 전용(주면 log 제외). 값은 ? 바인딩, type은 브랜치 선택만이라 injection 안전. 신규 5 테스트. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
76
agent-office/tests/test_activity_feed_filters.py
Normal file
76
agent-office/tests/test_activity_feed_filters.py
Normal file
@@ -0,0 +1,76 @@
|
||||
# agent-office/tests/test_activity_feed_filters.py
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import gc
|
||||
|
||||
_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 app import db
|
||||
|
||||
db.DB_PATH = _TMP
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fresh_db():
|
||||
db.DB_PATH = _TMP
|
||||
gc.collect()
|
||||
if os.path.exists(_TMP):
|
||||
os.remove(_TMP)
|
||||
db.init_db()
|
||||
yield
|
||||
gc.collect()
|
||||
if os.path.exists(_TMP):
|
||||
try:
|
||||
os.remove(_TMP)
|
||||
except PermissionError:
|
||||
pass
|
||||
|
||||
|
||||
def test_filter_by_agent_id():
|
||||
db.create_task("lotto", "curate", {})
|
||||
db.create_task("stock", "brief", {})
|
||||
db.add_log("stock", "stock 로그")
|
||||
feed = db.get_activity_feed(limit=50, offset=0, agent_id="lotto")
|
||||
assert feed["total"] == 1
|
||||
assert all(i["agent_id"] == "lotto" for i in feed["items"])
|
||||
|
||||
|
||||
def test_filter_type_task_excludes_logs():
|
||||
db.create_task("lotto", "curate", {})
|
||||
db.add_log("lotto", "로그 한 줄")
|
||||
feed = db.get_activity_feed(limit=50, offset=0, type="task")
|
||||
assert feed["total"] == 1
|
||||
assert all(i["type"] == "task" for i in feed["items"])
|
||||
|
||||
|
||||
def test_filter_type_log_excludes_tasks():
|
||||
db.create_task("lotto", "curate", {})
|
||||
db.add_log("lotto", "로그 한 줄")
|
||||
feed = db.get_activity_feed(limit=50, offset=0, type="log")
|
||||
assert feed["total"] == 1
|
||||
assert all(i["type"] == "log" for i in feed["items"])
|
||||
|
||||
|
||||
def test_filter_status_tasks_only():
|
||||
t1 = db.create_task("lotto", "curate", {})
|
||||
t2 = db.create_task("lotto", "curate", {})
|
||||
db.update_task_status(t1, "succeeded", {})
|
||||
db.update_task_status(t2, "failed", {})
|
||||
db.add_log("lotto", "로그 한 줄") # status 필터 시 log는 제외돼야 함
|
||||
feed = db.get_activity_feed(limit=50, offset=0, status="succeeded")
|
||||
assert feed["total"] == 1
|
||||
assert all(i["type"] == "task" and i["status"] == "succeeded" for i in feed["items"])
|
||||
|
||||
|
||||
def test_no_filters_returns_all():
|
||||
db.create_task("lotto", "curate", {})
|
||||
db.add_log("stock", "로그")
|
||||
feed = db.get_activity_feed(limit=50, offset=0)
|
||||
assert feed["total"] == 2
|
||||
Reference in New Issue
Block a user