Merge pull request #20 from KatharaFramework/develop

Version 0.1.7

README.md

@@ -88,9 +88,10 @@ In the following you will find the possible values for the configuration file.
     },
     "protocols": { # Checks on routing protocols
       "bgpd": { # Specific checks for BGP
-        "peerings": { # Check that a peering is up
+        "neighbors": { # Check that a peering is up
           "<device_name>": [
-            "<neighbour_ip>", # Check that a peering is up between the device and the specified neighbour ip
+            {"ip": <neighbour_ip>, "asn": <neighbor_asn>}, # Check that a peering is up between the device and 
+                                                           # the specified neighbour ip
           ],
         },
         "networks": {

examples/palabra/correction.json

@@ -171,32 +171,32 @@
 		},
 		"protocols": {
 			"bgpd": {
-				"peerings": {
+				"neighbors": {
 					"as1r1": [
-						"1.0.0.2",
-						"10.20.0.2"
+						{"ip": "1.0.0.2", "asn":  1},
+						{"ip": "10.20.0.2", "asn":  2}
 					],
 					"as1r2": [
-						"1.0.0.1",
-						"10.20.1.2"
+						{"ip": "1.0.0.1", "asn":  1},
+						{"ip": "10.20.1.2", "asn":  2}
 					],
 					"as2r1": [
-						"2.0.0.2",
-						"10.20.0.1",
-						"20.30.0.2"
+						{"ip": "2.0.0.2", "asn":  2},
+						{"ip": "10.20.0.1", "asn":  1},
+						{"ip": "20.30.0.2", "asn":  3}
 					],
 					"as2r2": [
-						"2.0.0.1",
-						"10.20.1.1",
-						"20.30.1.2"
+						{"ip": "2.0.0.1", "asn":  2},
+						{"ip": "10.20.1.1", "asn":  1},
+						{"ip": "20.30.1.2", "asn":  3}
 					],
 					"as3r1": [
-						"3.0.0.2",
-						"20.30.0.1"
+						{"ip": "3.0.0.2", "asn":  3},
+						{"ip": "20.30.0.1", "asn":  2}
 					],
 					"as3r2": [
-						"3.0.0.1",
-						"20.30.1.1"
+						{"ip": "3.0.0.1", "asn":  3},
+						{"ip": "20.30.1.1", "asn":  2}
 					]
 				},
 				"networks": {

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "kathara_lab_checker"
-    version = "0.1.6"
+    version = "0.1.7"
 description = "Tool to automatically check Kathará network scenarios based on a configuration file."
 readme = "README.md"
 requires-python = ">=3.11"
@@ -42,4 +42,10 @@ kathara_lab_checker = "kathara_lab_checker.__main__:main"
 
 [build-system]
 requires = ["setuptools>=61.2", "wheel"]
-build-backend = "setuptools.build_meta"
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+where = ["src/kathara_lab_checker"]
+
+[tool.setuptools.package-data]
+schemas = ["*.json"]

requirements.txt

@@ -2,4 +2,5 @@ kathara>=3.7.7
 jc>=1.24.0
 openpyxl>=3.1.2
 tqdm>=4
-coloredlogs~=15.0.1
+coloredlogs~=15.0.1
+jsonschema>=4.23.0

src/kathara_lab_checker/TestCollector.py

@@ -1,13 +1,16 @@
+from .model.CheckResult import CheckResult
+
+
 class TestCollector:
     def __init__(self):
         self.tests = {}
 
-    def add_check_result(self, lab_name: str, check_result: 'CheckResultPackage.CheckResult') -> None:
+    def add_check_result(self, lab_name: str, check_result: CheckResult) -> None:
         if lab_name not in self.tests:
             self.tests[lab_name] = []
         self.tests[lab_name].append(check_result)
 
-    def add_check_results(self, lab_name: str, check_results: list['CheckResultPackage.CheckResult']) -> None:
+    def add_check_results(self, lab_name: str, check_results: list[CheckResult]) -> None:
         for result in check_results:
             self.add_check_result(lab_name, result)
 

src/kathara_lab_checker/__main__.py

@@ -18,7 +18,6 @@
 
 from .TestCollector import TestCollector
 from .checks.BridgeCheck import BridgeCheck
-from .checks.CheckResult import CheckResult
 from .checks.CollisionDomainCheck import CollisionDomainCheck
 from .checks.CustomCommandCheck import CustomCommandCheck
 from .checks.DaemonCheck import DaemonCheck
@@ -34,14 +33,16 @@
 from .checks.applications.dns.LocalNSCheck import LocalNSCheck
 from .checks.protocols.AnnouncedNetworkCheck import AnnouncedNetworkCheck
 from .checks.protocols.ProtocolRedistributionCheck import ProtocolRedistributionCheck
-from .checks.protocols.bgp.BGPPeeringCheck import BGPPeeringCheck
+from .checks.protocols.bgp.BGPNeighborCheck import BGPNeighborCheck
+from .checks.protocols.bgp.BGPRoutesCheck import BGPRoutesCheck
 from .checks.protocols.evpn.AnnouncedVNICheck import AnnouncedVNICheck
 from .checks.protocols.evpn.EVPNSessionCheck import EVPNSessionCheck
 from .checks.protocols.evpn.VTEPCheck import VTEPCheck
 from .excel_utils import write_final_results_to_excel, write_result_to_excel
+from .model.CheckResult import CheckResult
 from .utils import reverse_dictionary
 
-VERSION = "0.1.6"
+VERSION = "0.1.7"
 CURRENT_LAB: Optional[Lab] = None
 
 
@@ -116,59 +117,66 @@ def run_on_single_network_scenario(
     logger.info(f"Verifying lab structure using lab.conf template in: {configuration['structure']}")
 
     logger.info("Checking that all devices exist...")
-    check_results = DeviceExistenceCheck().run(list(lab_template.machines.keys()), lab)
+    check_results = DeviceExistenceCheck(lab).run(list(lab_template.machines.keys()))
     test_collector.add_check_results(lab_name, check_results)
 
     logger.info("Checking collision domains...")
-    check_results = CollisionDomainCheck().run(list(lab_template.machines.values()), lab)
+    check_results = CollisionDomainCheck(lab).run(list(lab_template.machines.values()))
     test_collector.add_check_results(lab_name, check_results)
 
     if "requiring_startup" in configuration["test"]:
         logger.info("Checking that all required startup files exist...")
-        check_results = StartupExistenceCheck().run(configuration["test"]["requiring_startup"], lab)
+        check_results = StartupExistenceCheck(lab).run(configuration["test"]["requiring_startup"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "ipv6_enabled" in configuration["test"]:
         logger.info(f"Checking that IPv6 is enabled on devices: {configuration['test']['ipv6_enabled']}")
-        check_results = IPv6EnabledCheck().run(configuration["test"]["ipv6_enabled"], lab)
+        check_results = IPv6EnabledCheck(lab).run(configuration["test"]["ipv6_enabled"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "sysctls" in configuration["test"]:
         logger.info(f"Checking sysctl configurations on devices...")
-        check_results = SysctlCheck().run(configuration["test"]["sysctls"], lab)
+        check_results = SysctlCheck(lab).run(configuration["test"]["sysctls"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "ip_mapping" in configuration["test"]:
         logger.info("Verifying the IP addresses assigned to devices...")
-        check_results = InterfaceIPCheck().run(configuration["test"]["ip_mapping"], lab)
+        check_results = InterfaceIPCheck(lab).run(configuration["test"]["ip_mapping"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "bridges" in configuration["test"]:
         logger.info("Verifying the bridges inside devices...")
-        check_results = BridgeCheck().run(configuration["test"]["bridges"], lab)
+        check_results = BridgeCheck(lab).run(configuration["test"]["bridges"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "reachability" in configuration["test"]:
         logger.info(f"Starting reachability test...")
-        check_results = ReachabilityCheck().run(configuration["test"]["reachability"], lab)
+        check_results = ReachabilityCheck(lab).run(configuration["test"]["reachability"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "daemons" in configuration["test"]:
         logger.info(f"Checking if daemons are running...")
-        check_results = DaemonCheck().run(configuration["test"]["daemons"], lab)
+        check_results = DaemonCheck(lab).run(configuration["test"]["daemons"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "protocols" in configuration["test"]:
         logger.info("Checking routing daemons configurations...")
         for daemon_name, daemon_test in configuration["test"]["protocols"].items():
             if daemon_name == "bgpd":
                 logger.info(f"Check BGP peerings configurations...")
-                check_results = BGPPeeringCheck().run(daemon_test["peerings"], lab)
-                test_collector.add_check_results(lab_name, check_results)
+
+                if "neighbors" in daemon_test:
+                    check_results = BGPNeighborCheck(lab).run(daemon_test["neighbors"])
+                    test_collector.add_check_results(lab_name, check_results)
 
                 if "networks" in daemon_test:
                     logger.info(f"Checking BGP announces...")
-                    check_results = AnnouncedNetworkCheck().run(daemon_name, daemon_test["networks"], lab)
+                    check_results = AnnouncedNetworkCheck(lab).run(daemon_name, daemon_test["networks"])
+                    test_collector.add_check_results(lab_name, check_results)
+
+                if "routes" in daemon_test:
+                    logger.info(f"Checking BGP Routes...")
+                    check_results = BGPRoutesCheck(lab).run(daemon_test["routes"])
                     test_collector.add_check_results(lab_name, check_results)
 
                 if "evpn" in daemon_test:
@@ -177,58 +185,57 @@ def run_on_single_network_scenario(
                     for test in evpn_test:
                         if "evpn_sessions" in test:
                             logger.info(f"Checking EVPN session configuration...")
-                            check_results = EVPNSessionCheck().run(evpn_test["evpn_sessions"], lab)
+                            check_results = EVPNSessionCheck(lab).run(evpn_test["evpn_sessions"])
                             test_collector.add_check_results(lab_name, check_results)
 
                         if "vtep_devices" in test:
                             logger.info(f"Checking VTEP devices configuration...")
-                            check_results = VTEPCheck().run(evpn_test["vtep_devices"], lab)
+                            check_results = VTEPCheck(lab).run(evpn_test["vtep_devices"])
                             test_collector.add_check_results(lab_name, check_results)
 
                             logger.info(f"Checking BGP VNIs configurations...")
-                            check_results = AnnouncedVNICheck().run(
-                                evpn_test["vtep_devices"], evpn_test["evpn_sessions"], lab
+                            check_results = AnnouncedVNICheck(lab).run(
+                                evpn_test["vtep_devices"], evpn_test["evpn_sessions"]
                             )
                             test_collector.add_check_results(lab_name, check_results)
 
             if "injections" in daemon_test:
                 logger.info(f"Checking {daemon_name} protocols redistributions...")
-                check_results = ProtocolRedistributionCheck().run(daemon_name, daemon_test["injections"], lab)
+                check_results = ProtocolRedistributionCheck(lab).run(daemon_name, daemon_test["injections"])
                 test_collector.add_check_results(lab_name, check_results)
 
     if "kernel_routes" in configuration["test"]:
         logger.info(f"Checking Routing Tables...")
-        check_results = KernelRouteCheck().run(configuration["test"]["kernel_routes"], lab)
+        check_results = KernelRouteCheck(lab).run(configuration["test"]["kernel_routes"])
         test_collector.add_check_results(lab_name, check_results)
 
     if "applications" in configuration["test"]:
         for application_name, application in configuration["test"]["applications"].items():
             if application_name == "dns":
                 if "authoritative" in application:
                     logger.info("Checking DNS configurations...")
-                    check_results = DNSAuthorityCheck().run(
+                    check_results = DNSAuthorityCheck(lab).run(
                         application["authoritative"],
                         list(application["local_ns"].keys()),
                         configuration["test"]["ip_mapping"],
-                        lab,
                     )
                     test_collector.add_check_results(lab_name, check_results)
 
                 if "local_ns" in application:
                     logger.info("Checking local name servers configurations...")
-                    check_results = LocalNSCheck().run(application["local_ns"], lab)
+                    check_results = LocalNSCheck(lab).run(application["local_ns"])
                     test_collector.add_check_results(lab_name, check_results)
 
                 if "records" in application:
                     logger.info(f"Starting test for DNS records...")
-                    check_results = DNSRecordCheck().run(
-                        application["records"], reverse_dictionary(application["local_ns"]).keys(), lab
+                    check_results = DNSRecordCheck(lab).run(
+                        application["records"], reverse_dictionary(application["local_ns"]).keys()
                     )
                     test_collector.add_check_results(lab_name, check_results)
 
     if "custom_commands" in configuration["test"]:
         logger.info("Checking custom commands output...")
-        check_results = CustomCommandCheck().run(configuration["test"]["custom_commands"], lab)
+        check_results = CustomCommandCheck(lab).run(configuration["test"]["custom_commands"])
         test_collector.add_check_results(lab_name, check_results)
 
     if not live and not keep_open:
@@ -269,7 +276,8 @@ def run_on_multiple_network_scenarios(
                         lambda x: os.path.isdir(os.path.join(labs_path, x)) and x != ".DS_Store",
                         os.listdir(labs_path),
                     )
-                )
+                ),
+                key=str.casefold,
             )
     ):
         test_results = run_on_single_network_scenario(
@@ -288,7 +296,7 @@ def parse_arguments():
     parser = argparse.ArgumentParser(
         description="A tool for automatically check Kathará network scenarios",
         prog="kathara_lab_checker",
-        add_help=True
+        add_help=True,
     )
 
     parser.add_argument(
@@ -355,7 +363,6 @@ def main():
     signal.signal(signal.SIGINT, partial(handler, live=args.live))
 
     logger = logging.getLogger("kathara-lab-checker")
-    # logger.addHandler(TqdmLoggingHandler())
 
     coloredlogs.install(fmt="%(message)s", level="INFO", logger=logger)
 

src/kathara_lab_checker/checks/AbstractCheck.py

@@ -1,9 +1,15 @@
 import logging
 from abc import ABC
+from Kathara.manager.Kathara import Kathara
+from Kathara.model.Lab import Lab
 
 
 class AbstractCheck(ABC):
 
-    def __init__(self, description: str = None):
+    __slots__ = ["description", "logger", "kathara_manager", "lab"]
+
+    def __init__(self, lab: Lab, description: str = None):
         self.description: str = description
         self.logger = logging.getLogger("kathara-lab-checker")
+        self.kathara_manager: Kathara = Kathara.get_instance()
+        self.lab: Lab = lab