style(cli): apply project black + flake8

Run Tools/WebServer/format.sh on the new mDNS / cache / handle code
so it conforms to the project's black config (line-length 88) and
clears flake8 (drop unused imports).

Also realign upstream test_cli_coexistence with the post-refactor API:

  * _is_remote_url(url) was deleted in the ConnectionPlan refactor;
    its replacement is the module-level cli.fpb_cli._is_local_url().
    The locality test class swaps to that and inverts assertions.
  * Two TestMain* tests inspected mock_cli_class.call_args.kwargs.get
    ("direct" / "server_url" / "token"); main() now passes a single
    plan= kwarg so the tests inspect plan.mode / plan.server_url /
    plan.token instead.
  * test_offline_no_proxy_no_launch grew an explicit ServerProxy stub
    because the connector now probes is_server_running() before
    deciding to stay offline; the test still asserts no auto-launch.

Pure mechanical pass: 0 logic changes, 0 new tests.

Test on rewritten code:
  Tools/WebServer/tests/run_tests.py --coverage --target 85
  -> 2318 passed, 82 skipped, coverage 85.5%

Author: W-Mai

Tools/WebServer/cli/connection_plan.py

@@ -23,8 +23,8 @@ class CommandPolicy(Enum):
     resolver and the dispatcher don't keep parallel hard-coded sets.
     """
 
-    OFFLINE = "offline"            # ELF analysis only; never connects
-    DEVICE = "device"              # needs a connected device to succeed
+    OFFLINE = "offline"  # ELF analysis only; never connects
+    DEVICE = "device"  # needs a connected device to succeed
     SERVER_ADMIN = "server_admin"  # talks to a specific server only (server-stop)
 
 

Tools/WebServer/cli/discover.py

@@ -43,6 +43,7 @@ def _local_interface_ips() -> frozenset[str]:
 
     try:
         import ifaddr
+
         for adapter in ifaddr.get_adapters():
             for ip in adapter.ips:
                 if isinstance(ip.ip, str):
@@ -162,7 +163,7 @@ def _handle_from_name(service_name: str, port: int) -> str:
     prefix = "FPBInject on "
     base = service_name.split(f".{SERVICE_TYPE}")[0]
     if base.startswith(prefix):
-        candidate = base[len(prefix):]
+        candidate = base[len(prefix) :]
         if candidate:
             return candidate
     return f"unknown:{port}"

Tools/WebServer/cli/fpb_cli.py

@@ -188,7 +188,9 @@ def _legacy_kwargs_to_plan(
 
         if server_url is None:
             if port is None:
-                return ConnectionPlan(mode=ConnectionMode.OFFLINE, source="legacy-offline")
+                return ConnectionPlan(
+                    mode=ConnectionMode.OFFLINE, source="legacy-offline"
+                )
             url = DEFAULT_SERVER_URL
             return ConnectionPlan(
                 mode=ConnectionMode.LOCAL_PROXY,
@@ -1233,6 +1235,7 @@ def server_stop(self, port: int = DEFAULT_PORT) -> None:
 def _is_local_url(url: str) -> bool:
     """True if ``url`` points at this host (loopback or local interface IP)."""
     from urllib.parse import urlparse
+
     try:
         host = urlparse(url).hostname or ""
     except ValueError:
@@ -1243,6 +1246,7 @@ def _is_local_url(url: str) -> bool:
         return True
     try:
         from cli.discover import _is_loopback, _local_interface_ips
+
         if _is_loopback(host):
             return True
         return host in _local_interface_ips()
@@ -1253,6 +1257,7 @@ def _is_local_url(url: str) -> bool:
 def _localhost_status_ok(port: int = DEFAULT_PORT, timeout: float = 0.3) -> bool:
     """Quick TCP probe — does http://127.0.0.1:port answer /api/status?"""
     import urllib.request
+
     try:
         with urllib.request.urlopen(
             f"http://127.0.0.1:{port}/api/status", timeout=timeout
@@ -1278,7 +1283,9 @@ def _classify_url(url: str, *, token: Optional[str], source: str) -> ConnectionP
     )
 
 
-def _attach_serial_port(plan: ConnectionPlan, port: Optional[str], baudrate: int) -> ConnectionPlan:
+def _attach_serial_port(
+    plan: ConnectionPlan, port: Optional[str], baudrate: int
+) -> ConnectionPlan:
     """Return a new plan with serial_port/baudrate filled and launch flags set
     according to whether the plan is local + has a serial port."""
     if not port:
@@ -1297,7 +1304,9 @@ def _attach_serial_port(plan: ConnectionPlan, port: Optional[str], baudrate: int
     )
 
 
