Skip to content
Snippets Groups Projects
Select Git revision
  • 10500dadcfb208f8b50c9484181092a6e48d7e66
  • master default protected
  • enxhi_issue460_remove_TOAR-I_access
  • michael_issue459_preprocess_german_stations
  • sh_pollutants
  • develop protected
  • release_v2.4.0
  • michael_issue450_feat_load-ifs-data
  • lukas_issue457_feat_set-config-paths-as-parameter
  • lukas_issue454_feat_use-toar-statistics-api-v2
  • lukas_issue453_refac_advanced-retry-strategy
  • lukas_issue452_bug_update-proj-version
  • lukas_issue449_refac_load-era5-data-from-toar-db
  • lukas_issue451_feat_robust-apriori-estimate-for-short-timeseries
  • lukas_issue448_feat_load-model-from-path
  • lukas_issue447_feat_store-and-load-local-clim-apriori-data
  • lukas_issue445_feat_data-insight-plot-monthly-distribution
  • lukas_issue442_feat_bias-free-evaluation
  • lukas_issue444_feat_choose-interp-method-cams
  • 414-include-crps-analysis-and-other-ens-verif-methods-or-plots
  • lukas_issue384_feat_aqw-data-handler
  • v2.4.0 protected
  • v2.3.0 protected
  • v2.2.0 protected
  • v2.1.0 protected
  • Kleinert_etal_2022_initial_submission
  • v2.0.0 protected
  • v1.5.0 protected
  • v1.4.0 protected
  • v1.3.0 protected
  • v1.2.1 protected
  • v1.2.0 protected
  • v1.1.0 protected
  • IntelliO3-ts-v1.0_R1-submit
  • v1.0.0 protected
  • v0.12.2 protected
  • v0.12.1 protected
  • v0.12.0 protected
  • v0.11.0 protected
  • v0.10.0 protected
  • IntelliO3-ts-v1.0_initial-submit
41 results

test_join.py

Blame
  • user avatar
    lukas leufen authored
    609a4a7c
    History
    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"