github/krkn
1
0
Fork 0
mirror of https://github.com/krkn-chaos/krkn.git synced 2026-02-14 18:10:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

pycodestyle fixes: tests/test_ingress_network_plugin.py

Browse Source 
Signed-off-by: Sandro Bonazzola <sbonazzo@redhat.com>
This commit is contained in:
Sandro Bonazzola
2022-09-02 15:51:53 +02:00
committed by Sandro Bonazzola
parent ed1c486c85
commit 79b92fc395
1 changed files with 29 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
tests/test_ingress_network_plugin.py
View File

@@ -1,44 +1,60 @@
import unittest import unittest
import os
import os
import logging import logging
from arcaflow_plugin_sdk import plugin from arcaflow_plugin_sdk import plugin
import kraken.plugins.network.kubernetes_functions as kube_helper
from kraken.plugins.network import ingress_shaping from kraken.plugins.network import ingress_shaping
class NetworkScenariosTest(unittest.TestCase): class NetworkScenariosTest(unittest.TestCase):
def test_serialization(self): def test_serialization(self):
plugin.test_object_serialization( plugin.test_object_serialization(
ingress_shaping.NetworkScenarioConfig(node_interface_name={"foo": ['bar']}, network_params={ ingress_shaping.NetworkScenarioConfig(
"latency": "50ms" , "loss": "0.02", "bandwidth": "100mbit"}), node_interface_name={"foo": ['bar']},
network_params={
"latency": "50ms",
"loss": "0.02",
"bandwidth": "100mbit"
}
),
self.fail, self.fail,
) )
plugin.test_object_serialization( plugin.test_object_serialization(
ingress_shaping.NetworkScenarioSuccessOutput( ingress_shaping.NetworkScenarioSuccessOutput(
filter_direction="ingress", test_interfaces= {"foo": ['bar']}, network_parameters={ filter_direction="ingress",
"latency": "50ms" , "loss": "0.02", "bandwidth": "100mbit"}, execution_type="parallel"), test_interfaces={"foo": ['bar']},
network_parameters={
"latency": "50ms",
"loss": "0.02",
"bandwidth": "100mbit"
},
execution_type="parallel"),
self.fail, self.fail,
) )
plugin.test_object_serialization( plugin.test_object_serialization(
ingress_shaping.NetworkScenarioErrorOutput( ingress_shaping.NetworkScenarioErrorOutput(
error="Hello World", error="Hello World",
), ),
self.fail, self.fail,
) )
def test_network_chaos(self): def test_network_chaos(self):
output_id, output_data = ingress_shaping.network_chaos( output_id, output_data = ingress_shaping.network_chaos(
ingress_shaping.NetworkScenarioConfig( ingress_shaping.NetworkScenarioConfig(
label_selector="node-role.kubernetes.io/master", instance_count=1, label_selector="node-role.kubernetes.io/master",
network_params = {"latency": "50ms" , "loss": "0.02", "bandwidth": "100mbit"} instance_count=1,
network_params={
"latency": "50ms",
"loss": "0.02",
"bandwidth": "100mbit"
}
) )
) )
if output_id == "error": if output_id == "error":
logging.error(output_data.error) logging.error(output_data.error)
self.fail("The network chaos scenario did not complete successfully because an error/exception occurred") self.fail(
"The network chaos scenario did not complete successfully "
"because an error/exception occurred"
)
if __name__ == "__main__": if __name__ == "__main__":