Skip to content

Add NTP tests #60

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.

Sign up for GitHub

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

Merged
jboddey merged 7 commits into from
Jul 12, 2023
Merged

Add NTP tests #60

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
00254f7
Add ntp support test
jboddey Jun 30, 2023
185c7e1
Add extra log message
jboddey Jun 30, 2023
9f00040
Merge branch 'dev' into ntp
jboddey Jul 4, 2023
459ae30
Modify descriptions
jboddey Jul 7, 2023
085785d
Pylint
jboddey Jul 7, 2023
79acb96
Pylint (#69)
jhughesoti Jul 8, 2023
55433dc
Merge branch 'dev' into ntp
jboddey Jul 12, 2023
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
12 changes: 6 additions & 6 deletions framework/python/src/net_orc/network_validator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,7 @@ def _get_os_user(self):
LOGGER.error('An OS error occurred while retrieving the login name.')
except Exception as error:
# Catch any other unexpected exceptions
LOGGER.error('An exception occurred:', error)
LOGGER.error('An exception occurred:', error)
return user

def _get_user(self):
Expand All @@ -203,15 +203,15 @@ def _get_user(self):
except (KeyError, ImportError, ModuleNotFoundError, OSError) as e:
# Handle specific exceptions individually
if isinstance(e, KeyError):
LOGGER.error("USER environment variable not set or unavailable.")
LOGGER.error('USER environment variable not set or unavailable.')
elif isinstance(e, ImportError):
LOGGER.error("Unable to import the getpass module.")
LOGGER.error('Unable to import the getpass module.')
elif isinstance(e, ModuleNotFoundError):
LOGGER.error("The getpass module was not found.")
LOGGER.error('The getpass module was not found.')
elif isinstance(e, OSError):
LOGGER.error("An OS error occurred while retrieving the username.")
LOGGER.error('An OS error occurred while retrieving the username.')
else:
LOGGER.error("An exception occurred:", e)
LOGGER.error('An exception occurred:', e)
return user

def _get_device_status(self, module):
Expand Down
17 changes: 8 additions & 9 deletions framework/python/src/test_orc/test_orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,14 @@
# limitations under the License.

"""Provides high level management of the test orchestrator."""
import getpass
import os
import json
import time
import shutil
import docker
from docker.types import Mount
from common import logger
from common import logger, util
from test_orc.module import TestModule
from common import util

LOG_NAME = "test_orc"
LOGGER = logger.get_logger("test_orc")
Expand Down Expand Up @@ -61,7 +59,7 @@ def start(self):
# Setup the output directory
self._host_user = util.get_host_user()
os.makedirs(RUNTIME_DIR, exist_ok=True)
util.run_command(f'chown -R {self._host_user} {RUNTIME_DIR}')
util.run_command(f"chown -R {self._host_user} {RUNTIME_DIR}")

self._load_test_modules()
self.build_test_modules()
Expand Down Expand Up @@ -102,15 +100,15 @@ def _generate_results(self, device):
results[module.name] = module_results
except (FileNotFoundError, PermissionError,
json.JSONDecodeError) as results_error:
LOGGER.error("Error occured whilst obbtaining results for module " + module.name)
LOGGER.error(f"Error occured whilst obbtaining results for module {module.name}")
LOGGER.debug(results_error)

out_file = os.path.join(
self._root_path,
"runtime/test/" + device.mac_addr.replace(":", "") + "/results.json")
with open(out_file, "w", encoding="utf-8") as f:
json.dump(results, f, indent=2)
util.run_command(f'chown -R {self._host_user} {out_file}')
util.run_command(f"chown -R {self._host_user} {out_file}")
return results

def test_in_progress(self):
Expand Down Expand Up @@ -140,18 +138,19 @@ def _run_test_module(self, module, device):
container_runtime_dir = os.path.join(
self._root_path, "runtime/test/" + device.mac_addr.replace(":", "") +
"/" + module.name)
network_runtime_dir = os.path.join(self._root_path, "runtime/network")
os.makedirs(container_runtime_dir)

network_runtime_dir = os.path.join(self._root_path, "runtime/network")

device_startup_capture = os.path.join(
self._root_path, "runtime/test/" + device.mac_addr.replace(":", "") +
"/startup.pcap")
util.run_command(f'chown -R {self._host_user} {device_startup_capture}')
util.run_command(f"chown -R {self._host_user} {device_startup_capture}")

device_monitor_capture = os.path.join(
self._root_path, "runtime/test/" + device.mac_addr.replace(":", "") +
"/monitor.pcap")
util.run_command(f'chown -R {self._host_user} {device_monitor_capture}')
util.run_command(f"chown -R {self._host_user} {device_monitor_capture}")

client = docker.from_env()

Expand Down
47 changes: 29 additions & 18 deletions modules/network/dhcp-1/python/src/grpc_server/dhcp_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,7 @@

LOG_NAME = 'dhcp_config'
LOGGER = None

CONFIG_FILE = '/etc/dhcp/dhcpd.conf'

DEFAULT_LEASE_TIME_KEY = 'default-lease-time'


Expand Down Expand Up @@ -186,13 +184,18 @@ def __str__(self):
config += '\tprimary;' if self.primary else 'secondary;'
config += '\n\t{ADDRESS_KEY} {ADDRESS};' if self.address is not None else ''
config += '\n\t{PORT_KEY} {PORT};' if self.port is not None else ''
config += '\n\t{PEER_ADDRESS_KEY} {PEER_ADDRESS};' if self.peer_address is not None else ''
config += '\n\t{PEER_PORT_KEY} {PEER_PORT};' if self.peer_port is not None else ''
config += '\n\t{MAX_RESPONSE_DELAY_KEY} {MAX_RESPONSE_DELAY};' if self.max_response_delay is not None else ''
config += '\n\t{MAX_UNACKED_UPDATES_KEY} {MAX_UNACKED_UPDATES};' if self.max_unacked_updates is not None else ''
config += ('\n\t{PEER_ADDRESS_KEY} {PEER_ADDRESS};'
if self.peer_address is not None else '')
config += ('\n\t{PEER_PORT_KEY} {PEER_PORT};'
if self.peer_port is not None else '')
config += ('\n\t{MAX_RESPONSE_DELAY_KEY} {MAX_RESPONSE_DELAY};'
if self.max_response_delay is not None else '')
config += ('\n\t{MAX_UNACKED_UPDATES_KEY} {MAX_UNACKED_UPDATES};'
if self.max_unacked_updates is not None else '')
config += '\n\t{MCLT_KEY} {MCLT};' if self.mclt is not None else ''
config += '\n\t{SPLIT_KEY} {SPLIT};' if self.split is not None else ''
config += '\n\t{LOAD_BALANCE_MAX_SECONDS_KEY} {LOAD_BALANCE_MAX_SECONDS};' if self.load_balance_max_seconds is not None else ''
config += ('\n\t{LOAD_BALANCE_MAX_SECONDS_KEY} {LOAD_BALANCE_MAX_SECONDS};'
if self.load_balance_max_seconds is not None else '')
config += '\n\r}}'

config = config.format(
Expand Down Expand Up @@ -220,9 +223,9 @@ def __str__(self):

if not self.enabled:
lines = config.strip().split('\n')
for i in range(len(lines)-1):
for i in range(len(lines) - 1):
lines[i] = '#' + lines[i]
lines[-1] = '#' + lines[-1].strip() # Handle the last line separately
lines[-1] = '#' + lines[-1].strip() # Handle the last line separately
config = '\n'.join(lines)

return config
Expand Down Expand Up @@ -302,15 +305,20 @@ def __init__(self, subnet):

def __str__(self):
config = 'subnet {SUBNET_OPTION} netmask {SUBNET_MASK_OPTION} {{'
config += '\n\t{NTP_OPTION_KEY} {NTP_OPTION};' if self._ntp_servers is not None else ''
config += '\n\t{SUBNET_MASK_OPTION_KEY} {SUBNET_MASK_OPTION};' if self._subnet_mask is not None else ''
config += '\n\t{BROADCAST_OPTION_KEY} {BROADCAST_OPTION};' if self._broadcast is not None else ''
config += '\n\t{ROUTER_OPTION_KEY} {ROUTER_OPTION};' if self._routers is not None else ''
config += '\n\t{DNS_OPTION_KEY} {DNS_OPTION};' if self._dns_servers is not None else ''
config += '\n\t{INTERFACE_KEY} {INTERFACE_OPTION};' if self._interface is not None else ''
config += ('\n\t{NTP_OPTION_KEY} {NTP_OPTION};'
if self._ntp_servers is not None else '')
config += ('\n\t{SUBNET_MASK_OPTION_KEY} {SUBNET_MASK_OPTION};'
if self._subnet_mask is not None else '')
config += ('\n\t{BROADCAST_OPTION_KEY} {BROADCAST_OPTION};'
if self._broadcast is not None else '')
config += ('\n\t{ROUTER_OPTION_KEY} {ROUTER_OPTION};'
if self._routers is not None else '')
config += ('\n\t{DNS_OPTION_KEY} {DNS_OPTION};'
if self._dns_servers is not None else '')
config += ('\n\t{INTERFACE_KEY} {INTERFACE_OPTION};'
if self._interface is not None else '')
config += '\n\t{AUTHORITATIVE_KEY};' if self._authoritative else ''


config = config.format(length='multi-line',
SUBNET_OPTION=self._subnet,
NTP_OPTION_KEY=NTP_OPTION_KEY,
Expand Down Expand Up @@ -407,8 +415,11 @@ def __init__(self, pool):

def __str__(self):
config = 'pool {{'
config += '\n\t\t{FAILOVER_KEY} "{FAILOVER}";' if self.failover_peer is not None else ''
config += '\n\t\t{RANGE_KEY} {RANGE_START} {RANGE_END};' if self.range_start is not None and self.range_end is not None else ''
config += ('\n\t\t{FAILOVER_KEY} "{FAILOVER}";'
if self.failover_peer is not None else '')
config += ('\n\t\t{RANGE_KEY} {RANGE_START} {RANGE_END};'
if self.range_start is not None and self.range_end is not None
else '')
config += '\n\t}}'

config = config.format(
Expand Down
150 changes: 76 additions & 74 deletions modules/network/dhcp-1/python/src/grpc_server/dhcp_config_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,87 +17,89 @@
import os

CONFIG_FILE = 'conf/dhcpd.conf'

DHCP_CONFIG = None

def get_config_file_path():
dhcp_config = DHCPConfig()
current_dir = os.path.dirname(os.path.abspath(__file__))
module_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(current_dir))))
conf_file = os.path.join(module_dir,CONFIG_FILE)
return conf_file
current_dir = os.path.dirname(os.path.abspath(__file__))
module_dir = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.abspath(current_dir))))
conf_file = os.path.join(module_dir, CONFIG_FILE)
return conf_file


