Files
吴让宇 c06e350f6d
CI / test (3.10) (push) Waiting to run
CI / test (3.11) (push) Waiting to run
CI / test (3.12) (push) Waiting to run
feat: bgRFC/FM 支撑能力 + ADT 后缀剥离与 CLAS 激活修复(B1 配套)
新增能力:
- 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)
2026-09-15 22:51:35 +08:00

275 lines
8.9 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""目录扫描模块。
扫描项目目录,识别 ABAP 开发对象。支持两种布局:
1. **三层布局(推荐,sap-cli-skill 铁律 5**
`src/<开发包>/<对象类型>/<文件名>.abap`
例:`src/TMP/class/zcl_demo.abap`
本地开发包固定写 `TMP`(对应 SAP 端 `$TMP` 本地包,目录名去掉 `$`)。
2. **扁平布局(旧项目,向后兼容)**:
`<对象类型>/<文件名>.abap`(项目根目录下,对象类型目录用复数)
例:`classes/zcl_demo.abap`
对象类型目录名**单复数均接受、大小写不敏感**`class/` 与 `classes/` 等价)。
函数模块(function 类型)两种命名均支持:
- 子目录式:`function/<函数组>/<函数模块>.abap`
- ADT 式:`function/<函数组>.fugr.<函数模块>.abap`abapGit 惯例)
对象名统一为 `函数组/函数模块`(大写)。
"""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass
logger = logging.getLogger("sapcli.scanner")
# 对象类型目录名(复数,历史命名)→ 对象类型
DIRECTORY_TYPE_MAP: dict[str, str] = {
"reports": "report",
"classes": "class",
"interfaces": "interface",
"functions": "function",
"domains": "domain",
"dataelements": "dataelement",
"tables": "table",
"structures": "structure",
"tabletypes": "tabletype",
"includes": "include",
"cdsviews": "cdsview",
"messageclasses": "messageclass",
"views": "view",
"searchhelps": "searchhelp",
"lockobjects": "lockobject",
"dcl": "dcl",
"ddlx": "ddlX",
}
# 单数形式目录名(sap-cli-skill 结构规范使用单数)→ 对象类型
_SINGULAR_TYPE_MAP: dict[str, str] = {
"report": "report",
"class": "class",
"interface": "interface",
"function": "function",
"domain": "domain",
"dataelement": "dataelement",
"table": "table",
"structure": "structure",
"tabletype": "tabletype",
"include": "include",
"cdsview": "cdsview",
"messageclass": "messageclass",
"view": "view",
"searchhelp": "searchhelp",
"lockobject": "lockobject",
"dcl": "dcl",
"ddlx": "ddlX",
}
# 查找用总表:小写目录名 → 对象类型
TYPE_DIR_LOOKUP: dict[str, str] = {**DIRECTORY_TYPE_MAP, **_SINGULAR_TYPE_MAP}
FUNCTIONS_DIR = "functions"
# 源码根目录(三层布局的第 1 层父目录)
SRC_DIR = "src"
# ADT 式函数模块文件名分隔符:<fugr>.fugr.<fm>.abap
FUNC_ADT_SEP = ".fugr."
# ADT/abapGit 风格的**对象类型后缀**`<name>.<suffix>.abap`)。
# 来源:ADT 链接格式与 abapGit 序列化——项目仓命名规范(如 ABAP_Integrated
# 《命名规范》§4)即按此写文件名,故扫描器必须剥掉后缀才能得到对象名。
# 仅剥与所在目录 obj_type 匹配的那一个后缀,避免误伤含点的对象名。
ADT_TYPE_SUFFIX: dict[str, str] = {
"class": "clas",
"interface": "intf",
"report": "prog",
"domain": "doma",
"dataelement": "dtel",
"table": "tabl",
"structure": "stru",
"tabletype": "ttyp",
"include": "cinc",
"cdsview": "ddls",
"messageclass": "msag",
"searchhelp": "shlp",
"lockobject": "enqu",
"dcl": "dcls",
"ddlX": "ddlx",
}
def _strip_adt_suffix(base: str, obj_type: str) -> str:
"""剥掉 ADT 类型后缀(`zcl_demo.clas` → `zcl_demo`)。
只在该类型有约定后缀、且文件名确实以 `.<该后缀>` 结尾时剥,
其余情况原样返回(含点的对象名不受影响)。
"""
suffix = ADT_TYPE_SUFFIX.get(obj_type)
if not suffix:
return base
tail = "." + suffix
if base.lower().endswith(tail) and len(base) > len(tail):
return base[: -len(tail)]
return base
@dataclass
class ScannedObject:
"""扫描到的对象。"""
name: str # 大写对象名,function 类型含 / (GROUP/FUNC)
type: str # 对象类型
file: str # 相对于项目根目录的路径
def scan_project(project_path: str) -> list[ScannedObject]:
"""扫描项目目录,返回识别到的所有对象列表。
先扫项目根目录下的对象类型目录(旧布局),
再扫 src/<开发包>/ 下的对象类型目录(三层布局)。
同名同类型对象只保留第一次扫到的。
"""
results: list[ScannedObject] = []
seen: set[tuple[str, str]] = set()
# 扫描范围:(目录绝对路径, 该目录相对项目根的路径前缀)
scopes: list[tuple[str, str]] = [(project_path, "")]
src_path = os.path.join(project_path, SRC_DIR)
if os.path.isdir(src_path):
for pkg_name in sorted(os.listdir(src_path)):
if pkg_name.startswith("."):
continue
pkg_path = os.path.join(src_path, pkg_name)
if os.path.isdir(pkg_path):
scopes.append((pkg_path, f"{SRC_DIR}/{pkg_name}"))
for root_path, rel_prefix in scopes:
try:
entries = sorted(os.listdir(root_path))
except OSError as exc: # 权限/竞态
logger.debug("目录不可读,跳过: %s (%s)", root_path, exc)
continue
for dir_name in entries:
if dir_name.startswith("."):
continue
obj_type = TYPE_DIR_LOOKUP.get(dir_name.lower())
if obj_type is None:
continue
dir_path = os.path.join(root_path, dir_name)
if not os.path.isdir(dir_path):
continue
if obj_type == "function":
_scan_functions(dir_path, dir_name, rel_prefix, results, seen)
else:
_scan_flat_directory(dir_path, dir_name, obj_type, rel_prefix, results, seen)
logger.info("扫描完成: %d 个对象", len(results))
return results
def _rel(rel_prefix: str, *parts: str) -> str:
"""拼相对项目根的路径(统一用 / 分隔,去掉空段)。"""
segs = [rel_prefix, *parts]
return "/".join(s for s in segs if s)
def _add(
results: list[ScannedObject],
seen: set[tuple[str, str]],
name: str,
obj_type: str,
rel_path: str,
) -> None:
key = (name, obj_type)
if key in seen:
logger.debug("重复对象,跳过: %s (%s) ← %s", name, obj_type, rel_path)
return
seen.add(key)
results.append(ScannedObject(name=name, type=obj_type, file=rel_path))
logger.debug("扫描到: %s (%s) → %s", name, obj_type, rel_path)
def _scan_flat_directory(
dir_path: str,
dir_name: str,
obj_type: str,
rel_prefix: str,
results: list[ScannedObject],
seen: set[tuple[str, str]],
) -> None:
"""扫描普通目录(一对一:文件名 = 对象名)。"""
for filename in sorted(os.listdir(dir_path)):
if not filename.endswith(".abap"):
continue
if filename.startswith("."):
continue
if not os.path.isfile(os.path.join(dir_path, filename)):
continue
obj_name = _strip_adt_suffix(filename[:-5], obj_type).upper() # 去 .abap + ADT 类型后缀
_add(results, seen, obj_name, obj_type, _rel(rel_prefix, dir_name, filename))
def _scan_functions(
dir_path: str,
dir_name: str,
rel_prefix: str,
results: list[ScannedObject],
seen: set[tuple[str, str]],
) -> None:
"""扫描函数模块目录。
支持两种命名:
- 子目录式: <函数组>/<函数模块>.abap
- ADT 式: <函数组>.fugr.<函数模块>.abap
对象名: <函数组大写>/<函数模块大写>
"""
for entry in sorted(os.listdir(dir_path)):
if entry.startswith("."):
continue
entry_path = os.path.join(dir_path, entry)
if os.path.isdir(entry_path):
# 子目录式:子目录名 = 函数组名
group_name = entry.upper()
for filename in sorted(os.listdir(entry_path)):
if not filename.endswith(".abap") or filename.startswith("."):
continue
if not os.path.isfile(os.path.join(entry_path, filename)):
continue
func_name = filename[:-5].upper()
_add(
results, seen,
f"{group_name}/{func_name}", "function",
_rel(rel_prefix, dir_name, entry, filename),
)
continue
if not entry.endswith(".abap"):
continue
# ADT 式:<fugr>.fugr.<fm>.abap
base = entry[:-5]
low = base.lower()
if FUNC_ADT_SEP not in low:
logger.debug("函数模块文件名不符合约定,跳过: %s", entry)
continue
idx = low.index(FUNC_ADT_SEP)
group_name = base[:idx].upper()
func_name = base[idx + len(FUNC_ADT_SEP):].upper()
if not group_name or not func_name:
logger.debug("函数模块文件名解析失败,跳过: %s", entry)
continue
_add(
results, seen,
f"{group_name}/{func_name}", "function",
_rel(rel_prefix, dir_name, entry),
)