sap-cli-skill/tests/unit/test_client.py

"""client.py 单元测试 — 30 个 ADT API 方法

运行: python tests/unit/test_client.py
"""
from __future__ import annotations

import os
import sys
import unittest
import xml.etree.ElementTree as ET
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.exceptions import (
    LoginError, SapCliError, CreateError, DeleteError,
    ActivationError, SyntaxCheckError, LockError,
)

# ADT XML 命名空间
TM_NS = "http://www.sap.com/cts/adt/tm"
ADTCORE_NS = "http://www.sap.com/adt/core"
CHK_NS = "http://www.sap.com/adt/checklist"
LOCK_NS = "http://www.sap.com/adt/lock"


def _make_client() -> ADTClient:
    """创建 mock 好的 ADTClient。"""
    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 _mock_resp(status_code=200, text="", content=b"", headers=None):
    """快速创建 mock response。"""
    resp = MagicMock()
    resp.status_code = status_code
    resp.text = text
    resp.content = content
    resp.headers = headers or {}
    resp.raise_for_status = MagicMock()
    return resp


# ═══════════════════════════════════════════
#  基础方法
# ═══════════════════════════════════════════

class TestHeaders(unittest.TestCase):

    def test_default(self):
        h = _make_client()._headers()
        self.assertEqual(h["content-type"], "application/xml")
        self.assertEqual(h["x-csrf-token"], "test-csrf-token")

    def test_custom(self):
        h = _make_client()._headers("text/plain")
        self.assertEqual(h["content-type"], "text/plain")


class TestContextManager(unittest.TestCase):

    def test_enter_calls_login(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, headers={"x-csrf-token": "TOK"})
        result = client.__enter__()
        self.assertIs(result, client)

    def test_exit_closes_session(self):
        client = _make_client()
        client.__exit__(None, None, None)
        client.session.close.assert_called_once()


# ═══════════════════════════════════════════
#  认证
# ═══════════════════════════════════════════

