Add scada_model test

2023-02-10 00:47:53 +08:00
parent e18a2a529e
commit 14c57bb313
1 changed files with 193 additions and 0 deletions
--- a/test_tjnetwork.py
+++ b/test_tjnetwork.py
@@ -3650,5 +3650,198 @@ class TestApi:
        self.leave(p)


+    # 28 end
+
+
+    # 29 scada_model
+
+
+    def test_scada_model(self):
+        p = 'test_scada_model'
+        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_pipe(p, ChangeSet({'id': 'p1', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
+
+        sm = get_scada_model(p, 'sm')
+        assert sm == {}
+
+        add_scada_model(p, ChangeSet({'id': 'sm', 'x': 0.0, 'y': 10.0, 'device_id': 'sm_device'}))
+        sm = get_scada_model(p, 'sm')
+        assert sm['id'] == 'sm'
+        assert sm['x'] == 0.0
+        assert sm['y'] == 10.0
+        assert sm['device_id'] == 'sm_device'
+        assert sm['device_name'] == None
+        assert sm['address'] == None
+        assert sm['sm_type'] == None
+        assert sm['model_id'] == None
+
+        add_scada_model(p, ChangeSet({'id': 'sm0', 'x': 0.0, 'y': 10.0, 'device_id': 'sm_device_0', 'device_name': 'sm_device_name', 'address': 'xxx', 'sm_type': SCADA_TYPE_PRESSURE, 'model_id': 'j0'}))
+        sm0 = get_scada_model(p, 'sm0')
+        assert sm0 == {}
+
+        add_scada_model(p, ChangeSet({'id': 'sm0', 'x': 0.0, 'y': 10.0, 'device_id': 'sm_device_0', 'device_name': 'sm_device_name', 'address': 'xxx', 'sm_type': SCADA_TYPE_PRESSURE, 'model_id': 'p0'}))
+        sm0 = get_scada_model(p, 'sm0')
+        assert sm0 == {}
+
+        add_scada_model(p, ChangeSet({'id': 'sm1', 'x': 0.0, 'y': 10.0, 'device_id': 'sm_device_1', 'device_name': 'sm_device_name', 'address': 'xxx', 'sm_type': SCADA_TYPE_PRESSURE, 'model_id': 'j1'}))
+        sm1 = get_scada_model(p, 'sm1')
+        assert sm1['id'] == 'sm1'
+        assert sm1['x'] == 0.0
+        assert sm1['y'] == 10.0
+        assert sm1['device_id'] == 'sm_device_1'
+        assert sm1['device_name'] == 'sm_device_name'
+        assert sm1['address'] == 'xxx'
+        assert sm1['sm_type'] == SCADA_TYPE_PRESSURE
+        assert sm1['model_id'] == 'j1'
+
+        add_scada_model(p, ChangeSet({'id': 'sm2', 'x': 0.0, 'y': 10.0, 'device_id': 'sm_device_2', 'device_name': 'sm_device_name', 'address': 'xxx', 'sm_type': SCADA_TYPE_PRESSURE, 'model_id': 'p1'}))
+        sm2 = get_scada_model(p, 'sm2')
+        assert sm2['id'] == 'sm2'
+        assert sm2['x'] == 0.0
+        assert sm2['y'] == 10.0
+        assert sm2['device_id'] == 'sm_device_2'
+        assert sm2['device_name'] == 'sm_device_name'
+        assert sm2['address'] == 'xxx'
+        assert sm2['sm_type'] == SCADA_TYPE_PRESSURE
+        assert sm2['model_id'] == 'p1'
+
+        set_scada_model(p, ChangeSet({'id': 'sm', 'device_name': 'sm_device_name', 'address': 'xxx', 'sm_type': SCADA_TYPE_PRESSURE, 'model_id': 'j0'}))
+        sm = get_scada_model(p, 'sm')
+        assert sm['id'] == 'sm'
+        assert sm['x'] == 0.0
+        assert sm['y'] == 10.0
+        assert sm['device_id'] == 'sm_device'
+        assert sm['device_name'] == None
+        assert sm['address'] == None
+        assert sm['sm_type'] == None
+        assert sm['model_id'] == None
+
+        set_scada_model(p, ChangeSet({'id': 'sm', 'device_name': 'sm_device_name', 'address': 'xxx', 'sm_type': SCADA_TYPE_PRESSURE, 'model_id': 'j1'}))
+        sm = get_scada_model(p, 'sm')
+        assert sm['id'] == 'sm'
+        assert sm['x'] == 0.0
+        assert sm['y'] == 10.0
+        assert sm['device_id'] == 'sm_device'
+        assert sm['device_name'] == 'sm_device_name'
+        assert sm['address'] == 'xxx'
+        assert sm['sm_type'] == SCADA_TYPE_PRESSURE
+        assert sm['model_id'] == 'j1'
+
+        delete_scada_model(p, ChangeSet({'id': 'sm'}))
+        sm = get_scada_model(p, 'sm')
+        assert sm == {}
+
+        delete_scada_model(p, ChangeSet({'id': 'sm1'}))
+        sm1 = get_scada_model(p, 'sm1')
+        assert sm1 == {}
+
+        delete_scada_model(p, ChangeSet({'id': 'sm2'}))
+        sm2 = get_scada_model(p, 'sm2')
+        assert sm2 == {}
+
+        self.leave(p)
+
+
+    def test_scada_model_op(self):
+        p = 'test_scada_model_op'
+
+        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}))
+
+        cs = add_scada_model(p, ChangeSet({'id': 'sm', 'x': 0.0, 'y': 10.0, 'device_id': 'sm_device', 'device_name': 'sm_device_name', 'address': 'xxx', 'sm_type': SCADA_TYPE_PRESSURE, 'model_id': 'j1'})).operations[0]
+        assert cs['operation'] == API_ADD
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+        assert cs['x'] == 0.0
+        assert cs['y'] == 10.0
+        assert cs['device_id'] == 'sm_device'
+        assert cs['device_name'] == 'sm_device_name'
+        assert cs['address'] == 'xxx'
+        assert cs['sm_type'] == SCADA_TYPE_PRESSURE
+        assert cs['model_id'] == 'j1'
+
+        cs = execute_undo(p).operations[0]
+        assert cs['operation'] == API_DELETE
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+
+        cs = execute_redo(p).operations[0]
+        assert cs['operation'] == API_ADD
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+        assert cs['x'] == 0.0
+        assert cs['y'] == 10.0
+        assert cs['device_id'] == 'sm_device'
+        assert cs['device_name'] == 'sm_device_name'
+        assert cs['address'] == 'xxx'
+        assert cs['sm_type'] == SCADA_TYPE_PRESSURE
+        assert cs['model_id'] == 'j1'
+
+        cs = set_scada_model(p, ChangeSet({'id': 'sm', 'device_name': 'sm_device_name_', 'address': 'xxx_', 'sm_type': SCADA_TYPE_DEMAND, 'model_id': 'j2'})).operations[0]
+        assert cs['operation'] == API_UPDATE
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+        assert cs['x'] == 0.0
+        assert cs['y'] == 10.0
+        assert cs['device_id'] == 'sm_device'
+        assert cs['device_name'] == 'sm_device_name_'
+        assert cs['address'] == 'xxx_'
+        assert cs['sm_type'] == SCADA_TYPE_DEMAND
+        assert cs['model_id'] == 'j2'
+
+        cs = execute_undo(p).operations[0]
+        assert cs['operation'] == API_UPDATE
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+        assert cs['x'] == 0.0
+        assert cs['y'] == 10.0
+        assert cs['device_id'] == 'sm_device'
+        assert cs['device_name'] == 'sm_device_name'
+        assert cs['address'] == 'xxx'
+        assert cs['sm_type'] == SCADA_TYPE_PRESSURE
+        assert cs['model_id'] == 'j1'
+
+        cs = execute_redo(p).operations[0]
+        assert cs['operation'] == API_UPDATE
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+        assert cs['x'] == 0.0
+        assert cs['y'] == 10.0
+        assert cs['device_id'] == 'sm_device'
+        assert cs['device_name'] == 'sm_device_name_'
+        assert cs['address'] == 'xxx_'
+        assert cs['sm_type'] == SCADA_TYPE_DEMAND
+        assert cs['model_id'] == 'j2'
+
+        cs = delete_scada_model(p, ChangeSet({'id': 'sm'})).operations[0]
+        assert cs['operation'] == API_DELETE
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+
+        cs = execute_undo(p).operations[0]
+        assert cs['operation'] == API_ADD
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+        assert cs['x'] == 0.0
+        assert cs['y'] == 10.0
+        assert cs['device_id'] == 'sm_device'
+        assert cs['device_name'] == 'sm_device_name_'
+        assert cs['address'] == 'xxx_'
+        assert cs['sm_type'] == SCADA_TYPE_DEMAND
+        assert cs['model_id'] == 'j2'
+
+        cs = execute_redo(p).operations[0]
+        assert cs['operation'] == API_DELETE
+        assert cs['type'] == 'scada_model'
+        assert cs['id'] == 'sm'
+
+        self.leave(p)
+
+
 if __name__ == '__main__':
    pytest.main()