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
102
103
104
105
106
107
108
109
110
111
112
113
114
|
#!/usr/bin/python
#
# Copyright (C) 2010, Nokia <ivan.frade@nokia.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.
#
"""
Send concurrent inserts and queries to the daemon to check the concurrency.
"""
import sys
import os
import dbus
import unittest
import time
import random
import commands
import signal
from gi.repository import GObject
from dbus.mainloop.glib import DBusGMainLoop
from common.utils import configuration as cfg
import unittest as ut
from common.utils.storetest import CommonTrackerStoreTest as CommonTrackerStoreTest
AMOUNT_OF_TEST_INSTANCES = 100
AMOUNT_OF_QUERIES = 10
class TestConcurrentQuery (CommonTrackerStoreTest):
"""
Send a bunch of queries to the daemon asynchronously, to test the queue
holding those queries
"""
def setUp(self):
super(TestConcurrentQuery, self).setUp()
self.main_loop = GObject.MainLoop()
self.mock_data_insert()
self.finish_counter = 0
def mock_data_insert(self):
query = "INSERT {\n"
for i in range(0, AMOUNT_OF_TEST_INSTANCES):
query += "<test-09:instance-%d> a nco:PersonContact ; nco:fullname 'moe %d'.\n" % (
i, i)
query += "}"
self.tracker.update(query)
super(TestConcurrentQuery, self).tearDown()
def mock_data_delete(self):
query = "DELETE {\n"
for i in range(0, AMOUNT_OF_TEST_INSTANCES):
query += "<test-09:instance-%d> a rdfs:Resource.\n" % (i)
query += "}"
self.tracker.update(query)
query = "DELETE {\n"
for i in range(0, AMOUNT_OF_QUERIES):
query += "<test-09:picture-%d> a rdfs:Resource.\n" % (i)
query += "}"
self.tracker.update(query)
def test_async_queries(self):
QUERY = "SELECT ?u WHERE { ?u a nco:PersonContact. FILTER regex (?u, 'test-09:ins')}"
UPDATE = "INSERT { <test-09:picture-%d> a nmm:Photo. }"
for i in range(0, AMOUNT_OF_QUERIES):
self.tracker.get_tracker_iface().SparqlQuery(QUERY,
reply_handler=self.reply_cb,
error_handler=self.error_handler)
self.tracker.get_tracker_iface().SparqlUpdate(UPDATE % (i),
reply_handler=self.update_cb,
error_handler=self.error_handler)
# Safeguard of 50 seconds. The last reply should quit the loop
GObject.timeout_add_seconds(60, self.timeout_cb)
self.main_loop.run()
def reply_cb(self, results):
self.finish_counter += 1
self.assertEquals(len(results), AMOUNT_OF_TEST_INSTANCES)
if (self.finish_counter >= AMOUNT_OF_QUERIES):
self.timeout_cb()
def update_cb(self):
self.assertTrue(True)
def error_handler(self):
print "ERROR in DBus call"
self.assertTrue(False)
def timeout_cb(self):
self.mock_data_delete()
self.main_loop.quit()
return False
if __name__ == "__main__":
ut.main()
|
