Passed
Pull Request — main (#48)
by Guy
01:40
created

  A

Complexity

Conditions 1

Size

Total Lines 3
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Importance

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