test: draft of moving unit tests to canonical test directory

Author: fleimgruber

tests/__init__.py

@@ -0,0 +1,113 @@
+import unittest
+
+import pandas as pd
+from bs4 import BeautifulSoup
+
+from entsoe import EntsoeRawClient, EntsoePandasClient
+from entsoe.exceptions import NoMatchingDataError
+from settings import api_key
+
+
+class EntsoeRawClientTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.client = EntsoeRawClient(api_key=api_key)
+        cls.start = pd.Timestamp('20180101', tz='Europe/Brussels')
+        cls.end = pd.Timestamp('20180107', tz='Europe/Brussels')
+        cls.country_code = 'BE'
+
+    def test_datetime_to_str(self):
+        start_str = self.client._datetime_to_str(dtm=self.start)
+        self.assertIsInstance(start_str, str)
+        self.assertEqual(start_str, '201712312300')
+
+    def test_basic_queries(self):
+        queries = [
+            self.client.query_day_ahead_prices,
+            self.client.query_load,
+            self.client.query_wind_and_solar_forecast,
+            self.client.query_load_forecast,
+            self.client.query_generation,
+            self.client.query_generation_forecast,
+            self.client.query_installed_generation_capacity,
+            self.client.query_imbalance_prices
+        ]
+        for query in queries:
+            text = query(country_code=self.country_code, start=self.start,
+                         end=self.end)
+            self.assertIsInstance(text, str)
+            try:
+                BeautifulSoup(text, 'html.parser')
+            except Exception as e:
+                self.fail(f'Parsing of response failed with exception: {e}')
+
+    def query_crossborder_flows(self):
+        text = self.client.query_crossborder_flows(
+            country_code_from='BE', country_code_to='NL', start=self.start,
+            end=self.end)
+        self.assertIsInstance(text, str)
+        try:
+            BeautifulSoup(text, 'html.parser')
+        except Exception as e:
+            self.fail(f'Parsing of response failed with exception: {e}')
+
+    def test_query_unavailability_of_generation_units(self):
+        text = self.client.query_unavailability_of_generation_units(
+            country_code='BE', start=self.start,
+            end=self.end)
+        self.assertIsInstance(text, bytes)
+
+    def test_query_withdrawn_unavailability_of_generation_units(self):
+        with self.assertRaises(NoMatchingDataError):
+            self.client.query_withdrawn_unavailability_of_generation_units(
+                country_code='BE', start=self.start, end=self.end)
+
+
+class EntsoePandasClientTest(EntsoeRawClientTest):
+    @classmethod
+    def setUpClass(cls):
+        cls.client = EntsoePandasClient(api_key=api_key)
+        cls.start = pd.Timestamp('20180101', tz='Europe/Brussels')
+        cls.end = pd.Timestamp('20180107', tz='Europe/Brussels')
+        cls.country_code = 'BE'
+
+    def test_basic_queries(self):
+        pass
+
+    def test_basic_series(self):
+        queries = [
+            self.client.query_day_ahead_prices,
+            self.client.query_load,
+            self.client.query_load_forecast,
+            self.client.query_generation_forecast
+        ]
+        for query in queries:
+            ts = query(country_code=self.country_code, start=self.start,
+                       end=self.end)
+            self.assertIsInstance(ts, pd.Series)
+
+    def query_crossborder_flows(self):
+        ts = self.client.query_crossborder_flows(
+            country_code_from='BE', country_code_to='NL', start=self.start,
+            end=self.end)
+        self.assertIsInstance(ts, pd.Series)
+
+    def test_basic_dataframes(self):
+        queries = [
+            self.client.query_wind_and_solar_forecast,
+            self.client.query_generation,
+            self.client.query_installed_generation_capacity,
+            self.client.query_imbalance_prices,
+            self.client.query_unavailability_of_generation_units,
+        ]
+        for query in queries:
+            ts = query(country_code=self.country_code, start=self.start,
+                       end=self.end)
+            self.assertIsInstance(ts, pd.DataFrame)
+
+    def test_query_unavailability_of_generation_units(self):
+        pass
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_clients.py

@@ -0,0 +1,48 @@
+import pandas as pd
+from settings import api_key
+from entsoe import EntsoeRawClient as Entsoe
+
+
+def test_czech_unavailability_of_production_units():
+    e = Entsoe(api_key=api_key, retry_count=20, retry_delay=30)
+
+    start = pd.Timestamp("20170601", tz="Europe/Brussels")
+    end = pd.Timestamp("20170603", tz="Europe/Brussels")
+
+    # s = e.query_imbalance_prices(country_code='BE', start=start, end=end, as_dataframe=True)
+
+    """domains = [["10YIT-GRTN-----B", "Italy, IT CA / MBA"],
+            ["10Y1001A1001A885", "Italy_Saco_AC"],
+            ["10Y1001A1001A893", "Italy_Saco_DC"],
+            ["10Y1001A1001A699", "IT-Brindisi BZ"],
+            ["10Y1001A1001A70O", "IT-Centre-North BZ"],
+            ["10Y1001A1001A71M", "IT-Centre-South BZ"],
+            ["10Y1001A1001A72K", "IT-Foggia BZ"],
+            ["10Y1001A1001A66F", "IT-GR BZ"],
+            ["10Y1001A1001A84D", "IT-MACROZONE NORTH MBA"],
+            ["10Y1001A1001A85B", "IT-MACROZONE SOUTH MBA"],
+            ["10Y1001A1001A877", "IT-Malta BZ"],
+            ["10Y1001A1001A73I", "IT-North BZ"],
+            ["10Y1001A1001A80L", "IT-North-AT BZ"],
+            ["10Y1001A1001A68B", "IT-North-CH BZ"],
+            ["10Y1001A1001A81J", "IT-North-FR BZ"],
+            ["10Y1001A1001A67D", "IT-North-SI BZ"],
+            ["10Y1001A1001A76C", "IT-Priolo BZ"],
+            ["10Y1001A1001A77A", "IT-Rossano BZ"],
+            ["10Y1001A1001A74G", "IT-Sardinia BZ"],
+            ["10Y1001A1001A75E", "IT-Sicily BZ"],
+            ["10Y1001A1001A788", "IT-South BZ"]]
+    """
+
+    domains = [["CZ", "Czech bidding zone"]]
+
+    lst = []
+    for bzn in domains:
+        s = e.query_unavailability_of_production_units(
+            country_code=bzn[0], docstatus=None, start=start, end=end
+        )
+        if s is not None:
+            lst.append(s)
+
+    result = pd.concat(lst)
+    result.to_csv("result.csv")