htsworkflow/util/test/test_rdfhelp.py

   1 import os
   2 import unittest
   3 import types
   4
   5
   6 from datetime import datetime
   7
   8 from htsworkflow.util.rdfhelp import \
   9      blankOrUri, \
  10      dump_model, \
  11      fromTypedNode, \
  12      get_model, \
  13      guess_parser, \
  14      guess_parser_by_extension, \
  15      load_string_into_model, \
  16      rdfsNS, \
  17      toTypedNode, \
  18      stripNamespace, \
  19      simplify_uri, \
  20      sanitize_literal, \
  21      xsdNS
  22
  23 try:
  24     import RDF
  25
  26     class TestRDFHelp(unittest.TestCase):
  27         def test_from_none(self):
  28           self.failUnlessEqual(fromTypedNode(None), None)
  29
  30         def test_typed_node_boolean(self):
  31             node = toTypedNode(True)
  32             self.failUnlessEqual(node.literal_value['string'], u'1')
  33             self.failUnlessEqual(str(node.literal_value['datatype']),
  34                                  'http://www.w3.org/2001/XMLSchema#boolean')
  35
  36         def test_bad_boolean(self):
  37             node = RDF.Node(literal='bad', datatype=xsdNS['boolean'].uri)
  38             self.failUnlessRaises(ValueError, fromTypedNode, node)
  39
  40         def test_typed_node_string(self):
  41             node = toTypedNode('hello')
  42             self.failUnlessEqual(node.literal_value['string'], u'hello')
  43             self.failUnless(node.literal_value['datatype'] is None)
  44
  45         def test_typed_real_like(self):
  46             num = 3.14
  47             node = toTypedNode(num)
  48             self.failUnlessEqual(fromTypedNode(node), num)
  49
  50         def test_typed_integer(self):
  51             num = 3
  52             node = toTypedNode(num)
  53             self.failUnlessEqual(fromTypedNode(node), num)
  54             self.failUnlessEqual(type(fromTypedNode(node)), type(num))
  55
  56         def test_typed_node_string(self):
  57             s = "Argh matey"
  58             node = toTypedNode(s)
  59             self.failUnlessEqual(fromTypedNode(node), s)
  60             self.failUnlessEqual(type(fromTypedNode(node)), types.UnicodeType)
  61
  62         def test_blank_or_uri_blank(self):
  63             node = blankOrUri()
  64             self.failUnlessEqual(node.is_blank(), True)
  65
  66         def test_blank_or_uri_url(self):
  67             s = 'http://google.com'
  68             node = blankOrUri(s)
  69             self.failUnlessEqual(node.is_resource(), True)
  70             self.failUnlessEqual(str(node.uri), s)
  71
  72         def test_blank_or_uri_node(self):
  73             s = RDF.Node(RDF.Uri('http://google.com'))
  74             node = blankOrUri(s)
  75             self.failUnlessEqual(node.is_resource(), True)
  76             self.failUnlessEqual(node, s)
  77
  78         def test_unicode_node_roundtrip(self):
  79             literal = u'\u5927'
  80             roundtrip = fromTypedNode(toTypedNode(literal))
  81             self.failUnlessEqual(roundtrip, literal)
  82             self.failUnlessEqual(type(roundtrip), types.UnicodeType)
  83
  84         def test_datetime_no_microsecond(self):
  85             dateTimeType = xsdNS['dateTime'].uri
  86             short_isostamp = '2011-12-20T11:44:25'
  87             short_node = RDF.Node(literal=short_isostamp,
  88                                  datatype=dateTimeType)
  89             short_datetime = datetime(2011,12,20,11,44,25)
  90
  91             self.assertEqual(fromTypedNode(short_node), short_datetime)
  92             self.assertEqual(toTypedNode(short_datetime), short_node)
  93             self.assertEqual(fromTypedNode(toTypedNode(short_datetime)),
  94                              short_datetime)
  95
  96         def test_datetime_with_microsecond(self):
  97             dateTimeType = xsdNS['dateTime'].uri
  98             long_isostamp = '2011-12-20T11:44:25.081776'
  99             long_node = RDF.Node(literal=long_isostamp,
 100                                  datatype=dateTimeType)
 101             long_datetime = datetime(2011,12,20,11,44,25,81776)
 102
 103             self.assertEqual(fromTypedNode(long_node), long_datetime)
 104             self.assertEqual(toTypedNode(long_datetime), long_node)
 105             self.assertEqual(fromTypedNode(toTypedNode(long_datetime)),
 106                              long_datetime)
 107
 108         def test_strip_namespace_uri(self):
 109             nsOrg = RDF.NS('example.org/example#')
 110             nsCom = RDF.NS('example.com/example#')
 111
 112             term = 'foo'
 113             node = nsOrg[term]
 114             self.failUnlessEqual(stripNamespace(nsOrg, node), term)
 115             self.failUnlessEqual(stripNamespace(nsCom, node), None)
 116             self.failUnlessEqual(stripNamespace(nsOrg, node.uri), term)
 117
 118         def test_strip_namespace_exceptions(self):
 119             nsOrg = RDF.NS('example.org/example#')
 120             nsCom = RDF.NS('example.com/example#')
 121
 122             node = toTypedNode('bad')
 123             self.failUnlessRaises(ValueError, stripNamespace, nsOrg, node)
 124             self.failUnlessRaises(ValueError, stripNamespace, nsOrg, nsOrg)
 125
 126         def test_simplify_uri(self):
 127             DATA = [('http://asdf.org/foo/bar', 'bar'),
 128                     ('http://asdf.org/foo/bar#bleem', 'bleem'),
 129                     ('http://asdf.org/foo/bar/', 'bar'),
 130                     ('http://asdf.org/foo/bar?was=foo', 'was=foo')]
 131
 132             for uri, expected in DATA:
 133                 self.assertEqual(simplify_uri(uri), expected)
 134
 135             for uri, expected in DATA:
 136                 n = RDF.Uri(uri)
 137                 self.assertEqual(simplify_uri(n), expected)
 138
 139             for uri, expected in DATA:
 140                 n = RDF.Node(RDF.Uri(uri))
 141                 self.assertEqual(simplify_uri(n), expected)
 142
 143             # decoding literals is questionable
 144             n = toTypedNode('http://foo/bar')
 145             self.assertRaises(ValueError, simplify_uri, n)
 146
 147         def test_owl_import(self):
 148             path, name = os.path.split(__file__)
 149             #loc = 'file://'+os.path.abspath(path)+'/'
 150             loc = os.path.abspath(path)+'/'
 151             model = get_model()
 152             fragment = '''
 153 @prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
 154 @prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
 155 @prefix owl: <http://www.w3.org/2002/07/owl#> .
 156
 157 _:a owl:imports "{loc}extra.turtle" .
 158 '''.format(loc=loc)
 159             load_string_into_model(model, 'turtle', fragment, loc)
 160             tc = RDF.Node(RDF.Uri('http://jumpgate.caltech.edu/wiki/TestCase'))
 161             query = RDF.Statement(tc, rdfsNS['label'], None)
 162             result = list(model.find_statements(query))
 163             self.failUnlessEqual(len(result), 1)
 164             self.failUnlessEqual(str(result[0].object), 'TestCase')
 165
 166         def test_sanitize_literal_text(self):
 167             self.failUnlessRaises(ValueError, sanitize_literal, "hi")
 168             hello_text = "hello"
 169             hello_none = RDF.Node(hello_text)
 170             self.failUnlessEqual(str(sanitize_literal(hello_none)),
 171                                  hello_text)
 172             hello_str = RDF.Node(literal=hello_text,
 173                                  datatype=xsdNS['string'].uri)
 174             hello_clean = sanitize_literal(hello_str)
 175             self.failUnlessEqual(hello_clean.literal_value['string'],
 176                                  hello_text)
 177
 178         def test_sanitize_literal_empty_string(self):
 179             value = ""
 180             value_node = RDF.Node(value)
 181             self.assertEqual(str(sanitize_literal(value_node)), value)
 182
 183         def test_sanitize_literal_html(self):
 184             hello = "hello <a onload='javascript:alert(\"foo\");' href='http://google.com'>google.com</a>, whats up?"
 185             hello_clean = 'hello <a href="http://google.com">google.com</a>, whats up?'
 186             hello_node = RDF.Node(literal=hello,
 187                                   datatype=xsdNS['string'].uri)
 188             hello_sanitized = sanitize_literal(hello_node)
 189             self.failUnlessEqual(hello_sanitized.literal_value['string'],
 190                                  hello_clean)
 191
 192             hostile = "hi <b>there</b><script type='text/javascript>alert('boo');</script><a href='javascript:alert('poke')>evil</a> scammer"
 193             hostile_node = RDF.Node(hostile)
 194             hostile_sanitized = sanitize_literal(hostile_node)
 195             # so it drops the stuff after the javascript link.
 196             # I suppose it could be worse
 197             hostile_result = """hi <b>there</b>"""
 198             self.failUnlessEqual(str(hostile_sanitized), hostile_result)
 199
 200         def test_guess_parser_from_file(self):
 201             DATA = [
 202                 ('/a/b/c.rdf', 'rdfxml'),
 203                 ('/a/b/c.xml', 'rdfxml'),
 204                 ('/a/b/c.html', 'rdfa'),
 205                 ('/a/b/c.turtle', 'turtle')]
 206             for path, parser in DATA:
 207                 self.assertEqual(guess_parser_by_extension(path), parser)
 208                 self.assertEqual(guess_parser(None, path), parser)
 209
 210             DATA = [
 211                 ('application/rdf+xml', 'http://a.org/b/c', 'rdfxml'),
 212                 ('application/x-turtle', 'http://a.org/b/c', 'turtle'),
 213                 ('text/html', 'http://a.org/b/c', 'rdfa')
 214             ]
 215             for contenttype, url, parser in DATA:
 216                 self.assertEqual(guess_parser(contenttype, url), parser)
 217
 218     def suite():
 219         return unittest.makeSuite(TestRDFHelp, 'test')
 220 except ImportError, e:
 221     print "Unable to test rdfhelp"
 222
 223     def suite():
 224         return None
 225
 226 if __name__ == "__main__":
 227     unittest.main(defaultTest='suite')