Codex
72c666f165
Align the generated Kingdee token request with the official enhanced-token document: client_id, client_secret, username, accountId, language, automatic nonce, and automatic current timestamp. Treat expires_in as milliseconds by default and update docs/tests accordingly. Validation: python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests; git diff --check.
276 lines
11 KiB
Python
276 lines
11 KiB
Python
import json
|
|
import os
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
|
|
from obsidian_wiki.kingdee_api import (
|
|
DEFAULT_SALORDER_API_URL,
|
|
ensure_kingdee_config,
|
|
fetch_kingdee_sales_orders,
|
|
get_kingdee_access_token,
|
|
)
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def read(self):
|
|
return json.dumps(self.payload).encode("utf-8")
|
|
|
|
|
|
class RecordingUrlopen:
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def __call__(self, request, timeout=60):
|
|
self.calls.append({"request": request, "timeout": timeout})
|
|
if not self.responses:
|
|
raise AssertionError("unexpected HTTP request")
|
|
return FakeResponse(self.responses.pop(0))
|
|
|
|
|
|
class KingdeeApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tempdir.name)
|
|
|
|
def tearDown(self):
|
|
self.tempdir.cleanup()
|
|
|
|
def test_config_command_creates_private_template_without_overwriting(self):
|
|
config_path = self.root / ".runtime" / "kingdee.env"
|
|
|
|
created = ensure_kingdee_config(config_path)
|
|
config_path.write_text(config_path.read_text(encoding="utf-8") + "\nKINGDEE_TOKEN_URL=https://token.example\n", encoding="utf-8")
|
|
created_again = ensure_kingdee_config(config_path)
|
|
|
|
self.assertTrue(created)
|
|
self.assertFalse(created_again)
|
|
self.assertEqual(stat.S_IMODE(config_path.stat().st_mode), 0o600)
|
|
text = config_path.read_text(encoding="utf-8")
|
|
self.assertIn(f"KINGDEE_SALORDER_API_URL={DEFAULT_SALORDER_API_URL}", text)
|
|
self.assertIn("KINGDEE_TOKEN_BODY_JSON=", text)
|
|
self.assertIn("KINGDEE_TOKEN_URL=https://token.example", text)
|
|
self.assertIn("是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。", text)
|
|
self.assertIn("用途：销售订单 API 要求的 x-acgw-identity 请求头。", text)
|
|
self.assertIn("是否必填：可选，手动测试用；长期定时同步建议使用自动登录配置。", text)
|
|
self.assertIn("是否必填：可选，默认从 data.access_token 读取。", text)
|
|
self.assertIn("变量：KINGDEE_CLIENT_ID", text)
|
|
self.assertIn("变量：KINGDEE_USERNAME", text)
|
|
self.assertIn("变量：KINGDEE_ACCOUNT_ID", text)
|
|
self.assertIn("nonce 和 timestamp 会按官方文档自动生成", text)
|
|
|
|
def test_default_config_path_accepts_none(self):
|
|
cwd = Path.cwd()
|
|
try:
|
|
os.chdir(self.root)
|
|
created = ensure_kingdee_config(None)
|
|
finally:
|
|
os.chdir(cwd)
|
|
|
|
self.assertTrue(created)
|
|
self.assertTrue((self.root / ".runtime" / "kingdee.env").is_file())
|
|
|
|
def test_cli_config_kingdee_writes_template(self):
|
|
from obsidian_wiki.cli import main
|
|
|
|
config_path = self.root / ".runtime" / "kingdee.env"
|
|
output = StringIO()
|
|
with redirect_stdout(output):
|
|
code = main(["business", "config", "kingdee", "--config", str(config_path)])
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(config_path.is_file())
|
|
self.assertIn("Kingdee config", output.getvalue())
|
|
|
|
def test_access_token_is_fetched_and_reused_from_private_cache(self):
|
|
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
|
config = {
|
|
"KINGDEE_TOKEN_URL": "https://token.example/access",
|
|
"KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc","app_secret":"secret"}',
|
|
"KINGDEE_TOKEN_HEADERS_JSON": '{"X-App":"demo"}',
|
|
"KINGDEE_TOKEN_FIELD": "data.access_token",
|
|
"KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
|
|
"KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
|
|
}
|
|
http = RecordingUrlopen(
|
|
[
|
|
{
|
|
"data": {
|
|
"access_token": "access-1",
|
|
"expires_in": 3600000,
|
|
"refresh_token": "refresh-1",
|
|
}
|
|
}
|
|
]
|
|
)
|
|
|
|
token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
|
|
cached = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1100)
|
|
|
|
self.assertEqual(token, "access-1")
|
|
self.assertEqual(cached, "access-1")
|
|
self.assertEqual(len(http.calls), 1)
|
|
request = http.calls[0]["request"]
|
|
self.assertEqual(request.get_method(), "POST")
|
|
self.assertEqual(request.full_url, "https://token.example/access")
|
|
self.assertEqual(json.loads(request.data.decode("utf-8"))["app_key"], "abc")
|
|
self.assertEqual(request.get_header("X-app"), "demo")
|
|
self.assertEqual(stat.S_IMODE(cache_path.stat().st_mode), 0o600)
|
|
|
|
def test_access_token_body_defaults_to_official_fields_with_nonce_and_current_time(self):
|
|
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
|
config = {
|
|
"KINGDEE_TOKEN_URL": "https://token.example/access",
|
|
"KINGDEE_CLIENT_ID": "thirdapp",
|
|
"KINGDEE_CLIENT_SECRET": "secret",
|
|
"KINGDEE_USERNAME": "zhangSan",
|
|
"KINGDEE_ACCOUNT_ID": "1355633519610561531",
|
|
"KINGDEE_LANGUAGE": "zh_CN",
|
|
"KINGDEE_TOKEN_FIELD": "data.access_token",
|
|
}
|
|
http = RecordingUrlopen([{"data": {"access_token": "access-1"}}])
|
|
|
|
token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
|
|
|
|
self.assertEqual(token, "access-1")
|
|
body = json.loads(http.calls[0]["request"].data.decode("utf-8"))
|
|
self.assertEqual(body["client_id"], "thirdapp")
|
|
self.assertEqual(body["client_secret"], "secret")
|
|
self.assertEqual(body["username"], "zhangSan")
|
|
self.assertEqual(body["accountId"], "1355633519610561531")
|
|
self.assertEqual(body["language"], "zh_CN")
|
|
self.assertRegex(body["nonce"], r"^[0-9a-f]{32}$")
|
|
self.assertEqual(body["timestamp"], "1970-01-01T00:16:40.000Z")
|
|
|
|
def test_expired_token_uses_refresh_endpoint(self):
|
|
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
|
cache_path.parent.mkdir(parents=True)
|
|
cache_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"access_token": "old-token",
|
|
"refresh_token": "refresh-1",
|
|
"expires_at": 999,
|
|
"obtained_at": 100,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
config = {
|
|
"KINGDEE_TOKEN_URL": "https://token.example/access",
|
|
"KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc"}',
|
|
"KINGDEE_REFRESH_URL": "https://token.example/refresh",
|
|
"KINGDEE_REFRESH_BODY_JSON": '{"refresh_token":"{refresh_token}"}',
|
|
"KINGDEE_TOKEN_FIELD": "data.access_token",
|
|
"KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
|
|
"KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
|
|
}
|
|
http = RecordingUrlopen(
|
|
[
|
|
{
|
|
"data": {
|
|
"access_token": "new-token",
|
|
"expires_in": 3600000,
|
|
"refresh_token": "refresh-2",
|
|
}
|
|
}
|
|
]
|
|
)
|
|
|
|
token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
|
|
|
|
self.assertEqual(token, "new-token")
|
|
request = http.calls[0]["request"]
|
|
self.assertEqual(request.full_url, "https://token.example/refresh")
|
|
self.assertEqual(json.loads(request.data.decode("utf-8"))["refresh_token"], "refresh-1")
|
|
|
|
def test_sales_order_fetch_posts_filters_headers_and_paginates(self):
|
|
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
|
cache_path.parent.mkdir(parents=True)
|
|
cache_path.write_text(
|
|
json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
|
|
encoding="utf-8",
|
|
)
|
|
config = {
|
|
"KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
|
|
"KINGDEE_ACGW_IDENTITY": "identity-1",
|
|
"KINGDEE_PAGE_SIZE": "2",
|
|
}
|
|
page_1 = {
|
|
"status": True,
|
|
"errorCode": "",
|
|
"message": "",
|
|
"data": {
|
|
"pageNo": 1,
|
|
"pageSize": 2,
|
|
"rows": [{"id": "1", "billno": "SO-001"}, {"id": "2", "billno": "SO-002"}],
|
|
"totalCount": 3,
|
|
},
|
|
}
|
|
page_2 = {
|
|
"status": True,
|
|
"errorCode": "",
|
|
"message": "",
|
|
"data": {
|
|
"pageNo": 2,
|
|
"pageSize": 2,
|
|
"rows": [{"id": "3", "billno": "SO-003"}],
|
|
"totalCount": 3,
|
|
},
|
|
}
|
|
http = RecordingUrlopen([page_1, page_2])
|
|
|
|
rows, payload = fetch_kingdee_sales_orders(
|
|
config,
|
|
filters={"start_bizdate": "2026-06-01", "billstatus": ["C"]},
|
|
token_cache_path=cache_path,
|
|
urlopen=http,
|
|
now_epoch=1000,
|
|
)
|
|
|
|
self.assertEqual([row["billno"] for row in rows], ["SO-001", "SO-002", "SO-003"])
|
|
self.assertEqual(payload["pages_fetched"], 2)
|
|
self.assertEqual(len(http.calls), 2)
|
|
first = http.calls[0]["request"]
|
|
self.assertEqual(first.get_method(), "POST")
|
|
self.assertEqual(first.get_header("Accesstoken"), "access-1")
|
|
self.assertEqual(first.get_header("X-acgw-identity"), "identity-1")
|
|
body = json.loads(first.data.decode("utf-8"))
|
|
self.assertEqual(body["data"]["start_bizdate"], "2026-06-01")
|
|
self.assertEqual(body["data"]["billstatus"], ["C"])
|
|
self.assertEqual(body["pageNo"], 1)
|
|
self.assertEqual(body["pageSize"], 2)
|
|
|
|
def test_sales_order_api_failure_reports_code_and_message(self):
|
|
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
|
cache_path.parent.mkdir(parents=True)
|
|
cache_path.write_text(
|
|
json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
|
|
encoding="utf-8",
|
|
)
|
|
config = {
|
|
"KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
|
|
"KINGDEE_ACGW_IDENTITY": "identity-1",
|
|
}
|
|
http = RecordingUrlopen([{"status": False, "errorCode": "E001", "message": "bad filter"}])
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "E001.*bad filter"):
|
|
fetch_kingdee_sales_orders(config, filters={}, token_cache_path=cache_path, urlopen=http, now_epoch=1000)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|