« back to all changes in this revision
Viewing changes to GTG/tests/test_backend_tomboy.py
- browse files at revision 1362
- compare with another revision
- download diff
- download tarball
- view history from revision 1362
- Committer: Nimit Shah
- Date: 2014-03-10 03:37:32 UTC
- mfrom: (1360.2.18 test-cleanup)
- Revision ID: nimit.svnit@gmail.com-20140310033732-n4od1z8npybs4ooc
Rework of test-suite, by Izidor Matušov
- files added:
- requirements.txt
- tests/core
- tests/tools
- files removed:
- GTG/tests
- GTG/tests/xdg
- GTG/tests/xdg/config
- GTG/tests/xdg/config/gtg
- files renamed:
- test/data/ => data/test-data/
- test/ => tests/
- GTG/tests/__init__.py => tests/__init__.py
- GTG/tests/test_search_filter.py => tests/core/test_search_filter.py
- GTG/tests/test_search_query.py => tests/core/test_search_query.py
- GTG/tests/test_tag.py => tests/core/test_tag.py
- GTG/tests/test_dates.py => tests/tools/test_dates.py
- GTG/tests/test_interruptible.py => tests/tools/test_interruptible.py
- GTG/tests/test_networkmanager.py => tests/tools/test_networkmanager.py
- GTG/tests/test_urlregex.py => tests/tools/test_urlregex.py
- files modified:
- CHANGELOG
added
removed
GTG/tests/test_backend_tomboy.py
1
# -*- coding: utf-8 -*-
2
# -----------------------------------------------------------------------------
3
# Getting Things GNOME! - a personal organizer for the GNOME desktop
4
# Copyright (c) 2008-2013 - Lionel Dricot & Bertrand Rousseau
5
#
6
# This program is free software: you can redistribute it and/or modify it under
7
# the terms of the GNU General Public License as published by the Free Software
8
# Foundation, either version 3 of the License, or (at your option) any later
9
# version.
10
#
11
# This program is distributed in the hope that it will be useful, but WITHOUT
12
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
13
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
14
# details.
15
#
16
# You should have received a copy of the GNU General Public License along with
17
# this program. If not, see <http://www.gnu.org/licenses/>.
18
# -----------------------------------------------------------------------------
19
20
""" Tests for the tomboy backend """
21
22
from datetime import datetime
23
from dbus.mainloop.glib import DBusGMainLoop
24
import dbus
25
import dbus.glib
26
import dbus.service
27
import errno
28
from gi.repository import GObject
29
import math
30
import os
31
import random
32
import signal
33
import sys
34
import tempfile
35
import threading
36
import time
37
import unittest
38
import uuid
39
40
from GTG.backends import BackendFactory
41
from GTG.backends.genericbackend import GenericBackend
42
from GTG.core.datastore import DataStore
43
44
PID_TOMBOY = False
45
46
47
class TestBackendTomboy(unittest.TestCase):
48
""" Tests for the tomboy backend """
49
50
def setUp(self):
51
thread_tomboy = threading.Thread(target=self.spawn_fake_tomboy_server)
52
thread_tomboy.start()
53
thread_tomboy.join()
54
# only the test process should go further, the dbus server one should
55
# stop here
56
if not PID_TOMBOY:
57
return
58
# we create a custom dictionary listening to the server, and register
59
# it in GTG.
60
additional_dic = {}
61
additional_dic["use this fake connection instead"] = (
62
FakeTomboy.BUS_NAME, FakeTomboy.BUS_PATH, FakeTomboy.BUS_INTERFACE)
63
additional_dic[GenericBackend.KEY_ATTACHED_TAGS] = \
64
[GenericBackend.ALLTASKS_TAG]
65
additional_dic[GenericBackend.KEY_DEFAULT_BACKEND] = True
66
dic = BackendFactory().get_new_backend_dict('backend_tomboy',
67
additional_dic)
68
self.datastore = DataStore()
69
self.backend = self.datastore.register_backend(dic)
70
# waiting for the "start_get_tasks" to settle
71
time.sleep(1)
72
# we create a dbus session to speak with the server
73
self.bus = dbus.SessionBus()
74
obj = self.bus.get_object(FakeTomboy.BUS_NAME, FakeTomboy.BUS_PATH)
75
self.tomboy = dbus.Interface(obj, FakeTomboy.BUS_INTERFACE)
76
77
def spawn_fake_tomboy_server(self):
78
# the fake tomboy server has to be in a different process,
79
# otherwise it will lock on the GIL.
80
# For details, see
81
# http://lists.freedesktop.org/archives/dbus/2007-January/006921.html
82
83
# we use a lockfile to make sure the server is running before we start
84
# the test
85
global PID_TOMBOY
86
lockfile_fd, lockfile_path = tempfile.mkstemp()
87
PID_TOMBOY = os.fork()
88
if PID_TOMBOY:
89
# we wait in polling that the server has been started
90
while True:
91
try:
92
fd = os.open(lockfile_path,
93
os.O_CREAT | os.O_EXCL | os.O_RDWR)
94
except OSError as e:
95
if e.errno != errno.EEXIST:
96
raise
97
time.sleep(0.3)
98
continue
99
os.close(fd)
100
break
101
else:
102
FakeTomboy()
103
os.close(lockfile_fd)
104
os.unlink(lockfile_path)
105
106
def tearDown(self):
107
if not PID_TOMBOY:
108
return
109
self.datastore.save(quit=True)
110
time.sleep(0.5)
111
self.tomboy.FakeQuit()
112
# FIXME: self.bus.close()
113
os.kill(PID_TOMBOY, signal.SIGKILL)
114
os.waitpid(PID_TOMBOY, 0)
115
116
def test_everything(self):
117
# we cannot use separate test functions because we only want a single
118
# FakeTomboy dbus server running
119
if not PID_TOMBOY:
120
return
121
for function in dir(self):
122
if function.startswith("TEST_"):
123
getattr(self, function)()
124
self.tomboy.Reset()
125
for tid in self.datastore.get_all_tasks():
126
self.datastore.request_task_deletion(tid)
127
time.sleep(0.1)
128
129
def TEST_processing_tomboy_notes(self):
130
self.backend.set_attached_tags([GenericBackend.ALLTASKS_TAG])
131
# adding a note
132
note = self.tomboy.CreateNamedNote(str(uuid.uuid4()))
133
self.backend._process_tomboy_note(note)
134
self.assertEqual(len(self.datastore.get_all_tasks()), 1)
135
tid = self.backend.sync_engine.sync_memes.get_local_id(note)
136
task = self.datastore.get_task(tid)
137
# re-adding that (should not change anything)
138
self.backend._process_tomboy_note(note)
139
self.assertEqual(len(self.datastore.get_all_tasks()), 1)
140
self.assertEqual(
141
self.backend.sync_engine.sync_memes.get_local_id(note), tid)
142
# removing the note and updating gtg
143
self.tomboy.DeleteNote(note)
144
self.backend.set_task(task)
145
self.assertEqual(len(self.datastore.get_all_tasks()), 0)
146
147
def TEST_set_task(self):
148
self.backend.set_attached_tags([GenericBackend.ALLTASKS_TAG])
149
# adding a task
150
task = self.datastore.requester.new_task()
151
task.set_title("title")
152
self.backend.set_task(task)
153
self.assertEqual(len(self.tomboy.ListAllNotes()), 1)
154
note = self.tomboy.ListAllNotes()[0]
155
self.assertEqual(str(self.tomboy.GetNoteTitle(note)), task.get_title())
156
# re-adding that (should not change anything)
157
self.backend.set_task(task)
158
self.assertEqual(len(self.tomboy.ListAllNotes()), 1)
159
self.assertEqual(note, self.tomboy.ListAllNotes()[0])
160
# removing the task and updating tomboy
161
self.datastore.request_task_deletion(task.get_id())
162
self.backend._process_tomboy_note(note)
163
self.assertEqual(len(self.tomboy.ListAllNotes()), 0)
164
165
def TEST_update_newest(self):
166
self.backend.set_attached_tags([GenericBackend.ALLTASKS_TAG])
167
task = self.datastore.requester.new_task()
168
task.set_title("title")
169
self.backend.set_task(task)
170
note = self.tomboy.ListAllNotes()[0]
171
gtg_modified = task.get_modified()
172
tomboy_modified = self._modified_string_to_datetime(
173
self.tomboy.GetNoteChangeDate(note))
174
# no-one updated, nothing should happen
175
self.backend.set_task(task)
176
self.assertEqual(gtg_modified, task.get_modified())
177
actual_modified = self._modified_string_to_datetime(
178
self.tomboy.GetNoteChangeDate(note))
179
self.assertEqual(tomboy_modified, actual_modified)
180
# we update the GTG task
181
UPDATED_GTG_TITLE = "UPDATED_GTG_TITLE"
182
task.set_title(UPDATED_GTG_TITLE)
183
self.backend.set_task(task)
184
self.assertTrue(gtg_modified < task.get_modified())
185
actual_modified = self._modified_string_to_datetime(
186
self.tomboy.GetNoteChangeDate(note))
187
self.assertTrue(tomboy_modified <= actual_modified)
188
self.assertEqual(task.get_title(), UPDATED_GTG_TITLE)
189
self.assertEqual(self.tomboy.GetNoteTitle(note), UPDATED_GTG_TITLE)
190
gtg_modified = task.get_modified()
191
tomboy_modified = self._modified_string_to_datetime(
192
self.tomboy.GetNoteChangeDate(note))
193
# we update the TOMBOY task
194
UPDATED_TOMBOY_TITLE = "UPDATED_TOMBOY_TITLE"
195
# the resolution of tomboy notes changed time is 1 second, so we need
196
# to wait. This *shouldn't* be needed in the actual code because
197
# tomboy signals are always a few seconds late.
198
time.sleep(1)
199
self.tomboy.SetNoteContents(note, UPDATED_TOMBOY_TITLE)
200
self.backend._process_tomboy_note(note)
201
self.assertTrue(gtg_modified <= task.get_modified())
202
actual_modified = self._modified_string_to_datetime(
203
self.tomboy.GetNoteChangeDate(note))
204
self.assertTrue(tomboy_modified <= actual_modified)
205
self.assertEqual(task.get_title(), UPDATED_TOMBOY_TITLE)
206
self.assertEqual(self.tomboy.GetNoteTitle(note), UPDATED_TOMBOY_TITLE)
207
208
def TEST_processing_tomboy_notes_with_tags(self):
209
self.backend.set_attached_tags(['@a'])
210
# adding a not syncable note
211
note = self.tomboy.CreateNamedNote("title" + str(uuid.uuid4()))
212
self.backend._process_tomboy_note(note)
213
self.assertEqual(len(self.datastore.get_all_tasks()), 0)
214
# re-adding that (should not change anything)
215
self.backend._process_tomboy_note(note)
216
self.assertEqual(len(self.datastore.get_all_tasks()), 0)
217
# adding a tag to that note
218
self.tomboy.SetNoteContents(note, "something with @a")
219
self.backend._process_tomboy_note(note)
220
self.assertEqual(len(self.datastore.get_all_tasks()), 1)
221
# removing the tag and resyncing
222
self.tomboy.SetNoteContents(note, "something with no tags")
223
self.backend._process_tomboy_note(note)
224
self.assertEqual(len(self.datastore.get_all_tasks()), 0)
225
# adding a syncable note
226
note = self.tomboy.CreateNamedNote("title @a" + str(uuid.uuid4()))
227
self.backend._process_tomboy_note(note)
228
self.assertEqual(len(self.datastore.get_all_tasks()), 1)
229
tid = self.backend.sync_engine.sync_memes.get_local_id(note)
230
task = self.datastore.get_task(tid)
231
# re-adding that (should not change anything)
232
self.backend._process_tomboy_note(note)
233
self.assertEqual(len(self.datastore.get_all_tasks()), 1)
234
self.assertEqual(
235
self.backend.sync_engine.sync_memes.get_local_id(note), tid)
236
# removing the note and updating gtg
237
self.tomboy.DeleteNote(note)
238
self.backend.set_task(task)
239
self.assertEqual(len(self.datastore.get_all_tasks()), 0)
240
241
def TEST_set_task_with_tags(self):
242
self.backend.set_attached_tags(['@a'])
243
# adding a not syncable task
244
task = self.datastore.requester.new_task()
245
task.set_title("title")
246
self.backend.set_task(task)
247
self.assertEqual(len(self.tomboy.ListAllNotes()), 0)
248
# making that task syncable
249
task.set_title("something else")
250
task.add_tag("@a")
251
self.backend.set_task(task)
252
self.assertEqual(len(self.tomboy.ListAllNotes()), 1)
253
note = self.tomboy.ListAllNotes()[0]
254
self.assertEqual(str(self.tomboy.GetNoteTitle(note)), task.get_title())
255
# re-adding that (should not change anything)
256
self.backend.set_task(task)
257
self.assertEqual(len(self.tomboy.ListAllNotes()), 1)
258
self.assertEqual(note, self.tomboy.ListAllNotes()[0])
259
# removing the syncable property and updating tomboy
260
task.remove_tag("@a")
261
self.backend.set_task(task)
262
self.assertEqual(len(self.tomboy.ListAllNotes()), 0)
263
264
def TEST_multiple_task_same_title(self):
265
self.backend.set_attached_tags(['@a'])
266
how_many_tasks = int(math.ceil(20 * random.random()))
267
for iteration in range(0, how_many_tasks):
268
task = self.datastore.requester.new_task()
269
task.set_title("title")
270
task.add_tag('@a')
271
self.backend.set_task(task)
272
self.assertEqual(len(self.tomboy.ListAllNotes()), how_many_tasks)
273
274
def _modified_string_to_datetime(self, modified_string):
275
return datetime.fromtimestamp(modified_string)
276
277
278
def test_suite():
279
return unittest.TestLoader().loadTestsFromTestCase(TestBackendTomboy)
280
281
282
class FakeTomboy(dbus.service.Object):
283
"""
284
D-Bus service object that mimics TOMBOY
285
"""
286
287
# We don't directly use the tomboy dbus path to avoid conflicts
288
# if tomboy is running during the test
289
BUS_NAME = "Fake.Tomboy"
290
BUS_PATH = "/Fake/Tomboy"
291
BUS_INTERFACE = "Fake.Tomboy.RemoteControl"
292
293
def __init__(self):
294
# Attach the object to D-Bus
295
DBusGMainLoop(set_as_default=True)
296
self.bus = dbus.SessionBus()
297
bus_name = dbus.service.BusName(self.BUS_NAME, bus=self.bus)
298
dbus.service.Object.__init__(self, bus_name, self.BUS_PATH)
299
self.notes = {}
300
threading.Thread(target=self.fake_main_loop).start()
301
302
@dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
303
def GetNoteContents(self, note):
304
return self.notes[note]['content']
305
306
@dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="b")
307
def NoteExists(self, note):
308
return note in self.notes
309
310
@dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="d")
311
def GetNoteChangeDate(self, note):
312
return self.notes[note]['changed']
313
314
@dbus.service.method(BUS_INTERFACE, in_signature="ss")
315
def SetNoteContents(self, note, text):
316
self.fake_update_note(note)
317
self.notes[note]['content'] = text
318
319
@dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
320
def GetNoteTitle(self, note):
321
return self._GetNoteTitle(note)
322
323
def _GetNoteTitle(self, note):
324
content = self.notes[note]['content']
325
try:
326
end_of_title = content.index('\n')
327
except ValueError:
328
return content
329
return content[:end_of_title]
330
331
@dbus.service.method(BUS_INTERFACE, in_signature="s")
332
def DeleteNote(self, note):
333
del self.notes[note]
334
335
@dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
336
def CreateNamedNote(self, title):
337
# this is to mimic the way tomboy handles title clashes
338
if self._FindNote(title) != '':
339
return ''
340
note = str(uuid.uuid4())
341
self.notes[note] = {'content': title}
342
self.fake_update_note(note)
343
return note
344
345
@dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
346
def FindNote(self, title):
347
return self._FindNote(title)
348
349
def _FindNote(self, title):
350
for note in self.notes:
351
if self._GetNoteTitle(note) == title:
352
return note
353
return ''
354
355
@dbus.service.method(BUS_INTERFACE, out_signature="as")
356
def ListAllNotes(self):
357
return list(self.notes)
358
359
@dbus.service.signal(BUS_INTERFACE, signature='s')
360
def NoteSaved(self, note):
361
pass
362
363
@dbus.service.signal(BUS_INTERFACE, signature='s')
364
def NoteDeleted(self, note):
365
pass
366
367
###############################################################################
368
### Function with the fake_ prefix are here to assist in testing, they do not
369
### need to be present in the real class
370
###############################################################################
371
def fake_update_note(self, note):
372
self.notes[note]['changed'] = time.mktime(datetime.now().timetuple())
373
374
def fake_main_loop(self):
375
GObject.threads_init()
376
dbus.glib.init_threads()
377
self.main_loop = GObject.MainLoop()
378
self.main_loop.run()
379
380
@dbus.service.method(BUS_INTERFACE)
381
def Reset(self):
382
self.notes = {}
383
384
@dbus.service.method(BUS_INTERFACE)
385
def FakeQuit(self):
386
threading.Timer(0.2, self._fake_quit).start()
387
388
def _fake_quit(self):
389
self.main_loop.quit()
390
sys.exit(0)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1