21403831778

Committed 27 Jan 2026 03:46PM UTC coverage: 24.281%. First build

Build # 21403831778

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #83 from CBIIT/unit-testing

Unit Testing & Repository Standardization

Run Details

93 of 145 new or added lines in 4 files covered. (64.14%)

1140 of 4695 relevant lines covered (24.28%)

0.49 hits per line

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

88.18

/tests/test_loader.py

import unittest
import os
from unittest.mock import Mock, MagicMock, patch
from bento.common.utils import get_logger, removeTrailingSlash, UUID
from data_loader import DataLoader
from icdc_schema import ICDC_Schema
from props import Props
from neo4j import Driver


class MockNeo4jDriver(Driver):
    """Mock Neo4j driver for testing without a database connection"""
    def __init__(self):
        # Don't call super().__init__ to avoid actual connection
        pass
    
    def session(self, **kwargs):
        session = Mock()
        session.__enter__ = Mock(return_value=session)
        session.__exit__ = Mock(return_value=False)
        return session
    
    def close(self):
        pass
    
    def verify_connectivity(self):
        return True


class TestLoader(unittest.TestCase):
    def setUp(self):
        # Use mock Neo4j driver instead of requiring environment variable
        self.driver = MockNeo4jDriver()
        
        # Use paths relative to the tests directory
        test_dir = os.path.dirname(os.path.abspath(__file__))
        self.data_folder = os.path.join(test_dir, 'data', 'COTC007B')
        props_path = os.path.join(test_dir, 'data', 'props-icdc.yml')
        model_path = os.path.join(test_dir, 'data', 'icdc-model.yml')
        model_props_path = os.path.join(test_dir, 'data', 'icdc-model-props.yml')
        
        props = Props(props_path)
        self.schema = ICDC_Schema([model_path, model_props_path], props)
        self.log = get_logger('Test Loader')
        self.loader = DataLoader(self.driver, self.schema)
        self.file_list = [
            os.path.join(test_dir, "data/Dataset/COP-program.txt"),
            os.path.join(test_dir, "data/Dataset/NCATS-COP01-case.txt"),
            os.path.join(test_dir, "data/Dataset/NCATS-COP01-diagnosis.txt"),
            os.path.join(test_dir, "data/Dataset/NCATS-COP01_cohort_file.txt"),
            os.path.join(test_dir, "data/Dataset/NCATS-COP01_study_file.txt")
        ]

    def test_remove_traling_slash(self):
        self.assertEqual('abc', removeTrailingSlash('abc/'))
        self.assertEqual('abc', removeTrailingSlash('abc'))
        self.assertEqual('abc', removeTrailingSlash('abc//'))
        self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78'))
        self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78/'))
        self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78//'))
        self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78////'))

    def test_loader_construction(self):
        self.assertRaises(Exception, DataLoader, None, None, None)
        self.assertRaises(Exception, DataLoader, self.driver, None, None)
        self.assertIsInstance(self.loader, DataLoader)

    @unittest.skip("Skipping test that requires actual Neo4j database interaction")
    def test_validate_parents_exist_in_file(self):
        test_dir = os.path.dirname(os.path.abspath(__file__))
        load_result = self.loader.load(self.file_list, True, False, 'upsert', False, 1, '/tmp', False)
        self.assertIsInstance(load_result, dict, msg='Load data failed!')
        result = self.loader.validate_parents_exist_in_file(os.path.join(test_dir, 'data/pathology-reports-failure.txt'), 100)
        self.assertFalse(result)
        result = self.loader.validate_parents_exist_in_file(os.path.join(test_dir, 'data/pathology-reports-success.txt'), 100)
        self.assertTrue(result)

    def test_duplicated_ids(self):
        test_dir = os.path.dirname(os.path.abspath(__file__))
        self.assertTrue(self.loader.validate_file(os.path.join(test_dir, 'data/Dataset/NCATS-COP01-case.txt'), 10, False))
        self.assertFalse(self.loader.validate_file(os.path.join(test_dir, 'data/NCATS01-case-dup.txt'), 10, False))

    def test_get_signature(self):
        self.assertEqual(self.loader.get_signature({}), '{  }')
        self.assertEqual(self.loader.get_signature({'key1': 'value1'}), '{ key1: value1 }')
        self.assertEqual(self.loader.get_signature({'key1': 'value1', 'key2': 'value2'}), '{ key1: value1, key2: value2 }')

    def test_cleanup_node(self):
        #Test UUIDs
        self.assertRaises(SystemExit, self.loader.prepare_node, {}, 'test_file.txt')
        result = self.loader.prepare_node({'type': 'case', 'case_id': '123', ' key1 ': ' value1  '}, 'test_file.txt')
        self.assertEqual(result['key1'], 'value1')
        self.assertEqual(result['type'], 'case')
        self.assertEqual(result['case_id'], '123')
        self.assertEqual(len(result['uuid']), 36)  # Check UUID format, not exact value
        
        result = self.loader.prepare_node({'type': 'file', 'uuid': '123', ' key1 ': ' value1  '}, 'test_file.txt')
        self.assertDictEqual(result, {'key1': 'value1', 'type': 'file', 'uuid': '123'})

        # Test parent ids
        obj = self.loader.prepare_node({'type': 'case', 'cohort.cohort_id': 'abc132'}, 'test_file.txt')
        self.assertEqual(obj['cohort_id'], 'abc132')
        obj = self.loader.prepare_node({'type': 'case', 'cohort.cohort_id': 'abc132', 'cohort_id': 'def333'}, 'test_file.txt')
        self.assertEqual(obj['cohort_id'], 'def333')
        self.assertEqual(obj['cohort_cohort_id'], 'abc132')
        self.assertEqual(len(obj[UUID]), 36)

        # Test Boolean values
        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'abc132'}, 'test_file.txt')
        self.assertIsNone(obj['ecg'])
        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'yes'}, 'test_file.txt')
        self.assertEqual(obj['ecg'], True)
        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'YeS'}, 'test_file.txt')
        self.assertEqual(obj['ecg'], True)
        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'YeS13'}, 'test_file.txt')
        self.assertEqual(obj['ecg'], True)

        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'no'}, 'test_file.txt')
        self.assertEqual(obj['ecg'], False)
        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'No'}, 'test_file.txt')
        self.assertEqual(obj['ecg'], False)
        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': ' No33 '}, 'test_file.txt')
        self.assertEqual(obj['ecg'], False)
        obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': ' Normal '}, 'test_file.txt')
        self.assertEqual(obj['ecg'], False)

        # Test Int values
        obj = self.loader.prepare_node({'type': 'physical_exam', 'day_in_cycle': ' Normal '}, 'test_file.txt')
        self.assertEqual(obj['day_in_cycle'], None)
        obj = self.loader.prepare_node({'type': 'physical_exam', 'day_in_cycle': ' 13 '}, 'test_file.txt')
        self.assertEqual(obj['day_in_cycle'], 13)
        self.assertNotEqual(obj['day_in_cycle'], '13')
        obj = self.loader.prepare_node({'type': 'physical_exam', 'day_in_cycle': ' 12 Normal '}, 'test_file.txt')
        self.assertEqual(obj['day_in_cycle'], None)

        #Test Float values
        obj = self.loader.prepare_node({'type': 'file', 'file_size': ' Normal '}, 'test_file.txt')
        self.assertEqual(obj['file_size'], None)
        obj = self.loader.prepare_node({'type': 'file', 'file_size': ' 1.5 Normal '}, 'test_file.txt')
        self.assertEqual(obj['file_size'], None)
        obj = self.loader.prepare_node({'type': 'file', 'file_size': ' 1.5 '}, 'test_file.txt')
        self.assertEqual(obj['file_size'], 1.5)
        obj = self.loader.prepare_node({'type': 'file', 'file_size': ' 15 '}, 'test_file.txt')
        self.assertEqual(obj['file_size'], 15)


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

1	import unittest	2✔
2	import os	2✔
3	from unittest.mock import Mock, MagicMock, patch	2✔
4	from bento.common.utils import get_logger, removeTrailingSlash, UUID	2✔
5	from data_loader import DataLoader	2✔
6	from icdc_schema import ICDC_Schema	2✔
7	from props import Props	2✔
8	from neo4j import Driver	2✔
9
10
11	class MockNeo4jDriver(Driver):	2✔
12	"""Mock Neo4j driver for testing without a database connection"""
13	def __init__(self):	2✔
14	# Don't call super().__init__ to avoid actual connection
15	pass	2✔
16
17	def session(self, **kwargs):	2✔
NEW 18	session = Mock()	×
NEW 19	session.__enter__ = Mock(return_value=session)	×
NEW 20	session.__exit__ = Mock(return_value=False)	×
NEW 21	return session	×
22
23	def close(self):	2✔
24	pass	2✔
25
26	def verify_connectivity(self):	2✔
NEW 27	return True	×
28
29
30	class TestLoader(unittest.TestCase):	2✔
31	def setUp(self):	2✔
32	# Use mock Neo4j driver instead of requiring environment variable
33	self.driver = MockNeo4jDriver()	2✔
34
35	# Use paths relative to the tests directory
36	test_dir = os.path.dirname(os.path.abspath(__file__))	2✔
37	self.data_folder = os.path.join(test_dir, 'data', 'COTC007B')	2✔
38	props_path = os.path.join(test_dir, 'data', 'props-icdc.yml')	2✔
39	model_path = os.path.join(test_dir, 'data', 'icdc-model.yml')	2✔
40	model_props_path = os.path.join(test_dir, 'data', 'icdc-model-props.yml')	2✔
41
42	props = Props(props_path)	2✔
43	self.schema = ICDC_Schema([model_path, model_props_path], props)	2✔
44	self.log = get_logger('Test Loader')	2✔
45	self.loader = DataLoader(self.driver, self.schema)	2✔
46	self.file_list = [	2✔
47	os.path.join(test_dir, "data/Dataset/COP-program.txt"),
48	os.path.join(test_dir, "data/Dataset/NCATS-COP01-case.txt"),
49	os.path.join(test_dir, "data/Dataset/NCATS-COP01-diagnosis.txt"),
50	os.path.join(test_dir, "data/Dataset/NCATS-COP01_cohort_file.txt"),
51	os.path.join(test_dir, "data/Dataset/NCATS-COP01_study_file.txt")
52	]
53
54	def test_remove_traling_slash(self):	2✔
55	self.assertEqual('abc', removeTrailingSlash('abc/'))	2✔
56	self.assertEqual('abc', removeTrailingSlash('abc'))	2✔
57	self.assertEqual('abc', removeTrailingSlash('abc//'))	2✔
58	self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78'))	2✔
59	self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78/'))	2✔
60	self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78//'))	2✔
61	self.assertEqual('bolt://12.34.56.78', removeTrailingSlash('bolt://12.34.56.78////'))	2✔
62
63	def test_loader_construction(self):	2✔
64	self.assertRaises(Exception, DataLoader, None, None, None)	2✔
65	self.assertRaises(Exception, DataLoader, self.driver, None, None)	2✔
66	self.assertIsInstance(self.loader, DataLoader)	2✔
67
68	@unittest.skip("Skipping test that requires actual Neo4j database interaction")	2✔
69	def test_validate_parents_exist_in_file(self):	2✔
NEW 70	test_dir = os.path.dirname(os.path.abspath(__file__))	×
NEW 71	load_result = self.loader.load(self.file_list, True, False, 'upsert', False, 1, '/tmp', False)	×
72	self.assertIsInstance(load_result, dict, msg='Load data failed!')	×
NEW 73	result = self.loader.validate_parents_exist_in_file(os.path.join(test_dir, 'data/pathology-reports-failure.txt'), 100)	×
74	self.assertFalse(result)	×
NEW 75	result = self.loader.validate_parents_exist_in_file(os.path.join(test_dir, 'data/pathology-reports-success.txt'), 100)	×
76	self.assertTrue(result)	×
77
78	def test_duplicated_ids(self):	2✔
79	test_dir = os.path.dirname(os.path.abspath(__file__))	2✔
80	self.assertTrue(self.loader.validate_file(os.path.join(test_dir, 'data/Dataset/NCATS-COP01-case.txt'), 10, False))	2✔
81	self.assertFalse(self.loader.validate_file(os.path.join(test_dir, 'data/NCATS01-case-dup.txt'), 10, False))	2✔
82
83	def test_get_signature(self):	2✔
84	self.assertEqual(self.loader.get_signature({}), '{ }')	2✔
85	self.assertEqual(self.loader.get_signature({'key1': 'value1'}), '{ key1: value1 }')	2✔
86	self.assertEqual(self.loader.get_signature({'key1': 'value1', 'key2': 'value2'}), '{ key1: value1, key2: value2 }')	2✔
87
88	def test_cleanup_node(self):	2✔
89	#Test UUIDs
90	self.assertRaises(SystemExit, self.loader.prepare_node, {}, 'test_file.txt')	2✔
91	result = self.loader.prepare_node({'type': 'case', 'case_id': '123', ' key1 ': ' value1 '}, 'test_file.txt')	2✔
92	self.assertEqual(result['key1'], 'value1')	2✔
93	self.assertEqual(result['type'], 'case')	2✔
94	self.assertEqual(result['case_id'], '123')	2✔
95	self.assertEqual(len(result['uuid']), 36) # Check UUID format, not exact value	2✔
96
97	result = self.loader.prepare_node({'type': 'file', 'uuid': '123', ' key1 ': ' value1 '}, 'test_file.txt')	2✔
98	self.assertDictEqual(result, {'key1': 'value1', 'type': 'file', 'uuid': '123'})	2✔
99
100	# Test parent ids
101	obj = self.loader.prepare_node({'type': 'case', 'cohort.cohort_id': 'abc132'}, 'test_file.txt')	2✔
102	self.assertEqual(obj['cohort_id'], 'abc132')	2✔
103	obj = self.loader.prepare_node({'type': 'case', 'cohort.cohort_id': 'abc132', 'cohort_id': 'def333'}, 'test_file.txt')	2✔
104	self.assertEqual(obj['cohort_id'], 'def333')	2✔
105	self.assertEqual(obj['cohort_cohort_id'], 'abc132')	2✔
106	self.assertEqual(len(obj[UUID]), 36)	2✔
107
108	# Test Boolean values
109	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'abc132'}, 'test_file.txt')	2✔
110	self.assertIsNone(obj['ecg'])	2✔
111	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'yes'}, 'test_file.txt')	2✔
112	self.assertEqual(obj['ecg'], True)	2✔
113	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'YeS'}, 'test_file.txt')	2✔
114	self.assertEqual(obj['ecg'], True)	2✔
115	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'YeS13'}, 'test_file.txt')	2✔
116	self.assertEqual(obj['ecg'], True)	2✔
117
118	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'no'}, 'test_file.txt')	2✔
119	self.assertEqual(obj['ecg'], False)	2✔
120	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': 'No'}, 'test_file.txt')	2✔
121	self.assertEqual(obj['ecg'], False)	2✔
122	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': ' No33 '}, 'test_file.txt')	2✔
123	self.assertEqual(obj['ecg'], False)	2✔
124	obj = self.loader.prepare_node({'type': 'vital_signs', 'ecg': ' Normal '}, 'test_file.txt')	2✔
125	self.assertEqual(obj['ecg'], False)	2✔
126
127	# Test Int values
128	obj = self.loader.prepare_node({'type': 'physical_exam', 'day_in_cycle': ' Normal '}, 'test_file.txt')	2✔
129	self.assertEqual(obj['day_in_cycle'], None)	2✔
130	obj = self.loader.prepare_node({'type': 'physical_exam', 'day_in_cycle': ' 13 '}, 'test_file.txt')	2✔
131	self.assertEqual(obj['day_in_cycle'], 13)	2✔
132	self.assertNotEqual(obj['day_in_cycle'], '13')	2✔
133	obj = self.loader.prepare_node({'type': 'physical_exam', 'day_in_cycle': ' 12 Normal '}, 'test_file.txt')	2✔
134	self.assertEqual(obj['day_in_cycle'], None)	2✔
135
136	#Test Float values
137	obj = self.loader.prepare_node({'type': 'file', 'file_size': ' Normal '}, 'test_file.txt')	2✔
138	self.assertEqual(obj['file_size'], None)	2✔
139	obj = self.loader.prepare_node({'type': 'file', 'file_size': ' 1.5 Normal '}, 'test_file.txt')	2✔
140	self.assertEqual(obj['file_size'], None)	2✔
141	obj = self.loader.prepare_node({'type': 'file', 'file_size': ' 1.5 '}, 'test_file.txt')	2✔
142	self.assertEqual(obj['file_size'], 1.5)	2✔
143	obj = self.loader.prepare_node({'type': 'file', 'file_size': ' 15 '}, 'test_file.txt')	2✔
144	self.assertEqual(obj['file_size'], 15)	2✔
145
146
147	if __name__ == '__main__':	2✔
148	unittest.main()	×

CBIIT / CCDI-C3DC-Dataloader / 21403831778

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