Skip to content
Snippets Groups Projects
Commit d6a48a27 authored by Martin Lange's avatar Martin Lange
Browse files

add a check to avoid giving infos that already have rules

parent 6f3b8395
No related branches found
No related tags found
1 merge request!154Auto-transfer infos
......@@ -111,7 +111,7 @@ class ConnectHelper(Loggable):
with ErrorLogger(self.logger):
for name in pull_data or []:
if name not in self._inputs:
raise ValueError(
raise KeyError(
f"No input named '{name}' available to get info for."
)
......@@ -273,6 +273,7 @@ class ConnectHelper(Loggable):
with ErrorLogger(self.logger):
self._check_names(exchange_infos, push_infos, push_data)
self._check_in_rules(exchange_infos, push_infos)
exchange_infos = {
k: v for k, v in exchange_infos.items() if self.in_infos[k] is None
......@@ -359,15 +360,29 @@ class ConnectHelper(Loggable):
def _check_names(self, exchange_infos, push_infos, push_data):
for name in exchange_infos:
if name not in self._inputs:
raise ValueError(
raise KeyError(
f"No input named '{name}' available to exchange info for."
)
for name in push_infos:
if name not in self._outputs:
raise ValueError(f"No output named '{name}' available to push info.")
raise KeyError(f"No output named '{name}' available to push info.")
for name in push_data:
if name not in self._outputs:
raise ValueError(f"No output named '{name}' available to push data.")
raise KeyError(f"No output named '{name}' available to push data.")
def _check_in_rules(self, exchange_infos, push_infos):
for name in exchange_infos:
if name in self._in_info_rules:
raise ValueError(
f"There are info transfer rules given for input `{name}`. "
f"Can't provide the info directly."
)
for name in push_infos:
if name in self._out_info_rules:
raise ValueError(
f"There are info transfer rules given for output `{name}`. "
f"Can't provide the info directly."
)
def _exchange_in_infos(self):
any_done = False
......
......@@ -119,16 +119,16 @@ class TestConnectHelper(unittest.TestCase):
outputs.add(name="Out1")
outputs.add(name="Out2")
with self.assertRaises(ValueError):
with self.assertRaises(KeyError):
_connector = ConnectHelper("TestLogger", inputs, outputs, pull_data=["In3"])
connector = ConnectHelper("TestLogger", inputs, outputs)
with self.assertRaises(ValueError):
with self.assertRaises(KeyError):
connector.connect(time=None, exchange_infos={"In3": Info(time, NoGrid())})
with self.assertRaises(ValueError):
with self.assertRaises(KeyError):
connector.connect(time=None, push_infos={"Out3": Info(time, NoGrid())})
with self.assertRaises(ValueError):
with self.assertRaises(KeyError):
connector.connect(time=None, push_data={"Out3": 0.0})
def test_connect_caching(self):
......@@ -320,8 +320,7 @@ class TestConnectHelper(unittest.TestCase):
{"Out1": Info(time=time, grid=UniformGrid((10, 10)), units="m")},
)
def test_connect_transfer_fail(self):
def test_connect_constructor_fail(self):
inputs = IOList(None, "INPUT")
inputs.add(name="In1")
outputs = IOList(None, "OUTPUT")
......@@ -366,3 +365,27 @@ class TestConnectHelper(unittest.TestCase):
outputs,
in_info_rules={"In1": [0]},
)
def test_connect_rules_fail(self):
inputs = IOList(None, "INPUT")
inputs.add(name="In1")
outputs = IOList(None, "OUTPUT")
outputs.add(name="Out1")
connector: ConnectHelper = ConnectHelper(
"TestLogger",
inputs,
outputs,
in_info_rules={"In1": [FromOutput("Out1")]},
out_info_rules={"Out1": [FromInput("In1")]},
)
with self.assertRaises(ValueError):
connector.connect(
time=None, exchange_infos={"In1": Info(time=None, grid=NoGrid())}
)
with self.assertRaises(ValueError):
connector.connect(
time=None, push_infos={"Out1": Info(time=None, grid=NoGrid())}
)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment