test_esipp_network.py

        # net.remove_edge(u="E1", v="B")

        # *********************************************************************

        # create a new import node

        net.add_import_node("I1", prices={(0, 0, 0): resource_price})

        # create an arc ending in that import node

        error_raised = False
        try:
            net.add_directed_arc(node_key_a="A", node_key_b="I1", arcs=lossless_arcs)
        except ValueError:
            error_raised = True
        assert error_raised

        # # remove the troublesome arc
        # net.remove_edge(u="A", v="I1")

        # *********************************************************************

        # check non-existent arc
        net.arc_is_undirected(("X", "Y", 1))
        
    # *************************************************************************
    # *************************************************************************
            
    def test_undirected_arc_import_error(self):
        
        # network    
        mynet = Network()
    
        # import node    
        imp_node_key = generate_pseudo_unique_key(mynet.nodes())    
        mynet.add_import_node(
            imp_node_key,
            prices={
                (0, 0, 0): ResourcePrice(prices=1+0.05, volumes=None)
            },
        )
    
        # other nodes
        node_A = generate_pseudo_unique_key(mynet.nodes())
        mynet.add_source_sink_node(
            node_A,
            # base_flow=[1, -1, 0.5, -0.5]
            base_flow={(0, 0): 1, (0, 1): -1, (0, 2): 0.5, (0, 3): -0.5},
        )
        node_B = generate_pseudo_unique_key(mynet.nodes())
        mynet.add_source_sink_node(
            node_B,
            # base_flow=[-1, 1, -0.5, 0.5]
            base_flow={(0, 0): -1, (0, 1): 1, (0, 2): -0.5, (0, 3): 0.5},
        )
    
        # add arcs
    
        # import arc
        arc_tech_IA = Arcs(
            name="any",
            # efficiency=[1, 1, 1, 1],
            efficiency={(0, 0): 1, (0, 1): 1, (0, 2): 1, (0, 3): 1},
            capacity=[0.5, 0.75, 1.0, 1.25, 1.5, 2.0],
            minimum_cost=[10, 10.1, 10.2, 10.3, 10.4, 10.5],
            specific_capacity_cost=1,
            capacity_is_instantaneous=False,
            efficiency_reverse=None,
            static_loss=None,
            validate=False,
        )
        
        error_raised = False
        try:
            # ValueError: Undirected arcs cannot involve import or export nodes.
            mynet.add_undirected_arc(
                node_key_a=imp_node_key, node_key_b=node_A, arcs=arc_tech_IA
            )
        except ValueError:
            error_raised = True
        assert error_raised
    
        # *********************************************************************
        # *********************************************************************
        
    # *************************************************************************
    # *************************************************************************
            
    def test_undirected_arc_export_error(self):
            
        # 4 nodes: one import, one export, two supply/demand nodes
        mynet = Network()
    
        # export node
        exp_node_key = generate_pseudo_unique_key(mynet.nodes())
        mynet.add_export_node(
            exp_node_key,
            prices={
                (0, 0, 0): ResourcePrice(prices=0.1+0.05, volumes=None)
            },
        )
    
        # other nodes
        node_B = generate_pseudo_unique_key(mynet.nodes())
        mynet.add_source_sink_node(
            node_B,
            # base_flow=[-1, 1, -0.5, 0.5]
            base_flow={(0, 0): -1, (0, 1): 1, (0, 2): -0.5, (0, 3): 0.5},
        )    
        # export arc
        arc_tech_BE = Arcs(
            name="any",
            # efficiency=[1, 1, 1, 1],
            efficiency={(0, 0): 1, (0, 1): 1, (0, 2): 1, (0, 3): 1},
            capacity=[0.5, 0.75, 1.0, 1.25, 1.5, 2.0],
            minimum_cost=[10, 10.1, 10.2, 10.3, 10.4, 10.5],
            specific_capacity_cost=1,
            capacity_is_instantaneous=False,
            efficiency_reverse=None,
            static_loss=None,
            validate=False,
        )
        
        error_raised = False
        try:
            # ValueError: Undirected arcs cannot involve import or export nodes.
            mynet.add_undirected_arc(
                node_key_a=node_B, node_key_b=exp_node_key, arcs=arc_tech_BE
            )
        except ValueError:
            error_raised = True
        assert error_raised
        
    # *************************************************************************
    # *************************************************************************

    def test_tree_topology(self):
        
        # create a network object with a tree topology
        tree_network = binomial_tree(3, create_using=MultiDiGraph)
        network = Network(network_type=Network.NET_TYPE_TREE, incoming_graph_data=tree_network)
        for edge_key in network.edges(keys=True):
            arc = ArcsWithoutLosses(
                name=str(edge_key),
                capacity=[5, 10],
                minimum_cost=[3, 6],
                specific_capacity_cost=0,
                capacity_is_instantaneous=False,
            )
            network.add_edge(*edge_key, **{Network.KEY_ARC_TECH: arc})
            
        # assert that it should have a tree topology
        assert network.should_be_tree_network()
            
        # assert that it does not have a tree topology
        assert not network.has_tree_topology()

        # select all the nodes
        for edge_key in network.edges(keys=True):
            network.edges[edge_key][Network.KEY_ARC_TECH].options_selected[0] = True
            
        # assert that it has a tree topology
        assert network.has_tree_topology()
        
        

    # *************************************************************************
    # *************************************************************************

    def test_pseudo_unique_key_generation(self):
        
        # create network
        network = Network()

        # add nodes A and B
        network.add_nodes_from(['A','B'])

        # add arcs
        key_list = [
            "3e225573-4e78-48c8-bb08-efbeeb795c22",
            "f6d30428-15d1-41e9-a952-0742eaaa5a31",
            "8c29b906-2518-41c5-ada8-07b83508b5b8",
            "f9a72a39-1422-4a02-af97-906ce79c32a3",
            "b6941a48-10cc-465d-bf53-178bd2939bd1",
        ]

        for key in key_list:
            network.add_edge(
                u_for_edge="A",
                v_for_edge="B",
                key=key,
                **{network.KEY_ARC_UND: False, network.KEY_ARC_TECH: None}
            )

        # use a seed number to trigger more iterations

        import uuid

        rand = random.Random()
        rand.seed(360)
        uuid.uuid4 = lambda: uuid.UUID(int=rand.getrandbits(128), version=4)

        error_raised = False
        try:
            _ = network.get_pseudo_unique_arc_key(
                node_key_start="A", node_key_end="B", max_iterations=len(key_list) - 1
            )
        except Exception:
            error_raised = True
        assert error_raised
        
    # *************************************************************************
    # *************************************************************************
    
    def test_imp_exp_static_losses(self):
                
        # assessment
        q = 0
        # 4 nodes: one import, one export, two supply/demand nodes
        mynet = Network()
    
        # import node
        imp_node_key = generate_pseudo_unique_key(mynet.nodes())
        imp_prices = {
            qpk: ResourcePrice(
                prices=0.5,
                volumes=None,
            )
            for qpk in [(0,0,0),(0,0,1),(0,1,0),(0,1,1)]
            }
        mynet.add_import_node(
            imp_node_key,
            prices=imp_prices
        )
    
        # export node
        exp_node_key = generate_pseudo_unique_key(mynet.nodes())
        exp_prices = {
            qpk: ResourcePrice(
                prices=1.5,
                volumes=None,
            )
            for qpk in [(0,0,0),(0,0,1),(0,1,0),(0,1,1)]
            }
        mynet.add_export_node(
            exp_node_key,
            prices=exp_prices,
        )
        
        # add arc with fixed losses from import node to export

        arc_tech_IE_fix = Arcs(
            name="IE_fix",
            # efficiency=[1, 1, 1, 1],
            efficiency={(q, 0): 1, (q, 1): 1},
            efficiency_reverse=None,
            validate=False,
            capacity=[0.5, 1.0, 2.0],
            minimum_cost=[5, 5.1, 5.2],
            specific_capacity_cost=1,
            capacity_is_instantaneous=False,
            # static_losses=[
            #     [0.10, 0.15, 0.20, 0.25],
            #     [0.15, 0.20, 0.25, 0.30],
            #     [0.20, 0.25, 0.30, 0.35]]
            static_loss={
                (0, q, 0): 0.10,
                (0, q, 1): 0.15,
                (1, q, 0): 0.15,
                (1, q, 1): 0.20,
                (2, q, 0): 0.20,
                (2, q, 1): 0.25,
            },
        )
    
        error_raised = False
        try:
            # ValueError: Arcs between import and export nodes cannot have static losses.
            mynet.add_directed_arc(
                node_key_a=imp_node_key, node_key_b=exp_node_key, arcs=arc_tech_IE_fix
            )
        except ValueError:
            error_raised = True
        assert error_raised
        
    # *************************************************************************
    # *************************************************************************
    
    def test_antiparallel_arcs(self):
        
        # create network        
        net = Network()
        
        # add nodes
        node_a = 'A'
        # net.add_waypoint_node(node_a)
        node_b = 'B'
        # net.add_waypoint_node(node_b)
        node_c = 'C'
        # net.add_waypoint_node(node_c)
        net.add_nodes_from([node_a,node_b,node_c])
        
        # add arcs
        node_pairs = ((node_a, node_b), (node_b, node_a),)
        
        # test network
        for node_pair in node_pairs:
            net.add_preexisting_directed_arc(
                *node_pair,
                efficiency=None, 
                static_loss=None, 
                capacity=1, 
                capacity_is_instantaneous=False
                )
        
        # assert that it can detected the selected antiparallel arcs
        assert net.has_selected_antiparallel_arcs()
        # check that it finds the right node pairs
        identified_node_pairs = net.find_selected_antiparallel_arcs()
        assert (node_a, node_b) in identified_node_pairs
        assert (node_b, node_a) in identified_node_pairs
        
    # *************************************************************************
    # *************************************************************************
    
    def test_add_nodes(self):
        
        # create network        
        net = Network()
        
        # add nodes
        node_a = 'A'
        net.add_node(node_a)
        assert net.is_waypoint_node(node_a)
        # source
        node_b = 'B'
        net.add_node(node_b, **{net.KEY_NODE_BASE_FLOW: {(0,0):-1}})
        assert net.is_source_sink_node(node_b)
        # sink
        node_c = 'C'
        net.add_node(node_c, **{net.KEY_NODE_BASE_FLOW: {(0,0):1}})
        assert net.is_source_sink_node(node_c)
        # import node
        node_d = 'D'
        net.add_node(node_d, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[1, 2], volumes=[1,None])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP})
        assert net.is_import_node(node_d)
        # export node
        node_e = 'E'
        net.add_node(node_e, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[2, 3], volumes=[4,None])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP})
        assert net.is_export_node(node_e)            
        
        # modify nodes
        # from waypoint to source/sink
        net.modify_node(node_a, **{net.KEY_NODE_BASE_FLOW: {(0,0):-2}})
        assert not net.is_waypoint_node(node_a)
        assert net.is_source_sink_node(node_a)
        # from source/sink to waypoint
        net.modify_node(node_a)
        assert not net.is_source_sink_node(node_a)
        assert net.is_waypoint_node(node_a)
        # from waypoint to import node
        net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[5, 3.5], volumes=[2,4])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP})
        assert not net.is_waypoint_node(node_a)
        assert net.is_import_node(node_a)
        # from import node to waypoint
        net.modify_node(node_a)
        assert not net.is_import_node(node_a)
        assert net.is_waypoint_node(node_a)
        # from waypoint node to export node
        net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP})
        assert not net.is_waypoint_node(node_a)
        assert net.is_export_node(node_a)
        # from export node to sink/source
        net.modify_node(node_a, **{net.KEY_NODE_BASE_FLOW: {(0,0):-1}})
        assert not net.is_export_node(node_a)
        assert net.is_source_sink_node(node_a)
        # from sink/source node to import node
        net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[5, 3.5], volumes=[2,4])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP})
        assert not net.is_source_sink_node(node_a)
        assert net.is_import_node(node_a)
        # from import node to export node
        net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP})
        assert not net.is_import_node(node_a)
        assert net.is_export_node(node_a)
        # from export node to waypoint node
        net.modify_node(node_a)
        assert not net.is_export_node(node_a)
        assert net.is_waypoint_node(node_a)
        
        # *********************************************************************
        
        # test modifying nodes with preexisting arcs
        
        # add arcs
        # add arc between two waypoint nodes
        net.add_preexisting_directed_arc(
            node_key_a=node_a, 
            node_key_b=node_b, 
            efficiency=None, 
            static_loss=None,  
            capacity=3, 
            capacity_is_instantaneous=False
            )
        
        # modify nodes
        # try to change the start node to an export node
        with pytest.raises(ValueError):
            net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP})
        # try to change the end node to an import node
        with pytest.raises(ValueError):
            net.modify_node(node_b, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP})
        
    # *************************************************************************
    # *************************************************************************

# *****************************************************************************
# *****************************************************************************