« back to all changes in this revision
Viewing changes to test/dbus/engine-test.py
- browse files at revision 1.1.13
- compare with another revision
- download diff
- download tarball
- view history from revision 1.1.13
- Committer: Package Import Robot
- Author(s): Didier Roche
- Date: 2011-11-15 11:15:56 UTC
- mto: (6.2.2 experimental) (1.3.1)
- mto: This revision was merged to the branch mainline in revision 19.
- Revision ID: package-import@ubuntu.com-20111115111556-so7cmhfbqongw7hf
Tags: upstream-0.8.99~alpha1
Import upstream version 0.8.99~alpha1
added
removed
test/dbus/engine-test.py
1
#! /usr/bin/python
2
# -.- coding: utf-8 -.-
3
4
# remote-test.py
5
#
6
# Copyright © 2009-2011 Seif Lotfy <seif@lotfy.com>
7
# Copyright © 2009-2011 Siegfried-Angel Gevatter Pujals <siegfried@gevatter.com>
8
# Copyright © 2009-2011 Mikkel Kamstrup Erlandsen <mikkel.kamstrup@gmail.com>
9
# Copyright © 2009-2011 Markus Korn <thekorn@gmx.de>
10
# Copyright © 2011 Collabora Ltd.
11
# By Siegfried-Angel Gevatter Pujals <siegfried@gevatter.com>
12
# By Seif Lotfy <seif@lotfy.com>
13
#
14
# This program is free software: you can redistribute it and/or modify
15
# it under the terms of the GNU Lesser General Public License as published by
16
# the Free Software Foundation, either version 2.1 of the License, or
17
# (at your option) any later version.
18
#
19
# This program is distributed in the hope that it will be useful,
20
# but WITHOUT ANY WARRANTY; without even the implied warranty of
21
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
22
# GNU Lesser General Public License for more details.
23
#
24
# You should have received a copy of the GNU Lesser General Public License
25
# along with this program. If not, see <http://www.gnu.org/licenses/>.
26
27
import unittest
28
import os
29
import sys
30
import logging
31
import signal
32
import time
33
import tempfile
34
import shutil
35
import pickle
36
from subprocess import Popen, PIPE
37
38
# DBus setup
39
import gobject
40
from dbus.mainloop.glib import DBusGMainLoop
41
DBusGMainLoop(set_as_default=True)
42
from dbus.exceptions import DBusException
43
44
from zeitgeist.datamodel import (Event, Subject, Interpretation, Manifestation,
45
TimeRange, StorageState, DataSource, NULL_EVENT, ResultType)
46
47
import testutils
48
from testutils import parse_events, import_events
49
50
TEST_ACTOR = "/usr/share/applications/gnome-about.desktop"
51
52
# FIXME: move this to a .js file
53
test_event_1 = None
54
def create_test_event_1():
55
ev = Event()
56
ev.timestamp = 0
57
ev.interpretation = Manifestation.USER_ACTIVITY
58
ev.manifestation = Interpretation.CREATE_EVENT
59
ev.actor = TEST_ACTOR
60
subj = Subject()
61
subj.uri = u"test://mytest"
62
subj.manifestation = "lala"
63
subj.interpretation = "tinky winky"
64
subj.origin = "test://"
65
subj.mimetype = "YOMAMA"
66
subj.text = "SUCKS"
67
subj.storage = "MyStorage"
68
subj.current_uri = u"test://mytest"
69
70
ev.append_subject(subj)
71
return ev
72
73
74
class ZeitgeistEngineTest(testutils.RemoteTestCase):
75
76
def testSingleInsertGet(self):
77
test_event_1 = create_test_event_1()
78
# Insert item and event
79
ids = self.insertEventsAndWait([test_event_1])
80
self.assertEquals(1, len(ids))
81
82
result = self.getEventsAndWait(ids)
83
resulting_event = result.pop()
84
self.assertEquals(len(resulting_event), len(test_event_1))
85
86
# fixing id, the initial event does not have any id set
87
test_event_1[0][0] = ids[0]
88
resulting_event[2] = ""
89
90
self.assertEqual(resulting_event, test_event_1)
91
92
def testInsertGetWithoutTimestamp(self):
93
# We test two things, that Event creates a default timestamp
94
# and that the engine provides one for us if don't do our selves
95
96
subj = Subject.new_for_values(interpretation="foo://interp",
97
manifestation="foo://manif",
98
uri="nowhere")
99
ev = Event.new_for_values(interpretation="foo://bar",
100
manifestation="foo://quiz",
101
actor="actor://myself",
102
subjects=[subj])
103
104
# Assert that timestamp is set
105
self.assertTrue(ev.timestamp)
106
107
# Clear the timestamp and insert event
108
ev.timestamp = ""
109
ids = self.insertEventsAndWait([ev])
110
result = self.getEventsAndWait(ids)
111
112
self.assertEquals(1, len(result))
113
resulting_event = Event(result.pop())
114
self.assertEquals("foo://bar", resulting_event.interpretation)
115
self.assertTrue(resulting_event.timestamp) # We should have a timestamp again
116
117
def testDuplicateEventInsertion(self):
118
self.testSingleInsertGet()
119
120
# Inserting the same event again should be ok, but not
121
# cause duplicates
122
self.testSingleInsertGet()
123
124
# Find all events, and make sure that this is exactly one event
125
result = self.findEventIdsAndWait([])
126
self.assertEquals(1, len(result))
127
self.assertEquals(1, result[0]) # The single event must have id 1
128
129
def testDeleteSingle(self):
130
self.testSingleInsertGet()
131
self.deleteEventsAndWait([1])
132
result = self.getEventsAndWait([1])
133
self.assertEquals(0, len(filter(None, result)))
134
135
def testIllegalPredefinedEventId(self):
136
event = Event()
137
event[0][0] = str(23) # This is illegal, we assert the erro later
138
event.timestamp = 0
139
event.interpretation = Manifestation.USER_ACTIVITY
140
event.manifestation = Interpretation.CREATE_EVENT
141
event.actor = "/usr/share/applications/gnome-about.desktop"
142
143
subject = Subject()
144
subject.uri = "file:///tmp/file.txt"
145
subject.manifestation = Manifestation.FILE_DATA_OBJECT
146
subject.interpretation = Interpretation.DOCUMENT
147
subject.origin = "test://"
148
subject.mimetype = "text/plain"
149
subject.text = "This subject has no text"
150
subject.storage = "368c991f-8b59-4018-8130-3ce0ec944157" # UUID of home partition
151
152
event.append_subject(subject)
153
154
# Insert item and event
155
ids = self.insertEventsAndWait([event,])
156
self.assertEquals(len(ids), 1)
157
# event is not inserted, id == 0 means error
158
self.assertEquals(ids[0], 0)
159
# check if really not events were inserted
160
ids = self.findEventIdsAndWait([])
161
self.assertEquals(len(ids), 0)
162
163
def testGetNonExisting(self):
164
events = self.getEventsAndWait([23,45,65])
165
self.assertEquals(3, len(events))
166
for ev in events : self.assertEquals(None, ev)
167
168
def testGetDuplicateEventIds(self):
169
ids = import_events("test/data/five_events.js", self)
170
self.assertEquals(5, len(ids))
171
172
events = self.getEventsAndWait([1, 1])
173
self.assertEqual(2, len(events))
174
self.assertEqual(2, len(filter(None, events))) #FIXME:FAILS HERE
175
self.assertTrue(events[0].id == events[1].id == 1)
176
177
def testFindEventsId(self):
178
test_event_1 = create_test_event_1()
179
self.testSingleInsertGet()
180
result = self.findEventIdsAndWait([])
181
self.assertEquals(1, len(result))
182
test_event_1[0][0] = 1
183
self.assertEqual(result[0], test_event_1.id)
184
185
def testFindNothing(self):
186
result = self.findEventIdsAndWait([])
187
self.assertEquals(0, len(result))
188
189
def testFindNothingBackwards(self):
190
result = self.findEventIdsAndWait([], timerange=(1000000,1))
191
self.assertEquals(0, len(result))
192
193
def testFindFilteredByEventButNotSubject(self):
194
# revision rainct@ubuntu.com-20091128164327-j8ez3fsifd1gygkr (1185)
195
# Fix _build_templates so that it works when the Subject is empty.
196
self.testSingleInsertGet()
197
result = self.findEventIdsAndWait([Event.new_for_values(interpretation=Interpretation.LEAVE_EVENT)])
198
self.assertEquals(0, len(result))
199
200
def testFindFive(self):
201
import_events("test/data/five_events.js", self)
202
result = self.findEventIdsAndWait([])
203
self.assertEquals(5, len(result))
204
205
def testFindFiveWithStorageState(self):
206
import_events("test/data/five_events.js", self)
207
# The event's storage is unknown, so we get them back always.
208
result = self.findEventIdsAndWait([], storage_state = 1)
209
self.assertEquals(5, len(result))
210
result = self.findEventIdsAndWait([], storage_state = 0)
211
self.assertEquals(5, len(result))
212
213
def testFindWithNonExistantActor(self):
214
# Bug 496109: filtering by timerange and a non-existing actor gave an
215
# incorrect result.
216
import_events("test/data/twenty_events.js", self)
217
# The event's storage is unknown, so we get them back always.
218
result = self.findEventIdsAndWait([Event.new_for_values(actor="fake://foobar")])
219
self.assertEquals(0, len(result))
220
221
def testFindWithSubjectText(self):
222
import_events("test/data/five_events.js", self)
223
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='this is not real')])
224
self.assertEquals(0, len(result))
225
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='some text')])
226
self.assertEquals(1, len(result))
227
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='this *')])
228
self.assertEquals(0, len(result)) #We don't support wildcards for text
229
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='this item *')])
230
self.assertEquals(1, len(result))
231
232
def testSortFindByTimeAsc(self):
233
import_events("test/data/twenty_events.js", self)
234
result = self.findEventIdsAndWait([], num_events = 2, result_type = ResultType.LeastRecentEvents)
235
event1 = self.getEventsAndWait([result[0]])[0]
236
event2 = self.getEventsAndWait([result[1]])[0]
237
self.assertEquals(True, event1.timestamp < event2.timestamp)
238
239
def testSortFindByTimeDesc(self):
240
import_events("test/data/twenty_events.js", self)
241
result = self.findEventIdsAndWait([], num_events = 2, result_type = ResultType.MostRecentEvents)
242
event1 = self.getEventsAndWait([result[0]])[0]
243
event2 = self.getEventsAndWait([result[1]])[0]
244
self.assertEquals(True, event1.timestamp > event2.timestamp)
245
246
def testFindWithActor(self):
247
test_event_1 = create_test_event_1()
248
self.testSingleInsertGet()
249
subj = Subject()
250
event_template = Event.new_for_values(actor=TEST_ACTOR, subjects=[subj,])
251
result = self.findEventIdsAndWait([event_template], num_events = 0, result_type = 1)
252
self.assertEquals(1, len(result))
253
test_event_1[0][0] = 1
254
self.assertEqual(result[0], test_event_1.id)
255
256
def testFindWithInterpretation(self):
257
import_events("test/data/five_events.js", self)
258
subj = Subject()
259
event_template = Event.new_for_values(interpretation="stfu:OpenEvent", subjects=[subj])
260
result = self.findEventIdsAndWait([event_template], num_events = 0, result_type = 1)
261
self.assertEquals(2, len(result))
262
events = self.getEventsAndWait(result)
263
for event in events:
264
self.assertEqual(event.interpretation, "stfu:OpenEvent")
265
266
def testFindEventTwoInterpretations(self):
267
import_events("test/data/twenty_events.js", self)
268
result = self.findEventIdsAndWait([
269
Event.new_for_values(interpretation="stfu:OpenEvent"),
270
Event.new_for_values(interpretation="stfu:EvilEvent")],
271
timerange = (102, 117), num_events = 0, result_type = 0)
272
self.assertEquals(15, len(result))
273
274
def testFindWithFakeInterpretation(self):
275
import_events("test/data/twenty_events.js", self)
276
result = self.findEventIdsAndWait([Event.new_for_values(interpretation="this-is-not-an-intrprettin")],
277
num_events = 0, result_type = 0)
278
self.assertEquals(0, len(result))
279
280
def testFindWithManifestation(self):
281
import_events("test/data/five_events.js", self)
282
subj = Subject()
283
event_template = Event.new_for_values(manifestation="stfu:EpicFailActivity", subjects=[subj])
284
285
result = self.findEventIdsAndWait([event_template],
286
num_events = 0, result_type = 1)
287
self.assertEquals(1, len(result))
288
events = self.getEventsAndWait(result)
289
for event in events:
290
self.assertEqual(event.manifestation, "stfu:EpicFailActivity")
291
292
def testFindWithEventOrigin(self):
293
import_events("test/data/twenty_events.js", self)
294
event_template = Event.new_for_values(origin="origin3")
295
result = self.findEventIdsAndWait([event_template],
296
num_events = 0, result_type = 1)
297
events = self.getEventsAndWait(result)
298
299
self.assertTrue(len(events) > 0)
300
self.assertTrue(all(ev.origin == "origin3" for ev in events))
301
302
def testFindWithEventOriginNegatedWildcard(self):
303
import_events("test/data/twenty_events.js", self)
304
event_template = Event.new_for_values(origin="!origin*")
305
result = self.findEventIdsAndWait([event_template],
306
num_events = 0, result_type = 1)
307
events = self.getEventsAndWait(result)
308
309
self.assertTrue(len(events) > 0)
310
self.assertFalse(any(ev.origin.startswith("origin") for ev in events))
311
312
def testFindWithSubjectOrigin(self):
313
import_events("test/data/five_events.js", self)
314
subj = Subject.new_for_values(origin="file:///tmp")
315
event_template = Event.new_for_values(subjects=[subj])
316
result = self.findEventIdsAndWait([event_template], num_events = 0, result_type = 1)
317
events = self.getEventsAndWait(result)
318
for event in events:
319
test = any(subj.origin == "file:///tmp" for subj in event.subjects)
320
self.assertTrue(test)
321
322
def testFindMultipleEvents(self):
323
import_events("test/data/five_events.js", self)
324
subj1 = Subject.new_for_values(uri="file:///home/foo.txt")
325
event_template1 = Event.new_for_values(subjects=[subj1])
326
subj2 = Subject.new_for_values(uri="file:///tmp/foo.txt")
327
event_template2 = Event.new_for_values(subjects=[subj2])
328
result = self.findEventIdsAndWait([event_template1, event_template2], num_events = 0, result_type = 4)
329
self.assertEquals(2, len(result))
330
events = self.getEventsAndWait(result)
331
332
def testGetWithMultipleSubjects(self):
333
subj1 = Subject.new_for_values(uri="file:///tmp/foo.txt")
334
subj2 = Subject.new_for_values(uri="file:///tmp/loo.txt")
335
event_template = Event.new_for_values(subjects=[subj1, subj2])
336
result = self.insertEventsAndWait([event_template])
337
events = self.getEventsAndWait(result)
338
self.assertEquals(2, len(events[0].subjects))
339
self.assertEquals("file:///tmp/foo.txt", events[0].subjects[0].uri)
340
self.assertEquals("file:///tmp/loo.txt", events[0].subjects[1].uri)
341
342
def testFindEventIdsWithMultipleSubjects(self):
343
subj1 = Subject.new_for_values(uri="file:///tmp/foo.txt")
344
subj2 = Subject.new_for_values(uri="file:///tmp/loo.txt")
345
event = Event.new_for_values(subjects=[subj1, subj2])
346
orig_ids = self.insertEventsAndWait([event])
347
result_ids = self.findEventIdsAndWait([Event()], num_events = 0, result_type = 1)
348
self.assertEquals(orig_ids, list(result_ids)) #FIXME: We need subjects of the same event to be merged
349
350
def testFindEventsEventTemplate(self):
351
import_events("test/data/five_events.js", self)
352
subj = Subject.new_for_values(interpretation="stfu:Bee")
353
subj1 = Subject.new_for_values(interpretation="stfu:Bar")
354
event_template = Event.new_for_values(subjects=[subj, subj1])
355
result = self.findEventIdsAndWait(
356
[event_template, ],
357
timerange = (0, 200),
358
num_events = 100,
359
result_type = 0)
360
self.assertEquals(0, len(result)) # no subject with two different
361
# interpretations at the same time
362
subj = Subject.new_for_values(uri="file:///tmp/foo.txt")
363
subj1 = Subject.new_for_values(interpretation="stfu:Image")
364
event_template = Event.new_for_values(subjects=[subj, subj1])
365
result = self.findEventIdsAndWait(
366
[event_template, ],
367
timerange = (0, 200),
368
num_events = 100,
369
result_type = 0)
370
self.assertEquals(1, len(result))
371
372
def testJsonImport(self):
373
import_events("test/data/single_event.js", self)
374
results = self.getEventsAndWait([1])
375
self.assertEquals(1, len(results))
376
ev = results[0]
377
self.assertEquals(1, ev.id)
378
self.assertEquals("123", ev.timestamp)
379
self.assertEquals("stfu:OpenEvent", ev.interpretation)
380
self.assertEquals("stfu:UserActivity", ev.manifestation)
381
self.assertEquals("firefox", ev.actor)
382
self.assertEquals(1, len(ev.subjects))
383
384
subj = ev.subjects[0]
385
self.assertEquals("file:///tmp/foo.txt", subj.uri)
386
self.assertEquals("stfu:Document", subj.interpretation)
387
self.assertEquals("stfu:File", subj.manifestation)
388
self.assertEquals("text/plain", subj.mimetype)
389
self.assertEquals("this item has no text... rly!", subj.text)
390
self.assertEquals("368c991f-8b59-4018-8130-3ce0ec944157", subj.storage)
391
392
def testInsertSubjectOptionalAttributes(self):
393
ev = Event.new_for_values(
394
timestamp=123,
395
interpretation=Interpretation.ACCESS_EVENT,
396
manifestation=Manifestation.USER_ACTIVITY,
397
actor="Freak Mamma"
398
)
399
subj = Subject.new_for_values(
400
uri="void://foobar",
401
interpretation=Interpretation.DOCUMENT,
402
manifestation=Manifestation.FILE_DATA_OBJECT,
403
)
404
ev.append_subject(subj)
405
406
ids = self.insertEventsAndWait([ev,])
407
result = self.getEventsAndWait(ids)
408
self.assertEquals(len(ids), len(result))
409
410
def testEventWithoutSubject(self):
411
ev = Event.new_for_values(timestamp=123,
412
interpretation=Interpretation.ACCESS_EVENT,
413
manifestation=Manifestation.USER_ACTIVITY,
414
actor="Freak Mamma")
415
ids = self.insertEventsAndWait([ev,])
416
self.assertEquals(len(ids), 1)
417
# event is not inserted, id == 0 means error
418
self.assertEquals(ids[0], 0)
419
# check if really not events were inserted
420
ids = self.findEventIdsAndWait([ev],
421
num_events = 0,
422
result_type = ResultType.MostRecentEvents)
423
self.assertEquals(len(ids), 0)
424
425
def testUnicodeEventInsert(self):
426
# Insert and get a unicode event
427
ids = import_events("test/data/unicode_event.js", self)
428
self.assertEquals(len(ids), 1)
429
result = self.getEventsAndWait(ids)
430
self.assertEquals(1, len(result))
431
event = result[0]
432
self.assertEquals(1, len(event.subjects))
433
self.assertEquals(u"hällö, I'm gürmen - åge drikker øl - ☠", event.subjects[0].text)
434
self.assertEquals(u"http://live.gnome.org/☠", event.subjects[0].uri)
435
436
# update the event we got from the DB's timestamp and insert
437
# it again, we want to to test some ping-pong back and forth
438
event[0][Event.Id] = "" #FIXME: It used to be None but it did not work until i passed an empty string
439
event.timestamp = str(243)
440
ids = self.insertEventsAndWait([event])
441
result = self.getEventsAndWait(ids)
442
self.assertEquals(1, len(result))
443
event = result[0]
444
self.assertEquals(1, len(event.subjects))
445
self.assertEquals(u"hällö, I'm gürmen - åge drikker øl - ☠", event.subjects[0].text)
446
self.assertEquals(u"http://live.gnome.org/☠", event.subjects[0].uri)
447
448
# try and find a unicode event, we use unicode and not
449
# inconsequently on deliberation
450
subj = Subject.new_for_values(text="hällö, I'm gürmen - åge drikker øl - ☠",
451
origin="file:///åges_øl í",
452
uri=u"http://live.gnome.org/☠")
453
event_template = Event.new_for_values(subjects=[subj,])
454
455
456
result = self.findEventIdsAndWait([event_template],
457
timerange = (0,200),
458
num_events = 100,
459
result_type = 0)
460
self.assertEquals(len(result), 1)
461
462
def testEventWithBinaryPayload(self):
463
ev = Event()
464
subject = Subject()
465
ev.actor = "application:///firefox.desktop"
466
ev.manifestation = Manifestation.USER_ACTIVITY
467
ev.interpretation = Interpretation.ACCESS_EVENT
468
subject.uri = "http://www.google.com"
469
subject.interpretation = Interpretation #InterPretation.UNKNOWN
470
subject.manifestation = Manifestation #Manifestation.WEB_HISTORY
471
subject.text = ""
472
subject.mimetype = "text/html"
473
subject.origin = ""
474
subject.storage = ""
475
ev.subjects.append(subject)
476
477
sampleString = """
478
<Content name="Telepathy" class="Text">
479
<header>johnsmith@foo.bar</header>
480
<body>
481
John: Here is a talking point
482
You: Ok that looks fine
483
</body>
484
<launcher command="{application} johnsmith@foo.bar"/>
485
</Content>"""
486
487
ev.payload = sampleString.encode("UTF-8")
488
ids = self.insertEventsAndWait([ev])
489
_ev = self.getEventsAndWait(ids)[0]
490
_ev.payload = "".join(map(str, _ev.payload)).decode('utf-8')
491
self.assertEquals(ev.payload, _ev.payload)
492
493
# Note: engine.insert_events() sets the id of the Event objects
494
ev[0][0] = _ev.id
495
self.assertEquals(ev.payload, _ev.payload)
496
497
def testQueryByParent (self):
498
ev = Event.new_for_values(subject_interpretation=Interpretation.AUDIO)
499
_ids = self.insertEventsAndWait([ev])
500
501
tmpl = Event.new_for_values(subject_interpretation=Interpretation.MEDIA)
502
ids = self.findEventIdsAndWait([tmpl],
503
num_events = 10,
504
result_type = ResultType.MostRecentEvents)
505
506
self.assertEquals(1, len(ids))
507
self.assertEquals(_ids, list(ids))
508
509
def testNegation(self):
510
import_events("test/data/five_events.js", self)
511
512
template = Event.new_for_values(
513
interpretation = "!stfu:OpenEvent"
514
)
515
ids = self.findEventIdsAndWait([template,],
516
num_events = 10,
517
result_type = ResultType.MostRecentEvents
518
)
519
self.assertEquals(3, len(ids))
520
521
template = Event.new_for_values(
522
manifestation = "!stfu:YourActivity"
523
)
524
ids = self.findEventIdsAndWait([template,],
525
num_events = 10,
526
result_type = ResultType.MostRecentEvents
527
)
528
self.assertEquals(4, len(ids))
529
530
template = Event.new_for_values(
531
actor = "!firefox"
532
)
533
ids = self.findEventIdsAndWait([template,],
534
num_events = 10,
535
result_type = ResultType.MostRecentEvents
536
)
537
self.assertEquals(2, len(ids))
538
539
template = Event.new_for_values(
540
subject_uri = "!file:///tmp/foo.txt"
541
)
542
ids = self.findEventIdsAndWait([template,],
543
num_events = 10,
544
result_type = ResultType.MostRecentEvents
545
)
546
self.assertEquals(3, len(ids))
547
548
template = Event.new_for_values(
549
subject_interpretation = "!stfu:Document"
550
)
551
ids = self.findEventIdsAndWait([template,],
552
num_events = 10,
553
result_type = ResultType.MostRecentEvents
554
)
555
self.assertEquals(4, len(ids))
556
557
template = Event.new_for_values(
558
subject_manifestation = "!stfu:File"
559
)
560
ids = self.findEventIdsAndWait([template,],
561
num_events = 10,
562
result_type = ResultType.MostRecentEvents
563
)
564
self.assertEquals(1, len(ids))
565
566
template = Event.new_for_values(
567
subject_origin = "!file:///tmp"
568
)
569
ids = self.findEventIdsAndWait([template,],
570
num_events = 10,
571
result_type = ResultType.MostRecentEvents
572
)
573
self.assertEquals(1, len(ids))
574
575
template = Event.new_for_values(
576
subject_mimetype = "!text/plain"
577
)
578
ids = self.findEventIdsAndWait([template,],
579
num_events = 10,
580
result_type = ResultType.MostRecentEvents
581
)
582
self.assertEquals(2, len(ids))
583
584
# the next two fields do not support negation, '!' is treated as
585
# content
586
587
template = Event.new_for_values(
588
subject_text = "!boo"
589
)
590
ids = self.findEventIdsAndWait([template,],
591
num_events = 10,
592
result_type = ResultType.MostRecentEvents
593
)
594
self.assertEquals(0, len(ids))
595
596
# searching by subject_storage is not working
597
#~ template = Event.new_for_values(
598
#~ subject_storage = "!boo"
599
#~ )
600
#~ ids = self.engine.find_eventids(TimeRange.always(),
601
#~ [template,], StorageState.Any, 10, ResultType.MostRecentEvents
602
#~ )
603
#~ self.assertEquals(0, len(ids))
604
605
def testNegationCombination(self):
606
import_events("test/data/five_events.js", self)
607
608
template = Event.new_for_values(
609
interpretation = "!stfu:OpenEvent",
610
actor = "!firefox"
611
)
612
ids = self.findEventIdsAndWait([template,],
613
num_events = 10,
614
result_type = ResultType.MostRecentEvents
615
)
616
self.assertEquals(2, len(ids))
617
618
template = Event.new_for_values(
619
interpretation = "!stfu:OpenEvent",
620
manifestation = "!stfu:YourActivity"
621
)
622
ids = self.findEventIdsAndWait([template,],
623
num_events = 10,
624
result_type = ResultType.MostRecentEvents
625
)
626
self.assertEquals(3, len(ids))
627
628
def testFindStorageNotExistant(self):
629
events = [
630
Event.new_for_values(timestamp=1000, subject_storage="sometext"),
631
Event.new_for_values(timestamp=2000, subject_storage="anotherplace")
632
]
633
ids_in = self.insertEventsAndWait(events)
634
template = Event.new_for_values(subject_storage="xxx")
635
results = self.findEventIdsAndWait([template,],
636
num_events = 10,
637
result_type = ResultType.MostRecentEvents
638
)
639
self.assertEquals(0, len(results))
640
641
def testFindStorage(self):
642
events = [
643
Event.new_for_values(timestamp=1000, subject_storage="sometext"),
644
Event.new_for_values(timestamp=2000, subject_storage="anotherplace")
645
]
646
ids_in = self.insertEventsAndWait(events)
647
template = Event.new_for_values(subject_storage="sometext")
648
results = self.findEventIdsAndWait([template,],
649
num_events = 10,
650
result_type = ResultType.MostRecentEvents
651
)
652
self.assertEquals(1, len(results))
653
654
def testWildcard(self):
655
import_events("test/data/five_events.js", self)
656
657
template = Event.new_for_values(
658
actor = "ge*"
659
)
660
ids = self.findEventIdsAndWait([template,],
661
num_events = 10,
662
result_type = ResultType.MostRecentEvents
663
)
664
self.assertEquals(2, len(ids))
665
666
template = Event.new_for_values(
667
actor = "!ge*"
668
)
669
ids = self.findEventIdsAndWait([template,],
670
num_events = 10,
671
result_type = ResultType.MostRecentEvents
672
)
673
self.assertEquals(3, len(ids))
674
675
template = Event.new_for_values(
676
subject_mimetype = "text/*"
677
)
678
ids = self.findEventIdsAndWait([template,],
679
num_events = 10,
680
result_type = ResultType.MostRecentEvents
681
)
682
self.assertEquals(3, len(ids))
683
684
template = Event.new_for_values(
685
subject_uri = "http://*"
686
)
687
ids = self.findEventIdsAndWait([template,],
688
num_events = 10,
689
result_type = ResultType.MostRecentEvents
690
)
691
self.assertEquals(1, len(ids))
692
693
template = Event.new_for_values(
694
subject_current_uri = "http://*"
695
)
696
ids = self.findEventIdsAndWait([template,],
697
num_events = 10,
698
result_type = ResultType.MostRecentEvents
699
)
700
self.assertEquals(1, len(ids))
701
702
template = Event.new_for_values(
703
subject_origin = "file://*"
704
)
705
ids = self.findEventIdsAndWait([template,],
706
num_events = 10,
707
result_type = ResultType.MostRecentEvents
708
)
709
self.assertEquals(4, len(ids))
710
711
class ResultTypeTest(testutils.RemoteTestCase):
712
713
def testResultTypesMostRecentEvents(self):
714
import_events("test/data/five_events.js", self)
715
716
# MostRecentEvents - new -> old
717
ids = self.findEventIdsAndWait([],
718
num_events = 0,
719
result_type = ResultType.MostRecentEvents
720
)
721
events = self.getEventsAndWait(ids)
722
sorted_event_ids = [
723
event.id for event in sorted(events,
724
cmp=lambda x, y: cmp(int(x.timestamp), int(y.timestamp)),
725
reverse=True
726
)
727
]
728
self.assertEquals(list(ids), sorted_event_ids)
729
730
def testResultTypesLeastRecentEvents(self):
731
import_events("test/data/five_events.js", self)
732
733
# LeastRecentEvents - old -> new
734
ids = self.findEventIdsAndWait([],
735
num_events = 0,
736
result_type = ResultType.LeastRecentEvents)
737
events = self.getEventsAndWait(ids)
738
sorted_event_ids = [
739
event.id for event in sorted(events,
740
cmp=lambda x, y: cmp(int(x.timestamp), int(y.timestamp)))
741
]
742
self.assertEquals(list(ids), sorted_event_ids)
743
744
def testResultTypesMostPopularActor(self):
745
import_events("test/data/twenty_events.js", self)
746
747
ids = self.findEventIdsAndWait([],
748
num_events = 0,
749
result_type = ResultType.MostPopularActor)
750
events = self.getEventsAndWait(ids)
751
self.assertEquals([e[0][4] for e in events], ["firefox", "icedove",
752
"frobnicator"])
753
self.assertEquals([e.timestamp for e in events], ["119", "114", "105"])
754
755
def testResultTypesMostPopularActor2(self):
756
import_events("test/data/twenty_events.js", self)
757
758
ids = self.findEventIdsAndWait([],
759
timerange = (105,107),
760
num_events = 0,
761
result_type = ResultType.MostPopularActor)
762
events = self.getEventsAndWait(ids)
763
self.assertEquals(len(events), 2)
764
self.assertEquals([e[0][4] for e in events], ["firefox", "frobnicator"])
765
self.assertEquals([e.timestamp for e in events], ["107", "105"])
766
767
def testResultTypesLeastPopularActor(self):
768
import_events("test/data/twenty_events.js", self)
769
770
ids = self.findEventIdsAndWait([],
771
num_events = 0,
772
result_type = ResultType.LeastPopularActor)
773
events = self.getEventsAndWait(ids)
774
775
self.assertEquals([e[0][4] for e in events], ["frobnicator", "icedove",
776
"firefox"])
777
self.assertEquals([e.timestamp for e in events], ["105", "114", "119"])
778
779
def testResultTypesLeastPopularActor2(self):
780
import_events("test/data/twenty_events.js", self)
781
782
ids = self.findEventIdsAndWait([],
783
timerange = (105,107),
784
num_events = 0,
785
result_type = ResultType.LeastPopularActor)
786
events = self.getEventsAndWait(ids)
787
788
self.assertEquals(len(events), 2)
789
self.assertEquals([e[0][4] for e in events], ["frobnicator", "firefox"])
790
self.assertEquals([e.timestamp for e in events], ["105", "107"])
791
792
def testResultTypesMostRecentSubject(self):
793
import_events("test/data/five_events.js", self)
794
795
ids = self.findEventIdsAndWait([],
796
num_events = 0,
797
result_type = ResultType.MostRecentSubjects)
798
events = self.getEventsAndWait(ids)
799
800
self.assertEquals([e.timestamp for e in events],
801
["163", "153", "143", "123"])
802
803
def testResultTypesLeastRecentSubject(self):
804
import_events("test/data/five_events.js", self)
805
806
ids = self.findEventIdsAndWait([],
807
num_events = 0,
808
result_type = ResultType.LeastRecentSubjects)
809
events = self.getEventsAndWait(ids)
810
self.assertEquals([e.timestamp for e in events],
811
["123", "143", "153", "163"])
812
813
def testResultTypesMostPopularSubject(self):
814
import_events("test/data/five_events.js", self)
815
816
ids = self.findEventIdsAndWait([],
817
num_events = 0,
818
result_type = ResultType.MostPopularSubjects)
819
events = self.getEventsAndWait(ids)
820
self.assertEquals([e.timestamp for e in events],
821
["143", "163", "153", "123"])
822
823
def testResultTypesLeastPopularSubject(self):
824
import_events("test/data/five_events.js", self)
825
826
ids = self.findEventIdsAndWait([],
827
num_events = 0,
828
result_type = ResultType.LeastPopularSubjects)
829
events = self.getEventsAndWait(ids)
830
831
self.assertEquals([e.timestamp for e in events],
832
["123", "153", "163", "143"])
833
834
def testResultTypesMostRecentCurrentUri(self):
835
import_events("test/data/five_events.js", self)
836
import_events("test/data/five_events_ext_move.js", self)
837
838
ids = self.findEventIdsAndWait([],
839
num_events = 0,
840
result_type = ResultType.MostRecentCurrentUri)
841
events = self.getEventsAndWait(ids)
842
843
self.assertEquals([e.timestamp for e in events],
844
["200", "153", "123"])
845
846
def testResultTypesLeastRecentCurrentUri(self):
847
import_events("test/data/five_events.js", self)
848
import_events("test/data/five_events_ext_move.js", self)
849
850
ids = self.findEventIdsAndWait([],
851
num_events = 0,
852
result_type = ResultType.LeastRecentCurrentUri)
853
events = self.getEventsAndWait(ids)
854
self.assertEquals([e.timestamp for e in events],
855
["123", "153", "200"])
856
857
def testResultTypesMostPopularCurrentUri(self):
858
import_events("test/data/five_events.js", self)
859
import_events("test/data/five_events_ext_move.js", self)
860
861
ids = self.findEventIdsAndWait([],
862
num_events = 0,
863
result_type = ResultType.MostPopularCurrentUri)
864
events = self.getEventsAndWait(ids)
865
self.assertEquals([e.timestamp for e in events],
866
["200", "123", "153"])
867
868
def testResultTypesLeastPopularCurrentUri(self):
869
import_events("test/data/five_events.js", self)
870
import_events("test/data/five_events_ext_move.js", self)
871
872
ids = self.findEventIdsAndWait([],
873
num_events = 0,
874
result_type = ResultType.LeastPopularCurrentUri)
875
events = self.getEventsAndWait(ids)
876
self.assertEquals([e.timestamp for e in events],
877
["153", "123", "200"]) #Zeitgeist 0.8 does this test wrong.
878
#This is the expected results
879
880
def testResultTypesMostRecentActor(self):
881
import_events("test/data/twenty_events.js", self)
882
883
ids = self.findEventIdsAndWait([],
884
num_events = 0,
885
result_type = ResultType.MostRecentActor)
886
events = self.getEventsAndWait(ids)
887
888
self.assertEquals([e.timestamp for e in events], ["119", "114", "105"])
889
890
def testResultTypesMostRecentActor2(self):
891
import_events("test/data/twenty_events.js", self)
892
893
ids = self.findEventIdsAndWait([],
894
timerange = (105, 107),
895
num_events = 0,
896
result_type = ResultType.MostRecentActor)
897
events = self.getEventsAndWait(ids)
898
899
self.assertEquals([e.timestamp for e in events], ["107", "105"])
900
901
def testResultTypesOldestActorBug641968(self):
902
events = [
903
Event.new_for_values(timestamp=1, actor="boo", subject_uri="tmp/boo"),
904
Event.new_for_values(timestamp=2, actor="boo", subject_uri="home/boo"),
905
Event.new_for_values(timestamp=3, actor="bar", subject_uri="tmp/boo"),
906
Event.new_for_values(timestamp=4, actor="baz", subject_uri="tmp/boo"),
907
]
908
self.insertEventsAndWait(events)
909
910
# Get the least recent actors
911
ids = self.findEventIdsAndWait([],
912
num_events = 0,
913
result_type = ResultType.OldestActor)
914
events = self.getEventsAndWait(ids)
915
self.assertEquals(list(ids), [1, 3, 4])
916
917
# Get the least recent actors for "home/boo"
918
template = Event.new_for_values(subject_uri="home/boo")
919
ids = self.findEventIdsAndWait([template],
920
num_events = 0,
921
result_type = ResultType.OldestActor)
922
self.assertEquals(list(ids), [2])
923
924
# Let's also try the same with MostRecentActor... Although there
925
# should be no problem here.
926
template = Event.new_for_values(subject_uri="home/boo")
927
ids = self.findEventIdsAndWait([template],
928
num_events = 0,
929
result_type = ResultType.OldestActor)
930
self.assertEquals(list(ids), [2])
931
932
def testResultTypesOldestActor(self):
933
import_events("test/data/twenty_events.js", self)
934
935
ids = self.findEventIdsAndWait(
936
[Event.new_for_values(subject_manifestation="stfu:File")],
937
num_events = 0,
938
result_type = ResultType.OldestActor)
939
events = self.getEventsAndWait(ids)
940
self.assertEquals([e.timestamp for e in events], ["100", "101", "105"])
941
942
def testResultTypesLeastRecentActor(self):
943
import_events("test/data/twenty_events.js", self)
944
945
ids = self.findEventIdsAndWait(
946
[Event.new_for_values(subject_manifestation="stfu:File")],
947
num_events = 0,
948
result_type = ResultType.LeastRecentActor)
949
events = self.getEventsAndWait(ids)
950
951
self.assertEquals([e.timestamp for e in events], ['105', '114', '119'])
952
953
def testResultTypesLeastRecentActor2(self):
954
# The same test as before, but this time with fewer events so that
955
# it is actually understandable.
956
events = [
957
Event.new_for_values(timestamp=1, actor="gedit", subject_uri="oldFile"),
958
Event.new_for_values(timestamp=2, actor="banshee", subject_uri="oldMusic"),
959
Event.new_for_values(timestamp=3, actor="banshee", subject_uri="newMusic"),
960
Event.new_for_values(timestamp=4, actor="gedit", subject_uri="newFile"),
961
]
962
self.insertEventsAndWait(events)
963
964
ids = self.findEventIdsAndWait([],
965
num_events = 0,
966
result_type = ResultType.LeastRecentActor)
967
recv_events = self.getEventsAndWait(ids)
968
self.assertEquals([e.timestamp for e in recv_events], ['3', '4'])
969
970
def testResultTypesMostPopularEventOrigin(self):
971
import_events("test/data/twenty_events.js", self)
972
973
ids = self.findEventIdsAndWait([],
974
num_events = 0,
975
result_type = ResultType.MostPopularEventOrigin)
976
events = self.getEventsAndWait(ids)
977
978
self.assertEquals([e[0][5] for e in events],
979
["origin1", "origin3", "origin2"])
980
self.assertEquals([e.timestamp for e in events], ["102", "103", "100"])
981
982
def testResultTypesLeastPopularEventOrigin(self):
983
import_events("test/data/twenty_events.js", self)
984
985
ids = self.findEventIdsAndWait([],
986
num_events = 0,
987
result_type = ResultType.LeastPopularEventOrigin)
988
events = self.getEventsAndWait(ids)
989
990
self.assertEquals([e[0][5] for e in events],
991
["origin2", "origin3", "origin1"])
992
self.assertEquals([e.timestamp for e in events], ["100", "103", "102"])
993
994
def testResultTypesMostRecentEventOrigin(self):
995
import_events("test/data/twenty_events.js", self)
996
997
ids = self.findEventIdsAndWait([],
998
num_events = 0,
999
result_type = ResultType.MostRecentEventOrigin)
1000
events = self.getEventsAndWait(ids)
1001
1002
self.assertEquals([e.timestamp for e in events], ["103", "102", "100"])
1003
1004
def testResultTypesLeastRecentEventOrigin(self):
1005
import_events("test/data/twenty_events.js", self)
1006
1007
ids = self.findEventIdsAndWait([],
1008
num_events = 0,
1009
result_type = ResultType.LeastRecentEventOrigin)
1010
events = self.getEventsAndWait(ids)
1011
1012
self.assertEquals([e.timestamp for e in events], ["100", "102", "103"])
1013
1014
def testResultTypesMostPopularSubjectOrigin(self):
1015
import_events("test/data/twenty_events.js", self)
1016
1017
ids = self.findEventIdsAndWait([],
1018
num_events = 0,
1019
result_type = ResultType.MostPopularOrigin)
1020
events = self.getEventsAndWait(ids)
1021
1022
self.assertEquals([e[1][0][3] for e in events], ["file:///tmp", "file:///home",
1023
"file:///etc"])
1024
self.assertEquals([e.timestamp for e in events], ["116", "118", "119"])
1025
1026
def testResultTypesLeastPopularSubjectOrigin(self):
1027
import_events("test/data/twenty_events.js", self)
1028
1029
ids = self.findEventIdsAndWait([],
1030
num_events = 0,
1031
result_type = ResultType.LeastPopularOrigin)
1032
events = self.getEventsAndWait(ids)
1033
1034
self.assertEquals([e[1][0][3] for e in events], ["file:///etc", "file:///home",
1035
"file:///tmp"])
1036
self.assertEquals([e.timestamp for e in events], ["119", "118", "116"])
1037
1038
def testResultTypesMostRecentSubjectOrigin(self):
1039
import_events("test/data/twenty_events.js", self)
1040
1041
ids = self.findEventIdsAndWait([],
1042
num_events = 0,
1043
result_type = ResultType.MostRecentOrigin)
1044
events = self.getEventsAndWait(ids)
1045
1046
self.assertEquals([e.timestamp for e in events], ["119", "118", "116"])
1047
1048
def testResultTypesLeastRecentSubjectOrigin(self):
1049
import_events("test/data/twenty_events.js", self)
1050
1051
ids = self.findEventIdsAndWait([],
1052
num_events = 0,
1053
result_type = ResultType.LeastRecentOrigin)
1054
events = self.getEventsAndWait(ids)
1055
1056
self.assertEquals([e.timestamp for e in events], ["116", "118", "119"])
1057
1058
def testResultTypesMostRecentMimeType(self):
1059
import_events("test/data/twenty_events.js", self)
1060
1061
ids = self.findEventIdsAndWait([],
1062
num_events = 0,
1063
result_type = ResultType.MostRecentMimeType)
1064
events = self.getEventsAndWait(ids)
1065
1066
self.assertEquals([e.timestamp for e in events], ['119', '114', '110', '107'])
1067
1068
def testResultTypesLeastRecentMimeType(self):
1069
import_events("test/data/twenty_events.js", self)
1070
1071
ids = self.findEventIdsAndWait([],
1072
num_events = 0,
1073
result_type = ResultType.LeastRecentMimeType)
1074
events = self.getEventsAndWait(ids)
1075
1076
self.assertEquals([e.timestamp for e in events], ['107', '110', '114', '119'])
1077
1078
def testResultTypesMostPopularMimeType(self):
1079
import_events("test/data/twenty_events.js", self)
1080
1081
ids = self.findEventIdsAndWait([],
1082
num_events = 0,
1083
result_type = ResultType.MostPopularMimeType)
1084
events = self.getEventsAndWait(ids)
1085
1086
self.assertEquals([e.timestamp for e in events], ['119', '110', '107', '114'])
1087
1088
def testResultTypesLeastPopularMimeType(self):
1089
import_events("test/data/twenty_events.js", self)
1090
1091
ids = self.findEventIdsAndWait([],
1092
num_events = 0,
1093
result_type = ResultType.LeastPopularMimeType)
1094
events = self.getEventsAndWait(ids)
1095
1096
self.assertEquals([e.timestamp for e in events], ['114', '107', '110', '119'])
1097
1098
def testResultTypesMostRecentSubjectInterpretation(self):
1099
import_events("test/data/twenty_events.js", self)
1100
1101
ids = self.findEventIdsAndWait([],
1102
num_events = 0,
1103
result_type = ResultType.MostRecentSubjectInterpretation)
1104
events = self.getEventsAndWait(ids)
1105
1106
self.assertEquals([e.timestamp for e in events], ['119', '118', '116', '106'])
1107
1108
def testResultTypesLeastRecentSubjectInterpretation(self):
1109
import_events("test/data/twenty_events.js", self)
1110
1111
ids = self.findEventIdsAndWait([],
1112
num_events = 0,
1113
result_type = ResultType.LeastRecentSubjectInterpretation)
1114
events = self.getEventsAndWait(ids)
1115
1116
self.assertEquals([e.timestamp for e in events], ['106', '116', '118', '119'])
1117
1118
def testResultTypesMostPopularSubjectInterpretation(self):
1119
import_events("test/data/twenty_events.js", self)
1120
1121
ids = self.findEventIdsAndWait([],
1122
num_events = 0,
1123
result_type = ResultType.MostPopularSubjectInterpretation)
1124
events = self.getEventsAndWait(ids)
1125
1126
self.assertEquals([e.timestamp for e in events], ['119', '116', '106', '118'])
1127
1128
def testResultTypesLeastPopularSubjectInterpretation(self):
1129
import_events("test/data/twenty_events.js", self)
1130
1131
ids = self.findEventIdsAndWait([],
1132
num_events = 0,
1133
result_type = ResultType.LeastPopularSubjectInterpretation)
1134
events = self.getEventsAndWait(ids)
1135
1136
self.assertEquals([e.timestamp for e in events], ['118', '106', '116', '119'])
1137
1138
if __name__ == "__main__":
1139
unittest.main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.1