src/python_testing/test_testing/TestCommissioningStatusDetection.py

#
#    Copyright (c) 2025 Project CHIP Authors
#    All rights reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

"""
Unit tests for commissioning status detection functions in commissioning.py.

Tests cover:
- DNS-SD discovery (_is_device_operational_via_dnssd)
- Parallel session establishment (_establish_pase_or_case_session)
- is_commissioned() integration scenarios
- get_commissioned_fabric_count() integration scenarios

Scenarios include factory fresh devices, devices commissioned on the current fabric,
and devices commissioned on other fabrics.
"""

import asyncio
import contextlib
import sys
from dataclasses import dataclass
from unittest.mock import AsyncMock, MagicMock, patch

from python_path import PythonPath

# Add the python_testing directory to path so mdns_discovery module can be found
with PythonPath('..', relative_to=__file__):
    import mdns_discovery.mdns_discovery  # noqa: F401

# Test constants
TEST_NODE_ID = 1234
TEST_COMPRESSED_FABRIC_ID = 0x1234567890ABCDEF
TEST_ENDPOINT = 0
TEST_DISCRIMINATOR = 3840
TEST_PASSCODE = 20202021


@dataclass
class MockMdnsServiceInfo:
    """Mock MdnsServiceInfo for testing DNS-SD discovery."""
    instance_name: str


class MockDeviceController:
    """Mock ChipDeviceController for testing."""

    def __init__(self, compressed_fabric_id: int = TEST_COMPRESSED_FABRIC_ID):
        self._compressed_fabric_id = compressed_fabric_id
        # Mock methods that will be configured per-test
        self.EstablishPASESession = AsyncMock()
        self.GetConnectedDevice = AsyncMock()
        self.ReadAttribute = AsyncMock()
        self.CreateManualCode = MagicMock(return_value="MT:YNJV7VSC00KA0648G00")
        self.FindOrEstablishPASESession = AsyncMock()

    def GetCompressedFabricId(self) -> int:
        return self._compressed_fabric_id


def build_expected_instance_name(compressed_fabric_id: int, node_id: int) -> str:
    """Build the expected DNS-SD instance name for a device on a fabric."""
    return f'{compressed_fabric_id:016X}-{node_id:016X}'


def create_mock_attribute_result(fabric_count: int):
    """Create a mock ReadAttribute result with CommissionedFabrics == fabric_count.

    OperationalCredentials is node-scoped and always on endpoint 0.
    """
    try:
        import matter.clusters as Clusters
        return {
            0: {
                Clusters.OperationalCredentials: {
                    Clusters.OperationalCredentials.Attributes.CommissionedFabrics: fabric_count
                }
            }
        }
    except ImportError:
        mock_opcreds = MagicMock()
        mock_opcreds_attrs = MagicMock()
        mock_opcreds_attrs.CommissionedFabrics = fabric_count
        return {
            0: {
                mock_opcreds: {
                    mock_opcreds_attrs: fabric_count
                }
            }
        }


@contextlib.contextmanager
def mock_mdns_module():
    """Context manager to inject a mock mdns_discovery module into sys.modules.

    This is necessary because the real mdns_discovery depends on zeroconf,
    which may not be installed in unit test environments. By injecting a mock
    module, the `from mdns_discovery.mdns_discovery import MdnsDiscovery`
    inside _is_device_operational_via_dnssd resolves correctly.
    """
    mock_mdns_mod = MagicMock()
    mock_parent = MagicMock()
    mock_parent.mdns_discovery = mock_mdns_mod

    saved_parent = sys.modules.get('mdns_discovery')
    saved_child = sys.modules.get('mdns_discovery.mdns_discovery')

    sys.modules['mdns_discovery'] = mock_parent
    sys.modules['mdns_discovery.mdns_discovery'] = mock_mdns_mod

    try:
        yield mock_mdns_mod
    finally:
        # Restore original modules
        if saved_parent is not None:
            sys.modules['mdns_discovery'] = saved_parent
        else:
            sys.modules.pop('mdns_discovery', None)
        if saved_child is not None:
            sys.modules['mdns_discovery.mdns_discovery'] = saved_child
        else:
            sys.modules.pop('mdns_discovery.mdns_discovery', None)


