Add new tests

Author: uvchik

tests/test_data_handling.py

@@ -3,6 +3,72 @@
 SPDX-License-Identifier: MIT
 """
 
+import os
+from datetime import datetime
+from shutil import copyfile
+
+import pandas as pd
+import pytest
+
+from windpowerlib.data import check_imported_data, check_data_integretiy
+
+
+class TestDataCheck:
+    @classmethod
+    def setup_class(cls):
+        cls.path = os.path.join(os.path.dirname(__file__), "oedb")
+        cls.filename = os.path.join(cls.path, "{0}.csv")
+        cls.df = pd.read_csv(cls.filename.format("turbine_data"), index_col=0)
+        cls.time_stamp = datetime.now().strftime("%Y%m%d%H%M%S")
+        cls.broken_fn = os.path.join(cls.path, "power_curves_broken.csv")
+        cls.backup_fn = os.path.join(cls.path, "power_curves_backup.csv")
+        cls.orig_path = os.path.join(os.path.dirname(__file__), os.pardir,
+                                     "windpowerlib", "oedb")
+        cls.orig_fn = os.path.join(cls.orig_path, "power_curves.csv")
+
+    @classmethod
+    def teardown_class(cls):
+        cls.path = os.path.join(os.path.dirname(__file__), "oedb")
+        cls.backup_fn = os.path.join(cls.path, "power_curves_backup.csv")
+        orig_path = os.path.join(os.path.dirname(__file__), os.pardir,
+                                 "windpowerlib", "oedb")
+        cls.orig_fn = os.path.join(orig_path, "power_curves.csv")
+        copyfile(cls.backup_fn, cls.orig_fn)
+        for f in os.listdir(cls.path):
+            if "error" in f:
+                os.remove(os.path.join(cls.path, f))
+
+    def test_normal_data_check(self):
+        copyfile(
+            self.filename.format("turbine_data"),
+            self.filename.format("turbine_data_{0}".format(self.time_stamp)),
+        )
+        for curve_type in ["power_curve", "power_coefficient_curve"]:
+            copyfile(
+                self.filename.format("{}s".format(curve_type)),
+                self.filename.format("{0}s_{1}".format(curve_type,
+                                                       self.time_stamp)),
+            )
+        check_imported_data(self.df, self.filename, self.time_stamp)
+
+    def test_data_check_logging_warnings(self, caplog):
+        self.df.loc["GE158/4800", "has_power_curve"] = True
+        self.df.loc["GE100/2750", "has_cp_curve"] = True
+        check_data_integretiy(self.df, min_pc_length=26)
+        assert "E48/800: power_curve is to short (25 values)" in caplog.text
+        assert "GE158/4800: No power curve" in caplog.text
+        assert "GE100/2750: No cp-curve but has_cp_curve" in caplog.text
+
+    def test_global_error(self):
+        msg = "Must have equal len keys"
+        with pytest.raises(ValueError, match=msg):
+            self.df.loc["GE158/4800", "has_cp_curve"] = [5, 3]
+            check_imported_data(self.df, self.filename, self.time_stamp)
+
+    def test_broken_pwr_curve(self):
+        copyfile(self.orig_fn, self.backup_fn)
+        copyfile(self.broken_fn, self.orig_fn)
+        msg = "could not convert string to float"
+        with pytest.raises(ValueError, match=msg):
+            check_imported_data(self.df, self.filename, self.time_stamp)
 
-def test_integretiy_check():
-    pass

windpowerlib/data.py

@@ -5,9 +5,9 @@
 SPDX-License-Identifier: MIT
 """
 
+import logging
 import os
 import warnings
-import logging
 from datetime import datetime
 from shutil import copyfile
 
@@ -184,7 +184,7 @@ def store_turbine_data_from_oedb(
     """
     turbine_data = fetch_turbine_data_from_oedb(schema=schema, table=table)
     # standard file name for saving data
-    filename = os.path.join(os.path.dirname(__file__), "oedb", "{}.csv")
+    filename = os.path.join(os.path.dirname(__file__), "oedb", "{0}.csv")
 
     time_stamp = datetime.now().strftime("%Y%m%d%H%M%S")
     # get all power (coefficient) curves and save to file
@@ -260,15 +260,10 @@ def store_turbine_data_from_oedb(
 def remove_tmp_file(filename, time_stamp):
     os.remove(filename.format("turbine_data_{0}".format(time_stamp)))
     for curve_type in ["power_curve", "power_coefficient_curve"]:
-        copyfile(
-            filename.format("{0}s_{1}".format(curve_type, time_stamp)),
-            filename.format("{}s".format(curve_type)),
-        )
         os.remove(filename.format("{0}s_{1}".format(curve_type, time_stamp)))
 
 
 def check_imported_data(data, filename, time_stamp):
-    data.to_csv(filename.format("turbine_data"))
     try:
         data = check_data_integretiy(data)
     except Exception as e:
@@ -296,7 +291,7 @@ def check_imported_data(data, filename, time_stamp):
     return data
 
 
-def check_data_integretiy(data):
+def check_data_integretiy(data, min_pc_length=5):
     for dataset in data.iterrows():
         ttype = dataset[0]
         enercon_e126 = {"turbine_type": "{0}".format(ttype), "hub_height": 135}
@@ -316,8 +311,8 @@ def check_data_integretiy(data):
                 logging.warning(
                     "{0}: No cp-curve but has_cp_curve=True.".format(ttype)
                 )
-            if dataset[1].has_power_curve is True:
-                if len(wt.power_curve) < 5:
+            if wt.power_curve is not None:
+                if len(wt.power_curve) < min_pc_length:
                     logging.warning(
                         "{0}: power_curve is to short ({1} values),".format(
                             ttype, len(wt.power_curve)