24222694482

Committed 10 Apr 2026 02:03AM UTC coverage: 88.95% (-0.006%) from 88.956%

Build # 24222694482

Build Type

Pull #1351

github

Committed by

web-flow

Commit Message

Merge f4be4f05e into ad380fcdb

Pull Request Pull Request #1351: RA Toolkit updates

Coverage Stats

496 of 574 new or added lines in 27 files covered. (86.41%)

1 existing line in 1 file now uncovered.

27869 of 31331 relevant lines covered (88.95%)

0.89 hits per line

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

89.66

/tests/test_data_toolkit/system/test_create_sync_load_input_csvs.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 unittest

from db.create_database import main as create_database_main
from data_toolkit.load_raw_data import main as load_raw_data_main
from data_toolkit.system.create_sync_load_input_csvs import (
    main as create_sync_load_input_csvs_main,
)


class TestCreateSyncLoadInputCsvs(unittest.TestCase):
    """
    Test create_sync_load_input_csvs script
    """

    @classmethod
    def setUpClass(cls):
        """Set up test environment"""
        os.chdir(os.path.join(os.path.dirname(__file__), "..", "..", "..", "db"))
        cls.db_path = "ra_toolkit_test_steps_temp.db"

        # Clean up temp database if it exists
        if os.path.exists(cls.db_path):
            os.remove(cls.db_path)

        # Create database first
        create_db_args = [
            "--database",
            cls.db_path,
            "--db_schema",
            "../data_toolkit/raw_data_db_schema.sql",
            "--quiet",
        ]
        create_database_main(create_db_args)

        # Load raw data
        load_data_args = [
            "--database",
            cls.db_path,
            "--csv_location",
            "./csvs_test_examples/raw_data_ra_toolkit/",
            "--quiet",
        ]
        load_raw_data_main(load_data_args)

    def test_create_sync_load_input_csvs(self):
        """Test create_sync_load_input_csvs with hardcoded arguments"""
        args = [
            "--database",
            self.db_path,
            "--output_directory",
            "./csvs_test_examples/system_load/system_load",
            "--load_scenario_id",
            "13",
            "--load_scenario_name",
            "ra_toolkit_module_tests_sync",
            "--load_components_scenario_id",
            "7",
            "--load_components_scenario_name",
            "ra_toolkit_module_tests_sync",
            "--load_levels_scenario_id",
            "14",
            "--load_levels_scenario_name",
            "ra_toolkit_module_tests_sync",
            "--load_levels_overwrite",
            "--load_components_overwrite",
            "--load_scenario_overwrite",
            "--quiet",
        ]
        create_sync_load_input_csvs_main(args)

    @classmethod
    def tearDownClass(cls):
        """Clean up test database"""
        if os.path.exists(cls.db_path):
            os.remove(cls.db_path)
        for temp_file_ext in ["-shm", "-wal"]:
            temp_file = f"{cls.db_path}{temp_file_ext}"
            if os.path.exists(temp_file):
                os.remove(temp_file)


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	1✔
16	import unittest	1✔
17
18	from db.create_database import main as create_database_main	1✔
19	from data_toolkit.load_raw_data import main as load_raw_data_main	1✔
20	from data_toolkit.system.create_sync_load_input_csvs import (	1✔
21	main as create_sync_load_input_csvs_main,
22	)
23
24
25	class TestCreateSyncLoadInputCsvs(unittest.TestCase):	1✔
26	"""
27	Test create_sync_load_input_csvs script
28	"""
29
30	@classmethod	1✔
31	def setUpClass(cls):	1✔
32	"""Set up test environment"""
33	os.chdir(os.path.join(os.path.dirname(__file__), "..", "..", "..", "db"))	1✔
34	cls.db_path = "ra_toolkit_test_steps_temp.db"	1✔
35
36	# Clean up temp database if it exists
37	if os.path.exists(cls.db_path):	1✔
NEW 38	os.remove(cls.db_path)	×
39
40	# Create database first
41	create_db_args = [	1✔
42	"--database",
43	cls.db_path,
44	"--db_schema",
45	"../data_toolkit/raw_data_db_schema.sql",
46	"--quiet",
47	]
48	create_database_main(create_db_args)	1✔
49
50	# Load raw data
51	load_data_args = [	1✔
52	"--database",
53	cls.db_path,
54	"--csv_location",
55	"./csvs_test_examples/raw_data_ra_toolkit/",
56	"--quiet",
57	]
58	load_raw_data_main(load_data_args)	1✔
59
60	def test_create_sync_load_input_csvs(self):	1✔
61	"""Test create_sync_load_input_csvs with hardcoded arguments"""
62	args = [	1✔
63	"--database",
64	self.db_path,
65	"--output_directory",
66	"./csvs_test_examples/system_load/system_load",
67	"--load_scenario_id",
68	"13",
69	"--load_scenario_name",
70	"ra_toolkit_module_tests_sync",
71	"--load_components_scenario_id",
72	"7",
73	"--load_components_scenario_name",
74	"ra_toolkit_module_tests_sync",
75	"--load_levels_scenario_id",
76	"14",
77	"--load_levels_scenario_name",
78	"ra_toolkit_module_tests_sync",
79	"--load_levels_overwrite",
80	"--load_components_overwrite",
81	"--load_scenario_overwrite",
82	"--quiet",
83	]
84	create_sync_load_input_csvs_main(args)	1✔
85
86	@classmethod	1✔
87	def tearDownClass(cls):	1✔
88	"""Clean up test database"""
89	if os.path.exists(cls.db_path):	1✔
90	os.remove(cls.db_path)	1✔
91	for temp_file_ext in ["-shm", "-wal"]:	1✔
92	temp_file = f"{cls.db_path}{temp_file_ext}"	1✔
93	if os.path.exists(temp_file):	1✔
NEW 94	os.remove(temp_file)	×
95
96
97	if __name__ == "__main__":	1✔
NEW 98	unittest.main()	×

blue-marble / gridpath / 24222694482

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