2002 lines
68 KiB
Python
2002 lines
68 KiB
Python
import pytest
|
|
from tjnetwork import *
|
|
|
|
class TestApi:
|
|
def enter(self, p):
|
|
if is_project_open(p):
|
|
close_project(p)
|
|
|
|
if have_project(p):
|
|
delete_project(p)
|
|
|
|
create_project(p)
|
|
open_project(p)
|
|
|
|
|
|
def leave(self, p):
|
|
close_project(p)
|
|
delete_project(p)
|
|
|
|
|
|
def test_project(self):
|
|
p = 'test_project'
|
|
|
|
assert not have_project(p)
|
|
assert not is_project_open(p)
|
|
assert get_project_open_count(p) == 0
|
|
|
|
create_project(p)
|
|
|
|
assert have_project(p)
|
|
assert not is_project_open(p)
|
|
assert get_project_open_count(p) == 0
|
|
|
|
open_project(p)
|
|
|
|
assert have_project(p)
|
|
assert is_project_open(p)
|
|
assert get_project_open_count(p) == 1
|
|
|
|
open_project(p)
|
|
|
|
assert have_project(p)
|
|
assert is_project_open(p)
|
|
assert get_project_open_count(p) == 2
|
|
|
|
close_project(p)
|
|
|
|
assert have_project(p)
|
|
assert is_project_open(p)
|
|
assert get_project_open_count(p) == 1
|
|
|
|
close_project(p)
|
|
|
|
assert have_project(p)
|
|
assert not is_project_open(p)
|
|
assert get_project_open_count(p) == 0
|
|
|
|
delete_project(p)
|
|
|
|
assert not have_project(p)
|
|
assert not is_project_open(p)
|
|
assert get_project_open_count(p) == 0
|
|
|
|
|
|
def test_title(self):
|
|
p = 'test_title'
|
|
self.enter(p)
|
|
|
|
assert get_title(p)['value'] == ''
|
|
|
|
change = set_title(p, ChangeSet({'value': 'title'})).operations[0]
|
|
assert change['operation'] == 'update'
|
|
assert change['type'] == 'title'
|
|
assert get_title(p)['value'] == 'title'
|
|
|
|
set_title(p, ChangeSet({'value': 'test'}))
|
|
assert get_title(p)['value'] == 'test'
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_title_op(self):
|
|
p = 'test_title_op'
|
|
self.enter(p)
|
|
|
|
assert get_title(p)['value'] == ''
|
|
|
|
cs = set_title(p, ChangeSet({'value': 'title'})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == 'title'
|
|
assert cs['value'] == 'title'
|
|
assert get_title(p)['value'] == 'title'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == 'title'
|
|
assert cs['value'] == ''
|
|
assert get_title(p)['value'] == ''
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == 'title'
|
|
assert cs['value'] == 'title'
|
|
assert get_title(p)['value'] == 'title'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == 'title'
|
|
assert cs['value'] == ''
|
|
assert get_title(p)['value'] == ''
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
assert get_title(p)['value'] == ''
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_junction(self):
|
|
p = 'test_junction'
|
|
self.enter(p)
|
|
|
|
add_junction(p, ChangeSet({'id': 'j0', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
|
|
j0 = get_junction(p, 'j0')
|
|
assert j0['x'] == 0.0
|
|
assert j0['y'] == 10.0
|
|
assert j0['elevation'] == 20.0
|
|
assert j0['demand'] == None
|
|
assert j0['pattern'] == None
|
|
assert j0['links'] == []
|
|
|
|
set_junction(p, ChangeSet({'id': 'j0', 'x': 100.0, 'y': 200.0}))
|
|
j0 = get_junction(p, 'j0')
|
|
assert j0['x'] == 100.0
|
|
assert j0['y'] == 200.0
|
|
|
|
set_junction(p, ChangeSet({'id': 'j0', 'elevation': 100.0}))
|
|
j0 = get_junction(p, 'j0')
|
|
assert j0['elevation'] == 100.0
|
|
|
|
set_junction(p, ChangeSet({'id': 'j0', 'demand': 100.0}))
|
|
j0 = get_junction(p, 'j0')
|
|
assert j0['demand'] == 100.0
|
|
|
|
# TODO: pattern
|
|
|
|
add_junction(p, ChangeSet({'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 2
|
|
assert nodes[0] == 'j0'
|
|
assert nodes[1] == 'j1'
|
|
assert is_junction(p, 'j0')
|
|
assert is_junction(p, 'j1')
|
|
|
|
delete_junction(p, ChangeSet({'id': 'j1'}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
assert nodes[0] == 'j0'
|
|
|
|
delete_junction(p, ChangeSet({'id': 'j0'}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 0
|
|
|
|
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_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 }))
|
|
add_pump(p, ChangeSet({'id': 'p2', 'node1': 'j1', 'node2': 'j2', 'power': 0.0}))
|
|
add_valve(p, ChangeSet({'id': 'v1', 'node1': 'j2', 'node2': 'j3', 'diameter': 10.0, 'v_type': VALVES_TYPE_FCV, 'setting': 0.1, 'minor_loss': 0.5 }))
|
|
assert get_junction(p, 'j1')['links'] == ['p1', 'p2']
|
|
assert get_junction(p, 'j2')['links'] == ['p1', 'p2', 'v1']
|
|
assert get_junction(p, 'j3')['links'] == ['v1']
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_junction_op(self):
|
|
p = 'test_junction_op'
|
|
self.enter(p)
|
|
|
|
cs = add_junction(p, ChangeSet({'id': 'j0', 'x': 0.0, 'y': 10.0, 'elevation': 20.0})).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['demand'] == None
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['demand'] == None
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 0
|
|
|
|
add_junction(p, ChangeSet({'id': 'j0', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'demand': 100.0}))
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
|
|
cs = delete_junction(p, ChangeSet({'id': 'j0'})).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['demand'] == 100.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['demand'] == 100.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
|
|
cs = set_junction(p, ChangeSet({'id': 'j0', 'x': 100.0, 'y': 200.0})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
assert cs['x'] == 100.0
|
|
assert cs['y'] == 200.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['demand'] == 100.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == JUNCTION
|
|
assert cs['id'] == 'j0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['demand'] == 100.0
|
|
assert cs['pattern'] == None
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_reservoir(self):
|
|
p = 'test_reservoir'
|
|
self.enter(p)
|
|
|
|
add_reservoir(p, ChangeSet({'id': 'r0', 'x': 0.0, 'y': 10.0, 'head': 20.0}))
|
|
r0 = get_reservoir(p, 'r0')
|
|
assert r0['x'] == 0.0
|
|
assert r0['y'] == 10.0
|
|
assert r0['head'] == 20.0
|
|
assert r0['pattern'] == None
|
|
assert r0['links'] == []
|
|
|
|
set_reservoir(p, ChangeSet({'id': 'r0', 'x': 100.0, 'y': 200.0}))
|
|
r0 = get_reservoir(p, 'r0')
|
|
assert r0['x'] == 100.0
|
|
assert r0['y'] == 200.0
|
|
|
|
set_reservoir(p, ChangeSet({'id': 'r0', 'head': 100.0}))
|
|
r0 = get_reservoir(p, 'r0')
|
|
assert r0['head'] == 100.0
|
|
|
|
# TODO: pattern
|
|
|
|
add_reservoir(p, ChangeSet({'id': 'r1', 'x': 0.0, 'y': 10.0, 'head': 20.0}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 2
|
|
assert nodes[0] == 'r0'
|
|
assert nodes[1] == 'r1'
|
|
assert is_reservoir(p, 'r0')
|
|
assert is_reservoir(p, 'r1')
|
|
|
|
delete_reservoir(p, ChangeSet({'id': 'r1'}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
assert nodes[0] == 'r0'
|
|
|
|
delete_reservoir(p, ChangeSet({'id': 'r0'}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 0
|
|
|
|
add_reservoir(p, ChangeSet({'id': 'r1', 'x': 0.0, 'y': 10.0, 'head': 20.0}))
|
|
add_reservoir(p, ChangeSet({'id': 'r2', 'x': 0.0, 'y': 10.0, 'head': 20.0}))
|
|
add_reservoir(p, ChangeSet({'id': 'r3', 'x': 0.0, 'y': 10.0, 'head': 20.0}))
|
|
add_pipe(p, ChangeSet({'id': 'p1', 'node1': 'r1', 'node2': 'r2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
add_pump(p, ChangeSet({'id': 'p2', 'node1': 'r1', 'node2': 'r2', 'power': 0.0}))
|
|
add_valve(p, ChangeSet({'id': 'v1', 'node1': 'r2', 'node2': 'r3', 'diameter': 10.0, 'v_type': VALVES_TYPE_FCV, 'setting': 0.1, 'minor_loss': 0.5 }))
|
|
assert get_reservoir(p, 'r1')['links'] == ['p1', 'p2']
|
|
assert get_reservoir(p, 'r2')['links'] == ['p1', 'p2', 'v1']
|
|
assert get_reservoir(p, 'r3')['links'] == ['v1']
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_reservoir_op(self):
|
|
p = 'test_reservoir_op'
|
|
self.enter(p)
|
|
|
|
cs = add_reservoir(p, ChangeSet({'id': 'r0', 'x': 0.0, 'y': 10.0, 'head': 20.0})).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['head'] == 20.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['head'] == 20.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 0
|
|
|
|
add_reservoir(p, ChangeSet({'id': 'r0', 'x': 0.0, 'y': 10.0, 'head': 20.0}))
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
|
|
cs = delete_reservoir(p, ChangeSet({'id': 'r0'})).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['head'] == 20.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['head'] == 20.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
|
|
cs = set_reservoir(p, ChangeSet({'id': 'r0', 'x': 100.0, 'y': 200.0})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
assert cs['x'] == 100.0
|
|
assert cs['y'] == 200.0
|
|
assert cs['head'] == 20.0
|
|
assert cs['pattern'] == None
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == RESERVOIR
|
|
assert cs['id'] == 'r0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['head'] == 20.0
|
|
assert cs['pattern'] == None
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_tank(self):
|
|
p = 'test_tank'
|
|
self.enter(p)
|
|
|
|
add_tank(p, ChangeSet({'id': 't0', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'init_level': 1.0, 'min_level': 0.0, 'max_level': 2.0, 'diameter': 10.0, 'min_vol': 100.0, 'vol_curve': None, 'overflow': OVERFLOW_NO}))
|
|
t0 = get_tank(p, 't0')
|
|
assert t0['x'] == 0.0
|
|
assert t0['y'] == 10.0
|
|
assert t0['elevation'] == 20.0
|
|
assert t0['init_level'] == 1.0
|
|
assert t0['min_level'] == 0.0
|
|
assert t0['max_level'] == 2.0
|
|
assert t0['diameter'] == 10.0
|
|
assert t0['min_vol'] == 100.0
|
|
assert t0['vol_curve'] == None
|
|
assert t0['overflow'] == OVERFLOW_NO
|
|
assert t0['links'] == []
|
|
|
|
set_tank(p, ChangeSet({'id': 't0', 'x': 100.0, 'y': 200.0}))
|
|
t0 = get_tank(p, 't0')
|
|
assert t0['x'] == 100.0
|
|
assert t0['y'] == 200.0
|
|
|
|
set_tank(p, ChangeSet({'id': 't0', 'elevation': 100.0}))
|
|
t0 = get_tank(p, 't0')
|
|
assert t0['elevation'] == 100.0
|
|
|
|
set_tank(p, ChangeSet({'id': 't0', 'init_level': 100.0, 'min_level': 50.0, 'max_level': 200.0}))
|
|
t0 = get_tank(p, 't0')
|
|
assert t0['init_level'] == 100.0
|
|
assert t0['min_level'] == 50.0
|
|
assert t0['max_level'] == 200.0
|
|
|
|
set_tank(p, ChangeSet({'id': 't0', 'diameter': 100.0}))
|
|
t0 = get_tank(p, 't0')
|
|
assert t0['diameter'] == 100.0
|
|
|
|
set_tank(p, ChangeSet({'id': 't0', 'min_vol': 200.0}))
|
|
t0 = get_tank(p, 't0')
|
|
assert t0['min_vol'] == 200.0
|
|
|
|
# TODO: vol_curve
|
|
|
|
set_tank(p, ChangeSet({'id': 't0', 'overflow': OVERFLOW_YES}))
|
|
t0 = get_tank(p, 't0')
|
|
assert t0['overflow'] == OVERFLOW_YES
|
|
|
|
add_tank(p, ChangeSet({'id': 't1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'init_level': 1.0, 'min_level': 0.0, 'max_level': 2.0, 'diameter': 10.0, 'min_vol': 100.0, 'vol_curve': None, 'overflow': OVERFLOW_NO}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 2
|
|
assert nodes[0] == 't0'
|
|
assert nodes[1] == 't1'
|
|
assert is_tank(p, 't0')
|
|
assert is_tank(p, 't1')
|
|
|
|
delete_tank(p, ChangeSet({'id': 't1'}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
assert nodes[0] == 't0'
|
|
|
|
delete_tank(p, ChangeSet({'id': 't0'}))
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 0
|
|
|
|
add_tank(p, ChangeSet({'id': 't1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'init_level': 1.0, 'min_level': 0.0, 'max_level': 2.0, 'diameter': 10.0, 'min_vol': 100.0, 'vol_curve': None, 'overflow': OVERFLOW_NO}))
|
|
add_tank(p, ChangeSet({'id': 't2', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'init_level': 1.0, 'min_level': 0.0, 'max_level': 2.0, 'diameter': 10.0, 'min_vol': 100.0, 'vol_curve': None, 'overflow': OVERFLOW_NO}))
|
|
add_tank(p, ChangeSet({'id': 't3', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'init_level': 1.0, 'min_level': 0.0, 'max_level': 2.0, 'diameter': 10.0, 'min_vol': 100.0, 'vol_curve': None, 'overflow': OVERFLOW_NO}))
|
|
add_pipe(p, ChangeSet({'id': 'p1', 'node1': 't1', 'node2': 't2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
add_pump(p, ChangeSet({'id': 'p2', 'node1': 't1', 'node2': 't2', 'power': 0.0}))
|
|
add_valve(p, ChangeSet({'id': 'v1', 'node1': 't2', 'node2': 't3', 'diameter': 10.0, 'v_type': VALVES_TYPE_FCV, 'setting': 0.1, 'minor_loss': 0.5 }))
|
|
assert get_tank(p, 't1')['links'] == ['p1', 'p2']
|
|
assert get_tank(p, 't2')['links'] == ['p1', 'p2', 'v1']
|
|
assert get_tank(p, 't3')['links'] == ['v1']
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_tank_op(self):
|
|
p = 'test_tank_op'
|
|
self.enter(p)
|
|
|
|
cs = add_tank(p, ChangeSet({'id': 't0', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'init_level': 1.0, 'min_level': 0.0, 'max_level': 2.0, 'diameter': 10.0, 'min_vol': 100.0, 'vol_curve': None, 'overflow': OVERFLOW_NO})).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['init_level'] == 1.0
|
|
assert cs['min_level'] == 0.0
|
|
assert cs['max_level'] == 2.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['min_vol'] == 100.0
|
|
assert cs['vol_curve'] == None
|
|
assert cs['overflow'] == OVERFLOW_NO
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['init_level'] == 1.0
|
|
assert cs['min_level'] == 0.0
|
|
assert cs['max_level'] == 2.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['min_vol'] == 100.0
|
|
assert cs['vol_curve'] == None
|
|
assert cs['overflow'] == OVERFLOW_NO
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 0
|
|
|
|
add_tank(p, ChangeSet({'id': 't0', 'x': 0.0, 'y': 10.0, 'elevation': 20.0, 'init_level': 1.0, 'min_level': 0.0, 'max_level': 2.0, 'diameter': 10.0, 'min_vol': 100.0, 'vol_curve': None, 'overflow': OVERFLOW_NO}))
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
|
|
cs = delete_tank(p, ChangeSet({'id': 't0'})).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['init_level'] == 1.0
|
|
assert cs['min_level'] == 0.0
|
|
assert cs['max_level'] == 2.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['min_vol'] == 100.0
|
|
assert cs['vol_curve'] == None
|
|
assert cs['overflow'] == OVERFLOW_NO
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['init_level'] == 1.0
|
|
assert cs['min_level'] == 0.0
|
|
assert cs['max_level'] == 2.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['min_vol'] == 100.0
|
|
assert cs['vol_curve'] == None
|
|
assert cs['overflow'] == OVERFLOW_NO
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
|
|
cs = set_tank(p, ChangeSet({'id': 't0', 'x': 100.0, 'y': 200.0})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
assert cs['x'] == 100.0
|
|
assert cs['y'] == 200.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['init_level'] == 1.0
|
|
assert cs['min_level'] == 0.0
|
|
assert cs['max_level'] == 2.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['min_vol'] == 100.0
|
|
assert cs['vol_curve'] == None
|
|
assert cs['overflow'] == OVERFLOW_NO
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == TANK
|
|
assert cs['id'] == 't0'
|
|
assert cs['x'] == 0.0
|
|
assert cs['y'] == 10.0
|
|
assert cs['elevation'] == 20.0
|
|
assert cs['init_level'] == 1.0
|
|
assert cs['min_level'] == 0.0
|
|
assert cs['max_level'] == 2.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['min_vol'] == 100.0
|
|
assert cs['vol_curve'] == None
|
|
assert cs['overflow'] == OVERFLOW_NO
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pipe(self):
|
|
p = 'test_pipe'
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
add_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
p0 = get_pipe(p, 'p0')
|
|
assert p0['node1'] == 'j1'
|
|
assert p0['node2'] == 'j2'
|
|
assert p0['length'] == 100.0
|
|
assert p0['diameter'] == 10.0
|
|
assert p0['roughness'] == 0.1
|
|
assert p0['minor_loss'] == 0.5
|
|
assert p0['status'] == PIPE_STATUS_OPEN
|
|
|
|
set_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j3', 'node2': 'j4'}))
|
|
p0 = get_pipe(p, 'p0')
|
|
assert p0['node1'] == 'j3'
|
|
assert p0['node2'] == 'j4'
|
|
|
|
set_pipe(p, ChangeSet({'id': 'p0', 'length': 200.0}))
|
|
p0 = get_pipe(p, 'p0')
|
|
assert p0['length'] == 200.0
|
|
|
|
set_pipe(p, ChangeSet({'id': 'p0', 'diameter': 100.0}))
|
|
p0 = get_pipe(p, 'p0')
|
|
assert p0['diameter'] == 100.0
|
|
|
|
set_pipe(p, ChangeSet({'id': 'p0', 'roughness': 0.2}))
|
|
p0 = get_pipe(p, 'p0')
|
|
assert p0['roughness'] == 0.2
|
|
|
|
set_pipe(p, ChangeSet({'id': 'p0', 'minor_loss': 0.1}))
|
|
p0 = get_pipe(p, 'p0')
|
|
assert p0['minor_loss'] == 0.1
|
|
|
|
set_pipe(p, ChangeSet({'id': 'p0', 'status': PIPE_STATUS_CLOSED}))
|
|
p0 = get_pipe(p, 'p0')
|
|
assert p0['status'] == PIPE_STATUS_CLOSED
|
|
|
|
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 }))
|
|
links = get_links(p)
|
|
assert len(links) == 2
|
|
assert links[0] == 'p0'
|
|
assert links[1] == 'p1'
|
|
assert is_pipe(p, 'p0')
|
|
assert is_pipe(p, 'p1')
|
|
|
|
delete_pipe(p, ChangeSet({'id': 'p1'}))
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
assert links[0] == 'p0'
|
|
|
|
delete_pipe(p, ChangeSet({'id': 'p0'}))
|
|
links = get_links(p)
|
|
assert len(links) == 0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pipe_op(self):
|
|
p = 'test_pipe_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}))
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
cs = add_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN })).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['length'] == 100.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['roughness'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
assert cs['status'] == PIPE_STATUS_OPEN
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['length'] == 100.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['roughness'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
assert cs['status'] == PIPE_STATUS_OPEN
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 0
|
|
|
|
add_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
|
|
cs = delete_pipe(p, ChangeSet({'id': 'p0'})).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['length'] == 100.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['roughness'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
assert cs['status'] == PIPE_STATUS_OPEN
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['length'] == 100.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['roughness'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
assert cs['status'] == PIPE_STATUS_OPEN
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
|
|
cs = set_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j3', 'node2': 'j4'})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j3'
|
|
assert cs['node2'] == 'j4'
|
|
assert cs['length'] == 100.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['roughness'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
assert cs['status'] == PIPE_STATUS_OPEN
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == PIPE
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['length'] == 100.0
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['roughness'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
assert cs['status'] == PIPE_STATUS_OPEN
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pump(self):
|
|
p = 'test_pump'
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
add_pump(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'power': 0.0}))
|
|
p0 = get_pump(p, 'p0')
|
|
assert p0['node1'] == 'j1'
|
|
assert p0['node2'] == 'j2'
|
|
assert p0['power'] == 0.0
|
|
assert p0['head'] == None
|
|
assert p0['speed'] == None
|
|
assert p0['pattern'] == None
|
|
|
|
set_pump(p, ChangeSet({'id': 'p0', 'node1': 'j3', 'node2': 'j4'}))
|
|
p0 = get_pump(p, 'p0')
|
|
assert p0['node1'] == 'j3'
|
|
assert p0['node2'] == 'j4'
|
|
|
|
set_pump(p, ChangeSet({'id': 'p0', 'power': 100.0}))
|
|
p0 = get_pump(p, 'p0')
|
|
assert p0['power'] == 100.0
|
|
|
|
add_pump(p, ChangeSet({'id': 'p1', 'node1': 'j1', 'node2': 'j2', 'power': 0.0}))
|
|
links = get_links(p)
|
|
assert len(links) == 2
|
|
assert links[0] == 'p0'
|
|
assert links[1] == 'p1'
|
|
assert is_pump(p, 'p0')
|
|
assert is_pump(p, 'p1')
|
|
|
|
delete_pump(p, ChangeSet({'id': 'p1'}))
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
assert links[0] == 'p0'
|
|
|
|
delete_pump(p, ChangeSet({'id': 'p0'}))
|
|
links = get_links(p)
|
|
assert len(links) == 0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pump_op(self):
|
|
p = 'test_pump_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}))
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
cs = add_pump(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'power': 0.0})).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['power'] == 0.0
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 0
|
|
|
|
add_pump(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'power': 0.0}))
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
|
|
cs = delete_pump(p, ChangeSet({'id': 'p0'})).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
|
|
cs = set_pump(p, ChangeSet({'id': 'p0', 'node1': 'j3', 'node2': 'j4'})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j3'
|
|
assert cs['node2'] == 'j4'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
|
|
cs = set_pump(p, ChangeSet({'id': 'p0', 'power': 100.0})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['power'] == 100.0
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == PUMP
|
|
assert cs['id'] == 'p0'
|
|
assert cs['power'] == 0.0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_valve(self):
|
|
p = 'test_valve'
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
add_valve(p, ChangeSet({'id': 'v0', 'node1': 'j1', 'node2': 'j2', 'diameter': 10.0, 'v_type': VALVES_TYPE_FCV, 'setting': 0.1, 'minor_loss': 0.5 }))
|
|
v0 = get_valve(p, 'v0')
|
|
assert v0['node1'] == 'j1'
|
|
assert v0['node2'] == 'j2'
|
|
assert v0['diameter'] == 10.0
|
|
assert v0['v_type'] == VALVES_TYPE_FCV
|
|
assert v0['setting'] == 0.1
|
|
assert v0['minor_loss'] == 0.5
|
|
|
|
set_valve(p, ChangeSet({'id': 'v0', 'node1': 'j3', 'node2': 'j4'}))
|
|
v0 = get_valve(p, 'v0')
|
|
assert v0['node1'] == 'j3'
|
|
assert v0['node2'] == 'j4'
|
|
|
|
set_valve(p, ChangeSet({'id': 'v0', 'diameter': 100.0}))
|
|
v0 = get_valve(p, 'v0')
|
|
assert v0['diameter'] == 100.0
|
|
|
|
set_valve(p, ChangeSet({'id': 'v0', 'v_type': VALVES_TYPE_GPV}))
|
|
v0 = get_valve(p, 'v0')
|
|
assert v0['v_type'] == VALVES_TYPE_GPV
|
|
|
|
set_valve(p, ChangeSet({'id': 'v0', 'setting': 0.2}))
|
|
v0 = get_valve(p, 'v0')
|
|
assert v0['setting'] == 0.2
|
|
|
|
set_valve(p, ChangeSet({'id': 'v0', 'minor_loss': 0.1}))
|
|
v0 = get_valve(p, 'v0')
|
|
assert v0['minor_loss'] == 0.1
|
|
|
|
add_valve(p, ChangeSet({'id': 'v1', 'node1': 'j1', 'node2': 'j2', 'diameter': 10.0, 'v_type': VALVES_TYPE_FCV, 'setting': 0.1, 'minor_loss': 0.5 }))
|
|
links = get_links(p)
|
|
assert len(links) == 2
|
|
assert links[0] == 'v0'
|
|
assert links[1] == 'v1'
|
|
assert is_valve(p, 'v0')
|
|
assert is_valve(p, 'v1')
|
|
|
|
delete_valve(p, ChangeSet({'id': 'v1'}))
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
assert links[0] == 'v0'
|
|
|
|
delete_valve(p, ChangeSet({'id': 'v0'}))
|
|
links = get_links(p)
|
|
assert len(links) == 0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_valve_op(self):
|
|
p = 'test_valve_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}))
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
cs = add_valve(p, ChangeSet({'id': 'v0', 'node1': 'j1', 'node2': 'j2', 'diameter': 10.0, 'v_type': VALVES_TYPE_FCV, 'setting': 0.1, 'minor_loss': 0.5 })).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['v_type'] == VALVES_TYPE_FCV
|
|
assert cs['setting'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['v_type'] == VALVES_TYPE_FCV
|
|
assert cs['setting'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 0
|
|
|
|
add_valve(p, ChangeSet({'id': 'v0', 'node1': 'j1', 'node2': 'j2', 'diameter': 10.0, 'v_type': VALVES_TYPE_FCV, 'setting': 0.1, 'minor_loss': 0.5 }))
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
|
|
cs = delete_valve(p, ChangeSet({'id': 'v0'})).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['v_type'] == VALVES_TYPE_FCV
|
|
assert cs['setting'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == 'delete'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
|
|
cs = execute_undo(p, True).operations[0]
|
|
assert cs['operation'] == 'add'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['v_type'] == VALVES_TYPE_FCV
|
|
assert cs['setting'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
|
|
cs = execute_redo(p)
|
|
assert len(cs.operations) == 0
|
|
|
|
links = get_links(p)
|
|
assert len(links) == 1
|
|
|
|
cs = set_valve(p, ChangeSet({'id': 'v0', 'node1': 'j3', 'node2': 'j4'})).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
assert cs['node1'] == 'j3'
|
|
assert cs['node2'] == 'j4'
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['v_type'] == VALVES_TYPE_FCV
|
|
assert cs['setting'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == 'update'
|
|
assert cs['type'] == VALVE
|
|
assert cs['id'] == 'v0'
|
|
assert cs['node1'] == 'j1'
|
|
assert cs['node2'] == 'j2'
|
|
assert cs['diameter'] == 10.0
|
|
assert cs['v_type'] == VALVES_TYPE_FCV
|
|
assert cs['setting'] == 0.1
|
|
assert cs['minor_loss'] == 0.5
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_tag(self):
|
|
p = 'test_tag'
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
|
|
add_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
assert is_pipe(p, 'p0')
|
|
|
|
t = get_tag(p, TAG_TYPE_NODE, 'j1')
|
|
assert t['tag'] == None
|
|
set_tag(p, ChangeSet({'t_type': TAG_TYPE_NODE, 'id': 'j1', 'tag': 'j1t' }))
|
|
t = get_tag(p, TAG_TYPE_NODE, 'j1')
|
|
assert t['tag'] == 'j1t'
|
|
|
|
t = get_tag(p, TAG_TYPE_NODE, 'j2')
|
|
assert t['tag'] == None
|
|
set_tag(p, ChangeSet({'t_type': TAG_TYPE_NODE, 'id': 'j2', 'tag': 'j2t' }))
|
|
t = get_tag(p, TAG_TYPE_NODE, 'j2')
|
|
assert t['tag'] == 'j2t'
|
|
|
|
t = get_tag(p, TAG_TYPE_LINK, 'p0')
|
|
assert t['tag'] == None
|
|
set_tag(p, ChangeSet({'t_type': TAG_TYPE_LINK, 'id': 'p0', 'tag': 'p0t' }))
|
|
t = get_tag(p, TAG_TYPE_LINK, 'p0')
|
|
assert t['tag'] == 'p0t'
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_tag_op(self):
|
|
p = 'test_tag_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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
|
|
add_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
assert is_pipe(p, 'p0')
|
|
|
|
cs = set_tag(p, ChangeSet({'t_type': TAG_TYPE_NODE, 'id': 'j1', 'tag': 'j1t' })).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'tag'
|
|
assert cs['t_type'] == TAG_TYPE_NODE
|
|
assert cs['id'] == 'j1'
|
|
assert cs['tag'] == 'j1t'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'tag'
|
|
assert cs['t_type'] == TAG_TYPE_NODE
|
|
assert cs['id'] == 'j1'
|
|
assert cs['tag'] == None
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'tag'
|
|
assert cs['t_type'] == TAG_TYPE_NODE
|
|
assert cs['id'] == 'j1'
|
|
assert cs['tag'] == 'j1t'
|
|
|
|
cs = set_tag(p, ChangeSet({'t_type': TAG_TYPE_LINK, 'id': 'p0', 'tag': 'p0t' })).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'tag'
|
|
assert cs['t_type'] == TAG_TYPE_LINK
|
|
assert cs['id'] == 'p0'
|
|
assert cs['tag'] == 'p0t'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'tag'
|
|
assert cs['t_type'] == TAG_TYPE_LINK
|
|
assert cs['id'] == 'p0'
|
|
assert cs['tag'] == None
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'tag'
|
|
assert cs['t_type'] == TAG_TYPE_LINK
|
|
assert cs['id'] == 'p0'
|
|
assert cs['tag'] == 'p0t'
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_control(self):
|
|
p = 'test_control'
|
|
self.enter(p)
|
|
|
|
assert get_control(p)['control'] == ''
|
|
|
|
set_control(p, ChangeSet({'control': 'x'}))
|
|
assert get_control(p)['control'] == 'x'
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_control_op(self):
|
|
p = 'test_control_op'
|
|
self.enter(p)
|
|
|
|
cs = set_control(p, ChangeSet({'control': 'x'})).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'control'
|
|
assert cs['control'] == 'x'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'control'
|
|
assert cs['control'] == ''
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'control'
|
|
assert cs['control'] == 'x'
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_demand(self):
|
|
p = 'test_demand'
|
|
self.enter(p)
|
|
|
|
add_junction(p, ChangeSet({'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
|
|
assert is_junction(p, 'j1')
|
|
|
|
d = get_demand(p, 'j1')
|
|
assert d['junction'] == 'j1'
|
|
assert d['demands'] == []
|
|
|
|
set_demand(p, ChangeSet({'junction': 'j1', 'demands': [{'demand': 10.0, 'pattern': None, 'category': 'x'},
|
|
{'demand': 20.0, 'pattern': None, 'category': None}]}))
|
|
|
|
d = get_demand(p, 'j1')
|
|
assert d['junction'] == 'j1'
|
|
ds = d['demands']
|
|
assert len(ds) == 2
|
|
assert ds[0]['demand'] == 10.0
|
|
assert ds[0]['pattern'] == None
|
|
assert ds[0]['category'] == 'x'
|
|
assert ds[1]['demand'] == 20.0
|
|
assert ds[1]['pattern'] == None
|
|
assert ds[1]['category'] == None
|
|
|
|
set_demand(p, ChangeSet({'junction': 'j1', 'demands': []}))
|
|
|
|
d = get_demand(p, 'j1')
|
|
assert d['junction'] == 'j1'
|
|
assert d['demands'] == []
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_demand_op(self):
|
|
p = 'test_demand_op'
|
|
self.enter(p)
|
|
|
|
add_junction(p, ChangeSet({'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
|
|
assert is_junction(p, 'j1')
|
|
|
|
cs = set_demand(p, ChangeSet({'junction': 'j1', 'demands': [{'demand': 10.0, 'pattern': None, 'category': 'x'},
|
|
{'demand': 20.0, 'pattern': None, 'category': None}]})).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'demand'
|
|
assert cs['junction'] == 'j1'
|
|
ds = cs['demands']
|
|
assert len(ds) == 2
|
|
assert ds[0]['demand'] == 10.0
|
|
assert ds[0]['pattern'] == None
|
|
assert ds[0]['category'] == 'x'
|
|
assert ds[1]['demand'] == 20.0
|
|
assert ds[1]['pattern'] == None
|
|
assert ds[1]['category'] == None
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'demand'
|
|
assert cs['junction'] == 'j1'
|
|
assert len(cs['demands']) == 0
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'demand'
|
|
assert cs['junction'] == 'j1'
|
|
ds = cs['demands']
|
|
assert len(ds) == 2
|
|
assert ds[0]['demand'] == 10.0
|
|
assert ds[0]['pattern'] == None
|
|
assert ds[0]['category'] == 'x'
|
|
assert ds[1]['demand'] == 20.0
|
|
assert ds[1]['pattern'] == None
|
|
assert ds[1]['category'] == None
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_status(self):
|
|
p = 'test_status'
|
|
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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
|
|
add_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
assert is_pipe(p, 'p0')
|
|
|
|
s = get_status(p, 'p0')
|
|
assert s['link'] == 'p0'
|
|
assert s['status'] == None
|
|
assert s['setting'] == None
|
|
|
|
set_status(p, ChangeSet({'link': 'p0', 'status': LINK_STATUS_OPEN, 'setting': 10.0}))
|
|
s = get_status(p, 'p0')
|
|
assert s['link'] == 'p0'
|
|
assert s['status'] == LINK_STATUS_OPEN
|
|
assert s['setting'] == 10.0
|
|
|
|
set_status(p, ChangeSet({'link': 'p0', 'status': None, 'setting': None}))
|
|
s = get_status(p, 'p0')
|
|
assert s['link'] == 'p0'
|
|
assert s['status'] == None
|
|
assert s['setting'] == None
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_status_op(self):
|
|
p = 'test_status_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}))
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
|
|
add_pipe(p, ChangeSet({'id': 'p0', 'node1': 'j1', 'node2': 'j2', 'length': 100.0, 'diameter': 10.0, 'roughness': 0.1, 'minor_loss': 0.5, 'status': PIPE_STATUS_OPEN }))
|
|
assert is_pipe(p, 'p0')
|
|
|
|
s = get_status(p, 'p0')
|
|
assert s['link'] == 'p0'
|
|
assert s['status'] == None
|
|
assert s['setting'] == None
|
|
|
|
cs = set_status(p, ChangeSet({'link': 'p0', 'status': LINK_STATUS_OPEN, 'setting': 10.0})).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'status'
|
|
assert cs['link'] == 'p0'
|
|
assert cs['status'] == LINK_STATUS_OPEN
|
|
assert cs['setting'] == 10.0
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'status'
|
|
assert cs['link'] == 'p0'
|
|
assert cs['status'] == None
|
|
assert cs['setting'] == None
|
|
|
|
s = get_status(p, 'p0')
|
|
assert s['link'] == 'p0'
|
|
assert s['status'] == None
|
|
assert s['setting'] == None
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'status'
|
|
assert cs['link'] == 'p0'
|
|
assert cs['status'] == LINK_STATUS_OPEN
|
|
assert cs['setting'] == 10.0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pattern(self):
|
|
p = 'test_pattern'
|
|
self.enter(p)
|
|
|
|
assert is_pattern(p, 'p0') == False
|
|
p0 = get_pattern(p, 'p0')
|
|
assert p0['id'] == 'p0'
|
|
assert p0['factors'] == []
|
|
|
|
set_pattern(p, ChangeSet({'id' : 'p0', 'factors': [1.0, 2.0, 3.0]}))
|
|
|
|
assert is_pattern(p, 'p0')
|
|
p0 = get_pattern(p, 'p0')
|
|
assert p0['id'] == 'p0'
|
|
assert p0['factors'] == [1.0, 2.0, 3.0]
|
|
|
|
set_pattern(p, ChangeSet({'id' : 'p0', 'factors': []}))
|
|
|
|
assert is_pattern(p, 'p0') == False
|
|
p0 = get_pattern(p, 'p0')
|
|
assert p0['id'] == 'p0'
|
|
assert p0['factors'] == []
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pattern_op(self):
|
|
p = 'test_pattern_op'
|
|
self.enter(p)
|
|
|
|
cs = set_pattern(p, ChangeSet({'id' : 'p0', 'factors': [1.0, 2.0, 3.0]})).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == PATTERN
|
|
assert cs['id'] == 'p0'
|
|
assert cs['factors'] == [1.0, 2.0, 3.0]
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == PATTERN
|
|
assert cs['id'] == 'p0'
|
|
assert cs['factors'] == []
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == PATTERN
|
|
assert cs['id'] == 'p0'
|
|
assert cs['factors'] == [1.0, 2.0, 3.0]
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_curve(self):
|
|
p = 'test_curve'
|
|
self.enter(p)
|
|
|
|
assert is_curve(p, 'c0') == False
|
|
c0 = get_curve(p, 'c0')
|
|
assert c0['id'] == 'c0'
|
|
assert c0['coords'] == []
|
|
|
|
set_curve(p, ChangeSet({'id' : 'c0', 'coords': [{'x': 1.0, 'y': 2.0}, {'x': 2.0, 'y': 1.0}]}))
|
|
|
|
assert is_curve(p, 'c0')
|
|
c0 = get_curve(p, 'c0')
|
|
assert c0['id'] == 'c0'
|
|
xys = c0['coords']
|
|
assert len(xys) == 2
|
|
assert xys[0]['x'] == 1.0
|
|
assert xys[0]['y'] == 2.0
|
|
assert xys[1]['x'] == 2.0
|
|
assert xys[1]['y'] == 1.0
|
|
|
|
set_curve(p, ChangeSet({'id' : 'c0', 'coords': []}))
|
|
|
|
assert is_curve(p, 'c0') == False
|
|
c0 = get_curve(p, 'c0')
|
|
assert c0['id'] == 'c0'
|
|
assert c0['coords'] == []
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_curve_op(self):
|
|
p = 'test_curve_op'
|
|
self.enter(p)
|
|
|
|
cs = set_curve(p, ChangeSet({'id' : 'c0', 'coords': [{'x': 1.0, 'y': 2.0}, {'x': 2.0, 'y': 1.0}]})).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == CURVE
|
|
assert cs['id'] == 'c0'
|
|
xys = cs['coords']
|
|
assert len(xys) == 2
|
|
assert xys[0]['x'] == 1.0
|
|
assert xys[0]['y'] == 2.0
|
|
assert xys[1]['x'] == 2.0
|
|
assert xys[1]['y'] == 1.0
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == CURVE
|
|
assert cs['id'] == 'c0'
|
|
assert cs['coords'] == []
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == CURVE
|
|
assert cs['id'] == 'c0'
|
|
xys = cs['coords']
|
|
assert len(xys) == 2
|
|
assert xys[0]['x'] == 1.0
|
|
assert xys[0]['y'] == 2.0
|
|
assert xys[1]['x'] == 2.0
|
|
assert xys[1]['y'] == 1.0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_emitter(self):
|
|
p = 'test_emitter'
|
|
self.enter(p)
|
|
|
|
add_junction(p, ChangeSet({'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
|
|
assert is_junction(p, 'j1')
|
|
|
|
e = get_emitter(p, 'j1')
|
|
assert e['junction'] == 'j1'
|
|
assert e['coefficient'] == None
|
|
|
|
set_emitter(p, ChangeSet({'junction': 'j1', 'coefficient': 10.0}))
|
|
|
|
e = get_emitter(p, 'j1')
|
|
assert e['junction'] == 'j1'
|
|
assert e['coefficient'] == 10.0
|
|
|
|
set_emitter(p, ChangeSet({'junction': 'j1', 'coefficient': None}))
|
|
|
|
e = get_emitter(p, 'j1')
|
|
assert e['junction'] == 'j1'
|
|
assert e['coefficient'] == None
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_emitter_op(self):
|
|
p = 'test_emitter_op'
|
|
self.enter(p)
|
|
|
|
add_junction(p, ChangeSet({'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}))
|
|
assert is_junction(p, 'j1')
|
|
|
|
cs = set_emitter(p, ChangeSet({'junction': 'j1', 'coefficient': 10.0})).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'emitter'
|
|
assert cs['junction'] == 'j1'
|
|
assert cs['coefficient'] == 10.0
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'emitter'
|
|
assert cs['junction'] == 'j1'
|
|
assert cs['coefficient'] == None
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'emitter'
|
|
assert cs['junction'] == 'j1'
|
|
assert cs['coefficient'] == 10.0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_time(self):
|
|
p = 'test_time'
|
|
self.enter(p)
|
|
|
|
t = get_time(p)
|
|
assert t['DURATION'] == '0:00'
|
|
assert t['HYDRAULIC TIMESTEP'] == '1:00'
|
|
assert t['QUALITY TIMESTEP'] == '0:05'
|
|
assert t['RULE TIMESTEP'] == '0:05'
|
|
assert t['PATTERN TIMESTEP'] == '1:00'
|
|
assert t['PATTERN START'] == '0:00'
|
|
assert t['REPORT TIMESTEP'] == '1:00'
|
|
assert t['REPORT START'] == '0:00'
|
|
assert t['START CLOCKTIME'] == '12:00 AM'
|
|
assert t['STATISTIC'] == TIME_STATISTIC_NONE
|
|
|
|
t['STATISTIC'] = TIME_STATISTIC_AVERAGED
|
|
set_time(p, ChangeSet(t))
|
|
|
|
t = get_time(p)
|
|
assert t['DURATION'] == '0:00'
|
|
assert t['HYDRAULIC TIMESTEP'] == '1:00'
|
|
assert t['QUALITY TIMESTEP'] == '0:05'
|
|
assert t['RULE TIMESTEP'] == '0:05'
|
|
assert t['PATTERN TIMESTEP'] == '1:00'
|
|
assert t['PATTERN START'] == '0:00'
|
|
assert t['REPORT TIMESTEP'] == '1:00'
|
|
assert t['REPORT START'] == '0:00'
|
|
assert t['START CLOCKTIME'] == '12:00 AM'
|
|
assert t['STATISTIC'] == TIME_STATISTIC_AVERAGED
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_time_op(self):
|
|
p = 'test_time_op'
|
|
self.enter(p)
|
|
|
|
t = get_time(p)
|
|
assert t['DURATION'] == '0:00'
|
|
assert t['HYDRAULIC TIMESTEP'] == '1:00'
|
|
assert t['QUALITY TIMESTEP'] == '0:05'
|
|
assert t['RULE TIMESTEP'] == '0:05'
|
|
assert t['PATTERN TIMESTEP'] == '1:00'
|
|
assert t['PATTERN START'] == '0:00'
|
|
assert t['REPORT TIMESTEP'] == '1:00'
|
|
assert t['REPORT START'] == '0:00'
|
|
assert t['START CLOCKTIME'] == '12:00 AM'
|
|
assert t['STATISTIC'] == TIME_STATISTIC_NONE
|
|
|
|
t['STATISTIC'] = TIME_STATISTIC_AVERAGED
|
|
cs = set_time(p, ChangeSet(t)).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'time'
|
|
assert cs['DURATION'] == '0:00'
|
|
assert cs['HYDRAULIC TIMESTEP'] == '1:00'
|
|
assert cs['QUALITY TIMESTEP'] == '0:05'
|
|
assert cs['RULE TIMESTEP'] == '0:05'
|
|
assert cs['PATTERN TIMESTEP'] == '1:00'
|
|
assert cs['PATTERN START'] == '0:00'
|
|
assert cs['REPORT TIMESTEP'] == '1:00'
|
|
assert cs['REPORT START'] == '0:00'
|
|
assert cs['START CLOCKTIME'] == '12:00 AM'
|
|
assert cs['STATISTIC'] == TIME_STATISTIC_AVERAGED
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'time'
|
|
assert cs['DURATION'] == '0:00'
|
|
assert cs['HYDRAULIC TIMESTEP'] == '1:00'
|
|
assert cs['QUALITY TIMESTEP'] == '0:05'
|
|
assert cs['RULE TIMESTEP'] == '0:05'
|
|
assert cs['PATTERN TIMESTEP'] == '1:00'
|
|
assert cs['PATTERN START'] == '0:00'
|
|
assert cs['REPORT TIMESTEP'] == '1:00'
|
|
assert cs['REPORT START'] == '0:00'
|
|
assert cs['START CLOCKTIME'] == '12:00 AM'
|
|
assert cs['STATISTIC'] == TIME_STATISTIC_NONE
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'time'
|
|
assert cs['DURATION'] == '0:00'
|
|
assert cs['HYDRAULIC TIMESTEP'] == '1:00'
|
|
assert cs['QUALITY TIMESTEP'] == '0:05'
|
|
assert cs['RULE TIMESTEP'] == '0:05'
|
|
assert cs['PATTERN TIMESTEP'] == '1:00'
|
|
assert cs['PATTERN START'] == '0:00'
|
|
assert cs['REPORT TIMESTEP'] == '1:00'
|
|
assert cs['REPORT START'] == '0:00'
|
|
assert cs['START CLOCKTIME'] == '12:00 AM'
|
|
assert cs['STATISTIC'] == TIME_STATISTIC_AVERAGED
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_option(self):
|
|
p = 'test_option'
|
|
self.enter(p)
|
|
|
|
o = get_option(p)
|
|
assert o['UNITS'] == OPTION_UNITS_GPM
|
|
assert o['HEADLOSS'] == OPTION_HEADLOSS_HW
|
|
assert o['VISCOSITY'] == '1.0'
|
|
assert o['SPECIFIC GRAVITY'] == '1.0'
|
|
assert o['TRIALS'] == '40'
|
|
assert o['ACCURACY'] == '0.001'
|
|
assert o['FLOWCHANGE'] == '0'
|
|
assert o['HEADERROR'] == '0'
|
|
assert o['CHECKFREQ'] == '2'
|
|
assert o['MAXCHECK'] == '10'
|
|
assert o['DAMPLIMIT'] == '0'
|
|
assert o['UNBALANCED'] == OPTION_UNBALANCED_STOP
|
|
assert o['DEMAND MODEL'] == OPTION_DEMAND_MODEL_DDA
|
|
assert o['MINIMUM PRESSURE'] == '0'
|
|
assert o['REQUIRED PRESSURE'] == '0.1'
|
|
assert o['PRESSURE EXPONENT'] == '0.5'
|
|
assert o['PATTERN'] == '1'
|
|
assert o['DEMAND MULTIPLIER'] == '1.0'
|
|
assert o['EMITTER EXPONENT'] == '0.5'
|
|
assert o['QUALITY'] == OPTION_QUALITY_NONE
|
|
assert o['DIFFUSIVITY'] == '1.0'
|
|
assert o['TOLERANCE'] == '0.01'
|
|
|
|
o['UNITS'] = OPTION_UNITS_LPS
|
|
set_option(p, ChangeSet(o))
|
|
|
|
o = get_option(p)
|
|
assert o['UNITS'] == OPTION_UNITS_LPS
|
|
assert o['HEADLOSS'] == OPTION_HEADLOSS_HW
|
|
assert o['VISCOSITY'] == '1.0'
|
|
assert o['SPECIFIC GRAVITY'] == '1.0'
|
|
assert o['TRIALS'] == '40'
|
|
assert o['ACCURACY'] == '0.001'
|
|
assert o['FLOWCHANGE'] == '0'
|
|
assert o['HEADERROR'] == '0'
|
|
assert o['CHECKFREQ'] == '2'
|
|
assert o['MAXCHECK'] == '10'
|
|
assert o['DAMPLIMIT'] == '0'
|
|
assert o['UNBALANCED'] == OPTION_UNBALANCED_STOP
|
|
assert o['DEMAND MODEL'] == OPTION_DEMAND_MODEL_DDA
|
|
assert o['MINIMUM PRESSURE'] == '0'
|
|
assert o['REQUIRED PRESSURE'] == '0.1'
|
|
assert o['PRESSURE EXPONENT'] == '0.5'
|
|
assert o['PATTERN'] == '1'
|
|
assert o['DEMAND MULTIPLIER'] == '1.0'
|
|
assert o['EMITTER EXPONENT'] == '0.5'
|
|
assert o['QUALITY'] == OPTION_QUALITY_NONE
|
|
assert o['DIFFUSIVITY'] == '1.0'
|
|
assert o['TOLERANCE'] == '0.01'
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_option_op(self):
|
|
p = 'test_option_op'
|
|
self.enter(p)
|
|
|
|
o = get_option(p)
|
|
assert o['UNITS'] == OPTION_UNITS_GPM
|
|
assert o['HEADLOSS'] == OPTION_HEADLOSS_HW
|
|
assert o['VISCOSITY'] == '1.0'
|
|
assert o['SPECIFIC GRAVITY'] == '1.0'
|
|
assert o['TRIALS'] == '40'
|
|
assert o['ACCURACY'] == '0.001'
|
|
assert o['FLOWCHANGE'] == '0'
|
|
assert o['HEADERROR'] == '0'
|
|
assert o['CHECKFREQ'] == '2'
|
|
assert o['MAXCHECK'] == '10'
|
|
assert o['DAMPLIMIT'] == '0'
|
|
assert o['UNBALANCED'] == OPTION_UNBALANCED_STOP
|
|
assert o['DEMAND MODEL'] == OPTION_DEMAND_MODEL_DDA
|
|
assert o['MINIMUM PRESSURE'] == '0'
|
|
assert o['REQUIRED PRESSURE'] == '0.1'
|
|
assert o['PRESSURE EXPONENT'] == '0.5'
|
|
assert o['PATTERN'] == '1'
|
|
assert o['DEMAND MULTIPLIER'] == '1.0'
|
|
assert o['EMITTER EXPONENT'] == '0.5'
|
|
assert o['QUALITY'] == OPTION_QUALITY_NONE
|
|
assert o['DIFFUSIVITY'] == '1.0'
|
|
assert o['TOLERANCE'] == '0.01'
|
|
|
|
o['UNITS'] = OPTION_UNITS_LPS
|
|
cs = set_option(p, ChangeSet(o)).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'option'
|
|
assert cs['UNITS'] == OPTION_UNITS_LPS
|
|
assert cs['HEADLOSS'] == OPTION_HEADLOSS_HW
|
|
assert cs['VISCOSITY'] == '1.0'
|
|
assert cs['SPECIFIC GRAVITY'] == '1.0'
|
|
assert cs['TRIALS'] == '40'
|
|
assert cs['ACCURACY'] == '0.001'
|
|
assert cs['FLOWCHANGE'] == '0'
|
|
assert cs['HEADERROR'] == '0'
|
|
assert cs['CHECKFREQ'] == '2'
|
|
assert cs['MAXCHECK'] == '10'
|
|
assert cs['DAMPLIMIT'] == '0'
|
|
assert cs['UNBALANCED'] == OPTION_UNBALANCED_STOP
|
|
assert cs['DEMAND MODEL'] == OPTION_DEMAND_MODEL_DDA
|
|
assert cs['MINIMUM PRESSURE'] == '0'
|
|
assert cs['REQUIRED PRESSURE'] == '0.1'
|
|
assert cs['PRESSURE EXPONENT'] == '0.5'
|
|
assert cs['PATTERN'] == '1'
|
|
assert cs['DEMAND MULTIPLIER'] == '1.0'
|
|
assert cs['EMITTER EXPONENT'] == '0.5'
|
|
assert cs['QUALITY'] == OPTION_QUALITY_NONE
|
|
assert cs['DIFFUSIVITY'] == '1.0'
|
|
assert cs['TOLERANCE'] == '0.01'
|
|
|
|
cs = execute_undo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'option'
|
|
assert cs['UNITS'] == OPTION_UNITS_GPM
|
|
assert cs['HEADLOSS'] == OPTION_HEADLOSS_HW
|
|
assert cs['VISCOSITY'] == '1.0'
|
|
assert cs['SPECIFIC GRAVITY'] == '1.0'
|
|
assert cs['TRIALS'] == '40'
|
|
assert cs['ACCURACY'] == '0.001'
|
|
assert cs['FLOWCHANGE'] == '0'
|
|
assert cs['HEADERROR'] == '0'
|
|
assert cs['CHECKFREQ'] == '2'
|
|
assert cs['MAXCHECK'] == '10'
|
|
assert cs['DAMPLIMIT'] == '0'
|
|
assert cs['UNBALANCED'] == OPTION_UNBALANCED_STOP
|
|
assert cs['DEMAND MODEL'] == OPTION_DEMAND_MODEL_DDA
|
|
assert cs['MINIMUM PRESSURE'] == '0'
|
|
assert cs['REQUIRED PRESSURE'] == '0.1'
|
|
assert cs['PRESSURE EXPONENT'] == '0.5'
|
|
assert cs['PATTERN'] == '1'
|
|
assert cs['DEMAND MULTIPLIER'] == '1.0'
|
|
assert cs['EMITTER EXPONENT'] == '0.5'
|
|
assert cs['QUALITY'] == OPTION_QUALITY_NONE
|
|
assert cs['DIFFUSIVITY'] == '1.0'
|
|
assert cs['TOLERANCE'] == '0.01'
|
|
|
|
cs = execute_redo(p).operations[0]
|
|
assert cs['operation'] == API_UPDATE
|
|
assert cs['type'] == 'option'
|
|
assert cs['UNITS'] == OPTION_UNITS_LPS
|
|
assert cs['HEADLOSS'] == OPTION_HEADLOSS_HW
|
|
assert cs['VISCOSITY'] == '1.0'
|
|
assert cs['SPECIFIC GRAVITY'] == '1.0'
|
|
assert cs['TRIALS'] == '40'
|
|
assert cs['ACCURACY'] == '0.001'
|
|
assert cs['FLOWCHANGE'] == '0'
|
|
assert cs['HEADERROR'] == '0'
|
|
assert cs['CHECKFREQ'] == '2'
|
|
assert cs['MAXCHECK'] == '10'
|
|
assert cs['DAMPLIMIT'] == '0'
|
|
assert cs['UNBALANCED'] == OPTION_UNBALANCED_STOP
|
|
assert cs['DEMAND MODEL'] == OPTION_DEMAND_MODEL_DDA
|
|
assert cs['MINIMUM PRESSURE'] == '0'
|
|
assert cs['REQUIRED PRESSURE'] == '0.1'
|
|
assert cs['PRESSURE EXPONENT'] == '0.5'
|
|
assert cs['PATTERN'] == '1'
|
|
assert cs['DEMAND MULTIPLIER'] == '1.0'
|
|
assert cs['EMITTER EXPONENT'] == '0.5'
|
|
assert cs['QUALITY'] == OPTION_QUALITY_NONE
|
|
assert cs['DIFFUSIVITY'] == '1.0'
|
|
assert cs['TOLERANCE'] == '0.01'
|
|
|
|
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_batch_commands'
|
|
self.enter(p)
|
|
|
|
cs = ChangeSet()
|
|
cs.add({'type': JUNCTION, 'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0})
|
|
cs.add({'type': JUNCTION, 'id': 'j2', 'x': 0.0, 'y': 10.0, 'elevation': 20.0})
|
|
cs.add({'type': JUNCTION, 'id': 'j2', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}) # fail
|
|
|
|
cs = execute_batch_commands(p, cs)
|
|
assert len(cs.operations) == 2
|
|
|
|
cs = ChangeSet()
|
|
cs.delete({'type': JUNCTION, 'id': 'j1'})
|
|
cs.delete({'type': JUNCTION, 'id': 'j2'})
|
|
|
|
cs = execute_batch_commands(p, cs)
|
|
assert len(cs.operations) == 2
|
|
|
|
cs = execute_undo(p)
|
|
assert len(cs.operations) == 1
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_batch_command(self):
|
|
p = 'test_batch_command'
|
|
self.enter(p)
|
|
|
|
cs = ChangeSet()
|
|
cs.add({'type': JUNCTION, 'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0})
|
|
cs.add({'type': JUNCTION, 'id': 'j2', 'x': 0.0, 'y': 10.0, 'elevation': 20.0})
|
|
cs.add({'type': JUNCTION, 'id': 'j2', 'x': 0.0, 'y': 10.0, 'elevation': 20.0}) # fail
|
|
|
|
cs = execute_batch_command(p, cs)
|
|
assert len(cs.operations) == 0
|
|
|
|
assert get_current_operation(p) == 0
|
|
|
|
cs = ChangeSet()
|
|
cs.add({'type': JUNCTION, 'id': 'j1', 'x': 0.0, 'y': 10.0, 'elevation': 20.0})
|
|
cs.add({'type': JUNCTION, 'id': 'j2', 'x': 0.0, 'y': 10.0, 'elevation': 20.0})
|
|
|
|
cs = execute_batch_command(p, cs)
|
|
|
|
assert get_current_operation(p) == 1
|
|
|
|
cs = ChangeSet()
|
|
cs.delete({'type': JUNCTION, 'id': 'j1'})
|
|
cs.delete({'type': JUNCTION, 'id': 'j2'})
|
|
|
|
cs = execute_batch_command(p, cs)
|
|
|
|
assert get_current_operation(p) == 2
|
|
|
|
cs = execute_undo(p)
|
|
assert get_current_operation(p) == 1
|
|
|
|
cs = execute_undo(p)
|
|
assert get_current_operation(p) == 0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main()
|