def get_config():
dhcp_config = DHCPConfig()
dhcp_config.resolve_config(get_config_file_path())
return dhcp_config
dhcp_config = DHCPConfig()
dhcp_config.resolve_config(get_config_file_path())
return dhcp_config


class DHCPConfigTest(unittest.TestCase):

@classmethod
def setUpClass(cls):
# Resolve the config
global DHCP_CONFIG
DHCP_CONFIG = get_config()

def test_resolve_config(self):
print('Test Resolve Config:\n' + str(DHCP_CONFIG))

# Resolve the raw config file
with open(get_config_file_path(),'r') as f:
lines = f.readlines()

# Get the resolved config as a
conf_parts = str(DHCP_CONFIG).split('\n')

# dhcpd conf is not picky about spacing so we just
# need to check contents of each line for matching
# to make sure evertying matches
for i in range(len(lines)):
self.assertEqual(lines[i].strip(),conf_parts[i].strip())

def test_disable_failover(self):
DHCP_CONFIG.disable_failover()
print('Test Disable Config:\n' + str(DHCP_CONFIG))
config_lines = str(DHCP_CONFIG._peer).split('\n')
for line in config_lines:
self.assertTrue(line.startswith('#'))

def test_enable_failover(self):
DHCP_CONFIG.enable_failover()
print('Test Enable Config:\n' + str(DHCP_CONFIG))
config_lines = str(DHCP_CONFIG._peer).split('\n')
for line in config_lines:
self.assertFalse(line.startswith('#'))

