refactor(management): add composable connection-based procedure cores

docs/changelog.md

@@ -22,6 +22,15 @@ nav_order: 2
 - `CEMIFlags.EXTENDED_FRAME_FORMAT` was removed; its value `0x0001` was reserved, not an "extended frame format" indicator - `0x0000` is used for standard frames as well as for long extended frames. `CEMIFlags.LTE_FRAME_FORMAT` and `CEMIFlags.EXTENDED_FRAME_FORMAT_MASK` were added instead.
 - Add explicit length checks to every remaining APCI `from_knx` (and the top-level `APCI.from_knx` dispatcher) as defense-in-depth on top of the broad `except (IndexError, struct.error, ValueError)` added in 3.17.0: each service now raises `ConversionError` with a specific "Invalid length for A_X in CEMI" message for a truncated, malformed or overlong frame instead of relying solely on the generic dispatcher-level catch.
 
+### Breaking Changes
+
+- Remove the `nm_invididual_address_write` typo alias for `nm_individual_address_write` in `xknx.management.procedures` — the misspelled name is no longer exported.
+
+### New Features
+
+- Add `dm_restart_r_co(conn)` and `nm_individual_address_check_conn(conn)` to `xknx.management.procedures` — variants of `dm_restart`/`nm_individual_address_check` that operate on an already-open `P2PConnection` instead of opening and closing their own, for chaining several procedures over one connection. `dm_restart_r_co` is the actual KNX v02.01.02 - Management Procedures 03.05.02 - §3.7.3 procedure name; the `_conn` suffix on the others is an xknx-only naming convention, not a KNX spec name. All existing top-level procedures (`dm_restart`, `nm_individual_address_check`, `nm_individual_address_write`, `nm_individual_address_read`, `nm_individual_address_serial_number_read`/`_write`) keep their `(xknx, ...)` signature unchanged.
+- Add `P2PConnection.send_data(payload, wait_for_ack=True)` for sending a telegram, optionally without waiting for an ACK (used internally by `dm_restart`/`dm_restart_r_co` and `nm_individual_address_write` instead of open-coding the same `TDataConnected` construction in multiple procedures).
+
 # 3.17.0 APCIs and DPTs 2026-07-25
 
 ### Deprecation notes

test/management_tests/management_test.py

@@ -10,6 +10,7 @@
     CommunicationError,
     ConfirmationError,
     ManagementConnectionError,
+    ManagementConnectionRefused,
     ManagementConnectionTimeout,
 )
 from xknx.management.management import MANAGAMENT_ACK_TIMEOUT
@@ -138,6 +139,28 @@ async def test_failed_connect_disconnect() -> None:
         await conn_1.disconnect()
 
 
+async def test_send_on_disconnected_connection() -> None:
+    """Test send_data and request raise once the connection is closed."""
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    ia = IndividualAddress("4.0.1")
+
+    conn = await xknx.management.connect(ia)
+    await conn.disconnect()
+
+    with pytest.raises(ManagementConnectionRefused):
+        await conn.send_data(apci.Restart(), wait_for_ack=False)
+
+    with pytest.raises(ManagementConnectionRefused):
+        await conn.send_data(apci.Restart())
+
+    with pytest.raises(ManagementConnectionRefused):
+        await conn.request(
+            payload=apci.DeviceDescriptorRead(descriptor=0),
+            expected=apci.DeviceDescriptorResponse,
+        )
+
+
 async def test_reject_incoming_connection() -> None:
     """Test rejecting incoming transport connections."""
     # Note: incoming L_DATA.ind indication connection requests are rejected

test/management_tests/procedures/device/test_dm_restart_r_co.py

@@ -1,14 +1,42 @@
-"""Tests for dm_restart — KNX 03.05.02 §3.7.3 DM_Restart_RCo."""
+"""Tests for dm_restart — KNX v02.01.02 - Management Procedures 03.05.02 - §3.7.3 DM_Restart_RCo."""
 
 from unittest.mock import AsyncMock, call
 
 from xknx import XKNX
-from xknx.management.procedures.device.dm_restart_r_co import dm_restart
+from xknx.management.procedures.device.dm_restart_r_co import (
+    dm_restart,
+    dm_restart_r_co,
+)
 from xknx.telegram import IndividualAddress, Telegram, apci, tpci
 
 