class TestLogin(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, headers={"x-csrf-token": "NEW_TOK"})
        self.assertTrue(client.login())
        self.assertEqual(client.csrf_token, "NEW_TOK")

    def test_failure(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(401)
        with self.assertRaises(LoginError):
            client.login()


# ═══════════════════════════════════════════
#  对象存在检查
# ═══════════════════════════════════════════

class TestObjectExists(unittest.TestCase):

    def test_exists(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200)
        self.assertTrue(client.object_exists("/uri"))

    def test_not_exists(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(404)
        self.assertFalse(client.object_exists("/uri"))


# ═══════════════════════════════════════════
#  源码读取
# ═══════════════════════════════════════════

class TestGetSource(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, text="REPORT ztest.")
        self.assertEqual(client.get_source("/src"), "REPORT ztest.")

    def test_not_found(self):
        client = _make_client()
        mock = _mock_resp(404)
        mock.raise_for_status.side_effect = Exception("404")
        client.session.get.return_value = mock
        with self.assertRaises(Exception):
            client.get_source("/nonexistent")


# ═══════════════════════════════════════════
#  锁管理
# ═══════════════════════════════════════════

class TestLock(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=(
            f'<lock xmlns="{LOCK_NS}">'
            f'<lockHandle>lh_123</lockHandle>'
            f'<corrNr>DEVK001</corrNr></lock>'
        ).encode())
        handle, corr = client.lock("/uri")
        self.assertEqual(handle, "lh_123")
        self.assertEqual(corr, "DEVK001")

    def test_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(403)
        with self.assertRaises(LockError):
            client.lock("/uri")

    def test_no_handle(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=(
            f'<lock xmlns="{LOCK_NS}"></lock>'
        ).encode())
        with self.assertRaises(LockError):
            client.lock("/uri")


class TestUnlock(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200)
        self.assertTrue(client.unlock("/uri", "lh"))

    def test_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500)
        self.assertFalse(client.unlock("/uri", "lh"))


# ═══════════════════════════════════════════
#  源码写入
# ═══════════════════════════════════════════

class TestSetSource(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.put.return_value = _mock_resp(200)
        self.assertTrue(client.set_source("/src", "code", "lh"))

    def test_with_corr_nr(self):
        client = _make_client()
        client.session.put.return_value = _mock_resp(200)
        self.assertTrue(client.set_source("/src", "code", "lh", corr_nr="DEVK001"))


# ═══════════════════════════════════════════
#  传输请求
# ═══════════════════════════════════════════

_TRANSPORT_XML = (
    f'<?xml version="1.0"?>'
    f'<tm:transportorganizer xmlns:tm="{TM_NS}">'
    f'<tm:request tm:number="DEVK001" tm:status="D" tm:description="Test"/>'
    f'</tm:transportorganizer>'
).encode()

_TRANSPORT_EMPTY = (
    f'<?xml version="1.0"?>'
    f'<tm:transportorganizer xmlns:tm="{TM_NS}"/>'
).encode()

# create_transport_request 返回的 XML（带 tm:number 属性）
_CREATE_TRANSPORT_RESP = (
    f'<?xml version="1.0"?>'
    f'<tm:transportorganizer xmlns:tm="{TM_NS}">'
    f'<tm:request tm:number="DEVK002" tm:status="D"/>'
    f'</tm:transportorganizer>'
).encode()


class TestGetTransportRequest(unittest.TestCase):

    def test_found(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=_TRANSPORT_XML)
        self.assertEqual(client.get_transport_request(), "DEVK001")

    def test_none(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=_TRANSPORT_EMPTY)
        self.assertIsNone(client.get_transport_request())


class TestListTransportRequests(unittest.TestCase):

    def test_found(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=_TRANSPORT_XML)
        self.assertIsInstance(client.list_transport_requests(), list)

    def test_empty(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=_TRANSPORT_EMPTY)
        self.assertEqual(client.list_transport_requests(), [])


class TestCreateTransportRequest(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(201, content=_CREATE_TRANSPORT_RESP)
        result = client.create_transport_request("My request")
        self.assertEqual(result, "DEVK002")

    def test_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500, text="Error")
        with self.assertRaises(CreateError):
            client.create_transport_request("Fail")


# ═══════════════════════════════════════════
#  激活 — 解析 type="E" 属性
# ═══════════════════════════════════════════

class TestActivate(unittest.TestCase):

    def test_success_empty(self):
        """空响应视为成功。"""
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, text="", content=b"")
        ok, msgs = client.activate("ztest", "/uri")
        self.assertTrue(ok)

    def test_failure_with_error_type(self):
        """包含 type="E" 属性的元素视为失败。"""
        client = _make_client()
        xml_body = (
            f'<?xml version="1.0"?>'
            f'<adtcore:activation xmlns:adtcore="{ADTCORE_NS}">'
            f'<adtcore:objectReference type="E" line="1" href="">'
            f'<adtcore:txt>Activation error</adtcore:txt>'
            f'</adtcore:objectReference>'
            f'</adtcore:activation>'
        )
        client.session.post.return_value = _mock_resp(200, text=xml_body, content=xml_body.encode())
        ok, msgs = client.activate("ztest", "/uri")
        self.assertFalse(ok)

    def test_http_error(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500)
        with self.assertRaises(ActivationError):
            client.activate("ztest", "/uri")

    # ── NW 7.40 double-activate workaround（空响应验证路径）──────────

    def _verify_xml(self, version):
        """构造 verify GET 响应的 XML（bytes），adtcore:version 指定。"""
        return (
            f'<?xml version="1.0" encoding="UTF-8"?>'
            f'<abapsource:abapSource xmlns:abapsource="http://www.sap.com/adt/abapsource"'
            f' xmlns:adtcore="{ADTCORE_NS}"'
            f' adtcore:version="{version}">'
            f'</abapsource:abapSource>'
        ).encode()

    def test_empty_response_already_active(self):
        """空响应 + verify 显示 active → 成功，无需重试（post 仅 1 次）。"""
        client = _make_client()
        # 激活 POST → 200 空响应
        client.session.post.return_value = _mock_resp(200, text="", content=b"")
        # verify GET → 200, version="active"
        client.session.get.return_value = _mock_resp(200, content=self._verify_xml("active"))

        ok, msgs = client.activate("ztest", "/uri")
        self.assertTrue(ok)
        self.assertEqual(msgs, [])
        # 区分真/假实现：已 active 时不应重试激活
        self.assertEqual(client.session.post.call_count, 1)

    def test_double_activate_success(self):
        """空响应 + inactive → 重试一次 → active（double-activate 成功路径）。"""
        client = _make_client()
        # 激活 POST + 重试 POST → 均 200 空
        client.session.post.side_effect = [
            _mock_resp(200, text="", content=b""),
            _mock_resp(200, text="", content=b""),
        ]
        # verify GET → inactive；再 verify GET → active
        client.session.get.side_effect = [
            _mock_resp(200, content=self._verify_xml("inactive")),
            _mock_resp(200, content=self._verify_xml("active")),
        ]

        ok, msgs = client.activate("ztest", "/uri")
        self.assertTrue(ok)
        self.assertEqual(msgs, [])
        # 区分真/假实现：确实重试了，post 被调用 2 次
        self.assertEqual(client.session.post.call_count, 2)
        # 两次 verify GET
        self.assertEqual(client.session.get.call_count, 2)

    def test_double_activate_still_inactive(self):
        """重试后仍 inactive → 失败并返回 SE09 提示（post 调用 2 次）。"""
        client = _make_client()
        client.session.post.side_effect = [
            _mock_resp(200, text="", content=b""),
            _mock_resp(200, text="", content=b""),
        ]
        # 两次 verify GET 都返回 inactive
        client.session.get.side_effect = [
            _mock_resp(200, content=self._verify_xml("inactive")),
            _mock_resp(200, content=self._verify_xml("inactive")),
        ]

        ok, msgs = client.activate("ztest", "/uri")
        self.assertFalse(ok)
        # 重试过一次，共 2 次 POST
        self.assertEqual(client.session.post.call_count, 2)
        # 返回的消息含 SE09 或 inactive 提示
        msg_text = " ".join(m.get("text", "") for m in msgs)
        self.assertTrue("SE09" in msg_text or "inactive" in msg_text)


# ═══════════════════════════════════════════
#  语法检查 — 同样解析 type="E"
# ═══════════════════════════════════════════

class TestSyntaxCheck(unittest.TestCase):

    def test_clean(self):
        """空响应视为通过。"""
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, text="", content=b"")
        ok, msgs = client.syntax_check("ztest", "/uri")
        self.assertTrue(ok)

    def test_with_errors(self):
        client = _make_client()
        xml_body = (
            f'<?xml version="1.0"?>'
            f'<chk:checks xmlns:chk="{CHK_NS}">'
            f'<chk:msg type="E" line="1" href="">'
            f'<chk:txt>Syntax error at line 1</chk:txt>'
            f'</chk:msg>'
            f'</chk:checks>'
        )
        client.session.post.return_value = _mock_resp(200, text=xml_body, content=xml_body.encode())
        ok, msgs = client.syntax_check("ztest", "/uri")
        self.assertFalse(ok)

    def test_http_error(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500)
        with self.assertRaises(SyntaxCheckError):
            client.syntax_check("ztest", "/uri")


