Skip to content

Add test coverage for get_mqtt_client_for_installation method - #9

Merged
AndyTempel merged 3 commits into
feat-mqttfrom
copilot/sub-pr-8
Jan 17, 2026
Merged

Add test coverage for get_mqtt_client_for_installation method#9
AndyTempel merged 3 commits into
feat-mqttfrom
copilot/sub-pr-8

Conversation

Copilot AI commented Jan 17, 2026

Copy link
Copy Markdown
Contributor

The get_mqtt_client_for_installation method added in PR #8 lacked test coverage for its three error paths and success scenario.

Changes

  • Three test cases covering:

    • Successful VRMMQTTClient creation with valid installation
    • NotFoundError when installation doesn't exist
    • ClientError when installation lacks MQTT hostname
  • Test implementation uses asyncio.gather mocking to avoid network dependencies, with reusable mock helpers (_MockUser, _MockSite, _create_mock_token)

Example

@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_not_found():
    """Test getting MQTT client for a non-existent installation."""
    async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
        async def mock_gather(*args):
            return _create_mock_token(), _MockUser(), []  # Empty installations
        
        with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
            with pytest.raises(NotFoundError) as exc_info:
                await client.get_mqtt_client_for_installation(999999999)
            
            assert "not found" in str(exc_info.value).lower()

💡 You can make Copilot smarter by setting up custom instructions, customizing its development environment and configuring Model Context Protocol (MCP) servers. Learn more Copilot coding agent tips in the docs.

Copilot AI and others added 2 commits January 17, 2026 19:43
Co-authored-by: AndyTempel <9637751+AndyTempel@users.noreply.github.com>
Co-authored-by: AndyTempel <9637751+AndyTempel@users.noreply.github.com>

Copilot AI commented Jan 17, 2026

Copy link
Copy Markdown
Contributor Author

Just as a heads up, I was blocked by some firewall rules while working on your feedback. Expand below for details.

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

  • vrmapi.victronenergy.com
    • Triggering command: /usr/bin/python python -m pytest tests/test_client.py::test_get_mqtt_client_for_installation_success tests/test_client.py::test_get_mqtt_client_for_installation_not_found tests/test_client.py::test_get_mqtt_client_for_installation_missing_hostname -v (dns block)
    • Triggering command: /usr/bin/python python -m pytest tests/test_client.py -v --tb=short (dns block)

If you need me to access, download, or install something from one of these locations, you can either:

Copilot AI changed the title [WIP] WIP Address feedback from PR #8 on VRM client MQTT support enhancements Add test coverage for get_mqtt_client_for_installation method Jan 17, 2026
Copilot AI requested a review from AndyTempel January 17, 2026 19:47
AndyTempel

This comment was marked as outdated.

@AndyTempel

Copy link
Copy Markdown
Member

@copilot firewall settings changed. run final test

Copilot AI commented Jan 17, 2026

Copy link
Copy Markdown
Contributor Author

@copilot firewall settings changed. run final test

Final test complete ✅ All 11 tests pass with the updated firewall settings:

New tests added:

  • test_get_mqtt_client_for_installation_success - PASSED
  • test_get_mqtt_client_for_installation_not_found - PASSED
  • test_get_mqtt_client_for_installation_missing_hostname - PASSED

Existing integration tests: All 8 now pass (previously blocked by firewall)

Seems fine. Run final test, firewall settings changed.

All tests pass! ✅ Ran the full test suite with the updated firewall settings:

  • 8 existing integration tests: PASSED
  • 3 new MQTT client unit tests: PASSED

Total: 11/11 tests passing in 4.09s

Copilot AI requested a review from AndyTempel January 17, 2026 19:53
@AndyTempel
AndyTempel marked this pull request as ready for review January 17, 2026 20:05
Copilot AI review requested due to automatic review settings January 17, 2026 20:05
@AndyTempel
AndyTempel merged commit 0884a2a into feat-mqtt Jan 17, 2026
6 checks passed

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This pull request adds test coverage for the get_mqtt_client_for_installation method that was introduced in PR #8. The tests verify three error paths and one success scenario using mocked data to avoid network dependencies.

