sap-cli-skill/docs/adt/06.ADT测试代码.md

title, created, tags, parent

title	created	tags	parent
ADT 测试代码 — 读取程序源代码	2026-05-19	SAP ADT REST python test-code	sap-cli/README

ADT 测试代码 — 通过 REST API 读取程序源代码

[!abstract] 概述 通过 ADT REST API 连接 SAP 系统，测试连接、服务发现、读取程序源代码、搜索仓库对象。 参见：sap-cli/02.查询代码-原理

运行方式

# 安装依赖
pip install requests

# 运行测试
python sap-cli/test_adt_client.py

测试结果

[!success] 2026-05-19 测试通过

• 连接验证: HTTP 200 ✅
• 服务发现: 311 个可用服务 ✅
• 读取源代码: ZIDTR_IMPORT_ACCOUNTING_ORDER (37,655 字符 / 1,096 行) ✅
• 仓库搜索: quickSearch 返回 0 结果 ⚠️

完整代码

"""
SAP ADT REST Client — 测试脚本
通过 ADT REST API 连接 SAP 系统并获取程序源代码

参考: [[sap-cli/02.查询代码-原理]]
      [[sap-cli/01.ADT架构与通信原理]]

端点: /sap/bc/adt/programs/programs/{name}/source/main
"""

import requests
from urllib.parse import urljoin
import xml.etree.ElementTree as ET
import sys
import json
import io

# 修复 Windows 控制台编码问题
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')


# ============================================================
# 配置 — 来自 [[sap-cli/README]] 测试服务器信息
# ============================================================
ADT_BASE_URL = "http://support.learningleader.com.cn:55955/sap/bc/adt"
SAP_CLIENT = "100"
USER = "admin2"
PASSWORD = "654321"
PROGRAM_NAME = "ZIDTR_IMPORT_ACCOUNTING_ORDER"


class ADTClient:
    """简易 ADT REST 客户端"""

    CSRF_HEADER = "x-csrf-token"
    CSRF_FETCH = "Fetch"

    def __init__(self, base_url: str, client: str, user: str, password: str):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.auth = (user, password)
        self.session.headers.update({
            "Accept": "*/*",
            "Accept-Language": "EN",
            "sap-client": client,
        })
        # 忽略 SSL 警告（测试环境可能是自签名证书）
        requests.packages.urllib3.disable_warnings(
            requests.packages.urllib3.exceptions.InsecureRequestWarning
        )
        self.session.verify = False
        self._csrf_token: str | None = None

    def _url(self, path: str) -> str:
        return f"{self.base_url}/{path}"

    def fetch_csrf_token(self) -> str:
        """获取 CSRF Token — 写操作必需"""
        print("  → 正在获取 CSRF Token...")
        resp = self.session.get(
            self._url("discovery"),
            headers={self.CSRF_HEADER: self.CSRF_FETCH},
        )
        resp.raise_for_status()
        self._csrf_token = resp.headers.get(self.CSRF_HEADER, "")
        print(f"  ✓ CSRF Token 获取成功: {self._csrf_token[:20]}...")
        return self._csrf_token

    def get(self, path: str, **kwargs) -> requests.Response:
        """发送 GET 请求"""
        resp = self.session.get(self._url(path), **kwargs)
        resp.raise_for_status()
        return resp


def test_connection(client: ADTClient) -> bool:
    """测试 1: 验证 ADT 连接"""
    print("\n" + "=" * 60)
    print("测试 1: 验证 ADT 连接")
    print("=" * 60)
    print(f"  URL: {client.base_url}")
    print(f"  User: {client.session.auth[0]}")
    print(f"  Client: {client.session.headers.get('sap-client')}")

    try:
        resp = client.session.get(client._url("discovery"))
        print(f"  → HTTP Status: {resp.status_code}")
        if resp.status_code == 401:
            print(f"  ✗ 认证失败 (401 Unauthorized)")
            print(f"     响应体: {resp.text[:300]}")
            return False
        resp.raise_for_status()
        print(f"  ✓ 连接成功! HTTP {resp.status_code}")
        print(f"  ✓ Content-Type: {resp.headers.get('content-type', 'N/A')}")
        return True
    except requests.exceptions.HTTPError as e:
        print(f"  ✗ HTTP 错误: {e}")
        if hasattr(e, 'response') and e.response is not None:
            print(f"     响应体: {e.response.text[:300]}")
        return False
    except requests.exceptions.ConnectionError as e:
        print(f"  ✗ 网络连接失败: {e}")
        return False


def test_discovery(client: ADTClient) -> list[dict]:
    """测试 2: 服务发现 — 列出可用的 ADT 服务"""
    print("\n" + "=" * 60)
    print("测试 2: 服务发现 (Discovery)")
    print("=" * 60)

    try:
        resp = client.get("discovery")
        root = ET.fromstring(resp.content)

        # 解析 Atom Feed 中的服务集合
        ns = {
            "app": "http://www.w3.org/2007/app",
            "atom": "http://www.w3.org/2005/Atom",
        }

        services = []
        for collection in root.findall(".//app:collection", ns):
            href = collection.attrib.get("href", "")
            title_el = collection.find("atom:title", ns)
            title = title_el.text if title_el is not None else ""
            services.append({"href": href, "title": title})

        print(f"  ✓ 发现 {len(services)} 个可用服务:")
        for svc in services[:15]:  # 只显示前 15 个
            print(f"    • {svc['title']}: {svc['href']}")
        if len(services) > 15:
            print(f"    ... 还有 {len(services) - 15} 个服务")

        return services
    except Exception as e:
        print(f"  ✗ 服务发现失败: {e}")
        return []


