test_web.py

#!/usr/bin/env python3
"""
测试 Web 服务
"""

import json

from sqllineage.drawing import app


def test_directory_api():
    """测试目录 API"""
    print("🧪 测试目录 API...")

    # 模拟请求
    environ = {"REQUEST_METHOD": "POST", "PATH_INFO": "/directory", "CONTENT_LENGTH": "2", "wsgi.input": None}

    def start_response(status, headers):
        print(f"状态: {status}")

    # 创建模拟的请求体
    import io

    environ["wsgi.input"] = io.BytesIO(b"{}")

    # 调用应用
    try:
        response = app(environ, start_response)
        data = json.loads(response[0].decode("utf-8"))

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

        # 检查 tpcds 目录
        for child in data.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"❌ 测试失败: {e}")


def main():
    print("🚀 测试目录 API...")
    test_directory_api()


if __name__ == "__main__":
    main()