1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
# Copyright (C) 2024, Sam Thursfield <sam@afuera.me.uk>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.
"""
Test HTTP endpoint
"""
import unittest
import configuration
import fixtures
import json
import os
from pathlib import Path
import random
import shutil
from tempfile import mkdtemp
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen
COMMAND_NAME = "tinysparql"
class TestEndpointHttp(fixtures.TrackerCommandLineTestCase):
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def setUp(self):
super().setUp()
self._dirpath = Path(mkdtemp())
port = random.randint(32000, 65000)
self.address = "http://127.0.0.1:%d/sparql/" % port
# Create the database
self.run_background(
[
COMMAND_NAME,
"endpoint",
"--database",
self._dirpath,
"--ontology",
"nepomuk",
"--http-port",
port,
],
"Listening",
)
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def tearDown(self):
super().tearDown()
shutil.rmtree(self._dirpath)
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def test_http_query_cli(self):
"""Query the HTTP endpoint via `tracker3 sparql`"""
stdout = self.run_cli(
[
COMMAND_NAME,
"query",
"--remote-service",
self.address,
"ASK { ?u a rdfs:Resource }",
]
)
self.assertIn("true", stdout)
@staticmethod
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def example_ask_query() -> str:
"""Simple query that should always return 'true'."""
return "ASK { ?u a rdfs:Resource }"
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def validate_ask_query_response(self, data):
self.assertEqual(len(data["head"]["vars"]), 1);
self.assertEqual(len(data["results"]["bindings"]), 1);
var_name = data["head"]["vars"][0]
row = data["results"]["bindings"][0]
self.assertEqual(row[var_name]["type"], "literal")
self.assertEqual(row[var_name]["value"], "true")
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def test_http_get_without_query(self):
"""Get the endpoint descriptor, returned when there's no query."""
with urlopen(self.address) as response:
data = response.read().decode()
# Don't check the entire output, just make sure it's got some of the
# expected info.
self.assertIn("format:JSON-LD", data)
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def test_http_get_query(self):
query = quote(self.example_ask_query())
request = Request(f"{self.address}?query={query}")
request.add_header("Accept", "application/sparql-results+json");
with urlopen(request) as response:
text = response.read().decode()
data = json.loads(text)
self.validate_ask_query_response(data)
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def test_http_post_query(self):
query = "ASK { ?u a rdfs:Resource }"
request = Request(self.address, data=query.encode())
request.add_header("Accept", "application/sparql-results+json");
with urlopen(request) as response:
text = response.read().decode()
data = json.loads(text)
self.validate_ask_query_response(data)
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def test_missing_accept_header(self):
"""Ensure error code when there is no valid response format specified."""
query = "ASK { ?u a rdfs:Resource }"
request = Request(self.address, data=query.encode())
request.add_header("Accept", "Invalid");
request.add_header("Accept", r"Nonsense !\0");
with self.assertRaises(HTTPError) as error_context:
urlopen(request)
error = error_context.exception
self.assertEqual(error.code, 400);
self.assertIn(error.msg, "No recognized accepted formats");
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def test_http_invalid_utf8(self):
"""Ensure error code when the query is not valid UTF-8"""
query = "AB\xfc\0\0\0\xef"
request = Request(self.address, data=query.encode())
request.add_header("Accept", "application/sparql-results+json");
with self.assertRaises(HTTPError) as error_context:
urlopen(request)
error = error_context.exception
self.assertEqual(error.code, 400);
self.assertIn("invalid UTF-8", error.msg);
@unittest.skipIf ('AUTOPKGTEST_TMP' in os.environ, "It errors in autopkgtest")
def test_http_invalid_sparql(self):
"""Ensure error code when the query is not valid SPARQL"""
query = "Bananas in space"
request = Request(self.address, data=query.encode())
request.add_header("Accept", "application/sparql-results+json");
with self.assertRaises(HTTPError) as error_context:
urlopen(request)
error = error_context.exception
self.assertEqual(error.code, 400);
self.assertIn("Parser error", error.msg);
if __name__ == "__main__":
fixtures.tracker_test_main()
|