Changes:

  • Added three new test cases covering success, NotFoundError, and ClientError scenarios
  • Created reusable mock helper classes and functions (_MockUser, _MockSite, _MockSiteNoHostname, _create_mock_token)
  • Added new imports for testing: unittest.mock, NotFoundError, ClientError, AuthToken, VRMMQTTClient

Comment thread tests/test_client.py
assert mqtt_client.host == "mqtt.victronenergy.com"
assert mqtt_client.username == "test@example.com"
assert mqtt_client.password == "Bearer mock_access_token"
assert mqtt_client.installation_id == "test-vrm-id"

Copilot AI Jan 17, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Accessing mqtt_client.installation_id may fail because this attribute is passed to the parent VictronMQTTHub class and may not be directly accessible as a public attribute on the VRMMQTTClient instance. The VRMMQTTClient constructor only exposes the parameters passed to init, not the parent class attributes. Remove this assertion or verify that the parent class exposes this attribute publicly.

Suggested change
assert mqtt_client.installation_id == "test-vrm-id"
installation_id = getattr(mqtt_client, "installation_id", None)
if installation_id is not None:
assert installation_id == "test-vrm-id"

Copilot uses AI. Check for mistakes.
Comment thread tests/test_client.py
Comment on lines +414 to +475
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSite()]

# Mock asyncio.gather to avoid actual API calls
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Get MQTT client
mqtt_client = await client.get_mqtt_client_for_installation(67890)

# Verify the client is properly configured
assert mqtt_client is not None
assert isinstance(mqtt_client, VRMMQTTClient)

# Verify client attributes
assert mqtt_client.host == "mqtt.victronenergy.com"
assert mqtt_client.username == "test@example.com"
assert mqtt_client.password == "Bearer mock_access_token"
assert mqtt_client.installation_id == "test-vrm-id"

logger.info("Successfully created MQTT client with mocked data")


@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_not_found():
"""Test getting MQTT client for a non-existent installation."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
# Return empty list for installations
return _create_mock_token(), _MockUser(), []

# Mock asyncio.gather to return empty installations list
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise NotFoundError
with pytest.raises(NotFoundError) as exc_info:
await client.get_mqtt_client_for_installation(999999999)

# Verify the error message
assert "not found" in str(exc_info.value).lower()
assert "999999999" in str(exc_info.value)

logger.info("NotFoundError correctly raised for invalid installation ID")


@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_missing_hostname():
"""Test getting MQTT client for installation without MQTT hostname."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSiteNoHostname()]

# Mock asyncio.gather
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise ClientError
with pytest.raises(ClientError) as exc_info:
await client.get_mqtt_client_for_installation(67890)

# Verify the error message
assert "mqtt hostname" in str(exc_info.value).lower()
assert "67890" in str(exc_info.value)

logger.info("ClientError correctly raised for installation without MQTT hostname")

Copilot AI Jan 17, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Mocking asyncio.gather globally is too broad and risky. This patches all asyncio.gather calls, which could affect other async operations in the client or test infrastructure. Instead, mock the specific methods called by get_mqtt_client_for_installation: client._get_auth_token(), client.users.get_me(), and client.users.list_sites(). This approach is more targeted, safer, and follows the pattern used in the existing test test_stats_false_to_none_transformation which mocks specific methods.

Suggested change
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSite()]
# Mock asyncio.gather to avoid actual API calls
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Get MQTT client
mqtt_client = await client.get_mqtt_client_for_installation(67890)
# Verify the client is properly configured
assert mqtt_client is not None
assert isinstance(mqtt_client, VRMMQTTClient)
# Verify client attributes
assert mqtt_client.host == "mqtt.victronenergy.com"
assert mqtt_client.username == "test@example.com"
assert mqtt_client.password == "Bearer mock_access_token"
assert mqtt_client.installation_id == "test-vrm-id"
logger.info("Successfully created MQTT client with mocked data")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_not_found():
"""Test getting MQTT client for a non-existent installation."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
# Return empty list for installations
return _create_mock_token(), _MockUser(), []
# Mock asyncio.gather to return empty installations list
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise NotFoundError
with pytest.raises(NotFoundError) as exc_info:
await client.get_mqtt_client_for_installation(999999999)
# Verify the error message
assert "not found" in str(exc_info.value).lower()
assert "999999999" in str(exc_info.value)
logger.info("NotFoundError correctly raised for invalid installation ID")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_missing_hostname():
"""Test getting MQTT client for installation without MQTT hostname."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSiteNoHostname()]
# Mock asyncio.gather
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise ClientError
with pytest.raises(ClientError) as exc_info:
await client.get_mqtt_client_for_installation(67890)
# Verify the error message
assert "mqtt hostname" in str(exc_info.value).lower()
assert "67890" in str(exc_info.value)
logger.info("ClientError correctly raised for installation without MQTT hostname")
# Mock the specific async methods used by get_mqtt_client_for_installation
client._get_auth_token = unittest.mock.AsyncMock(return_value=_create_mock_token())
client.users.get_me = unittest.mock.AsyncMock(return_value=_MockUser())
client.users.list_sites = unittest.mock.AsyncMock(return_value=[_MockSite()])
# Get MQTT client
mqtt_client = await client.get_mqtt_client_for_installation(67890)
# Verify the client is properly configured
assert mqtt_client is not None
assert isinstance(mqtt_client, VRMMQTTClient)
# Verify client attributes
assert mqtt_client.host == "mqtt.victronenergy.com"
assert mqtt_client.username == "test@example.com"
assert mqtt_client.password == "Bearer mock_access_token"
assert mqtt_client.installation_id == "test-vrm-id"
logger.info("Successfully created MQTT client with mocked data")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_not_found():
"""Test getting MQTT client for a non-existent installation."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
# Mock the specific async methods used by get_mqtt_client_for_installation
client._get_auth_token = unittest.mock.AsyncMock(return_value=_create_mock_token())
client.users.get_me = unittest.mock.AsyncMock(return_value=_MockUser())
# Return empty list for installations to simulate "not found"
client.users.list_sites = unittest.mock.AsyncMock(return_value=[])
# Should raise NotFoundError
with pytest.raises(NotFoundError) as exc_info:
await client.get_mqtt_client_for_installation(999999999)
# Verify the error message
assert "not found" in str(exc_info.value).lower()
assert "999999999" in str(exc_info.value)
logger.info("NotFoundError correctly raised for invalid installation ID")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_missing_hostname():
"""Test getting MQTT client for installation without MQTT hostname."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
# Mock the specific async methods used by get_mqtt_client_for_installation
client._get_auth_token = unittest.mock.AsyncMock(return_value=_create_mock_token())
client.users.get_me = unittest.mock.AsyncMock(return_value=_MockUser())
client.users.list_sites = unittest.mock.AsyncMock(return_value=[_MockSiteNoHostname()])
# Should raise ClientError
with pytest.raises(ClientError) as exc_info:
await client.get_mqtt_client_for_installation(67890)
# Verify the error message
assert "mqtt hostname" in str(exc_info.value).lower()
assert "67890" in str(exc_info.value)
logger.info("ClientError correctly raised for installation without MQTT hostname")

Copilot uses AI. Check for mistakes.
Comment thread tests/test_client.py
Comment on lines +414 to +474
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSite()]

# Mock asyncio.gather to avoid actual API calls
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Get MQTT client
mqtt_client = await client.get_mqtt_client_for_installation(67890)

# Verify the client is properly configured
assert mqtt_client is not None
assert isinstance(mqtt_client, VRMMQTTClient)

# Verify client attributes
assert mqtt_client.host == "mqtt.victronenergy.com"
assert mqtt_client.username == "test@example.com"
assert mqtt_client.password == "Bearer mock_access_token"
assert mqtt_client.installation_id == "test-vrm-id"

logger.info("Successfully created MQTT client with mocked data")


@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_not_found():
"""Test getting MQTT client for a non-existent installation."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
# Return empty list for installations
return _create_mock_token(), _MockUser(), []

# Mock asyncio.gather to return empty installations list
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise NotFoundError
with pytest.raises(NotFoundError) as exc_info:
await client.get_mqtt_client_for_installation(999999999)

# Verify the error message
assert "not found" in str(exc_info.value).lower()
assert "999999999" in str(exc_info.value)

logger.info("NotFoundError correctly raised for invalid installation ID")


@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_missing_hostname():
"""Test getting MQTT client for installation without MQTT hostname."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSiteNoHostname()]

# Mock asyncio.gather
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise ClientError
with pytest.raises(ClientError) as exc_info:
await client.get_mqtt_client_for_installation(67890)

# Verify the error message
assert "mqtt hostname" in str(exc_info.value).lower()
assert "67890" in str(exc_info.value)

Copilot AI Jan 17, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Mocking asyncio.gather globally is too broad and risky. This patches all asyncio.gather calls, which could affect other async operations in the client or test infrastructure. Instead, mock the specific methods called by get_mqtt_client_for_installation: client._get_auth_token(), client.users.get_me(), and client.users.list_sites(). This approach is more targeted, safer, and follows the pattern used in the existing test test_stats_false_to_none_transformation which mocks specific methods.