+async def test_dm_restart_r_co() -> None:
+    """Test dm_restart_r_co on an already-open connection."""
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    individual_address = IndividualAddress("4.0.10")
+
+    connect = Telegram(destination_address=individual_address, tpci=tpci.TConnect())
+    restart = Telegram(
+        destination_address=individual_address,
+        tpci=tpci.TDataConnected(0),
+        payload=apci.Restart(),
+    )
+    disconnect = Telegram(
+        destination_address=individual_address,
+        tpci=tpci.TDisconnect(),
+    )
+    async with xknx.management.connection(individual_address) as conn:
+        await dm_restart_r_co(conn)
+    assert xknx.cemi_handler.send_telegram.call_args_list == [
+        call(connect),
+        call(restart),
+        call(disconnect),
+    ]
+
+
 async def test_dm_restart() -> None:
-    """Test dm_restart."""
+    """Test dm_restart opens and closes its own connection."""
     xknx = XKNX()
     xknx.cemi_handler = AsyncMock()
     individual_address = IndividualAddress("4.0.10")

test/management_tests/procedures/network/test_nm_individual_address_check.py

@@ -1,11 +1,13 @@
-"""Tests for nm_individual_address_check — KNX 03.05.02 §2.19 NM_IndividualAddress_Check."""
+"""Tests for nm_individual_address_check — KNX v02.01.02 - Management Procedures 03.05.02 - §2.19 NM_IndividualAddress_Check."""
 
 import asyncio
 from unittest.mock import AsyncMock, call
 
 from xknx import XKNX
+from xknx.exceptions import ManagementConnectionRefused
 from xknx.management.procedures.network.nm_individual_address_check import (
     nm_individual_address_check,
+    nm_individual_address_check_conn,
 )
 from xknx.telegram import (
     IndividualAddress,
@@ -16,8 +18,84 @@
 )
 
 
+async def test_nm_individual_address_check_conn_success() -> None:
+    """Test nm_individual_address_check_conn when device responds normally."""
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    individual_address = IndividualAddress("4.0.10")
+
+    connect = Telegram(destination_address=individual_address, tpci=tpci.TConnect())
+    device_desc_read = Telegram(
+        destination_address=individual_address,
+        tpci=tpci.TDataConnected(0),
+        payload=apci.DeviceDescriptorRead(descriptor=0),
+    )
+    ack = Telegram(
+        source_address=individual_address,
+        destination_address=IndividualAddress(0),
+        direction=TelegramDirection.INCOMING,
+        tpci=tpci.TAck(0),
+    )
+    device_desc_resp = Telegram(
+        source_address=individual_address,
+        destination_address=IndividualAddress(0),
+        direction=TelegramDirection.INCOMING,
+        tpci=tpci.TDataConnected(0),
+        payload=apci.DeviceDescriptorResponse(),
+    )
+    async with xknx.management.connection(individual_address) as conn:
+        task = asyncio.create_task(nm_individual_address_check_conn(conn))
+        await asyncio.sleep(0)
+        assert xknx.cemi_handler.send_telegram.call_args_list == [
+            call(connect),
+            call(device_desc_read),
+        ]
+        xknx.management.process(ack)
+        xknx.management.process(device_desc_resp)
+        assert await task
+
+
+async def test_nm_individual_address_check_conn_refused() -> None:
+    """Test nm_individual_address_check_conn when device refuses the connection."""
+    xknx = XKNX()
+    xknx.cemi_handler = AsyncMock()
+    individual_address = IndividualAddress("4.0.10")
+
+    connect = Telegram(destination_address=individual_address, tpci=tpci.TConnect())
+    device_desc_read = Telegram(
+        destination_address=individual_address,
+        tpci=tpci.TDataConnected(0),
+        payload=apci.DeviceDescriptorRead(descriptor=0),
+    )
+    ack = Telegram(
+        source_address=individual_address,
+        destination_address=IndividualAddress(0),
+        direction=TelegramDirection.INCOMING,
+        tpci=tpci.TAck(0),
+    )
+    disconnect = Telegram(
+        source_address=individual_address,
+        destination_address=IndividualAddress(0),
+        direction=TelegramDirection.INCOMING,
+        tpci=tpci.TDisconnect(),
+    )
+    try:
+        async with xknx.management.connection(individual_address) as conn:
+            task = asyncio.create_task(nm_individual_address_check_conn(conn))
+            await asyncio.sleep(0)
+            assert xknx.cemi_handler.send_telegram.call_args_list == [
+                call(connect),
+                call(device_desc_read),
+            ]
+            xknx.management.process(disconnect)
+            xknx.management.process(ack)
+            assert await task
+    except ManagementConnectionRefused:
+        pass
+
+
 async def test_nm_individual_address_check_success() -> None:
-    """Test nm_individual_address_check."""
+    """Test nm_individual_address_check opens and closes its own connection."""
     xknx = XKNX()
     xknx.cemi_handler = AsyncMock()
     individual_address = IndividualAddress("4.0.10")
@@ -34,27 +112,46 @@ async def test_nm_individual_address_check_success() -> None:
         direction=TelegramDirection.INCOMING,
         tpci=tpci.TAck(0),
     )
+    ack_out = Telegram(
+        source_address=IndividualAddress(0),
+        destination_address=individual_address,
+        tpci=tpci.TAck(0),
+    )
     device_desc_resp = Telegram(
         source_address=individual_address,
         destination_address=IndividualAddress(0),
         direction=TelegramDirection.INCOMING,
         tpci=tpci.TDataConnected(0),
         payload=apci.DeviceDescriptorResponse(),
     )
+    disconnect = Telegram(
+        destination_address=individual_address,
+        tpci=tpci.TDisconnect(),
+    )
+
     task = asyncio.create_task(nm_individual_address_check(xknx, individual_address))
     await asyncio.sleep(0)
+    xknx.management.process(ack)
+    xknx.management.process(device_desc_resp)
+
+    assert await task
     assert xknx.cemi_handler.send_telegram.call_args_list == [
         call(connect),
         call(device_desc_read),
+        call(ack_out),
+        call(disconnect),
     ]
-    # receive response
-    xknx.management.process(ack)
-    xknx.management.process(device_desc_resp)
-    assert await task
 
 
-async def test_nm_individual_address_check_refused() -> None:
-    """Test nm_individual_address_check."""
+async def test_nm_individual_address_check_occupied_by_disconnect() -> None:
+    """
+    Test nm_individual_address_check when the peer disconnects during the check.
+
+    The device sends TDisconnect before answering, so nm_individual_address_check_conn
+    returns True internally; the connection context manager's own disconnect() then
+    raises ManagementConnectionRefused (peer already disconnected), which is swallowed
+    and the address is still reported as found/occupied.
+    """
     xknx = XKNX()
     xknx.cemi_handler = AsyncMock()
     individual_address = IndividualAddress("4.0.10")
@@ -65,24 +162,27 @@ async def test_nm_individual_address_check_refused() -> None:
         tpci=tpci.TDataConnected(0),
         payload=apci.DeviceDescriptorRead(descriptor=0),
     )
-    ack = Telegram(
+    disconnect_from_device = Telegram(
         source_address=individual_address,
         destination_address=IndividualAddress(0),
         direction=TelegramDirection.INCOMING,
-        tpci=tpci.TAck(0),
+        tpci=tpci.TDisconnect(),
     )
-    disconnect = Telegram(
+    ack_from_device = Telegram(
         source_address=individual_address,
         destination_address=IndividualAddress(0),
         direction=TelegramDirection.INCOMING,
-        tpci=tpci.TDisconnect(),
+        tpci=tpci.TAck(0),
     )
+
     task = asyncio.create_task(nm_individual_address_check(xknx, individual_address))
     await asyncio.sleep(0)
     assert xknx.cemi_handler.send_telegram.call_args_list == [
         call(connect),
         call(device_desc_read),
     ]
-    xknx.management.process(disconnect)
-    xknx.management.process(ack)
+    xknx.management.process(disconnect_from_device)
+    xknx.management.process(ack_from_device)
+
     assert await task
+    assert len(xknx.cemi_handler.send_telegram.call_args_list) == 2