← Back to branch summary

~jplacerda/zeitgeist/slots

« back to all changes in this revision

Viewing changes to test/remote-test.py

  • Committer: Markus Korn
  • Date: 2010-04-25 09:21:29 UTC
  • mfrom: (1444.1.2)
  • mto: This revision was merged to the branch mainline in revision 1447.
  • Revision ID: thekorn@gmx.de-20100425092129-jm9v28jihkq5htv1
merged changes from lp:zeitgeist

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/remote-test.py
2
2
 
3
3
import unittest
4
4
import os
5
 
import time
6
5
import sys
7
 
import signal
8
 
from subprocess import Popen, PIPE
9
6
 
10
7
# DBus setup
11
8
import gobject
17
14
from zeitgeist.client import ZeitgeistDBusInterface, ZeitgeistClient
18
15
from zeitgeist.datamodel import (Event, Subject, Interpretation, Manifestation,
19
16
        TimeRange, StorageState)
 
17
 
20
18
import testutils
21
 
 
22
19
from testutils import parse_events
23
20
 
24
21
class ZeitgeistRemoteAPITest(testutils.RemoteTestCase):
162
159
                
163
160
                self.assertEquals(2, len(result))
164
161
        
 
162
        def testMonitorDeleteNonExistingEvent(self):
 
163
                result = []
 
164
                mainloop = gobject.MainLoop()
 
165
                events = parse_events("test/data/five_events.js")
 
166
                
 
167
                def timeout():
 
168
                        # We want this timeout - we should not get informed
 
169
                        # about deletions of non-existing events
 
170
                        mainloop.quit()
 
171
                        return False
 
172
 
 
173
                def notify_insert_handler(time_range, events):
 
174
                        event_ids = map(lambda ev : ev.id, events)
 
175
                        self.client.delete_events([9999999])
 
176
                
 
177
                def notify_delete_handler(time_range, event_ids):
 
178
                        mainloop.quit()
 
179
                        self.fail("Notified about deletion of non-existing events %s", events)
 
180
                        
 
181
                        
 
182
                self.client.install_monitor(TimeRange(125, 145), [],
 
183
                        notify_insert_handler, notify_delete_handler)
 
184
                
 
185
                gobject.timeout_add_seconds(5, timeout)
 
186
                self.client.insert_events(events)
 
187
                mainloop.run()
 
188
        
 
189
        def testTwoMonitorsDeleteEvents(self):
 
190
                result1 = []
 
191
                result2 = []
 
192
                mainloop = gobject.MainLoop()
 
193
                events = parse_events("test/data/five_events.js")
 
194
                
 
195
                def timeout ():
 
196
                        mainloop.quit()
 
197
                        self.fail("Test case timed out")
 
198
                        return False
 
199
                
 
200
                def check_ok():
 
201
                        if len(result1) == 2 and len(result2) == 2:
 
202
                                mainloop.quit()
 
203
 
 
204
                def notify_insert_handler1(time_range, events):
 
205
                        event_ids = map(lambda ev : ev.id, events)
 
206
                        self.client.delete_events(event_ids)
 
207
                
 
208
                def notify_delete_handler1(time_range, event_ids):
 
209
                        result1.extend(event_ids)
 
210
                        check_ok()
 
211
                
 
212
                def notify_delete_handler2(time_range, event_ids):
 
213
                        result2.extend(event_ids)
 
214
                        check_ok()
 
215
                        
 
216
                self.client.install_monitor(TimeRange(125, 145), [],
 
217
                        notify_insert_handler1, notify_delete_handler1)
 
218
                
 
219
                self.client.install_monitor(TimeRange(125, 145), [],
 
220
                        lambda x, y: x, notify_delete_handler2)
 
221
                
 
222
                self.client.insert_events(events)
 
223
                gobject.timeout_add_seconds (5, timeout)
 
224
                mainloop.run()
 
225
                
 
226
                self.assertEquals(2, len(result1))
 
227
                self.assertEquals(2, len(result1))
 
228
 
165
229
        def testMonitorInstallRemoval(self):
166
230
                result = []
167
231
                mainloop = gobject.MainLoop()
224
288
                
225
289
                def callback(uris):
226
290
                        mainloop.quit()
227
 
                        self.assertEquals(uris, ["i2", "i1", "i5", "i3"])
 
291
                        self.assertEquals(uris, ["i2", "i1", "i3", "i5"])
228
292
                
229
293
                result = self.client.find_related_uris_for_uris(["i4"], callback, num_events=4, result_type=0)
230
294
                mainloop.run()
241
305
                
242
306
                result = self.client.find_events_for_values(callback, actor="firefox", num_events=1)
243
307
                mainloop.run()
244
 
                
 
308
 
 
309
        def testDataSourcesRegistry(self):
 
310
                """ Ensure that the DataSourceRegistry extension is there. If we'd want
 
311
                    to do any actual value checking we need to change testutils.py to
 
312
                    use a ZEITGEIST_DATA_PATH other than ~/.local/share. """
 
313
                iface = ZeitgeistDBusInterface()
 
314
                registry = iface.get_extension("DataSourceRegistry", "data_source_registry")
 
315
                registry.GetDataSources()
245
316
        
246
317
if __name__ == "__main__":
247
318
        unittest.main()