Merge pull request #375 from fschrader1992/rdf

Terminology Cache

Author: mpsonntag

odml/terminology.py

@@ -24,12 +24,13 @@
 CACHE_AGE = datetime.timedelta(days=1)
 
 
-def cache_load(url):
+def cache_load(url, replace_file=False):
     """
     Loads the url and store it in a temporary cache directory
     subsequent requests for this url will use the cached version.
 
     :param url: URL from where to load an odML terminology file from.
+    :param replace_file: True, if file should be reloaded
     """
     filename = '.'.join([md5(url.encode()).hexdigest(), os.path.basename(url)])
     cache_dir = os.path.join(tempfile.gettempdir(), "odml.cache")
@@ -41,8 +42,9 @@ def cache_load(url):
                 raise
     cache_file = os.path.join(cache_dir, filename)
     if not os.path.exists(cache_file) \
-       or datetime.datetime.fromtimestamp(os.path.getmtime(cache_file)) < \
-       datetime.datetime.now() - CACHE_AGE:
+            or replace_file \
+            or datetime.datetime.fromtimestamp(os.path.getmtime(cache_file)) < \
+            datetime.datetime.now() - CACHE_AGE:
         try:
             data = urllib2.urlopen(url).read()
             if sys.version_info.major > 2:
@@ -64,6 +66,7 @@ class Terminologies(dict):
     browsing and importing of full or partial odML terminologies.
     """
     loading = {}
+    reload_cache = False
 
     def load(self, url):
         """
