Skip to content

Commit

Permalink
Group Grouping tests
Browse files Browse the repository at this point in the history
  • Loading branch information
p-snft committed Sep 27, 2023
1 parent 8a2204b commit d9dddb6
Show file tree
Hide file tree
Showing 3 changed files with 71 additions and 50 deletions.
70 changes: 70 additions & 0 deletions tests/test_energy_system.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
# -*- coding: utf-8 -

"""Basic tests.
This file is part of project oemof.network (github.com/oemof/oemof-network).
SPDX-FileCopyrightText: Stephan Günther <>
SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert <>
SPDX-FileCopyrightText: Cord Kaldemeyer <>
SPDX-FileCopyrightText: Patrik Schönfeldt <[email protected]>
SPDX-License-Identifier: MIT
"""
from itertools import chain

from oemof.network import energy_system as es
from oemof.network.groupings import Flows
from oemof.network.groupings import FlowsWithNodes as FWNs
from oemof.network.network import Bus
from oemof.network.network.nodes import Node


class TestsEnergySystem:
def setup_method(self):
self.es = es.EnergySystem()

def test_flows(self):
key = object()
ensys = es.EnergySystem(groupings=[Flows(key)])
bus = Bus(label="A Bus")
node = Node(label="A Node", inputs={bus: None}, outputs={bus: None})
ensys.add(bus, node)
assert ensys.groups[key] == set(
chain(bus.inputs.values(), bus.outputs.values())
)

def test_flows_with_nodes(self):
key = object()
ensys = es.EnergySystem(groupings=[FWNs(key)])
bus = Bus(label="A Bus")
node = Node(label="A Node", inputs={bus: None}, outputs={bus: None})
ensys.add(bus, node)
assert ensys.groups[key], {
(bus, node, bus.outputs[node]),
(node, bus, node.outputs[bus]),
}

def test_that_node_additions_are_signalled(self):
"""
When a node gets `add`ed, a corresponding signal should be emitted.
"""
node = Node(label="Node")

def subscriber(sender, **kwargs):
assert sender is node
assert kwargs["EnergySystem"] is self.es
subscriber.called = True

subscriber.called = False

es.EnergySystem.signals[es.EnergySystem.add].connect(
subscriber, sender=node
)
self.es.add(node)
assert subscriber.called, (
"\nExpected `subscriber.called` to be `True`.\n"
"Got {}.\n"
"Probable reason: `subscriber` didn't get called."
).format(subscriber.called)
File renamed without changes.
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,7 @@

"""Basic tests.
This file is part of project oemof (github.com/oemof/oemof). It's copyrighted
by the contributors recorded in the version control history of the file,
available from its original location oemof/tests/basic_tests.py
This file is part of project oemof.network (github.com/oemof/oemof-network).
SPDX-FileCopyrightText: Stephan Günther <>
SPDX-FileCopyrightText: Uwe Krien <[email protected]>
Expand All @@ -15,13 +13,10 @@
SPDX-License-Identifier: MIT
"""
from collections.abc import Iterable
from itertools import chain
from pprint import pformat

from oemof.network import energy_system as es
from oemof.network.groupings import Entities
from oemof.network.groupings import Flows
from oemof.network.groupings import FlowsWithNodes as FWNs
from oemof.network.groupings import Grouping
from oemof.network.network import Bus
from oemof.network.network.nodes import Node
Expand Down Expand Up @@ -188,47 +183,3 @@ def everything():
assert everything in ensys.groups
assert ensys.groups[everything] == {node}
assert everything() == "everything"

def test_flows(self):
key = object()
ensys = es.EnergySystem(groupings=[Flows(key)])
bus = Bus(label="A Bus")
node = Node(label="A Node", inputs={bus: None}, outputs={bus: None})
ensys.add(bus, node)
assert ensys.groups[key] == set(
chain(bus.inputs.values(), bus.outputs.values())
)

def test_flows_with_nodes(self):
key = object()
ensys = es.EnergySystem(groupings=[FWNs(key)])
bus = Bus(label="A Bus")
node = Node(label="A Node", inputs={bus: None}, outputs={bus: None})
ensys.add(bus, node)
assert ensys.groups[key], {
(bus, node, bus.outputs[node]),
(node, bus, node.outputs[bus]),
}

def test_that_node_additions_are_signalled(self):
"""
When a node gets `add`ed, a corresponding signal should be emitted.
"""
node = Node(label="Node")

def subscriber(sender, **kwargs):
assert sender is node
assert kwargs["EnergySystem"] is self.es
subscriber.called = True

subscriber.called = False

es.EnergySystem.signals[es.EnergySystem.add].connect(
subscriber, sender=node
)
self.es.add(node)
assert subscriber.called, (
"\nExpected `subscriber.called` to be `True`.\n"
"Got {}.\n"
"Probable reason: `subscriber` didn't get called."
).format(subscriber.called)

0 comments on commit d9dddb6

Please sign in to comment.