吴让宇
c06e350f6d
新增能力： - remote-enable 子命令：设置函数模块处理类型为远程启用（bgRFC/RFC 执行体必需）。 实现要点均按实机验证：PUT fmodule:processingType="rfc"（枚举值仅小写 rfc 有效， remoteEnabled/remote/RFC 等均 400）；lockHandle 走 query 参数（放 header 报 ParameterNotFound）；PUT 后 GET 复核回显；DDIC 结构参数才可远程启用（类本地类型 报"针对 RFC 不允许使用类或接口的类型"）。 - create --type functiongroup 透传 --package / --corr_nr（此前 package 被静默写成 $TMP， 导致函数组无法归目标包）。 修复： - activate 请求体补重复引用：ADT CLAS 激活端点对 objectReferences 只有 1 个引用时 返回 HTTP 200 + 空 body 且不执行激活；引用数 ≥2 才真正激活（同对象重复亦可）。 - scanner 剥离 ADT 文件后缀（xxx.clas.abap 解析错误）。 守卫测试： - tests/unit/test_bgrfc_support.py（17 例）：函数组 package 透传、processingType 助手、 remote-enable 命令（含 lockHandle 走 query/小写 rfc/错误透传/复核失配判失败）、 函数 URI 组装不得用组名冒充模块名。 - test_activate.py 增加激活请求体引用数守卫；test_repo_guards.py 增加后缀剥离守卫。 测试：tests/unit 676 + tests/test_sapcli.py 69 = 745 全绿 （PYTHONPATH=assets python -m unittest discover -s tests/unit -t tests）
416 lines
18 KiB
Python
416 lines
18 KiB
Python
"""SourceMixin — source code read/write, lock/unlock, activate, syntax-check."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import xml.etree.ElementTree as ET
|
||
|
||
from sapcli.exceptions import (
|
||
ActivationError,
|
||
LockError,
|
||
SyntaxCheckError,
|
||
)
|
||
|
||
logger = logging.getLogger("sapcli.client")
|
||
|
||
|
||
class SourceMixin:
|
||
"""Source-code read/write and lock management."""
|
||
|
||
# ------------------------------------------------------------------
|
||
# Source read / write
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_source(self, src_uri: str) -> str:
|
||
url = f"{self.host}{src_uri}"
|
||
hdrs = self._headers()
|
||
hdrs["Accept"] = "text/plain"
|
||
logger.info("GET SOURCE: GET %s", url)
|
||
resp = self.session.get(url, headers=hdrs)
|
||
logger.info(
|
||
"GET SOURCE RESPONSE: HTTP %s, %d bytes",
|
||
resp.status_code,
|
||
len(resp.content) if resp.content else 0,
|
||
)
|
||
resp.raise_for_status()
|
||
return resp.text
|
||
|
||
def set_source(
|
||
self,
|
||
src_uri: str,
|
||
source: str,
|
||
lock_handle: str,
|
||
corr_nr: str | None = None,
|
||
content_type: str | None = None,
|
||
) -> bool:
|
||
params: dict[str, str] = {}
|
||
if lock_handle:
|
||
params["lockHandle"] = lock_handle
|
||
if corr_nr:
|
||
params["corrNr"] = corr_nr
|
||
url = f"{self.host}{src_uri}"
|
||
ct = content_type or "text/plain; charset=utf-8"
|
||
hdrs = self._headers(ct)
|
||
logger.info("SET SOURCE: PUT %s (%d chars, ct=%s)", url, len(source), ct)
|
||
resp = self.session.put(
|
||
url, headers=hdrs, params=params, data=source.encode("utf-8")
|
||
)
|
||
logger.info("SET SOURCE RESPONSE: HTTP %s", resp.status_code)
|
||
if resp.status_code >= 400:
|
||
logger.error("SET SOURCE ERROR BODY: %s", resp.text[:2000])
|
||
existing_nr = self._extract_locked_corrnr(resp.text)
|
||
if existing_nr and existing_nr != corr_nr:
|
||
logger.info("使用 corrNr=%s 重试 SET SOURCE", existing_nr)
|
||
params["corrNr"] = existing_nr
|
||
resp = self.session.put(
|
||
url, headers=hdrs, params=params, data=source.encode("utf-8")
|
||
)
|
||
logger.info("SET SOURCE RETRY RESPONSE: HTTP %s", resp.status_code)
|
||
if resp.status_code >= 400:
|
||
resp.raise_for_status()
|
||
return True
|
||
|
||
# ------------------------------------------------------------------
|
||
# Lock / unlock
|
||
# ------------------------------------------------------------------
|
||
|
||
def lock(
|
||
self,
|
||
obj_uri: str,
|
||
corr_nr: str | None = None,
|
||
accept: str | None = None,
|
||
) -> tuple[str, str | None]:
|
||
"""锁定对象。
|
||
|
||
Returns:
|
||
(lock_handle, effective_corr_nr) — 实际使用的传输请求号。
|
||
如果对象已绑定在某请求中，effective_corrnr 为该请求号；
|
||
否则为传入的 corr_nr（可能为 None）。
|
||
"""
|
||
self._stateful = True
|
||
url = f"{self.host}{obj_uri}"
|
||
params: dict[str, str] = {"_action": "LOCK", "accessMode": "MODIFY"}
|
||
effective_corr_nr = corr_nr
|
||
if corr_nr:
|
||
params["corrNr"] = corr_nr
|
||
hdrs = self._headers()
|
||
hdrs["Accept"] = accept or "*/*"
|
||
logger.info("LOCK: POST %s (corrNr=%s)", url, corr_nr or "none")
|
||
resp = self.session.post(url, headers=hdrs, params=params)
|
||
logger.info("LOCK RESPONSE: HTTP %s", resp.status_code)
|
||
if resp.status_code == 500:
|
||
existing_nr = self._extract_locked_corrnr(resp.text)
|
||
if existing_nr:
|
||
effective_corr_nr = existing_nr
|
||
logger.info("对象已锁定在请求 %s 中，使用该请求重试", existing_nr)
|
||
params["corrNr"] = existing_nr
|
||
resp = self.session.post(url, headers=hdrs, params=params, data="")
|
||
logger.info("LOCK RETRY RESPONSE: HTTP %s", resp.status_code)
|
||
if resp.status_code >= 400:
|
||
raise LockError(f"锁定失败: HTTP {resp.status_code}", obj_uri=obj_uri)
|
||
root = ET.fromstring(resp.content)
|
||
handle = root.findtext(".//LOCK_HANDLE")
|
||
if not handle:
|
||
for el in root.iter():
|
||
if "HANDLE" in el.tag.upper():
|
||
handle = el.text
|
||
break
|
||
if not handle:
|
||
raise LockError("锁定失败: 未获取到 lock handle", obj_uri=obj_uri)
|
||
|
||
# 从成功响应中提取 corrNr（对象可能已绑定传输请求）
|
||
if not effective_corr_nr:
|
||
extracted = self._extract_corrnr_from_lock_response(root)
|
||
if extracted:
|
||
effective_corr_nr = extracted
|
||
logger.info("从 lock 响应中检测到 corrNr: %s", extracted)
|
||
|
||
logger.info(
|
||
"Lock handle: %s (corrNr=%s)",
|
||
handle[:20] if len(handle) > 20 else handle,
|
||
effective_corr_nr or "none",
|
||
)
|
||
return handle, effective_corr_nr
|
||
|
||
def _extract_locked_corrnr(self, error_body: str) -> str | None:
|
||
try:
|
||
root = ET.fromstring(error_body)
|
||
for entry in root.iter():
|
||
if entry.attrib.get("key") == "corrNr":
|
||
corrnr = entry.text
|
||
if corrnr and corrnr != "*":
|
||
return corrnr
|
||
for el in root.iter():
|
||
if el.tag.endswith("message") and el.text:
|
||
m = re.search(r"request\s+(\w+)", el.text, re.IGNORECASE)
|
||
if m:
|
||
return m.group(1)
|
||
except Exception:
|
||
logger.debug("XML 解析提取 locked corrnr 失败", exc_info=True)
|
||
return None
|
||
|
||
def _extract_corrnr_from_lock_response(self, root: ET.Element) -> str | None:
|
||
"""从成功的 lock 响应 XML 中提取传输请求号。
|
||
|
||
常见格式:
|
||
<lock:lock ...>
|
||
<lock:LOCK_HANDLE>...</lock:LOCK_HANDLE>
|
||
<lock:CORRN>DEVK901362</lock:CORRN>
|
||
</lock:lock>
|
||
也可能是:
|
||
<adtcore:property adtcore:key="corrNr">DEVK901362</adtcore:property>
|
||
"""
|
||
# 策略 1: 直接找 CORRN 标签
|
||
for tag_name in ("CORRN", "corrNr", "corr_nr"):
|
||
text = root.findtext(f".//{tag_name}")
|
||
if text and text.strip() and text.strip() != "*":
|
||
return text.strip()
|
||
|
||
# 策略 2: 遍历所有元素，找标签或属性包含 corrNr 的
|
||
for el in root.iter():
|
||
# 检查属性
|
||
for attr_key in el.attrib:
|
||
if "corr" in attr_key.lower() and "nr" in attr_key.lower():
|
||
val = el.attrib[attr_key]
|
||
if val and val.strip() and val.strip() != "*":
|
||
return val.strip()
|
||
# 检查标签名包含 CORRN
|
||
if el.text and "corr" in el.tag.lower() and el.text.strip() and el.text.strip() != "*":
|
||
return el.text.strip()
|
||
|
||
# 策略 3: 找 key="corrNr" 的 property 元素
|
||
for el in root.iter():
|
||
if el.attrib.get("key") == "corrNr" and el.text:
|
||
val = el.text.strip()
|
||
if val and val != "*":
|
||
return val
|
||
|
||
return None
|
||
|
||
def unlock(self, obj_uri: str, lock_handle: str) -> bool:
|
||
url = f"{self.host}{obj_uri}"
|
||
params = {"_action": "UNLOCK", "lockHandle": lock_handle}
|
||
hdrs = self._headers("text/plain; charset=utf-8")
|
||
logger.info("UNLOCK: POST %s", url)
|
||
resp = self.session.post(url, headers=hdrs, params=params, data="")
|
||
logger.info("UNLOCK RESPONSE: HTTP %s", resp.status_code)
|
||
self._stateful = False
|
||
return resp.status_code == 200
|
||
|
||
# ------------------------------------------------------------------
|
||
# Activate / syntax check
|
||
# ------------------------------------------------------------------
|
||
|
||
def activate(
|
||
self,
|
||
name: str,
|
||
obj_uri: str,
|
||
corr_nr: str | None = None,
|
||
) -> tuple[bool, list[dict[str, str]]]:
|
||
# 实测（本机 SAP_BASIS 752，2026-09-15 受控实验）：
|
||
# objectReferences 里只有 1 个 objectReference 时，服务端返回
|
||
# HTTP 200 + 空 body（无 content-type），且根本不执行激活；
|
||
# 引用数为 2 及以上才真正激活（同一对象重复 2 次亦可）。
|
||
# 故单对象请求补一个重复引用，使请求体始终含 2 个引用。
|
||
refs = (
|
||
f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"/>'
|
||
f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"/>'
|
||
)
|
||
body = (
|
||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||
'<adtcore:objectReferences xmlns:adtcore="http://www.sap.com/adt/core">'
|
||
f'{refs}'
|
||
"</adtcore:objectReferences>"
|
||
)
|
||
self._stateful = False
|
||
url = f"{self.host}/sap/bc/adt/activation"
|
||
params: dict[str, str] = {"method": "activate"}
|
||
if corr_nr:
|
||
params["corrNr"] = corr_nr
|
||
hdrs = self._headers()
|
||
logger.info("ACTIVATE: POST %s name=%s corrNr=%s", url, name, corr_nr or "none")
|
||
resp = self.session.post(url, headers=hdrs, params=params, data=body)
|
||
logger.info(
|
||
"ACTIVATE RESPONSE: HTTP %s CT=%s",
|
||
resp.status_code,
|
||
resp.headers.get("content-type", "")[:80],
|
||
)
|
||
|
||
if resp.status_code != 200:
|
||
logger.error("Activation HTTP %s", resp.status_code)
|
||
raise ActivationError(f"激活失败: HTTP {resp.status_code}")
|
||
|
||
ct = resp.headers.get("content-type", "")
|
||
if "inactivectsobjects" in ct:
|
||
root = ET.fromstring(resp.content)
|
||
ioc_ns = "http://www.sap.com/abapxml/inactiveCtsObjects"
|
||
inactive_objects: list[str] = []
|
||
for entry in root.findall(f".//{{{ioc_ns}}}entry"):
|
||
obj_elem = entry.find(f"{{{ioc_ns}}}object")
|
||
if obj_elem is not None:
|
||
ref = obj_elem.find(".//{http://www.sap.com/adt/core}ref")
|
||
if ref is None:
|
||
for child in obj_elem:
|
||
if child.tag.endswith("}ref"):
|
||
ref = child
|
||
break
|
||
if ref is not None:
|
||
inactive_name = ref.attrib.get(
|
||
"{http://www.sap.com/adt/core}name", ""
|
||
)
|
||
inactive_type = ref.attrib.get(
|
||
"{http://www.sap.com/adt/core}type", ""
|
||
)
|
||
if inactive_name:
|
||
inactive_objects.append(f"{inactive_name} ({inactive_type})")
|
||
if inactive_objects:
|
||
logger.warning("激活返回未激活对象: %s", ", ".join(inactive_objects))
|
||
msg_text = f"Objects still inactive: {', '.join(inactive_objects)}"
|
||
messages: list[dict[str, str]] = [
|
||
{"type": "E", "line": "?", "text": msg_text, "href": ""}
|
||
]
|
||
logger.info("Activation result: FAILED (inactive objects)")
|
||
return False, messages
|
||
logger.info("Activation result: SUCCESS (inactiveObjects 响应但无对象列出)")
|
||
return True, []
|
||
|
||
if not resp.content or not resp.text.strip():
|
||
# NW 7.40 返回空响应但可能并未真正激活
|
||
# 验证对象实际状态
|
||
logger.info("Activation returned empty response, verifying object status...")
|
||
try:
|
||
verify_resp = self.session.get(
|
||
f"{self.host}{obj_uri}",
|
||
headers=self._headers(),
|
||
)
|
||
if verify_resp.status_code == 200:
|
||
vroot = ET.fromstring(verify_resp.content)
|
||
version = vroot.attrib.get("{http://www.sap.com/adt/core}version", "")
|
||
if version == "inactive":
|
||
# NW 7.40: first activation may only "stage" the change.
|
||
# Retry activation once, then re-verify.
|
||
logger.info("Object still inactive, retrying activation (NW 7.40 double-activate workaround)...")
|
||
retry_resp = self.session.post(
|
||
url, headers=hdrs, params=params, data=body,
|
||
)
|
||
logger.info(
|
||
"RETRY ACTIVATE: HTTP %s len=%s",
|
||
retry_resp.status_code, len(retry_resp.content),
|
||
)
|
||
if retry_resp.status_code == 200:
|
||
# Re-verify after retry
|
||
verify_resp2 = self.session.get(
|
||
f"{self.host}{obj_uri}",
|
||
headers=self._headers(),
|
||
)
|
||
if verify_resp2.status_code == 200:
|
||
vroot2 = ET.fromstring(verify_resp2.content)
|
||
version2 = vroot2.attrib.get(
|
||
"{http://www.sap.com/adt/core}version", ""
|
||
)
|
||
if version2 == "active":
|
||
logger.info("Retry activation succeeded!")
|
||
return True, []
|
||
logger.warning("Object still inactive after retry (NW 7.40)")
|
||
messages = [
|
||
{
|
||
"type": "W",
|
||
"line": "?",
|
||
"text": "ADT 激活返回空响应，对象仍为 inactive（NW 7.40 已知限制）。请在 SAP GUI SE09 手动激活。",
|
||
"href": "",
|
||
}
|
||
]
|
||
return False, messages
|
||
elif version == "active":
|
||
logger.info("Verified: object is active after activation")
|
||
except Exception as e:
|
||
logger.warning("Could not verify activation status: %s", e)
|
||
logger.info("Activation success (空响应)")
|
||
return True, []
|
||
|
||
root = ET.fromstring(resp.content)
|
||
messages = []
|
||
for msg in root.iter():
|
||
msg_type = msg.attrib.get("type", "")
|
||
if msg_type in ("E", "W", "I", "S"):
|
||
line = msg.attrib.get("line", "?")
|
||
href = msg.attrib.get("href", "")
|
||
txt = ""
|
||
for child in msg.iter():
|
||
if child.text and child.tag.endswith("}txt"):
|
||
txt = child.text
|
||
break
|
||
if not txt:
|
||
for child in msg.iter():
|
||
if child.text and len(child.text.strip()) > 3:
|
||
txt = child.text.strip()
|
||
break
|
||
messages.append({"type": msg_type, "line": line, "text": txt, "href": href})
|
||
logger.info("Activation msg [%s] line=%s: %s", msg_type, line, txt)
|
||
|
||
errors = [m for m in messages if m["type"] == "E"]
|
||
success = len(errors) == 0
|
||
logger.info(
|
||
"Activation result: %s (%d errors)",
|
||
"SUCCESS" if success else "FAILED",
|
||
len(errors),
|
||
)
|
||
return success, messages
|
||
|
||
def syntax_check(
|
||
self,
|
||
name: str,
|
||
obj_uri: str,
|
||
) -> tuple[bool, list[dict[str, str]]]:
|
||
body = (
|
||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||
'<adtcore:objectReferences xmlns:adtcore="http://www.sap.com/adt/core">'
|
||
f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"/>'
|
||
"</adtcore:objectReferences>"
|
||
)
|
||
self._stateful = False
|
||
url = f"{self.host}/sap/bc/adt/activation"
|
||
params = {"method": "check"}
|
||
hdrs = self._headers()
|
||
logger.info("SYNTAX CHECK: POST %s name=%s", url, name)
|
||
resp = self.session.post(url, headers=hdrs, params=params, data=body)
|
||
logger.info("SYNTAX CHECK RESPONSE: HTTP %s", resp.status_code)
|
||
|
||
if resp.status_code != 200:
|
||
logger.error("Syntax check HTTP %s", resp.status_code)
|
||
raise SyntaxCheckError(f"语法检查失败: HTTP {resp.status_code}")
|
||
|
||
if not resp.content or not resp.text.strip():
|
||
logger.info("Syntax check: OK (空响应)")
|
||
return True, []
|
||
|
||
root = ET.fromstring(resp.content)
|
||
messages: list[dict[str, str]] = []
|
||
for msg in root.iter():
|
||
msg_type = msg.attrib.get("type", "")
|
||
if msg_type in ("E", "W", "I", "S"):
|
||
line = msg.attrib.get("line", "?")
|
||
href = msg.attrib.get("href", "")
|
||
txt = ""
|
||
for child in msg.iter():
|
||
if child.text and child.tag.endswith("}txt"):
|
||
txt = child.text
|
||
break
|
||
if not txt:
|
||
for child in msg.iter():
|
||
if child.text and len(child.text.strip()) > 3:
|
||
txt = child.text.strip()
|
||
break
|
||
messages.append({"type": msg_type, "line": line, "text": txt, "href": href})
|
||
logger.info("Syntax check msg [%s] line=%s: %s", msg_type, line, txt)
|
||
|
||
errors = [m for m in messages if m["type"] == "E"]
|
||
success = len(errors) == 0
|
||
logger.info(
|
||
"Syntax check result: %s (%d errors)",
|
||
"OK" if success else "ERRORS",
|
||
len(errors),
|
||
)
|
||
return success, messages
|