daemon: LinkOptions now leverage dataclass and has type hinting, improve test_gui type hinting
This commit is contained in:
parent
18044f9474
commit
b5e53e573a
4 changed files with 27 additions and 39 deletions
|
@ -86,13 +86,8 @@ def add_link_data(
|
||||||
"""
|
"""
|
||||||
interface_one = link_interface(link_proto.interface_one)
|
interface_one = link_interface(link_proto.interface_one)
|
||||||
interface_two = link_interface(link_proto.interface_two)
|
interface_two = link_interface(link_proto.interface_two)
|
||||||
|
link_type = LinkTypes(link_proto.type)
|
||||||
link_type = None
|
options = LinkOptions(type=link_type)
|
||||||
link_type_value = link_proto.type
|
|
||||||
if link_type_value is not None:
|
|
||||||
link_type = LinkTypes(link_type_value)
|
|
||||||
|
|
||||||
options = LinkOptions(_type=link_type)
|
|
||||||
options_data = link_proto.options
|
options_data = link_proto.options
|
||||||
if options_data:
|
if options_data:
|
||||||
options.delay = options_data.delay
|
options.delay = options_data.delay
|
||||||
|
@ -106,7 +101,6 @@ def add_link_data(
|
||||||
options.unidirectional = options_data.unidirectional
|
options.unidirectional = options_data.unidirectional
|
||||||
options.key = options_data.key
|
options.key = options_data.key
|
||||||
options.opaque = options_data.opaque
|
options.opaque = options_data.opaque
|
||||||
|
|
||||||
return interface_one, interface_two, options
|
return interface_one, interface_two, options
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -772,7 +772,7 @@ class CoreHandler(socketserver.BaseRequestHandler):
|
||||||
if link_type_value is not None:
|
if link_type_value is not None:
|
||||||
link_type = LinkTypes(link_type_value)
|
link_type = LinkTypes(link_type_value)
|
||||||
|
|
||||||
link_options = LinkOptions(_type=link_type)
|
link_options = LinkOptions(type=link_type)
|
||||||
link_options.delay = message.get_tlv(LinkTlvs.DELAY.value)
|
link_options.delay = message.get_tlv(LinkTlvs.DELAY.value)
|
||||||
link_options.bandwidth = message.get_tlv(LinkTlvs.BANDWIDTH.value)
|
link_options.bandwidth = message.get_tlv(LinkTlvs.BANDWIDTH.value)
|
||||||
link_options.session = message.get_tlv(LinkTlvs.SESSION.value)
|
link_options.session = message.get_tlv(LinkTlvs.SESSION.value)
|
||||||
|
|
|
@ -93,34 +93,28 @@ class NodeOptions:
|
||||||
self.alt = alt
|
self.alt = alt
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
class LinkOptions:
|
class LinkOptions:
|
||||||
"""
|
"""
|
||||||
Options for creating and updating links within core.
|
Options for creating and updating links within core.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, _type: LinkTypes = LinkTypes.WIRED) -> None:
|
type: LinkTypes = LinkTypes.WIRED
|
||||||
"""
|
session: int = None
|
||||||
Create a LinkOptions object.
|
delay: int = None
|
||||||
|
bandwidth: int = None
|
||||||
:param _type: type of link, defaults to
|
per: float = None
|
||||||
wired
|
dup: int = None
|
||||||
"""
|
jitter: int = None
|
||||||
self.type = _type
|
mer: int = None
|
||||||
self.session = None
|
burst: int = None
|
||||||
self.delay = None
|
mburst: int = None
|
||||||
self.bandwidth = None
|
gui_attributes: str = None
|
||||||
self.per = None
|
unidirectional: bool = None
|
||||||
self.dup = None
|
emulation_id: int = None
|
||||||
self.jitter = None
|
network_id: int = None
|
||||||
self.mer = None
|
key: int = None
|
||||||
self.burst = None
|
opaque: str = None
|
||||||
self.mburst = None
|
|
||||||
self.gui_attributes = None
|
|
||||||
self.unidirectional = None
|
|
||||||
self.emulation_id = None
|
|
||||||
self.network_id = None
|
|
||||||
self.key = None
|
|
||||||
self.opaque = None
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|
|
@ -511,7 +511,7 @@ class TestGui:
|
||||||
EventTypes.DEFINITION_STATE,
|
EventTypes.DEFINITION_STATE,
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
def test_event_state(self, coretlv, state):
|
def test_event_state(self, coretlv: CoreHandler, state: EventTypes):
|
||||||
message = coreapi.CoreEventMessage.create(0, [(EventTlvs.TYPE, state.value)])
|
message = coreapi.CoreEventMessage.create(0, [(EventTlvs.TYPE, state.value)])
|
||||||
|
|
||||||
coretlv.handle_message(message)
|
coretlv.handle_message(message)
|
||||||
|
@ -536,7 +536,7 @@ class TestGui:
|
||||||
|
|
||||||
coretlv.session.add_event.assert_called_once()
|
coretlv.session.add_event.assert_called_once()
|
||||||
|
|
||||||
def test_event_save_xml(self, coretlv, tmpdir):
|
def test_event_save_xml(self, coretlv: CoreHandler, tmpdir):
|
||||||
xml_file = tmpdir.join("coretlv.session.xml")
|
xml_file = tmpdir.join("coretlv.session.xml")
|
||||||
file_path = xml_file.strpath
|
file_path = xml_file.strpath
|
||||||
coretlv.session.add_node(CoreNode)
|
coretlv.session.add_node(CoreNode)
|
||||||
|
@ -549,7 +549,7 @@ class TestGui:
|
||||||
|
|
||||||
assert os.path.exists(file_path)
|
assert os.path.exists(file_path)
|
||||||
|
|
||||||
def test_event_open_xml(self, coretlv, tmpdir):
|
def test_event_open_xml(self, coretlv: CoreHandler, tmpdir):
|
||||||
xml_file = tmpdir.join("coretlv.session.xml")
|
xml_file = tmpdir.join("coretlv.session.xml")
|
||||||
file_path = xml_file.strpath
|
file_path = xml_file.strpath
|
||||||
node = coretlv.session.add_node(CoreNode)
|
node = coretlv.session.add_node(CoreNode)
|
||||||
|
@ -573,7 +573,7 @@ class TestGui:
|
||||||
EventTypes.RECONFIGURE,
|
EventTypes.RECONFIGURE,
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
def test_event_service(self, coretlv, state):
|
def test_event_service(self, coretlv: CoreHandler, state: EventTypes):
|
||||||
coretlv.session.broadcast_event = mock.MagicMock()
|
coretlv.session.broadcast_event = mock.MagicMock()
|
||||||
node = coretlv.session.add_node(CoreNode)
|
node = coretlv.session.add_node(CoreNode)
|
||||||
message = coreapi.CoreEventMessage.create(
|
message = coreapi.CoreEventMessage.create(
|
||||||
|
@ -599,7 +599,7 @@ class TestGui:
|
||||||
EventTypes.RECONFIGURE,
|
EventTypes.RECONFIGURE,
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
def test_event_mobility(self, coretlv, state):
|
def test_event_mobility(self, coretlv: CoreHandler, state: EventTypes):
|
||||||
message = coreapi.CoreEventMessage.create(
|
message = coreapi.CoreEventMessage.create(
|
||||||
0, [(EventTlvs.TYPE, state.value), (EventTlvs.NAME, "mobility:ns2script")]
|
0, [(EventTlvs.TYPE, state.value), (EventTlvs.NAME, "mobility:ns2script")]
|
||||||
)
|
)
|
||||||
|
@ -610,7 +610,7 @@ class TestGui:
|
||||||
message = coreapi.CoreRegMessage.create(0, [(RegisterTlvs.GUI, "gui")])
|
message = coreapi.CoreRegMessage.create(0, [(RegisterTlvs.GUI, "gui")])
|
||||||
coretlv.handle_message(message)
|
coretlv.handle_message(message)
|
||||||
|
|
||||||
def test_register_xml(self, coretlv, tmpdir):
|
def test_register_xml(self, coretlv: CoreHandler, tmpdir):
|
||||||
xml_file = tmpdir.join("coretlv.session.xml")
|
xml_file = tmpdir.join("coretlv.session.xml")
|
||||||
file_path = xml_file.strpath
|
file_path = xml_file.strpath
|
||||||
node = coretlv.session.add_node(CoreNode)
|
node = coretlv.session.add_node(CoreNode)
|
||||||
|
@ -625,7 +625,7 @@ class TestGui:
|
||||||
|
|
||||||
assert coretlv.coreemu.sessions[1].get_node(node.id, CoreNode)
|
assert coretlv.coreemu.sessions[1].get_node(node.id, CoreNode)
|
||||||
|
|
||||||
def test_register_python(self, coretlv, tmpdir):
|
def test_register_python(self, coretlv: CoreHandler, tmpdir):
|
||||||
xml_file = tmpdir.join("test.py")
|
xml_file = tmpdir.join("test.py")
|
||||||
file_path = xml_file.strpath
|
file_path = xml_file.strpath
|
||||||
with open(file_path, "w") as f:
|
with open(file_path, "w") as f:
|
||||||
|
|
Loading…
Add table
Reference in a new issue