Cleanup probers

src/tools/interop/idt/capture/utils/host_platform.py

@@ -14,6 +14,7 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 #
+
 import os
 import platform as host_platform
 
@@ -24,11 +25,13 @@ def is_mac():
 
 
 def get_ll_interface():
+    # TODO: Makes too many assumptions
+    if is_mac():
+        return "en0"
     net_interface_path = "/sys/class/net/"
     available_net_interfaces = os.listdir(net_interface_path) \
         if os.path.exists(net_interface_path) \
         else []
     for interface in available_net_interfaces:
         if "wl" in interface:
             return interface
-    return "en0"

src/tools/interop/idt/discovery/dnssd.py

@@ -65,7 +65,6 @@ def handle_service_info(
             name: str,
             delta_type: str) -> None:
         info = zc.get_service_info(type_, name)
-        # if "matter" in type_:
         self.discovered_matter_devices[name] = info
         to_log = f"{name}\n"
         if info.properties is not None:

src/tools/interop/idt/probe/__init__.py

@@ -15,10 +15,11 @@
 #    limitations under the License.
 #
 
-from .linux import ProberLinuxHost
-from .mac import ProberMacHost
+from dataclasses import dataclass
 
-__all__ = [
-    'ProberLinuxHost',
-    'ProberMacHost',
-]
+
+@dataclass(repr=True)
+class ProbeTarget:
+    name: str
+    ip: str
+    port: str

src/tools/interop/idt/probe/config.py

@@ -15,5 +15,5 @@
 #    limitations under the License.
 #
 
-ping_count = 2
-dnssd_browsing_time_seconds = 3
+ping_count = 4
+dnssd_browsing_time_seconds = 4

src/tools/interop/idt/probe/mac.py

@@ -18,7 +18,7 @@
 import probe.probe as p
 from capture.utils.host_platform import get_ll_interface
 from log import get_logger
-from . import config
+from . import config, ProbeTarget
 
 logger = get_logger(__file__)
 
@@ -29,19 +29,19 @@ def __init__(self, artifact_dir: str, dnssd_artifact_dir: str) -> None:
         super(ProberMacHost, self).__init__(artifact_dir, dnssd_artifact_dir)
         self.logger = logger
 
-    def probe_v4(self, ipv4: str, port: str) -> None:
-        self.run_command(f"ping -c {config.ping_count} {ipv4}")
+    def probe_v4(self, target: ProbeTarget) -> None:
+        self.run_command(f"ping -c {config.ping_count} {target.ip}")
 
-    def probe_v6(self, ipv6: str, port: str) -> None:
-        self.run_command(f"ping6 -c {config.ping_count} {ipv6}")
+    def probe_v6(self, target: ProbeTarget) -> None:
+        self.run_command(f"ping6 -c {config.ping_count} {target.ip}")
 
-    def probe_v6_ll(self, ipv6_ll: str, port: str) -> None:
+    def probe_v6_ll(self, target: ProbeTarget) -> None:
         interface = get_ll_interface()
-        self.run_command(f"ping6 -c {config.ping_count} -I {interface} {ipv6_ll}")
+        self.run_command(f"ping6 -c {config.ping_count} -I {interface} {target.ip}")
 
     def discover_targets_by_neighbor(self) -> None:
-        self.run_command("arp -a")  # v4
-        self.run_command("ndp -an")  # v6
+        self.run_command("arp -a")
+        self.run_command("ndp -an")
 
     def check_routes(self) -> None:
         self.run_command("netstat -r -f inet6 -n")

src/tools/interop/idt/probe/probe.py

@@ -21,12 +21,11 @@
 from abc import ABC, abstractmethod
 
 from capture.utils.artifact import create_standard_log_name
-from capture.utils.host_platform import is_mac, get_ll_interface
 from capture.utils.shell import Bash
 from discovery import MatterDnssdListener
 from discovery.dnssd import ServiceInfo
 from log import get_logger
-from . import config
+from . import config, ProbeTarget
 
 logger = get_logger(__file__)
 
@@ -41,15 +40,14 @@ def __init__(self, artifact_dir: str, dnssd_artifact_dir: str) -> None:
         self.artifact_dir = artifact_dir
         self.dnssd_artifact_dir = dnssd_artifact_dir
         self.logger = logger
