[modbus] Fix parsing & split out server mode (#11969)

tests/integration/fixtures/uart_mock_modbus.yaml

@@ -49,15 +49,16 @@ modbus_controller:
   - address: 1
     id: modbus_controller_ok
     max_cmd_retries: 2
-    update_interval: 1s
+    # Update interval is set to never to prevent automatic polling: the test will trigger requests by pressing the "Start Scenario" button
+    update_interval: never
   - address: 2
     id: modbus_controller_slow
     max_cmd_retries: 0
-    update_interval: 1s
+    update_interval: never
   - address: 3
     id: modbus_controller_offline
     max_cmd_retries: 0
-    update_interval: 1s
+    update_interval: never
 
 sensor:
   - platform: modbus_controller
@@ -91,4 +92,11 @@ button:
     name: "Start Scenario"
     id: start_scenario_btn
     on_press:
-      - lambda: "id(virtual_uart_dev).start_scenario();"
+      - lambda: |-
+          id(virtual_uart_dev).start_scenario();
+          id(modbus_controller_ok).set_update_interval(1000);
+          id(modbus_controller_ok).start_poller();
+          id(modbus_controller_slow).set_update_interval(1000);
+          id(modbus_controller_slow).start_poller();
+          id(modbus_controller_offline).set_update_interval(1000);
+          id(modbus_controller_offline).start_poller();

tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml

@@ -54,7 +54,11 @@ modbus:
 sensor:
   - platform: sdm_meter
     address: 2
-    update_interval: 1s
+    id: sdm_meter_1
+    # update_interval is set to never to avoid automatic polling before the test starts the scenario.
+    # The test will manually start the poller after subscribing to states, to ensure no state changes are missed.
+    # This also allows us to assert there are no modbus errors/warnings during the initial request/response.
+    update_interval: never
     phase_a:
       voltage:
         name: sdm_voltage
@@ -64,4 +68,7 @@ button:
     name: "Start Scenario"
     id: start_scenario_btn
     on_press:
-      - lambda: "id(virtual_uart_dev).start_scenario();"
+      - lambda: |-
+          id(virtual_uart_dev).start_scenario();
+          id(sdm_meter_1).set_update_interval(1000);
+          id(sdm_meter_1).start_poller();

tests/integration/fixtures/uart_mock_modbus_server_controller.yaml

@@ -53,8 +53,8 @@ modbus:
 modbus_controller:
   - address: 1
     modbus_id: virtual_modbus_controller
-    update_interval: 1s
     id: modbus_controller_1
+    update_interval: 1s
 
 modbus_server:
   - address: 1
@@ -176,6 +176,4 @@ button:
   - platform: template
     name: "Start Scenario"
     id: start_scenario_btn
-    on_press:
-      - lambda: "id(virtual_uart_server).start_scenario();"
-      - lambda: "id(virtual_uart_controller).start_scenario();"
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server_controller_multiple.yaml

@@ -113,7 +113,4 @@ button:
   - platform: template
     name: "Start Scenario"
     id: start_scenario_btn
-    on_press:
-      - lambda: "id(virtual_uart_server).start_scenario();"
-      - lambda: "id(virtual_uart_server_2).start_scenario();"
-      - lambda: "id(virtual_uart_controller).start_scenario();"
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server_controller_write.yaml

@@ -326,6 +326,4 @@ button:
   - platform: template
     name: "Start Scenario"
     id: start_scenario_btn
-    on_press:
-      - lambda: "id(virtual_uart_server).start_scenario();"
-      - lambda: "id(virtual_uart_controller).start_scenario();"
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_timing.yaml

@@ -53,7 +53,11 @@ modbus:
 sensor:
   - platform: sdm_meter
     address: 2
-    update_interval: 1s
+    id: sdm_meter_1
+    # update_interval is set to never to avoid automatic polling before the test starts the scenario.
+    # The test will manually start the poller after subscribing to states, to ensure no state changes are missed.
+    # This also allows us to assert there are no modbus errors/warnings during the initial request/response.
+    update_interval: never
     phase_a:
       voltage:
         name: sdm_voltage
@@ -63,4 +67,7 @@ button:
     name: "Start Scenario"
     id: start_scenario_btn
     on_press:
-      - lambda: "id(virtual_uart_dev).start_scenario();"
+      - lambda: |-
+          id(virtual_uart_dev).start_scenario();
+          id(sdm_meter_1).set_update_interval(1000);
+          id(sdm_meter_1).start_poller();

tests/integration/test_uart_mock_modbus.py

@@ -127,15 +127,18 @@ async def test_uart_mock_modbus_timing(
 ) -> None:
     """Test modbus timing with multi-register SDM meter response."""
 
+    line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
+
     tracker = SensorTracker(["sdm_voltage"])
     voltage_changed = tracker.expect_any("sdm_voltage")
 
     async with (
-        run_compiled(yaml_config),
+        run_compiled(yaml_config, line_callback=line_callback),
         api_client_connected() as client,
     ):
         await tracker.setup_and_start_scenario(client)
         await tracker.await_change(voltage_changed, "sdm_voltage")
+        _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
 @pytest.mark.asyncio
@@ -148,26 +151,25 @@ async def test_uart_mock_modbus_no_threshold(
 
     Without the 50ms fallback timeout, the chunked response with a 40ms gap
     between USB packets would cause a false timeout and CRC failure cascade.
-    Bus-level warnings (CRC failures, buffer clears) are expected during
-    chunked reassembly — the test only verifies the final value arrives.
+    Bus-level warnings (CRC/parse failures, buffer clears) are NOT expected during
+    chunked reassembly, if timeouts are set properly — these warnings indicate undersized timeouts.
     """
 
+    line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
+
     tracker = SensorTracker(["sdm_voltage"])
     voltage_changed = tracker.expect_any("sdm_voltage")
 
     async with (
-        run_compiled(yaml_config),
+        run_compiled(yaml_config, line_callback=line_callback),
         api_client_connected() as client,
     ):
         await tracker.setup_and_start_scenario(client)
         await tracker.await_change(voltage_changed, "sdm_voltage")
+        _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
 @pytest.mark.asyncio
-@pytest.mark.xfail(
-    reason="Modbus parser cannot handle server responses from other devices on the bus. Fix tracked in PR #11969.",
-    strict=True,
-)
 async def test_uart_mock_modbus_server(
     yaml_config: str,
     run_compiled: RunCompiledFunction,
@@ -308,10 +310,6 @@ async def test_uart_mock_modbus_server_controller_write(
 
 
 @pytest.mark.asyncio
-@pytest.mark.xfail(
-    reason="Modbus parser cannot handle server responses from other devices on the bus. Fix tracked in PR #11969.",
-    strict=True,
-)
 async def test_uart_mock_modbus_server_controller_multiple(
     yaml_config: str,
     run_compiled: RunCompiledFunction,