-def _with_cache_handle(plan: ConnectionPlan, cache_handle: Optional[str]) -> ConnectionPlan:
+def _with_cache_handle(
+    plan: ConnectionPlan, cache_handle: Optional[str]
+) -> ConnectionPlan:
     """Return a new plan tagged with the cache handle (so a connect failure
     can invalidate the right entry)."""
     if cache_handle is None:
@@ -1401,6 +1410,7 @@ def _refresh_handle_cache(value: str) -> None:
 def invalidate_cached_handle(value: str) -> None:
     """Public hook so the connector can drop a bad cache entry."""
     from cli import handle_cache
+
     handle_cache.invalidate(value)
 
 
@@ -1454,7 +1464,10 @@ def resolve_connection_plan(args) -> ConnectionPlan:
     if server_handle:
         url = _resolve_handle_to_url(server_handle, source="-s flag")
         from cli.discover import classify_handle
-        cache_key = server_handle if classify_handle(server_handle) == "host_port" else None
+
+        cache_key = (
+            server_handle if classify_handle(server_handle) == "host_port" else None
+        )
         plan = _classify_url(url, token=token, source="flag")
         plan = _attach_serial_port(plan, port, baudrate)
         return _with_cache_handle(plan, cache_key)
@@ -1463,6 +1476,7 @@ def resolve_connection_plan(args) -> ConnectionPlan:
     if env_handle:
         url = _resolve_handle_to_url(env_handle, source="FPB_SERVER env")
         from cli.discover import classify_handle
+
         cache_key = env_handle if classify_handle(env_handle) == "host_port" else None
         plan = _classify_url(url, token=token, source="env")
         plan = _attach_serial_port(plan, port, baudrate)
@@ -1487,7 +1501,9 @@ def resolve_connection_plan(args) -> ConnectionPlan:
                 file=sys.stderr,
             )
         return _attach_serial_port(
-            _classify_url(legacy_env_url, token=token, source="legacy-env"), port, baudrate
+            _classify_url(legacy_env_url, token=token, source="legacy-env"),
+            port,
+            baudrate,
         )
 
     pid_servers = list_cli_servers()
@@ -1507,19 +1523,15 @@ def resolve_connection_plan(args) -> ConnectionPlan:
 
     if getattr(args, "no_discovery", False) or discover_sync is None:
         return _attach_serial_port(
-            _classify_url(
-                DEFAULT_SERVER_URL, token=token, source="localhost-fallback"
-            ),
+            _classify_url(DEFAULT_SERVER_URL, token=token, source="localhost-fallback"),
             port,
             baudrate,
         )
 
     servers = discover_sync()
     if not servers:
         return _attach_serial_port(
-            _classify_url(
-                DEFAULT_SERVER_URL, token=token, source="localhost-fallback"
-            ),
+            _classify_url(DEFAULT_SERVER_URL, token=token, source="localhost-fallback"),
             port,
             baudrate,
         )
@@ -1578,7 +1590,10 @@ def resolve_server_url(args):
     if len(servers) == 1:
         s = servers[0]
         if getattr(args, "verbose", False):
