Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: make the interfaces up after configuration #108

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 31 additions & 1 deletion src/upf_network.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,28 @@ def addresses_are_set(self) -> bool:
logger.warning("Interface %s not found in the network database", self.name)
return False

def interface_is_up(self) -> bool:
"""Check if the given network interface is up."""
interfaces = self.network_db.interfaces # type: ignore[reportAttributeAccessIssue]
if iface_record := interfaces.get(self.name):
return iface_record["state"] == "up"
logger.warning(
"Checking the state of network interface is failed: Interface %s not found in the network database", # noqa: E501
self.name,
)
return False

def bring_up_interface(self) -> None:
"""Set the network interface status to up."""
interfaces = self.network_db.interfaces # type: ignore[reportAttributeAccessIssue]
if iface_record := interfaces.get(self.name):
iface_record.set(state="up").commit()
gatici marked this conversation as resolved.
Show resolved Hide resolved
return
logger.warning(
"Setting the interface state to up is failed: Interface %s not found in the network database", # noqa: E501
self.name,
)

def set_ip_address(self) -> None:
"""Clean all unrequired IPs and set the IP address for the given network interface."""
interfaces = self.network_db.interfaces # type: ignore[reportAttributeAccessIssue]
Expand Down Expand Up @@ -296,10 +318,14 @@ def configure(self) -> None:
self.access_interface.set_ip_address()
if not self.access_interface.mtu_size_is_set():
self.access_interface.set_mtu_size()
if not self.access_interface.interface_is_up():
self.access_interface.bring_up_interface()
if not self.core_interface.addresses_are_set():
self.core_interface.set_ip_address()
if not self.core_interface.mtu_size_is_set():
self.core_interface.set_mtu_size()
if not self.core_interface.interface_is_up():
self.core_interface.bring_up_interface()
if not self.default_route.exists():
logger.info("Default route does not exist")
self.default_route.create()
Expand All @@ -321,7 +347,11 @@ def is_configured(self) -> bool:
and self.ran_route.exists()
and self.ip_tables_rule.exists()
)
return ifaces_are_configured and routes_are_configured
interfaces_are_up = (
self.access_interface.interface_is_up()
and self.core_interface.interface_is_up()
)
return ifaces_are_configured and routes_are_configured and interfaces_are_up

def clean_configuration(self) -> None:
"""Remove the configured IPs/routes from the networking."""
Expand Down
94 changes: 92 additions & 2 deletions tests/unit/test_upf_network.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,13 +23,23 @@ def __init__(self, ipv4_address: str = "", ipv6_address: str = ""):

class MockInterface:
def __init__(
self, name: str, ipv4_address: str = "", ipv6_address: str = "", mtu_size: int = 1500
self,
name: str,
ipv4_address: str = "",
ipv6_address: str = "",
mtu_size: int = 1500,
state: str = "down",
):
self.name = name
self.ipaddr = [MockIPAddr(ipv4_address=ipv4_address, ipv6_address=ipv6_address)]
self.mtu = mtu_size
self.state = state

def get(self, key):
return self.__getitem__(key)

def __getitem__(self, key):
"""Return the given attribute from the interface."""
return getattr(self, key)


Expand Down Expand Up @@ -151,13 +161,30 @@ def test_given_interface_has_ipv4_address_when_get_interface_ip_address_ip_is_re

assert address == self.interface_ipv4_address

def test_given_interface_status_is_up_when_get_interface_status_then_state_is_up(self):
self.network_interface.network_db.interfaces = MockInterfaces(
interfaces=[
MockInterface(
ipv4_address=self.interface_ipv4_address,
name=self.network_interface_name,
state="up",
)
]
)
assert self.network_interface.interface_is_up()

def test_given_interface_doesnt_exist_when_get_interface_ip_address_then_empty_string_is_returned(self): # noqa: E501
self.network_interface.network_db.interfaces = MockInterfaces(interfaces=[])

address = self.network_interface.get_ip_address()

assert address == ""

def test_given_interface_doesnt_exist_when_check_interface_states_then_false_is_returned(self): # noqa: E501
self.network_interface.network_db.interfaces = MockInterfaces(interfaces=[])

assert self.network_interface.interface_is_up() is False

def test_given_interface_doesnt_have_ipv4_address_when_get_interface_ip_address_then_empty_string_is_returned(self): # noqa: E501
self.network_interface.network_db.interfaces = MockInterfaces(
interfaces=[
Expand Down Expand Up @@ -1163,5 +1190,68 @@ def test_given_core_mtu_set_when_configure_then_set_mtu_size_is_not_called(self)
)

upf_network.configure()

mock_core_interface_instance.set_mtu_size.assert_not_called()

def test_given_interfaces_are_down_when_configure_then_bring_up_interface_is_called_for_both_interfaces(self): # noqa: E501
mock_access_interface_instance = MagicMock()
mock_access_interface_instance.is_valid.return_value = True
mock_access_interface_instance.addresses_are_set.return_value = True
mock_access_interface_instance.mtu_size_is_set.return_value = True
mock_access_interface_instance.interface_is_up.return_value = False
mock_core_interface_instance = MagicMock()
mock_core_interface_instance.is_valid.return_value = True
mock_core_interface_instance.addresses_are_set.return_value = True
mock_core_interface_instance.mtu_size_is_set.return_value = True
mock_core_interface_instance.interface_is_up.return_value = False
self.mock_network_interface.side_effect = [
mock_access_interface_instance,
mock_core_interface_instance,
]
upf_network = UPFNetwork(
access_interface_name=self.access_interface_name,
access_ip=self.access_ip,
access_gateway_ip=self.access_gateway_ip,
access_mtu_size=self.access_interface_mtu_size,
core_interface_name=self.core_interface_name,
core_ip=self.core_ip,
core_gateway_ip=self.core_gateway_ip,
core_mtu_size=self.core_interface_mtu_size,
gnb_subnet=self.gnb_subnet,
)

upf_network.configure()

mock_access_interface_instance.bring_up_interface.assert_called_once()
mock_core_interface_instance.bring_up_interface.assert_called_once()

def test_given_interfaces_are_up_when_configure_then_bring_up_interface_is_not_called(self):
mock_access_interface_instance = MagicMock()
mock_access_interface_instance.is_valid.return_value = True
mock_access_interface_instance.addresses_are_set.return_value = True
mock_access_interface_instance.mtu_size_is_set.return_value = True
mock_access_interface_instance.interface_is_up.return_value = True
mock_core_interface_instance = MagicMock()
mock_core_interface_instance.is_valid.return_value = True
mock_core_interface_instance.addresses_are_set.return_value = True
mock_core_interface_instance.mtu_size_is_set.return_value = True
mock_core_interface_instance.interface_is_up.return_value = True
self.mock_network_interface.side_effect = [
mock_access_interface_instance,
mock_core_interface_instance,
]
upf_network = UPFNetwork(
access_interface_name=self.access_interface_name,
access_ip=self.access_ip,
access_gateway_ip=self.access_gateway_ip,
access_mtu_size=self.access_interface_mtu_size,
core_interface_name=self.core_interface_name,
core_ip=self.core_ip,
core_gateway_ip=self.core_gateway_ip,
core_mtu_size=self.core_interface_mtu_size,
gnb_subnet=self.gnb_subnet,
)

upf_network.configure()

mock_access_interface_instance.bring_up_interface.assert_not_called()
mock_core_interface_instance.bring_up_interface.assert_not_called()
Loading