吴让宇
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）
220 lines
7.6 KiB
Python
220 lines
7.6 KiB
Python
"""Tests for sapcli.commands.activate — activate command."""
|
||
import argparse
|
||
import unittest
|
||
from unittest.mock import MagicMock
|
||
|
||
from sapcli.commands.activate import cmd_activate
|
||
|
||
|
||
def _make_client():
|
||
"""创建完整 mock 的 ADTClient。"""
|
||
client = MagicMock()
|
||
client.host = "https://sap.example.com"
|
||
client.sap_client = "100"
|
||
client.csrf_token = "test-csrf-token"
|
||
client.user = "TESTUSER"
|
||
client.session = MagicMock()
|
||
client._headers.return_value = {
|
||
"content-type": "application/xml",
|
||
"x-csrf-token": "test-csrf-token",
|
||
}
|
||
return client
|
||
|
||
|
||
class TestCmdActivateSingle(unittest.TestCase):
|
||
"""单对象激活测试。"""
|
||
|
||
def test_activate_success(self):
|
||
"""激活成功，无消息。"""
|
||
client = _make_client()
|
||
client.activate.return_value = (True, [])
|
||
args = argparse.Namespace(
|
||
name="ZTEST", type="report",
|
||
names=None, types=None, corr_nr=None, config=None,
|
||
)
|
||
cmd_activate(args, client)
|
||
client.activate.assert_called_once_with(
|
||
"ZTEST", "/sap/bc/adt/programs/programs/ztest", None,
|
||
)
|
||
|
||
def test_activate_success_with_corr_nr(self):
|
||
"""带传输请求号激活。"""
|
||
client = _make_client()
|
||
client.activate.return_value = (True, [])
|
||
args = argparse.Namespace(
|
||
name="ZTEST", type="class",
|
||
names=None, types=None, corr_nr="DEVK901362", config=None,
|
||
)
|
||
cmd_activate(args, client)
|
||
client.activate.assert_called_once_with(
|
||
"ZTEST", "/sap/bc/adt/oo/classes/ztest", "DEVK901362",
|
||
)
|
||
|
||
def test_activate_with_errors(self):
|
||
"""激活失败，返回错误消息。"""
|
||
client = _make_client()
|
||
client.activate.return_value = (
|
||
False,
|
||
[{"type": "E", "line": "10", "text": "Syntax error", "href": ""}],
|
||
)
|
||
args = argparse.Namespace(
|
||
name="ZMY_CLASS", type="class",
|
||
names=None, types=None, corr_nr=None, config=None,
|
||
)
|
||
cmd_activate(args, client)
|
||
client.activate.assert_called_once()
|
||
|
||
def test_activate_with_warnings(self):
|
||
"""激活成功但有警告。"""
|
||
client = _make_client()
|
||
client.activate.return_value = (
|
||
True,
|
||
[{"type": "W", "line": "5", "text": "Unused variable", "href": ""}],
|
||
)
|
||
args = argparse.Namespace(
|
||
name="ZTEST", type="report",
|
||
names=None, types=None, corr_nr=None, config=None,
|
||
)
|
||
cmd_activate(args, client) # 不抛异常
|
||
|
||
def test_activate_exception(self):
|
||
"""activate 抛出异常时优雅处理。"""
|
||
client = _make_client()
|
||
client.activate.side_effect = Exception("HTTP 500")
|
||
args = argparse.Namespace(
|
||
name="ZTEST", type="report",
|
||
names=None, types=None, corr_nr=None, config=None,
|
||
)
|
||
cmd_activate(args, client) # 不抛异常，打印错误
|
||
|
||
|
||
class TestCmdActivateBatch(unittest.TestCase):
|
||
"""批量激活测试。"""
|
||
|
||
def test_batch_activate_all_success(self):
|
||
"""批量激活全部成功。"""
|
||
client = _make_client()
|
||
client.activate.return_value = (True, [])
|
||
args = argparse.Namespace(
|
||
name=None, type=None,
|
||
names="ZCLS1,ZCLS2,ZCLS3",
|
||
types="class,class,class",
|
||
corr_nr=None, config=None,
|
||
)
|
||
cmd_activate(args, client)
|
||
self.assertEqual(client.activate.call_count, 3)
|
||
|
||
def test_batch_activate_with_corr_nr(self):
|
||
"""批量激活带传输请求号。"""
|
||
client = _make_client()
|
||
client.activate.return_value = (True, [])
|
||
args = argparse.Namespace(
|
||
name=None, type=None,
|
||
names="ZCLS1,ZREP1",
|
||
types="class,report",
|
||
corr_nr="DEVK901368", config=None,
|
||
)
|
||
cmd_activate(args, client)
|
||
calls = client.activate.call_args_list
|
||
self.assertEqual(len(calls), 2)
|
||
# 验证都传了 corr_nr
|
||
for call in calls:
|
||
self.assertEqual(call[0][2], "DEVK901368")
|
||
|
||
def test_batch_activate_mismatch_count(self):
|
||
"""names 和 types 数量不匹配时报错。"""
|
||
client = _make_client()
|
||
args = argparse.Namespace(
|
||
name=None, type=None,
|
||
names="ZCLS1,ZCLS2",
|
||
types="class",
|
||
corr_nr=None, config=None,
|
||
)
|
||
cmd_activate(args, client)
|
||
client.activate.assert_not_called()
|
||
|
||
def test_batch_activate_partial_failure(self):
|
||
"""批量激活部分失败。"""
|
||
client = _make_client()
|
||
client.activate.side_effect = [
|
||
(True, []),
|
||
(False, [{"type": "E", "line": "1", "text": "Error", "href": ""}]),
|
||
]
|
||
args = argparse.Namespace(
|
||
name=None, type=None,
|
||
names="ZCLS1,ZCLS2",
|
||
types="class,class",
|
||
corr_nr=None, config=None,
|
||
)
|
||
cmd_activate(args, client)
|
||
self.assertEqual(client.activate.call_count, 2)
|
||
|
||
|
||
class TestCmdActivateNoneMode(unittest.TestCase):
|
||
"""既没有 --name 也没有 --names 的情况。"""
|
||
|
||
def test_no_name_no_names(self):
|
||
"""没有提供 name 或 names 时，访问 args.name 应为 None，触发 AttributeError。"""
|
||
client = _make_client()
|
||
args = argparse.Namespace(
|
||
name=None, type=None,
|
||
names=None, types=None, corr_nr=None, config=None,
|
||
)
|
||
# name=None 传给 parse_object_name 会失败
|
||
with self.assertRaises((TypeError, AttributeError, ValueError)):
|
||
cmd_activate(args, client)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|
||
|
||
class TestActivateRequestBodyHasTwoRefs(unittest.TestCase):
|
||
"""守卫：activation 请求体必须含 ≥2 个 objectReference 引用。
|
||
|
||
实测（SAP_BASIS 752，2026-09-15 受控实验）：objectReferences 里只有 1 个
|
||
objectReference 时，服务端返回 HTTP 200 + 空 body（无 content-type）且
|
||
根本不执行激活；引用数为 2 及以上才真正激活（同一对象重复 2 次亦可）。
|
||
单对象 activate 因此必须补重复引用，否则「激活成功」是假象。
|
||
"""
|
||
|
||
def _capture_body(self, name, uri, corr=None):
|
||
from sapcli.client._source import SourceMixin
|
||
|
||
class _Client(SourceMixin):
|
||
def __init__(self):
|
||
self.host = "https://sap.example.com"
|
||
self._stateful = False
|
||
self.session = MagicMock()
|
||
self._headers = MagicMock(return_value={})
|
||
resp = MagicMock()
|
||
resp.status_code = 200
|
||
resp.content = b""
|
||
resp.text = ""
|
||
resp.headers = {"content-type": "text/plain"}
|
||
self.session.post.return_value = resp
|
||
self.session.get.return_value = resp
|
||
|
||
c = _Client()
|
||
try:
|
||
c.activate(name, uri, corr)
|
||
except Exception:
|
||
pass
|
||
body = c.session.post.call_args[1].get("data", "")
|
||
return body
|
||
|
||
def test_single_object_request_has_two_refs(self):
|
||
body = self._capture_body(
|
||
"ZINT_CL_LOG_API", "/sap/bc/adt/oo/classes/zint_cl_log_api")
|
||
n = body.count("<adtcore:objectReference ")
|
||
self.assertGreaterEqual(
|
||
n, 2,
|
||
f"单对象 activate 请求体只有 {n} 个引用 —— 服务端将返回空响应且不激活")
|
||
|
||
def test_request_includes_target_object(self):
|
||
body = self._capture_body(
|
||
"ZINT_CL_LOG_API", "/sap/bc/adt/oo/classes/zint_cl_log_api")
|
||
self.assertIn("zint_cl_log_api", body)
|
||
self.assertIn("ZINT_CL_LOG_API", body)
|
||
|