Marc-Antoine Parent committed 144feaa

tighten the regular expressions. Remove dead debugging code. tentatively set formula_aware on the store

Comments (0)

Files changed (2)


 import os
 import unittest
-from nose.plugins.skip import SkipTest
+#from nose.plugins.skip import SkipTest
 class Test00Plugin(unittest.TestCase):
     def test_get_plugin(self):
              "We keep talking and talking, but what are we saying? Precisely nothing the "
              "whole idea is to have a bunch of characters here. Blah blah, yadda yadda, "
              "etc. This is probably enough. Hopefully. One more sentence to make certain.")),
-    (URIRef(""), RDFS["label"], Literal(3)), # Fails because comes back untyped
+    (URIRef(""), RDFS["label"], Literal(3)),
     (URIRef(""), RDFS["comment"], Literal(,
     (URIRef(""), RDFS["comment"], Literal(,
     (URIRef(""), RDFS["comment"], Literal(,
     (URIRef(""), RDFS["comment"], Literal("1970", datatype=XSD["gYear"])),
-    (URIRef(""), RDFS["label"], Literal("hello world", lang="en")), # Fails because comes back w/o language
+    (URIRef(""), RDFS["label"], Literal("hello world", lang="en")),
 ## special test that will induce a namespace creation for testing of serialisation
         q = "CONSTRUCT { ?s ?p ?o } WHERE { GRAPH %s { ?s ?p ?o } }" % (self.graph.identifier.n3(),)
         result =, q)
         assert isinstance(result, Graph) or isinstance(result, Result)
-        result = list(result)
-        print result
         assert test_statements[0] in result


 ## end hack
 import re
-_ask_re = re.compile(u'^SPARQL ([ \t\r\n]*DEFINE[ \t]+.*)*([ \t\r\n]*PREFIX[ \t]+[^ \t]*: <[^>]*>)*[ \t\r\n]*(ASK)[ \t\r\n]+(FROM|WHERE)', re.IGNORECASE)
-_construct_re = re.compile(u'^SPARQL ([ \t\r\n]*DEFINE[ \t]+.*)*([ \t\r\n]*PREFIX[ \t]+[^ \t]*: <[^>]*>)*[ \t\r\n]*(CONSTRUCT|DESCRIBE)', re.IGNORECASE)
-_select_re = re.compile(u'^SPARQL ([ \t\r\n]*DEFINE[ \t]+.*)*([ \t\r\n]*PREFIX[ \t]+[^ \t]*: <[^>]*>)*[ \t\r\n]*SELECT', re.IGNORECASE)
+_start_re = r'^SPARQL\s+' \
+            r'(DEFINE[ \t]+\S+[ \t]+("[^"]*"|<[^>]*>|[0-9]+)\s+)*' \
+            r'(PREFIX[ \t]+\w+:\s+<[^>]*>\s+)*'
+_ask_re = re.compile(_start_re + r'(ASK)\s+(FROM|WHERE)\b', re.IGNORECASE + re.MULTILINE)
+_construct_re = re.compile(_start_re + r'(CONSTRUCT|DESCRIBE)\b', re.IGNORECASE + re.MULTILINE)
+_select_re = re.compile(_start_re + r'SELECT\b', re.IGNORECASE + re.MULTILINE)
 class OperationalError(Exception):
     context_aware = True
     transaction_aware = True
+    formula_aware = True   # Not sure whether this is true; needed to read N3.
     def __init__(self, *av, **kw):
         super(Virtuoso, self).__init__(*av, **kw)
                 cursor = self.cursor()
             if _construct_re.match(q):
-                print "co"
                 return self._sparql_construct(q, cursor)
             elif _ask_re.match(q):
-                print "ask"
                 return self._sparql_ask(q, cursor)
             elif _select_re.match(q):
-                print "sel"
                 return self._sparql_select(q, cursor)
-                print "els", q
                 return self._sparql_ul(q, cursor, commit=commit)
             log.error(u"Exception running: %s" % q.decode("utf-8"))
             results = cursor.execute(q.encode("utf-8"))
             with self.cursor() as resolver:
                 for result in results:
-                    print result
                     g.add(resolve(resolver, x) for x in result)
         return vsparql.Result(g)
             # and ask -> true returns an single row
             results = cursor.execute(q.encode("utf-8"))
             result =
-            print result
             result = resolve(None, result[0])
             return result != 0
             results = cursor.execute(q.encode("utf-8"))
             with self.cursor() as resolver:
                 for result in results:
-                    print result
                     yield [resolve(resolver, x) for x in result]
     def _sparql_ul(self, q, cursor, commit):