# ═══════════════════════════════════════════
#  删除对象（内部调 lock）
# ═══════════════════════════════════════════

class TestDeleteObject(unittest.TestCase):

    def _mock_lock_ok(self, client):
        client.session.post.return_value = _mock_resp(200, content=(
            f'<lock xmlns="{LOCK_NS}"><lockHandle>lh_del</lockHandle></lock>'
        ).encode())

    def test_success_200(self):
        client = _make_client()
        self._mock_lock_ok(client)
        client.session.delete.return_value = _mock_resp(200)
        ok, msg = client.delete_object("/uri")
        self.assertTrue(ok)

    def test_success_204(self):
        client = _make_client()
        self._mock_lock_ok(client)
        client.session.delete.return_value = _mock_resp(204)
        ok, msg = client.delete_object("/uri")
        self.assertTrue(ok)

    def test_failure(self):
        client = _make_client()
        self._mock_lock_ok(client)
        client.session.delete.return_value = _mock_resp(500, text="Cannot delete")
        with self.assertRaises(DeleteError):
            client.delete_object("/uri")

    def test_ddic_lock_uses_special_accept_header(self):
        """delete 调用 lock 时传 DDIC lock 专用 Accept 头（缺陷4）。

        DDIC 对象（domain/dataelement/table/structure）的 lock 端点对默认
        Accept: */* 返回 HTTP 406。参考 abap-adt-api objectcontents.ts 的 lock
        实现，需用 application/*,application/vnd.sap.as+xml;charset=UTF-8;
        dataname=com.sap.adt.lock.result。
        """
        client = _make_client()
        self._mock_lock_ok(client)
        client.session.delete.return_value = _mock_resp(204)
        client.delete_object("/uri")
        # lock 的 POST 请求必须携带 DDIC lock 专用 Accept 头。
        # 注意：delete 成功后会再发一次 unlock 的 POST，故不能用 call_args
        # （那是最后一次调用 = unlock），须从调用列表中挑出 _action=LOCK 的请求。
        lock_calls = [
            kw for _, kw in client.session.post.call_args_list
            if kw.get("params", {}).get("_action") == "LOCK"
        ]
        self.assertEqual(len(lock_calls), 1, "delete 应恰好发起一次 LOCK 请求")
        accept = lock_calls[0]["headers"]["Accept"]
        self.assertIn("com.sap.adt.lock.result", accept)
        self.assertNotEqual(accept, "*/*")


# ═══════════════════════════════════════════
#  创建对象
# ═══════════════════════════════════════════

class TestCreateObject(unittest.TestCase):

    def test_report_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(201, headers={
            "Location": "/sap/bc/adt/programs/programs/ztest"
        })
        obj_uri, src_uri = client.create_object("report", "ZTEST", "Test report")
        self.assertIn("ztest", obj_uri)

    def test_class_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(201, headers={
            "Location": "/sap/bc/adt/oo/classes/zcl_test"
        })
        obj_uri, src_uri = client.create_object("class", "ZCL_TEST", "Test class")
        self.assertIn("zcl_test", obj_uri)

    def test_with_source(self):
        """有 source → lock → set_source → unlock → activate。"""
        client = _make_client()
        lock_resp = _mock_resp(200, content=(
            f'<lock xmlns="{LOCK_NS}"><lockHandle>lh</lockHandle></lock>'
        ).encode())
        create_resp = _mock_resp(201, headers={"Location": "/sap/bc/adt/programs/programs/ztest"})
        unlock_resp = _mock_resp(200)
        activate_resp = _mock_resp(200, text="", content=b"")
        client.session.post.side_effect = [create_resp, lock_resp, unlock_resp, activate_resp]
        client.session.put.return_value = _mock_resp(200)
        obj_uri, src_uri = client.create_object("report", "ZTEST", "Test", source="REPORT ztest.")
        self.assertIn("ztest", obj_uri)

    def test_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(409, text="Already exists")
        with self.assertRaises(CreateError):
            client.create_object("report", "ZTEST", "Test")


