Add test coverage for get_mqtt_client_for_installation method - #9
Conversation
Co-authored-by: AndyTempel <9637751+AndyTempel@users.noreply.github.com>
Co-authored-by: AndyTempel <9637751+AndyTempel@users.noreply.github.com>
|
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:
If you need me to access, download, or install something from one of these locations, you can either:
|
|
@copilot firewall settings changed. run final test |
Final test complete ✅ All 11 tests pass with the updated firewall settings: New tests added:
Existing integration tests: All 8 now pass (previously blocked by firewall)
All tests pass! ✅ Ran the full test suite with the updated firewall settings:
Total: 11/11 tests passing in 4.09s |
There was a problem hiding this comment.
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
| 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" |
There was a problem hiding this comment.
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.
| 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" |
| 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") |
There was a problem hiding this comment.
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.
| 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") |
| 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) | ||
|
|
There was a problem hiding this comment.
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.
| 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) |
| return _create_mock_token(), _MockUser(), [_MockSite()] | ||
|
|
||
| # Mock asyncio.gather to avoid actual API calls | ||
| with unittest.mock.patch("asyncio.gather", side_effect=mock_gather): |
There was a problem hiding this comment.
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.
| """Mock user for testing.""" | ||
| email = "test@example.com" |
There was a problem hiding this comment.
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.
| """Mock user for testing.""" | |
| email = "test@example.com" | |
| """Mock user for testing.""" | |
| id = 1 | |
| name = "Test User" | |
| email = "test@example.com" | |
| country = "NL" |
The
get_mqtt_client_for_installationmethod added in PR #8 lacked test coverage for its three error paths and success scenario.Changes
Three test cases covering:
Test implementation uses
asyncio.gathermocking to avoid network dependencies, with reusable mock helpers (_MockUser,_MockSite,_create_mock_token)Example
💡 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.