Fix flow_control: "software" enabling RTS/CTS instead of XON/XOFF

bellows/cli/opts.py

@@ -62,7 +62,10 @@
     default="software",
     type=click.Choice(["hardware", "software", "none"], case_sensitive=False),
     envvar="EZSP_FLOW_CONTROL",
-    help="use hardware flow control",
+    help=(
+        "serial port flow control: hardware (RTS/CTS) or software (XON/XOFF);"
+        " none currently behaves like software"
+    ),
 )
 
 pan = click.option("-P", "--pan-id", type=click.IntRange(0, 65535))

bellows/uart.py

@@ -113,18 +113,15 @@ async def _connect(config, api):
     gateway = Gateway(api, connection_done_future)
     protocol = AshProtocol(gateway)
 
-    if config[zigpy.config.CONF_DEVICE_FLOW_CONTROL] is None:
-        xon_xoff, rtscts = True, False
-    else:
-        xon_xoff, rtscts = False, True
+    flow_control = config[zigpy.config.CONF_DEVICE_FLOW_CONTROL]
 
     transport, _ = await zigpy.serial.create_serial_connection(
         loop,
         lambda: protocol,
         url=config[zigpy.config.CONF_DEVICE_PATH],
         baudrate=config[zigpy.config.CONF_DEVICE_BAUDRATE],
-        xonxoff=xon_xoff,
-        rtscts=rtscts,
+        xonxoff=flow_control in (None, "software"),
+        rtscts=flow_control == "hardware",
     )
 
     await gateway.wait_until_connected()

tests/test_uart.py

@@ -10,12 +10,21 @@
 import bellows.types as t
 
 
-@pytest.mark.parametrize("flow_control", ["software", "hardware"])
-async def test_connect(flow_control, monkeypatch):
+@pytest.mark.parametrize(
+    ("flow_control", "xonxoff", "rtscts"),
+    [
+        (None, True, False),
+        ("software", True, False),
+        ("hardware", False, True),
+    ],
+)
+async def test_connect(flow_control, xonxoff, rtscts, monkeypatch):
     appmock = MagicMock()
     transport = MagicMock()
+    connect_kwargs = {}
 
     async def mockconnect(loop, protocol_factory, **kwargs):
+        connect_kwargs.update(kwargs)
         protocol = protocol_factory()
         loop.call_soon(protocol.connection_made, transport)
         return None, protocol
@@ -33,6 +42,9 @@ async def mockconnect(loop, protocol_factory, **kwargs):
         use_thread=False,
     )
 
+    assert connect_kwargs["xonxoff"] is xonxoff
+    assert connect_kwargs["rtscts"] is rtscts
+
     threads = [t for t in threading.enumerate() if "bellows" in t.name]
     assert len(threads) == 0
     gw.close()