class TestBuildCreateBody(unittest.TestCase):

    def test_report(self):
        body = _make_client()._build_create_body("report", "ZTEST", "Desc")
        self.assertIn("ZTEST", body)

    def test_class(self):
        body = _make_client()._build_create_body("class", "ZCL_TEST", "Desc")
        self.assertIn("ZCL_TEST", body)

    def test_unsupported(self):
        with self.assertRaises(ValueError):
            _make_client()._build_create_body("unknown", "ZTEST", "Desc")


# ═══════════════════════════════════════════
#  DDIC 对象 — definition_body 是 XML 字符串
# ═══════════════════════════════════════════

_DDIC_DOMAIN_XML = (
    f'<?xml version="1.0" encoding="UTF-8"?>'
    f'<abap xmlns="http://www.sap.com/abapxml/types">'
    f'<domain ddic:version="1.0" xmlns:ddic="http://www.sap.com/abapxml/types">'
    f'</domain></abap>'
)

class TestCreateDDICObject(unittest.TestCase):

    def test_domain_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(201)
        # mock activate（create_ddic_object 内部调用）
        client.session.post.side_effect = [
            _mock_resp(201),  # create POST
            _mock_resp(200, text="", content=b""),  # activate POST
        ]
        obj_uri, src_uri = client.create_ddic_object("domain", "ZTEST_DOM", _DDIC_DOMAIN_XML)
        self.assertIn("ztest_dom", obj_uri)

    def test_table_success(self):
        """table 类型先 POST 创建实体，再 PUT 写入 DDL（缺陷3 修复后）。"""
        client = _make_client()
        client.session.post.return_value = _mock_resp(201)
        # mock _put_ddl_source
        with patch.object(client, "_put_ddl_source") as mock_ddl, \
             patch.object(client, "activate", return_value=(True, [])):
            obj_uri, src_uri = client.create_ddic_object(
                "table", "ZTEST_TAB", "<ddl/>"
            )
        # 先 POST 创建实体，再 PUT DDL
        self.assertTrue(client.session.post.called)
        mock_ddl.assert_called_once()
        self.assertIn("ztest_tab", obj_uri)

    def test_create_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500, text="Error")
        with self.assertRaises(CreateError):
            client.create_ddic_object("domain", "ZTEST", _DDIC_DOMAIN_XML)


# ═══════════════════════════════════════════
#  对象状态
# ═══════════════════════════════════════════

class TestGetObjectStatus(unittest.TestCase):

    def test_active(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<adtcore:object xmlns:adtcore="{ADTCORE_NS}">'
            f'<adtcore:packageRef adtcore:name="$TMP"/>'
            f'</adtcore:object>'
        ).encode())
        self.assertTrue(client.get_object_status("/uri")["exists"])

    def test_not_exists(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(404)
        self.assertFalse(client.get_object_status("/uri")["exists"])


# ═══════════════════════════════════════════
#  函数组
# ═══════════════════════════════════════════

class TestFunctionGroup(unittest.TestCase):

    def test_exists_true(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200)
        self.assertTrue(client.function_group_exists("ZGROUP"))

    def test_exists_false(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(404)
        self.assertFalse(client.function_group_exists("ZGROUP"))

    def test_create_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(201)
        self.assertTrue(client.create_function_group("ZGROUP"))

    def test_create_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500, text="Error")
        with self.assertRaises(CreateError):
            client.create_function_group("ZGROUP")


# ═══════════════════════════════════════════
#  搜索浏览
# ═══════════════════════════════════════════

_LIST_TWO = (
    f'<?xml version="1.0"?>'
    f'<adtcore:objectReferences xmlns:adtcore="{ADTCORE_NS}">'
    f'<adtcore:objectReference adtcore:name="Z_TEST1" adtcore:type="PROG/P">'
    f'<adtcore:packageRef adtcore:name="$TMP"/></adtcore:objectReference>'
    f'<adtcore:objectReference adtcore:name="Z_TEST2" adtcore:type="PROG/P">'
    f'<adtcore:packageRef adtcore:name="$TMP"/></adtcore:objectReference>'
    f'</adtcore:objectReferences>'
).encode()

_LIST_EMPTY = (
    f'<?xml version="1.0"?>'
    f'<adtcore:objectReferences xmlns:adtcore="{ADTCORE_NS}"/>'
).encode()


class TestListObjects(unittest.TestCase):

    def test_found(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=_LIST_TWO)
        result = client.list_objects(obj_type="report")
        self.assertEqual(len(result), 2)
        self.assertEqual(result[0]["name"], "Z_TEST1")

    def test_empty(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=_LIST_EMPTY)
        self.assertEqual(client.list_objects(), [])

    def test_quickssearch_params(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=_LIST_TWO)
        client.list_objects(obj_type="PROG/P", prefix="Z_TEST")
        _, kwargs = client.session.get.call_args
        params = kwargs["params"]
        self.assertEqual(params["operation"], "quickSearch")
        self.assertEqual(params["query"], "Z_TEST")
        self.assertEqual(params["maxResults"], "200")
        self.assertEqual(params["objectType"], "PROG/P")
        for bad in ("maxrow", "name", "type"):
            self.assertNotIn(bad, params)

    def test_strips_type_label_from_name(self):
        client = _make_client()
        xml = (
            f'<?xml version="1.0"?>'
            f'<adtcore:objectReferences xmlns:adtcore="{ADTCORE_NS}">'
            f'<adtcore:objectReference adtcore:name="ZMM_BIP_001_HEADER (Structure)" adtcore:type="TABL/ST"/>'
            f'</adtcore:objectReferences>'
        ).encode()
        client.session.get.return_value = _mock_resp(200, content=xml)
        result = client.list_objects()
        self.assertEqual(result[0]["name"], "ZMM_BIP_001_HEADER")


class TestWhereUsed(unittest.TestCase):

    def test_found(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<adtcore:objectReferences xmlns:adtcore="{ADTCORE_NS}">'
            f'<adtcore:objectReference adtcore:name="Z_USER" adtcore:type="PROG/P"/>'
            f'</adtcore:objectReferences>'
        ).encode())
        result = client.where_used("Z_TEST", "/uri", "PROG/P")
        self.assertEqual(len(result), 1)

    def test_empty(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=_LIST_EMPTY)
        self.assertEqual(client.where_used("Z", "/uri"), [])


