104 lines
5.2 KiB
Python
104 lines
5.2 KiB
Python
# coding=utf-8
|
|
import unittest
|
|
import inspect
|
|
from imp import reload
|
|
|
|
import arcpy
|
|
from functions import *
|
|
from sys_config import SysConfig
|
|
from logger import Logger
|
|
from sde_sync import SdeSynchronous
|
|
|
|
reload(sys)
|
|
sys.setdefaultencoding('utf8')
|
|
|
|
|
|
class TestSdeSynchronous(unittest.TestCase):
|
|
def setUp(self):
|
|
# 获取当前路径的父级路径
|
|
work_pathname = os.path.abspath(os.path.dirname(os.getcwd()))
|
|
# 同步开始时间
|
|
curr_sync_date = time.strftime("%Y-%m-%d", time.localtime())
|
|
curr_sync_time = time.strftime("%H-%M-%S", time.localtime())
|
|
# 设置当前工作路径
|
|
config = SysConfig()
|
|
config.pathname = work_pathname
|
|
config.curr_sync_datetime = curr_sync_date + " " + curr_sync_time
|
|
# 日志记录
|
|
create_dir(work_pathname + "\\log")
|
|
config.logfile_name = work_pathname + "\\log\\" + config.curr_sync_datetime + "-SdeSynchronous.txt"
|
|
logger = Logger(config.logfile_name)
|
|
logger.log("==SdeSynchronous Started==")
|
|
logger.log("Process begin at " + str(curr_sync_date) + " " + str(curr_sync_time) + ".")
|
|
self.sde_db_sync = SdeSynchronous(config, logger)
|
|
self.sde_db_sync.src_sde = self.sde_db_sync.get_connection(str(u"源库SDE连接"))
|
|
arcpy.env.workspace = self.sde_db_sync.src_sde
|
|
self.sde_db_sync.config.last_sync_datetime = self.sde_db_sync.get_last_sync_datetime()
|
|
self.edit = self.sde_db_sync.start_edit()
|
|
self.sde_db_sync.create_gdb_files()
|
|
|
|
def tearDown(self):
|
|
self.sde_db_sync.stop_edit(self.edit, False)
|
|
self.sde_db_sync.logger.close()
|
|
# 删除log日志文件
|
|
os.remove(self.sde_db_sync.config.logfile_name)
|
|
|
|
def test_get_last_max_object_id(self):
|
|
self.assertEqual(self.sde_db_sync.get_last_max_object_id("Building_A"), 11)
|
|
|
|
def test_get_last_sync_datetime(self):
|
|
print(self.sde_db_sync.get_last_sync_datetime())
|
|
|
|
def test_get_projects_info(self):
|
|
self.sde_db_sync.config.last_sync_datetime = "2022-10-01 00:00:00"
|
|
print(self.sde_db_sync.get_projects_info())
|
|
|
|
def test_create_gdb_files(self):
|
|
self.assertTrue(os.path.exists(self.sde_db_sync.out_gdb_path))
|
|
self.assertTrue(os.path.exists(self.sde_db_sync.out_gdb_path + "\\" + self.sde_db_sync.out_gdb_files["Add"]))
|
|
self.assertTrue(os.path.exists(self.sde_db_sync.out_gdb_path + "\\" + self.sde_db_sync.out_gdb_files["Edit"]))
|
|
self.assertTrue(os.path.exists(self.sde_db_sync.out_gdb_path + "\\" + self.sde_db_sync.out_gdb_files["Delete"]))
|
|
arcpy.Delete_management(self.sde_db_sync.out_gdb_path + "\\" + self.sde_db_sync.out_gdb_files["Add"])
|
|
arcpy.Delete_management(self.sde_db_sync.out_gdb_path + "\\" + self.sde_db_sync.out_gdb_files["Edit"])
|
|
arcpy.Delete_management(self.sde_db_sync.out_gdb_path + "\\" + self.sde_db_sync.out_gdb_files["Delete"])
|
|
|
|
def test_get_max_object_id(self):
|
|
self.assertEqual(self.sde_db_sync.get_max_object_id("Building_A"), 1)
|
|
|
|
def test_set_last_max_object_id(self):
|
|
self.sde_db_sync.set_last_max_object_id("Building_A", 11)
|
|
|
|
def test_get_sys_time_condition(self):
|
|
ld = self.sde_db_sync.config.last_sync_datetime
|
|
cd = self.sde_db_sync.config.curr_sync_datetime
|
|
str_cond1 = "UpdateTime>=TO_DATE('" + ld + "','YYYY-MM-DD HH24:MI:SS') and UpdateTime<TO_DATE('" + cd + "','YYYY-MM-DD HH24:MI:SS')"
|
|
str_cond2 = "DeleteTime>=TO_DATE('" + ld + "','YYYY-MM-DD HH24:MI:SS') and DeleteTime<TO_DATE('" + cd + "','YYYY-MM-DD HH24:MI:SS')"
|
|
self.assertEqual(self.sde_db_sync.get_sys_time_condition(False), str_cond1)
|
|
self.assertEqual(self.sde_db_sync.get_sys_time_condition(True), str_cond2)
|
|
|
|
def test_get_feature_expression(self):
|
|
ld = self.sde_db_sync.config.last_sync_datetime
|
|
cd = self.sde_db_sync.config.curr_sync_datetime
|
|
ss = self.sde_db_sync.get_sys_time_condition(False)
|
|
cond = "{} and FeatureGuid {} (SELECT FeatureGuid FROM historyBuilding_A WHERE DeleteTime>=TO_DATE('{}','YYYY-MM-DD HH24:MI:SS') and DeleteTime<TO_DATE('{}','YYYY-MM-DD HH24:MI:SS') and (UpdateTime<TO_DATE('{}','YYYY-MM-DD HH24:MI:SS') or UpdateTime IS NULL))"
|
|
add_cond = cond.format(ss, "NOT IN", ld, cd, ld)
|
|
edit_cond = cond.format(ss, "IN", ld, cd, ld)
|
|
ss = self.sde_db_sync.get_sys_time_condition(True)
|
|
cond = "{} and (historyBuilding_A.UpdateTime<TO_DATE('{last_time}','YYYY-MM-DD HH24:MI:SS') or historyBuilding_A.UpdateTime IS NULL) and (FeatureGuid NOT IN (SELECT FeatureGuid FROM Building_A))"
|
|
del_cond = cond.format(ss, last_time=ld)
|
|
c1, c2, c3 = self.sde_db_sync.get_export_expression("Building_A")
|
|
self.assertEqual(c1, add_cond)
|
|
self.assertEqual(c2, edit_cond)
|
|
self.assertEqual(c3, del_cond)
|
|
|
|
def test_featureclass_to_featureclass(self):
|
|
self.sde_db_sync.create_gdb_files()
|
|
self.sde_db_sync.featureclass_to_featureclass("Building_A", self.sde_db_sync.out_gdb_path + "\\" + self.sde_db_sync.out_gdb_files["Add"], "Building_A", "ObjectID>0")
|
|
|
|
def test_copy_to_mdo(self):
|
|
self.sde_db_sync.copy_to_mdo(self.sde_db_sync.out_gdb_path)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|