Version 0.1.6

Version 0.1.6

.gitignore

@@ -96,15 +96,10 @@ celerybeat-schedule
 Dockerfile~
 *.py~
 
-# builds
-/src/kathara.build/
-/src/kathara.dist/
-
 # Packages
 *.rpm
 *.tar.zst
 
-
 /config
 /.vscode/
 /.vs/

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "kathara_lab_checker"
-    version = "0.1.5"
+    version = "0.1.6"
 description = "Tool to automatically check Kathará network scenarios based on a configuration file."
 readme = "README.md"
 requires-python = ">=3.11"

src/kathara_lab_checker/__main__.py

@@ -16,32 +16,32 @@
 from Kathara.setting.Setting import Setting
 from tqdm import tqdm
 
-from kathara_lab_checker.TestCollector import TestCollector
-from kathara_lab_checker.checks.BridgeCheck import BridgeCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.checks.CollisionDomainCheck import CollisionDomainCheck
-from kathara_lab_checker.checks.CustomCommandCheck import CustomCommandCheck
-from kathara_lab_checker.checks.DaemonCheck import DaemonCheck
-from kathara_lab_checker.checks.DeviceExistenceCheck import DeviceExistenceCheck
-from kathara_lab_checker.checks.IPv6EnabledCheck import IPv6EnabledCheck
-from kathara_lab_checker.checks.InterfaceIPCheck import InterfaceIPCheck
-from kathara_lab_checker.checks.KernelRouteCheck import KernelRouteCheck
-from kathara_lab_checker.checks.ReachabilityCheck import ReachabilityCheck
-from kathara_lab_checker.checks.StartupExistenceCheck import StartupExistenceCheck
-from kathara_lab_checker.checks.SysctlCheck import SysctlCheck
-from kathara_lab_checker.checks.applications.dns.DNSAuthorityCheck import DNSAuthorityCheck
-from kathara_lab_checker.checks.applications.dns.DNSRecordCheck import DNSRecordCheck
-from kathara_lab_checker.checks.applications.dns.LocalNSCheck import LocalNSCheck
-from kathara_lab_checker.checks.protocols.AnnouncedNetworkCheck import AnnouncedNetworkCheck
-from kathara_lab_checker.checks.protocols.ProtocolRedistributionCheck import ProtocolRedistributionCheck
-from kathara_lab_checker.checks.protocols.bgp.BGPPeeringCheck import BGPPeeringCheck
-from kathara_lab_checker.checks.protocols.evpn.AnnouncedVNICheck import AnnouncedVNICheck
-from kathara_lab_checker.checks.protocols.evpn.EVPNSessionCheck import EVPNSessionCheck
-from kathara_lab_checker.checks.protocols.evpn.VTEPCheck import VTEPCheck
-from kathara_lab_checker.excel_utils import write_final_results_to_excel, write_result_to_excel
-from kathara_lab_checker.utils import reverse_dictionary
-
-VERSION = "0.1.5"
+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
+from .checks.DeviceExistenceCheck import DeviceExistenceCheck
+from .checks.IPv6EnabledCheck import IPv6EnabledCheck
+from .checks.InterfaceIPCheck import InterfaceIPCheck
+from .checks.KernelRouteCheck import KernelRouteCheck
+from .checks.ReachabilityCheck import ReachabilityCheck
+from .checks.StartupExistenceCheck import StartupExistenceCheck
+from .checks.SysctlCheck import SysctlCheck
+from .checks.applications.dns.DNSAuthorityCheck import DNSAuthorityCheck
+from .checks.applications.dns.DNSRecordCheck import DNSRecordCheck
+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.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 .utils import reverse_dictionary
+
+VERSION = "0.1.6"
 CURRENT_LAB: Optional[Lab] = None
 
 
