# -*- coding: utf8 -*- #!/usr/bin/python import inspect import os import sys # prefer local copy to the one which is installed # hack from http://stackoverflow.com/a/6098238/280539 _top_level_path = os.path.realpath(os.path.abspath(os.path.join( os.path.split(inspect.getfile(inspect.currentframe()))[0], ".." ))) if _top_level_path not in sys.path: sys.path.insert(0, _top_level_path) # end of hack import unittest try: from rdflib.graph import ConjunctiveGraph except ImportError: from rdflib import ConjunctiveGraph from SPARQLWrapper import SPARQLWrapper, XML, N3, JSONLD, JSON, POST, GET, SELECT, CONSTRUCT, ASK, DESCRIBE from SPARQLWrapper.Wrapper import _SPARQL_DEFAULT, _SPARQL_XML, _SPARQL_JSON, _SPARQL_POSSIBLE, _RDF_XML, _RDF_N3, _RDF_JSONLD, _RDF_POSSIBLE from SPARQLWrapper.SPARQLExceptions import QueryBadFormed try: from urllib.error import HTTPError # Python 3 except ImportError: from urllib2 import HTTPError # Python 2 try: bytes # Python 2.6 and above except NameError: bytes = str import logging logging.basicConfig() endpoint = "http://dbpedia.org/sparql" prefixes = """ PREFIX rdf: PREFIX rdfs: """ selectQuery = """ SELECT ?label WHERE { rdfs:label ?label . } """ constructQuery = """ CONSTRUCT { _:v rdfs:label ?label . _:v rdfs:comment "this is only a mock node to test library" } WHERE { rdfs:label ?label . } """ queryBadFormed = """ PREFIX prop: PREFIX res: FROM SELECT ?lat ?long WHERE { res:Budapest prop:latitude ?lat; prop:longitude ?long. } """ queryManyPrefixes = """ PREFIX conf: PREFIX meta: PREFIX rdf: PREFIX rdfs: PREFIX xsd: PREFIX owl: PREFIX dc: PREFIX dcterms: PREFIX foaf: PREFIX skos: PREFIX geo: PREFIX dbpedia: PREFIX o: PREFIX p: PREFIX yago: PREFIX units: PREFIX geonames: PREFIX prv: PREFIX prvTypes: PREFIX foo: SELECT ?label WHERE { rdfs:label ?label . } """ class SPARQLWrapperTests(unittest.TestCase): def __generic(self, query, returnFormat, method): sparql = SPARQLWrapper(endpoint) sparql.setQuery(prefixes + query) sparql.setReturnFormat(returnFormat) sparql.setMethod(method) try: result = sparql.query() except HTTPError: # An ugly way to get the exception, but the only one that works # both on Python 2.5 and Python 3. e = sys.exc_info()[1] if e.code == 400: sys.stdout.write("400 Bad Request, probably query is not well formed") elif e.code == 406: sys.stdout.write("406 Not Acceptable, maybe query is not well formed") else: sys.stdout.write(str(e)) sys.stdout.write("\n") return False else: return result def testSelectByGETinXML(self): result = self.__generic(selectQuery, XML, GET) ct = result.info()["content-type"] assert True in [one in ct for one in _SPARQL_XML], ct results = result.convert() results.toxml() def testSelectByPOSTinXML(self): result = self.__generic(selectQuery, XML, POST) ct = result.info()["content-type"] assert True in [one in ct for one in _SPARQL_XML], ct results = result.convert() results.toxml() # Virtuoso returns text/rdf+n3. It MUST return SPARQL Results Document in XML (sparql-results+xml), JSON (sparql-results+json), or CSV/TSV (text/csv or text/tab-separated-values) see http://www.w3.org/TR/sparql11-protocol/#query-success # URI generated http://dbpedia.org/sparql?query=%0A++++PREFIX+rdf%3A+%3Chttp%3A%2F%2Fwww.w3.org%2F1999%2F02%2F22-rdf-syntax-ns%23%3E%0A++++PREFIX+rdfs%3A+%3Chttp%3A%2F%2Fwww.w3.org%2F2000%2F01%2Frdf-schema%23%3E%0A%0A++++SELECT+%3Flabel%0A++++WHERE+%7B%0A++++%3Chttp%3A%2F%2Fdbpedia.org%2Fresource%2FAsturias%3E+rdfs%3Alabel+%3Flabel+.%0A++++%7D%0A def _testSelectByGETinN3(self): result = self.__generic(selectQuery, N3, GET) ct = result.info()["content-type"] assert True in [one in ct for one in _SPARQL_XML], ct assert True in [one in ct for one in _SPARQL_POSSIBLE], ct results = result.convert() self.assertEqual(type(results), bytes) def testSelectByGETinJSON(self): result = self.__generic(selectQuery, JSON, GET) ct = result.info()["content-type"] assert True in [one in ct for one in _SPARQL_JSON], ct results = result.convert() self.assertEqual(type(results), dict) def testSelectByPOSTinJSON(self): result = self.__generic(selectQuery, JSON, POST) ct = result.info()["content-type"] assert True in [one in ct for one in _SPARQL_JSON], ct results = result.convert() self.assertEqual(type(results), dict) def testSelectByGETinUnknow(self): result = self.__generic(selectQuery, "foo", GET) ct = result.info()["content-type"] assert True in [one in ct for one in _SPARQL_POSSIBLE], ct results = result.convert() def testSelectByPOSTinUnknow(self): result = self.__generic(selectQuery, "bar", POST) ct = result.info()["content-type"] assert True in [one in ct for one in _SPARQL_POSSIBLE], ct results = result.convert() def testConstructByGETinXML(self): result = self.__generic(constructQuery, XML, GET) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_XML], ct results = result.convert() self.assertEqual(type(results), ConjunctiveGraph) def testConstructByPOSTinXML(self): result = self.__generic(constructQuery, XML, POST) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_XML], ct results = result.convert() self.assertEqual(type(results), ConjunctiveGraph) def testConstructByGETinN3(self): result = self.__generic(constructQuery, N3, GET) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_N3], ct results = result.convert() self.assertEqual(type(results), bytes) def testConstructByPOSTinN3(self): result = self.__generic(constructQuery, N3, POST) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_N3], ct results = result.convert() self.assertEqual(type(results), bytes) # Virtuoso returns application/sparql-results+json. It MUST return an RDF graph [RDF-CONCEPTS] serialized, for example, in the RDF/XML syntax [RDF-XML], or an equivalent RDF graph serialization, for SPARQL Query forms DESCRIBE and CONSTRUCT). See http://www.w3.org/TR/sparql11-protocol/#query-success def _testConstructByGETinJSON(self): result = self.__generic(constructQuery, JSON, GET) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_POSSIBLE], ct results = result.convert() self.assertEqual(type(results), ConjunctiveGraph) def _testConstructByPOSTinJSON(self): result = self.__generic(constructQuery, _RDF_JSONLD, POST) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_POSSIBLE], ct results = result.convert() self.assertEqual(type(results), bytes) def testConstructByGETinUnknow(self): result = self.__generic(constructQuery, "foo", GET) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_POSSIBLE], ct results = result.convert() self.assertEqual(type(results), ConjunctiveGraph) def testConstructByPOSTinUnknow(self): result = self.__generic(constructQuery, "bar", POST) ct = result.info()["content-type"] assert True in [one in ct for one in _RDF_POSSIBLE], ct results = result.convert() self.assertEqual(type(results), ConjunctiveGraph) def testQueryBadFormed(self): self.assertRaises(QueryBadFormed, self.__generic, queryBadFormed, XML, GET) # def testQueryManyPrefixes(self): # result = self.__generic(queryManyPrefixes, XML, GET) def testKeepAlive(self): sparql = SPARQLWrapper(endpoint) sparql.setQuery('SELECT * WHERE {?s ?p ?o} LIMIT 10') sparql.setReturnFormat(JSON) sparql.setMethod(GET) sparql.setUseKeepAlive() sparql.query() sparql.query() if __name__ == "__main__": unittest.main()