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
|
from example.main import app
from fastapi.testclient import TestClient
from rdflib import Graph
from rdflib_endpoint.sparql_router import SD
# Use app defined in example folder
endpoint = TestClient(app)
def test_service_description():
# Check GET turtle
response = endpoint.get("/", headers={"accept": "text/turtle"})
print(response.text)
assert response.status_code == 200
g = Graph()
g.parse(data=response.text, format="turtle")
assert any(g.triples((None, SD.endpoint, None))), "Missing sd:endpoint in service description"
assert any(g.triples((None, SD.extensionFunction, None))), "Missing sd:extensionFunction in service description"
assert len(list(g.triples((None, SD.extensionFunction, None)))) == 2, "Expected exactly 2 extension functions"
assert len(list(g.triples((None, SD.namedGraph, None)))) == 2, "Expected exactly 2 named graphs"
# Check POST turtle
response = endpoint.post("/", headers={"accept": "text/turtle"})
assert response.status_code == 200
g = Graph()
g.parse(data=response.text, format="turtle")
assert any(g.triples((None, SD.endpoint, None))), "Missing sd:endpoint in service description"
assert any(g.triples((None, SD.extensionFunction, None))), "Missing sd:extensionFunction in service description"
assert len(list(g.triples((None, SD.extensionFunction, None)))) == 2, "Expected exactly 2 extension functions"
# Check POST XML
response = endpoint.post("/", headers={"accept": "application/xml"})
assert response.status_code == 200
g = Graph()
g.parse(data=response.text, format="xml")
assert any(g.triples((None, SD.endpoint, None))), "Missing sd:endpoint in service description"
assert any(g.triples((None, SD.extensionFunction, None))), "Missing sd:extensionFunction in service description"
assert len(list(g.triples((None, SD.extensionFunction, None)))) == 2, "Expected exactly 2 extension functions"
def test_custom_concat():
response = endpoint.get("/", params={"query": custom_concat_query}, headers={"accept": "application/json"})
# print(response.json())
assert response.status_code == 200
assert response.json()["results"]["bindings"][0]["concat"]["value"] == "Firstlast"
response = endpoint.post(
"/",
data={"query": custom_concat_query},
headers={"accept": "application/json"},
)
assert response.status_code == 200
assert response.json()["results"]["bindings"][0]["concat"]["value"] == "Firstlast"
def test_bad_request():
response = endpoint.get("/?query=figarofigarofigaro", headers={"accept": "application/json"})
assert response.status_code == 400
custom_concat_query = """PREFIX myfunctions: <https://w3id.org/sparql-functions/>
SELECT ?concat ?concatLength WHERE {
BIND("First" AS ?first)
BIND(myfunctions:custom_concat(?first, "last") AS ?concat)
}"""
|