class TestSearchCode(unittest.TestCase):

    def test_found(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<adtcore:objectReferences xmlns:adtcore="{ADTCORE_NS}">'
            f'<adtcore:objectReference adtcore:name="Z_RESULT" adtcore:type="PROG/P"/>'
            f'</adtcore:objectReferences>'
        ).encode())
        self.assertEqual(len(client.search_code("SELECT", obj_type="report")), 1)

    def test_empty(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=_LIST_EMPTY)
        self.assertEqual(client.search_code("NONEXISTENT"), [])


class TestReadSourceForDiff(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        with patch.object(client, "get_source", return_value="REPORT ztest."):
            self.assertEqual(client.read_source_for_diff("ZTEST", "report"), "REPORT ztest.")

    def test_no_src_uri_raises(self):
        client = _make_client()
        with self.assertRaises(SapCliError):
            client.read_source_for_diff("ZTEST", "functiongroup")


# ═══════════════════════════════════════════
#  CDS
# ═══════════════════════════════════════════

class TestGetCdsSource(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, text="define view Z_TEST as select from mara { mara.matnr };")
        self.assertIn("define view", client.get_cds_source("Z_TEST"))

    def test_not_found(self):
        client = _make_client()
        mock = _mock_resp(404)
        mock.raise_for_status.side_effect = Exception("404")
        client.session.get.return_value = mock
        with self.assertRaises(Exception):
            client.get_cds_source("NONEXISTENT")


class TestCreateCDS(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        lock_resp = _mock_resp(200, content=(
            f'<lock xmlns="{LOCK_NS}"><lockHandle>lh_cds</lockHandle></lock>'
        ).encode())
        create_resp = _mock_resp(201, headers={"Location": "/sap/bc/adt/dds/ddl/sources/z_test"})
        unlock_resp = _mock_resp(200)
        client.session.post.side_effect = [create_resp, lock_resp, unlock_resp]
        client.session.put.return_value = _mock_resp(200)
        obj_uri, src_uri = client.create_cds("Z_TEST", "Test", "define view Z_TEST as select from mara { mara.matnr };")
        self.assertIn("z_test", obj_uri)

    def test_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500, text="Error")
        with self.assertRaises(CreateError):
            client.create_cds("Z_TEST", "Test", "ddl source")


# ═══════════════════════════════════════════
#  包管理
# ═══════════════════════════════════════════

class TestPackage(unittest.TestCase):

    def test_create_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(201)
        self.assertTrue(client.create_package("Z_TEST_PKG"))

    def test_create_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500, text="Error")
        with self.assertRaises(CreateError):
            client.create_package("Z_TEST_PKG")

    def test_get_info_success(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<pkg:package xmlns:pkg="http://www.sap.com/adt/packages" pkg:name="Z_TEST">'
            f'<pkg:description>Test</pkg:description></pkg:package>'
        ).encode())
        self.assertIsInstance(client.get_package_info("Z_TEST"), dict)

    def test_get_info_not_found(self):
        client = _make_client()
        mock = _mock_resp(404)
        mock.raise_for_status.side_effect = Exception("404")
        client.session.get.return_value = mock
        with self.assertRaises(Exception):
            client.get_package_info("NONEXISTENT")


# ═══════════════════════════════════════════
#  传输管理
# ═══════════════════════════════════════════

