14431500873

Committed 09 Apr 2025 10:33PM UTC coverage: 89.009%. Remained the same

Build # 14431500873

Build Type

push

github

Committed by

anamileva

Commit Message

Catch empty cursor description

Coverage Stats

26562 of 29842 relevant lines covered (89.01%)

2.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.92

/tests/test_run_data_toolkit.py

# Copyright 2016-2024 Blue Marble Analytics LLC.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import pandas as pd
import unittest

from data_toolkit import run_data_toolkit

RA_SETTINGS_CSV = "../tests/test_data/data_toolkit_ra_settings.csv"
RA_SETTINGS_STEPS_CSV = "../tests/test_data/data_toolkit_ra_settings_steps.csv"
OPEN_DATA_SETTINGS_CSV = "../tests/test_data/data_toolkit_open_data_settings" ".csv"


class TestDataToolkit(unittest.TestCase):
    """
    Check if the database is created with no errors.
    """

    @classmethod
    def setUpClass(cls):
        """
        Set up the testing database
        :return:
        """
        os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))
        temp_db_paths = get_temp_db_paths()

        for p in temp_db_paths:
            if os.path.exists(p):
                os.remove(p)

    def test_data_toolkit_ra_steps(self):
        os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))
        for step in [
            "create_database",
            "load_raw_data",
            "create_sync_load_input_csvs",
            "create_sync_var_gen_input_csvs",
            "create_monte_carlo_weather_draws",
            "create_monte_carlo_load_input_csvs",
            "create_monte_carlo_var_gen_input_csvs",
            "create_hydro_iteration_input_csvs",
            "create_availability_iteration_input_csvs",
            "create_sync_gen_weather_derate_input_csvs",
            "create_monte_carlo_gen_weather_derate_input_csvs",
            "create_temporal_scenarios",
        ]:
            run_data_toolkit.main(
                [
                    "--settings_csv",
                    RA_SETTINGS_STEPS_CSV,
                    "--quiet",
                    "--single_step_only",
                    step,
                ]
            )

    def test_data_toolkit_ra(self):
        os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))
        run_data_toolkit.main(["--settings_csv", RA_SETTINGS_CSV, "--quiet"])

    def test_data_toolkit_open_data(self):
        os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))
        run_data_toolkit.main(["--settings_csv", OPEN_DATA_SETTINGS_CSV, "--quiet"])

    @classmethod
    def tearDownClass(cls):
        temp_db_paths = get_temp_db_paths()

        for p in temp_db_paths:
            if os.path.exists(p):
                os.remove(p)
            for temp_file_ext in ["-shm", "-wal"]:
                temp_file = "{}{}".format(p, temp_file_ext)
                if os.path.exists(temp_file):
                    os.remove(temp_file)


def get_temp_db_paths():
    ra_settings_df = pd.read_csv(RA_SETTINGS_CSV)
    ra_settings_df.set_index(["script", "setting"])
    settings_db_path = os.path.join(
        os.getcwd(),
        run_data_toolkit.get_setting(ra_settings_df, "create_database", "database"),
    )

    ra_settings_steps_df = pd.read_csv(RA_SETTINGS_STEPS_CSV)
    ra_settings_steps_df.set_index(["script", "setting"])
    settings_steps_db_path = os.path.join(
        os.getcwd(),
        run_data_toolkit.get_setting(
            ra_settings_steps_df, "create_database", "database"
        ),
    )

    open_data_settings_df = pd.read_csv(OPEN_DATA_SETTINGS_CSV)
    open_data_settings_df.set_index(["script", "setting"])
    open_data_settings_db_path = os.path.join(
        os.getcwd(),
        run_data_toolkit.get_setting(
            open_data_settings_df, "create_database", "database"
        ),
    )

    temp_db_paths = [
        settings_db_path,
        settings_steps_db_path,
        open_data_settings_db_path,
    ]

    return temp_db_paths


if __name__ == "__main__":
    unittest.main()

