chore: sync skill to v2.2.1
- NW 7.40 E2E-verified compatibility matrix in SKILL.md - DDIC fixes: include namespace, tabletype exists check, DDIC delete lock Accept - New commands: clone, enhancement, history, unit-test - 665 tests, 96% coverage
This commit is contained in:
@@ -12,6 +12,41 @@ from sapcli.types import ObjectTypeConfig, get_type_config
|
||||
logger = logging.getLogger("sapcli.client")
|
||||
|
||||
|
||||
def _local(tag: str) -> str:
|
||||
"""返回 XML 标签/属性名的本地部分（去掉 ``{namespace}`` 前缀）。"""
|
||||
if tag and tag[0] == "{":
|
||||
return tag.split("}", 1)[1]
|
||||
return tag
|
||||
|
||||
|
||||
def _attr_local(el: ET.Element, name: str) -> str:
|
||||
"""按本地名查找元素属性值（忽略命名空间）。
|
||||
|
||||
例如 ``adtcore:name`` 与 ``name`` 都能匹配 ``name``。
|
||||
"""
|
||||
for key, val in el.attrib.items():
|
||||
if _local(key) == name:
|
||||
return val
|
||||
return ""
|
||||
|
||||
|
||||
def _find_local(container: ET.Element, name: str) -> ET.Element | None:
|
||||
"""返回容器（含自身）下首个本地标签名为 ``name`` 的元素。"""
|
||||
for el in container.iter():
|
||||
if _local(el.tag) == name:
|
||||
return el
|
||||
return None
|
||||
|
||||
|
||||
# ADT lock 端点返回 ABAP 结构 XML。DDIC 对象（domain/dataelement/table/structure）
|
||||
# 的 lock 端点对默认 Accept: */* 返回 HTTP 406，必须显式请求 lock 结果类型。
|
||||
# 参考 abap-adt-api objectcontents.ts 的 lock 实现。
|
||||
LOCK_RESULT_ACCEPT = (
|
||||
"application/*,application/vnd.sap.as+xml;charset=UTF-8;"
|
||||
"dataname=com.sap.adt.lock.result"
|
||||
)
|
||||
|
||||
|
||||
class DdicMixin:
|
||||
"""Object creation / deletion, DDIC operations, CDS, packages, ATC, pretty-print."""
|
||||
|
||||
@@ -24,7 +59,8 @@ class DdicMixin:
|
||||
obj_uri: str,
|
||||
corr_nr: str | None = None,
|
||||
) -> tuple[bool, str]:
|
||||
lock_handle, _ = self.lock(obj_uri, corr_nr)
|
||||
# DDIC 对象的 lock 端点对 Accept: */* 返回 406，需传 lock 结果专用 Accept 头
|
||||
lock_handle, _ = self.lock(obj_uri, corr_nr, accept=LOCK_RESULT_ACCEPT)
|
||||
url = f"{self.host}{obj_uri}"
|
||||
params: dict[str, str] = {"lockHandle": lock_handle}
|
||||
if corr_nr:
|
||||
@@ -193,6 +229,19 @@ class DdicMixin:
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</tabl:table>"
|
||||
)
|
||||
elif obj_type == "structure":
|
||||
# 结构体（TABL/DS）：参考 table，但使用 structures 命名空间
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<struc:structure xmlns:struc="http://www.sap.com/adt/dictionary/structures"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:type="TABL/DS"'
|
||||
f' adtcore:description="{desc}"'
|
||||
' adtcore:language="EN">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</struc:structure>"
|
||||
)
|
||||
elif obj_type == "tabletype":
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
@@ -204,6 +253,75 @@ class DdicMixin:
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</ttyp:abapTableType>"
|
||||
)
|
||||
elif obj_type == "include":
|
||||
# POST /programs/includes/{name} — include 使用 programs/includes 命名空间，
|
||||
# 不能复用 report 的 <program:abapProgram>（否则端点返回 400）。
|
||||
# 参考 abap-adt-api objectcreator.ts:
|
||||
# rootName="include:abapInclude"
|
||||
# nameSpace="http://www.sap.com/adt/programs/includes"
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<include:abapInclude xmlns:include="http://www.sap.com/adt/programs/includes"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="PROG/I"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</include:abapInclude>"
|
||||
)
|
||||
elif obj_type == "messageclass":
|
||||
# POST /oo/t100/messages/classes/{name} — body 含 class name + description
|
||||
# // TODO: verify XML structure on live SAP system
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<msgcls:t100MessageClass xmlns:msgcls="http://www.sap.com/adt/t100/message/classes"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
' adtcore:type="MSAG"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</msgcls:t100MessageClass>"
|
||||
)
|
||||
elif obj_type == "view":
|
||||
# POST /ddic/views/{name}（数据库视图）— body 结构参考 table 的创建
|
||||
# // TODO: verify XML structure on live SAP system
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<view:view xmlns:view="http://www.sap.com/adt/dictionary/views"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:type="VIEW"'
|
||||
f' adtcore:description="{desc}"'
|
||||
' adtcore:language="EN">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</view:view>"
|
||||
)
|
||||
elif obj_type == "searchhelp":
|
||||
# POST /ddic/searchhelps/{name} — body 含 search help 基本定义
|
||||
# // TODO: verify XML structure on live SAP system
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<shlp:searchHelp xmlns:shlp="http://www.sap.com/adt/dictionary/searchhelps"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:type="SHLP"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</shlp:searchHelp>"
|
||||
)
|
||||
elif obj_type == "lockobject":
|
||||
# POST /ddic/lockobjects/{name} — body 含 lock mode + table name
|
||||
# // TODO: verify XML structure on live SAP system (lock mode + table name)
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<lock:lockObject xmlns:lock="http://www.sap.com/adt/dictionary/lockobjects"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core"'
|
||||
f' adtcore:name="{name}"'
|
||||
f' adtcore:type="LOCK"'
|
||||
f' adtcore:description="{desc}">'
|
||||
f'<adtcore:packageRef adtcore:name="{package}"/>'
|
||||
"</lock:lockObject>"
|
||||
)
|
||||
raise ValueError(f"不支持的对象类型: {obj_type}")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
@@ -222,6 +340,31 @@ class DdicMixin:
|
||||
src_uri = config.format_src_uri(name.lower())
|
||||
|
||||
if obj_type in ("table", "structure"):
|
||||
# 先 POST 集合端点创建对象实体，再 PUT 写入 DDL 定义。
|
||||
# 之前直接 _put_ddl_source，PUT 到尚未创建的对象 /source/main 返回 405。
|
||||
obj_name = name.upper()
|
||||
body = self._build_create_body(obj_type, obj_name, obj_name)
|
||||
collection_url = config.collection_uri
|
||||
params: dict[str, str] = {}
|
||||
if corr_nr:
|
||||
params["corrNr"] = corr_nr
|
||||
hdrs = self._headers(config.create_content_type)
|
||||
logger.info(
|
||||
"CREATE DDIC ENTITY: POST %s name=%s",
|
||||
collection_url,
|
||||
obj_name,
|
||||
)
|
||||
resp = self.session.post(
|
||||
f"{self.host}{collection_url}",
|
||||
headers=hdrs,
|
||||
params=params,
|
||||
data=body.encode("utf-8"),
|
||||
)
|
||||
logger.info("CREATE DDIC ENTITY RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code >= 400:
|
||||
raise CreateError(
|
||||
f"创建 DDIC 对象实体失败: HTTP {resp.status_code} — {resp.text[:500]}"
|
||||
)
|
||||
self._put_ddl_source(src_uri, definition_body, "", corr_nr)
|
||||
else:
|
||||
obj_name = name.upper()
|
||||
@@ -754,3 +897,288 @@ class DdicMixin:
|
||||
logger.info("RUN PROGRAM RESPONSE: HTTP %s (%d bytes)", resp.status_code, len(resp.content))
|
||||
resp.raise_for_status()
|
||||
return resp.text
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# ABAP Unit tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def run_unit_test(self, obj_uri: str) -> dict[str, Any]:
|
||||
"""执行 ABAP Unit 测试。
|
||||
|
||||
Args:
|
||||
obj_uri: 被测对象的 ADT URI。
|
||||
|
||||
Returns:
|
||||
``{"summary": {...}, "classes": [{"name", "methods": [...]}]}``。
|
||||
每个方法项含 ``name``, ``duration``, ``alert``（失败原因）, ``line``。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/abapunit/testruns"
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<aunit:runConfiguration xmlns:aunit="http://www.sap.com/adt/aunit">'
|
||||
"<external><coverage active=\"false\"/></external>"
|
||||
"<options>"
|
||||
'<uriType value="semantic"/>'
|
||||
'<testDeterminationStrategy sameProgram="true" assignedTests="false"/>'
|
||||
'<testRiskLevels harmless="true" dangerous="true" critical="true"/>'
|
||||
'<testDurations short="true" medium="true" long="true"/>'
|
||||
'<withNavigationUri enabled="true"/>'
|
||||
"</options>"
|
||||
'<adtcore:objectSets xmlns:adtcore="http://www.sap.com/adt/core">'
|
||||
'<objectSet kind="inclusive">'
|
||||
"<adtcore:objectReferences>"
|
||||
f'<adtcore:objectReference adtcore:uri="{obj_uri}"/>'
|
||||
"</adtcore:objectReferences>"
|
||||
"</objectSet>"
|
||||
"</adtcore:objectSets>"
|
||||
"</aunit:runConfiguration>"
|
||||
)
|
||||
hdrs = self._headers("application/*")
|
||||
hdrs["Accept"] = "application/*"
|
||||
logger.info("RUN UNIT TEST: POST %s uri=%s", url, obj_uri)
|
||||
resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
|
||||
logger.info("RUN UNIT TEST RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
return self._parse_aunit_result(resp.content)
|
||||
|
||||
def _parse_aunit_result(self, content: bytes) -> dict[str, Any]:
|
||||
"""解析 ABAP Unit 测试结果 XML。
|
||||
|
||||
响应结构（aunit + adtcore 命名空间）::
|
||||
|
||||
<aunit:runResult>
|
||||
<summary tests="3" failures="1" errors="0" skipped="0" duration="450"/>
|
||||
<program adtcore:name="...">
|
||||
<testClasses>
|
||||
<testClass adtcore:name="ZCL_MY_TEST">
|
||||
<testMethods>
|
||||
<testMethod adtcore:name="test_foo" duration="0.45">
|
||||
<alert kind="t" title="assertion error">
|
||||
<stackInfo><info line="42"/></stackInfo>
|
||||
</alert>
|
||||
</testMethod>
|
||||
</testMethods>
|
||||
</testClass>
|
||||
</testClasses>
|
||||
</program>
|
||||
</aunit:runResult>
|
||||
|
||||
解析逻辑按本地标签名匹配，兼容命名空间前缀变化。
|
||||
"""
|
||||
# TODO: verify XML structure on live SAP system
|
||||
root = ET.fromstring(content)
|
||||
|
||||
summary = {
|
||||
"tests": "0", "failures": "0", "errors": "0",
|
||||
"skipped": "0", "duration": "0",
|
||||
}
|
||||
summary_el = _find_local(root, "summary")
|
||||
if summary_el is not None:
|
||||
for key in summary:
|
||||
summary[key] = summary_el.attrib.get(key, summary[key])
|
||||
|
||||
classes: list[dict[str, Any]] = []
|
||||
for tc in root.iter():
|
||||
if _local(tc.tag) != "testClass":
|
||||
continue
|
||||
cls_name = _attr_local(tc, "name")
|
||||
methods: list[dict[str, str]] = []
|
||||
for tm in tc.iter():
|
||||
if _local(tm.tag) != "testMethod":
|
||||
continue
|
||||
m_name = _attr_local(tm, "name")
|
||||
duration = tm.attrib.get("duration", "")
|
||||
alert_title = ""
|
||||
alert_line = ""
|
||||
for alert in tm.iter():
|
||||
if _local(alert.tag) != "alert":
|
||||
continue
|
||||
if not alert_title:
|
||||
alert_title = alert.attrib.get("title", "")
|
||||
info = _find_local(alert, "info")
|
||||
if info is not None:
|
||||
line = info.attrib.get("line", "")
|
||||
if line:
|
||||
alert_line = line
|
||||
methods.append({
|
||||
"name": m_name,
|
||||
"duration": duration,
|
||||
"alert": alert_title,
|
||||
"line": alert_line,
|
||||
})
|
||||
classes.append({"name": cls_name, "methods": methods})
|
||||
|
||||
return {"summary": summary, "classes": classes}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Version history
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_object_versions(self, obj_uri: str) -> list[dict[str, str]]:
|
||||
"""查询对象的版本历史。
|
||||
|
||||
Args:
|
||||
obj_uri: 对象 ADT URI。
|
||||
|
||||
Returns:
|
||||
版本列表，每项含 ``version``, ``author``, ``date``, ``versionTitle``。
|
||||
"""
|
||||
versions_href = self._get_versions_link(obj_uri)
|
||||
if not versions_href:
|
||||
return []
|
||||
|
||||
url = (
|
||||
versions_href
|
||||
if versions_href.startswith("http")
|
||||
else f"{self.host}{versions_href}"
|
||||
)
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "application/xml"
|
||||
logger.info("GET VERSIONS: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("GET VERSIONS RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
return self._parse_versions(resp.content)
|
||||
|
||||
def _get_versions_link(self, obj_uri: str) -> str | None:
|
||||
"""GET 对象结构 XML，提取 versions 关系链接的 href。"""
|
||||
url = f"{self.host}{obj_uri}"
|
||||
hdrs = self._headers()
|
||||
hdrs["Accept"] = "application/xml"
|
||||
logger.info("GET OBJECT STRUCTURE: GET %s", url)
|
||||
resp = self.session.get(url, headers=hdrs)
|
||||
logger.info("GET OBJECT STRUCTURE RESPONSE: HTTP %s", resp.status_code)
|
||||
if resp.status_code != 200:
|
||||
return None
|
||||
try:
|
||||
root = ET.fromstring(resp.content)
|
||||
except ET.ParseError:
|
||||
return None
|
||||
for el in root.iter():
|
||||
if _local(el.tag) != "link":
|
||||
continue
|
||||
if "versions" in el.attrib.get("rel", ""):
|
||||
href = el.attrib.get("href", "")
|
||||
if href:
|
||||
return href
|
||||
return None
|
||||
|
||||
def _parse_versions(self, content: bytes) -> list[dict[str, str]]:
|
||||
"""解析版本历史 feed。
|
||||
|
||||
典型结构（Atom feed，每个 entry 含版本元数据属性）::
|
||||
|
||||
<atom:feed>
|
||||
<atom:entry>
|
||||
<app:control>
|
||||
<d:version d:versionId="0001" d:author="DEVUSER"
|
||||
d:date="2026-06-10T14:20:00" d:versionTitle="Initial"/>
|
||||
</app:control>
|
||||
</atom:entry>
|
||||
</atom:feed>
|
||||
|
||||
按本地名（忽略命名空间）匹配属性，兼容 ``versionId`` / ``version``。
|
||||
"""
|
||||
# TODO: verify XML structure on live SAP system
|
||||
try:
|
||||
root = ET.fromstring(content)
|
||||
except ET.ParseError:
|
||||
return []
|
||||
|
||||
fields = ("version", "author", "date", "versionTitle")
|
||||
alias = {"versionId": "version"}
|
||||
|
||||
def extract(container: ET.Element) -> dict[str, str]:
|
||||
v = {f: "" for f in fields}
|
||||
for el in container.iter():
|
||||
for key, val in el.attrib.items():
|
||||
field = alias.get(_local(key), _local(key))
|
||||
if field in fields and not v[field]:
|
||||
v[field] = val
|
||||
return v
|
||||
|
||||
entries = [e for e in root.iter() if _local(e.tag) == "entry"]
|
||||
if entries:
|
||||
return [extract(e) for e in entries]
|
||||
|
||||
# 回退：无 entry 时，凡携带 version/versionId 的元素各成一行
|
||||
versions: list[dict[str, str]] = []
|
||||
for el in root.iter():
|
||||
if any(_local(k) in ("version", "versionId") for k in el.attrib):
|
||||
versions.append(extract(el))
|
||||
return versions
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Enhancements
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_enhancements(self, obj_uri: str) -> list[dict[str, Any]]:
|
||||
"""查询对象的增强实现（ENHO）。
|
||||
|
||||
Args:
|
||||
obj_uri: 被增强对象的 ADT URI。
|
||||
|
||||
Returns:
|
||||
增强实现列表，每项含 ``name``, ``type``, ``enhanced_name``,
|
||||
``enhanced_type``, ``elements``（含 ``name``/``type``/``mode``/``replacing``）。
|
||||
"""
|
||||
url = f"{self.host}/sap/bc/adt/enhancements"
|
||||
body = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<enh:enhancementSearch xmlns:enh="http://www.sap.com/adt/enhancements"'
|
||||
' xmlns:adtcore="http://www.sap.com/adt/core">'
|
||||
"<adtcore:objectReferences>"
|
||||
f'<adtcore:objectReference adtcore:uri="{obj_uri}"/>'
|
||||
"</adtcore:objectReferences>"
|
||||
"</enh:enhancementSearch>"
|
||||
)
|
||||
hdrs = self._headers()
|
||||
logger.info("GET ENHANCEMENTS: POST %s uri=%s", url, obj_uri)
|
||||
resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
|
||||
logger.info("GET ENHANCEMENTS RESPONSE: HTTP %s", resp.status_code)
|
||||
resp.raise_for_status()
|
||||
return self._parse_enhancements(resp.content)
|
||||
|
||||
def _parse_enhancements(self, content: bytes) -> list[dict[str, Any]]:
|
||||
"""解析增强实现列表 XML。
|
||||
|
||||
元素标签名与命名空间在不同 SAP 版本差异较大，故按本地名 + ENHO 类型码
|
||||
匹配增强实现，再在其下查找被增强对象与插件元素。
|
||||
"""
|
||||
# TODO: verify XML structure on live SAP system
|
||||
try:
|
||||
root = ET.fromstring(content)
|
||||
except ET.ParseError:
|
||||
return []
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
for el in root.iter():
|
||||
local = _local(el.tag)
|
||||
otype = _attr_local(el, "type")
|
||||
is_enh = local in (
|
||||
"enhancement", "enhancementImplementation", "enhancementSpotUse",
|
||||
) or "ENHO" in otype
|
||||
if not is_enh:
|
||||
continue
|
||||
|
||||
enhanced_obj = _find_local(el, "enhancedObject")
|
||||
if enhanced_obj is None:
|
||||
enhanced_obj = _find_local(el, "enhanced")
|
||||
elements: list[dict[str, str]] = []
|
||||
for pe in el.iter():
|
||||
if _local(pe.tag) not in ("pluginElement", "element", "sourceCodePluginElement"):
|
||||
continue
|
||||
elements.append({
|
||||
"name": _attr_local(pe, "elementName") or _attr_local(pe, "name"),
|
||||
"type": _attr_local(pe, "elementType") or _attr_local(pe, "type"),
|
||||
"mode": _attr_local(pe, "mode"),
|
||||
"replacing": _attr_local(pe, "replacing"),
|
||||
})
|
||||
results.append({
|
||||
"name": _attr_local(el, "name"),
|
||||
"type": otype or "ENHO",
|
||||
"enhanced_name": _attr_local(enhanced_obj, "name") if enhanced_obj is not None else "",
|
||||
"enhanced_type": _attr_local(enhanced_obj, "type") if enhanced_obj is not None else "",
|
||||
"elements": elements,
|
||||
})
|
||||
return results
|
||||
|
||||
Reference in New Issue
Block a user