@@ -120,7 +120,7 @@ def run_on_single_network_scenario(
     test_collector.add_check_results(lab_name, check_results)
 
     logger.info("Checking collision domains...")
-    check_results = CollisionDomainCheck().run(list(lab_template.links.values()), lab)
+    check_results = CollisionDomainCheck().run(list(lab_template.machines.values()), lab)
     test_collector.add_check_results(lab_name, check_results)
 
     if "requiring_startup" in configuration["test"]:
@@ -263,10 +263,12 @@ def run_on_multiple_network_scenarios(
 
     test_collector = TestCollector()
     for lab_name in tqdm(
-            list(
-                filter(
-                    lambda x: os.path.isdir(os.path.join(labs_path, x)) and x != ".DS_Store",
-                    os.listdir(labs_path),
+            sorted(
+                list(
+                    filter(
+                        lambda x: os.path.isdir(os.path.join(labs_path, x)) and x != ".DS_Store",
+                        os.listdir(labs_path),
+                    )
                 )
             )
     ):

src/kathara_lab_checker/checks/CheckResult.py

@@ -1,6 +1,6 @@
 import logging
 
-from kathara_lab_checker.utils import green, red
+from ..utils import green, red
 
 
 class CheckResult:

src/kathara_lab_checker/checks/CollisionDomainCheck.py

@@ -1,31 +1,37 @@
 from Kathara.exceptions import LinkNotFoundError
-from Kathara.model import Link
 from Kathara.model.Lab import Lab
+from Kathara.model.Machine import Machine
 
 from .AbstractCheck import AbstractCheck
 from .CheckResult import CheckResult
 
 
 class CollisionDomainCheck(AbstractCheck):
-    def check(self, cd_t: Link, lab: Lab) -> CheckResult:
-        self.description = f"Checking collision domain `{cd_t.name}`"
+    def check(self, machine_t: Machine, lab: Lab) -> list[CheckResult]:
 
+        results = []
         try:
-            cd = lab.get_link(cd_t.name)
-            if cd.machines.keys() != cd_t.machines.keys():
-                reason = (
-                    f"Devices connected to collision domain {cd.name} {list(cd.machines.keys())} "
-                    f"are different from the one in the template {list(cd_t.machines.keys())}."
-                )
-                return CheckResult(self.description, False, reason)
+            machine = lab.get_machine(machine_t.name)
+            for iface_num, interface in machine.interfaces.items():
+                self.description = f"Checking the collision domain attached to interface `eth{iface_num}` of `{machine_t.name}`"
+                interface_t = machine_t.interfaces[iface_num]
+                if interface_t.link.name != interface.link.name:
+                    reason = (
+                        f"Interface `{iface_num}` of device {machine_t.name} is connected to collision domain "
+                        f"`{interface.link.name}` instead of `{interface_t.link.name}`"
+                    )
+                    results.append(CheckResult(self.description, False, reason))
+                else:
+                    results.append(CheckResult(self.description, True, "OK"))
+
 
-            return CheckResult(self.description, True, "OK")
         except LinkNotFoundError as e:
-            return CheckResult(self.description, False, str(e))
+            results.append(CheckResult(self.description, False, str(e)))
+        return results
 
-    def run(self, template_cds: list[Link], lab: Lab) -> list[CheckResult]:
+    def run(self, template_machines: list[Machine], lab: Lab) -> list[CheckResult]:
         results = []
-        for cd_t in template_cds:
-            check_result = self.check(cd_t, lab)
-            results.append(check_result)
+        for machine_t in template_machines:
+            check_result = self.check(machine_t, lab)
+            results.extend(check_result)
         return results

src/kathara_lab_checker/checks/DaemonCheck.py

@@ -2,7 +2,7 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.utils import get_output
+from ..utils import get_output
 from .AbstractCheck import AbstractCheck
 from .CheckResult import CheckResult
 

src/kathara_lab_checker/checks/InterfaceIPCheck.py

@@ -3,7 +3,7 @@
 from Kathara.exceptions import MachineNotRunningError
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.utils import get_interfaces_addresses
+from ..utils import get_interfaces_addresses
 from .AbstractCheck import AbstractCheck
 from .CheckResult import CheckResult
 

src/kathara_lab_checker/checks/ReachabilityCheck.py

@@ -2,7 +2,7 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.utils import get_output
+from ..utils import get_output
 from .AbstractCheck import AbstractCheck
 from .CheckResult import CheckResult
 

src/kathara_lab_checker/checks/applications/dns/DNSAuthorityCheck.py

@@ -6,9 +6,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output, find_lines_with_string, find_device_name_from_ip
+from ...AbstractCheck import AbstractCheck
+from ...CheckResult import CheckResult
+from ....utils import get_output, find_lines_with_string, find_device_name_from_ip
 
 
 class DNSAuthorityCheck(AbstractCheck):
@@ -52,11 +52,12 @@ def check(self, domain: str, authority_ip: str, device_name: str, device_ip: str
                 )
                 return CheckResult(self.description, False, reason)
         else:
-            if os.path.exists(f"{device_name}.startup"):
+            if lab.fs.exists(f"{device_name}.startup"):
                 with lab.fs.open(f"{device_name}.startup", "r") as startup_file:
-                    systemctl_lines = find_lines_with_string(startup_file.readline(), "systemctl")
+                    lines = startup_file.readlines()
 
-                for line in systemctl_lines:
+                for line in lines:
+                    line = line.strip()
                     if re.search(rf"^\s*systemctl\s*start\s*named\s*$", line):
                         exec_output_gen = kathara_manager.exec(
                             machine_name=device_name,
@@ -65,19 +66,22 @@ def check(self, domain: str, authority_ip: str, device_name: str, device_ip: str
                         )
 
                         output = get_output(exec_output_gen)
-
                         date_pattern = (
                             r"\d{2}-[Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec]{3}-\d{4} \d{2}:\d{2}:\d{2}\.\d{3}"
                         )
 
                         reason_list = find_lines_with_string(output, "could not")
+                        reason_list.extend(find_lines_with_string(output, "/etc/bind/named.conf"))
                         reason_list_no_dates = [re.sub(date_pattern, "", line) for line in reason_list]
                         reason = "\n".join(reason_list_no_dates)
 
-                        return CheckResult(self.description, False, reason)
+                        return CheckResult(self.description, False, "Configuration Error:\n" + reason)
 
-            reason = f"named not started in the startup file of `{device_name}`"
-            return CheckResult(self.description, False, reason)
+                reason = f"named not started in `{device_name}`.startup`"
+                return CheckResult(self.description, False, reason)
+            else:
+                reason = f"There is no `.startup` file for device `{device_name}`"
+                return CheckResult(self.description, False, reason)
 
     def run(
         self,

src/kathara_lab_checker/checks/applications/dns/DNSRecordCheck.py

@@ -1,9 +1,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ...AbstractCheck import AbstractCheck
+from ...CheckResult import CheckResult
+from ....utils import get_output
 
 
 class DNSRecordCheck(AbstractCheck):

src/kathara_lab_checker/checks/applications/dns/LocalNSCheck.py

@@ -3,9 +3,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ...AbstractCheck import AbstractCheck
+from ...CheckResult import CheckResult
+from ....utils import get_output
 
 
 class LocalNSCheck(AbstractCheck):
@@ -26,13 +26,18 @@ def check(self, local_ns_ip: str, device_name: str, lab: Lab) -> CheckResult:
         if not lines:
             reason = f"`resolv.conf` file not found for device `{device_name}`"
             return CheckResult(self.description, False, reason)
+        actual_ips = []
         for line in lines:
-            if re.search(rf"^nameserver {local_ns_ip}$", line):
-                return CheckResult(self.description, True, "OK")
-            else:
-                reason = f"The local name server for device `{device_name}` has ip `{local_ns_ip}`"
-                return CheckResult(self.description, False, reason)
-
+            match = re.search(rf"^nameserver (.*)$", line)
+            if match:
+                actual_ns_ip = match.group(1)
+                if actual_ns_ip == local_ns_ip:
+                    return CheckResult(self.description, True, "OK")
+                actual_ips.append(actual_ns_ip)
+
+        reason = (f"There is no local name server for device `{device_name}` with IP `{local_ns_ip}`. "
+                  f"Actual nameservers: {actual_ips}")
+        return CheckResult(self.description, False, reason)
     def run(self, local_nameservers_to_devices: dict[str, list[str]], lab: Lab) -> list[CheckResult]:
         results = []
         for local_ns, managed_devices in local_nameservers_to_devices.items():

src/kathara_lab_checker/checks/protocols/AnnouncedNetworkCheck.py

@@ -3,9 +3,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ..AbstractCheck import AbstractCheck
+from ..CheckResult import CheckResult
+from ...utils import get_output
 
 
 class AnnouncedNetworkCheck(AbstractCheck):

src/kathara_lab_checker/checks/protocols/ProtocolRedistributionCheck.py

@@ -3,9 +3,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ..AbstractCheck import AbstractCheck
+from ..CheckResult import CheckResult
+from ...utils import get_output
 
 
 class ProtocolRedistributionCheck(AbstractCheck):

src/kathara_lab_checker/checks/protocols/bgp/BGPPeeringCheck.py

@@ -4,9 +4,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ...AbstractCheck import AbstractCheck
+from ...CheckResult import CheckResult
+from ....utils import get_output
 
 
 class BGPPeeringCheck(AbstractCheck):

src/kathara_lab_checker/checks/protocols/evpn/AnnouncedVNICheck.py

@@ -3,9 +3,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ...AbstractCheck import AbstractCheck
+from ...CheckResult import CheckResult
+from ....utils import get_output
 
 
 class AnnouncedVNICheck(AbstractCheck):

src/kathara_lab_checker/checks/protocols/evpn/EVPNSessionCheck.py

@@ -4,9 +4,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ...AbstractCheck import AbstractCheck
+from ...CheckResult import CheckResult
+from ....utils import get_output
 
 
 class EVPNSessionCheck(AbstractCheck):

src/kathara_lab_checker/checks/protocols/evpn/VTEPCheck.py

@@ -4,9 +4,9 @@
 from Kathara.manager.Kathara import Kathara
 from Kathara.model.Lab import Lab
 
-from kathara_lab_checker.checks.AbstractCheck import AbstractCheck
-from kathara_lab_checker.checks.CheckResult import CheckResult
-from kathara_lab_checker.utils import get_output
+from ...AbstractCheck import AbstractCheck
+from ...CheckResult import CheckResult
+from ....utils import get_output
 
 
 class VTEPCheck(AbstractCheck):

src/kathara_lab_checker/excel_utils.py

@@ -29,7 +29,7 @@ def write_final_results_to_excel(test_collector: "TestCollectorPackage.TestColle
         if failed_tests:
             failed_string = ""
             for idx, failed in enumerate(failed_tests):
-                failed_string += f"{(idx + 1)}: {failed.reason}\n"
+                failed_string += f"{(idx + 1)}: {failed.description}: {failed.reason}\n"
             if len(failed_string) >= 32767:
                 raise Exception("ERROR: Excel cell too big")
             sheet["E" + str(index + 2)] = failed_string

src/main.py

@@ -0,0 +1,4 @@
+from kathara_lab_checker.__main__ import main
+
+if __name__ == '__main__':
+    main()