-
Notifications
You must be signed in to change notification settings - Fork 30
/
test_data_source.py
52 lines (35 loc) · 1.49 KB
/
test_data_source.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import pytest
from flexmeasures.data.models.reporting import Reporter
from datetime import datetime
from pytz import UTC
def test_get_reporter_from_source(
db, app, aggregator_reporter_data_source, add_nearby_weather_sensors
):
reporter = aggregator_reporter_data_source.data_generator
assert isinstance(reporter, Reporter)
assert reporter.__class__.__name__ == "TestReporter"
res = reporter.compute(
start=datetime(2023, 1, 1, tzinfo=UTC), end=datetime(2023, 1, 2, tzinfo=UTC)
)
assert res.lineage.sources[0] == reporter.data_source
with pytest.raises(AttributeError):
reporter.compute(start=datetime(2023, 1, 1, tzinfo=UTC), end="not a date")
def test_creation_of_new_data_source(
db, app, aggregator_reporter_data_source, add_nearby_weather_sensors
):
sensor1 = add_nearby_weather_sensors.get("temperature")
sensor2 = add_nearby_weather_sensors.get("farther_temperature")
TestReporter = app.data_generators["TestReporter"]
ds1 = TestReporter(config={"sensor": sensor1.id})
db.session.add(ds1.data_source)
db.session.commit()
ds2 = TestReporter(sensor=sensor1)
assert ds1.data_source == ds2.data_source
assert ds1.data_source.attributes.get("config") == ds2.data_source.attributes.get(
"config"
)
ds3 = TestReporter(config={"sensor": sensor2.id})
assert ds3.data_source != ds2.data_source
assert ds3.data_source.attributes.get("config") != ds2.data_source.attributes.get(
"config"
)