60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
import sys
|
|
import types
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
base_module = types.ModuleType("app.jobs.base")
|
|
|
|
|
|
class BaseJob:
|
|
pass
|
|
|
|
|
|
base_module.BaseJob = BaseJob
|
|
sys.modules["app.jobs.base"] = base_module
|
|
|
|
api_module = types.ModuleType("extensions.sync_ehr_to_ad.api")
|
|
api_module.ActiveDirectoryClient = object
|
|
api_module.SyncEhrToAdApi = object
|
|
sys.modules["extensions.sync_ehr_to_ad.api"] = api_module
|
|
|
|
spec = importlib.util.spec_from_file_location("sync_ehr_to_ad_job", ROOT / "extensions/sync_ehr_to_ad/job.py")
|
|
job_module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(job_module)
|
|
|
|
_is_syncable_employee_record = job_module._is_syncable_employee_record
|
|
|
|
|
|
def _ehr_item(*, employee_status="离职", last_work_date="", emp_flags=None, rec_flags=None):
|
|
emp_flags = emp_flags or {}
|
|
rec_flags = rec_flags or {}
|
|
return {
|
|
"employeeInfo": {
|
|
"translateProperties": {"EmployeeStatus": employee_status},
|
|
"customProperties": {"extADAccountName_606508_511687157": "zhangsan"},
|
|
**emp_flags,
|
|
},
|
|
"recordInfo": {
|
|
"lastWorkDate": last_work_date,
|
|
**rec_flags,
|
|
},
|
|
}
|
|
|
|
|
|
def test_syncable_employee_record_includes_any_employee_status():
|
|
assert _is_syncable_employee_record(_ehr_item(employee_status="离职")) is True
|
|
|
|
|
|
def test_syncable_employee_record_includes_records_with_last_work_date():
|
|
assert _is_syncable_employee_record(_ehr_item(last_work_date="2026-07-01")) is True
|
|
|
|
|
|
def test_syncable_employee_record_excludes_deleted_or_disabled_records():
|
|
assert _is_syncable_employee_record(_ehr_item(emp_flags={"isDeleted": True})) is False
|
|
assert _is_syncable_employee_record(_ehr_item(rec_flags={"disabled": True})) is False
|