aboutsummaryrefslogtreecommitdiffstats
path: root/bindings/python/test_serd.py
diff options
context:
space:
mode:
Diffstat (limited to 'bindings/python/test_serd.py')
-rw-r--r--bindings/python/test_serd.py994
1 files changed, 994 insertions, 0 deletions
diff --git a/bindings/python/test_serd.py b/bindings/python/test_serd.py
new file mode 100644
index 00000000..74f125fd
--- /dev/null
+++ b/bindings/python/test_serd.py
@@ -0,0 +1,994 @@
+# Copyright 2020 David Robillard <d@drobilla.net>
+#
+# Permission to use, copy, modify, and/or distribute this software for any
+# purpose with or without fee is hereby granted, provided that the above
+# copyright notice and this permission notice appear in all copies.
+#
+# THIS SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+import base64
+import math
+import os
+import serd
+import shutil
+import tempfile
+import textwrap
+import unittest
+
+
+class StringTests(unittest.TestCase):
+ def testStrerror(self):
+ self.assertEqual(serd.strerror(serd.Status.SUCCESS), "Success")
+ self.assertEqual(serd.strerror(99999), "Unknown error")
+
+ with self.assertRaises(OverflowError):
+ serd.strerror(-1)
+
+ self.assertEqual(
+ serd.strerror(serd.Status.ERR_BAD_WRITE), "Error writing to file"
+ )
+
+ def testStrlen(self):
+ self.assertEqual(serd.strlen("hello"), (5, 0))
+ self.assertEqual(
+ serd.strlen("new\nline"), (8, serd.NodeFlags.HAS_NEWLINE)
+ )
+
+ def testStrtod(self):
+ self.assertEqual(serd.strtod("42"), 42.0)
+ self.assertEqual(serd.strtod("1.234 hello"), 1.234)
+ self.assertTrue(math.isnan(serd.strtod("not a number")))
+
+
+class Base64Tests(unittest.TestCase):
+ def testShortBase64(self):
+ data = "foobar".encode("utf-8")
+ encoded = "Zm9vYmFy"
+
+ self.assertEqual(serd.base64_encode(data), encoded)
+ self.assertEqual(serd.base64_encode(data, True), encoded)
+ self.assertEqual(serd.base64_decode(encoded), data)
+
+ def testLongBase64(self):
+ data = ("foobar" * 20).encode("utf-8")
+ oneline_encoded = "Zm9vYmFy" * 20
+ multiline_encoded = "\n".join(textwrap.wrap("Zm9vYmFy" * 20, width=76))
+
+ self.assertEqual(serd.base64_encode(data), oneline_encoded)
+ self.assertEqual(serd.base64_encode(data, True), multiline_encoded)
+ self.assertEqual(serd.base64_decode(oneline_encoded), data)
+ self.assertEqual(serd.base64_decode(multiline_encoded), data)
+
+
+class SyntaxTests(unittest.TestCase):
+ def testSyntaxByName(self):
+ self.assertEqual(serd.syntax_by_name("TuRtLe"), serd.Syntax.TURTLE)
+ self.assertEqual(serd.syntax_by_name("wat"), serd.Syntax.EMPTY)
+
+ def testGuessSyntax(self):
+ self.assertEqual(serd.guess_syntax("foo.nq"), serd.Syntax.NQUADS)
+ self.assertEqual(serd.guess_syntax("foo.txt"), serd.Syntax.EMPTY)
+
+ def testSyntaxHasGraphs(self):
+ self.assertFalse(serd.syntax_has_graphs(serd.Syntax.EMPTY))
+ self.assertFalse(serd.syntax_has_graphs(serd.Syntax.TURTLE))
+ self.assertFalse(serd.syntax_has_graphs(serd.Syntax.NTRIPLES))
+ self.assertTrue(serd.syntax_has_graphs(serd.Syntax.NQUADS))
+ self.assertTrue(serd.syntax_has_graphs(serd.Syntax.TRIG))
+
+
+class WorldTests(unittest.TestCase):
+ def setUp(self):
+ self.world = serd.World()
+
+ def testGetBlank(self):
+ self.assertEqual(self.world.get_blank(), serd.blank("b1"))
+ self.assertEqual(self.world.get_blank(), serd.blank("b2"))
+
+
+class NodeTests(unittest.TestCase):
+ @staticmethod
+ def _throughSyntax(n):
+ return serd.Node.from_syntax(n.to_syntax())
+
+ def testConstruction(self):
+ self.assertEqual(serd.Node("hello"), serd.plain_literal("hello"))
+ self.assertEqual(serd.Node(False), serd.boolean(False))
+ self.assertEqual(serd.Node(True), serd.boolean(True))
+ self.assertEqual(serd.Node(42), serd.integer(42))
+ self.assertEqual(serd.Node(42.34), serd.double(42.34))
+
+ def testString(self):
+ n = serd.string("hello")
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "hello")
+ self.assertEqual(len(n), 5)
+ self.assertEqual(repr(n), 'serd.string("hello")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testPlainLiteral(self):
+ n = serd.plain_literal("hallo", "de")
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "hallo")
+ self.assertEqual(len(n), 5)
+ self.assertEqual(repr(n), 'serd.plain_literal("hallo", "de")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertEqual(n.language(), serd.string("de"))
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_LANGUAGE)
+
+ def testTypedLiteral(self):
+ datatype = serd.uri("http://example.org/ns#Hex")
+ n = serd.typed_literal("ABCD", datatype)
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "ABCD")
+ self.assertEqual(len(n), 4)
+ self.assertEqual(
+ repr(n), 'serd.typed_literal("ABCD", "http://example.org/ns#Hex")'
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(n.datatype(), datatype)
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ def testBlank(self):
+ n = serd.blank("b0")
+ self.assertEqual(n.type(), serd.NodeType.BLANK)
+ self.assertEqual(n, "b0")
+ self.assertEqual(len(n), 2)
+ self.assertEqual(repr(n), 'serd.blank("b0")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testCurie(self):
+ n = serd.curie("ns:name")
+ self.assertEqual(n.type(), serd.NodeType.CURIE)
+ self.assertEqual(n, "ns:name")
+ self.assertEqual(len(n), 7)
+ self.assertEqual(repr(n), 'serd.curie("ns:name")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testUri(self):
+ n = serd.uri("http://example.org/")
+ self.assertEqual(n.type(), serd.NodeType.URI)
+ self.assertEqual(n, "http://example.org/")
+ self.assertEqual(len(n), 19)
+ self.assertEqual(repr(n), 'serd.uri("http://example.org/")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testRelativeUri(self):
+ n = serd.uri("rel/uri")
+ self.assertEqual(n.type(), serd.NodeType.URI)
+ self.assertEqual(n, "rel/uri")
+ self.assertEqual(len(n), 7)
+ self.assertEqual(repr(n), 'serd.uri("rel/uri")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testResolvedUri(self):
+ base = serd.uri("http://example.org/")
+ n = serd.resolved_uri("name", base)
+ self.assertEqual(n.type(), serd.NodeType.URI)
+ self.assertEqual(n, "http://example.org/name")
+ self.assertEqual(len(n), 23)
+ self.assertEqual(repr(n), 'serd.uri("http://example.org/name")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testLocalFileUri(self):
+ n = serd.file_uri("/foo/bar")
+ self.assertEqual(n.type(), serd.NodeType.URI)
+ self.assertEqual(n, "file:///foo/bar")
+ self.assertEqual(len(n), 15)
+ self.assertEqual(repr(n), 'serd.uri("file:///foo/bar")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testFileUriWithHostname(self):
+ n = serd.file_uri("/foo/bar", "host")
+ self.assertEqual(n.type(), serd.NodeType.URI)
+ self.assertEqual(n, "file://host/foo/bar")
+ self.assertEqual(len(n), 19)
+ self.assertEqual(repr(n), 'serd.uri("file://host/foo/bar")')
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testDecimal(self):
+ xsd_decimal = "http://www.w3.org/2001/XMLSchema#decimal"
+
+ n = serd.decimal(12.34, 7, 4, None)
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "12.34")
+ self.assertEqual(len(n), 5)
+ self.assertEqual(
+ repr(n),
+ 'serd.typed_literal("12.34", "http://www.w3.org/2001/XMLSchema#decimal")',
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(n.datatype(), serd.uri(xsd_decimal))
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ n = serd.decimal(12.34)
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "12.34")
+ self.assertEqual(len(n), 5)
+ self.assertEqual(
+ repr(n), 'serd.typed_literal("12.34", "{}")'.format(xsd_decimal)
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(n.datatype(), serd.uri(xsd_decimal))
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ datatype = "http://example.org/ns#Decimal"
+ d = serd.decimal(1234, datatype=serd.uri(datatype))
+ self.assertEqual(d.datatype(), serd.uri(datatype))
+ self.assertEqual(d, "1234.0")
+ self.assertEqual(len(d), 6)
+ self.assertEqual(
+ repr(d), 'serd.typed_literal("1234.0", "%s")' % datatype
+ )
+ self.assertEqual(d, eval(repr(d)))
+ self.assertEqual(n, self._throughSyntax(n))
+
+ def testDouble(self):
+ xsd_double = "http://www.w3.org/2001/XMLSchema#double"
+ n = serd.double(12.34)
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "1.234E1")
+ self.assertEqual(len(n), 7)
+ self.assertEqual(
+ repr(n), 'serd.typed_literal("1.234E1", "{}")'.format(xsd_double)
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(n.datatype(), serd.uri(xsd_double))
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ def testFloat(self):
+ xsd_float = "http://www.w3.org/2001/XMLSchema#float"
+ n = serd.float(234.5)
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "2.345E2")
+ self.assertEqual(len(n), 7)
+ self.assertEqual(
+ repr(n), 'serd.typed_literal("2.345E2", "{}")'.format(xsd_float)
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(n.datatype(), serd.uri(xsd_float))
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ def testInteger(self):
+ xsd_integer = "http://www.w3.org/2001/XMLSchema#integer"
+ n = serd.integer(42)
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(n, "42")
+ self.assertEqual(len(n), 2)
+ self.assertEqual(
+ repr(n), 'serd.typed_literal("42", "{}")'.format(xsd_integer)
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(n.datatype(), serd.uri(xsd_integer))
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ datatype = "http://example.org/ns#Integer"
+ d = serd.decimal(1234, datatype=serd.uri(datatype))
+ self.assertEqual(d.datatype(), serd.uri(datatype))
+ self.assertEqual(d, "1234.0")
+ self.assertEqual(len(d), 6)
+ self.assertEqual(
+ repr(d), 'serd.typed_literal("1234.0", "{}")'.format(datatype)
+ )
+ self.assertEqual(d, eval(repr(d)))
+ self.assertEqual(n, self._throughSyntax(n))
+
+ def testBoolean(self):
+ xsd_boolean = "http://www.w3.org/2001/XMLSchema#boolean"
+ t = serd.boolean(True)
+ self.assertEqual(t.type(), serd.NodeType.LITERAL)
+ self.assertEqual(t, "true")
+ self.assertEqual(len(t), 4)
+ self.assertEqual(repr(t), "serd.boolean(True)")
+ self.assertEqual(t, eval(repr(t)))
+ self.assertEqual(t, self._throughSyntax(t))
+ self.assertEqual(t.datatype(), serd.uri(xsd_boolean))
+ self.assertIsNone(t.language())
+ self.assertEqual(t.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ f = serd.boolean(False)
+ self.assertEqual(f.type(), serd.NodeType.LITERAL)
+ self.assertEqual(f, "false")
+ self.assertEqual(len(f), 5)
+ self.assertEqual(repr(f), "serd.boolean(False)")
+ self.assertEqual(f, eval(repr(f)))
+ self.assertEqual(f, self._throughSyntax(f))
+ self.assertEqual(f.datatype(), serd.uri(xsd_boolean))
+ self.assertIsNone(f.language())
+ self.assertEqual(f.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ def testBlob(self):
+ xsd_base64Binary = "http://www.w3.org/2001/XMLSchema#base64Binary"
+ n = serd.blob(b"DEAD")
+ n_bytes = base64.b64encode(b"DEAD")
+ self.assertEqual(n.type(), serd.NodeType.LITERAL)
+ self.assertEqual(bytes(str(n), "utf-8"), n_bytes)
+ self.assertEqual(len(n), 8)
+ self.assertEqual(
+ repr(n),
+ 'serd.typed_literal("{}", "{}")'.format(
+ n_bytes.decode("utf-8"), xsd_base64Binary
+ ),
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(n.datatype(), serd.uri(xsd_base64Binary))
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ datatype = "http://example.org/ns#Blob"
+ t = serd.blob(b"BEEF", datatype=serd.uri(datatype))
+ t_bytes = base64.b64encode(b"BEEF")
+ self.assertEqual(t.type(), serd.NodeType.LITERAL)
+ self.assertEqual(bytes(str(t), "utf-8"), t_bytes)
+ self.assertEqual(len(t), 8)
+ self.assertEqual(
+ repr(t),
+ 'serd.typed_literal("{}", "{}")'.format(
+ t_bytes.decode("utf-8"), datatype
+ ),
+ )
+ self.assertEqual(n, eval(repr(n)))
+ self.assertEqual(n, self._throughSyntax(n))
+ self.assertEqual(t.datatype(), serd.uri(datatype))
+ self.assertIsNone(t.language())
+ self.assertEqual(t.flags(), serd.NodeFlags.HAS_DATATYPE)
+
+ def testVariable(self):
+ n = serd.variable("foo")
+ self.assertEqual(n.type(), serd.NodeType.VARIABLE)
+ self.assertEqual(n, "foo")
+ self.assertEqual(len(n), 3)
+ self.assertEqual(repr(n), 'serd.variable("foo")')
+ self.assertEqual(n, eval(repr(n)))
+ # self.assertEqual(n, self._throughSyntax(n))
+ self.assertIsNone(n.datatype())
+ self.assertIsNone(n.language())
+ self.assertEqual(n.flags(), 0)
+
+ def testComparison(self):
+ a = serd.string("Aardvark")
+ b = serd.string("Banana")
+
+ self.assertEqual(a, a)
+ self.assertNotEqual(a, b)
+ self.assertLess(a, b)
+ self.assertLessEqual(a, b)
+ self.assertLessEqual(a, a)
+ self.assertGreater(b, a)
+ self.assertGreaterEqual(b, a)
+ self.assertGreaterEqual(b, b)
+
+
+class Env(unittest.TestCase):
+ def testEquality(self):
+ uri = serd.uri("http://example.org/")
+ env1 = serd.Env()
+ env2 = serd.Env()
+ self.assertEqual(env1, env2)
+
+ env2.set_base_uri(uri)
+ self.assertNotEqual(env1, env2)
+
+ env2.set_base_uri(None)
+ self.assertEqual(env1, env2)
+
+ env2.set_prefix("eg", uri)
+ self.assertNotEqual(env1, env2)
+
+ env1.set_prefix(serd.string("eg"), uri)
+ self.assertEqual(env1, env2)
+
+ def testBaseUri(self):
+ env = serd.Env()
+ self.assertIsNone(env.base_uri())
+
+ base = serd.uri("http://example.org/")
+ env.set_base_uri(base)
+ self.assertEqual(env.base_uri(), base)
+
+ def testInitialBaseUri(self):
+ base = serd.uri("http://example.org/")
+ env = serd.Env(base)
+ self.assertEqual(env.base_uri(), base)
+
+ def testQualify(self):
+ base = serd.uri("http://example.org/")
+ uri = serd.uri("http://example.org/name")
+ env = serd.Env(base)
+
+ self.assertIsNone(env.qualify(uri))
+
+ env.set_prefix("eg", base)
+ self.assertEqual(env.qualify(uri), "eg:name")
+
+ def testExpand(self):
+ base = serd.uri("http://example.org/")
+ curie = serd.curie("eg:name")
+ env = serd.Env(base)
+
+ self.assertIsNone(env.expand(curie))
+
+ env.set_prefix("eg", base)
+ self.assertEqual(env.expand(curie), serd.uri("http://example.org/name"))
+
+
+class ModelTests(unittest.TestCase):
+ def setUp(self):
+ self.world = serd.World()
+ self.s = serd.uri("http://example.org/s")
+ self.p = serd.uri("http://example.org/p")
+ self.o = serd.uri("http://example.org/o")
+ self.o1 = serd.uri("http://example.org/o1")
+ self.o2 = serd.uri("http://example.org/o2")
+ self.g = serd.uri("http://example.org/g")
+ self.x = serd.uri("http://example.org/x")
+
+ def testConstruction(self):
+ flags = serd.ModelFlags.INDEX_SPO | serd.ModelFlags.INDEX_GRAPHS
+ model = serd.Model(self.world, flags)
+ self.assertEqual(model.flags(), flags)
+ self.assertNotEqual(model.flags(), serd.ModelFlags.INDEX_SPO)
+ self.assertEqual(model.world(), self.world)
+
+ def testInsertErase(self):
+ model = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+
+ model.insert((self.s, self.p, self.o))
+ self.assertEqual(len(model), 1)
+ model.erase(iter(model))
+ self.assertEqual(len(model), 0)
+
+ statement = serd.Statement(self.s, self.p, self.o)
+ model += statement
+ self.assertEqual(len(model), 1)
+ del model[statement]
+ self.assertEqual(len(model), 0)
+
+ def testSize(self):
+ model = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+ self.assertEqual(model.size(), 0)
+ self.assertEqual(len(model), 0)
+ self.assertTrue(model.empty())
+
+ model.insert((self.s, self.p, self.o))
+ self.assertEqual(model.size(), 1)
+ self.assertEqual(len(model), 1)
+ self.assertFalse(model.empty())
+
+ model.erase(iter(model))
+ self.assertEqual(model.size(), 0)
+ self.assertEqual(len(model), 0)
+ self.assertTrue(model.empty())
+
+ def testBeginEnd(self):
+ s, p, o, g = self.s, self.p, self.o, self.g
+ model = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+
+ self.assertEqual(model.begin(), model.end())
+
+ model.insert((s, p, o, g))
+ self.assertNotEqual(model.begin(), model.end())
+
+ def testFind(self):
+ s, p, o, g, x = self.s, self.p, self.o, self.g, self.x
+ flags = serd.ModelFlags.INDEX_SPO | serd.ModelFlags.INDEX_GRAPHS
+ model = serd.Model(self.world, flags)
+ in_statement = serd.Statement(s, p, o, g)
+ out_statement = serd.Statement(x, p, o, g)
+
+ model += in_statement
+ self.assertEqual(model.find(out_statement), model.end())
+ self.assertNotEqual(model.find(in_statement), model.end())
+
+ def testGet(self):
+ s, p, o, g = self.s, self.p, self.o, self.g
+ flags = serd.ModelFlags.INDEX_SPO | serd.ModelFlags.INDEX_GRAPHS
+ model = serd.Model(self.world, flags)
+
+ model.insert((s, p, o, g))
+ self.assertEqual(model.get(None, p, o, g), s)
+ self.assertEqual(model.get(s, None, o, g), p)
+ self.assertEqual(model.get(s, p, None, g), o)
+ self.assertEqual(model.get(s, p, o, None), g)
+
+ def testAsk(self):
+ s, p, o, g, x = self.s, self.p, self.o, self.g, self.x
+ flags = serd.ModelFlags.INDEX_SPO | serd.ModelFlags.INDEX_GRAPHS
+ model = serd.Model(self.world, flags)
+ model.insert((s, p, o, g))
+
+ self.assertTrue(model.ask(s, p, o, g))
+ self.assertIn(serd.Statement(s, p, o, g), model)
+ self.assertIn((s, p, o, g), model)
+
+ self.assertFalse(model.ask(x, p, o, g))
+ self.assertNotIn(serd.Statement(x, p, o, g), model)
+ self.assertNotIn((x, p, o, g), model)
+
+ self.assertTrue(model.ask(None, p, o, g))
+ self.assertTrue(model.ask(s, None, o, g))
+ self.assertTrue(model.ask(s, p, None, g))
+ self.assertTrue(model.ask(s, p, o, None))
+
+ self.assertFalse(model.ask(None, x, o, g))
+ self.assertFalse(model.ask(s, None, x, g))
+ self.assertFalse(model.ask(s, p, None, x))
+ self.assertFalse(model.ask(x, p, o, None))
+
+ def testCount(self):
+ s, p, o1, o2, g, x = self.s, self.p, self.o1, self.o2, self.g, self.x
+ flags = serd.ModelFlags.INDEX_SPO | serd.ModelFlags.INDEX_GRAPHS
+ model = serd.Model(self.world, flags)
+ model.insert((s, p, o1, g))
+ model.insert((s, p, o2, g))
+
+ self.assertEqual(model.count(s, p, o1, g), 1)
+ self.assertEqual(model.count(s, p, None, g), 2)
+ self.assertEqual(model.count(s, p, x, g), 0)
+
+
+class StatementTests(unittest.TestCase):
+ def setUp(self):
+ self.s = serd.uri("http://example.org/s")
+ self.p = serd.uri("http://example.org/p")
+ self.o = serd.uri("http://example.org/o")
+ self.g = serd.uri("http://example.org/g")
+ self.cursor = serd.Cursor("foo.ttl", 1, 0)
+
+ def testAllFields(self):
+ s, p, o, g = self.s, self.p, self.o, self.g
+ statement = serd.Statement(s, p, o, g, self.cursor)
+
+ self.assertEqual(statement.node(serd.Field.SUBJECT), s)
+ self.assertEqual(statement.node(serd.Field.PREDICATE), p)
+ self.assertEqual(statement.node(serd.Field.OBJECT), o)
+ self.assertEqual(statement.node(serd.Field.GRAPH), g)
+
+ self.assertEqual(statement.subject(), s)
+ self.assertEqual(statement.predicate(), p)
+ self.assertEqual(statement.object(), o)
+ self.assertEqual(statement.graph(), g)
+
+ self.assertEqual(statement.cursor(), self.cursor)
+
+ def testNoGraph(self):
+ s, p, o = self.s, self.p, self.o
+ statement = serd.Statement(s, p, o, None, self.cursor)
+
+ self.assertEqual(statement.node(serd.Field.SUBJECT), s)
+ self.assertEqual(statement.node(serd.Field.PREDICATE), p)
+ self.assertEqual(statement.node(serd.Field.OBJECT), o)
+ self.assertIsNone(statement.node(serd.Field.GRAPH))
+
+ self.assertEqual(statement.subject(), s)
+ self.assertEqual(statement.predicate(), p)
+ self.assertEqual(statement.object(), o)
+ self.assertIsNone(statement.graph())
+
+ self.assertEqual(statement.cursor(), self.cursor)
+
+ def testNoCursor(self):
+ s, p, o, g = self.s, self.p, self.o, self.g
+ statement = serd.Statement(s, p, o, g)
+
+ self.assertEqual(statement.node(serd.Field.SUBJECT), s)
+ self.assertEqual(statement.node(serd.Field.PREDICATE), p)
+ self.assertEqual(statement.node(serd.Field.OBJECT), o)
+ self.assertEqual(statement.node(serd.Field.GRAPH), g)
+
+ self.assertEqual(statement.subject(), s)
+ self.assertEqual(statement.predicate(), p)
+ self.assertEqual(statement.object(), o)
+ self.assertEqual(statement.graph(), g)
+
+ self.assertIsNone(statement.cursor())
+
+ def testNoGraphOrCursor(self):
+ s, p, o = self.s, self.p, self.o
+ statement = serd.Statement(s, p, o)
+
+ self.assertEqual(statement.node(serd.Field.SUBJECT), s)
+ self.assertEqual(statement.node(serd.Field.PREDICATE), p)
+ self.assertEqual(statement.node(serd.Field.OBJECT), o)
+ self.assertIsNone(statement.node(serd.Field.GRAPH))
+
+ self.assertEqual(statement.subject(), s)
+ self.assertEqual(statement.predicate(), p)
+ self.assertEqual(statement.object(), o)
+ self.assertIsNone(statement.graph())
+
+ self.assertIsNone(statement.cursor())
+
+ def testComparison(self):
+ s, p, o, g = self.s, self.p, self.o, self.g
+ statement1 = serd.Statement(s, p, o, g)
+ statement2 = serd.Statement(o, p, s, g)
+
+ self.assertEqual(statement1, statement1)
+ self.assertNotEqual(statement1, statement2)
+
+ def testMatches(self):
+ s, p, o, g = self.s, self.p, self.o, self.g
+ x = serd.uri("http://example.org/x")
+ statement = serd.Statement(s, p, o, g)
+
+ self.assertTrue(statement.matches(s, p, o, g))
+ self.assertTrue(statement.matches(None, p, o, g))
+ self.assertTrue(statement.matches(s, None, o, g))
+ self.assertTrue(statement.matches(s, p, None, g))
+ self.assertTrue(statement.matches(s, p, o, None))
+
+ self.assertFalse(statement.matches(x, p, o, g))
+ self.assertFalse(statement.matches(s, x, o, g))
+ self.assertFalse(statement.matches(s, p, x, g))
+ self.assertFalse(statement.matches(s, p, o, x))
+
+ def testStr(self):
+ self.assertEqual(
+ str(serd.Statement(self.s, self.p, self.o)),
+ "<http://example.org/s> <http://example.org/p> <http://example.org/o>",
+ )
+
+ self.assertEqual(
+ str(serd.Statement(self.s, self.p, self.o, self.g)),
+ "<http://example.org/s> <http://example.org/p> <http://example.org/o> <http://example.org/g>",
+ )
+
+ def testRepr(self):
+ self.assertEqual(
+ repr(serd.Statement(self.s, self.p, self.o)),
+ 'serd.Statement(serd.uri("http://example.org/s"), serd.uri("http://example.org/p"), serd.uri("http://example.org/o"))',
+ )
+
+ self.assertEqual(
+ repr(serd.Statement(self.s, self.p, self.o, self.g)),
+ 'serd.Statement(serd.uri("http://example.org/s"), serd.uri("http://example.org/p"), serd.uri("http://example.org/o"), serd.uri("http://example.org/g"))',
+ )
+
+
+class RangeTests(unittest.TestCase):
+ def setUp(self):
+ self.world = serd.World()
+ self.s = serd.uri("http://example.org/s")
+ self.p = serd.uri("http://example.org/p")
+ self.p1 = serd.uri("http://example.org/p1")
+ self.p2 = serd.uri("http://example.org/p2")
+ self.o1 = serd.uri("http://example.org/o1")
+ self.o2 = serd.uri("http://example.org/o2")
+ self.g = serd.uri("http://example.org/g")
+
+ def testFront(self):
+ model = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+
+ model.insert((self.s, self.p, self.o1))
+ self.assertEqual(
+ model.all().front(), serd.Statement(self.s, self.p, self.o1)
+ )
+
+ def testEmpty(self):
+ model = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+
+ self.assertTrue(model.all().empty())
+ self.assertFalse(model.all())
+
+ model.insert((self.s, self.p, self.o1))
+ self.assertFalse(model.all().empty())
+ self.assertTrue(model.all())
+
+ def testIteration(self):
+ model = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+
+ model.insert((self.s, self.p, self.o1))
+ model.insert((self.s, self.p, self.o2))
+
+ i = iter(model.all())
+ self.assertEqual(next(i), serd.Statement(self.s, self.p, self.o1))
+ self.assertEqual(next(i), serd.Statement(self.s, self.p, self.o2))
+ with self.assertRaises(StopIteration):
+ next(i)
+
+ def testEmptyIteration(self):
+ model = serd.Model(self.world)
+ count = 0
+
+ for s in model:
+ count += 1
+
+ self.assertEqual(count, 0)
+
+ for s in model.all():
+ count += 1
+
+ self.assertEqual(count, 0)
+
+ def testInsertErase(self):
+ model1 = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+ model2 = serd.Model(self.world, serd.ModelFlags.INDEX_SPO)
+
+ model1.insert((self.s, self.p1, self.o1))
+ model1.insert((self.s, self.p1, self.o2))
+ model1.insert((self.s, self.p2, self.o1))
+ model1.insert((self.s, self.p2, self.o2))
+
+ model2.insert(model1.range((self.s, self.p1, None)))
+
+ self.assertEqual(
+ [s for s in model2],
+ [
+ serd.Statement(self.s, self.p1, self.o1),
+ serd.Statement(self.s, self.p1, self.o2),
+ ],
+ )
+
+ model1.erase(model1.range((self.s, self.p2, None)))
+ self.assertEqual(model1, model2)
+
+
+class CursorTests(unittest.TestCase):
+ def testStringConstruction(self):
+ cur = serd.Cursor("foo.ttl", 3, 4)
+ self.assertEqual(cur.name(), "foo.ttl")
+ self.assertEqual(cur.line(), 3)
+ self.assertEqual(cur.column(), 4)
+
+ def testNodeConstruction(self):
+ name = serd.string("foo.ttl")
+ cur = serd.Cursor(name, 5, 6)
+ self.assertEqual(cur.name(), name)
+ self.assertEqual(cur.line(), 5)
+ self.assertEqual(cur.column(), 6)
+
+ def testComparison(self):
+ self.assertEqual(
+ serd.Cursor("foo.ttl", 1, 2), serd.Cursor("foo.ttl", 1, 2)
+ )
+ self.assertNotEqual(
+ serd.Cursor("foo.ttl", 9, 2), serd.Cursor("foo.ttl", 1, 2)
+ )
+ self.assertNotEqual(
+ serd.Cursor("foo.ttl", 1, 9), serd.Cursor("foo.ttl", 1, 2)
+ )
+ self.assertNotEqual(
+ serd.Cursor("bar.ttl", 1, 2), serd.Cursor("foo.ttl", 1, 2)
+ )
+
+
+class ReaderTests(unittest.TestCase):
+ def setUp(self):
+ self.world = serd.World()
+ self.temp_dir = tempfile.mkdtemp()
+ self.ttl_path = os.path.join(self.temp_dir, "input.ttl")
+ self.s = serd.curie("eg:s")
+ self.p1 = serd.curie("eg:p1")
+ self.p2 = serd.curie("eg:p2")
+ self.o1 = serd.curie("eg:o1")
+ self.o2 = serd.curie("eg:o2")
+
+ self.ttl_document = """@prefix eg: <http://example.org/> .
+@base <http://example.org/base> .
+eg:s eg:p1 eg:o1 ;
+eg:p2 eg:o2 .
+"""
+ self.events = [
+ serd.Event.prefix("eg", "http://example.org/"),
+ serd.Event.base("http://example.org/base"),
+ serd.Event.statement(serd.Statement(self.s, self.p1, self.o1)),
+ serd.Event.statement(serd.Statement(self.s, self.p2, self.o2)),
+ ]
+
+ with open(self.ttl_path, "w") as f:
+ f.write(self.ttl_document)
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir)
+
+ def testReadFileToSink(self):
+ class TestSink(serd.Sink):
+ def __init__(self):
+ super().__init__()
+ self.events = []
+
+ def __call__(self, event):
+ self.events += [event]
+ return serd.Status.SUCCESS
+
+ s, p1, p2, o1, o2 = self.s, self.p1, self.p2, self.o1, self.o2
+
+ source = serd.FileSource(self.ttl_path)
+ sink = TestSink()
+ reader = serd.Reader(self.world, serd.Syntax.TURTLE, 0, sink, 4096)
+
+ self.assertEqual(reader.start(source), serd.Status.SUCCESS)
+ self.assertEqual(sink.events, [])
+ self.assertEqual(reader.read_document(), serd.Status.SUCCESS)
+ self.assertEqual(reader.finish(), serd.Status.SUCCESS)
+ self.assertEqual(sink.events, self.events)
+
+ def testReadFileToFunction(self):
+ captured_events = []
+
+ def sink(event):
+ captured_events.append(event)
+
+ s, p1, p2, o1, o2 = self.s, self.p1, self.p2, self.o1, self.o2
+
+ source = serd.FileSource(self.ttl_path)
+ reader = serd.Reader(self.world, serd.Syntax.TURTLE, 0, sink, 4096)
+
+ self.assertEqual(reader.start(source), serd.Status.SUCCESS)
+ self.assertEqual(captured_events, [])
+ self.assertEqual(reader.read_document(), serd.Status.SUCCESS)
+ self.assertEqual(reader.finish(), serd.Status.SUCCESS)
+ self.assertEqual(captured_events, self.events)
+
+ def testReadStringToFunction(self):
+ captured_events = []
+
+ def sink(event):
+ captured_events.append(event)
+
+ s, p1, p2, o1, o2 = self.s, self.p1, self.p2, self.o1, self.o2
+
+ source = serd.StringSource(self.ttl_document)
+ reader = serd.Reader(self.world, serd.Syntax.TURTLE, 0, sink, 4096)
+
+ self.assertEqual(reader.start(source), serd.Status.SUCCESS)
+ self.assertEqual(captured_events, [])
+ self.assertEqual(reader.read_document(), serd.Status.SUCCESS)
+ self.assertEqual(reader.finish(), serd.Status.SUCCESS)
+ self.assertEqual(captured_events, self.events)
+
+
+class LoadTests(unittest.TestCase):
+ def setUp(self):
+ self.world = serd.World()
+ self.temp_dir = tempfile.mkdtemp()
+ self.ttl_path = os.path.join(self.temp_dir, "input.ttl")
+ self.ttl_document = r"""@prefix eg: <http://example.org/> .
+@base <http://example.org/base> .
+eg:s eg:p1 eg:o1 ;
+ eg:p2 eg:o2 .
+"""
+
+ with open(self.ttl_path, "w") as f:
+ f.write(self.ttl_document)
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir)
+
+ def testLoad(self):
+ s = serd.uri("http://example.org/s")
+ p1 = serd.uri("http://example.org/p1")
+ p2 = serd.uri("http://example.org/p2")
+ o1 = serd.uri("http://example.org/o1")
+ o2 = serd.uri("http://example.org/o2")
+
+ model = self.world.load(self.ttl_path)
+
+ self.assertEqual(
+ [statement for statement in model],
+ [
+ serd.Statement(s, p1, o1),
+ serd.Statement(s, p2, o2),
+ ],
+ )
+
+ def testLoadString(self):
+ s = serd.uri("http://example.org/s")
+ p1 = serd.uri("http://example.org/p1")
+ p2 = serd.uri("http://example.org/p2")
+ o1 = serd.uri("http://example.org/o1")
+ o2 = serd.uri("http://example.org/o2")
+
+ model = self.world.loads(self.ttl_document)
+
+ self.assertEqual(
+ [statement for statement in model],
+ [
+ serd.Statement(s, p1, o1),
+ serd.Statement(s, p2, o2),
+ ],
+ )
+
+
+class DumpTests(unittest.TestCase):
+ def setUp(self):
+ self.world = serd.World()
+ self.temp_dir = tempfile.mkdtemp()
+ self.ttl_path = os.path.join(self.temp_dir, "output.ttl")
+
+ self.ttl_document = r"""<http://example.org/s>
+ <http://example.org/p> <http://example.org/o1> ,
+ <http://example.org/o2> .
+"""
+
+ self.s = serd.uri("http://example.org/s")
+ self.p = serd.uri("http://example.org/p")
+ self.o = serd.uri("http://example.org/o")
+ self.o1 = serd.uri("http://example.org/o1")
+ self.o2 = serd.uri("http://example.org/o2")
+ self.g = serd.uri("http://example.org/g")
+ self.x = serd.uri("http://example.org/x")
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir)
+
+ def testDumpFile(self):
+ s, p, o, o1, o2 = self.s, self.p, self.o, self.o1, self.o2
+ g, x = self.g, self.x
+
+ flags = serd.ModelFlags.INDEX_SPO
+ model = serd.Model(self.world, flags)
+
+ model.insert((s, p, o1))
+ model.insert((s, p, o2))
+
+ self.world.dump(model, self.ttl_path)
+
+ with open(self.ttl_path, "r") as output:
+ self.assertEqual(output.read(), self.ttl_document)
+
+ def testDumpString(self):
+ s, p, o, o1, o2 = self.s, self.p, self.o, self.o1, self.o2
+ g, x = self.g, self.x
+
+ flags = serd.ModelFlags.INDEX_SPO
+ model = serd.Model(self.world, flags)
+
+ model.insert((s, p, o1))
+ model.insert((s, p, o2))
+
+ self.assertEqual(self.world.dumps(model), self.ttl_document)
generated by cgit v1.2.1 (git 2.18.0) at 2024-05-14 04:52:00 +0000