class TestTransportInfo(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<tm:transportorganizer xmlns:tm="{TM_NS}">'
            f'<tm:request tm:number="DEVK001" tm:description="Test" tm:status="D" tm:owner="TESTUSER"/>'
            f'</tm:transportorganizer>'
        ).encode())
        result = client.transport_info("DEVK001")
        self.assertEqual(result["number"], "DEVK001")

    def test_not_found(self):
        client = _make_client()
        mock = _mock_resp(404)
        mock.raise_for_status.side_effect = Exception("404")
        client.session.get.return_value = mock
        with self.assertRaises(Exception):
            client.transport_info("NONEXISTENT")


class TestTransportRelease(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200)
        self.assertTrue(client.transport_release("DEVK001"))

    def test_failure(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(500, text="Error")
        with self.assertRaises(SapCliError):
            client.transport_release("DEVK001")


class TestTransportObjects(unittest.TestCase):

    def test_found(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<cts:transport xmlns:cts="http://www.sap.com/adt/cts">'
            f'<cts:objects><cts:object cts:pgmid="R3TR" cts:object="PROG" cts:objName="Z_TEST"/></cts:objects>'
            f'</cts:transport>'
        ).encode())
        self.assertIsInstance(client.transport_objects("DEVK001"), list)

    def test_empty(self):
        client = _make_client()
        client.session.get.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<cts:transport xmlns:cts="http://www.sap.com/adt/cts"><cts:objects/></cts:transport>'
        ).encode())
        self.assertEqual(client.transport_objects("DEVK001"), [])


# ═══════════════════════════════════════════
#  代码质量
# ═══════════════════════════════════════════

class TestATCCheck(unittest.TestCase):

    def test_clean(self):
        client = _make_client()
        xml_body = (
            f'<?xml version="1.0"?>'
            f'<chk:checks xmlns:chk="{CHK_NS}"><chk:checkList/></chk:checks>'
        )
        client.session.post.return_value = _mock_resp(200, text=xml_body, content=xml_body.encode())
        ok, findings = client.atc_check("ZTEST", "/uri")
        self.assertTrue(ok)

    def test_with_findings(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=(
            f'<?xml version="1.0"?>'
            f'<chk:checks xmlns:chk="{CHK_NS}">'
            f'<chk:msg type="E" line="1" href="">'
            f'<chk:txt>Unused variable</chk:txt>'
            f'</chk:msg>'
            f'</chk:checks>'
        ).encode())
        ok, findings = client.atc_check("ZTEST", "/uri")
        self.assertFalse(ok)


class TestPrettyPrint(unittest.TestCase):

    def test_success(self):
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, text="FORMATTED CODE")
        self.assertEqual(client.pretty_print("raw code"), "FORMATTED CODE")

    def test_error(self):
        """pretty_print 用 raise_for_status() 抛 HTTPError。"""
        client = _make_client()
        mock = _mock_resp(500, text="Error")
        mock.raise_for_status.side_effect = Exception("500 Server Error")
        client.session.post.return_value = mock
        with self.assertRaises(Exception):
            client.pretty_print("raw code")


# ═══════════════════════════════════════════
#  _build_create_body 各类型
# ═══════════════════════════════════════════

class TestBuildCreateBodyTypes(unittest.TestCase):

    def _check(self, obj_type, name):
        body = _make_client()._build_create_body(obj_type, name, f"Test {obj_type}")
        self.assertIn(name, body)

    def test_report(self):       self._check("report", "ZTEST")
    def test_class(self):        self._check("class", "ZCL_TEST")
    def test_interface(self):    self._check("interface", "ZIF_TEST")
    def test_domain(self):       self._check("domain", "ZTEST_DOM")
    def test_dataelement(self):  self._check("dataelement", "ZTEST_DE")
    def test_table(self):        self._check("table", "ZTEST_TAB")
    def test_structure(self):    self._check("structure", "ZTEST_STR")
    def test_tabletype(self):    self._check("tabletype", "ZTEST_TT")

    def test_unsupported(self):
        with self.assertRaises(ValueError):
            _make_client()._build_create_body("bogus_type", "ZTEST", "Test")


# ═══════════════════════════════════════════
#  set_source 错误重试路径（行 59-69）
# ═══════════════════════════════════════════

class TestSetSourceRetry(unittest.TestCase):

    def _error_body(self, corrnr):
        return f'<error><entry key="corrNr">{corrnr}</entry></error>'

    def test_retry_with_extracted_corrnr_succeeds(self):
        """首次 400 + 错误体含不同 corrNr → 用该 corrNr 重试成功。"""
        client = _make_client()
        client.session.put.side_effect = [
            _mock_resp(400, text=self._error_body("DEVK999")),
            _mock_resp(200),
        ]
        self.assertTrue(client.set_source("/src", "code", "lh", corr_nr="DEVK001"))
        self.assertEqual(client.session.put.call_count, 2)

    def test_retry_fails_raises(self):
        """重试后仍 400 → raise_for_status。"""
        client = _make_client()
        first = _mock_resp(400, text=self._error_body("DEVK999"))
        retry = _mock_resp(400)
        retry.raise_for_status.side_effect = Exception("still 400")
        client.session.put.side_effect = [first, retry]
        with self.assertRaises(Exception):
            client.set_source("/src", "code", "lh", corr_nr="DEVK001")

    def test_no_corrnr_in_error_raises(self):
        """错误体无 corrNr → 不重试，直接 raise。"""
        client = _make_client()
        resp = _mock_resp(400, text="<error>boom</error>")
        resp.raise_for_status.side_effect = Exception("400")
        client.session.put.return_value = resp
        with self.assertRaises(Exception):
            client.set_source("/src", "code", "lh")
        self.assertEqual(client.session.put.call_count, 1)


