diff --git a/src/csv_utils.py b/src/csv_utils.py new file mode 100644 index 0000000..b27d5ec --- /dev/null +++ b/src/csv_utils.py @@ -0,0 +1,20 @@ +import csv + + +def convert_csv_to_json(file_path): + csv_rows = [] + with open(file_path, encoding='utf-8', errors='ignore') as csvfile: + reader = csv.DictReader(csvfile, skipinitialspace=True, quotechar='"') + + # Here we clean any Non-breaking spaces and convert to normal spacing. \xa0 converts to ' ' + for row in reader: + temp_overwritable_dict = {} + for key, value in row.items(): + cleaned_key = key.replace('\xa0', ' ') + cleaned_value = value.replace('\xa0', ' ') + temp_overwritable_dict.update({cleaned_key: cleaned_value}) + csv_rows.append(temp_overwritable_dict) + + if csv_rows == None or csv_rows == []: + raise ValueError('Failed to convert CSV file to JSON. Exiting script.') + return csv_rows diff --git a/src/juniper_ap_staging.py b/src/juniper_ap_staging.py index 081e988..8f00c5f 100644 --- a/src/juniper_ap_staging.py +++ b/src/juniper_ap_staging.py @@ -23,17 +23,19 @@ def validate_if_ap_defined_on_csv_exists_in_mist(mist_inventory: list['dict'], ap_mac_addresses_from_mist = [] for inventory_item in mist_inventory: if inventory_item['type'] == 'ap': - ap_mac_addresses_from_mist.append(inventory_item['mac']) + ap_mac_addresses_from_mist.append(inventory_item['mac'].upper()) ap_mac_addresses_from_csv = [] for ap_row in ap_csv: cleaned_mac_address = ap_row['MAC Address'].replace(":", "") ap_mac_addresses_from_csv.append(cleaned_mac_address) - for site_name in ap_mac_addresses_from_csv: - if site_name not in ap_mac_addresses_from_mist: + for ap_mac_address in ap_mac_addresses_from_csv: + if ap_mac_address in ap_mac_addresses_from_mist: + continue + else: raise ValueError( - f"Site name '{site_name}' found in CSV file but not in mist site configurations.") + f"AP Mac address '{ap_mac_address}' found in CSV file but not on mist inventory.") def find_site_by_name(site_name: str, diff --git a/src/main.py b/src/main.py index e739dda..c2f0d6b 100644 --- a/src/main.py +++ b/src/main.py @@ -1,35 +1,13 @@ from juniper_site_creation import juniper_script_site_creation from juniper_ap_staging import juniper_ap_assign import os -import csv - - -# Convert CSV file to JSON object. -def convert_csv_to_json(file_path): - csv_rows = [] - with open(file_path, encoding="ISO-8859-1") as csvfile: - reader = csv.DictReader(csvfile, skipinitialspace=True, quotechar='"') - - # Here we clean any Non-breaking spaces and convert to normal spacing. \xa0 converts to ' ' - for row in reader: - temp_overwritable_dict = {} - for key, value in row.items(): - cleaned_key = key.replace('\xa0', ' ') - cleaned_value = value.replace('\xa0', ' ') - temp_overwritable_dict.update({cleaned_key: cleaned_value}) - csv_rows.append(temp_overwritable_dict) - - if csv_rows == None or csv_rows == []: - raise ValueError('Failed to convert CSV file to JSON. Exiting script.') - return csv_rows - +import csv_utils if __name__ == '__main__': csv_file_path = os.getcwd() + '/../data_src/' + os.environ.get('CSV_FILE_NAME') - # Convert CSV to valid JSON - json_data = convert_csv_to_json(csv_file_path) + json_data = csv_utils.convert_csv_to_json(csv_file_path) operation = os.environ.get('OPERATION') diff --git a/test/test_csv_utils.py b/test/test_csv_utils.py new file mode 100644 index 0000000..55c00d3 --- /dev/null +++ b/test/test_csv_utils.py @@ -0,0 +1,161 @@ +from src.csv_utils import convert_csv_to_json +import unittest +import tempfile +import csv + + +class TestCsvToJsonSiteCreation(unittest.TestCase): + + def setUp(self): + # Create a temporary CSV file for testing + self.csv_data = [ + {'Site Name': 'Test location 1', 'Site Address': '40 Mayflower Dr, Plymouth PL2 3DG', + 'Enable GovWifi': ' "TRUE"', + 'Enable MoJWifi': ' "FALSE"', 'GovWifi Radius Key': '00000DD0000BC0EEE000', + 'Wired NACS Radius Key': '00000DD0000BC0EEE000'}, + {'Site Name': 'Test location 2', 'Site Address': '102 Petty France, London SW1H 9AJ', + 'Enable GovWifi': ' "TRUE"', + 'Enable MoJWifi': ' "FALSE"', 'GovWifi Radius Key': '0D0E0DDE000BC0EEE000', + 'Wired NACS Radius Key': '00000DD0000BC0EEE000'}, + {'Site Name': 'Test location 3', 'Site Address': 'Met Office, FitzRoy Road, Exeter, Devon, EX1 3PB', + 'Enable GovWifi': ' "TRUE"', + 'Enable MoJWifi': ' "FALSE"', 'GovWifi Radius Key': '0D0E0DDE080BC0EEE000', + 'Wired NACS Radius Key': '00000DD0000BC0EEE000'} + ] + self.csv_file = tempfile.NamedTemporaryFile( + mode='w', delete=False, newline='', suffix='.csv') + self.csv_writer = csv.DictWriter(self.csv_file, fieldnames=[ + 'Site Name', + 'Site Address', + 'Enable GovWifi', + 'Enable MoJWifi', + 'GovWifi Radius Key', + 'Wired NACS Radius Key' + ]) + self.csv_writer.writeheader() + self.csv_writer.writerows(self.csv_data) + self.csv_file.close() + + def test_convert_csv_to_json_valid_csv(self): + expected_json = self.csv_data + actual_json = convert_csv_to_json(self.csv_file.name) + self.assertEqual(actual_json, expected_json) + + def test_given_csv_file_when_csv_file_empty_then_raise_value_error(self): + empty_csv_file = tempfile.NamedTemporaryFile( + mode='w', delete=False, newline='', suffix='.csv') + empty_csv_file.close() + with self.assertRaises(ValueError) as error: + convert_csv_to_json(empty_csv_file.name) + self.assertEqual(str(error.exception), + 'Failed to convert CSV file to JSON. Exiting script.') + + def test_given_file_path_when_csv_file_not_found_then_raise_FileNotFoundError(self): + # Test if the function handles a nonexistent CSV file correctly + nonexistent_file_path = 'nonexistent.csv' + with self.assertRaises(FileNotFoundError): + convert_csv_to_json(nonexistent_file_path) + + def test_given_csv_while_when_csv_file_contains_nbsp_then_convert_to_normal_spacing(self): + csv_data = [ + {'Site\xa0Name': 'Test location 1', 'Site Address': '40\xa0Mayflower\xa0Dr, Plymouth PL2 3DG', + 'Enable GovWifi': ' "TRUE"', + 'Enable\xa0MoJWifi': ' "FALSE"', 'GovWifi\xa0Radius Key': '00000DD0000BC0EEE000', + 'Wired\xa0NACS Radius Key': '00000DD0000BC0EEE000'}, + {'Site\xa0Name': 'Test location 2', 'Site Address': '102\xa0Petty\xa0France, London SW1H 9AJ', + 'Enable GovWifi': ' "TRUE"', + 'Enable\xa0MoJWifi': ' "FALSE"', 'GovWifi\xa0Radius Key': '0D0E0DDE000BC0EEE000', + 'Wired\xa0NACS Radius Key': '00000DD0000BC0EEE000'}, + {'Site\xa0Name': 'Test location 3', + 'Site Address': 'Met\xa0Office, FitzRoy Road,\xa0Exeter, Devon, EX1 3PB', + 'Enable GovWifi': ' "TRUE"', + 'Enable\xa0MoJWifi': ' "FALSE"', 'GovWifi\xa0Radius Key': '0D0E0DDE080BC0EEE000', + 'Wired\xa0NACS Radius Key': '00000DD0000BC0EEE000'} + ] + csv_file = tempfile.NamedTemporaryFile( + mode='w', delete=False, newline='', suffix='.csv') + csv_writer = csv.DictWriter(csv_file, fieldnames=[ + 'Site\xa0Name', + 'Site Address', + 'Enable GovWifi', + 'Enable\xa0MoJWifi', + 'GovWifi\xa0Radius Key', + 'Wired\xa0NACS Radius Key' + ]) + csv_writer.writeheader() + csv_writer.writerows(csv_data) + csv_file.close() + + expected_json = self.csv_data + actual_json = convert_csv_to_json(csv_file.name) + self.assertEqual(expected_json, actual_json) + + +class TestCsvToJsonApOnBoaring(unittest.TestCase): + + def setUp(self): + # Create a temporary CSV file for testing + self.csv_data = [ + {'SITE FITS ID': 'FITS_0001', 'Site Name': 'MOJ-3231-HMP-Isis', 'Location': 'Ground Floor, KITCHEN', + 'Device Type': 'WAP', 'Make/Model': 'MIST AP32', 'Host Name': 'MOJ-AA-0000-WAP001', + 'Serial Number': 'A1234567899A1', 'MAC Address': 'fe925c207c90', 'Claim Code': 'AAAAA-AAAAA-1A1AA', + 'RF Report Reference': 'Isis - Skills & Workshop - 5.1. 00 Ground FloorÊ'}, + {'SITE FITS ID': 'FITS_0002', 'Site Name': 'MOJ-3231-HMP-Isis', 'Location': 'Ground Floor, WORKSHOP 2', + 'Device Type': 'WAP', 'Make/Model': 'MIST AP32', 'Host Name': 'MOJ-AA-0000-WAP002', + 'Serial Number': 'A1234567899A2', 'MAC Address': 'beae56460243', 'Claim Code': 'AAAAA-AAAAA-1A1AB', + 'RF Report Reference': 'Isis - Skills & Workshop - 5.1. 00 Ground FloorÊ'}, + {'SITE FITS ID': 'FITS_0003', 'Site Name': 'MOJ-3231-HMP-Isis', 'Location': 'Ground Floor, WORKSHOP 3', + 'Device Type': 'WAP', 'Make/Model': 'MIST AP32', 'Host Name': 'MOJ-AA-0000-WAP003', + 'Serial Number': 'A1234567899A3', 'MAC Address': '3a53ef88ae86', 'Claim Code': 'AAAAA-AAAAA-1A1AC', + 'RF Report Reference': 'Isis - Skills & Workshop - 5.1. 00 Ground FloorÊ'}] + self.csv_file = tempfile.NamedTemporaryFile( + mode='w', delete=False, newline='', suffix='.csv', encoding='iso-8859-1') + self.csv_writer = csv.DictWriter(self.csv_file, fieldnames=[ + 'SITE FITS ID', + 'Site Name', + 'Location', + 'Device Type', + 'Make/Model', + 'Host Name', + 'Serial Number', + 'MAC Address', + 'Claim Code', + 'RF Report Reference' + ]) + self.csv_writer.writeheader() + self.csv_writer.writerows(self.csv_data) + self.csv_file.close() + + def test_given_iso_8859_1_file_when_csv_file_is_opened_then_convert_to_json(self): + expected_json = [{'Claim Code': 'AAAAA-AAAAA-1A1AA', + 'Device Type': 'WAP', + 'Host Name': 'MOJ-AA-0000-WAP001', + 'Location': 'Ground Floor, KITCHEN', + 'MAC Address': 'fe925c207c90', + 'Make/Model': 'MIST AP32', + 'RF Report Reference': 'Isis - Skills & Workshop - 5.1. 00 Ground Floor', + 'SITE FITS ID': 'FITS_0001', + 'Serial Number': 'A1234567899A1', + 'Site Name': 'MOJ-3231-HMP-Isis'}, + {'Claim Code': 'AAAAA-AAAAA-1A1AB', + 'Device Type': 'WAP', + 'Host Name': 'MOJ-AA-0000-WAP002', + 'Location': 'Ground Floor, WORKSHOP 2', + 'MAC Address': 'beae56460243', + 'Make/Model': 'MIST AP32', + 'RF Report Reference': 'Isis - Skills & Workshop - 5.1. 00 Ground Floor', + 'SITE FITS ID': 'FITS_0002', + 'Serial Number': 'A1234567899A2', + 'Site Name': 'MOJ-3231-HMP-Isis'}, + {'Claim Code': 'AAAAA-AAAAA-1A1AC', + 'Device Type': 'WAP', + 'Host Name': 'MOJ-AA-0000-WAP003', + 'Location': 'Ground Floor, WORKSHOP 3', + 'MAC Address': '3a53ef88ae86', + 'Make/Model': 'MIST AP32', + 'RF Report Reference': 'Isis - Skills & Workshop - 5.1. 00 Ground Floor', + 'SITE FITS ID': 'FITS_0003', + 'Serial Number': 'A1234567899A3', + 'Site Name': 'MOJ-3231-HMP-Isis'}] + actual_json = convert_csv_to_json(self.csv_file.name) + self.assertEqual(expected_json, actual_json) diff --git a/test/test_main.py b/test/test_main.py deleted file mode 100644 index 9a63390..0000000 --- a/test/test_main.py +++ /dev/null @@ -1,90 +0,0 @@ -from src.main import convert_csv_to_json -import unittest -import tempfile -import csv - - -class TestCsvToJson(unittest.TestCase): - - def setUp(self): - # Create a temporary CSV file for testing - self.csv_data = [ - {'Site Name': 'Test location 1', 'Site Address': '40 Mayflower Dr, Plymouth PL2 3DG', - 'Enable GovWifi': ' "TRUE"', - 'Enable MoJWifi': ' "FALSE"', 'GovWifi Radius Key': '00000DD0000BC0EEE000', - 'Wired NACS Radius Key': '00000DD0000BC0EEE000'}, - {'Site Name': 'Test location 2', 'Site Address': '102 Petty France, London SW1H 9AJ', - 'Enable GovWifi': ' "TRUE"', - 'Enable MoJWifi': ' "FALSE"', 'GovWifi Radius Key': '0D0E0DDE000BC0EEE000', - 'Wired NACS Radius Key': '00000DD0000BC0EEE000'}, - {'Site Name': 'Test location 3', 'Site Address': 'Met Office, FitzRoy Road, Exeter, Devon, EX1 3PB', - 'Enable GovWifi': ' "TRUE"', - 'Enable MoJWifi': ' "FALSE"', 'GovWifi Radius Key': '0D0E0DDE080BC0EEE000', - 'Wired NACS Radius Key': '00000DD0000BC0EEE000'} - ] - self.csv_file = tempfile.NamedTemporaryFile( - mode='w', delete=False, newline='', suffix='.csv') - self.csv_writer = csv.DictWriter(self.csv_file, fieldnames=[ - 'Site Name', - 'Site Address', - 'Enable GovWifi', - 'Enable MoJWifi', - 'GovWifi Radius Key', - 'Wired NACS Radius Key' - ]) - self.csv_writer.writeheader() - self.csv_writer.writerows(self.csv_data) - self.csv_file.close() - - def test_convert_csv_to_json_valid_csv(self): - expected_json = self.csv_data - actual_json = convert_csv_to_json(self.csv_file.name) - self.assertEqual(actual_json, expected_json) - - def test_given_csv_file_when_csv_file_empty_then_raise_value_error(self): - empty_csv_file = tempfile.NamedTemporaryFile( - mode='w', delete=False, newline='', suffix='.csv') - empty_csv_file.close() - with self.assertRaises(ValueError) as error: - convert_csv_to_json(empty_csv_file.name) - self.assertEqual(str(error.exception), - 'Failed to convert CSV file to JSON. Exiting script.') - - def test_given_file_path_when_csv_file_not_found_then_raise_FileNotFoundError(self): - # Test if the function handles a nonexistent CSV file correctly - nonexistent_file_path = 'nonexistent.csv' - with self.assertRaises(FileNotFoundError): - convert_csv_to_json(nonexistent_file_path) - - def test_given_csv_while_when_csv_file_contains_nbsp_then_convert_to_normal_spacing(self): - csv_data = [ - {'Site\xa0Name': 'Test location 1', 'Site Address': '40\xa0Mayflower\xa0Dr, Plymouth PL2 3DG', - 'Enable GovWifi': ' "TRUE"', - 'Enable\xa0MoJWifi': ' "FALSE"', 'GovWifi\xa0Radius Key': '00000DD0000BC0EEE000', - 'Wired\xa0NACS Radius Key': '00000DD0000BC0EEE000'}, - {'Site\xa0Name': 'Test location 2', 'Site Address': '102\xa0Petty\xa0France, London SW1H 9AJ', - 'Enable GovWifi': ' "TRUE"', - 'Enable\xa0MoJWifi': ' "FALSE"', 'GovWifi\xa0Radius Key': '0D0E0DDE000BC0EEE000', - 'Wired\xa0NACS Radius Key': '00000DD0000BC0EEE000'}, - {'Site\xa0Name': 'Test location 3', 'Site Address': 'Met\xa0Office, FitzRoy Road,\xa0Exeter, Devon, EX1 3PB', - 'Enable GovWifi': ' "TRUE"', - 'Enable\xa0MoJWifi': ' "FALSE"', 'GovWifi\xa0Radius Key': '0D0E0DDE080BC0EEE000', - 'Wired\xa0NACS Radius Key': '00000DD0000BC0EEE000'} - ] - csv_file = tempfile.NamedTemporaryFile( - mode='w', delete=False, newline='', suffix='.csv') - csv_writer = csv.DictWriter(csv_file, fieldnames=[ - 'Site\xa0Name', - 'Site Address', - 'Enable GovWifi', - 'Enable\xa0MoJWifi', - 'GovWifi\xa0Radius Key', - 'Wired\xa0NACS Radius Key' - ]) - csv_writer.writeheader() - csv_writer.writerows(csv_data) - csv_file.close() - - expected_json = self.csv_data - actual_json = convert_csv_to_json(csv_file.name) - self.assertEqual(actual_json, expected_json)