NikolajDanger/meow_base
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Added previous work to new repo

Browse Source
This commit is contained in:
NikolajDanger
2023-05-06 21:06:53 +02:00
parent 933d568fb2
commit 83ee6b2d55
2 changed files with 231 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
patterns/network_event_pattern.py Normal file
View File

@ -0,0 +1,51 @@
from typing import Any, Dict
from meow_base.functionality.validation import valid_string, valid_dict
from meow_base.core.vars import VALID_RECIPE_NAME_CHARS, VALID_VARIABLE_NAME_CHARS
from meow_base.core.base_recipe import BaseRecipe
from meow_base.core.base_monitor import BaseMonitor
from meow_base.core.base_pattern import BasePattern
class NetworkEventPattern(BasePattern):
# The port to monitor
triggering_port:int
def __init__(self, name: str, triggering_port:int, recipe: str, parameters: Dict[str, Any] = {}, outputs: Dict[str, Any] = {}, sweep: Dict[str, Any] = {}):
super().__init__(name, recipe, parameters, outputs, sweep)
self._is_valid_port(triggering_port)
self.triggering_port = triggering_port
def _is_valid_port(self, port:int)->None:
if not isinstance(port, int):
raise ValueError (
f"Port '{port}' is not of type int."
)
elif not (0 < port < 65536):
raise ValueError (
f"Port '{port}' is not valid."
)
def _is_valid_recipe(self, recipe:str)->None:
"""Validation check for 'recipe' variable from main constructor.
Called within parent BasePattern constructor."""
valid_string(recipe, VALID_RECIPE_NAME_CHARS)
def _is_valid_parameters(self, parameters:Dict[str,Any])->None:
"""Validation check for 'parameters' variable from main constructor.
Called within parent BasePattern constructor."""
valid_dict(parameters, str, Any, strict=False, min_length=0)
for k in parameters.keys():
valid_string(k, VALID_VARIABLE_NAME_CHARS)
def _is_valid_output(self, outputs:Dict[str,str])->None:
"""Validation check for 'output' variable from main constructor.
Called within parent BasePattern constructor."""
valid_dict(outputs, str, str, strict=False, min_length=0)
for k in outputs.keys():
valid_string(k, VALID_VARIABLE_NAME_CHARS)
class NetworkMonitor(BaseMonitor):
def __init__(self, patterns: Dict[str, NetworkEventPattern], recipes: Dict[str, BaseRecipe]) -> None:
super().__init__(patterns, recipes)
self.ports = set(pattern.triggering_port for pattern in patterns.values())

280
tests/test_patterns.py
View File

