tests.unit.test_ims_envista.TestIMSEnvista.test_get_latest_station_data() - Code Metrics - GuyKh/py-ims-envista - Measure and Improve Code Quality continuously with Scrutinizer

TestIMSEnvista.test_get_latest_station_data() A
last analyzed 2025-11-10 17:14 UTC

↳ Parent: tests.unit.test_ims_envista

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	1
dl	0
loc	9
rs	10
c	0
b	0
f	0

"""Test IMS Envista API."""

import os
import unittest
from datetime import date, datetime, timedelta

import pytest
import pytz
from aiohttp import ClientSession

from ims_envista import IMSEnvista


def to_date_time(d: date) -> datetime:
    """Convert date to datetime."""
    return datetime(d.year, d.month, d.day).astimezone()


class TestIMSEnvista(unittest.IsolatedAsyncioTestCase):
    """Test IMS Envista API."""

    async def asyncSetUp(self) -> None:
        """Do Setup."""
        self.token = os.environ.get("IMS_TOKEN")
        if not self.token:
            pytest.fail("Failed to load IMS Token")

        self.station_id = 178  # TEL AVIV COAST station
        self.region_id = 13
        self.channel_id = 7  # TD = Temperature Channel

        # Initialize the session in an async context
        self.session = ClientSession()
        self.tz = pytz.timezone("Asia/Jerusalem")
        self.ims = IMSEnvista(self.token, session=self.session)

    async def asyncTearDown(self) -> None:
        """Tear Down."""
        await self.session.close()


    async def test_get_all_regions_info(self) -> None:
        """Test get_all_regions_info endpoint."""
        regions = await self.ims.get_all_regions_info()

        assert regions is not None
        assert len(regions) > 0


    async def test_get_region_info(self) -> None:
        """Test get_regions_info endpoint."""
        region = await self.ims.get_region_info(self.region_id)

        assert region is not None
        assert region.region_id == self.region_id


    async def test_get_all_stations_info(self) -> None:
        """Test get_all_stations_info endpoint."""
        stations = await self.ims.get_all_stations_info()

        assert stations is not None
        assert len(stations) > 0


    async def test_get_station_info(self) -> None:
        """Test get_region_info endpoint."""
        station = await self.ims.get_station_info(self.station_id)

        assert station is not None
        assert station.station_id == self.station_id


    async def test_get_latest_station_data(self) -> None:
        """Test get_latest_station endpoint."""
        station_data = await self.ims.get_latest_station_data(self.station_id)

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        assert station_data.data[0].td > 0


    async def test_get_latest_station_data_with_channel(self) -> None:
        """Test get_latest_station_data endpoint with channel."""
        station_data = await self.ims.get_latest_station_data(
            self.station_id, self.channel_id
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        assert station_data.data[0].td > 0


    async def test_get_earliest_station_data(self) -> None:
        """Test get_earliest_station_data endpoint."""
        station_data = await self.ims.get_earliest_station_data(self.station_id)

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        assert station_data.data[0].td > 0


    async def test_get_earliest_station_data_with_channel(self) -> None:
        """Test get_earliest_station_data endpoint with channel."""
        station_data = await self.ims.get_earliest_station_data(
            self.station_id, self.channel_id
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        assert station_data.data[0].td > 0


    async def test_get_station_data_from_date(self) -> None:
        """Test get_station_data_from_date endpoint."""
        station_data = await self.ims.get_station_data_from_date(
            self.station_id, self.tz.localize(datetime.now())
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        for station_reading in station_data.data:
            assert station_reading.datetime.date() == self.tz.localize(datetime.now()).date()


    async def test_get_station_data_from_date_with_channel(self) -> None:
        """Test get_station_data_from_date endpoint with channel."""
        station_data = await self.ims.get_station_data_from_date(
            self.station_id, self.tz.localize(datetime.now()), self.channel_id
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        for station_reading in station_data.data:
            assert station_reading.datetime.date() == self.tz.localize(datetime.now()).date()


    async def test_get_station_data_by_date_range(self) -> None:

        """Test get_station_data_by_date_range endpoint."""
        today = self.tz.localize(datetime.now())
        today = today.replace(hour=0, minute=0, second=0, microsecond=0)
        yesterday = today - timedelta(days=1)
        # `hour=1` for DST fix cases
        today = today.replace(hour=2, minute=0, second=0, microsecond=0)
        station_data = await self.ims.get_station_data_by_date_range(
            self.station_id, from_date=yesterday, to_date=today
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        for station_reading in station_data.data:
            assert station_reading.datetime >= to_date_time(yesterday)
            assert station_reading.datetime < today
            assert station_reading.td > 0


    async def test_get_station_data_by_date_range_with_channel(self) -> None:

        """Test get_station_data_by_date_range endpoint with channel."""
        today = self.tz.localize(datetime.now())
        today = today.replace(hour=0, minute=0, second=0, microsecond=0)
        yesterday = today - timedelta(days=1)
        # `hour=1` for DST fix cases
        today = today.replace(hour=2, minute=0, second=0, microsecond=0)
        station_data = await self.ims.get_station_data_by_date_range(
            self.station_id,
            from_date=yesterday,
            to_date=today,
            channel_id=self.channel_id,
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        for station_reading in station_data.data:
            assert station_reading.datetime >= to_date_time(yesterday)
            assert station_reading.datetime < today
            assert station_reading.td > 0


    async def test_get_monthly_station_data(self) -> None:

        """Test get_monthly_station_data endpoint."""
        year = self.tz.localize(datetime.now()).strftime("%Y")
        month = self.tz.localize(datetime.now()).strftime("%m")
        station_data = await self.ims.get_monthly_station_data(
            self.station_id, month=month, year=year
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        for station_reading in station_data.data:
            assert station_reading.datetime.date().strftime("%Y") == year
            assert station_reading.datetime.date().strftime("%m") == month
            assert station_reading.td > 0


    async def test_get_monthly_station_data_with_channel(self) -> None:

        """Test get_monthly_station_data endpoint with channel."""
        year = self.tz.localize(datetime.now()).strftime("%Y")
        month = self.tz.localize(datetime.now()).strftime("%m")
        station_data = await self.ims.get_monthly_station_data(
            self.station_id, channel_id=self.channel_id, month=month, year=year
        )

        assert station_data is not None
        assert station_data.station_id == self.station_id
        assert station_data.data is not None
        assert len(station_data.data) > 0
        for station_reading in station_data.data:
            assert station_reading.datetime.date().strftime("%m") == month
            assert station_reading.datetime.date().strftime("%Y") == year
            assert station_reading.td > 0

    def test_get_metrics_descriptions(self) -> None:
        metrics = self.ims.get_metrics_descriptions()

        assert metrics is not None
        assert len(metrics) > 0


1		"""Test IMS Envista API."""
2
3		import os
4		import unittest
5		from datetime import date, datetime, timedelta
6
7		import pytest
8		import pytz
9		from aiohttp import ClientSession
10
11		from ims_envista import IMSEnvista
12
13
14		def to_date_time(d: date) -> datetime:
15		"""Convert date to datetime."""
16		return datetime(d.year, d.month, d.day).astimezone()
17
18
19		class TestIMSEnvista(unittest.IsolatedAsyncioTestCase):
20		"""Test IMS Envista API."""
21
22		async def asyncSetUp(self) -> None:
23		"""Do Setup."""
24		self.token = os.environ.get("IMS_TOKEN")
25		if not self.token:
26		pytest.fail("Failed to load IMS Token")
27
28		self.station_id = 178 # TEL AVIV COAST station
29		self.region_id = 13
30		self.channel_id = 7 # TD = Temperature Channel
31
32		# Initialize the session in an async context
33		self.session = ClientSession()
34		self.tz = pytz.timezone("Asia/Jerusalem")
35		self.ims = IMSEnvista(self.token, session=self.session)
36
37		async def asyncTearDown(self) -> None:
38		"""Tear Down."""
39		await self.session.close()
40
41
42		async def test_get_all_regions_info(self) -> None:
43		"""Test get_all_regions_info endpoint."""
44		regions = await self.ims.get_all_regions_info()
45
46		assert regions is not None
47		assert len(regions) > 0
48
49
50		async def test_get_region_info(self) -> None:
51		"""Test get_regions_info endpoint."""
52		region = await self.ims.get_region_info(self.region_id)
53
54		assert region is not None
55		assert region.region_id == self.region_id
56
57
58		async def test_get_all_stations_info(self) -> None:
59		"""Test get_all_stations_info endpoint."""
60		stations = await self.ims.get_all_stations_info()
61
62		assert stations is not None
63		assert len(stations) > 0
64
65
66		async def test_get_station_info(self) -> None:
67		"""Test get_region_info endpoint."""
68		station = await self.ims.get_station_info(self.station_id)
69
70		assert station is not None
71		assert station.station_id == self.station_id
72
73
74		async def test_get_latest_station_data(self) -> None:
75		"""Test get_latest_station endpoint."""
76		station_data = await self.ims.get_latest_station_data(self.station_id)
77
78		assert station_data is not None
79		assert station_data.station_id == self.station_id
80		assert station_data.data is not None
81		assert len(station_data.data) > 0
82		assert station_data.data[0].td > 0
83
84
85		async def test_get_latest_station_data_with_channel(self) -> None:
86		"""Test get_latest_station_data endpoint with channel."""
87		station_data = await self.ims.get_latest_station_data(
88		self.station_id, self.channel_id
89		)
90
91		assert station_data is not None
92		assert station_data.station_id == self.station_id
93		assert station_data.data is not None
94		assert len(station_data.data) > 0
95		assert station_data.data[0].td > 0
96
97
98		async def test_get_earliest_station_data(self) -> None:
99		"""Test get_earliest_station_data endpoint."""
100		station_data = await self.ims.get_earliest_station_data(self.station_id)
101
102		assert station_data is not None
103		assert station_data.station_id == self.station_id
104		assert station_data.data is not None
105		assert len(station_data.data) > 0
106		assert station_data.data[0].td > 0
107
108
109		async def test_get_earliest_station_data_with_channel(self) -> None:
110		"""Test get_earliest_station_data endpoint with channel."""
111		station_data = await self.ims.get_earliest_station_data(
112		self.station_id, self.channel_id
113		)
114
115		assert station_data is not None
116		assert station_data.station_id == self.station_id
117		assert station_data.data is not None
118		assert len(station_data.data) > 0
119		assert station_data.data[0].td > 0
120
121
122		async def test_get_station_data_from_date(self) -> None:
123		"""Test get_station_data_from_date endpoint."""
124		station_data = await self.ims.get_station_data_from_date(
125		self.station_id, self.tz.localize(datetime.now())
126		)
127
128		assert station_data is not None
129		assert station_data.station_id == self.station_id
130		assert station_data.data is not None
131		assert len(station_data.data) > 0
132		for station_reading in station_data.data:
133		assert station_reading.datetime.date() == self.tz.localize(datetime.now()).date()
134
135
136		async def test_get_station_data_from_date_with_channel(self) -> None:
137		"""Test get_station_data_from_date endpoint with channel."""
138		station_data = await self.ims.get_station_data_from_date(
139		self.station_id, self.tz.localize(datetime.now()), self.channel_id
140		)
141
142		assert station_data is not None
143		assert station_data.station_id == self.station_id
144		assert station_data.data is not None
145		assert len(station_data.data) > 0
146		for station_reading in station_data.data:
147		assert station_reading.datetime.date() == self.tz.localize(datetime.now()).date()
148
149
150	View Code Duplication	async def test_get_station_data_by_date_range(self) -> None:
		0 ignored issues – show Duplication introduced 2023-03-01 09:08 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
151		"""Test get_station_data_by_date_range endpoint."""
152		today = self.tz.localize(datetime.now())
153		today = today.replace(hour=0, minute=0, second=0, microsecond=0)
154		yesterday = today - timedelta(days=1)
155		# `hour=1` for DST fix cases
156		today = today.replace(hour=2, minute=0, second=0, microsecond=0)
157		station_data = await self.ims.get_station_data_by_date_range(
158		self.station_id, from_date=yesterday, to_date=today
159		)
160
161		assert station_data is not None
162		assert station_data.station_id == self.station_id
163		assert station_data.data is not None
164		assert len(station_data.data) > 0
165		for station_reading in station_data.data:
166		assert station_reading.datetime >= to_date_time(yesterday)
167		assert station_reading.datetime < today
168		assert station_reading.td > 0
169
170
171	View Code Duplication	async def test_get_station_data_by_date_range_with_channel(self) -> None:
		0 ignored issues – show Duplication introduced 2023-03-01 09:08 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
172		"""Test get_station_data_by_date_range endpoint with channel."""
173		today = self.tz.localize(datetime.now())
174		today = today.replace(hour=0, minute=0, second=0, microsecond=0)
175		yesterday = today - timedelta(days=1)
176		# `hour=1` for DST fix cases
177		today = today.replace(hour=2, minute=0, second=0, microsecond=0)
178		station_data = await self.ims.get_station_data_by_date_range(
179		self.station_id,
180		from_date=yesterday,
181		to_date=today,
182		channel_id=self.channel_id,
183		)
184
185		assert station_data is not None
186		assert station_data.station_id == self.station_id
187		assert station_data.data is not None
188		assert len(station_data.data) > 0
189		for station_reading in station_data.data:
190		assert station_reading.datetime >= to_date_time(yesterday)
191		assert station_reading.datetime < today
192		assert station_reading.td > 0
193
194
195	View Code Duplication	async def test_get_monthly_station_data(self) -> None:
		0 ignored issues – show Duplication introduced 2024-09-17 10:33 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
196		"""Test get_monthly_station_data endpoint."""
197		year = self.tz.localize(datetime.now()).strftime("%Y")
198		month = self.tz.localize(datetime.now()).strftime("%m")
199		station_data = await self.ims.get_monthly_station_data(
200		self.station_id, month=month, year=year
201		)
202
203		assert station_data is not None
204		assert station_data.station_id == self.station_id
205		assert station_data.data is not None
206		assert len(station_data.data) > 0
207		for station_reading in station_data.data:
208		assert station_reading.datetime.date().strftime("%Y") == year
209		assert station_reading.datetime.date().strftime("%m") == month
210		assert station_reading.td > 0
211
212
213	View Code Duplication	async def test_get_monthly_station_data_with_channel(self) -> None:
		0 ignored issues – show Duplication introduced 2024-09-17 10:33 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
214		"""Test get_monthly_station_data endpoint with channel."""
215		year = self.tz.localize(datetime.now()).strftime("%Y")
216		month = self.tz.localize(datetime.now()).strftime("%m")
217		station_data = await self.ims.get_monthly_station_data(
218		self.station_id, channel_id=self.channel_id, month=month, year=year
219		)
220
221		assert station_data is not None
222		assert station_data.station_id == self.station_id
223		assert station_data.data is not None
224		assert len(station_data.data) > 0
225		for station_reading in station_data.data:
226		assert station_reading.datetime.date().strftime("%m") == month
227		assert station_reading.datetime.date().strftime("%Y") == year
228		assert station_reading.td > 0
229
230		def test_get_metrics_descriptions(self) -> None:
231		metrics = self.ims.get_metrics_descriptions()
232
233		assert metrics is not None
234		assert len(metrics) > 0
235

GuyKh / py-ims-envista

TestIMSEnvista.test_get_latest_station_data() A last analyzed 2025-11-10 17:14 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestIMSEnvista.test_get_latest_station_data() A
last analyzed 2025-11-10 17:14 UTC