Testing Referential Opacity in RDF* triple stores

test_ref_opacity.py

#!/usr/bin/env python3
"""
This script checks different triple stores implementing RDF*,
to see how opaque/transparent are the terms used in embedded triples.
"""
from sys import argv, stderr
import base64
try:
    import requests
except:
    print("""
You need to install the 'requests' python module; e.g. with
        apt-get install python-requests
or
        pip install requests""",
        file=stderr)
    exit(-1)

ENDPOINTS = [
    ('GraphDB', 'http://localhost:7200/repositories/superman'),
    ('Stardog', 'http://localhost:5820/superman'),
    ('Corese', 'http://localhost:8080/sparql'),
    ('Fuseki', 'http://localhost:3030/superman'),
];

STARDOG_CREDENTIALS = "admin:admin"

def main():
    for endpoint in ENDPOINTS:
        if argv[1:] and endpoint[0] not in argv[1:]:
            continue
        print(endpoint[0]);
        sparql(endpoint, update="DELETE {?s ?p ?o} WHERE {?s ?p ?o}\n");
        sparql(endpoint, update="""INSERT DATA {
            << :superman :can :fly >> :reportedBy :clark.
            << _:x rdf:type :FlyingMan >> :reportedBy :clark.
            :superman owl:sameAs :clark.
            :superman :birthPlace :krypton.
            
            << :lifeUniverseAndEverything :anwser "042"^^xsd:integer >> :accordingTo :deepThought.
        }""")
        print("\tloaded", file=stderr)

        print("\tsupports owl:sameAs:\t", sparql(endpoint, """
            ASK { :clark :birthPlace :krypton }
        """))

        print("\tRDF* mode:\t", "PG" if sparql(endpoint, """
            ASK { :superman :can :fly }
        """) else "SA")

        print("\tref. transp. IRI:\t", sparql(endpoint, """
            ASK { << :clark :can :fly >> :reportedBy :clark }
        """))

        print("\tref. transp. lit.:\t", sparql(endpoint, """
            ASK { << :lifeUniverseAndEverything :anwser "42"^^xsd:integer >> :accordingTo :deepThought. }
        """))

        print("\tref. transp. bnode:\t", sparql(endpoint, """
            ASK { << _:x :can :fly >> :reportedBy :clark }
        """))

        print("\tref. transp. bnode2:\t", sparql(endpoint, """
            ASK { << _:y rdf:type :FlyingMan >> :reportedBy :clark }
        """))

        #print("\tquery emb. triple:\t", len(sparql(endpoint, """
        #    SELECT ?o { <<?s rdf:type ?o>> :reportedBy :clark }
        #"""))==1)


S = requests.session()
HEADERS = {
    'accept': 'application/sparql-results+json',
    'connection': 'keep-alive',
}

PREFIXES = """
    PREFIX : <http://example.org/ns/>
    PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
    PREFIX owl: <http://www.w3.org/2002/07/owl#>
    PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
"""


def sparql(endpoint, query=None, update=None) -> bool:
    if query is None and update is None:
        raise TypeError('query() missing argument query or update')
    if update is not None and update is not True:
        if query is None and isinstance(update, str):
            query = update
            update = True
        else:
            raise TypeError('incompatible mix of arguments query and update')

    name, url = endpoint
    config = CONFIG.get(name, {})
    url += config.get('update_url' if update else 'query_url', '')
    headers = HEADERS.copy()
    headers.update(config.get('headers', {}))
    qparam = config.get('update_param', 'query') if update else 'query'
    data = { qparam: PREFIXES + query }
    data.update(config.get('params', {}))

    res = S.post(url, data, headers=headers)
    if res.status_code // 100 != 2:
        print(f"{res.status_code} {res.reason}: {res.text}", file=stderr)
        if update:
            res.raise_for_status()
        return "(ERROR)"
    if not res.content or update:
        return
    result = res.json()
    if "boolean" in result:
        return result['boolean']
    else:
        return result['results']['bindings']

STARDOG_CREDENTIALS = base64.encodebytes(STARDOG_CREDENTIALS.encode('ascii')).decode('ascii').strip()

CONFIG = {
    'Stardog': {
        'query_url': '/query?reasoning=true&flushResponseHeaders=true',
        'update_url': '/update?reasoning=true&flushResponseHeaders=true',
        'headers': {
            'authorization': 'Basic ' + STARDOG_CREDENTIALS,
        }
    },
    'GraphDB': {
        'update_url': '/statements',
        'update_param': 'update',
        'params': {
            'infer': 'true',
            'sameAs': 'true',
        },
    },
    'Fuseki': {
        'query_url': '/query',
        'update_url': '/update',
        'update_param': 'update',
    }
}

main()