@ -1,4 +1,3 @@
import io import io
import os import os
import unittest import unittest
@ -15,6 +14,7 @@ from meow_base.functionality.meow import create_rule
from meow_base.patterns.file_event_pattern import FileEventPattern, \ from meow_base.patterns.file_event_pattern import FileEventPattern, \
WatchdogMonitor, _DEFAULT_MASK, WATCHDOG_HASH, WATCHDOG_BASE, \ WatchdogMonitor, _DEFAULT_MASK, WATCHDOG_HASH, WATCHDOG_BASE, \
EVENT_TYPE_WATCHDOG, WATCHDOG_EVENT_KEYS, create_watchdog_event EVENT_TYPE_WATCHDOG, WATCHDOG_EVENT_KEYS, create_watchdog_event
from patterns.network_event_pattern import NetworkEventPattern
from meow_base.recipes.jupyter_notebook_recipe import JupyterNotebookRecipe from meow_base.recipes.jupyter_notebook_recipe import JupyterNotebookRecipe
from meow_base.recipes.python_recipe import PythonRecipe from meow_base.recipes.python_recipe import PythonRecipe
from shared import BAREBONES_NOTEBOOK, TEST_MONITOR_BASE, \ from shared import BAREBONES_NOTEBOOK, TEST_MONITOR_BASE, \
@ -26,9 +26,9 @@ def patterns_equal(tester, pattern_one, pattern_two):
tester.assertEqual(pattern_one.recipe, pattern_two.recipe) tester.assertEqual(pattern_one.recipe, pattern_two.recipe)
tester.assertEqual(pattern_one.parameters, pattern_two.parameters) tester.assertEqual(pattern_one.parameters, pattern_two.parameters)
tester.assertEqual(pattern_one.outputs, pattern_two.outputs) tester.assertEqual(pattern_one.outputs, pattern_two.outputs)
tester.assertEqual(pattern_one.triggering_path, tester.assertEqual(pattern_one.triggering_path,
pattern_two.triggering_path) pattern_two.triggering_path)
tester.assertEqual(pattern_one.triggering_file, tester.assertEqual(pattern_one.triggering_file,
pattern_two.triggering_file) pattern_two.triggering_file)
tester.assertEqual(pattern_one.event_mask, pattern_two.event_mask) tester.assertEqual(pattern_one.event_mask, pattern_two.event_mask)
tester.assertEqual(pattern_one.sweep, pattern_two.sweep) tester.assertEqual(pattern_one.sweep, pattern_two.sweep)
@ -140,15 +140,15 @@ class FileEventPatternTests(unittest.TestCase):
self.assertEqual(fep.event_mask, _DEFAULT_MASK) self.assertEqual(fep.event_mask, _DEFAULT_MASK)
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
fep = FileEventPattern("name", "path", "recipe", "file", fep = FileEventPattern("name", "path", "recipe", "file",
event_mask=FILE_CREATE_EVENT) event_mask=FILE_CREATE_EVENT)
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
fep = FileEventPattern("name", "path", "recipe", "file", fep = FileEventPattern("name", "path", "recipe", "file",
event_mask=["nope"]) event_mask=["nope"])
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
fep = FileEventPattern("name", "path", "recipe", "file", fep = FileEventPattern("name", "path", "recipe", "file",
event_mask=[FILE_CREATE_EVENT, "nope"]) event_mask=[FILE_CREATE_EVENT, "nope"])
# Test FileEventPattern created with valid parameter sweep # Test FileEventPattern created with valid parameter sweep
@ -176,7 +176,7 @@ class FileEventPatternTests(unittest.TestCase):
}, },
} }
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
fep = FileEventPattern("name", "path", "recipe", "file", fep = FileEventPattern("name", "path", "recipe", "file",
sweep=bad_sweep) sweep=bad_sweep)
bad_sweep = { bad_sweep = {
@ -187,7 +187,7 @@ class FileEventPatternTests(unittest.TestCase):
} }
} }
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
fep = FileEventPattern("name", "path", "recipe", "file", fep = FileEventPattern("name", "path", "recipe", "file",
sweep=bad_sweep) sweep=bad_sweep)
class WatchdogMonitorTests(unittest.TestCase): class WatchdogMonitorTests(unittest.TestCase):
@ -202,10 +202,10 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test creation of watchdog event dict # Test creation of watchdog event dict
def testCreateWatchdogEvent(self)->None: def testCreateWatchdogEvent(self)->None:
pattern = FileEventPattern( pattern = FileEventPattern(
"pattern", "pattern",
"file_path", "file_path",
"recipe_one", "recipe_one",
"infile", "infile",
parameters={ parameters={
"extra":"A line from a test Pattern", "extra":"A line from a test Pattern",
"outfile":"result_path" "outfile":"result_path"
@ -233,11 +233,11 @@ class WatchdogMonitorTests(unittest.TestCase):
self.assertEqual(event[WATCHDOG_HASH], "hash") self.assertEqual(event[WATCHDOG_HASH], "hash")
event = create_watchdog_event( event = create_watchdog_event(
"path2", "path2",
rule, rule,
"base", "base",
time(), time(),
"hash", "hash",
extras={"a":1} extras={"a":1}
) )
@ -260,7 +260,7 @@ class WatchdogMonitorTests(unittest.TestCase):
#TODO test valid watchdog event #TODO test valid watchdog event
# Test WatchdogMonitor created # Test WatchdogMonitor created
def testWatchdogMonitorMinimum(self)->None: def testWatchdogMonitorMinimum(self)->None:
from_monitor = Pipe() from_monitor = Pipe()
WatchdogMonitor(TEST_MONITOR_BASE, {}, {}, from_monitor[1]) WatchdogMonitor(TEST_MONITOR_BASE, {}, {}, from_monitor[1])
@ -297,7 +297,7 @@ class WatchdogMonitorTests(unittest.TestCase):
self.assertEqual(len(rules), 1) self.assertEqual(len(rules), 1)
rule = rules[list(rules.keys())[0]] rule = rules[list(rules.keys())[0]]
wm.start() wm.start()
open(os.path.join(TEST_MONITOR_BASE, "A"), "w") open(os.path.join(TEST_MONITOR_BASE, "A"), "w")
@ -308,12 +308,12 @@ class WatchdogMonitorTests(unittest.TestCase):
event = message event = message
self.assertIsNotNone(event) self.assertIsNotNone(event)
self.assertEqual(type(event), dict) self.assertEqual(type(event), dict)
self.assertTrue(EVENT_TYPE in event.keys()) self.assertTrue(EVENT_TYPE in event.keys())
self.assertTrue(EVENT_PATH in event.keys()) self.assertTrue(EVENT_PATH in event.keys())
self.assertTrue(WATCHDOG_BASE in event.keys()) self.assertTrue(WATCHDOG_BASE in event.keys())
self.assertTrue(EVENT_RULE in event.keys()) self.assertTrue(EVENT_RULE in event.keys())
self.assertEqual(event[EVENT_TYPE], EVENT_TYPE_WATCHDOG) self.assertEqual(event[EVENT_TYPE], EVENT_TYPE_WATCHDOG)
self.assertEqual(event[EVENT_PATH], self.assertEqual(event[EVENT_PATH],
os.path.join(TEST_MONITOR_BASE, "A")) os.path.join(TEST_MONITOR_BASE, "A"))
self.assertEqual(event[WATCHDOG_BASE], TEST_MONITOR_BASE) self.assertEqual(event[WATCHDOG_BASE], TEST_MONITOR_BASE)
self.assertEqual(event[EVENT_RULE].name, rule.name) self.assertEqual(event[EVENT_RULE].name, rule.name)
@ -330,10 +330,10 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor identifies expected events in sub directories # Test WatchdogMonitor identifies expected events in sub directories
def testMonitoring(self)->None: def testMonitoring(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("start", "A.txt"), os.path.join("start", "A.txt"),
"recipe_one", "recipe_one",
"infile", "infile",
parameters={}) parameters={})
recipe = JupyterNotebookRecipe( recipe = JupyterNotebookRecipe(
"recipe_one", BAREBONES_NOTEBOOK) "recipe_one", BAREBONES_NOTEBOOK)
@ -356,7 +356,7 @@ class WatchdogMonitorTests(unittest.TestCase):
from_monitor_reader, from_monitor_writer = Pipe() from_monitor_reader, from_monitor_writer = Pipe()
wm.to_runner_event = from_monitor_writer wm.to_runner_event = from_monitor_writer
wm.start() wm.start()
start_dir = os.path.join(TEST_MONITOR_BASE, "start") start_dir = os.path.join(TEST_MONITOR_BASE, "start")
@ -382,7 +382,7 @@ class WatchdogMonitorTests(unittest.TestCase):
self.assertIn(WATCHDOG_BASE, message) self.assertIn(WATCHDOG_BASE, message)
self.assertEqual(message[WATCHDOG_BASE], TEST_MONITOR_BASE) self.assertEqual(message[WATCHDOG_BASE], TEST_MONITOR_BASE)
self.assertIn(EVENT_PATH, message) self.assertIn(EVENT_PATH, message)
self.assertEqual(message[EVENT_PATH], self.assertEqual(message[EVENT_PATH],
os.path.join(start_dir, "A.txt")) os.path.join(start_dir, "A.txt"))
self.assertIn(EVENT_RULE, message) self.assertIn(EVENT_RULE, message)
self.assertEqual(message[EVENT_RULE].name, rule.name) self.assertEqual(message[EVENT_RULE].name, rule.name)
@ -392,10 +392,10 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor identifies directory content updates # Test WatchdogMonitor identifies directory content updates
def testMonitorDirectoryMonitoring(self)->None: def testMonitorDirectoryMonitoring(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("top"), os.path.join("top"),
"recipe_one", "recipe_one",
"dir_to_count", "dir_to_count",
parameters={}, parameters={},
event_mask=DIR_EVENTS event_mask=DIR_EVENTS
) )
@ -421,7 +421,7 @@ class WatchdogMonitorTests(unittest.TestCase):
from_monitor_reader, from_monitor_writer = Pipe() from_monitor_reader, from_monitor_writer = Pipe()
wm.to_runner_event = from_monitor_writer wm.to_runner_event = from_monitor_writer
wm.start() wm.start()
start_dir = os.path.join(TEST_MONITOR_BASE, "top") start_dir = os.path.join(TEST_MONITOR_BASE, "top")
@ -463,10 +463,10 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor identifies fake events for retroactive patterns # Test WatchdogMonitor identifies fake events for retroactive patterns
def testMonitoringRetroActive(self)->None: def testMonitoringRetroActive(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("start", "A.txt"), os.path.join("start", "A.txt"),
"recipe_one", "recipe_one",
"infile", "infile",
parameters={}) parameters={})
recipe = JupyterNotebookRecipe( recipe = JupyterNotebookRecipe(
"recipe_one", BAREBONES_NOTEBOOK) "recipe_one", BAREBONES_NOTEBOOK)
@ -493,7 +493,7 @@ class WatchdogMonitorTests(unittest.TestCase):
patterns, patterns,
recipes, recipes,
print=monitor_debug_stream, print=monitor_debug_stream,
logging=3, logging=3,
settletime=1 settletime=1
) )
@ -502,7 +502,7 @@ class WatchdogMonitorTests(unittest.TestCase):
from_monitor_reader, from_monitor_writer = Pipe() from_monitor_reader, from_monitor_writer = Pipe()
wm.to_runner_event = from_monitor_writer wm.to_runner_event = from_monitor_writer
wm.start() wm.start()
messages = [] messages = []
@ -520,7 +520,7 @@ class WatchdogMonitorTests(unittest.TestCase):
self.assertIn(WATCHDOG_BASE, message) self.assertIn(WATCHDOG_BASE, message)
self.assertEqual(message[WATCHDOG_BASE], TEST_MONITOR_BASE) self.assertEqual(message[WATCHDOG_BASE], TEST_MONITOR_BASE)
self.assertIn(EVENT_PATH, message) self.assertIn(EVENT_PATH, message)
self.assertEqual(message[EVENT_PATH], self.assertEqual(message[EVENT_PATH],
os.path.join(start_dir, "A.txt")) os.path.join(start_dir, "A.txt"))
self.assertIn(EVENT_RULE, message) self.assertIn(EVENT_RULE, message)
self.assertEqual(message[EVENT_RULE].name, rule.name) self.assertEqual(message[EVENT_RULE].name, rule.name)
@ -544,10 +544,10 @@ class WatchdogMonitorTests(unittest.TestCase):
) )
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("top"), os.path.join("top"),
"recipe_one", "recipe_one",
"dir_to_count", "dir_to_count",
parameters={}, parameters={},
event_mask=DIR_EVENTS event_mask=DIR_EVENTS
) )
@ -573,7 +573,7 @@ class WatchdogMonitorTests(unittest.TestCase):
from_monitor_reader, from_monitor_writer = Pipe() from_monitor_reader, from_monitor_writer = Pipe()
wm.to_runner_event = from_monitor_writer wm.to_runner_event = from_monitor_writer
wm.start() wm.start()
messages = [] messages = []
@ -600,16 +600,16 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor get_patterns function # Test WatchdogMonitor get_patterns function
def testMonitorGetPatterns(self)->None: def testMonitorGetPatterns(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("start", "A.txt"), os.path.join("start", "A.txt"),
"recipe_one", "recipe_one",
"infile", "infile",
parameters={}) parameters={})
pattern_two = FileEventPattern( pattern_two = FileEventPattern(
"pattern_two", "pattern_two",
os.path.join("start", "B.txt"), os.path.join("start", "B.txt"),
"recipe_two", "recipe_two",
"infile", "infile",
parameters={}) parameters={})
wm = WatchdogMonitor( wm = WatchdogMonitor(
@ -633,16 +633,16 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor add_pattern function # Test WatchdogMonitor add_pattern function
def testMonitorAddPattern(self)->None: def testMonitorAddPattern(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("start", "A.txt"), os.path.join("start", "A.txt"),
"recipe_one", "recipe_one",
"infile", "infile",
parameters={}) parameters={})
pattern_two = FileEventPattern( pattern_two = FileEventPattern(
"pattern_two", "pattern_two",
os.path.join("start", "B.txt"), os.path.join("start", "B.txt"),
"recipe_two", "recipe_two",
"infile", "infile",
parameters={}) parameters={})
wm = WatchdogMonitor( wm = WatchdogMonitor(
@ -684,16 +684,16 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor update_patterns function # Test WatchdogMonitor update_patterns function
def testMonitorUpdatePattern(self)->None: def testMonitorUpdatePattern(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("start", "A.txt"), os.path.join("start", "A.txt"),
"recipe_one", "recipe_one",
"infile", "infile",
parameters={}) parameters={})
pattern_two = FileEventPattern( pattern_two = FileEventPattern(
"pattern_two", "pattern_two",
os.path.join("start", "B.txt"), os.path.join("start", "B.txt"),
"recipe_two", "recipe_two",
"infile", "infile",
parameters={}) parameters={})
wm = WatchdogMonitor( wm = WatchdogMonitor(
@ -715,21 +715,21 @@ class WatchdogMonitorTests(unittest.TestCase):
self.assertIsInstance(patterns, dict) self.assertIsInstance(patterns, dict)
self.assertEqual(len(patterns), 1) self.assertEqual(len(patterns), 1)
self.assertIn(pattern_one.name, patterns) self.assertIn(pattern_one.name, patterns)
self.assertEqual(patterns[pattern_one.name].name, self.assertEqual(patterns[pattern_one.name].name,
pattern_one.name) pattern_one.name)
self.assertEqual(patterns[pattern_one.name].recipe, self.assertEqual(patterns[pattern_one.name].recipe,
"recipe_one") "recipe_one")
self.assertEqual(patterns[pattern_one.name].parameters, self.assertEqual(patterns[pattern_one.name].parameters,
pattern_one.parameters) pattern_one.parameters)
self.assertEqual(patterns[pattern_one.name].outputs, self.assertEqual(patterns[pattern_one.name].outputs,
pattern_one.outputs) pattern_one.outputs)
self.assertEqual(patterns[pattern_one.name].triggering_path, self.assertEqual(patterns[pattern_one.name].triggering_path,
pattern_one.triggering_path) pattern_one.triggering_path)
self.assertEqual(patterns[pattern_one.name].triggering_file, self.assertEqual(patterns[pattern_one.name].triggering_file,
pattern_one.triggering_file) pattern_one.triggering_file)
self.assertEqual(patterns[pattern_one.name].event_mask, self.assertEqual(patterns[pattern_one.name].event_mask,
pattern_one.event_mask) pattern_one.event_mask)
self.assertEqual(patterns[pattern_one.name].sweep, self.assertEqual(patterns[pattern_one.name].sweep,
pattern_one.sweep) pattern_one.sweep)
wm.update_pattern(pattern_one) wm.update_pattern(pattern_one)
@ -752,16 +752,16 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor remove_patterns function # Test WatchdogMonitor remove_patterns function
def testMonitorRemovePattern(self)->None: def testMonitorRemovePattern(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("start", "A.txt"), os.path.join("start", "A.txt"),
"recipe_one", "recipe_one",
"infile", "infile",
parameters={}) parameters={})
pattern_two = FileEventPattern( pattern_two = FileEventPattern(
"pattern_two", "pattern_two",
os.path.join("start", "B.txt"), os.path.join("start", "B.txt"),
"recipe_two", "recipe_two",
"infile", "infile",
parameters={}) parameters={})
wm = WatchdogMonitor( wm = WatchdogMonitor(
@ -893,15 +893,15 @@ class WatchdogMonitorTests(unittest.TestCase):
self.assertIsInstance(recipes, dict) self.assertIsInstance(recipes, dict)
self.assertEqual(len(recipes), 1) self.assertEqual(len(recipes), 1)
self.assertIn(recipe_one.name, recipes) self.assertIn(recipe_one.name, recipes)
self.assertEqual(recipes[recipe_one.name].name, self.assertEqual(recipes[recipe_one.name].name,
recipe_one.name) recipe_one.name)
self.assertEqual(recipes[recipe_one.name].recipe, self.assertEqual(recipes[recipe_one.name].recipe,
recipe_one.recipe) recipe_one.recipe)
self.assertEqual(recipes[recipe_one.name].parameters, self.assertEqual(recipes[recipe_one.name].parameters,
recipe_one.parameters) recipe_one.parameters)
self.assertEqual(recipes[recipe_one.name].requirements, self.assertEqual(recipes[recipe_one.name].requirements,
recipe_one.requirements) recipe_one.requirements)
self.assertEqual(recipes[recipe_one.name].source, self.assertEqual(recipes[recipe_one.name].source,
"") "")
wm.update_recipe(recipe_one) wm.update_recipe(recipe_one)
@ -963,16 +963,16 @@ class WatchdogMonitorTests(unittest.TestCase):
# Test WatchdogMonitor get_rules function # Test WatchdogMonitor get_rules function
def testMonitorGetRules(self)->None: def testMonitorGetRules(self)->None:
pattern_one = FileEventPattern( pattern_one = FileEventPattern(
"pattern_one", "pattern_one",
os.path.join("start", "A.txt"), os.path.join("start", "A.txt"),
"recipe_one", "recipe_one",
"infile", "infile",
parameters={}) parameters={})
pattern_two = FileEventPattern( pattern_two = FileEventPattern(
"pattern_two", "pattern_two",
os.path.join("start", "B.txt"), os.path.join("start", "B.txt"),
"recipe_two", "recipe_two",
"infile", "infile",
parameters={}) parameters={})
recipe_one = JupyterNotebookRecipe( recipe_one = JupyterNotebookRecipe(
"recipe_one", BAREBONES_NOTEBOOK) "recipe_one", BAREBONES_NOTEBOOK)
@ -998,3 +998,83 @@ class WatchdogMonitorTests(unittest.TestCase):
self.assertIsInstance(rules, dict) self.assertIsInstance(rules, dict)
self.assertEqual(len(rules), 2) self.assertEqual(len(rules), 2)
class NetworkEventPatternTests(unittest.TestCase):
def setUp(self)->None:
super().setUp()
setup()
def tearDown(self)->None:
super().tearDown()
teardown()
# Test NetworkEvent created
def testNetworkEventPatternCreationMinimum(self)->None:
NetworkEventPattern("name", 9000, "recipe")
# Test NetworkEventPattern not created with empty name
def testNetworkEventPatternCreationEmptyName(self)->None:
with self.assertRaises(ValueError):
NetworkEventPattern("", 9000, "recipe")
# Test NetworkEventPattern not created with empty recipe
def testNetworkEventPatternCreationEmptyRecipe(self)->None:
with self.assertRaises(ValueError):
NetworkEventPattern("name", 9000, "")
# Test NetworkEventPattern not created with invalid name
def testNetworkEventPatternCreationInvalidName(self)->None:
with self.assertRaises(ValueError):
NetworkEventPattern("@name", 9000, "recipe")
# Test NetworkEventPattern not created with invalid port
def testNetworkEventPatternCreationInvalidPort(self)->None:
with self.assertRaises(ValueError):
NetworkEventPattern("name", "9000", "recipe")
# Test NetworkEventPattern not created with invalid port
def testNetworkEventPatternCreationInvalidPort2(self)->None:
with self.assertRaises(ValueError):
NetworkEventPattern("name", 0, "recipe")
# Test NetworkEventPattern not created with invalid recipe
def testNetworkEventPatternCreationInvalidRecipe(self)->None:
with self.assertRaises(ValueError):
NetworkEventPattern("name", 9000, "@recipe")
# Test NetworkEventPattern created with valid name
def testNetworkEventPatternSetupName(self)->None:
name = "name"
nep = NetworkEventPattern(name, 9000, "recipe")
self.assertEqual(nep.name, name)
# Test NetworkEventPattern created with valid port
def testNetworkEventPatternSetupPort(self)->None:
port = 9000
nep = NetworkEventPattern("name", port, "recipe")
self.assertEqual(nep.triggering_port, port)
# Test NetworkEventPattern created with valid recipe
def testNetworkEventPatternSetupRecipe(self)->None:
recipe = "recipe"
nep = NetworkEventPattern("name", 9000, recipe)
self.assertEqual(nep.recipe, recipe)
# Test NetworkEventPattern created with valid parameters
def testNetworkEventPatternSetupParameters(self)->None:
parameters = {
"a": 1,
"b": True
}
fep = NetworkEventPattern("name", 9000, "recipe", parameters=parameters)
self.assertEqual(fep.parameters, parameters)
# Test NetworkEventPattern created with valid outputs
def testNetworkEventPatternSetupOutputs(self)->None:
outputs = {
"a": "a",
"b": "b"
}
fep = NetworkEventPattern("name", 9000, "recipe", outputs=outputs)
self.assertEqual(fep.outputs, outputs)