@@ -92,7 +95,7 @@ def _load(self, url):
                  It will silently return None, if any exceptions
                  occur to enable loading of nested odML files.
         """
-        file_obj = cache_load(url)
+        file_obj = cache_load(url, self.reload_cache)
         if file_obj is None:
             print("did not successfully load '%s'" % url)
             return
@@ -117,10 +120,23 @@ def deferred_load(self, url):
         self.loading[url] = threading.Thread(target=self._load, args=(url,))
         self.loading[url].start()
 
+    def refresh(self, url):
+        """
+        Deletes and reloads all cached odML XML files given in the
+        terminology file from a URL.
+
+        :param url: location of an odML XML file.
+        """
+        self.reload_cache = True
+        self.clear()
+        self.load(url)
+        self.reload_cache = False
+
 
 terminologies = Terminologies()
 load = terminologies.load
 deferred_load = terminologies.deferred_load
+refresh = terminologies.refresh
 
 
 if __name__ == "__main__":

test/test_doc.py

@@ -1,6 +1,9 @@
 import datetime
 import os
 import unittest
+import tempfile
+import odml.terminology
+from hashlib import md5
 try:
     from urllib.request import pathname2url
 except ImportError:
@@ -109,6 +112,32 @@ def test_get_terminology_equivalent(self):
         doc.repository = None
         self.assertIsNone(doc.get_terminology_equivalent())
 
+    def test_terminology_refresh(self):
+
+        cache_dir = os.path.join(tempfile.gettempdir(), "odml.cache")
+        url = "https://terminologies.g-node.org/v1.1/terminologies.xml"
+        term_doc = odml.terminology.load(url)
+
+        terms = []
+        for sec in term_doc.sections:
+            terms += ['.'.join([md5(sec.include.encode()).hexdigest(), os.path.basename(sec.include)])]
+
+        before = datetime.datetime.now()
+
+        for loaded_file in os.listdir(cache_dir):
+            for term in terms:
+                if term in loaded_file:
+                    assert datetime.datetime.fromtimestamp(
+                        os.path.getmtime(os.path.join(cache_dir, loaded_file))) < before
+
+        odml.terminology.refresh(url)
+
+        for replaced_file in os.listdir(cache_dir):
+            for term in terms:
+                if term in replaced_file:
+                    assert datetime.datetime.fromtimestamp(
+                        os.path.getmtime(os.path.join(cache_dir, replaced_file))) > before
+
     def test_append(self):
         doc = Document()
         self.assertListEqual(doc.sections, [])

test/test_parser_odml.py

@@ -8,6 +8,7 @@
 import tempfile
 import unittest
 
+from odml import Document, Section, Property
 from odml.tools import odmlparser
 
 
@@ -82,6 +83,10 @@ def test_json_file(self):
         self.assertEqual(json_doc, self.odml_doc)
 
     def test_rdf_file(self):
+        """
+        Test comparison of document before and after rdf-conversion
+        """
+
         self.rdf_writer.write_file(self.odml_doc, self.rdf_file)
         rdf_doc = self.rdf_reader.from_file(self.rdf_file, "xml")
 
@@ -101,6 +106,45 @@ def test_rdf_file(self):
         with self.assertRaises(ValueError):
             self.rdf_reader.from_file(self.rdf_file)
 
+        doc = Document()
+        sec1 = Section(name="sec1", type="int", parent=doc)
+        Property(name="prop11", values=[1, 2, 3], dtype="int", parent=sec1)
+        Property(name="prop12", values=[1.1, 2.2, 3.3], dtype="float", parent=sec1)
+        Property(name="prop13", values=["a", "b", "c"], dtype="string", parent=sec1)
+        sec2 = Section(name="sec2", type="int", parent=doc)
+        Property(name="prop21", values=[1, 2, 3], dtype="int", parent=sec2)
+        Property(name="prop22", values=[1.1, 2.2, 3.3], dtype="float", parent=sec2)
+        Property(name="prop23", values=["a", "b", "c"], dtype="string", parent=sec2)
+        sec3 = Section(name="sec3", type="int", parent=doc)
+        Property(name="prop31", values=[1, 2, 3], dtype="int", parent=sec3)
+        Property(name="prop32", values=[1.1, 2.2, 3.3], dtype="float", parent=sec3)
+        Property(name="prop33", values=["a", "b", "c"], dtype="string", parent=sec3)
+        self.rdf_writer.write_file(doc, self.rdf_file)
+        rdf_doc = self.rdf_reader.from_file(self.rdf_file, "xml")
+
+        self.assertEqual(doc, rdf_doc[0])
+        self.assertEqual(len(rdf_doc), 1)
+        self.assertEqual(len(rdf_doc[0].sections), 3)
+
+        self.assertIn(doc.sections[0].name, rdf_doc[0].sections)
+        self.assertIn(doc.sections[1].name, rdf_doc[0].sections)
+        self.assertIn(doc.sections[2].name, rdf_doc[0].sections)
+        rdf_sec1 = rdf_doc[0].sections[doc.sections[0].name]
+        self.assertEqual(len(rdf_sec1.properties), 3)
+        self.assertIn(doc.sections[0].properties[0].name, rdf_sec1.properties)
+        self.assertIn(doc.sections[0].properties[1].name, rdf_sec1.properties)
+        self.assertIn(doc.sections[0].properties[1].name, rdf_sec1.properties)
+        rdf_sec2 = rdf_doc[0].sections[doc.sections[1].name]
+        self.assertEqual(len(rdf_sec2.properties), 3)
+        self.assertIn(doc.sections[1].properties[0].name, rdf_sec2.properties)
+        self.assertIn(doc.sections[1].properties[1].name, rdf_sec2.properties)
+        self.assertIn(doc.sections[1].properties[1].name, rdf_sec2.properties)
+        rdf_sec3 = rdf_doc[0].sections[doc.sections[2].name]
+        self.assertEqual(len(rdf_sec3.properties), 3)
+        self.assertIn(doc.sections[2].properties[0].name, rdf_sec3.properties)
+        self.assertIn(doc.sections[2].properties[1].name, rdf_sec3.properties)
+        self.assertIn(doc.sections[2].properties[1].name, rdf_sec3.properties)
+
     def test_xml_string(self):
         # Read from string
         author = "HPL"