# =============================================================================
# CATEGORY A: DNS-SD Discovery Tests
# =============================================================================

async def test_dnssd_device_found_on_this_fabric():
    """
    Test: Device found operational on THIS fabric via DNS-SD.

    Scenario: Device is commissioned to this fabric and advertising.
    Expected: Returns True - device is operational on our fabric.

    Value: This is the fast path. When DNS-SD finds the device, we know
    immediately it's commissioned to this fabric without needing CASE/PASE.
    """
    from matter.testing.commissioning import _is_device_operational_via_dnssd

    mock_controller = MockDeviceController()
    expected_instance_name = build_expected_instance_name(TEST_COMPRESSED_FABRIC_ID, TEST_NODE_ID)

    with mock_mdns_module() as mock_mdns_mod:
        mock_mdns = MagicMock()
        mock_mdns.get_operational_services = AsyncMock(return_value=[
            MockMdnsServiceInfo(instance_name=expected_instance_name)
        ])
        mock_mdns_mod.MdnsDiscovery.return_value = mock_mdns

        result = await _is_device_operational_via_dnssd(mock_controller, TEST_NODE_ID)

        if not result:
            return "Expected True when device is operational on this fabric"
    return None


async def test_dnssd_device_on_different_fabric():
    """
    Test: Device advertising on different fabric (Scenario 3).

    Scenario: Device is commissioned to another fabric but not ours.
    Expected: Returns False - device not operational on OUR fabric.

    Value: Verifies we don't falsely detect devices on other fabrics as ours.
    The fabric ID in the instance name must match our compressed fabric ID.
    """
    from matter.testing.commissioning import _is_device_operational_via_dnssd

    mock_controller = MockDeviceController()

    with mock_mdns_module() as mock_mdns_mod:
        mock_mdns = MagicMock()
        # Device advertising on different fabric (different compressed fabric ID)
        mock_mdns.get_operational_services = AsyncMock(return_value=[
            MockMdnsServiceInfo(instance_name="AAAAAAAAAAAAAAAA-00000000000004D2")  # Different fabric, same node
        ])
        mock_mdns_mod.MdnsDiscovery.return_value = mock_mdns

        result = await _is_device_operational_via_dnssd(mock_controller, TEST_NODE_ID)

        if result:
            return "Expected False when device is on different fabric"
    return None


async def test_dnssd_no_services_found():
    """
    Test: No operational services found via DNS-SD.

    Scenario: Factory fresh device or device not advertising.
    Expected: Returns False - no operational advertisements found.

    Value: Verifies correct behavior when DNS-SD finds nothing,
    which triggers the parallel PASE/CASE fallback.
    """
    from matter.testing.commissioning import _is_device_operational_via_dnssd

    mock_controller = MockDeviceController()

    with mock_mdns_module() as mock_mdns_mod:
        mock_mdns = MagicMock()
        mock_mdns.get_operational_services = AsyncMock(return_value=[])
        mock_mdns_mod.MdnsDiscovery.return_value = mock_mdns

        result = await _is_device_operational_via_dnssd(mock_controller, TEST_NODE_ID)

        if result:
            return "Expected False when no operational services found"
    return None


async def test_dnssd_exception_handling():
    """
    Test: DNS-SD discovery raises a generic Exception.

    Scenario: Network error, timeout, or other DNS-SD failure not mapped to the
    handled types in _is_device_operational_via_dnssd.
    Expected: Exception propagates (only OSError/ValueError/RuntimeError/TypeError/ChipStackError are swallowed).

    Value: Callers see unexpected failures explicitly instead of a silent False.
    """
    from matter.testing.commissioning import _is_device_operational_via_dnssd

    mock_controller = MockDeviceController()

    with mock_mdns_module() as mock_mdns_mod:
        mock_mdns = MagicMock()
        mock_mdns.get_operational_services = AsyncMock(side_effect=Exception("DNS-SD network error"))
        mock_mdns_mod.MdnsDiscovery.return_value = mock_mdns

        try:
            await _is_device_operational_via_dnssd(mock_controller, TEST_NODE_ID)
            return "Expected generic Exception from DNS-SD to propagate"
        except Exception as e:
            if "DNS-SD network error" not in str(e):
                return f"Unexpected exception message: {e}"
    return None


