姚章浩
66 lines
4.2 KiB
Python
66 lines
4.2 KiB
Python
import importlib.util
|
|
import logging
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
spec=importlib.util.spec_from_file_location('qa_ops_directory_sync',Path(__file__).with_name('qa_ops_directory_sync.py'))
|
|
assert spec is not None and spec.loader is not None
|
|
sync=importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(sync)
|
|
|
|
class FakeClient:
|
|
class Config:
|
|
dry_run=False
|
|
def __init__(self):
|
|
self.config=self.Config()
|
|
|
|
def full_departments(client,logger,include,exclude):
|
|
return {10:{'id':10,'parent_id':1,'name':'示例业务'}},set(),set()
|
|
def full_users(client,logger,ids,enable_user_detail):
|
|
if enable_user_detail:
|
|
raise AssertionError('Must not inherit silent per-user detail failures or guessed primary department')
|
|
return {'fixture-user':{'user_id':'fixture-user','department_ids':{10},'raw_payload':{'unionid':'fixture-union'}}},set(),set()
|
|
def root_detail(client,logger,department_id):
|
|
return {'id':department_id,'parent_id':0,'name':'真实源端返回的示例根部门'},False,False
|
|
|
|
class DirectoryBridgeTests(unittest.TestCase):
|
|
def test_real_display_name_is_retained_without_guessing_or_requiring_it(self):
|
|
def users(*args,**kwargs):
|
|
return {'u':{'user_id':'u','name':'<示例员工>','department_ids':{10},'raw_payload':{}}},set(),set()
|
|
result=sync.collect_directory(FakeClient(),full_departments,users,root_detail,logging.getLogger('fixture'))
|
|
self.assertEqual(result['directory']['people'][0].get('name'),'<示例员工>')
|
|
result=sync.collect_directory(FakeClient(),full_departments,full_users,root_detail,logging.getLogger('fixture'))
|
|
self.assertNotIn('name',result['directory']['people'][0])
|
|
def test_fetches_the_real_missing_root_and_preserves_stable_identities(self):
|
|
result=sync.collect_directory(FakeClient(),full_departments,full_users,root_detail,logging.getLogger('fixture'))
|
|
self.assertTrue(result['complete'])
|
|
self.assertEqual(result['source'],'weekly_report_live')
|
|
self.assertEqual(result['directory']['departments'],[
|
|
{'id':'1','parent_id':'','name':'真实源端返回的示例根部门'},
|
|
{'id':'10','parent_id':'1','name':'示例业务'}])
|
|
self.assertEqual(result['directory']['people'],[{'id':'fixture-user','union_id':'fixture-union','department_ids':['10']}])
|
|
def test_inaccessible_root_is_not_replaced_with_a_fabricated_company(self):
|
|
def denied(*args):return None,True,False
|
|
with self.assertRaisesRegex(sync.DirectorySyncError,'directory_ancestor_unavailable'):
|
|
sync.collect_directory(FakeClient(),full_departments,full_users,denied,logging.getLogger('fixture'))
|
|
def test_partial_fetch_and_dry_run_do_not_become_complete_live_snapshots(self):
|
|
def partial(*args):return {},set(),{10}
|
|
with self.assertRaises(sync.DirectorySyncError):
|
|
sync.collect_directory(FakeClient(),partial,full_users,root_detail,logging.getLogger('fixture'))
|
|
client=FakeClient();client.config.dry_run=True
|
|
with self.assertRaises(sync.DirectorySyncError):
|
|
sync.collect_directory(client,full_departments,full_users,root_detail,logging.getLogger('fixture'))
|
|
def test_union_id_does_not_fall_back_to_open_id(self):
|
|
def users(*args,**kwargs):return {'u':{'user_id':'u','department_ids':{10},'union_id':'not-a-union','raw_payload':{'openId':'not-a-union'}}},set(),set()
|
|
result=sync.collect_directory(FakeClient(),full_departments,users,root_detail,logging.getLogger('fixture'))
|
|
self.assertEqual(result['directory']['people'][0]['union_id'],'')
|
|
def test_wrong_root_identity_and_cycles_are_rejected(self):
|
|
def wrong(*args):return {'id':999,'parent_id':0,'name':'wrong'},False,False
|
|
with self.assertRaises(sync.DirectorySyncError):
|
|
sync.collect_directory(FakeClient(),full_departments,full_users,wrong,logging.getLogger('fixture'))
|
|
def cyclic(*args):return {1:{'id':1,'parent_id':10,'name':'a'},10:{'id':10,'parent_id':1,'name':'b'}},set(),set()
|
|
with self.assertRaises(sync.DirectorySyncError):
|
|
sync.collect_directory(FakeClient(),cyclic,full_users,root_detail,logging.getLogger('fixture'))
|
|
|
|
if __name__=='__main__':unittest.main()
|