|
| 1 | +import logging |
| 2 | +import os |
| 3 | +import shutil |
| 4 | +import tempfile |
| 5 | +import unittest |
| 6 | + |
| 7 | +import rdflib |
| 8 | +import rdflib.compare |
| 9 | + |
| 10 | +import sbol3 |
| 11 | + |
| 12 | +MODULE_LOCATION = os.path.dirname(os.path.abspath(__file__)) |
| 13 | +# The SBOLTestSuite root is 3 directories up from this file |
| 14 | +SBOL_TEST_SUITE = os.path.join(MODULE_LOCATION, '..', '..', '..') |
| 15 | +SBOL3_LOCATION = os.path.join(SBOL_TEST_SUITE, 'SBOL3') |
| 16 | +TEST_RESOURCE_DIR = os.path.join(MODULE_LOCATION, 'resources') |
| 17 | + |
| 18 | +DEBUG_ENV_VAR = 'SBOL_TEST_DEBUG' |
| 19 | + |
| 20 | + |
| 21 | +class TestRoundTrip(unittest.TestCase): |
| 22 | + |
| 23 | + @classmethod |
| 24 | + def setUpClass(cls): |
| 25 | + # Skip reading these files |
| 26 | + # No files are skipped at this time. All SBOLTestSuite files can |
| 27 | + # be read. |
| 28 | + cls.skip_reading = [ |
| 29 | + ] |
| 30 | + # Skip round tripping these files |
| 31 | + cls.skip_round_trip = [ |
| 32 | + ] |
| 33 | + # Skip validating these files |
| 34 | + cls.skip_validation = [ |
| 35 | + ] |
| 36 | + |
| 37 | + def setUp(self): |
| 38 | + sbol3.set_defaults() |
| 39 | + # Create temp directory |
| 40 | + self.temp_out_dir = tempfile.mkdtemp() |
| 41 | + self.logger = logging.getLogger('sbol3.test') |
| 42 | + if not self.logger.hasHandlers(): |
| 43 | + logging.basicConfig() |
| 44 | + if DEBUG_ENV_VAR in os.environ: |
| 45 | + self.logger.setLevel(logging.DEBUG) |
| 46 | + |
| 47 | + def tearDown(self): |
| 48 | + # Remove directory after the test |
| 49 | + if self.temp_out_dir: |
| 50 | + shutil.rmtree(self.temp_out_dir) |
| 51 | + self.temp_out_dir = None |
| 52 | + sbol3.set_defaults() |
| 53 | + |
| 54 | + def test_BBa_F2620_PoPSReceiver(self): |
| 55 | + sbol_path = os.path.join(SBOL3_LOCATION, 'BBa_F2620_PoPSReceiver', |
| 56 | + 'BBa_F2620_PoPSReceiver.ttl') |
| 57 | + doc = sbol3.Document() |
| 58 | + doc.read(sbol_path, sbol3.TURTLE) |
| 59 | + uri = 'https://synbiohub.org/public/igem/BBa_F2620/SubComponent3/Range1' |
| 60 | + range1 = doc.find(uri) |
| 61 | + self.assertIsNotNone(range1) |
| 62 | + self.assertEqual(1, range1.start) |
| 63 | + self.assertEqual(53, range1.end) |
| 64 | + |
| 65 | + def find_all_files(self, dirname: str): |
| 66 | + for item in os.listdir(dirname): |
| 67 | + item_path = os.path.join(dirname, item) |
| 68 | + if os.path.isdir(item_path): |
| 69 | + for f in self.find_all_files(item_path): |
| 70 | + yield f |
| 71 | + elif os.path.isfile(item_path): |
| 72 | + yield item_path |
| 73 | + else: |
| 74 | + print(f'{item} is neither file nor directory') |
| 75 | + |
| 76 | + @staticmethod |
| 77 | + def rdf_type(filename: str): |
| 78 | + filename = os.path.basename(filename) |
| 79 | + ext = os.path.splitext(filename)[1] |
| 80 | + # drop the leading dot |
| 81 | + ext = ext[1:] |
| 82 | + ext_map = { |
| 83 | + 'nt': sbol3.NTRIPLES, |
| 84 | + 'ttl': sbol3.TURTLE, |
| 85 | + 'rdf': sbol3.RDF_XML, |
| 86 | + 'jsonld': sbol3.JSONLD, |
| 87 | + 'jsonld_expanded': sbol3.JSONLD, |
| 88 | + } |
| 89 | + if ext in ext_map: |
| 90 | + return ext_map[ext] |
| 91 | + else: |
| 92 | + return None |
| 93 | + |
| 94 | + def test_read_all(self): |
| 95 | + # In lieu of round tripping the files, just make sure we can |
| 96 | + # read them all. |
| 97 | + # |
| 98 | + # This was an early test, before the library was complete and |
| 99 | + # the files could be round tripped. |
| 100 | + skip_list = self.skip_reading |
| 101 | + for f in self.find_all_files(SBOL3_LOCATION): |
| 102 | + basename = os.path.basename(f) |
| 103 | + if os.path.splitext(basename)[0] in skip_list: |
| 104 | + # print(f'Skipping {f}') |
| 105 | + continue |
| 106 | + rdf_type = self.rdf_type(f) |
| 107 | + if rdf_type is None: |
| 108 | + # Skip file types we don't know |
| 109 | + # print(f'Skipping {f} of type {rdf_type}') |
| 110 | + continue |
| 111 | + # print(f'Reading {f}') |
| 112 | + doc = sbol3.Document() |
| 113 | + doc.read(f, rdf_type) |
| 114 | + |
| 115 | + def run_round_trip_file(self, test_path, file_format): |
| 116 | + """Runs a round trip test on the file at the given path. |
| 117 | + Path can be relative or absolute. |
| 118 | + """ |
| 119 | + filename = os.path.basename(test_path) |
| 120 | + test2_path = os.path.join(self.temp_out_dir, filename) |
| 121 | + # Read the document, then write it back to disk |
| 122 | + doc = sbol3.Document() |
| 123 | + doc.read(test_path, file_format) |
| 124 | + skip_list = self.skip_validation |
| 125 | + basename = os.path.basename(test_path) |
| 126 | + if os.path.splitext(basename)[0] not in skip_list: |
| 127 | + # Validate files that are not in the skip list |
| 128 | + report = doc.validate() |
| 129 | + if report: |
| 130 | + for item in report: |
| 131 | + print(item) |
| 132 | + self.fail(f'got {len(report)} validation errors') |
| 133 | + |
| 134 | + doc.write(test2_path, file_format) |
| 135 | + |
| 136 | + # Read the newly written document and compare results |
| 137 | + doc2 = sbol3.Document() |
| 138 | + doc2.read(test2_path, file_format) |
| 139 | + # TODO: what about Document.compare()? |
| 140 | + # self.assertTrue(doc.compare(doc2)) |
| 141 | + |
| 142 | + # Now compare the graphs in RDF |
| 143 | + g1 = rdflib.Graph() |
| 144 | + g1.parse(test_path, format=file_format) |
| 145 | + iso1 = rdflib.compare.to_isomorphic(g1) |
| 146 | + g2 = rdflib.Graph() |
| 147 | + g2.parse(test2_path, format=file_format) |
| 148 | + iso2 = rdflib.compare.to_isomorphic(g2) |
| 149 | + rdf_diff = rdflib.compare.graph_diff(iso1, iso2) |
| 150 | + if rdf_diff[1] or rdf_diff[2]: |
| 151 | + self.logger.warning('Detected %d different RDF triples in %s' % |
| 152 | + (len(rdf_diff[1]) + len(rdf_diff[2]), test_path)) |
| 153 | + if not self.logger.isEnabledFor(logging.DEBUG): |
| 154 | + self.logger.warning('Set environment variable %s to see details', |
| 155 | + DEBUG_ENV_VAR) |
| 156 | + for stmt in rdf_diff[1]: |
| 157 | + self.logger.debug('Only in original: %r', stmt) |
| 158 | + for stmt in rdf_diff[2]: |
| 159 | + self.logger.debug('Only in loaded: %r', stmt) |
| 160 | + self.fail('Differences in RDF detected') |
| 161 | + |
| 162 | + def test_sbol3_files(self): |
| 163 | + test_dir = SBOL3_LOCATION |
| 164 | + skip_list = self.skip_round_trip |
| 165 | + for test_file in self.find_all_files(test_dir): |
| 166 | + basename = os.path.basename(test_file) |
| 167 | + if os.path.splitext(basename)[0] in skip_list: |
| 168 | + self.logger.debug(f'Skipping {test_file}') |
| 169 | + continue |
| 170 | + file_format = self.rdf_type(test_file) |
| 171 | + if not file_format: |
| 172 | + continue |
| 173 | + with self.subTest(filename=test_file): |
| 174 | + self.setUp() |
| 175 | + self.run_round_trip_file(test_file, file_format) |
| 176 | + self.tearDown() |
| 177 | + |
| 178 | + |
| 179 | +if __name__ == '__main__': |
| 180 | + unittest.main() |
0 commit comments