Files
variet-agent/tests/test_pipeline_e2e.py

57 lines
1.6 KiB
Python

"""E2E Test: Task Pipeline with real Gemini CLI.
Tests Planner phase against the variet-agent project.
"""
import sys
import io
import asyncio
import json
if sys.stdout.encoding != "utf-8":
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.path.insert(0, r"C:\Users\CafeVariet-GL552VW\Desktop\source_diff\variet-agent")
from core.task_pipeline import TaskPipeline
PROJECT = r"C:\Users\CafeVariet-GL552VW\Desktop\source_diff\variet-agent"
async def test_planner():
print("=" * 60)
print("E2E TEST: Planner")
print("=" * 60)
pipeline = TaskPipeline(PROJECT, token_budget=30_000)
pipeline.setup()
plan = await pipeline.plan(
"project_indexer.py의 find_relevant 함수가 공백이 포함된 쿼리를 처리하지 못합니다. "
"'gemini caller'로 검색하면 gemini_caller.py를 찾지 못합니다. "
"밑줄과 공백을 동일하게 처리하도록 개선해주세요."
)
print(f"\n📋 Plan result:")
print(json.dumps(plan, ensure_ascii=False, indent=2))
if plan.get("tasks"):
print(f"\n✅ Planner returned {len(plan['tasks'])} tasks")
for t in plan["tasks"]:
print(f" - {t.get('title', t.get('id', '?'))}: {t.get('description', '')[:80]}")
else:
print(f"\n⚠️ No structured tasks, raw response:")
print(plan.get("raw", plan.get("summary", ""))[:500])
return plan
async def main():
plan = await test_planner()
print(f"\n{'=' * 60}")
print(f"Gemini calls: {1}")
print(f"✅ E2E Planner test complete!")
if __name__ == "__main__":
asyncio.run(main())