test_api_direct.py

#!/usr/bin/env python3
"""
直接测试 API 响应
"""

import json
import threading
import time

from sqllineage.runner import LineageRunner


def start_server():
    """启动服务"""
    sql = "select 1;"
    runner = LineageRunner(sql)
    runner.draw()


def test_api():
    """测试 API"""
    # 等待服务启动
    time.sleep(3)

    try:
        import urllib.parse
        import urllib.request

        # 准备请求数据
        data = json.dumps({}).encode("utf-8")

        # 创建请求
        req = urllib.request.Request(
            "http://localhost:5000/directory", data=data, headers={"Content-Type": "application/json"}
        )

        # 发送请求
        with urllib.request.urlopen(req) as response:
            result = json.loads(response.read().decode("utf-8"))

        print("API 响应:")
        print(json.dumps(result, indent=2, ensure_ascii=False))

        # 检查 tpcds 目录
        for child in result.get("children", []):
            if child["name"] == "tpcds" and child["is_dir"]:
                print("\n✅ 找到 tpcds 目录")
                print(f"tpcds children 数量: {len(child.get('children', []))}")
                if child.get("children"):
                    print("tpcds 前5个文件:")
                    for i, file in enumerate(child["children"][:5]):
                        print(f"  {i + 1}. {file['name']} (is_dir: {file['is_dir']})")
                break
        else:
            print("\n❌ 未找到 tpcds 目录")

    except Exception as e:
        print(f"❌ API 测试失败: {e}")


def main():
    print("🚀 启动服务并测试 API...")

    # 启动服务
    thread = threading.Thread(target=start_server, daemon=True)
    thread.start()

    # 测试 API
    test_api()


if __name__ == "__main__":
    main()