diff --git a/assets/sapcli/commands/crud.py b/assets/sapcli/commands/crud.py index 66d876e..dd24266 100644 --- a/assets/sapcli/commands/crud.py +++ b/assets/sapcli/commands/crud.py @@ -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"} +# 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") _PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -426,6 +455,101 @@ def _query_transport_request( return "(无法获取)" +def _sap_language_key(master_language: str) -> str: + """把 ADT masterLanguage(ISO 码,如 '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: """查询对象元数据信息。""" obj_type: str = args.type @@ -505,6 +629,15 @@ def cmd_info(args: argparse.Namespace, client: ADTClient) -> None: 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": status_icon = "✓ 已激活" elif info_version == "inactive": diff --git a/tests/unit/test_info_ddic_enhancement.py b/tests/unit/test_info_ddic_enhancement.py new file mode 100644 index 0000000..7103549 --- /dev/null +++ b/tests/unit/test_info_ddic_enhancement.py @@ -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'' + b'' + return ( + head + b'>' + b'' + b'' + ) + + +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()