feat(info): DDIC 类型补全描述与开发包(NW 7.40 ADT 不返回)

NW 7.40 的 */* 响应不含 description/packageRef,info 对 structure/table/tabletype/dataelement/domain 在 ADT 返回值为空时回退查文本表补描述(DD02T/DD40T/DD04T/DD01T)、查 TADIR 补开发包。语言按「登录语言 -> E -> 1 -> 首行」回退;补查异常静默降级不打断输出;TADIR 无行时保持 (未指定)。
This commit is contained in:
吴让宇
2026-09-11 01:39:05 +08:00
parent a58275aa51
commit d0d65db137
2 changed files with 355 additions and 0 deletions
+133
View File
@@ -23,6 +23,35 @@ from sapcli.types import get_r3tr_object_code, get_type_config, parse_object_nam
DDIC_TYPES = {"domain", "dataelement", "table", "structure", "tabletype"} DDIC_TYPES = {"domain", "dataelement", "table", "structure", "tabletype"}
# NW 7.40 ADT 对 DDIC 对象(structure/table/tabletype/dataelement/domain)的 */* 响应
# 不含 description / packageRef,需回退查文本表补描述、查 TADIR 补开发包。
# 描述表的语言列固定为 DDLANGUAGE、文本列固定为 DDTEXT。
_DDIC_DESC_TABLE: dict[str, tuple[str, str]] = {
"structure": ("DD02T", "TABNAME"),
"table": ("DD02T", "TABNAME"),
"tabletype": ("DD40T", "TYPENAME"),
"dataelement": ("DD04T", "ROLLNAME"),
"domain": ("DD01T", "DOMNAME"),
}
# DDIC 对象 → TADIR.OBJECT 代码(用于补开发包 DEVCLASS)。
_DDIC_PACKAGE_OBJECT: dict[str, str] = {
"structure": "TABL",
"table": "TABL",
"tabletype": "TTYP",
"dataelement": "DTEL",
"domain": "DOMA",
}
# ADT masterLanguage 返回 ISO 码(如 'ZH'/'EN'),文本表 DDLANGUAGE 用 SAP 单字符键。
_ISO_TO_SAP_LANGUAGE: dict[str, str] = {
"ZH": "1",
"EN": "E",
"DE": "D",
"JA": "J",
"KO": "3",
}
logger = logging.getLogger("sapcli.commands.crud") logger = logging.getLogger("sapcli.commands.crud")
_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) _PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -426,6 +455,101 @@ def _query_transport_request(
return "(无法获取)" return "(无法获取)"
def _sap_language_key(master_language: str) -> str:
"""把 ADT masterLanguageISO 码,如 'ZH'/'EN')转成 SAP 单字符语言键。
已是单字符时原样返回;无法映射时原样返回(回退顺序会自然跳过不匹配项)。
"""
lang = (master_language or "").strip().upper()
if not lang:
return ""
if len(lang) == 1:
return lang
return _ISO_TO_SAP_LANGUAGE.get(lang, lang)
def _pick_description_text(rows, preferred_lang: str = "") -> str:
"""按「登录语言 -> 'E' -> '1' -> 首行」回退,从 (语言, 文本) 行中挑描述文本。"""
if not rows:
return ""
by_lang: dict[str, str] = {}
order: list[str] = []
for lang, text in rows:
lang = (lang or "").strip().upper()
if lang not in by_lang:
by_lang[lang] = (text or "").strip()
order.append(lang)
priority = [p for p in (preferred_lang, "E", "1") if p]
for p in priority:
p = p.strip().upper()
for lang in order:
if lang == p and by_lang[lang]:
return by_lang[lang]
for lang in order:
if by_lang[lang]:
return by_lang[lang]
return ""
def _query_ddic_description(
client: ADTClient, obj_type: str, obj_name: str, preferred_lang: str
) -> str:
"""ADT 返回空描述时,查描述表补 DDIC 对象描述;失败返回 '' 不抛。"""
if obj_type not in _DDIC_DESC_TABLE:
return ""
table, key_col = _DDIC_DESC_TABLE[obj_type]
name_upper = obj_name.upper().split("/")[-1]
if not re.fullmatch(r"[A-Za-z0-9_/$]+", name_upper):
return ""
try:
sql = (
f"SELECT DDLANGUAGE, DDTEXT FROM {table} "
f"WHERE {key_col} IN ('{name_upper}')"
)
result = client.query_table_data(sql, max_rows=20)
columns = result.get("columns", [])
rows = result.get("rows", [])
lang_idx = columns.index("DDLANGUAGE") if "DDLANGUAGE" in columns else 0
text_idx = columns.index("DDTEXT") if "DDTEXT" in columns else 1
pairs = []
for row in rows:
if row is None:
continue
lang = row[lang_idx] if lang_idx < len(row) else ""
text = row[text_idx] if text_idx < len(row) else ""
pairs.append((lang, text))
return _pick_description_text(pairs, preferred_lang)
except Exception:
logger.debug("查询 %s 描述失败", table, exc_info=True)
return ""
def _query_ddic_package(client: ADTClient, obj_type: str, obj_name: str) -> str:
"""ADT 返回空开发包时,查 TADIR 补 DEVCLASS;失败返回 '' 不抛。"""
if obj_type not in _DDIC_PACKAGE_OBJECT:
return ""
obj_code = _DDIC_PACKAGE_OBJECT[obj_type]
name_upper = obj_name.upper().split("/")[-1]
if not re.fullmatch(r"[A-Za-z0-9_/$]+", name_upper):
return ""
try:
sql = (
f"SELECT DEVCLASS FROM TADIR "
f"WHERE OBJECT IN ('{obj_code}') AND OBJ_NAME IN ('{name_upper}')"
)
result = client.query_table_data(sql, max_rows=10)
columns = result.get("columns", [])
rows = result.get("rows", [])
dev_idx = columns.index("DEVCLASS") if "DEVCLASS" in columns else 0
for row in rows:
if row and dev_idx < len(row) and row[dev_idx]:
return str(row[dev_idx]).strip()
return ""
except Exception:
logger.debug("查询 TADIR 开发包失败", exc_info=True)
return ""
def cmd_info(args: argparse.Namespace, client: ADTClient) -> None: def cmd_info(args: argparse.Namespace, client: ADTClient) -> None:
"""查询对象元数据信息。""" """查询对象元数据信息。"""
obj_type: str = args.type obj_type: str = args.type
@@ -505,6 +629,15 @@ def cmd_info(args: argparse.Namespace, client: ADTClient) -> None:
else "" else ""
) )
# NW 7.40 DDIC 的 */* 响应不含 description/packageRef:仅在 ADT 返回值为空时补查,
# 补查失败静默降级(保持原样),不打断 info 输出。
if not info_desc:
info_desc = _query_ddic_description(
client, obj_type, parsed.display_name, _sap_language_key(info_language)
)
if not info_package:
info_package = _query_ddic_package(client, obj_type, parsed.display_name)
if info_version == "active": if info_version == "active":
status_icon = "✓ 已激活" status_icon = "✓ 已激活"
elif info_version == "inactive": elif info_version == "inactive":
+222
View File
@@ -0,0 +1,222 @@
"""info 命令 DDIC 描述/开发包补全 — 单元测试。
覆盖四项行为(协调者 TDD 要求):
1. ADT 已返回描述/开发包时不得补查
2. 补查成功填充描述/开发包
3. 补查失败保持原样且不抛
4. 描述文本语言回退顺序(登录语言 -> 'E' -> '1' -> 首行)
"""
from __future__ import annotations
import argparse
import os
import sys
import unittest
from unittest.mock import MagicMock, patch
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "assets"))
from sapcli.client import ADTClient
from sapcli.commands.crud import (
_pick_description_text,
_query_ddic_description,
_query_ddic_package,
_sap_language_key,
)
def _mock_resp(status_code=200, text="", content=b"", headers=None):
r = MagicMock()
r.status_code = status_code
r.text = text
r.content = content
r.headers = headers or {}
r.raise_for_status = MagicMock()
return r
def _make_client():
client = MagicMock(spec=ADTClient)
client.host = "https://sap.example.com"
client.sap_client = "100"
client.csrf_token = "test-csrf-token"
client.session = MagicMock()
client._headers.side_effect = lambda content_type="application/xml": {
"content-type": content_type,
"x-csrf-token": "test-csrf-token",
"Accept": "*/*",
}
return client
def _args(**kwargs):
defaults = {"type": "structure", "name": "ZMM_BIP_001_HEADER", "path": ".", "config": None}
defaults.update(kwargs)
return argparse.Namespace(**defaults)
def _ddic_xml(name="ZMM_BIP_001_HEADER", otype="TABL/DS", version="active",
description="", package=None, master_language="ZH"):
attrs = (
f' adtcore:name="{name}"'
f' adtcore:type="{otype}"'
f' adtcore:version="{version}"'
f' adtcore:masterLanguage="{master_language}"'
)
if description:
attrs += f' adtcore:description="{description}"'
head = (
b'<?xml version="1.0"?>'
b'<adtcore:object xmlns:adtcore="http://www.sap.com/adt/core"'
+ attrs.encode()
)
if package is None:
return head + b'/>'
return (
head + b'>'
b'<adtcore:packageRef adtcore:name="' + package.encode() + b'"/>'
b'</adtcore:object>'
)
class TestLanguageFallback(unittest.TestCase):
def test_prefers_login_language(self):
rows = [("1", "中文描述"), ("E", "English desc")]
self.assertEqual(_pick_description_text(rows, "1"), "中文描述")
def test_prefers_E_when_login_language_missing(self):
rows = [("1", "中文描述"), ("E", "English desc")]
self.assertEqual(_pick_description_text(rows, ""), "English desc")
def test_E_before_1(self):
rows = [("E", "English desc"), ("1", "中文描述")]
self.assertEqual(_pick_description_text(rows, "D"), "English desc")
def test_falls_back_to_first_row_when_no_priority_match(self):
rows = [("F", "French"), ("D", "Deutsch")]
self.assertEqual(_pick_description_text(rows, ""), "French")
def test_empty_rows_returns_empty(self):
self.assertEqual(_pick_description_text([], "E"), "")
self.assertEqual(_pick_description_text(None, "E"), "")
def test_sap_language_key_mapping(self):
self.assertEqual(_sap_language_key("ZH"), "1")
self.assertEqual(_sap_language_key("EN"), "E")
self.assertEqual(_sap_language_key(""), "")
self.assertEqual(_sap_language_key("E"), "E")
class TestQueryDdicDescription(unittest.TestCase):
def test_structure_queries_dd02t_and_fills(self):
client = _make_client()
client.query_table_data.return_value = {
"columns": ["DDLANGUAGE", "DDTEXT"],
"rows": [["E", "BIP vendor header"]],
}
result = _query_ddic_description(client, "structure", "ZMM_BIP_001_HEADER", "")
self.assertEqual(result, "BIP vendor header")
sql = client.query_table_data.call_args[0][0]
self.assertIn("DD02T", sql)
self.assertIn("TABNAME IN ('ZMM_BIP_001_HEADER')", sql)
def test_tabletype_queries_dd40t(self):
client = _make_client()
client.query_table_data.return_value = {
"columns": ["DDLANGUAGE", "DDTEXT"],
"rows": [["1", "BIP bank rows"]],
}
result = _query_ddic_description(client, "tabletype", "ZMM_BIP_001_BANK_T", "")
self.assertEqual(result, "BIP bank rows")
sql = client.query_table_data.call_args[0][0]
self.assertIn("DD40T", sql)
self.assertIn("TYPENAME IN ('ZMM_BIP_001_BANK_T')", sql)
def test_non_ddic_type_returns_empty_without_query(self):
client = _make_client()
self.assertEqual(_query_ddic_description(client, "class", "ZCL_X", ""), "")
client.query_table_data.assert_not_called()
def test_failure_returns_empty(self):
client = _make_client()
client.query_table_data.side_effect = Exception("boom")
self.assertEqual(_query_ddic_description(client, "structure", "X", ""), "")
class TestQueryDdicPackage(unittest.TestCase):
def test_structure_queries_tadir(self):
client = _make_client()
client.query_table_data.return_value = {
"columns": ["DEVCLASS"],
"rows": [["ZMM"]],
}
result = _query_ddic_package(client, "structure", "ZMM_BIP_001_HEADER")
self.assertEqual(result, "ZMM")
sql = client.query_table_data.call_args[0][0]
self.assertIn("TADIR", sql)
self.assertIn("OBJECT IN ('TABL')", sql)
self.assertIn("OBJ_NAME IN ('ZMM_BIP_001_HEADER')", sql)
def test_no_rows_returns_empty(self):
client = _make_client()
client.query_table_data.return_value = {"columns": ["DEVCLASS"], "rows": []}
self.assertEqual(_query_ddic_package(client, "structure", "ZMM_BIP_001_HEADER"), "")
def test_failure_returns_empty(self):
client = _make_client()
client.query_table_data.side_effect = Exception("boom")
self.assertEqual(_query_ddic_package(client, "structure", "X"), "")
class TestCmdInfoSupplement(unittest.TestCase):
def test_no_supplement_when_values_present(self):
from sapcli.commands.crud import cmd_info
client = _make_client()
client.session.get.return_value = _mock_resp(
200, content=_ddic_xml(description="My desc", package="ZPKG")
)
client.query_table_data.return_value = {"columns": [], "rows": []}
with patch("builtins.print"):
cmd_info(_args(type="structure", name="ZMM_BIP_001_HEADER"), client)
sqls = [str(c[0][0]) for c in client.query_table_data.call_args_list]
self.assertFalse(any(t in s for t in ("DD02T", "DD40T", "DD04T", "DD01T") for s in sqls))
self.assertFalse(any("TADIR" in s for s in sqls))
def test_supplements_description_and_package(self):
from sapcli.commands.crud import cmd_info
client = _make_client()
client.session.get.return_value = _mock_resp(200, content=_ddic_xml())
def fake_query(sql, max_rows=200):
if "DD02T" in sql:
return {"columns": ["DDLANGUAGE", "DDTEXT"], "rows": [["E", "BIP vendor header"]]}
if "TADIR" in sql:
return {"columns": ["DEVCLASS"], "rows": [["ZX_TEST_PKG"]]}
return {"columns": [], "rows": []}
client.query_table_data.side_effect = fake_query
with patch("builtins.print") as mock_print:
cmd_info(_args(type="structure", name="ZMM_BIP_001_HEADER"), client)
printed = " ".join(str(c) for c, _ in mock_print.call_args_list)
self.assertIn("BIP vendor header", printed)
self.assertIn("ZX_TEST_PKG", printed)
def test_supplement_failure_keeps_original(self):
from sapcli.commands.crud import cmd_info
client = _make_client()
client.session.get.return_value = _mock_resp(200, content=_ddic_xml())
def fake_query(sql, max_rows=200):
if "DD02T" in sql or "TADIR" in sql:
raise Exception("boom")
return {"columns": [], "rows": []}
client.query_table_data.side_effect = fake_query
with patch("builtins.print") as mock_print:
cmd_info(_args(type="structure", name="ZMM_BIP_001_HEADER"), client)
printed = " ".join(str(c) for c, _ in mock_print.call_args_list)
self.assertIn("(未指定)", printed)
if __name__ == "__main__":
unittest.main()