async def test_dnssd_import_error_propagates():
    """
    Test: ImportError propagates when mdns_discovery is unavailable.

    Scenario: Running in environment without mdns_discovery installed.
    Expected: ImportError propagates (NOT silently caught).

    Value: Ensures that a missing mdns_discovery module causes a clear,
    loud failure instead of silently degrading to a slower path.
    The module is always expected to be available in test environments.
    """
    from matter.testing.commissioning import _is_device_operational_via_dnssd

    mock_controller = MockDeviceController()

    import builtins

    # Simulate ImportError by intercepting the built-in import function
    original_import = builtins.__import__

    def mock_import(name, *args, **kwargs):
        if name == 'mdns_discovery' or name == 'mdns_discovery.mdns_discovery':
            raise ModuleNotFoundError(f"No module named '{name}'")
        return original_import(name, *args, **kwargs)

    # Pop them from sys.modules first so the mock is invoked
    saved_mod = sys.modules.pop('mdns_discovery', None)
    saved_submod = sys.modules.pop('mdns_discovery.mdns_discovery', None)

    try:
        with patch('builtins.__import__', side_effect=mock_import):
            try:
                await _is_device_operational_via_dnssd(mock_controller, TEST_NODE_ID)
                return "Expected ImportError to propagate when mdns_discovery module not available"
            except (ImportError, ModuleNotFoundError):
                pass  # Expected - ImportError should now propagate
    finally:
        if saved_mod:
            sys.modules['mdns_discovery'] = saved_mod
        if saved_submod:
            sys.modules['mdns_discovery.mdns_discovery'] = saved_submod

    return None


async def test_dnssd_multiple_services_finds_correct_one():
    """
    Test: Multiple services found, one matches our fabric+node.

    Scenario: Network has multiple Matter devices, we find ours.
    Expected: Returns True when our specific device is in the list.

    Value: Verifies correct filtering when multiple devices advertise.
    """
    from matter.testing.commissioning import _is_device_operational_via_dnssd

    mock_controller = MockDeviceController()
    expected_instance_name = build_expected_instance_name(TEST_COMPRESSED_FABRIC_ID, TEST_NODE_ID)

    with mock_mdns_module() as mock_mdns_mod:
        mock_mdns = MagicMock()
        mock_mdns.get_operational_services = AsyncMock(return_value=[
            MockMdnsServiceInfo(instance_name="OTHERFABRICID00000-0000000000005678"),
            MockMdnsServiceInfo(instance_name=expected_instance_name),  # Our device
            MockMdnsServiceInfo(instance_name="ANOTHERFABRIC0000-0000000000009999"),
        ])
        mock_mdns_mod.MdnsDiscovery.return_value = mock_mdns

        result = await _is_device_operational_via_dnssd(mock_controller, TEST_NODE_ID)

        if not result:
            return "Expected True when this fabric's service is in the list"
    return None


async def test_dnssd_same_fabric_different_node():
    """
    Test: Service for same fabric but different node ID.

    Scenario: Another device on our fabric, but not the one we're looking for.
    Expected: Returns False - wrong node ID.

    Value: Verifies both fabric AND node ID must match.
    """
    from matter.testing.commissioning import _is_device_operational_via_dnssd

    mock_controller = MockDeviceController()
    # Same fabric, different node
    different_node_instance = build_expected_instance_name(TEST_COMPRESSED_FABRIC_ID, 9999)

    with mock_mdns_module() as mock_mdns_mod:
        mock_mdns = MagicMock()
        mock_mdns.get_operational_services = AsyncMock(return_value=[
            MockMdnsServiceInfo(instance_name=different_node_instance)
        ])
        mock_mdns_mod.MdnsDiscovery.return_value = mock_mdns

        result = await _is_device_operational_via_dnssd(mock_controller, TEST_NODE_ID)

        if result:
            return "Expected False when service is for different node ID"
    return None


