2023-05-27 15:47:41 +08:00
# coding=utf-8
import unittest
import inspect
2023-05-29 18:10:37 +08:00
from imp import reload
2023-05-27 15:47:41 +08:00
import arcpy
from functions import *
from sys_config import SysConfig
from logger import Logger
from sde_sync import SdeSynchronous
2023-05-29 18:10:37 +08:00
reload ( sys )
sys . setdefaultencoding ( ' utf8 ' )
2023-05-27 15:47:41 +08:00
class TestSdeSynchronous ( unittest . TestCase ) :
def setUp ( self ) :
2023-05-29 18:10:37 +08:00
# 获取当前路径的父级路径
work_pathname = os . path . abspath ( os . path . dirname ( os . getcwd ( ) ) )
2023-05-27 15:47:41 +08:00
# 同步开始时间
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 )
2023-05-29 18:10:37 +08:00
self . sde_db_sync . src_sde = self . sde_db_sync . get_connection ( str ( u " 源库SDE连接 " ) )
2023-05-27 15:47:41 +08:00
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 ( )
2024-02-20 13:54:44 +08:00
self . sde_db_sync . create_gdb_files ( )
2023-05-27 15:47:41 +08:00
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 " )
2024-02-20 13:54:44 +08:00
def test_copy_to_mdo ( self ) :
self . sde_db_sync . copy_to_mdo ( self . sde_db_sync . out_gdb_path )
2023-05-27 15:47:41 +08:00
if __name__ == ' __main__ ' :
unittest . main ( )