# ═══════════════════════════════════════════
#  lock 高级路径（行 94, 101-107）
# ═══════════════════════════════════════════

class TestLockAdvanced(unittest.TestCase):

    def test_lock_with_corr_nr(self):
        """传 corr_nr → 写入 params 并原样返回（不提取）。"""
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, content=(
            f'<lock xmlns="{LOCK_NS}"><lockHandle>lh_123</lockHandle></lock>'
        ).encode())
        handle, corr = client.lock("/uri", corr_nr="DEVK001")
        self.assertEqual(handle, "lh_123")
        self.assertEqual(corr, "DEVK001")

    def test_lock_500_retry_with_existing_corrnr(self):
        """首次 500 + 错误体含 corrNr → 用该 corrNr 重试成功。"""
        client = _make_client()
        error_body = '<error><entry key="corrNr">DEVK999</entry></error>'
        client.session.post.side_effect = [
            _mock_resp(500, text=error_body),
            _mock_resp(200, content=(
                f'<lock xmlns="{LOCK_NS}"><lockHandle>lh</lockHandle></lock>'
            ).encode()),
        ]
        handle, corr = client.lock("/uri", corr_nr="DEVK001")
        self.assertEqual(handle, "lh")
        self.assertEqual(corr, "DEVK999")
        self.assertEqual(client.session.post.call_count, 2)


# ═══════════════════════════════════════════
#  _extract_locked_corrnr（行 135-149）
# ═══════════════════════════════════════════

class TestExtractLockedCorrnr(unittest.TestCase):

    def test_key_corrnr(self):
        client = _make_client()
        body = '<root><entry key="corrNr">DEVK001</entry></root>'
        self.assertEqual(client._extract_locked_corrnr(body), "DEVK001")

    def test_star_skipped_no_message(self):
        client = _make_client()
        body = '<root><entry key="corrNr">*</entry></root>'
        self.assertIsNone(client._extract_locked_corrnr(body))

    def test_message_branch(self):
        client = _make_client()
        body = '<root><message>Object is locked in request DEVK888</message></root>'
        self.assertEqual(client._extract_locked_corrnr(body), "DEVK888")

    def test_invalid_xml(self):
        client = _make_client()
        self.assertIsNone(client._extract_locked_corrnr("not valid xml <<"))


# ═══════════════════════════════════════════
#  _extract_corrnr_from_lock_response（行 166, 172-175, 183-185）
# ═══════════════════════════════════════════

class TestExtractCorrnrFromLockResponse(unittest.TestCase):

    def _extract(self, xml):
        return _make_client()._extract_corrnr_from_lock_response(ET.fromstring(xml))

    def test_strategy1_cornn_tag(self):
        self.assertEqual(
            self._extract('<lock><CORRN>DEVK001</CORRN></lock>'),
            "DEVK001",
        )

    def test_strategy1_star_skipped(self):
        self.assertIsNone(
            self._extract('<lock><CORRN>*</CORRN></lock>'),
        )

    def test_strategy2_attr(self):
        self.assertEqual(
            self._extract('<lock><meta myCorrNr="DEVK003"/></lock>'),
            "DEVK003",
        )

    def test_strategy2_tagname(self):
        self.assertEqual(
            self._extract('<lock><myCorrField>DEVK004</myCorrField></lock>'),
            "DEVK004",
        )

    def test_strategy3_property_key(self):
        self.assertEqual(
            self._extract('<lock><property key="corrNr">DEVK005</property></lock>'),
            "DEVK005",
        )

    def test_no_match_returns_none(self):
        self.assertIsNone(self._extract('<lock><other>val</other></lock>'))


# ═══════════════════════════════════════════
#  activate — inactivectsobjects 路径（行 235-265）
# ═══════════════════════════════════════════