# =============================================================================
# CATEGORY B: Parallel Session Establishment Tests
# =============================================================================

async def test_parallel_session_pase_wins():
    """
    Test: PASE completes before CASE in parallel attempt.

    Scenario: Factory fresh device - PASE succeeds, CASE fails.
    Expected: Session established via PASE, CASE cancelled.

    Value: Verifies PASE path works for uncommissioned devices.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    # PASE succeeds immediately
    async def pase_success(*args, **kwargs):
        return None

    # CASE fails after a delay (simulating real network behavior)
    async def case_fail_slow(*args, **kwargs):
        await asyncio.sleep(0.1)  # Delay so PASE wins the race
        raise RuntimeError("CASE failed - no fabric")

    mock_controller.FindOrEstablishPASESession = pase_success
    mock_controller.GetConnectedDevice = case_fail_slow

    pase_params = commissioning.PaseParams(
        setup_code="MT:YNJV7VSC00KA0648G00", passcode=TEST_PASSCODE, discriminator=TEST_DISCRIMINATOR)

    try:
        session_kind = await commissioning._establish_pase_or_case_session(mock_controller, TEST_NODE_ID, pase_params)
        if session_kind != commissioning.EstablishedSessionKind.PASE:
            return f"Expected PASE session, got {session_kind}"
    except Exception as e:
        return f"Should have succeeded via PASE: {e}"

    return None


async def test_parallel_session_case_wins():
    """
    Test: CASE completes before PASE in parallel attempt.

    Scenario: Device commissioned to this fabric, CASE faster than PASE.
    Expected: Session established via CASE, PASE cancelled.

    Value: Verifies CASE path works when device is on our fabric.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    # PASE is slow (will be cancelled)
    async def pase_slow(*args, **kwargs):
        await asyncio.sleep(10)  # Will be cancelled when CASE succeeds
        return

    # CASE succeeds immediately
    async def case_success(*args, **kwargs):
        return MagicMock()

    mock_controller.FindOrEstablishPASESession = pase_slow
    mock_controller.GetConnectedDevice = case_success

    pase_params = commissioning.PaseParams(
        setup_code="MT:YNJV7VSC00KA0648G00", passcode=TEST_PASSCODE, discriminator=TEST_DISCRIMINATOR)

    try:
        session_kind = await commissioning._establish_pase_or_case_session(mock_controller, TEST_NODE_ID, pase_params)
        if session_kind != commissioning.EstablishedSessionKind.CASE:
            return f"Expected CASE session, got {session_kind}"
    except Exception as e:
        return f"Should have succeeded via CASE: {e}"

    return None


async def test_parallel_session_first_fails_second_succeeds():
    """
    Test: First connection attempt fails, second succeeds.

    Scenario: CASE fails quickly, PASE succeeds after.
    Expected: Session established via PASE after CASE failure.

    Value: Verifies fallback to second attempt when first fails.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    # Track call order
    call_order = []

    async def pase_success(*args, **kwargs):
        call_order.append('pase')
        await asyncio.sleep(0.1)  # Slight delay
        return

    async def case_fail(*args, **kwargs):
        call_order.append('case')
        raise RuntimeError("CASE failed immediately")

    mock_controller.FindOrEstablishPASESession = pase_success
    mock_controller.GetConnectedDevice = case_fail

    pase_params = commissioning.PaseParams(
        setup_code="MT:YNJV7VSC00KA0648G00", passcode=TEST_PASSCODE, discriminator=TEST_DISCRIMINATOR)

    try:
        session_kind = await commissioning._establish_pase_or_case_session(mock_controller, TEST_NODE_ID, pase_params)
        if session_kind != commissioning.EstablishedSessionKind.PASE:
            return f"Expected PASE after CASE failed first, got {session_kind}"
    except Exception as e:
        return f"Should have succeeded via PASE fallback: {e}"

    return None


async def test_parallel_session_both_fail():
    """
    Test: Both PASE and CASE connection attempts fail.

    Scenario: Device unreachable or invalid credentials.
    Expected: RuntimeError raised with both error messages.

    Value: Verifies proper error handling when device can't be reached.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    # First task fails quickly, second task fails after delay
    # This ensures there's a pending task when first one fails
    async def pase_fail_fast(*args, **kwargs):
        raise RuntimeError("PASE failed")

    async def case_fail_slow(*args, **kwargs):
        await asyncio.sleep(0.1)  # Delay so PASE fails first
        raise RuntimeError("CASE failed")

    mock_controller.FindOrEstablishPASESession = pase_fail_fast
    mock_controller.GetConnectedDevice = case_fail_slow

    pase_params = commissioning.PaseParams(
        setup_code="MT:YNJV7VSC00KA0648G00", passcode=TEST_PASSCODE, discriminator=TEST_DISCRIMINATOR)

    try:
        await commissioning._establish_pase_or_case_session(mock_controller, TEST_NODE_ID, pase_params)
        return "Should have raised RuntimeError when both fail"
    except RuntimeError as e:
        error_msg = str(e)
        if "PASE" not in error_msg or "CASE" not in error_msg:
            return f"Error should mention both PASE and CASE failures: {e}"
    except Exception as e:
        return f"Expected RuntimeError, got {type(e).__name__}: {e}"

    return None


async def test_parallel_session_no_pase_params():
    """
    Test: No PASE params provided, only CASE attempted.

    Scenario: Caller doesn't have commissioning credentials.
    Expected: Only CASE is attempted (may succeed or fail).

    Value: Verifies graceful handling when PASE not possible.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    case_called = False

    async def case_success(*args, **kwargs):
        nonlocal case_called
        case_called = True
        return MagicMock()

    mock_controller.GetConnectedDevice = case_success

    try:
        session_kind = await commissioning._establish_pase_or_case_session(mock_controller, TEST_NODE_ID, None)
        if session_kind != commissioning.EstablishedSessionKind.CASE:
            return f"CASE-only path should yield CASE, got {session_kind}"
        if not case_called:
            return "CASE should have been attempted"
    except Exception as e:
        return f"Should succeed with CASE-only: {e}"

    return None


# =============================================================================
# CATEGORY C: is_commissioned() Tests
# =============================================================================

async def test_is_commissioned_on_current_fabric_scenario2_operational_shortcircuit():
    """
    Test: Device commissioned on THIS fabric.

    Behavior: DNS-SD finds device operational -> return True immediately.
    No PASE/CASE needed since DNS-SD confirms commissioning.

    Value: This is the FAST PATH optimization. Avoids connection overhead
    when DNS-SD already confirms the device is on our fabric.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    with patch.object(commissioning, '_is_device_operational_via_dnssd', new_callable=AsyncMock) as mock_dnssd:
        mock_dnssd.return_value = True

        result = await commissioning.is_commissioned(mock_controller, TEST_NODE_ID)

        if not result:
            return "Expected True when DNS-SD confirms operational"
        if not mock_dnssd.called:
            return "DNS-SD check should have been called"
    return None