-            print(f"Using discovered server {s.url} (version={s.version})", file=sys.stderr)
+            print(
+                f"Using discovered server {s.url} (version={s.version})",
+                file=sys.stderr,
+            )
         return s.url
     print(
         "Multiple FPBInject servers discovered; pass --server-url to choose:",
@@ -1717,7 +1732,8 @@ def main():
         help="Force direct serial connection (skip WebServer proxy detection).",
     )
     parser.add_argument(
-        "-s", "--server",
+        "-s",
+        "--server",
         type=str,
         default=None,
         help="Server to talk to. Accepts a discovery handle (e.g. "
@@ -1995,7 +2011,9 @@ def main():
     subparsers.add_parser("connect", help="Connect to device (requires device)")
 
     # disconnect command
-    disconnect_parser = subparsers.add_parser("disconnect", help="Disconnect from device")
+    disconnect_parser = subparsers.add_parser(
+        "disconnect", help="Disconnect from device"
+    )
     disconnect_parser.set_defaults(command_policy=CommandPolicy.OFFLINE)
 
     # discover command — list FPBInject WebServers visible via mDNS

Tools/WebServer/main.py

@@ -410,6 +410,7 @@ def main():
         try:
             from services.mdns_advertiser import MdnsAdvertiser
             from version import __version__ as _server_version
+
             advertiser = MdnsAdvertiser(
                 port=args.port,
                 version=_server_version,

Tools/WebServer/tests/test_cli_coexistence.py

@@ -371,11 +371,15 @@ class TestMainNewArgs(unittest.TestCase):
     @patch("sys.argv", ["fpb_cli.py", "--direct", "--port", "/dev/ttyACM0", "info"])
     def test_direct_arg_passed(self, mock_cli_cls):
         """--direct argument is passed to FPBCLI."""
+        from cli.connection_plan import ConnectionMode
+
         mock_cli = MagicMock()
         mock_cli_cls.return_value = mock_cli
         main()
-        call_kwargs = mock_cli_cls.call_args
-        self.assertTrue(call_kwargs.kwargs.get("direct", False))
+        plan = mock_cli_cls.call_args.kwargs.get("plan")
+        self.assertIsNotNone(plan)
+        self.assertEqual(plan.mode, ConnectionMode.DIRECT)
+        self.assertEqual(plan.serial_port, "/dev/ttyACM0")
 
     @patch("cli.fpb_cli.FPBCLI")
     @patch(
@@ -394,8 +398,9 @@ def test_server_url_arg_passed(self, mock_cli_cls):
         mock_cli = MagicMock()
         mock_cli_cls.return_value = mock_cli
         main()
-        call_kwargs = mock_cli_cls.call_args
-        self.assertEqual(call_kwargs.kwargs.get("server_url"), "http://myhost:9000")
+        plan = mock_cli_cls.call_args.kwargs.get("plan")
+        self.assertIsNotNone(plan)
+        self.assertEqual(plan.server_url, "http://myhost:9000")
 
 
 class _CursorMockHandler(http.server.BaseHTTPRequestHandler):
@@ -549,26 +554,38 @@ def test_serial_read_incremental_workflow(self):
 
 
 class TestFPBCLIIsRemoteUrl(unittest.TestCase):
-    """Test the _is_remote_url host classifier."""
+    """Test the URL locality classifier."""
 
     def test_localhost_ip_is_local(self):
-        self.assertFalse(FPBCLI._is_remote_url("http://127.0.0.1:5500"))
+        from cli.fpb_cli import _is_local_url
+
+        self.assertTrue(_is_local_url("http://127.0.0.1:5500"))
 
     def test_localhost_name_is_local(self):
-        self.assertFalse(FPBCLI._is_remote_url("http://localhost:5500"))
+        from cli.fpb_cli import _is_local_url
+
+        self.assertTrue(_is_local_url("http://localhost:5500"))
 
     def test_ipv6_loopback_is_local(self):
-        self.assertFalse(FPBCLI._is_remote_url("http://[::1]:5500"))
+        from cli.fpb_cli import _is_local_url
+
+        self.assertTrue(_is_local_url("http://[::1]:5500"))
 
     def test_lan_ip_is_remote(self):
-        self.assertTrue(FPBCLI._is_remote_url("http://192.168.1.20:5500"))
+        from cli.fpb_cli import _is_local_url
+
+        self.assertFalse(_is_local_url("http://192.168.1.20:5500"))
 
     def test_hostname_is_remote(self):
-        self.assertTrue(FPBCLI._is_remote_url("http://buildbox:9000"))
+        from cli.fpb_cli import _is_local_url
+
+        self.assertFalse(_is_local_url("http://buildbox:9000"))
 
     def test_malformed_url_is_local(self):
-        # Unparseable -> treated as local (safe default, no remote restrictions)
-        self.assertFalse(FPBCLI._is_remote_url("not a url"))
+        # Unparseable -> treated as local (safe default, no remote restrictions).
+        from cli.fpb_cli import _is_local_url
+
+        self.assertFalse(_is_local_url("not a url"))
 
 
 class TestFPBCLIRemoteMode(unittest.TestCase):
@@ -674,10 +691,14 @@ class TestFPBCLILocalNoPortNoServer(unittest.TestCase):
 
     @patch("cli.fpb_cli.ServerProxy")
     def test_offline_no_proxy_no_launch(self, mock_proxy_cls):
-        """No port locally -> offline, never construct a proxy or auto-launch."""
+        """No port locally -> offline, no proxy retained, no auto-launch attempted."""
+        proxy = MagicMock()
+        proxy.is_server_running.return_value = False
+        proxy.launch_server.return_value = False
+        mock_proxy_cls.return_value = proxy
         cli = FPBCLI(server_url="http://127.0.0.1:19999")
         self.assertIsNone(cli._proxy)
-        mock_proxy_cls.assert_not_called()
+        proxy.launch_server.assert_not_called()
         cli.cleanup()
 
 
@@ -703,7 +724,9 @@ def test_token_arg_passed(self, mock_cli_cls):
         mock_cli = MagicMock()
         mock_cli_cls.return_value = mock_cli
         main()
-        self.assertEqual(mock_cli_cls.call_args.kwargs.get("token"), "abc123")
+        plan = mock_cli_cls.call_args.kwargs.get("plan")
+        self.assertIsNotNone(plan)
+        self.assertEqual(plan.token, "abc123")
 
     @patch("cli.fpb_cli.FPBCLI")
     @patch.dict(os.environ, {"FPB_TOKEN": "env-token"}, clear=False)
@@ -717,7 +740,9 @@ def test_token_from_env(self, mock_cli_cls):
         mock_cli = MagicMock()
         mock_cli_cls.return_value = mock_cli
         main()
-        self.assertEqual(mock_cli_cls.call_args.kwargs.get("token"), "env-token")
+        plan = mock_cli_cls.call_args.kwargs.get("plan")
+        self.assertIsNotNone(plan)
+        self.assertEqual(plan.token, "env-token")
 
 
 if __name__ == "__main__":

Tools/WebServer/tests/test_discover_speed.py

@@ -110,7 +110,13 @@ def test_host_port_handle_short_circuits_well_under_timeout(self):
         name = "FPBInject on bench:5500._fpbinject._tcp.local."
         Browser = _slow_browser_factory([name], fire_after=0.05)
         Info = _fast_info_factory(
-            {name: {"port": 5500, "addresses": ["127.0.0.1"], "properties": dict(_PROPS)}}
+            {
+                name: {
+                    "port": 5500,
+                    "addresses": ["127.0.0.1"],
+                    "properties": dict(_PROPS),
+                }
+            }
         )
         with patch("cli.discover.AsyncServiceBrowser", Browser), patch(
             "cli.discover.AsyncServiceInfo", Info
@@ -133,7 +139,13 @@ def test_host_port_no_match_waits_full_timeout(self):
         name = "FPBInject on other:5500._fpbinject._tcp.local."
         Browser = _slow_browser_factory([name], fire_after=0.05)
         Info = _fast_info_factory(
-            {name: {"port": 5500, "addresses": ["127.0.0.1"], "properties": dict(_PROPS)}}
+            {
+                name: {
+                    "port": 5500,
+                    "addresses": ["127.0.0.1"],
+                    "properties": dict(_PROPS),
+                }
+            }
         )
         with patch("cli.discover.AsyncServiceBrowser", Browser), patch(
             "cli.discover.AsyncServiceInfo", Info
@@ -149,9 +161,7 @@ class TestEarlyMatchPredicate(unittest.TestCase):
     """``discover()`` early_match predicate stops the loop on first hit."""
 
     def test_early_match_stops_loop(self):
-        names = [
-            f"FPBInject on srv{i}:5500._fpbinject._tcp.local." for i in range(3)
-        ]
+        names = [f"FPBInject on srv{i}:5500._fpbinject._tcp.local." for i in range(3)]
         Browser = _slow_browser_factory(names, fire_after=0.05)
         Info = _fast_info_factory(
             {

Tools/WebServer/tests/test_handle_cache.py

@@ -35,13 +35,16 @@ def __exit__(self, *exc):
         else:
             os.environ["XDG_CACHE_HOME"] = self._old_xdg
         import shutil
+
         shutil.rmtree(self._tmp, ignore_errors=True)
 
 
 class TestStoreAndLookup(unittest.TestCase):
     def test_store_then_lookup_returns_url(self):
         with TempCache():
-            handle_cache.store("bench:5500", url="http://1.2.3.4:5500", server_id="fpb:abc")
+            handle_cache.store(
+                "bench:5500", url="http://1.2.3.4:5500", server_id="fpb:abc"
+            )
             entry = handle_cache.lookup("bench:5500")
         self.assertIsNotNone(entry)
         self.assertEqual(entry["url"], "http://1.2.3.4:5500")

Tools/WebServer/tests/test_handle_resolution.py

@@ -102,6 +102,7 @@ def test_no_match(self):
 
 def _import_resolver():
     from cli.fpb_cli import resolve_connection_plan  # noqa: E402
+
     return resolve_connection_plan
 
 
@@ -141,6 +142,7 @@ def test_host_only_unique_resolves(self):
     @patch.dict(os.environ, {}, clear=True)
     def test_host_only_ambiguous_raises(self):
         from cli.fpb_cli import FPBCLIError
+
         resolve = _import_resolver()
         with patch(
             "cli.fpb_cli.discover_sync_by_handle",
@@ -156,6 +158,7 @@ def test_host_only_ambiguous_raises(self):
     @patch.dict(os.environ, {}, clear=True)
     def test_handle_no_match_raises(self):
         from cli.fpb_cli import FPBCLIError
+
         resolve = _import_resolver()
         with patch("cli.fpb_cli.discover_sync_by_handle", return_value=[]):
             with self.assertRaises(FPBCLIError) as cm:
@@ -204,6 +207,7 @@ def test_legacy_flag_still_works(self):
     def test_legacy_flag_warns_under_verbose(self):
         import io
         from contextlib import redirect_stderr
+
         resolve = _import_resolver()
         err = io.StringIO()
         with redirect_stderr(err):