class TestActivateInactiveCtsObjects(unittest.TestCase):

    IOC_NS = "http://www.sap.com/abapxml/inactiveCtsObjects"

    def _activate_resp(self, xml):
        return _mock_resp(
            200, text=xml, content=xml.encode(),
            headers={"content-type": "application/vnd.sap.adt.inactivectsobjects+xml"},
        )

    def test_with_objects_via_fallback_ref(self):
        """ref 不在 adtcore ns → 走 endswith}ref 回退路径。"""
        xml = (
            f'<?xml version="1.0"?>'
            f'<asx:abap xmlns:asx="http://www.sap.com/abapxml"'
            f' xmlns:ioc="{self.IOC_NS}" xmlns:adtcore="{ADTCORE_NS}">'
            f'<ioc:entry><ioc:object>'
            f'<ioc:ref adtcore:name="Z_INACTIVE" adtcore:type="CLAS/OC"/>'
            f'</ioc:object></ioc:entry>'
            f'</asx:abap>'
        )
        client = _make_client()
        client.session.post.return_value = self._activate_resp(xml)
        ok, msgs = client.activate("ztest", "/uri")
        self.assertFalse(ok)
        self.assertTrue(any("Z_INACTIVE" in m["text"] for m in msgs))

    def test_with_objects_via_adtcore_ref(self):
        """ref 在 adtcore ns → obj_elem.find 直接命中。"""
        xml = (
            f'<?xml version="1.0"?>'
            f'<asx:abap xmlns:asx="http://www.sap.com/abapxml"'
            f' xmlns:ioc="{self.IOC_NS}" xmlns:adtcore="{ADTCORE_NS}">'
            f'<ioc:entry><ioc:object>'
            f'<adtcore:ref adtcore:name="Z_ADTC" adtcore:type="INTF/OI"/>'
            f'</ioc:object></ioc:entry>'
            f'</asx:abap>'
        )
        client = _make_client()
        client.session.post.return_value = self._activate_resp(xml)
        ok, msgs = client.activate("ztest", "/uri")
        self.assertFalse(ok)
        self.assertTrue(any("Z_ADTC" in m["text"] for m in msgs))

    def test_empty_no_objects(self):
        """inactivectsobjects 响应但无 entry → 视为成功。"""
        xml = (
            f'<?xml version="1.0"?>'
            f'<asx:abap xmlns:asx="http://www.sap.com/abapxml" xmlns:ioc="{self.IOC_NS}">'
            f'</asx:abap>'
        )
        client = _make_client()
        client.session.post.return_value = self._activate_resp(xml)
        ok, msgs = client.activate("ztest", "/uri")
        self.assertTrue(ok)
        self.assertEqual(msgs, [])


# ═══════════════════════════════════════════
#  activate — corr_nr / msg 回退 / verify 异常（行 219, 316-317, 334-337）
# ═══════════════════════════════════════════

class TestActivateMisc(unittest.TestCase):

    def test_activate_with_corr_nr(self):
        """传 corr_nr → 写入 activation params。"""
        client = _make_client()
        xml = f'<?xml version="1.0"?><adtcore:activation xmlns:adtcore="{ADTCORE_NS}"/>'
        client.session.post.return_value = _mock_resp(200, text=xml, content=xml.encode())
        ok, msgs = client.activate("ztest", "/uri", corr_nr="DEVK001")
        self.assertTrue(ok)
        self.assertEqual(msgs, [])

    def test_msg_without_txt_uses_fallback_text(self):
        """msg 无 }txt 子元素 → 用第一个长文本后代（行 334-337）。"""
        client = _make_client()
        xml = (
            f'<?xml version="1.0"?>'
            f'<adtcore:activation xmlns:adtcore="{ADTCORE_NS}">'
            f'<adtcore:objectReference type="E" line="1" href="">'
            f'<adtcore:description>Some long activation failure text</adtcore:description>'
            f'</adtcore:objectReference>'
            f'</adtcore:activation>'
        )
        client.session.post.return_value = _mock_resp(200, text=xml, content=xml.encode())
        ok, msgs = client.activate("ztest", "/uri")
        self.assertFalse(ok)
        self.assertTrue(msgs)
        self.assertIn("Some long", msgs[0]["text"])

    def test_empty_response_verify_raises(self):
        """空响应 + verify GET 抛异常 → 捕获后视为成功（行 316-317）。"""
        client = _make_client()
        client.session.post.return_value = _mock_resp(200, text="", content=b"")
        client.session.get.side_effect = Exception("verify boom")
        ok, msgs = client.activate("ztest", "/uri")
        self.assertTrue(ok)
        self.assertEqual(msgs, [])


# ═══════════════════════════════════════════
#  syntax_check — msg 回退文本（行 390-393）
# ═══════════════════════════════════════════

class TestSyntaxCheckMsgFallback(unittest.TestCase):

    def test_msg_without_txt_uses_fallback_text(self):
        client = _make_client()
        xml = (
            f'<?xml version="1.0"?>'
            f'<chk:checks xmlns:chk="{CHK_NS}">'
            f'<chk:msg type="E" line="1" href="">'
            f'<chk:note>Another long syntax failure note</chk:note>'
            f'</chk:msg>'
            f'</chk:checks>'
        )
        client.session.post.return_value = _mock_resp(200, text=xml, content=xml.encode())
        ok, msgs = client.syntax_check("ztest", "/uri")
        self.assertFalse(ok)
        self.assertTrue(msgs)
        self.assertIn("Another long", msgs[0]["text"])


if __name__ == "__main__":
    unittest.main(verbosity=2)