summaryrefslogtreecommitdiff
path: root/test/store_performance.py
blob: e9de8af8832482de7db1bfe7df1af1050bc8cef4 (plain)
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
import unittest
import os
import gc
import itertools
from time import time
from random import random

from tempfile import mkdtemp
import shutil

from rdflib.term import URIRef
from rdflib.graph import Graph


def random_uri():
    return URIRef("%s" % random())


class StoreTestCase(unittest.TestCase):
    """
    Test case for testing store performance... probably should be
    something other than a unit test... but for now we'll add it as a
    unit test.
    """
    store = 'default'
    tmppath = None
    configString = os.environ.get("DBURI", "dburi")

    def setUp(self):
        self.gcold = gc.isenabled()
        gc.collect()
        gc.disable()
        self.graph = Graph(store=self.store)
        if self.store == "MySQL":
            # from test.mysql import configString
            from rdflib.store.MySQL import MySQL
            path = self.configString
            MySQL().destroy(path)
        else:
            self.tmppath = mkdtemp()
        self.graph.open(self.tmppath, create=True)
        self.input = input = Graph()
        input.parse("http://eikeon.com")

    def tearDown(self):
        self.graph.close()
        if self.gcold:
            gc.enable()
        del self.graph
        shutil.rmtree(self.tmppath)

    def testTime(self):
        number = 1
        print self.store
        print "input:",
        for i in itertools.repeat(None, number):
            self._testInput()
        print "random:",
        for i in itertools.repeat(None, number):
            self._testRandom()
        print "."

    def _testRandom(self):
        number = len(self.input)
        store = self.graph

        def add_random():
            s = random_uri()
            p = random_uri()
            o = random_uri()
            store.add((s, p, o))

        it = itertools.repeat(None, number)
        t0 = time()
        for _i in it:
            add_random()
        t1 = time()
        print "%.3g" % (t1 - t0),

    def _testInput(self):
        number = 1
        store = self.graph

        def add_from_input():
            for t in self.input:
                store.add(t)

        it = itertools.repeat(None, number)
        t0 = time()
        for _i in it:
            add_from_input()
        t1 = time()
        print "%.3g" % (t1 - t0),


class MemoryStoreTestCase(StoreTestCase):
    store = "IOMemory"


if __name__ == '__main__':
    unittest.main()