async def test_is_commissioned_on_current_fabric_scenario1_factory_fresh():
    """
    Test: SCENARIO 1 - Device is commissionable (factory fresh, pairing window open).

    Behavior:
    - DNS-SD operational returns False (not on this fabric)
    - DNS-SD commissionable returns True (device advertising _matterc._udp)
    - is_commissioned returns False immediately (fast-fail)

    Value: Factory-fresh devices are detected without side-effects (no PASE session opened).
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    with patch.object(commissioning, '_is_device_operational_via_dnssd', new_callable=AsyncMock) as mock_op, \
            patch.object(commissioning, '_is_device_commissionable_via_dnssd', new_callable=AsyncMock) as mock_comm:

        mock_op.return_value = False
        mock_comm.return_value = True

        result = await commissioning.is_commissioned(mock_controller, TEST_NODE_ID)

        if result:
            return "Expected False when device is commissionable (factory fresh)"
        if not mock_comm.called:
            return "Commissionable DNS-SD check should have been called"
    return None


async def test_is_commissioned_neither_dnssd_conclusive():
    """
    Test: Neither DNS-SD check is conclusive (device off, broken, or on another fabric).

    Behavior:
    - DNS-SD operational returns False
    - DNS-SD commissionable returns False
    - is_commissioned returns False (device not reachable)

    Value: No side-effects when device is unreachable. No PASE or CASE attempted.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    with patch.object(commissioning, '_is_device_operational_via_dnssd', new_callable=AsyncMock) as mock_op, \
            patch.object(commissioning, '_is_device_commissionable_via_dnssd', new_callable=AsyncMock) as mock_comm:

        mock_op.return_value = False
        mock_comm.return_value = False

        result = await commissioning.is_commissioned(mock_controller, TEST_NODE_ID)

        if result:
            return "Expected False when neither DNS-SD check is conclusive"
    return None