1	# Copyright 2016-2024 Blue Marble Analytics LLC.
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# http://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	import os	3✔
16	import pandas as pd	3✔
17	import unittest	3✔
18
19	from data_toolkit import run_data_toolkit	3✔
20
21	RA_SETTINGS_CSV = "../tests/test_data/data_toolkit_ra_settings.csv"	3✔
22	RA_SETTINGS_STEPS_CSV = "../tests/test_data/data_toolkit_ra_settings_steps.csv"	3✔
23	OPEN_DATA_SETTINGS_CSV = "../tests/test_data/data_toolkit_open_data_settings" ".csv"	3✔
24
25
26	class TestDataToolkit(unittest.TestCase):	3✔
27	"""
28	Check if the database is created with no errors.
29	"""
30
31	@classmethod	3✔
32	def setUpClass(cls):	3✔
33	"""
34	Set up the testing database
35	:return:
36	"""
37	os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))	3✔
38	temp_db_paths = get_temp_db_paths()	3✔
39
40	for p in temp_db_paths:	3✔
41	if os.path.exists(p):	3✔
42	os.remove(p)	×
43
44	def test_data_toolkit_ra_steps(self):	3✔
45	os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))	3✔
46	for step in [	3✔
47	"create_database",
48	"load_raw_data",
49	"create_sync_load_input_csvs",
50	"create_sync_var_gen_input_csvs",
51	"create_monte_carlo_weather_draws",
52	"create_monte_carlo_load_input_csvs",
53	"create_monte_carlo_var_gen_input_csvs",
54	"create_hydro_iteration_input_csvs",
55	"create_availability_iteration_input_csvs",
56	"create_sync_gen_weather_derate_input_csvs",
57	"create_monte_carlo_gen_weather_derate_input_csvs",
58	"create_temporal_scenarios",
59	]:
60	run_data_toolkit.main(	3✔
61	[
62	"--settings_csv",
63	RA_SETTINGS_STEPS_CSV,
64	"--quiet",
65	"--single_step_only",
66	step,
67	]
68	)
69
70	def test_data_toolkit_ra(self):	3✔
71	os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))	3✔
72	run_data_toolkit.main(["--settings_csv", RA_SETTINGS_CSV, "--quiet"])	3✔
73
74	def test_data_toolkit_open_data(self):	3✔
75	os.chdir(os.path.join(os.path.dirname(__file__), "..", "db"))	3✔
76	run_data_toolkit.main(["--settings_csv", OPEN_DATA_SETTINGS_CSV, "--quiet"])	3✔
77
78	@classmethod	3✔
79	def tearDownClass(cls):	3✔
80	temp_db_paths = get_temp_db_paths()	3✔
81
82	for p in temp_db_paths:	3✔
83	if os.path.exists(p):	3✔
84	os.remove(p)	3✔
85	for temp_file_ext in ["-shm", "-wal"]:	3✔
86	temp_file = "{}{}".format(p, temp_file_ext)	3✔
87	if os.path.exists(temp_file):	3✔
88	os.remove(temp_file)	1✔
89
90
91	def get_temp_db_paths():	3✔
92	ra_settings_df = pd.read_csv(RA_SETTINGS_CSV)	3✔
93	ra_settings_df.set_index(["script", "setting"])	3✔
94	settings_db_path = os.path.join(	3✔
95	os.getcwd(),
96	run_data_toolkit.get_setting(ra_settings_df, "create_database", "database"),
97	)
98
99	ra_settings_steps_df = pd.read_csv(RA_SETTINGS_STEPS_CSV)	3✔
100	ra_settings_steps_df.set_index(["script", "setting"])	3✔
101	settings_steps_db_path = os.path.join(	3✔
102	os.getcwd(),
103	run_data_toolkit.get_setting(
104	ra_settings_steps_df, "create_database", "database"
105	),
106	)
107
108	open_data_settings_df = pd.read_csv(OPEN_DATA_SETTINGS_CSV)	3✔
109	open_data_settings_df.set_index(["script", "setting"])	3✔
110	open_data_settings_db_path = os.path.join(	3✔
111	os.getcwd(),
112	run_data_toolkit.get_setting(
113	open_data_settings_df, "create_database", "database"
114	),
115	)
116
117	temp_db_paths = [	3✔
118	settings_db_path,
119	settings_steps_db_path,
120	open_data_settings_db_path,
121	]
122
123	return temp_db_paths	3✔
124
125
126	if __name__ == "__main__":	3✔
127	unittest.main()	×

blue-marble / gridpath / 14431500873

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous