687 lines
26 KiB
Python
687 lines
26 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) == ""
|
|
|
|
change = set_title(p, "title").operations[0]
|
|
assert change['operation'] == 'update'
|
|
assert change['type'] == 'title'
|
|
assert change['id'] == ''
|
|
assert get_title(p) == "title"
|
|
|
|
set_title(p, "test")
|
|
assert get_title(p) == "test"
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_junction(self):
|
|
p = "test_junction"
|
|
self.enter(p)
|
|
|
|
change_set = add_junction(p, "j0", 0.0, 10.0, 20.0)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == JUNCTION
|
|
assert change_set.operations[0]['id'] == "j0"
|
|
coord = get_junction(p, 'j0')['coord']
|
|
assert coord['x'] == 0.0
|
|
assert coord['y'] == 10.0
|
|
z = get_junction(p, 'j0')['elevation']
|
|
assert z == 20.0
|
|
|
|
assert get_junction(p, 'j') == None
|
|
assert get_junction(p, 'j0')['demand'] == None
|
|
assert get_junction(p, 'j0')['demand'] == None
|
|
change_set = set_junction(p, 'j0', {'demand': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == JUNCTION
|
|
assert change_set.operations[0]['id'] == 'j0'
|
|
assert change_set.operations[0]['properties'] == ['demand']
|
|
assert get_junction(p, 'j0')['demand'] == 100.0
|
|
|
|
assert get_junction(p, 'j0')['coord']['x'] == 0.0
|
|
assert get_junction(p, 'j0')['coord']['y'] == 10.0
|
|
change_set = set_junction(p, 'j0', {'coord': {'x': 100.0, 'y': 200.0}})
|
|
assert get_junction(p, 'j0')['coord']['x'] == 100.0
|
|
assert get_junction(p, 'j0')['coord']['y'] == 200.0
|
|
|
|
# TODO: pattern
|
|
|
|
change_set = add_junction(p, "j1", 0.0, 10.0, 20.0)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == JUNCTION
|
|
assert change_set.operations[0]['id'] == "j1"
|
|
|
|
change_set = add_junction(p, "j1", 0.0, 10.0, 20.0)
|
|
assert len(change_set.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 2
|
|
assert nodes[0] == "j0"
|
|
assert nodes[1] == "j1"
|
|
|
|
change_set = delete_junction(p, 'j1')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == JUNCTION
|
|
assert change_set.operations[0]['id'] == "j1"
|
|
|
|
change_set = delete_junction(p, 'j1')
|
|
assert len(change_set.operations) == 0
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 1
|
|
assert nodes[0] == "j0"
|
|
|
|
change_set = delete_junction(p, 'j0')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == JUNCTION
|
|
assert change_set.operations[0]['id'] == "j0"
|
|
|
|
nodes = get_nodes(p)
|
|
assert len(nodes) == 0
|
|
|
|
add_junction(p, 'j1', 0.0, 0.0, 0.0)
|
|
add_junction(p, 'j2', 10.0, 10.0, 10.0)
|
|
add_junction(p, 'j3', 10.0, 10.0, 10.0)
|
|
add_pipe(p, 'p1', 'j1', 'j2')
|
|
add_pump(p, 'p2', 'j1', 'j2')
|
|
add_valve(p, 'v1', 'j2', 'j3')
|
|
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_reservoir(self):
|
|
p = "test_reservoir"
|
|
self.enter(p)
|
|
|
|
change_set = add_reservoir(p, "r0", 0.0, 10.0, 20.0)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == RESERVOIR
|
|
assert change_set.operations[0]['id'] == "r0"
|
|
|
|
change_set = add_reservoir(p, "r0", 0.0, 10.0, 20.0)
|
|
assert len(change_set.operations) == 0
|
|
|
|
assert get_reservoir(p, 'r0')['head'] == 20.0
|
|
assert get_reservoir(p, 'r0')['head'] == 20.0
|
|
change_set = set_reservoir(p, 'r0', {'head': 100.0})
|
|
assert change_set.operations[0]['type'] == RESERVOIR
|
|
assert change_set.operations[0]['id'] == 'r0'
|
|
assert change_set.operations[0]['properties'] == ['head']
|
|
assert get_reservoir(p, 'r0')['head'] == 100.0
|
|
|
|
assert get_reservoir(p, 'r0')['coord'] == { 'x': 0.0, 'y': 10.0 }
|
|
assert get_reservoir(p, 'r0')['coord'] == { 'x': 0.0, 'y': 10.0 }
|
|
change_set = set_reservoir(p, 'r0', {'coord': {'x': 100.0, 'y': 200.0}})
|
|
assert change_set.operations[0]['type'] == RESERVOIR
|
|
assert change_set.operations[0]['id'] == 'r0'
|
|
assert change_set.operations[0]['properties'] == ['coord']
|
|
assert get_reservoir(p, 'r0')['coord'] == { 'x': 100.0, 'y': 200.0 }
|
|
|
|
# TODO: pattern
|
|
|
|
change_set = delete_reservoir(p, 'r0')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == RESERVOIR
|
|
assert change_set.operations[0]['id'] == "r0"
|
|
|
|
change_set = delete_reservoir(p, 'r0')
|
|
assert len(change_set.operations) == 0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_tank(self):
|
|
p = "test_tank"
|
|
|
|
self.enter(p)
|
|
|
|
change_set = add_tank(p, "t0", 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
|
|
change_set = add_tank(p, "t1", 0.0, 10.0, 20.0)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t1"
|
|
|
|
change_set = add_tank(p, "t1", 0.0, 10.0, 20.0)
|
|
assert len(change_set.operations) == 0
|
|
|
|
tanks = get_nodes(p)
|
|
assert len(tanks) == 2
|
|
assert tanks[0] == 't0'
|
|
assert tanks[1] == 't1'
|
|
|
|
assert get_tank(p, 't0')['elevation'] == 10.0
|
|
change_set = set_tank(p, 't0', {'elevation': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
assert change_set.operations[0]['properties'] == ['elevation']
|
|
assert get_tank(p, 't0')['elevation'] == 100.0
|
|
|
|
assert get_tank(p, 't0')['init_level'] == 10.0
|
|
change_set = set_tank(p, 't0', {'init_level': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
assert change_set.operations[0]['properties'] == ['init_level']
|
|
assert get_tank(p, 't0')['init_level'] == 100.0
|
|
|
|
assert get_tank(p, 't0')['min_level'] == 10.0
|
|
change_set = set_tank(p, 't0', {'min_level': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
assert change_set.operations[0]['properties'] == ['min_level']
|
|
assert get_tank(p, 't0')['min_level'] == 100.0
|
|
|
|
assert get_tank(p, 't0')['max_level'] == 10.0
|
|
change_set = set_tank(p, 't0', {'max_level': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
assert change_set.operations[0]['properties'] == ['max_level']
|
|
assert get_tank(p, 't0')['max_level'] == 100.0
|
|
|
|
assert get_tank(p, 't0')['diameter'] == 10.0
|
|
change_set = set_tank(p, 't0', {'diameter': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
assert change_set.operations[0]['properties'] == ['diameter']
|
|
assert get_tank(p, 't0')['diameter'] == 100.0
|
|
|
|
assert get_tank(p, 't0')['min_vol'] == 10.0
|
|
change_set = set_tank(p, 't0', {'min_vol': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
assert change_set.operations[0]['properties'] == ['min_vol']
|
|
assert get_tank(p, 't0')['min_vol'] == 100.0
|
|
|
|
# TODO: vol_curve
|
|
|
|
assert get_tank(p, 't') == None
|
|
assert get_tank(p, 't0')['overflow'] == None
|
|
change_set = set_tank(p, 't0', {'overflow': "XXX"})
|
|
assert len(change_set.operations) == 0
|
|
change_set = set_tank(p, 't0', {'overflow': OVERFLOW_YES})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
assert change_set.operations[0]['properties'] == ['overflow']
|
|
assert get_tank(p, 't0')['overflow'] == OVERFLOW_YES
|
|
|
|
change_set = delete_tank(p, "t0")
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t0"
|
|
|
|
tanks = get_nodes(p)
|
|
assert len(tanks) == 1
|
|
|
|
change_set = delete_tank(p, "t1")
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == TANK
|
|
assert change_set.operations[0]['id'] == "t1"
|
|
|
|
change_set = delete_tank(p, "t1")
|
|
assert len(change_set.operations) == 0
|
|
|
|
tanks = get_nodes(p)
|
|
assert len(tanks) == 0
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pipe(self):
|
|
p = "test_pipe"
|
|
self.enter(p)
|
|
|
|
change_set = add_pipe(p, 'p1', 'j1', 'j2', 10.0, 10.0, 10.0, 10.0, PIPE_STATUS_CLOSED)
|
|
assert len(change_set.operations) == 0
|
|
|
|
add_junction(p, 'j1', 0.0, 0.0, 0.0)
|
|
add_junction(p, 'j2', 10.0, 10.0, 0.0)
|
|
add_junction(p, 'j3', 0.0, 0.0, 0.0)
|
|
add_junction(p, 'j4', 10.0, 10.0, 0.0)
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
assert not is_link(p, 'p1')
|
|
|
|
change_set = add_pipe(p, 'p1', 'j1', 'j2', 10.0, 10.0, 10.0, 10.0, "XXX")
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = add_pipe(p, 'p1', 'j1', 'j1', 10.0, 10.0, 10.0, 10.0, PIPE_STATUS_CLOSED)
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = add_pipe(p, 'p1', 'j1', 'j2', 10.0, 10.0, 10.0, 10.0, PIPE_STATUS_CLOSED)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == "p1"
|
|
|
|
change_set = add_pipe(p, 'p1', 'j1', 'j2', 10.0, 10.0, 10.0, 10.0, PIPE_STATUS_CLOSED)
|
|
assert len(change_set.operations) == 0
|
|
|
|
pipes = get_links(p)
|
|
assert len(pipes) == 1
|
|
|
|
assert get_pipe(p, 'p1')['node1'] == 'j1'
|
|
assert get_pipe(p, 'p1')['node2'] == 'j2'
|
|
assert get_pipe(p, 'p1')['length'] == 10.0
|
|
assert get_pipe(p, 'p1')['diameter'] == 10.0
|
|
assert get_pipe(p, 'p1')['roughness'] == 10.0
|
|
assert get_pipe(p, 'p1')['minor_loss'] == 10.0
|
|
assert get_pipe(p, 'p1')['status'] == PIPE_STATUS_CLOSED
|
|
|
|
change_set = set_pipe(p, 'p1', {'node1': 'j2'})
|
|
assert len(change_set.operations) == 0
|
|
change_set = set_pipe(p, 'p1', {'node2': 'j1'})
|
|
assert len(change_set.operations) == 0
|
|
change_set = set_pipe(p, 'p1', {'status': 'XXX'})
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = set_pipe(p, 'p1', {'node1': 'j3'})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['node1']
|
|
|
|
change_set = set_pipe(p, 'p1', {'node2': 'j4'})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['node2']
|
|
|
|
change_set = set_pipe(p, 'p1', {'length': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['length']
|
|
|
|
change_set = set_pipe(p, 'p1', {'diameter': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['diameter']
|
|
|
|
change_set = set_pipe(p, 'p1', {'roughness': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['roughness']
|
|
|
|
change_set = set_pipe(p, 'p1', {'minor_loss': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['minor_loss']
|
|
|
|
change_set = set_pipe(p, 'p1', {'status': PIPE_STATUS_OPEN})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['status']
|
|
|
|
assert get_pipe(p, 'p1')['node1'] == 'j3'
|
|
assert get_pipe(p, 'p1')['node2'] == 'j4'
|
|
assert get_pipe(p, 'p1')['length'] == 100.0
|
|
assert get_pipe(p, 'p1')['diameter'] == 100.0
|
|
assert get_pipe(p, 'p1')['roughness'] == 100.0
|
|
assert get_pipe(p, 'p1')['minor_loss'] == 100.0
|
|
assert get_pipe(p, 'p1')['status'] == PIPE_STATUS_OPEN
|
|
|
|
change_set = add_pipe(p, 'p2', 'j1', 'j2', 10.0, 10.0, 10.0, 10.0, PIPE_STATUS_CLOSED)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == "p2"
|
|
|
|
pipes = get_links(p)
|
|
assert len(pipes) == 2
|
|
|
|
change_set = delete_pipe(p, 'p')
|
|
assert len(change_set.operations) == 0
|
|
|
|
assert is_pipe(p, 'p1')
|
|
|
|
change_set = delete_pipe(p, 'p1')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == "p1"
|
|
|
|
pipes = get_links(p)
|
|
assert len(pipes) == 1
|
|
|
|
change_set = delete_pipe(p, 'p2')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == PIPE
|
|
assert change_set.operations[0]['id'] == "p2"
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_pump(self):
|
|
p = "test_pump"
|
|
self.enter(p)
|
|
|
|
change_set = add_pump(p, 'p1', 'j1', 'j2')
|
|
assert len(change_set.operations) == 0
|
|
|
|
add_junction(p, 'j1', 0.0, 0.0, 0.0)
|
|
add_junction(p, 'j2', 10.0, 10.0, 0.0)
|
|
add_junction(p, 'j3', 0.0, 0.0, 0.0)
|
|
add_junction(p, 'j4', 10.0, 10.0, 0.0)
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
assert not is_link(p, 'p1')
|
|
|
|
change_set = add_pump(p, 'p1', 'j1', 'j1')
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = add_pump(p, 'p1', 'j1', 'j2')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == PUMP
|
|
assert change_set.operations[0]['id'] == "p1"
|
|
|
|
change_set = add_pump(p, 'p1', 'j1', 'j2')
|
|
assert len(change_set.operations) == 0
|
|
|
|
pumps = get_links(p)
|
|
assert len(pumps) == 1
|
|
|
|
assert get_pump(p, 'p1')['node1'] == 'j1'
|
|
assert get_pump(p, 'p1')['node2'] == 'j2'
|
|
|
|
change_set = set_pump(p, 'p1', {'node1': 'j2'})
|
|
assert len(change_set.operations) == 0
|
|
change_set = set_pump(p, 'p1', {'node2': 'j1'})
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = set_pump(p, 'p1', {'node1': 'j3'})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PUMP
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['node1']
|
|
|
|
change_set = set_pump(p, 'p1', {'node2': 'j4'})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == PUMP
|
|
assert change_set.operations[0]['id'] == 'p1'
|
|
assert change_set.operations[0]['properties'] == ['node2']
|
|
|
|
assert get_pump(p, 'p1')['node1'] == 'j3'
|
|
assert get_pump(p, 'p1')['node2'] == 'j4'
|
|
|
|
change_set = add_pump(p, 'p2', 'j1', 'j2')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == PUMP
|
|
assert change_set.operations[0]['id'] == "p2"
|
|
|
|
pumps = get_links(p)
|
|
assert len(pumps) == 2
|
|
|
|
change_set = delete_pump(p, 'p')
|
|
assert len(change_set.operations) == 0
|
|
|
|
assert is_pump(p, 'p1')
|
|
|
|
change_set = delete_pump(p, 'p1')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == PUMP
|
|
assert change_set.operations[0]['id'] == "p1"
|
|
|
|
pumps = get_links(p)
|
|
assert len(pumps) == 1
|
|
|
|
change_set = delete_pump(p, 'p2')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == PUMP
|
|
assert change_set.operations[0]['id'] == "p2"
|
|
|
|
self.leave(p)
|
|
|
|
|
|
def test_valve(self):
|
|
p = "test_valve"
|
|
self.enter(p)
|
|
|
|
change_set = add_valve(p, 'v1', 'j1', 'j2', 10.0, VALVES_TYPE_FCV, 10.0, 10.0)
|
|
assert len(change_set.operations) == 0
|
|
|
|
add_junction(p, 'j1', 0.0, 0.0, 0.0)
|
|
add_junction(p, 'j2', 10.0, 10.0, 0.0)
|
|
add_junction(p, 'j3', 0.0, 0.0, 0.0)
|
|
add_junction(p, 'j4', 10.0, 10.0, 0.0)
|
|
assert is_junction(p, 'j1')
|
|
assert is_junction(p, 'j2')
|
|
assert is_junction(p, 'j3')
|
|
assert is_junction(p, 'j4')
|
|
|
|
assert not is_link(p, 'v1')
|
|
|
|
change_set = add_valve(p, 'v1', 'j1', 'j2', 10.0, "xxx", 10.0, 10.0)
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = add_valve(p, 'v1', 'j1', 'j1', 10.0, VALVES_TYPE_FCV, 10.0, 10.0)
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = add_valve(p, 'v1', 'j1', 'j2', 10.0, VALVES_TYPE_FCV, 10.0, 10.0)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == "v1"
|
|
|
|
change_set = add_valve(p, 'v1', 'j1', 'j2', 10.0, VALVES_TYPE_FCV, 10.0, 10.0)
|
|
assert len(change_set.operations) == 0
|
|
|
|
valves = get_links(p)
|
|
assert len(valves) == 1
|
|
|
|
assert get_valve(p, 'v1')['node1'] == 'j1'
|
|
assert get_valve(p, 'v1')['node2'] == 'j2'
|
|
assert get_valve(p, 'v1')['diameter'] == 10.0
|
|
assert get_valve(p, 'v1')['type'] == VALVES_TYPE_FCV
|
|
assert get_valve(p, 'v1')['setting'] == 10.0
|
|
assert get_valve(p, 'v1')['minor_loss'] == 10.0
|
|
|
|
change_set = set_valve(p, 'v1', {'node1': 'j2'})
|
|
assert len(change_set.operations) == 0
|
|
change_set = set_valve(p, 'v1', {'node2': 'j1'})
|
|
assert len(change_set.operations) == 0
|
|
change_set = set_valve(p, 'v1', {'type': "XXX"})
|
|
assert len(change_set.operations) == 0
|
|
|
|
change_set = set_valve(p, 'v1', {'node1': 'j3'})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == 'v1'
|
|
assert change_set.operations[0]['properties'] == ['node1']
|
|
|
|
change_set = set_valve(p, 'v1', {'node2': 'j4'})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == 'v1'
|
|
assert change_set.operations[0]['properties'] == ['node2']
|
|
|
|
change_set = set_valve(p, 'v1', {'diameter': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == 'v1'
|
|
assert change_set.operations[0]['properties'] == ['diameter']
|
|
|
|
change_set = set_valve(p, 'v1', {'type': VALVES_TYPE_GPV})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == 'v1'
|
|
assert change_set.operations[0]['properties'] == ['type']
|
|
|
|
change_set = set_valve(p, 'v1', {'setting': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == 'v1'
|
|
assert change_set.operations[0]['properties'] == ['setting']
|
|
|
|
change_set = set_valve(p, 'v1', {'minor_loss': 100.0})
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'update'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == 'v1'
|
|
assert change_set.operations[0]['properties'] == ['minor_loss']
|
|
|
|
assert get_valve(p, 'v1')['node1'] == 'j3'
|
|
assert get_valve(p, 'v1')['node2'] == 'j4'
|
|
assert get_valve(p, 'v1')['diameter'] == 100.0
|
|
assert get_valve(p, 'v1')['type'] == VALVES_TYPE_GPV
|
|
assert get_valve(p, 'v1')['setting'] == 100.0
|
|
assert get_valve(p, 'v1')['minor_loss'] == 100.0
|
|
|
|
change_set = add_valve(p, 'v2', 'j1', 'j2', 10.0, VALVES_TYPE_FCV, 10.0, 10.0)
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'add'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == "v2"
|
|
|
|
valves = get_links(p)
|
|
assert len(valves) == 2
|
|
|
|
change_set = delete_valve(p, 'v')
|
|
assert len(change_set.operations) == 0
|
|
|
|
assert is_valve(p, 'v1')
|
|
|
|
change_set = delete_valve(p, 'v1')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == "v1"
|
|
|
|
valves = get_links(p)
|
|
assert len(valves) == 1
|
|
|
|
change_set = delete_valve(p, 'v2')
|
|
assert len(change_set.operations) == 1
|
|
assert change_set.operations[0]['operation'] == 'delete'
|
|
assert change_set.operations[0]['type'] == VALVE
|
|
assert change_set.operations[0]['id'] == "v2"
|
|
|
|
self.leave(p)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main()
|