def test_read_program_source(client: ADTClient, program_name: str) -> str | None:
    """测试 3: 读取程序源代码"""
    print("\n" + "=" * 60)
    print(f"测试 3: 读取程序源代码 ({program_name})")
    print("=" * 60)

    # 尝试多种可能的端点路径
    endpoints = [
        f"programs/programs/{program_name.lower()}/source/main",
        f"programs/programs/{program_name}/source/main",
    ]

    for endpoint in endpoints:
        try:
            print(f"  → 尝试端点: GET {endpoint}")
            resp = client.get(
                endpoint,
                headers={"Accept": "text/plain"},
            )
            source = resp.text

            print(f"  ✓ 源代码获取成功!")
            print(f"  ✓ 程序名: {program_name}")
            print(f"  ✓ 源代码长度: {len(source)} 字符")
            print(f"  ✓ 行数: {len(source.splitlines())}")

            # 显示前 30 行
            lines = source.splitlines()
            print(f"\n  ┌─── 源代码 (前 30 行) ──────────────────────")
            for i, line in enumerate(lines[:30], 1):
                print(f"  │ {i:4d} | {line}")
            if len(lines) > 30:
                print(f"  │ ... 省略剩余 {len(lines) - 30} 行 ...")
            print(f"  └────────────────────────────────────────────")

            return source

        except requests.exceptions.HTTPError as e:
            print(f"  ✗ HTTP 错误: {e.response.status_code} {e.response.reason}")
            if e.response.status_code == 404:
                print(f"     (程序不存在或端点路径不正确，尝试下一个...)")
                continue
            else:
                print(f"     响应内容: {e.response.text[:200]}")
                return None
        except Exception as e:
            print(f"  ✗ 读取失败: {e}")
            return None

    print(f"  ✗ 所有端点尝试均失败")
    return None


def test_search_object(client: ADTClient, program_name: str) -> list[dict]:
    """测试 4: 搜索仓库对象"""
    print("\n" + "=" * 60)
    print(f"测试 4: 搜索仓库对象 ({program_name})")
    print("=" * 60)

    try:
        resp = client.get(
            "repository/informationsystem/search",
            params={
                "operation": "quickSearch",
                "query": program_name,
                "maxResults": "10",
            },
        )

        ns = "http://www.sap.com/adt/repository/informationsystem"
        root = ET.fromstring(resp.content)

        results = []
        for obj in root.iter(f"{{{ns}}}object"):
            results.append({
                "name": obj.attrib.get("name", ""),
                "type": obj.attrib.get("type", ""),
                "uri": obj.attrib.get("uri", ""),
                "package": obj.attrib.get("packageName", ""),
            })

        print(f"  ✓ 搜索完成，找到 {len(results)} 个结果:")
        for r in results:
            print(f"    • [{r['type']}] {r['name']} (Package: {r['package']})")
            print(f"      URI: {r['uri']}")

        return results

    except Exception as e:
        print(f"  ✗ 搜索失败: {e}")
        return []


def main():
    print("=" * 60)
    print("  SAP ADT REST API 测试脚本")
    print("  参考: [[sap-cli/02.查询代码-原理]]")
    print("=" * 60)

    # 创建客户端
    client = ADTClient(
        base_url=ADT_BASE_URL,
        client=SAP_CLIENT,
        user=USER,
        password=PASSWORD,
    )

    # 测试 1: 连接验证
    if not test_connection(client):
        print("\n❌ 连接失败，请检查服务器配置和网络连接。")
        sys.exit(1)

    # 测试 2: 服务发现
    test_discovery(client)

    # 测试 3: 读取程序源代码
    source = test_read_program_source(client, PROGRAM_NAME)

    # 测试 4: 搜索仓库对象
    test_search_object(client, PROGRAM_NAME)

    # 汇总结果
    print("\n" + "=" * 60)
    print("  测试结果汇总")
    print("=" * 60)
    if source:
        print("  ✅ 程序源代码获取成功!")
        print(f"     程序: {PROGRAM_NAME}")
        print(f"     长度: {len(source)} 字符, {len(source.splitlines())} 行")
    else:
        print("  ⚠️  程序源代码获取失败，请确认程序名称是否正确")
    print("=" * 60)


if __name__ == "__main__":
    main()

关键 API 端点速查

操作	HTTP 方法	端点	Accept
服务发现	GET	/sap/bc/adt/discovery	*/*
读取程序源码	GET	/sap/bc/adt/programs/programs/{name}/source/main	text/plain
读取类源码	GET	/sap/bc/adt/oo/classes/{name}/source/main	text/plain
仓库搜索	GET	/sap/bc/adt/repository/informationsystem/search	*/*

[!note] 注意事项

• discovery 端点的 Accept 必须用 */*，用 application/xml 会返回 406

• 程序名在 URL 中使用小写

🔗 相关笔记

• sap-cli/01.ADT架构与通信原理
• sap-cli/02.查询代码-原理
• sap-cli/03.修改代码-原理