diff options
author | Natanael Arndt <arndtn@gmail.com> | 2020-05-02 00:02:34 +0200 |
---|---|---|
committer | Natanael Arndt <arndtn@gmail.com> | 2020-05-02 00:02:34 +0200 |
commit | 0986737fc8170d3cd3a7c95d078fa8f689267355 (patch) | |
tree | 021fbd396446ed9a32f903857164d50397474e89 | |
parent | b71a00b239ec4d6a52281f6555fc14bdd16c0bfe (diff) | |
download | rdflib-0986737fc8170d3cd3a7c95d078fa8f689267355.tar.gz |
Add test for Content-Type: application/sparql-update
-rw-r--r-- | test/test_sparqlstore.py | 78 |
1 files changed, 77 insertions, 1 deletions
diff --git a/test/test_sparqlstore.py b/test/test_sparqlstore.py index 26a69460..a0a93b57 100644 --- a/test/test_sparqlstore.py +++ b/test/test_sparqlstore.py @@ -4,7 +4,10 @@ import os import unittest from nose import SkipTest from requests import HTTPError - +from http.server import BaseHTTPRequestHandler, HTTPServer +import socket +from threading import Thread +import requests try: assert len(urlopen("http://dbpedia.org/sparql").read()) > 0 @@ -67,5 +70,78 @@ class SPARQLStoreDBPediaTestCase(unittest.TestCase): assert type(i[0]) == Literal, i[0].n3() +class SPARQLStoreUpdateTestCase(unittest.TestCase): + def setUp(self): + port = self.setup_mocked_endpoint() + self.graph = Graph(store="SPARQLUpdateStore", identifier=URIRef("urn:ex")) + self.graph.open(("http://localhost:{port}/query".format(port=port), + "http://localhost:{port}/update".format(port=port)), create=False) + ns = list(self.graph.namespaces()) + assert len(ns) > 0, ns + + def setup_mocked_endpoint(self): + # Configure mock server. + s = socket.socket(socket.AF_INET, type=socket.SOCK_STREAM) + s.bind(('localhost', 0)) + address, port = s.getsockname() + s.close() + mock_server = HTTPServer(('localhost', port), SPARQL11ProtocolStoreMock) + + # Start running mock server in a separate thread. + # Daemon threads automatically shut down when the main process exits. + mock_server_thread = Thread(target=mock_server.serve_forever) + mock_server_thread.setDaemon(True) + mock_server_thread.start() + print("Started mocked sparql endpoint on http://localhost:{port}/".format(port=port)) + return port + + def tearDown(self): + self.graph.close() + + def test_Query(self): + query = "insert data {<urn:s> <urn:p> <urn:o>}" + res = self.graph.update(query) + print(res) + + +class SPARQL11ProtocolStoreMock(BaseHTTPRequestHandler): + def do_POST(self): + """ + If the body should be analysed as well, just use: + ``` + body = self.rfile.read(int(self.headers['Content-Length'])).decode() + print(body) + ``` + """ + contenttype = self.headers.get("Content-Type") + if self.path == "/query": + if self.headers.get("Content-Type") == "application/sparql-query": + pass + elif self.headers.get("Content-Type") == "application/x-www-form-urlencoded": + pass + else: + self.send_response(requests.codes.not_acceptable) + self.end_headers() + elif self.path == "/update": + if self.headers.get("Content-Type") == "application/sparql-update": + pass + elif self.headers.get("Content-Type") == "application/x-www-form-urlencoded": + pass + else: + self.send_response(requests.codes.not_acceptable) + self.end_headers() + else: + self.send_response(requests.codes.not_found) + self.end_headers() + self.send_response(requests.codes.ok) + self.end_headers() + return + + def do_GET(self): + # Process an HTTP GET request and return a response with an HTTP 200 status. + self.send_response(requests.codes.ok) + self.end_headers() + return + if __name__ == '__main__': unittest.main() |