Public Access
93 lines
3.8 KiB
Python
93 lines
3.8 KiB
Python
import io
|
|
import http.client
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import parse_qs, urlparse
|
|
from unittest.mock import patch
|
|
|
|
import tables
|
|
|
|
from damaris.data.Temperature import read_from_hdf
|
|
from damaris.gui.TemperatureProxyClient import TemperatureProxyClient
|
|
|
|
|
|
class TestTemperatureProxyClient(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = TemperatureProxyClient()
|
|
|
|
@staticmethod
|
|
def response(payload):
|
|
return io.BytesIO(json.dumps(payload).encode("utf-8"))
|
|
|
|
@patch("damaris.gui.TemperatureProxyClient.urlopen")
|
|
def test_get_history_uses_fixed_endpoint_and_converts_readings(self, mock_urlopen):
|
|
mock_urlopen.return_value = self.response({
|
|
"readings": [
|
|
{"t": 101.5, "pv": 293.15},
|
|
{"t": 104.0, "pv": 294.25},
|
|
]
|
|
})
|
|
|
|
result = self.client.get_history(100.0, 110.0)
|
|
|
|
args, kwargs = mock_urlopen.call_args
|
|
parsed_url = urlparse(args[0])
|
|
self.assertEqual(parsed_url.scheme, "http")
|
|
self.assertEqual(parsed_url.netloc, "localhost:8000")
|
|
self.assertEqual(parsed_url.path, "/v1/readings")
|
|
self.assertEqual(parse_qs(parsed_url.query), {"start": ["100.0"], "end": ["110.0"]})
|
|
self.assertEqual(kwargs, {"timeout": TemperatureProxyClient.TIMEOUT})
|
|
self.assertEqual(result.x, [1.5, 4.0])
|
|
self.assertEqual(result.y, [293.15, 294.25])
|
|
|
|
@patch("damaris.gui.TemperatureProxyClient.urlopen")
|
|
def test_empty_readings_return_none(self, mock_urlopen):
|
|
mock_urlopen.return_value = self.response({"readings": []})
|
|
|
|
self.assertIsNone(self.client.get_history(100.0, 110.0))
|
|
|
|
@patch("damaris.gui.TemperatureProxyClient.urlopen")
|
|
def test_invalid_responses_return_none_and_warn(self, mock_urlopen):
|
|
for payload in ({}, {"readings": "invalid"}, {"readings": [{"t": 101.0}]}):
|
|
with self.subTest(payload=payload):
|
|
mock_urlopen.return_value = self.response(payload)
|
|
with self.assertLogs("damaris.gui.TemperatureProxyClient", level="WARNING"):
|
|
self.assertIsNone(self.client.get_history(100.0, 110.0))
|
|
|
|
@patch("damaris.gui.TemperatureProxyClient.urlopen")
|
|
def test_unavailable_proxy_http_errors_and_malformed_json_return_none(self, mock_urlopen):
|
|
failures = (
|
|
OSError("connection refused"),
|
|
TimeoutError("request timed out"),
|
|
http.client.BadStatusLine("not an HTTP response"),
|
|
HTTPError(TemperatureProxyClient.URL, 500, "server error", {}, None),
|
|
)
|
|
for failure in failures:
|
|
with self.subTest(failure=failure):
|
|
mock_urlopen.side_effect = failure
|
|
with self.assertLogs("damaris.gui.TemperatureProxyClient", level="WARNING"):
|
|
self.assertIsNone(self.client.get_history(100.0, 110.0))
|
|
|
|
mock_urlopen.side_effect = None
|
|
mock_urlopen.return_value = io.BytesIO(b"not json")
|
|
with self.assertLogs("damaris.gui.TemperatureProxyClient", level="WARNING"):
|
|
self.assertIsNone(self.client.get_history(100.0, 110.0))
|
|
|
|
@patch("damaris.gui.TemperatureProxyClient.urlopen")
|
|
def test_result_round_trips_through_temperature_hdf_storage(self, mock_urlopen):
|
|
mock_urlopen.return_value = self.response({"readings": [{"t": 102.0, "pv": 295.0}]})
|
|
result = self.client.get_history(100.0, 110.0)
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".h5") as hdf_path:
|
|
with tables.open_file(hdf_path.name, mode="w") as hdf_file:
|
|
result.write_to_hdf(hdf_file, "/", "temperature", "Temperature")
|
|
restored = read_from_hdf(hdf_file.root.temperature)
|
|
|
|
self.assertEqual(restored.x, [2.0])
|
|
self.assertEqual(restored.y, [295.0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |