""" Test Milestone 17: Jira API client. Tests real API connectivity — your .env must have valid Jira credentials. """ import asyncio import os import sys # Fix Windows console encoding for emoji support if sys.platform == "win32": import codecs sys.stdout = codecs.getwriter("utf-8")(sys.stdout.buffer, errors="replace") sys.stderr = codecs.getwriter("utf-8")(sys.stderr.buffer, errors="replace") sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) async def test_config_loaded(): """Test that all Jira config vars are present.""" from backend.config import ( JIRA_BASE_URL, JIRA_EMAIL, JIRA_API_TOKEN, JIRA_DEFAULT_PROJECT, ENABLE_JIRA ) checks = { "JIRA_BASE_URL": JIRA_BASE_URL, "JIRA_EMAIL": JIRA_EMAIL, "JIRA_API_TOKEN": JIRA_API_TOKEN, "JIRA_DEFAULT_PROJECT": JIRA_DEFAULT_PROJECT, } all_pass = True for name, val in checks.items(): ok = bool(val and len(val) >= 2) status = "✅" if ok else "❌ MISSING" if not ok: all_pass = False print(f" {status} {name}: {val[:30] if val else '(empty)'}...") assert all_pass, "Fix missing Jira config vars in .env before continuing" print(f" ✅ ENABLE_JIRA: {ENABLE_JIRA}") async def test_connection(): """Test that credentials are valid and can reach the Jira API.""" from backend.integrations.jira_client import test_connection print("\nTesting Jira API connection...") result = await test_connection() assert result.get("ok"), f"Connection failed: {result.get('error')}" print(f" ✅ Connected as: {result['display_name']} ({result['email']})") print(f" Account ID: {result['account_id']}") async def test_list_projects(): """Test listing projects — must return at least one.""" from backend.integrations.jira_client import list_projects from backend.config import JIRA_DEFAULT_PROJECT print("\nTesting list_projects()...") projects = await list_projects() assert len(projects) > 0, "No projects returned. Make sure your account has at least one project." print(f" ✅ Found {len(projects)} project(s):") for p in projects[:5]: print(f" [{p['key']}] {p['name']}") keys = [p["key"] for p in projects] assert JIRA_DEFAULT_PROJECT in keys, ( f"JIRA_DEFAULT_PROJECT '{JIRA_DEFAULT_PROJECT}' not found in your projects: {keys}\n" "Update JIRA_DEFAULT_PROJECT in .env to one of the listed keys." ) print(f" ✅ Default project '{JIRA_DEFAULT_PROJECT}' exists") async def test_list_issue_types(): """Test listing issue types for the default project.""" from backend.integrations.jira_client import list_issue_types from backend.config import JIRA_DEFAULT_PROJECT print("\nTesting list_issue_types()...") types = await list_issue_types(JIRA_DEFAULT_PROJECT) assert len(types) > 0, f"No issue types returned for project {JIRA_DEFAULT_PROJECT}" names = [t["name"] for t in types] print(f" ✅ Issue types in '{JIRA_DEFAULT_PROJECT}': {names}") async def test_create_and_get_issue(): """Test creating a real Jira issue and then retrieving it.""" from backend.integrations.jira_client import create_issue, get_issue from backend.config import JIRA_DEFAULT_PROJECT print("\nTesting create_issue() and get_issue()...") result = await create_issue( project_key=JIRA_DEFAULT_PROJECT, summary="[ThirdEye Test] Milestone 17 verification ticket", description=( "This ticket was created automatically by the ThirdEye test suite.\n\n" "It verifies that the Jira API client can create issues successfully.\n\n" "Safe to close or delete." ), issue_type="Task", priority="Low", labels=["thirdeye", "test", "automated"], ) assert result.get("ok"), f"create_issue failed: {result.get('error')} — details: {result.get('details')}" issue_key = result["key"] print(f" ✅ Created issue: {issue_key}") print(f" URL: {result['url']}") # Retrieve it issue = await get_issue(issue_key) assert issue["key"] == issue_key assert "ThirdEye Test" in issue["summary"] assert issue["status"] in ("To Do", "Open", "Backlog", "New") print(f" ✅ Retrieved issue: [{issue['key']}] {issue['summary']}") print(f" Status: {issue['status']} | Priority: {issue['priority']}") return issue_key async def test_search_issues(issue_key: str): """Test searching issues by JQL.""" from backend.integrations.jira_client import search_issues from backend.config import JIRA_DEFAULT_PROJECT print("\nTesting search_issues() via JQL...") jql = f'project = {JIRA_DEFAULT_PROJECT} AND labels = "thirdeye" AND labels = "test" ORDER BY created DESC' results = await search_issues(jql, max_results=5) assert len(results) > 0, f"Expected at least one result for JQL: {jql}" keys = [r["key"] for r in results] assert issue_key in keys, f"Newly created issue {issue_key} not found in search results" print(f" ✅ JQL search returned {len(results)} result(s), including {issue_key}") async def test_add_comment(issue_key: str): """Test adding a comment to an existing issue.""" from backend.integrations.jira_client import add_comment print("\nTesting add_comment()...") result = await add_comment( issue_key, "ThirdEye test comment — verifying comment API works correctly." ) assert result.get("ok"), f"add_comment failed: {result.get('error')}" print(f" ✅ Comment added to {issue_key} (comment id: {result.get('id')})") async def test_adf_conversion(): """Test that _text_to_adf produces valid ADF structure.""" from backend.integrations.jira_client import _text_to_adf print("\nTesting ADF conversion...") text = "This is paragraph one.\n\nThis is paragraph two.\n\n- Bullet A\n- Bullet B" adf = _text_to_adf(text) assert adf["type"] == "doc" assert adf["version"] == 1 assert len(adf["content"]) >= 2 print(f" ✅ ADF produced {len(adf['content'])} content block(s)") # Empty text should not crash adf_empty = _text_to_adf("") assert adf_empty["type"] == "doc" print(" ✅ Empty text handled gracefully") async def main(): print("Running Milestone 17 tests...\n") await test_config_loaded() await test_connection() await test_list_projects() await test_list_issue_types() issue_key = await test_create_and_get_issue() await test_search_issues(issue_key) await test_add_comment(issue_key) await test_adf_conversion() print(f"\n🎉 MILESTONE 17 PASSED — Jira API client working. Test ticket: {issue_key}") asyncio.run(main())