test_logfire.py

#!/usr/bin/env python3
"""
Test script to verify Logfire tracing is working correctly
"""

import os
import asyncio
from dotenv import load_dotenv
import logfire
# from agent.code_review import logfire_client

load_dotenv()

# async def test_logfire_tracing():
#     """Test basic Logfire functionality"""
    
#     # Test basic span creation
#     with logfire_client.span("test_span", attributes={"test": "basic_functionality"}) as span:
#         span.add_event("test_event", attributes={"message": "Hello from Logfire!"})
#         print("✓ Basic span creation and event logging works")
    
#     # Test error handling
#     try:
#         with logfire_client.span("error_test_span") as span:
#             raise ValueError("Test error for tracing")
#     except ValueError:
#         print("✓ Error tracing works correctly")
    
#     # Test nested spans
#     with logfire_client.span("parent_span") as parent_span:
#         parent_span.add_event("parent_event")
        
#         with logfire_client.span("child_span") as child_span:
#             child_span.add_event("child_event")
#             print("✓ Nested spans work correctly")
    
#     print("✓ All Logfire tests passed!")

# def test_sync_tracing():
#     """Test synchronous tracing"""
#     with logfire_client.span("sync_test") as span:
#         span.add_event("sync_event", attributes={"type": "synchronous"})
#         print("✓ Synchronous tracing works")
    
#     # Test exception handling
#     try:
#         with logfire_client.span("sync_error_test") as span:
#             raise RuntimeError("Sync test error")
#     except RuntimeError:
#         print("✓ Synchronous error tracing works")

# if __name__ == "__main__":
#     print("Testing Logfire tracing functionality...")
#     print(f"Logfire API Key configured: {'Yes' if os.getenv('LOGFIRE_API_KEY') else 'No'}")
#     print()
    
#     # Test sync tracing
#     test_sync_tracing()
    
#     # Test async tracing
#     asyncio.run(test_logfire_tracing())
    
from tool import send_slack_message, list_slack_users, get_slack_usergroups, reply_slack_thread
from dotenv import load_dotenv

load_dotenv()

# report = f"""# [] PR #1: Change something PR
# Hello World

# **🌿 Branch Information:**
# - **Source Branch:**  → **Target Branch:** 

# **👤 Người tạo:** Hải Huỳnh @Hai Huynh

# **📅 Thời gian tạo:** 2024-08-20 14:30:00 +07:00

# **👥 Reviewers:**

# **🔗 Link Pull Request:** [PR #1: Implement user authentication system](https://google.com.vn)
# """


# send_slack_message(
#     message_text = "Hello, *World*",                
#     webhook_url = os.getenv("SLACK_WEBHOOK_URL"),
#     bot_token = os.getenv("SLACK_BOT_TOKEN"),
#     channel = "pull-request",
#     blocks = [
#         {
#             "type": "markdown",
#             "text": report
#         }
#     ]
# ) 

import requests

# Thay token bằng bot token Slack của bạn
SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")
url = "https://slack.com/api/users.list"

users = list_slack_users()
groups = get_slack_usergroups(SLACK_BOT_TOKEN)

# print(users)
# print(groups)

send_slack_message(
    message_text = "",
    channel = "pull-request",
    blocks = [
        {
            "type": "markdown",
            "text": "<@U099W8UMBAQ>"
        }
    ]
) 

resp = send_slack_message("Xin chào team! 🚀", channel="pull-request")
first_message_ts = resp["response"]["ts"]  # lấy ts để trả lời sau

reply_slack_thread(
    thread_ts=first_message_ts,
    message_text="Đây là câu trả lời trong thread 🌱",
    channel="pull-request",
)