|
7 | 7 | from sqlalchemy.engine import Engine |
8 | 8 |
|
9 | 9 | from flexmeasures import Sensor, Source, User |
| 10 | +from flexmeasures.api.v3_0.tests.conftest import GAS_MEASUREMENTS_10MIN |
10 | 11 | from flexmeasures.api.v3_0.tests.utils import make_sensor_data_request_for_gas_sensor |
11 | 12 |
|
12 | 13 |
|
@@ -73,9 +74,17 @@ def test_get_sensor_data( |
73 | 74 | print("Server responded with:\n%s" % response.json) |
74 | 75 | assert response.status_code == 200 |
75 | 76 | values = response.json["values"] |
76 | | - # We expect two data points (from conftest) followed by 2 null values (which are converted to None by .json) |
77 | | - # The first data point averages [91.3, 91.7], and the second data point averages [92.1, None]. |
78 | | - assert all(a == b for a, b in zip(values, [91.5, 92.1, None, None])) |
| 77 | + # GAS_MEASUREMENTS_10MIN stores 10-minute values; resampled to 20-minute resolution: |
| 78 | + # - 1st interval: average of [91.3, 91.7] = 91.5 |
| 79 | + # - 2nd interval: average of [92.1, None] = 92.1 (only one value present) |
| 80 | + # - 3rd and 4th intervals: no data → None |
| 81 | + expected = [ |
| 82 | + sum(GAS_MEASUREMENTS_10MIN[0:2]) / 2, # 91.5 |
| 83 | + GAS_MEASUREMENTS_10MIN[2], # 92.1 |
| 84 | + None, |
| 85 | + None, |
| 86 | + ] |
| 87 | + assert all(a == b for a, b in zip(values, expected)) |
79 | 88 |
|
80 | 89 |
|
81 | 90 | @pytest.mark.parametrize( |
@@ -114,6 +123,49 @@ def test_get_instantaneous_sensor_data( |
114 | 123 | assert all(a == b for a, b in zip(values, [815, 818, None, None])) |
115 | 124 |
|
116 | 125 |
|
| 126 | +@pytest.mark.parametrize( |
| 127 | + "requesting_user", ["[email protected]"], indirect=True |
| 128 | +) |
| 129 | +def test_get_sensor_data_filtered_by_source_account( |
| 130 | + client, |
| 131 | + setup_api_test_data: dict[str, Sensor], |
| 132 | + setup_roles_users: dict[str, User], |
| 133 | + requesting_user, |
| 134 | + db, |
| 135 | +): |
| 136 | + """Check that GET /sensors/<id>/data can filter by the account linked to a source.""" |
| 137 | + sensor = setup_api_test_data["some gas sensor"] |
| 138 | + source_user = db.session.get(User, setup_roles_users["Test Supplier User"]) |
| 139 | + assert source_user.account_id is not None |
| 140 | + message = { |
| 141 | + "start": "2021-05-02T00:00:00+02:00", |
| 142 | + "duration": "PT1H20M", |
| 143 | + "horizon": "PT0H", |
| 144 | + "unit": "m³/h", |
| 145 | + "account": source_user.account_id, |
| 146 | + "resolution": "PT20M", |
| 147 | + } |
| 148 | + response = client.get( |
| 149 | + url_for("SensorAPI:get_data", id=sensor.id), |
| 150 | + query_string=message, |
| 151 | + ) |
| 152 | + print("Server responded with:\n%s" % response.json) |
| 153 | + assert response.status_code == 200 |
| 154 | + values = response.json["values"] |
| 155 | + # The fixture also stores data from an accountless "Other source". |
| 156 | + # Filtering by the user account should exclude those points. |
| 157 | + # GAS_MEASUREMENTS_10MIN stores 10-minute values; resampled to 20-minute resolution: |
| 158 | + # - 1st interval: average of [91.3, 91.7] = 91.5 |
| 159 | + # - 2nd interval: average of [92.1, None] = 92.1 (only one value present) |
| 160 | + expected = [ |
| 161 | + sum(GAS_MEASUREMENTS_10MIN[0:2]) / 2, # 91.5 |
| 162 | + GAS_MEASUREMENTS_10MIN[2], # 92.1 |
| 163 | + None, |
| 164 | + None, |
| 165 | + ] |
| 166 | + assert all(a == b for a, b in zip(values, expected)) |
| 167 | + |
| 168 | + |
117 | 169 | @pytest.mark.parametrize( |
118 | 170 | "requesting_user, status_code", |
119 | 171 | [ |
|
0 commit comments