diff --git a/test_tjnetwork.py b/test_tjnetwork.py index c2c92c3..eb0eab7 100644 --- 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()