320 lines
12 KiB
Python
320 lines
12 KiB
Python
"""Unit tests for UTEClient class."""
|
|
|
|
import time
|
|
|
|
import pytest
|
|
import requests
|
|
from pytest_mock import MockerFixture
|
|
|
|
from src.ute_wrapper.ute import (
|
|
InvalidPlanException,
|
|
InvalidPowerFactorException,
|
|
MultipleDevicesException,
|
|
ReadingRequestFailedException,
|
|
ReadingResponseInvalidException,
|
|
TariffException,
|
|
UnsupportedMethodException,
|
|
UTEClient,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def ute_client(mocker: MockerFixture):
|
|
"""Return a UTEClient instance with mocked methods."""
|
|
mocker.patch.object(UTEClient, "_login", return_value="mocked_token")
|
|
mocker.patch.object(
|
|
UTEClient,
|
|
"get_devices_list",
|
|
return_value=[{"name": "Device 1", "accountServicePointId": "device_id_1"}],
|
|
)
|
|
mocker.patch.object(UTEClient, "get_account", return_value={"meterInfo": {"tariffType": "TRIPLE"}})
|
|
return UTEClient("test@example.com", "1234567890")
|
|
|
|
|
|
def test_make_request_success(ute_client, mocker: MockerFixture):
|
|
"""Test the _make_request method with a successful request."""
|
|
mocked_response = mocker.Mock(status_code=200, json=lambda: {"success": True})
|
|
mocker.patch("src.ute_wrapper.ute.requests.get", return_value=mocked_response)
|
|
response = ute_client._make_request("GET", "http://example.com")
|
|
assert response.status_code == requests.codes.ok
|
|
assert response.json()["success"] is True
|
|
|
|
|
|
def test_make_request_no_authorization(ute_client, mocker: MockerFixture):
|
|
"""Test the _make_request method when there is no authorization token."""
|
|
del ute_client.authorization
|
|
mocked_response = mocker.Mock(status_code=requests.codes.ok, json=lambda: {"success": True})
|
|
mocker.patch("requests.get", return_value=mocked_response)
|
|
response = ute_client._make_request("GET", "http://example.com")
|
|
assert response.status_code == requests.codes.ok
|
|
assert response.json()["success"] is True
|
|
|
|
|
|
def test_make_request_all_retries_failed(ute_client, mocker: MockerFixture):
|
|
"""Test the _make_request method when all retries fail."""
|
|
mocked_response = mocker.Mock()
|
|
mocked_response.raise_for_status.side_effect = requests.RequestException
|
|
mocker.patch("requests.get", return_value=mocked_response)
|
|
mocker.patch("time.sleep")
|
|
total_call_count = 3
|
|
with pytest.raises(Exception, match="All retries failed."):
|
|
ute_client._make_request("GET", "http://example.com", retries=total_call_count)
|
|
assert mocked_response.raise_for_status.call_count == total_call_count
|
|
assert time.sleep.call_count == total_call_count - 1
|
|
|
|
|
|
def test_method_not_supported(ute_client):
|
|
"""Test the _method_not_supported method."""
|
|
with pytest.raises(UnsupportedMethodException):
|
|
ute_client._method_not_supported()
|
|
|
|
|
|
def test_login(ute_client, mocker: MockerFixture):
|
|
"""Test the _login method."""
|
|
mocked_response = mocker.Mock(text="mocked_token")
|
|
mocker.patch("requests.post", return_value=mocked_response)
|
|
token = ute_client._login()
|
|
assert token == "mocked_token"
|
|
|
|
|
|
def test_validate_power_factor(ute_client):
|
|
"""Test the _validate_power_factor method."""
|
|
with pytest.raises(InvalidPowerFactorException):
|
|
ute_client.power_factor = 1.5
|
|
ute_client._validate_power_factor()
|
|
|
|
|
|
def test_initialize_device_id(ute_client, mocker: MockerFixture):
|
|
"""Test the _initialize_device_id method."""
|
|
ute_client.device_id = ""
|
|
mocker.patch.object(ute_client, "_select_device_id", return_value="selected_device_id")
|
|
ute_client._initialize_device_id()
|
|
assert ute_client.device_id == "selected_device_id"
|
|
|
|
|
|
def test_initialize_average_cost_per_kwh(ute_client, mocker: MockerFixture):
|
|
"""Test the _initialize_average_cost_per_kwh method."""
|
|
ute_client.average_cost_per_kwh = 0.0
|
|
mocker.patch.object(ute_client, "_determine_average_cost", return_value=5.0)
|
|
ute_client._initialize_average_cost_per_kwh()
|
|
assert ute_client.average_cost_per_kwh == 5.0
|
|
|
|
|
|
def test_select_device_id_multiple_devices(ute_client, mocker: MockerFixture):
|
|
"""Test the _select_device_id method when there are multiple devices."""
|
|
mocker.patch.object(
|
|
ute_client,
|
|
"get_devices_list",
|
|
return_value=[
|
|
{"name": "Device 1", "accountServicePointId": "device_id_1"},
|
|
{"name": "Device 2", "accountServicePointId": "device_id_2"},
|
|
],
|
|
)
|
|
with pytest.raises(MultipleDevicesException):
|
|
ute_client._select_device_id()
|
|
|
|
|
|
def test_determine_average_cost_invalid_tariff(ute_client, mocker: MockerFixture):
|
|
"""Test the _determine_average_cost method with an invalid tariff."""
|
|
mocker.patch.object(ute_client, "get_account", return_value={"meterInfo": {"tariffType": "INVALID"}})
|
|
with pytest.raises(InvalidPlanException):
|
|
ute_client._determine_average_cost()
|
|
|
|
|
|
def test_determine_average_cost_invalid_tariff_key(ute_client, mocker: MockerFixture):
|
|
"""Test the _determine_average_cost method with an invalid tariff key."""
|
|
mocker.patch.object(ute_client, "get_account", side_effect=KeyError)
|
|
with pytest.raises(TariffException, match="Tariff type not standard. Explicit definition required."):
|
|
ute_client._determine_average_cost()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"date_start, date_end",
|
|
[
|
|
("2023-05-01", "2023-05-02"),
|
|
(None, None),
|
|
],
|
|
)
|
|
def test_get_historic_consumption(ute_client, mocker: MockerFixture, date_start, date_end):
|
|
"""Test the get_historic_consumption method."""
|
|
mocked_response = {
|
|
"data": [
|
|
{"magnitudeVO": "IMPORT_ACTIVE_ENERGY", "date": "2023-05-01T00:00:00+00:00", "value": "10.0"},
|
|
{"magnitudeVO": "IMPORT_ACTIVE_ENERGY", "date": "2023-05-02T00:00:00+00:00", "value": "20.0"},
|
|
],
|
|
}
|
|
mocker.patch.object(ute_client, "_make_request", return_value=mocker.Mock(json=lambda: mocked_response))
|
|
result = ute_client.get_historic_consumption(date_start, date_end)
|
|
assert result["total"]["sum_in_kwh"] == 30.0
|
|
assert result["total"]["aproximated_cost_in_uyu"] == 136.579
|
|
assert len(result["dates"]) == 2
|
|
|
|
|
|
def test_get_current_usage_info_success(ute_client, mocker: MockerFixture):
|
|
"""Test the get_current_usage_info method with a successful request."""
|
|
readings = [
|
|
{"tipoLecturaMGMI": "I1", "valor": "10"},
|
|
{"tipoLecturaMGMI": "V1", "valor": "220"},
|
|
]
|
|
mocked_reading_request = mocker.Mock(status_code=requests.codes.ok)
|
|
mocked_reading_response = mocker.Mock(json=lambda: {"success": True, "data": {"readings": readings}})
|
|
mocker.patch.object(ute_client, "_make_request", side_effect=[mocked_reading_request, mocked_reading_response])
|
|
|
|
result = ute_client.get_current_usage_info()
|
|
|
|
assert result["success"] is True
|
|
assert result["data"]["readings"] == readings
|
|
assert result["data"]["power_in_watts"] == 2200.0
|
|
assert result["data"]["using_power_factor"] is True
|
|
|
|
|
|
def test_get_current_usage_info_failed_request(ute_client, mocker: MockerFixture):
|
|
"""Test the get_current_usage_info method with a failed request."""
|
|
mocker.patch.object(
|
|
ute_client,
|
|
"_make_request",
|
|
side_effect=[mocker.Mock(status_code=400), mocker.Mock(json=lambda: {"success": False})],
|
|
)
|
|
with pytest.raises(ReadingRequestFailedException):
|
|
ute_client.get_current_usage_info()
|
|
|
|
|
|
def test_get_current_usage_info_invalid_response(ute_client, mocker: MockerFixture):
|
|
"""Test the get_current_usage_info method with an invalid response."""
|
|
mocker.patch.object(
|
|
ute_client,
|
|
"_make_request",
|
|
side_effect=[
|
|
mocker.Mock(status_code=requests.codes.ok),
|
|
mocker.Mock(json=lambda: {"success": True, "data": {}}),
|
|
],
|
|
)
|
|
with pytest.raises(ReadingResponseInvalidException):
|
|
ute_client.get_current_usage_info()
|
|
|
|
|
|
def test_get_average_price_invalid_plan(ute_client):
|
|
"""Test the get_average_price method with an invalid plan."""
|
|
with pytest.raises(InvalidPlanException):
|
|
ute_client.get_average_price("invalid_plan")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"plan, expected_price",
|
|
[
|
|
("triple", 4.5526324),
|
|
("doble", 5.523887),
|
|
],
|
|
)
|
|
def test_get_average_price(ute_client, plan, expected_price):
|
|
"""Test the get_average_price method."""
|
|
result = ute_client.get_average_price(plan)
|
|
assert result == expected_price
|
|
|
|
|
|
def test_get_peak(ute_client, mocker: MockerFixture):
|
|
"""Test the get_peak method."""
|
|
expected_peak_data = {
|
|
"deviceId": "test_device_id",
|
|
"peak": {
|
|
"value": 123.45,
|
|
"date": "2023-06-01T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
mocked_response = mocker.Mock(json=lambda: {"data": expected_peak_data})
|
|
mocker.patch.object(ute_client, "_make_request", return_value=mocked_response)
|
|
|
|
result = ute_client.get_peak()
|
|
|
|
assert result == expected_peak_data
|
|
ute_client._make_request.assert_called_once_with(
|
|
"GET",
|
|
f"https://rocme.ute.com.uy/api/v1/accounts/{ute_client.device_id}/peak",
|
|
)
|
|
|
|
|
|
def test_get_network_status(ute_client, mocker: MockerFixture):
|
|
"""Test the get_network_status method."""
|
|
expected_network_status = [
|
|
{
|
|
"departmentId": 1,
|
|
"clientsWithoutService": 100,
|
|
"percentageClientsWithoutService": 0.5,
|
|
},
|
|
{
|
|
"departmentId": 2,
|
|
"clientsWithoutService": requests.codes.ok,
|
|
"percentageClientsWithoutService": 1.0,
|
|
},
|
|
]
|
|
|
|
mocked_response = mocker.Mock(json=lambda: {"data": {"summary": expected_network_status}})
|
|
mocker.patch.object(ute_client, "_make_request", return_value=mocked_response)
|
|
|
|
result = ute_client.get_network_status()
|
|
|
|
assert result == expected_network_status
|
|
ute_client._make_request.assert_called_once_with(
|
|
"GET",
|
|
"https://rocme.ute.com.uy/api/v1/info/network/status",
|
|
)
|
|
|
|
|
|
def test_get_renewable_sources(ute_client, mocker: MockerFixture):
|
|
"""Test the get_renewable_sources method."""
|
|
expected_renewable_sources = "50.0%"
|
|
|
|
mocked_response = mocker.Mock(json=lambda: {"data": {"renewableSources": expected_renewable_sources}})
|
|
mocker.patch.object(ute_client, "_make_request", return_value=mocked_response)
|
|
|
|
result = ute_client.get_renewable_sources()
|
|
|
|
assert result == expected_renewable_sources
|
|
ute_client._make_request.assert_called_once_with(
|
|
"GET",
|
|
"https://rocme.ute.com.uy/api/v1/info/demand/global",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"readings, expected_power",
|
|
[
|
|
(
|
|
[
|
|
{"tipoLecturaMGMI": "I1", "valor": "10"},
|
|
{"tipoLecturaMGMI": "V1", "valor": "220"},
|
|
],
|
|
2200.0,
|
|
),
|
|
(
|
|
[
|
|
{"tipoLecturaMGMI": "I1", "valor": "10"},
|
|
{"tipoLecturaMGMI": "I2", "valor": "10"},
|
|
{"tipoLecturaMGMI": "V1", "valor": "220"},
|
|
{"tipoLecturaMGMI": "V2", "valor": "220"},
|
|
],
|
|
2200.0,
|
|
),
|
|
(
|
|
[
|
|
{"tipoLecturaMGMI": "I1", "valor": "10"},
|
|
{"tipoLecturaMGMI": "I2", "valor": "10"},
|
|
{"tipoLecturaMGMI": "I3", "valor": "10"},
|
|
{"tipoLecturaMGMI": "V1", "valor": "220"},
|
|
{"tipoLecturaMGMI": "V2", "valor": "220"},
|
|
{"tipoLecturaMGMI": "V3", "valor": "220"},
|
|
],
|
|
3810.4,
|
|
),
|
|
(
|
|
[],
|
|
0.0,
|
|
),
|
|
],
|
|
)
|
|
def test_convert_powers_to_power_in_watts(ute_client, readings, expected_power):
|
|
"""Test the _convert_powers_to_power_in_watts method."""
|
|
result = ute_client._convert_powers_to_power_in_watts(readings)
|
|
assert result == expected_power
|