--- title: ADT 测试代码 — 读取程序源代码 created: 2026-05-19 tags: - SAP - ADT - REST - python - test-code parent: "[[sap-cli/README|SAP ADT 学习笔记总览]]" --- # ADT 测试代码 — 通过 REST API 读取程序源代码 > [!abstract] 概述 > 通过 ADT REST API 连接 SAP 系统，测试连接、服务发现、读取程序源代码、搜索仓库对象。 > 参见：[[sap-cli/02.查询代码-原理]] ## 运行方式 ```bash # 安装依赖 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 结果 ⚠️ ## 完整代码 ```python """ 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架构与通信原理|01.ADT架构与通信原理]] - [[sap-cli/02.查询代码-原理|02.查询代码 - 原理]] - [[sap-cli/03.修改代码-原理|03.修改代码 - 原理]]