async def test_is_commissioned_exception_propagation():
    """
    Test: Exception from DNS-SD operational check propagates correctly.

    Value: Ensures callers see infrastructure failures.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    with patch.object(commissioning, '_is_device_operational_via_dnssd', new_callable=AsyncMock) as mock_dnssd:
        mock_dnssd.side_effect = RuntimeError("DNS-SD infrastructure failure")

        try:
            await commissioning.is_commissioned(mock_controller, TEST_NODE_ID)
            return "Expected RuntimeError to propagate"
        except RuntimeError as e:
            if "DNS-SD infrastructure failure" not in str(e):
                return f"Unexpected error message: {e}"
    return None


# =============================================================================
# CATEGORY D: get_commissioned_fabric_count() Tests
# =============================================================================

async def test_get_fabric_count_scenario2_operational():
    """
    Test: SCENARIO 2 - Device operational on this fabric, get count via CASE.

    Value: Verifies correct count retrieval when device is on our fabric.
    """
    import matter.clusters as Clusters
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    with patch.object(commissioning, '_is_device_operational_via_dnssd', new_callable=AsyncMock) as mock_dnssd:

        mock_dnssd.return_value = True

        # Fabric count is len(TrustedRootCertificates); use two placeholder roots.
        mock_controller.ReadAttribute = AsyncMock(return_value={
            TEST_ENDPOINT: {
                Clusters.OperationalCredentials: {
                    Clusters.OperationalCredentials.Attributes.TrustedRootCertificates: [b'root_a', b'root_b']
                }
            }
        })

        result = await commissioning.get_commissioned_fabric_count(mock_controller, TEST_NODE_ID)

        if result != 2:
            return f"Expected fabric count 2, got {result}"
    return None


async def test_get_fabric_count_scenario1_factory_fresh():
    """
    Test: SCENARIO 1 - Factory fresh device, should return 0.

    Value: Verifies 0 count for uncommissioned devices.
    """
    import matter.clusters as Clusters
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    with patch.object(commissioning, '_is_device_operational_via_dnssd', new_callable=AsyncMock) as mock_dnssd, \
            patch.object(commissioning, '_establish_pase_or_case_session', new_callable=AsyncMock) as mock_parallel:

        mock_dnssd.return_value = False
        mock_parallel.return_value = commissioning.EstablishedSessionKind.PASE

        mock_controller.ReadAttribute = AsyncMock(return_value={
            TEST_ENDPOINT: {
                Clusters.OperationalCredentials: {
                    Clusters.OperationalCredentials.Attributes.TrustedRootCertificates: []
                }
            }
        })

        pase_params = commissioning.PaseParams(
            setup_code="MT:YNJV7VSC00KA0648G00", passcode=TEST_PASSCODE, discriminator=TEST_DISCRIMINATOR)
        result = await commissioning.get_commissioned_fabric_count(
            mock_controller, TEST_NODE_ID, pase_params=pase_params
        )

        if result != 0:
            return f"Expected fabric count 0 for factory fresh, got {result}"
    return None


async def test_get_fabric_count_not_operational_no_pase_params():
    """
    Test: Device not operational via DNS-SD and no PASE params provided.

    Value: Ensures ValueError is raised to prevent timeout.
    """
    from matter.testing import commissioning

    mock_controller = MockDeviceController()

    with patch.object(commissioning, '_is_device_operational_via_dnssd', new_callable=AsyncMock) as mock_dnssd:
        mock_dnssd.return_value = False

        try:
            await commissioning.get_commissioned_fabric_count(mock_controller, TEST_NODE_ID, pase_params=None)
            return "Expected ValueError when not operational and no PASE params"
        except ValueError as e:
            if "not operational on this fabric" not in str(e):
                return f"Unexpected error message: {e}"
    return None


# =============================================================================
# TEST RUNNER
# =============================================================================

def main():
    """Run all tests and report results."""
    tests = [
        # Category A: DNS-SD Discovery Tests
        ("A1. DNS-SD: device found on THIS fabric", test_dnssd_device_found_on_this_fabric),
        ("A2. DNS-SD: device on DIFFERENT fabric", test_dnssd_device_on_different_fabric),
        ("A3. DNS-SD: no services found", test_dnssd_no_services_found),
        ("A4. DNS-SD: exception handling", test_dnssd_exception_handling),
        ("A5. DNS-SD: import error propagates", test_dnssd_import_error_propagates),
        ("A6. DNS-SD: multiple services finds correct one", test_dnssd_multiple_services_finds_correct_one),
        ("A7. DNS-SD: same fabric different node", test_dnssd_same_fabric_different_node),

        # Category B: Parallel Session Establishment Tests
        ("B1. Parallel: PASE wins race", test_parallel_session_pase_wins),
        ("B2. Parallel: CASE wins race", test_parallel_session_case_wins),
        ("B3. Parallel: first fails, second succeeds", test_parallel_session_first_fails_second_succeeds),
        ("B4. Parallel: both fail", test_parallel_session_both_fail),
        ("B5. Parallel: no PASE params (CASE only)", test_parallel_session_no_pase_params),

        # Category C: is_commissioned() Tests
        ("C1. is_commissioned: operational shortcircuit",
         test_is_commissioned_on_current_fabric_scenario2_operational_shortcircuit),
        ("C2. is_commissioned: commissionable fast-fail (factory fresh)",
         test_is_commissioned_on_current_fabric_scenario1_factory_fresh),
        ("C3. is_commissioned: neither DNS-SD conclusive",
         test_is_commissioned_neither_dnssd_conclusive),
        ("C4. is_commissioned: exception propagation",
         test_is_commissioned_exception_propagation),

        # Category D: get_commissioned_fabric_count() Tests
        ("D1. get_fabric_count: SCENARIO 2 - operational", test_get_fabric_count_scenario2_operational),
        ("D2. get_fabric_count: SCENARIO 1 - factory fresh", test_get_fabric_count_scenario1_factory_fresh),
        ("D3. get_fabric_count: not operational, no PASE params", test_get_fabric_count_not_operational_no_pase_params),
    ]

    print("\n" + "=" * 70)
    print("COMMISSIONING STATUS DETECTION - UNIT TESTS")
    print("=" * 70)

    failures = []
    passed = 0
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    for test_name, test_func in tests:
        try:
            error = loop.run_until_complete(test_func())
            if error:
                failures.append(f"{test_name}: {error}")
                print(f"  FAIL: {test_name}")
                print(f"        {error}")
            else:
                print(f"  PASS: {test_name}")
                passed += 1
        except Exception as e:
            failures.append(f"{test_name}: Exception - {e}")
            print(f"  CRASH: {test_name}")
            print(f"         {type(e).__name__}: {e}")

    loop.close()

    # Summary
    print("\n" + "-" * 70)
    print(f"SUMMARY: {passed}/{len(tests)} tests passed, {len(failures)} failures")
    print("-" * 70)

    if failures:
        print("\nFailed tests:")
        for f in failures:
            print(f"  - {f}")

    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())