139 lines
5.7 KiB
Python
139 lines
5.7 KiB
Python
"""upload 命令 — 仅上传源代码到 SAP（锁定→写入→解锁，不检查不激活）。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import logging
|
||
import os
|
||
|
||
from sapcli.client import ADTClient
|
||
from sapcli.commands.crud import _select_transport_request
|
||
from sapcli.exceptions import (
|
||
ConfigError,
|
||
InvalidNameError,
|
||
LockError,
|
||
ObjectNotFoundError,
|
||
)
|
||
from sapcli.types import get_type_config, parse_object_name
|
||
|
||
logger = logging.getLogger("sapcli.commands.upload")
|
||
|
||
|
||
def cmd_upload(args: argparse.Namespace, client: ADTClient) -> None:
|
||
"""上传本地源代码到 SAP（仅锁定→写入→解锁，不检查不激活）。
|
||
|
||
与 sync 的区别：
|
||
- 不自动创建对象（对象必须已存在，否则报 ObjectNotFoundError）
|
||
- 不执行语法检查
|
||
- 不执行激活
|
||
|
||
适用于自动化测试场景：修改代码后先安全写入 SAP，避开 activate 的
|
||
锁定冲突，稍后再单独 activate。
|
||
"""
|
||
name: str = args.name
|
||
obj_type: str = args.type
|
||
file_path: str = args.path
|
||
corr_nr: str | None = getattr(args, "corr_nr", None)
|
||
part: str | None = getattr(args, "part", None)
|
||
|
||
parsed = parse_object_name(name, obj_type)
|
||
type_label = get_type_config(obj_type).label
|
||
|
||
if parsed.src_uri is None:
|
||
print(f"\n ✗ {type_label}没有源代码，不支持 upload 操作")
|
||
print(f" functiongroup 等是容器对象，不包含可编辑的源代码文件")
|
||
raise InvalidNameError(f"{type_label}没有源代码，不支持 upload 操作")
|
||
|
||
src_uri = parsed.src_uri
|
||
if part == "testclasses":
|
||
# Local test classes include (ABAP Unit): classes only.
|
||
# ADT unit testruns only discovers LOCAL test classes — global
|
||
# test classes are NOT run by the endpoint (S4T proven).
|
||
if obj_type != "class":
|
||
print(f"\n ✗ --part testclasses 仅支持 class 类型")
|
||
raise InvalidNameError("--part testclasses 仅支持 class 类型")
|
||
src_uri = f"/sap/bc/adt/oo/classes/{name.lower()}/source/testclasses"
|
||
|
||
if not os.path.isfile(file_path):
|
||
print(f"\n ✗ 文件不存在: {file_path}")
|
||
raise ConfigError(f"文件不存在: {file_path}")
|
||
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
source = f.read()
|
||
|
||
print("=" * 60)
|
||
print(" SAP ADT 源代码上传（不检查不激活）")
|
||
print("=" * 60)
|
||
print(f" 对象类型: {type_label}")
|
||
print(f" 对象名称: {parsed.display_name}")
|
||
print(f" 本地文件: {file_path}")
|
||
print(f" 源代码: {len(source)} 字符, {len(source.splitlines())} 行")
|
||
print(f" 流程: 检查 → 锁定 → 写入 → 解锁")
|
||
|
||
# 检查对象是否存在（upload 不自动创建，区别于 sync）
|
||
print(f"\n → 检查对象是否存在...")
|
||
if not client.object_exists(parsed.obj_uri):
|
||
print(f" ✗ 对象不存在: {parsed.display_name}")
|
||
print(f" upload 不会自动创建对象，请先用 create 或 sync 创建")
|
||
raise ObjectNotFoundError(parsed.display_name, obj_type)
|
||
print(" ✓ 对象存在")
|
||
|
||
# ── 锁定 → 写入 → 解锁（提取自 crud._sync_single，不含检查/激活）──
|
||
print("\n ── 锁定 → 写入 → 解锁 ──")
|
||
lock_handle: str
|
||
if corr_nr:
|
||
lock_handle, _ = client.lock(parsed.obj_uri, corr_nr)
|
||
print(f" ✓ 锁定成功（传输请求: {corr_nr}）")
|
||
else:
|
||
try:
|
||
lock_handle, detected_corr_nr = client.lock(parsed.obj_uri)
|
||
if detected_corr_nr:
|
||
corr_nr = detected_corr_nr
|
||
print(f" ✓ 锁定成功（对象已绑定传输请求: {corr_nr}）")
|
||
else:
|
||
print(f" ✓ 锁定成功（本地对象，无需传输请求）")
|
||
except LockError:
|
||
print(" ℹ 对象需要传输请求号")
|
||
corr_nr = _select_transport_request(client)
|
||
if not corr_nr:
|
||
print(" ✗ 锁定失败: 无法获取传输请求号")
|
||
raise LockError("锁定失败: 无法获取传输请求号", obj_uri=parsed.obj_uri)
|
||
lock_handle, _ = client.lock(parsed.obj_uri, corr_nr)
|
||
print(f" ✓ 锁定成功（传输请求: {corr_nr}）")
|
||
|
||
try:
|
||
if part == "testclasses":
|
||
# testclasses endpoint expects XML body: <abapClass> element
|
||
# in the oo/classes namespace wrapping the raw source.
|
||
from xml.sax.saxutils import escape as _esc
|
||
|
||
xml_body = (
|
||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||
'<cls:abapClass xmlns:cls="http://www.sap.com/adt/oo/classes" '
|
||
'xmlns:adtcore="http://www.sap.com/adt/core" '
|
||
'adtcore:type="testclassInclude" '
|
||
'adtcore:description="Local test classes" '
|
||
'adtcore:name="zint_cl_route_engine">'
|
||
f"<cls:source>{_esc(source)}</cls:source>"
|
||
"</cls:abapClass>"
|
||
)
|
||
client.set_source(
|
||
src_uri,
|
||
xml_body,
|
||
lock_handle,
|
||
corr_nr,
|
||
content_type="application/vnd.sap.adt.abap.classes+xml; charset=utf-8",
|
||
)
|
||
else:
|
||
client.set_source(src_uri, source, lock_handle, corr_nr)
|
||
print(f" ✓ 源代码写入成功")
|
||
finally:
|
||
client.unlock(parsed.obj_uri, lock_handle)
|
||
print(f" ✓ 解锁成功")
|
||
|
||
print(f"\n {'=' * 60}")
|
||
print(f" ✓ {parsed.display_name} 源代码已上传")
|
||
print(f" 文件: {file_path}")
|
||
print(f" 源代码: {len(source)} 字符, {len(source.splitlines())} 行")
|
||
print(f" 注意: 未执行语法检查和激活，请按需单独执行 syntax-check / activate")
|
||
print(f" {'=' * 60}")
|