姚章浩
b3c9331966
- filename_filter_rules table (000035) with source-root and per-directory rows - additive inheritance: source default + every ancestor directory contributes - filename_is_filtered() SQL predicate applied to catalog browsing, subtree counts, change-set diffing, routing fan-out and artifact fetch manifests - fetch_state 'filtered' (CHECK extended); reconcile cancels/restores fetch jobs, retires planned publications and blocks items when rules change - GET/PUT/DELETE /api/v1/filename-filters (Idempotency-Key, normalized extensions) + Settings and Catalog filter editors - failJobSQL no longer swallows the claimed row when the failure projection upserts (fixes spurious 'lease lost' retries) - DingTalk live fetch: adoc/axls now use the export-job pipeline (HTTP 403 from file/download fixed), header-signature downloads supported
48 lines
2.3 KiB
Python
48 lines
2.3 KiB
Python
import importlib.util
|
|
import pathlib
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
MODULE_PATH = pathlib.Path(__file__).with_name("artifact_fetch.py")
|
|
spec = importlib.util.spec_from_file_location("artifact_fetch", MODULE_PATH)
|
|
assert spec is not None and spec.loader is not None
|
|
artifact_fetch = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(artifact_fetch)
|
|
|
|
|
|
class DingTalkOnlineDocumentExportTest(unittest.TestCase):
|
|
def test_adoc_uses_export_job_instead_of_ordinary_file_download(self):
|
|
calls = []
|
|
|
|
def api(url, headers, payload=None, extra_headers=None, method=None):
|
|
calls.append((url, payload))
|
|
if "/dentry/info" in url:
|
|
return {"data": {"docKey": "doc-key", "dentryKey": "dentry-key", "extension": "adoc"}}
|
|
if "/api/document/data" in url:
|
|
return {"data": {"documentContent": {"checkpoint": {"content": "document-json"}}}}
|
|
if "/upload_info" in url:
|
|
return {"data": {"storagePath": "tmp/path", "uploadUrl": "https://oss.example/upload"}}
|
|
if "/submitExportJob" in url:
|
|
return {"data": {"jobId": "job-1"}}
|
|
if "/queryExportJobInfo" in url:
|
|
return {"data": {"status": "success", "ossUrl": "https://oss.example/result.docx"}}
|
|
self.fail("unexpected API " + url)
|
|
|
|
with mock.patch.object(artifact_fetch, "dingtalk_json_api", side_effect=api), \
|
|
mock.patch.object(artifact_fetch, "dingtalk_put_bytes") as put, \
|
|
mock.patch.object(artifact_fetch, "http_get", return_value=("/tmp/result", 123, "a" * 64)) as download:
|
|
result = artifact_fetch.live_export_dingtalk_document(
|
|
"raw-node", {"a-token": "x", "x-xsrf-token": "y", "cookie": "z", "referer": "https://alidocs.dingtalk.com/"},
|
|
{"name": "说明.adoc", "relative_path": "说明.adoc"}, 1_000_000,
|
|
)
|
|
|
|
self.assertEqual(result, ("/tmp/result", 123, "a" * 64, False))
|
|
self.assertTrue(any("submitExportJob" in url for url, _ in calls))
|
|
self.assertFalse(any("/file/download" in url for url, _ in calls))
|
|
put.assert_called_once()
|
|
download.assert_called_once_with("https://oss.example/result.docx", {}, 180, 1_000_000)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|