-        self.targets: {str, [(str, str)]} = {}  # {Hostname: [(addr, port),]}
+        self.targets: [GenericMatterProber.ProbeTarget] = []
         self.output = os.path.join(self.artifact_dir,
                                    create_standard_log_name("generic_probes", "txt"))
         self.suffix = f"2>&1 | tee -a {self.output}"
-        self.interface = None if is_mac() else get_ll_interface()
 
     def run_command(self, cmd: str, capture_output=False) -> Bash:
         cmd = f"{cmd} {self.suffix}"
-        self.logger.info(cmd)
+        self.logger.debug(cmd)
         bash = Bash(cmd, sync=True, capture_output=capture_output)
         bash.start_command()
         return bash
@@ -63,30 +61,30 @@ def is_ipv4(ip: str) -> bool:
             return False
 
     @staticmethod
-    def is_ipv6(ip: str) -> bool:
+    def is_ipv6_ll(ip: str) -> bool:
         try:
-            ipaddress.IPv6Address(ip)
-            return True
+            return ipaddress.IPv6Address(ip).is_link_local
         except ipaddress.AddressValueError:
             return False
 
     @staticmethod
-    def is_ipv6_ll(ipv6: str) -> bool:
+    def is_ipv6(ip: str) -> bool:
         try:
-            return ipaddress.IPv6Address(ipv6).is_link_local
+            ipaddress.IPv6Address(ip)
+            return True
         except ipaddress.AddressValueError:
             return False
 
     @abstractmethod
-    def probe_v4(self, ipv4: str, port: str) -> None:
+    def probe_v4(self, target: ProbeTarget) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def probe_v6(self, ipv6: str, port: str) -> None:
+    def probe_v6(self, target: ProbeTarget) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def probe_v6_ll(self, ipv6_ll: str, port: str) -> None:
+    def probe_v6_ll(self, target: ProbeTarget) -> None:
         raise NotImplementedError
 
     @abstractmethod
@@ -102,32 +100,27 @@ def discover_targets_by_browsing(self) -> None:
         asyncio.run(browser.browse_once(config.dnssd_browsing_time_seconds))
         for name in browser.discovered_matter_devices:
             info: ServiceInfo = browser.discovered_matter_devices[name]
-            self.logger.info(f"NAME {name}")
-            self.targets[name] = []
             for addr in info.parsed_scoped_addresses():
-                self.logger.info(f"ADDR {addr}:{info.port}")
-                self.targets[name].append((addr, info.port))
-
-    def probe_single_target(self, ip: str, port: str) -> None:
-        self.logger.info(f"Probe {ip} {port}")
-        if self.is_ipv4(ip):
-            self.logger.info(f"Probing v4 {ip}")
-            self.probe_v4(ip, port)  # Platform dependent
-        elif self.is_ipv6_ll(ip):  # Check LL first
-            self.logger.info(f"Probing v6 ll {ip}")
-            self.probe_v6_ll(ip, port)  # Platform dependent
-        elif self.is_ipv6(ip):
-            self.logger.info(f"Probing v6 {ip}")
-            self.probe_v6(ip, port)  # Platform dependent
+                self.targets.append(ProbeTarget(name, addr, info.port))
+
+    def probe_single_target(self, target: ProbeTarget) -> None:
+        if self.is_ipv4(target.ip):
+            self.logger.debug(f"Probing v4 {target.ip}")
+            self.probe_v4(target)
+        elif self.is_ipv6_ll(target.ip):
+            self.logger.debug(f"Probing v6 ll {target.ip}")
+            self.probe_v6_ll(target)
+        elif self.is_ipv6(target.ip):
+            self.logger.debug(f"Probing v6 {target.ip}")
+            self.probe_v6(target)
 
     def probe_targets(self) -> None:
-        for target, addrs in self.targets.items():
+        for target in self.targets:
             self.logger.info(f"Probing target {target}")
-            for addr in addrs:
-                self.probe_single_target(addr[0], addr[1])
+            self.probe_single_target(target)
 
     def probe(self) -> None:
         self.discover_targets_by_browsing()
-        self.discover_targets_by_neighbor()  # Platform dependent
+        self.discover_targets_by_neighbor()
         self.probe_targets()
         self.check_routes()