def test_add_reserved_host(self):
DHCP_CONFIG.add_reserved_host('test','00:11:22:33:44:55','192.168.10.5')
host = DHCP_CONFIG.get_reserved_host('00:11:22:33:44:55')
self.assertIsNotNone(host)
print('AddHostConfig:\n' + str(DHCP_CONFIG))

def test_delete_reserved_host(self):
DHCP_CONFIG.delete_reserved_host('00:11:22:33:44:55')
host = DHCP_CONFIG.get_reserved_host('00:11:22:33:44:55')
self.assertIsNone(host)
print('DeleteHostConfig:\n' + str(DHCP_CONFIG))

def test_resolve_config_with_hosts(self):
DHCP_CONFIG.add_reserved_host('test','00:11:22:33:44:55','192.168.10.5')
config_with_hosts = DHCPConfig()
config_with_hosts.make(str(DHCP_CONFIG))
host = config_with_hosts.get_reserved_host('00:11:22:33:44:55')
self.assertIsNotNone(host)
print("ResolveConfigWithHosts:\n" + str(config_with_hosts))
@classmethod
def setUpClass(cls):
# Resolve the config
global DHCP_CONFIG
DHCP_CONFIG = get_config()

def test_resolve_config(self):
print('Test Resolve Config:\n' + str(DHCP_CONFIG))

