1from __future__ import print_function
2
3import sys
4from encodings.utf_8 import StreamWriter
5
6import unittest
7
8import os
9import os.path
10
11from six.moves.urllib.request import url2pathname, urlopen
12
13from rdflib import RDF, RDFS, URIRef, BNode, Literal, Namespace, Graph
14from rdflib.exceptions import ParserError
15from rdflib.util import first
16
17
18import logging
19
20_logger = logging.getLogger("parser_rdfcore")
21
22verbose = 0
23
24
25sw = StreamWriter(sys.stdout)
26
27
28def write(msg):
29    _logger.info(msg + "\n")
30    # sw.write(msg+"\n")
31
32
33class TestStore(Graph):
34    __test__ = False
35
36    def __init__(self, expected):
37        super(TestStore, self).__init__()
38        self.expected = expected
39
40    def add(self, spo):
41        (s, p, o) = spo
42        if not isinstance(s, BNode) and not isinstance(o, BNode):
43            if not (s, p, o) in self.expected:
44                m = "Triple not in expected result: %s, %s, %s" % (
45                    s.n3(), p.n3(), o.n3())
46                if verbose:
47                    write(m)
48                # raise Exception(m)
49        super(TestStore, self).add((s, p, o))
50
51
52TEST = Namespace("http://www.w3.org/2000/10/rdf-tests/rdfcore/testSchema#")
53
54CACHE_DIR = os.path.join(os.path.dirname(__file__), "rdf")
55
56skipped = (
57    # "datatypes/Manifest.rdf#test002",
58    # "rdf-containers-syntax-vs-schema/Manifest.rdf#test004",
59    # "rdfms-xml-literal-namespaces/Manifest.rdf#test001",
60    # "rdfms-xml-literal-namespaces/Manifest.rdf#test002",
61    # "rdfms-xmllang/Manifest.rdf#test001",
62    # "rdfms-xmllang/Manifest.rdf#test002",
63    # "xml-canon/Manifest.rdf#test001"
64)
65
66
67def cached_file(url):
68    fname = url2pathname(relative(url))
69
70    fpath = os.path.join(CACHE_DIR, fname)
71    if not os.path.exists(fpath):
72        print("%s does not exist, fetching from %s" % (fpath, url))
73        folder = os.path.dirname(fpath)
74        if not os.path.exists(folder):
75            os.makedirs(folder)
76        f = open(fpath, 'w')
77        try:
78            f.write(urlopen(url).read())
79        finally:
80            f.close()
81    return fpath
82
83
84RDFCOREBASE = "http://www.w3.org/2000/10/rdf-tests/rdfcore/"
85
86
87def relative(url):
88    return url[len(RDFCOREBASE):]
89
90
91def resolve(rel):
92    return RDFCOREBASE + rel
93
94
95def _testPositive(uri, manifest):
96    if verbose:
97        write(u"TESTING: %s" % uri)
98    result = 0  # 1=failed, 0=passed
99    inDoc = first(manifest.objects(uri, TEST["inputDocument"]))
100    outDoc = first(manifest.objects(uri, TEST["outputDocument"]))
101    expected = Graph()
102    if outDoc[-3:] == ".nt":
103        format = "nt"
104    else:
105        format = "xml"
106    expected.parse(cached_file(outDoc), publicID=outDoc, format=format)
107    store = TestStore(expected)
108    if inDoc[-3:] == ".nt":
109        format = "nt"
110    else:
111        format = "xml"
112
113    try:
114        store.parse(cached_file(inDoc), publicID=inDoc, format=format)
115    except ParserError as pe:
116        write("Failed '")
117        write(inDoc)
118        write("' failed with")
119        raise pe
120        try:
121            write(type(pe))
122        except:
123            write("sorry could not dump out error.")
124        result = 1
125    else:
126        if not store.isomorphic(expected):
127            write(u"""Failed: '%s'""" % uri)
128            if verbose:
129                write("""  In:\n""")
130                for s, p, o in store:
131                    write("%s %s %s." % (repr(s), repr(p), repr(o)))
132                write("""  Out:\n""")
133                for s, p, o in expected:
134                    write("%s %s %s." % (repr(s), repr(p), repr(o)))
135            result += 1
136    return result
137
138
139def _testNegative(uri, manifest):
140    if verbose:
141        write(u"TESTING: %s" % uri)
142    result = 0  # 1=failed, 0=passed
143    inDoc = first(manifest.objects(uri, TEST["inputDocument"]))
144    store = Graph()
145
146    test = BNode()
147    results.add((test, RESULT["test"], uri))
148    results.add((test, RESULT["system"], system))
149
150    try:
151        if inDoc[-3:] == ".nt":
152            format = "nt"
153        else:
154            format = "xml"
155        store.parse(cached_file(inDoc), publicID=inDoc, format=format)
156    except ParserError:
157        results.add((test, RDF.type, RESULT["PassingRun"]))
158        # pass
159    else:
160        write(u"""Failed: '%s'""" % uri)
161        results.add((test, RDF.type, RESULT["FailingRun"]))
162        result = 1
163    return result
164
165
166class ParserTestCase(unittest.TestCase):
167    store = 'default'
168    path = 'store'
169    slow = True
170
171    def setUp(self):
172        self.manifest = manifest = Graph(store=self.store)
173        manifest.open(self.path)
174        manifest.load(cached_file(
175            "http://www.w3.org/2000/10/rdf-tests/rdfcore/Manifest.rdf"))
176
177    def tearDown(self):
178        self.manifest.close()
179
180    def testNegative(self):
181        manifest = self.manifest
182        num_failed = total = 0
183        negs = list(manifest.subjects(RDF.type, TEST["NegativeParserTest"]))
184        negs.sort()
185        for neg in negs:
186            status = first(manifest.objects(neg, TEST["status"]))
187            if status == Literal("APPROVED"):
188                result = _testNegative(neg, manifest)
189                total += 1
190                num_failed += result
191        self.assertEqual(
192            num_failed, 0, "Failed: %s of %s." % (num_failed, total))
193
194    def testPositive(self):
195        manifest = self.manifest
196        uris = list(manifest.subjects(RDF.type, TEST["PositiveParserTest"]))
197        uris.sort()
198        num_failed = total = 0
199        for uri in uris:
200            status = first(manifest.objects(uri, TEST["status"]))
201            # Failing tests, skipped
202            if uri[44:] in skipped:
203                status = Literal("Locally DISAPPROVED")
204                write("Skipping %s" % uri)
205            if status == Literal("APPROVED"):
206                result = _testPositive(uri, manifest)
207                test = BNode()
208                results.add((test, RESULT["test"], uri))
209                results.add((test, RESULT["system"], system))
210                if not result:
211                    results.add((test, RDF.type, RESULT["PassingRun"]))
212                else:
213                    results.add((test, RDF.type, RESULT["FailingRun"]))
214                total += 1
215                num_failed += result
216        self.assertEqual(
217            num_failed, 0, "Failed: %s of %s." % (num_failed, total))
218
219
220RESULT = Namespace("http://www.w3.org/2002/03owlt/resultsOntology#")
221FOAF = Namespace("http://xmlns.com/foaf/0.1/")
222
223
224results = Graph()
225
226system = BNode("system")
227results.add((system, FOAF["homepage"], URIRef("http://rdflib.net/")))
228results.add((system, RDFS.label, Literal("RDFLib")))
229results.add((system, RDFS.comment, Literal("")))
230
231
232if __name__ == "__main__":
233    manifest = Graph()
234    manifest.load(cached_file(
235        "http://www.w3.org/2000/10/rdf-tests/rdfcore/Manifest.rdf"))
236    import sys
237    import getopt
238    try:
239        optlist, args = getopt.getopt(sys.argv[1:], 'h:', ["help"])
240    except getopt.GetoptError as msg:
241        write(msg)
242        # usage()
243
244    try:
245        argv = sys.argv
246        if len(argv) > 1:
247            _logger.setLevel(logging.INFO)
248            _logger.addHandler(logging.StreamHandler())
249
250        for arg in argv[1:]:
251            verbose = 1
252            case = URIRef(arg)
253            write(u"Testing: %s" % case)
254            if (case, RDF.type, TEST["PositiveParserTest"]) in manifest:
255                result = _testPositive(case, manifest)
256                write(u"Positive test %s" % ["PASSED", "FAILED"][result])
257            elif (case, RDF.type, TEST["NegativeParserTest"]) in manifest:
258                result = _testNegative(case, manifest)
259                write(u"Negative test %s" % ["PASSED", "FAILED"][result])
260            else:
261                write(u"%s not ??" % case)
262
263        if len(argv) <= 1:
264            unittest.main()
265    finally:
266        results.serialize("results.rdf")
267