OrgTJWater/TJWaterServer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add test for snapshot

Browse Source
This commit is contained in:
WQY\qiong
2022-10-15 00:00:41 +08:00
parent fa34d86b8a
commit 37bafed958
1 changed files with 45 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
test_tjnetwork.py
View File

@@ -1153,6 +1153,51 @@ class TestApi:
self.leave(p)
def test_snapshot(self):
p = "test_snapshot"
self.enter(p)
add_junction(p, ChangeSet({'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
add_junction(p, ChangeSet({'id': 'j2', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
add_junction(p, ChangeSet({'id': 'j3', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
add_junction(p, ChangeSet({'id': 'j4', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
client_op = get_current_operation(p)
assert client_op == 4
assert take_snapshot(p, 'x') == 4
execute_undo(p)
execute_undo(p)
add_junction(p, ChangeSet({'id': 'j5', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
add_junction(p, ChangeSet({'id': 'j6', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
assert take_snapshot(p, 'xx') == 6
cs = sync_with_server(p, client_op).operations
cs[0]['operation'] = API_DELETE
cs[0]['id'] = 'j4'
cs[1]['operation'] = API_DELETE
cs[1]['id'] = 'j3'
cs[2]['operation'] = API_ADD
cs[2]['id'] = 'j5'
cs[3]['operation'] = API_ADD
cs[3]['id'] = 'j6'
cs = pick_snapshot(p, 'x').operations
cs[0]['operation'] = 'delete'
cs[0]['id'] = 'j6'
cs[1]['operation'] = 'delete'
cs[1]['id'] = 'j5'
cs[2]['operation'] = 'add'
cs[2]['id'] = 'j3'
cs[3]['operation'] = 'add'
cs[3]['id'] = 'j4'
assert get_nodes(p) == ['j1', 'j2', 'j3', 'j4']
self.leave(p)
def test_batch_commands(self):
p = 'test_valve_op'
self.enter(p)