姚章浩
52 lines
2.6 KiB
Python
52 lines
2.6 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
import dws_dingtalk_scan as scan
|
|
from source_catalog_import import build_dingtalk_sources, validate_source
|
|
|
|
|
|
class ScanTests(unittest.TestCase):
|
|
def test_follows_cursor_instead_of_accepting_truncated_inventory(self):
|
|
pages = [dict(nodes=[{'nodeId': 'f1', 'name': 'one', 'type': 'file'}], hasMore=True, nextCursor='next'),
|
|
dict(nodes=[{'nodeId': 'f2', 'name': 'two', 'type': 'file'}], hasMore=False, nextCursor='')]
|
|
with patch.object(scan, 'dws', side_effect=pages) as cli:
|
|
self.assertEqual(len(scan.list_wiki('wiki', '')), 2)
|
|
self.assertIn('--cursor', cli.call_args_list[1].args[0])
|
|
|
|
def test_rejects_missing_items_and_repeated_cursor(self):
|
|
for data in [{'hasMore': False}, {'nodes': [], 'hasMore': True, 'nextCursor': 'same'}]:
|
|
with patch.object(scan, 'dws', return_value=data):
|
|
with self.assertRaises(RuntimeError):
|
|
scan.list_wiki('wiki', '')
|
|
|
|
def test_files_with_children_are_not_walked_as_folders(self):
|
|
item = {'nodeId': 'doc', 'name': 'document', 'type': 'file', 'hasChildren': True}
|
|
self.assertFalse(scan.normalise(item, 'root')[2])
|
|
|
|
def test_output_retains_file_extension_and_valid_hierarchy(self):
|
|
with patch.object(scan, 'list_wiki', side_effect=[
|
|
[{'nodeId': 'dir', 'name': '目录', 'type': 'folder'}],
|
|
[{'nodeId': 'doc', 'name': '说明', 'type': 'file', 'extension': 'adoc', 'fileSize': 10}]
|
|
]):
|
|
nodes = scan.walk('space', 'wiki', 'root', lambda *args: None)
|
|
self.assertEqual(nodes['doc']['name'], '说明.adoc')
|
|
source = build_dingtalk_sources({'roots': {'space-space': {
|
|
'configured_root_id': 'root', 'live_space_name': '测试', 'nodes': nodes}}})[0]
|
|
validate_source(source)
|
|
self.assertEqual(source['file_count'], 1)
|
|
self.assertEqual(source['folder_count'], 2)
|
|
|
|
def test_child_environment_uses_shared_home_without_database_secret(self):
|
|
import os
|
|
with patch.dict(os.environ, {'DWS_HOME': '/shared/home', 'DATABASE_URL': 'private'}, clear=False):
|
|
with patch.object(scan.subprocess, 'run') as run:
|
|
run.return_value.returncode = 0
|
|
run.return_value.stdout = '{"ok":true,"data":{}}'
|
|
scan.dws(['wiki', '+node-list'])
|
|
env = run.call_args.kwargs['env']
|
|
self.assertEqual(env['HOME'], '/shared/home')
|
|
self.assertNotIn('DATABASE_URL', env)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|