吴让宇
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）
297 lines
12 KiB
Python
297 lines
12 KiB
Python
"""bgRFC 支撑能力守卫测试（B1 收尾并行件）。
|
||
|
||
守住三处「静默失效」类不变量，全部 mock/离线，不连接 SAP：
|
||
|
||
1. **create --type functiongroup 的 --package 透传**：包名必须进入创建 body 的
|
||
``packageRef``（否则 FUGR 静默落 $TMP，无法归 ZINT 包）。
|
||
2. **FM 远程启用**：``processingType`` 只认小写 ``rfc``；lockHandle 必须走 query
|
||
参数（放 header 报 ExceptionParameterNotFound）；PUT 后 GET 复核回显。
|
||
3. **function 类型的 '组/模块' URL 组装**：``fmodules/{name}`` 段必须是模块名，
|
||
不是组名（曾出现 URL 误用组名 → 400）。
|
||
|
||
运行: python -m unittest tests.unit.test_bgrfc_support -v
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import unittest
|
||
from argparse import Namespace
|
||
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.client._ddic import _set_processing_type, _get_processing_type
|
||
from sapcli.exceptions import InvalidNameError, SapCliError
|
||
from sapcli.types import parse_object_name
|
||
|
||
ADTCORE_NS = "http://www.sap.com/adt/core"
|
||
FM_NS = "http://www.sap.com/adt/functions/fmodules"
|
||
LOCK_NS = "http://www.sap.com/adt/lock"
|
||
|
||
|
||
def _make_client() -> ADTClient:
|
||
"""创建 mock 好的 ADTClient（绕过 __init__，手动注入依赖）。"""
|
||
client = ADTClient.__new__(ADTClient)
|
||
client.host = "https://sap.example.com"
|
||
client.sap_client = "100"
|
||
client.user = "TESTUSER"
|
||
client.password = "TESTPASS"
|
||
client.csrf_token = "test-csrf-token"
|
||
client.session = MagicMock()
|
||
client._stateful = False
|
||
return client
|
||
|
||
|
||
def _make_mock_client() -> MagicMock:
|
||
"""全 mock 客户端（命令级测试用，方法均为 MagicMock）。"""
|
||
client = MagicMock(spec=ADTClient)
|
||
client.host = "https://sap.example.com"
|
||
client.sap_client = "100"
|
||
client.csrf_token = "test-csrf-token"
|
||
client.user = "TESTUSER"
|
||
client.session = MagicMock()
|
||
return client
|
||
|
||
|
||
def _mock_resp(status_code=200, text="", content=b"", headers=None):
|
||
resp = MagicMock()
|
||
resp.status_code = status_code
|
||
resp.text = text
|
||
resp.content = content
|
||
resp.headers = headers or {}
|
||
resp.raise_for_status = MagicMock()
|
||
return resp
|
||
|
||
|
||
def _fm_xml(processing_type: str) -> str:
|
||
"""构造带 processingType 的 FM 元数据 XML。"""
|
||
return (
|
||
'<?xml version="1.0"?>'
|
||
f'<fm:abapFunctionModule xmlns:fm="{FM_NS}" xmlns:adtcore="{ADTCORE_NS}">'
|
||
f"<fm:processingType>{processing_type}</fm:processingType>"
|
||
"</fm:abapFunctionModule>"
|
||
)
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 改动 1：functiongroup 的 --package 透传
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestFunctionGroupPackagePassThrough(unittest.TestCase):
|
||
"""FUGR 创建必须把 package 写进 body 的 packageRef，否则静默落 $TMP。"""
|
||
|
||
def test_create_function_group_body_has_package_ref(self):
|
||
client = _make_client()
|
||
client.session.post.return_value = _mock_resp(201)
|
||
|
||
self.assertTrue(client.create_function_group("ZFG", "desc", package="ZINT"))
|
||
|
||
_, kwargs = client.session.post.call_args
|
||
body = kwargs["data"].decode("utf-8")
|
||
self.assertIn("packageRef", body, "FUGR 创建 body 必须含 packageRef")
|
||
self.assertIn("ZINT", body, "packageRef 必须写 ZINT，而非 $TMP")
|
||
self.assertNotIn("$TMP", body)
|
||
|
||
def test_create_function_group_default_package_is_tmp(self):
|
||
"""未显式传 package 时仍默认 $TMP（向后兼容）。"""
|
||
client = _make_client()
|
||
client.session.post.return_value = _mock_resp(201)
|
||
client.create_function_group("ZFG", "desc")
|
||
_, kwargs = client.session.post.call_args
|
||
body = kwargs["data"].decode("utf-8")
|
||
self.assertIn("$TMP", body)
|
||
|
||
@patch("sapcli.commands.crud._select_transport_request", return_value="DEVK001")
|
||
def test_cmd_create_functiongroup_passes_package(self, mock_transport):
|
||
"""cmd_create --type functiongroup --package ZINT → create_function_group 收到 package。"""
|
||
from sapcli.commands.crud import cmd_create
|
||
|
||
client = _make_mock_client()
|
||
client.function_group_exists.return_value = False
|
||
client.create_function_group.return_value = True
|
||
|
||
cmd_create(
|
||
Namespace(
|
||
type="functiongroup", name="ZFG", description="d",
|
||
source=None, definition=None, package="ZINT", corr_nr=None,
|
||
path=None,
|
||
),
|
||
client,
|
||
)
|
||
|
||
client.create_function_group.assert_called_once()
|
||
pos = client.create_function_group.call_args[0]
|
||
# 签名: (group_name, description, corr_nr, package)
|
||
self.assertEqual(pos[0], "ZFG")
|
||
self.assertEqual(pos[3], "ZINT", "package 必须透传到 create_function_group")
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 改动 2：FM 远程启用（processingType=rfc + lockHandle query）
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestProcessingTypeHelpers(unittest.TestCase):
|
||
"""_set_processing_type / _get_processing_type 的 XML 改写。"""
|
||
|
||
def test_element_form_set_and_get(self):
|
||
out = _set_processing_type(_fm_xml("normal"), "rfc")
|
||
self.assertEqual(_get_processing_type(out), "rfc")
|
||
self.assertIn("rfc", out)
|
||
self.assertNotIn(">normal<", out)
|
||
|
||
def test_attribute_form_set_and_get(self):
|
||
xml = (
|
||
'<?xml version="1.0"?>'
|
||
f'<fm:abapFunctionModule xmlns:fm="{FM_NS}" processingType="normal"/>'
|
||
)
|
||
out = _set_processing_type(xml, "rfc")
|
||
self.assertEqual(_get_processing_type(out), "rfc")
|
||
|
||
def test_missing_processing_type_raises(self):
|
||
xml = f'<fm:abapFunctionModule xmlns:fm="{FM_NS}"/>'
|
||
with self.assertRaises(SapCliError):
|
||
_set_processing_type(xml, "rfc")
|
||
|
||
def test_get_processing_type_missing_returns_empty(self):
|
||
self.assertEqual(_get_processing_type(f'<x xmlns:x="urn:x"/>'), "")
|
||
|
||
|
||
class TestSetFunctionRemoteEnabled(unittest.TestCase):
|
||
"""客户端 set_function_remote_enabled 的 GET→PUT→GET 闭环。"""
|
||
|
||
def _setup(self):
|
||
client = _make_client()
|
||
client.session.get.side_effect = [
|
||
_mock_resp(200, text=_fm_xml("normal")), # 首次 GET 原 XML
|
||
_mock_resp(200, text=_fm_xml("rfc")), # PUT 后 GET 复核
|
||
]
|
||
client.session.put.return_value = _mock_resp(200)
|
||
client.lock = MagicMock(return_value=("LH123", None))
|
||
client.unlock = MagicMock(return_value=True)
|
||
return client
|
||
|
||
def test_put_lockhandle_in_query_not_header(self):
|
||
client = self._setup()
|
||
ok, err = client.set_function_remote_enabled("ZZ_RFC_PROBE", "ZZ_RFC_PROBE_FM")
|
||
|
||
self.assertTrue(ok)
|
||
self.assertEqual(err, "")
|
||
_, put_kwargs = client.session.put.call_args
|
||
# 红线：lockHandle 必须走 query 参数，不能放 header
|
||
self.assertEqual(put_kwargs["params"].get("lockHandle"), "LH123")
|
||
self.assertNotIn("lockHandle", put_kwargs.get("headers", {}))
|
||
|
||
def test_put_body_uses_lowercase_rfc(self):
|
||
client = self._setup()
|
||
client.set_function_remote_enabled("ZZ_RFC_PROBE", "ZZ_RFC_PROBE_FM")
|
||
_, put_kwargs = client.session.put.call_args
|
||
body = put_kwargs["data"].decode("utf-8")
|
||
self.assertIn(">rfc<", body, "processingType 必须是元素形态小写 rfc")
|
||
self.assertNotIn(">normal<", body)
|
||
# 禁止 remoteEnabled/remote/RFC 等错误枚举值
|
||
self.assertNotIn("remoteEnabled", body)
|
||
self.assertNotIn(">RFC<", body)
|
||
|
||
def test_put_url_is_group_slash_module(self):
|
||
client = self._setup()
|
||
client.set_function_remote_enabled("ZZ_RFC_PROBE", "ZZ_RFC_PROBE_FM")
|
||
url = client.session.put.call_args[0][0]
|
||
self.assertIn(
|
||
"/sap/bc/adt/functions/groups/zz_rfc_probe/fmodules/zz_rfc_probe_fm",
|
||
url,
|
||
)
|
||
|
||
def test_put_failure_returns_error_transparently(self):
|
||
"""PUT 500（类本地类型参数）错误信息透传。"""
|
||
client = self._setup()
|
||
client.session.put.return_value = _mock_resp(
|
||
500, text="针对 RFC 不允许使用类或接口的类型"
|
||
)
|
||
ok, err = client.set_function_remote_enabled("ZZ_RFC_PROBE", "ZZ_RFC_PROBE_FM")
|
||
self.assertFalse(ok)
|
||
self.assertIn("类或接口", err)
|
||
# 失败也须解锁
|
||
client.unlock.assert_called_once()
|
||
|
||
def test_verify_mismatch_returns_error(self):
|
||
"""PUT 成功但 GET 复核回显不是 rfc → 判失败。"""
|
||
client = _make_client()
|
||
client.session.get.side_effect = [
|
||
_mock_resp(200, text=_fm_xml("normal")),
|
||
_mock_resp(200, text=_fm_xml("normal")), # 复核仍 normal
|
||
]
|
||
client.session.put.return_value = _mock_resp(200)
|
||
client.lock = MagicMock(return_value=("LH123", None))
|
||
client.unlock = MagicMock(return_value=True)
|
||
ok, err = client.set_function_remote_enabled("ZZ_RFC_PROBE", "ZZ_RFC_PROBE_FM")
|
||
self.assertFalse(ok)
|
||
self.assertIn("复核", err)
|
||
|
||
|
||
class TestCmdRemoteEnable(unittest.TestCase):
|
||
"""remote-enable 命令级守卫。"""
|
||
|
||
def _args(self, **kwargs):
|
||
defaults = {"name": "ZZ_RFC_PROBE/ZZ_RFC_PROBE_FM", "corr_nr": None}
|
||
defaults.update(kwargs)
|
||
return Namespace(**defaults)
|
||
|
||
def test_requires_group_slash_module(self):
|
||
from sapcli.commands.remote_enable import cmd_remote_enable
|
||
client = _make_mock_client()
|
||
with self.assertRaises(InvalidNameError):
|
||
cmd_remote_enable(self._args(name="ZZ_RFC_PROBE_FM"), client)
|
||
|
||
def test_success_path(self):
|
||
from sapcli.commands.remote_enable import cmd_remote_enable
|
||
client = _make_mock_client()
|
||
client.set_function_remote_enabled.return_value = (True, "")
|
||
with patch("builtins.print"):
|
||
cmd_remote_enable(self._args(), client)
|
||
client.set_function_remote_enabled.assert_called_once_with(
|
||
"ZZ_RFC_PROBE", "ZZ_RFC_PROBE_FM", None
|
||
)
|
||
|
||
def test_failure_raises_with_server_message(self):
|
||
from sapcli.commands.remote_enable import cmd_remote_enable
|
||
client = _make_mock_client()
|
||
client.set_function_remote_enabled.return_value = (
|
||
False, "远程启用失败: HTTP 500 — 针对 RFC 不允许使用类或接口的类型",
|
||
)
|
||
with patch("builtins.print"):
|
||
with self.assertRaises(SapCliError) as ctx:
|
||
cmd_remote_enable(self._args(), client)
|
||
self.assertIn("类或接口", str(ctx.exception))
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# 改动 3：function 类型 '组/模块' URL 组装守卫
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestFunctionUriAssembly(unittest.TestCase):
|
||
"""fmodules/{name} 段必须是模块名，绝不能误用组名。"""
|
||
|
||
def test_parse_object_name_not_group_swapped(self):
|
||
p = parse_object_name("ZZ_RFC_PROBE/ZZ_RFC_PROBE_FM", "function")
|
||
self.assertIn(
|
||
"/fmodules/zz_rfc_probe_fm", p.obj_uri,
|
||
"obj_uri 的 fmodules 段必须是模块名",
|
||
)
|
||
self.assertIn(
|
||
"/fmodules/zz_rfc_probe_fm/source/main", p.src_uri,
|
||
"src_uri 的 fmodules 段必须是模块名",
|
||
)
|
||
self.assertIn("/groups/zz_rfc_probe/", p.obj_uri, "groups 段是组名")
|
||
self.assertNotIn("/fmodules/zz_rfc_probe/source", p.src_uri,
|
||
"禁止用组名 zz_rfc_probe 当模块名")
|
||
|
||
def test_parse_object_name_missing_slash_raises(self):
|
||
with self.assertRaises(InvalidNameError):
|
||
parse_object_name("ZZ_RFC_PROBE_FM", "function")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|