Select Git revision
test_join.py
test_join.py 5.41 KiB
from typing import Iterable
import pytest
from src.join import *
from src.join import _save_to_pandas, _correct_stat_name, _lower_list
from src.join_settings import join_settings
class TestJoinUrlBase:
def test_url(self):
url, headers = join_settings()
assert url == 'https://join.fz-juelich.de/services/rest/surfacedata/'
assert headers == {}
class TestDownloadJoin:
def test_download_single_var(self):
data, meta = download_join("DEBW107", {"o3": "dma8eu"})
assert data.columns == "o3"
assert meta.columns == "DEBW107"
def test_download_empty(self):
with pytest.raises(EmptyQueryResult) as e:
download_join("DEBW107", {"o3": "dma8eu"}, "traffic")
assert e.value.args[-1] == "No data found in JOIN."
class TestCorrectDataFormat:
def test_correct_data_format(self):
list_data = [["2020-01-01 06:00:01", 23.], ["2020-01-01 06:00:11", 24.], ["2020-01-01 06:00:21", 25.],
["2020-01-01 06:00:31", 26.], ["2020-01-01 06:00:41", 27.], ["2020-01-01 06:00:51", 23.],
{"station": "test_station_001", "author": "ME", "success": True}]
dict_data = correct_data_format(list_data)
assert dict_data == {"datetime": ["2020-01-01 06:00:01", "2020-01-01 06:00:11", "2020-01-01 06:00:21",
"2020-01-01 06:00:31", "2020-01-01 06:00:41", "2020-01-01 06:00:51"],
"values": [23., 24., 25., 26., 27., 23.],
"metadata": {"station": "test_station_001", "author": "ME", "success": True}}
class TestGetData:
def test(self):
opts = {"base": join_settings()[0], "service": "series", "station_id": 'DEBW107', "network_name": "UBA",
"parameter_name": "o3,no2"}
assert get_data(opts, headers={}) == [[17057, 'UBA', 'DEBW107', 'O3'], [17058, 'UBA', 'DEBW107', 'NO2']]
class TestLoadSeriesInformation:
def test_standard_query(self):
expected_subset = {'o3': 23031, 'no2': 39002, 'temp--lubw': 17059, 'wspeed': 17060}
assert expected_subset.items() <= load_series_information(['DEBW107'], None, None, join_settings()[0], {}).items()
def test_empty_result(self):
assert load_series_information(['DEBW107'], "traffic", None, join_settings()[0], {}) == {}
class TestSaveToPandas:
@staticmethod
def convert_date(date):
return map(lambda s: dt.datetime.strptime(s, "%Y-%m-%d %H:%M"), date)
@pytest.fixture
def date(self):
return ['1997-01-01 00:00', '1997-01-02 00:00', '1997-01-03 00:00', '1997-01-04 00:00']
@pytest.fixture
def date_len19(self):
return ['1997-01-01 00:00:00', '1997-01-02 00:00:00', '1997-01-03 00:00:00', '1997-01-04 00:00:00']
@pytest.fixture
def values(self):
return [86.21, 94.76, 76.96, 99.89]
@pytest.fixture
def alternative_values(self):
return [20.0, 25.2, 25.1, 23.6]
@pytest.fixture
def create_df(self, date, values):
return pd.DataFrame(values, index=self.convert_date(date), columns=['cloudcover'])
def test_empty_df(self, date, values, create_df):
data = {'datetime': date, 'mean': values, 'metadata': None}
assert pd.testing.assert_frame_equal(create_df, _save_to_pandas(None, data, 'mean', 'cloudcover')) is None
def test_not_empty_df(self, date, alternative_values, create_df):
data = {'datetime': date, 'max': alternative_values, 'metadata': None}
next_df = pd.DataFrame(data["max"], index=self.convert_date(date), columns=['temperature'])
df_concat = pd.concat([create_df, next_df], axis=1)
assert pd.testing.assert_frame_equal(df_concat, _save_to_pandas(create_df, data, 'max', 'temperature')) is None
def test_alternative_date_format(self, date_len19, values, create_df):
data = {'datetime': date_len19, 'mean': values, 'metadata': None}
assert pd.testing.assert_frame_equal(create_df, _save_to_pandas(None, data, 'mean', 'cloudcover')) is None
class TestCorrectStatName:
def test_nothing_to_do(self):
assert _correct_stat_name("dma8eu") == "dma8eu"
assert _correct_stat_name("max") == "max"
def test_correct_string(self):
assert _correct_stat_name("maximum") == "max"
assert _correct_stat_name("minimum") == "min"
assert _correct_stat_name("average_values") == "mean"
class TestLowerList:
def test_string_lowering(self):
list_iterator = _lower_list(["Capitalised", "already_small", "UPPERCASE", "veRyStRaNGe"])
assert isinstance(list_iterator, Iterable)
assert list(list_iterator) == ["capitalised", "already_small", "uppercase", "verystrange"]
class TestCreateUrl:
def test_minimal_args_given(self):
url = create_url("www.base.edu", "testingservice")
assert url == "www.base.edu/testingservice/?"
def test_given_kwargs(self):
url = create_url("www.base2.edu/", "testingservice", mood="happy", confidence=0.98)
assert url == "www.base2.edu/testingservice/?mood=happy&confidence=0.98"
def test_single_kwargs(self):
url = create_url("www.base2.edu/", "testingservice", mood="undefined")
assert url == "www.base2.edu/testingservice/?mood=undefined"
def test_none_kwargs(self):
url = create_url("www.base2.edu/", "testingservice", mood="sad", happiness=None, stress_factor=100)
assert url == "www.base2.edu/testingservice/?mood=sad&stress_factor=100"