# Resolve the raw config file
with open(get_config_file_path(), 'r', encoding='UTF-8') as f:
lines = f.readlines()

# Get the resolved config as a
conf_parts = str(DHCP_CONFIG).split('\n')

# dhcpd conf is not picky about spacing so we just
# need to check contents of each line for matching
# to make sure evertying matches
for i in range(len(lines)):
self.assertEqual(lines[i].strip(), conf_parts[i].strip())

def test_disable_failover(self):
DHCP_CONFIG.disable_failover()
print('Test Disable Config:\n' + str(DHCP_CONFIG))
config_lines = str(DHCP_CONFIG._peer).split('\n')
for line in config_lines:
self.assertTrue(line.startswith('#'))

def test_enable_failover(self):
DHCP_CONFIG.enable_failover()
print('Test Enable Config:\n' + str(DHCP_CONFIG))
config_lines = str(DHCP_CONFIG._peer).split('\n')
for line in config_lines:
self.assertFalse(line.startswith('#'))

def test_add_reserved_host(self):
DHCP_CONFIG.add_reserved_host('test', '00:11:22:33:44:55', '192.168.10.5')
host = DHCP_CONFIG.get_reserved_host('00:11:22:33:44:55')
self.assertIsNotNone(host)
print('AddHostConfig:\n' + str(DHCP_CONFIG))

def test_delete_reserved_host(self):
DHCP_CONFIG.delete_reserved_host('00:11:22:33:44:55')
host = DHCP_CONFIG.get_reserved_host('00:11:22:33:44:55')
self.assertIsNone(host)
print('DeleteHostConfig:\n' + str(DHCP_CONFIG))

def test_resolve_config_with_hosts(self):
DHCP_CONFIG.add_reserved_host('test', '00:11:22:33:44:55', '192.168.10.5')
config_with_hosts = DHCPConfig()
config_with_hosts.make(str(DHCP_CONFIG))
host = config_with_hosts.get_reserved_host('00:11:22:33:44:55')
self.assertIsNotNone(host)
print('ResolveConfigWithHosts:\n' + str(config_with_hosts))


if __name__ == '__main__':
suite = unittest.TestSuite()
suite.addTest(DHCPConfigTest('test_resolve_config'))
suite.addTest(DHCPConfigTest('test_disable_failover'))
suite.addTest(DHCPConfigTest('test_enable_failover'))
suite.addTest(DHCPConfigTest('test_add_reserved_host'))
suite.addTest(DHCPConfigTest('test_delete_reserved_host'))
suite.addTest(DHCPConfigTest('test_resolve_config_with_hosts'))

runner = unittest.TextTestRunner()
runner.run(suite)
suite = unittest.TestSuite()
suite.addTest(DHCPConfigTest('test_resolve_config'))
suite.addTest(DHCPConfigTest('test_disable_failover'))
suite.addTest(DHCPConfigTest('test_enable_failover'))
suite.addTest(DHCPConfigTest('test_add_reserved_host'))
suite.addTest(DHCPConfigTest('test_delete_reserved_host'))
suite.addTest(DHCPConfigTest('test_resolve_config_with_hosts'))

runner = unittest.TextTestRunner()
runner.run(suite)
Loading