Suggested change
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSite()]
# Mock asyncio.gather to avoid actual API calls
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Get MQTT client
mqtt_client = await client.get_mqtt_client_for_installation(67890)
# Verify the client is properly configured
assert mqtt_client is not None
assert isinstance(mqtt_client, VRMMQTTClient)
# Verify client attributes
assert mqtt_client.host == "mqtt.victronenergy.com"
assert mqtt_client.username == "test@example.com"
assert mqtt_client.password == "Bearer mock_access_token"
assert mqtt_client.installation_id == "test-vrm-id"
logger.info("Successfully created MQTT client with mocked data")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_not_found():
"""Test getting MQTT client for a non-existent installation."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
# Return empty list for installations
return _create_mock_token(), _MockUser(), []
# Mock asyncio.gather to return empty installations list
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise NotFoundError
with pytest.raises(NotFoundError) as exc_info:
await client.get_mqtt_client_for_installation(999999999)
# Verify the error message
assert "not found" in str(exc_info.value).lower()
assert "999999999" in str(exc_info.value)
logger.info("NotFoundError correctly raised for invalid installation ID")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_missing_hostname():
"""Test getting MQTT client for installation without MQTT hostname."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
async def mock_gather(*args):
return _create_mock_token(), _MockUser(), [_MockSiteNoHostname()]
# Mock asyncio.gather
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):
# Should raise ClientError
with pytest.raises(ClientError) as exc_info:
await client.get_mqtt_client_for_installation(67890)
# Verify the error message
assert "mqtt hostname" in str(exc_info.value).lower()
assert "67890" in str(exc_info.value)
# Mock the specific async methods used by get_mqtt_client_for_installation
mock_get_auth_token = unittest.mock.AsyncMock(return_value=_create_mock_token())
mock_get_me = unittest.mock.AsyncMock(return_value=_MockUser())
mock_list_sites = unittest.mock.AsyncMock(return_value=[_MockSite()])
with unittest.mock.patch.object(client, "_get_auth_token", mock_get_auth_token), \
unittest.mock.patch.object(client.users, "get_me", mock_get_me), \
unittest.mock.patch.object(client.users, "list_sites", mock_list_sites):
# Get MQTT client
mqtt_client = await client.get_mqtt_client_for_installation(67890)
# Verify the client is properly configured
assert mqtt_client is not None
assert isinstance(mqtt_client, VRMMQTTClient)
# Verify client attributes
assert mqtt_client.host == "mqtt.victronenergy.com"
assert mqtt_client.username == "test@example.com"
assert mqtt_client.password == "Bearer mock_access_token"
assert mqtt_client.installation_id == "test-vrm-id"
logger.info("Successfully created MQTT client with mocked data")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_not_found():
"""Test getting MQTT client for a non-existent installation."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
# Mock the specific async methods used by get_mqtt_client_for_installation
mock_get_auth_token = unittest.mock.AsyncMock(return_value=_create_mock_token())
mock_get_me = unittest.mock.AsyncMock(return_value=_MockUser())
# Return empty list for installations
mock_list_sites = unittest.mock.AsyncMock(return_value=[])
with unittest.mock.patch.object(client, "_get_auth_token", mock_get_auth_token), \
unittest.mock.patch.object(client.users, "get_me", mock_get_me), \
unittest.mock.patch.object(client.users, "list_sites", mock_list_sites):
# Should raise NotFoundError
with pytest.raises(NotFoundError) as exc_info:
await client.get_mqtt_client_for_installation(999999999)
# Verify the error message
assert "not found" in str(exc_info.value).lower()
assert "999999999" in str(exc_info.value)
logger.info("NotFoundError correctly raised for invalid installation ID")
@pytest.mark.asyncio
async def test_get_mqtt_client_for_installation_missing_hostname():
"""Test getting MQTT client for installation without MQTT hostname."""
# Create a mock client
async with VictronVRMClient(token="mock_token", token_type="Bearer") as client:
# Mock the specific async methods used by get_mqtt_client_for_installation
mock_get_auth_token = unittest.mock.AsyncMock(return_value=_create_mock_token())
mock_get_me = unittest.mock.AsyncMock(return_value=_MockUser())
mock_list_sites = unittest.mock.AsyncMock(return_value=[_MockSiteNoHostname()])
with unittest.mock.patch.object(client, "_get_auth_token", mock_get_auth_token), \
unittest.mock.patch.object(client.users, "get_me", mock_get_me), \
unittest.mock.patch.object(client.users, "list_sites", mock_list_sites):
# Should raise ClientError
with pytest.raises(ClientError) as exc_info:
await client.get_mqtt_client_for_installation(67890)
# Verify the error message
assert "mqtt hostname" in str(exc_info.value).lower()
assert "67890" in str(exc_info.value)

Copilot uses AI. Check for mistakes.
Comment thread tests/test_client.py
return _create_mock_token(), _MockUser(), [_MockSite()]

# Mock asyncio.gather to avoid actual API calls
with unittest.mock.patch("asyncio.gather", side_effect=mock_gather):

Copilot AI Jan 17, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Mocking asyncio.gather globally is too broad and risky. This patches all asyncio.gather calls, which could affect other async operations in the client or test infrastructure. Instead, mock the specific methods called by get_mqtt_client_for_installation: client._get_auth_token(), client.users.get_me(), and client.users.list_sites(). This approach is more targeted, safer, and follows the pattern used in the existing test test_stats_false_to_none_transformation which mocks specific methods.

Copilot uses AI. Check for mistakes.
Comment thread tests/test_client.py
Comment on lines +29 to +30
"""Mock user for testing."""
email = "test@example.com"

Copilot AI Jan 17, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The _MockUser class is missing required attributes from the actual User model. According to the User model in victron_vrm/models/user.py, it requires id, name, email, and country fields. Add these missing fields to make the mock match the expected structure.

Suggested change
"""Mock user for testing."""
email = "test@example.com"
"""Mock user for testing."""
id = 1
name = "Test User"
email = "test@example.com"
country = "NL"

Copilot uses AI. Check for mistakes.
@AndyTempel
AndyTempel deleted the copilot/sub-pr-8 branch January 17, 2026 20:10
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants