← Back to branch summary

~saurabhanandiit/gtg/exportFixed

« back to all changes in this revision

Viewing changes to GTG/tests/test_backend_tomboy.py

Merge of my work on liblarch newbase and all the backends ported to liblarch
(which mainly means porting the datastore).
One failing test, will check it.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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-2009 - 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
'''
 
21
Tests for the tomboy backend
 
22
'''
 
23
 
 
24
import os
 
25
import sys
 
26
import errno
 
27
import unittest
 
28
import uuid
 
29
import signal
 
30
import time
 
31
import math
 
32
import dbus
 
33
import gobject
 
34
import random
 
35
import threading
 
36
import dbus.glib
 
37
import dbus.service
 
38
import tempfile
 
39
from datetime           import datetime
 
40
from dbus.mainloop.glib import DBusGMainLoop
 
41
 
 
42
from GTG.core.datastore          import DataStore
 
43
from GTG.backends                import BackendFactory
 
44
from GTG.backends.genericbackend import GenericBackend
 
45
 
 
46
 
 
47
 
 
48
class TestBackendTomboy(unittest.TestCase):
 
49
    '''
 
50
    Tests for the tomboy backend.
 
51
    '''
 
52
    
 
53
 
 
54
    def setUp(self):
 
55
        self.spawn_fake_tomboy_server()
 
56
        #only the test process should go further, the dbus server one should
 
57
        #stop here
 
58
        if not self.child_process_pid: return
 
59
        #we create a custom dictionary listening to the server, and register it
 
60
        # in GTG. 
 
61
        additional_dic = {}
 
62
        additional_dic["use this fake connection instead"] = \
 
63
                (FakeTomboy.BUS_NAME,
 
64
                 FakeTomboy.BUS_PATH,
 
65
                 FakeTomboy.BUS_INTERFACE)
 
66
        additional_dic[GenericBackend.KEY_ATTACHED_TAGS] = \
 
67
                        [GenericBackend.ALLTASKS_TAG]
 
68
        additional_dic[GenericBackend.KEY_DEFAULT_BACKEND] = True
 
69
        dic = BackendFactory().get_new_backend_dict('backend_tomboy', 
 
70
                                                   additional_dic)
 
71
        self.datastore = DataStore()
 
72
        self.backend = self.datastore.register_backend(dic)
 
73
        #waiting for the "start_get_tasks" to settle
 
74
        time.sleep(1)
 
75
        #we create a dbus session to speak with the server
 
76
        self.bus = dbus.SessionBus()
 
77
        obj = self.bus.get_object(FakeTomboy.BUS_NAME, FakeTomboy.BUS_PATH)
 
78
        self.tomboy = dbus.Interface(obj, FakeTomboy.BUS_INTERFACE)
 
79
        
 
80
    def spawn_fake_tomboy_server(self):
 
81
        #the fake tomboy server has to be in a different process,
 
82
        #otherwise it will lock on the GIL.
 
83
        #For details, see
 
84
        #http://lists.freedesktop.org/archives/dbus/2007-January/006921.html
 
85
 
 
86
        #we use a lockfile to make sure the server is running before we start
 
87
        # the test
 
88
        lockfile_fd, lockfile_path = tempfile.mkstemp()
 
89
        self.child_process_pid = os.fork()
 
90
        if self.child_process_pid:
 
91
            #we wait in polling that the server has been started
 
92
            while True:
 
93
                try:
 
94
                    fd = os.open(lockfile_path, os.O_CREAT | os.O_EXCL | os.O_RDWR)
 
95
                except OSError, e:
 
96
                    if e.errno != errno.EEXIST:
 
97
                        raise
 
98
                    time.sleep(0.3)
 
99
                    continue
 
100
                os.close(fd)
 
101
                break
 
102
        else:
 
103
            FakeTomboy()
 
104
            os.close(lockfile_fd)
 
105
            os.unlink(lockfile_path)
 
106
 
 
107
    def tearDown(self):
 
108
        if not self.child_process_pid: return
 
109
        self.datastore.save(quit = True) 
 
110
        print "QUIT"
 
111
        time.sleep(0.5)
 
112
        self.tomboy.FakeQuit()
 
113
        self.bus.close()
 
114
        os.kill(self.child_process_pid, signal.SIGKILL)
 
115
        os.waitpid(self.child_process_pid, 0)
 
116
    
 
117
    def test_everything(self):
 
118
        '''
 
119
        '''
 
120
        #we cannot use separate test functions because we only want a single
 
121
        # FakeTomboy dbus server running
 
122
        if not self.child_process_pid: return
 
123
        for function in dir(self):
 
124
            if function.startswith("TEST_"):
 
125
                getattr(self, function)()
 
126
                self.tomboy.Reset()
 
127
                for tid in self.datastore.get_all_tasks():
 
128
                    self.datastore.request_task_deletion(tid)
 
129
                time.sleep(0.1)
 
130
 
 
131
    def TEST_processing_tomboy_notes(self):
 
132
        self.backend.set_attached_tags([GenericBackend.ALLTASKS_TAG])
 
133
        #adding a note
 
134
        note = self.tomboy.CreateNamedNote(str(uuid.uuid4()))
 
135
        self.backend._process_tomboy_note(note)
 
136
        self.assertEqual(len(self.datastore.get_all_tasks()), 1)
 
137
        tid = self.backend.sync_engine.sync_memes.get_local_id(note)
 
138
        task = self.datastore.get_task(tid)
 
139
        #re-adding that (should not change anything)
 
140
        self.backend._process_tomboy_note(note)
 
141
        self.assertEqual(len(self.datastore.get_all_tasks()), 1)
 
142
        self.assertEqual( \
 
143
                self.backend.sync_engine.sync_memes.get_local_id(note), tid)
 
144
        #removing the note and updating gtg
 
145
        self.tomboy.DeleteNote(note)
 
146
        self.backend.set_task(task)
 
147
        self.assertEqual(len(self.datastore.get_all_tasks()), 0)
 
148
 
 
149
    def TEST_set_task(self):
 
150
        self.backend.set_attached_tags([GenericBackend.ALLTASKS_TAG])
 
151
        #adding a task
 
152
        task = self.datastore.requester.new_task()
 
153
        task.set_title("title")
 
154
        self.backend.set_task(task)
 
155
        self.assertEqual(len(self.tomboy.ListAllNotes()), 1)
 
156
        note = self.tomboy.ListAllNotes()[0]
 
157
        self.assertEqual(str(self.tomboy.GetNoteTitle(note)), task.get_title())
 
158
        #re-adding that (should not change anything)
 
159
        self.backend.set_task(task)
 
160
        self.assertEqual(len(self.tomboy.ListAllNotes()), 1)
 
161
        self.assertEqual(note, self.tomboy.ListAllNotes()[0])
 
162
        #removing the task and updating tomboy
 
163
        self.datastore.request_task_deletion(task.get_id())
 
164
        self.backend._process_tomboy_note(note)
 
165
        self.assertEqual(len(self.tomboy.ListAllNotes()), 0)
 
166
 
 
167
    def TEST_update_newest(self):
 
168
        self.backend.set_attached_tags([GenericBackend.ALLTASKS_TAG])
 
169
        task = self.datastore.requester.new_task()
 
170
        task.set_title("title")
 
171
        self.backend.set_task(task)
 
172
        note = self.tomboy.ListAllNotes()[0]
 
173
        gtg_modified = task.get_modified()
 
174
        tomboy_modified = self._modified_string_to_datetime( \
 
175
                    self.tomboy.GetNoteChangeDate(note))
 
176
        #no-one updated, nothing should happen
 
177
        self.backend.set_task(task)
 
178
        self.assertEqual(gtg_modified, task.get_modified())
 
179
        self.assertEqual(tomboy_modified, \
 
180
                         self._modified_string_to_datetime( \
 
181
                         self.tomboy.GetNoteChangeDate(note)))
 
182
        #we update the GTG task
 
183
        UPDATED_GTG_TITLE = "UPDATED_GTG_TITLE"
 
184
        task.set_title(UPDATED_GTG_TITLE)
 
185
        self.backend.set_task(task)
 
186
        self.assertTrue(gtg_modified < task.get_modified())
 
187
        self.assertTrue(tomboy_modified <=
 
188
                         self._modified_string_to_datetime( \
 
189
                         self.tomboy.GetNoteChangeDate(note)))
 
190
        self.assertEqual(task.get_title(), UPDATED_GTG_TITLE)
 
191
        self.assertEqual(self.tomboy.GetNoteTitle(note), UPDATED_GTG_TITLE)
 
192
        gtg_modified = task.get_modified()
 
193
        tomboy_modified = self._modified_string_to_datetime( \
 
194
                    self.tomboy.GetNoteChangeDate(note))
 
195
        #we update the TOMBOY task
 
196
        UPDATED_TOMBOY_TITLE = "UPDATED_TOMBOY_TITLE"
 
197
        #the resolution of tomboy notes changed time is 1 second, so we need
 
198
        # to wait. This *shouldn't* be needed in the actual code because
 
199
        # tomboy signals are always a few seconds late.
 
200
        time.sleep(1)
 
201
        self.tomboy.SetNoteContents(note, UPDATED_TOMBOY_TITLE)
 
202
        self.backend._process_tomboy_note(note)
 
203
        self.assertTrue(gtg_modified <= task.get_modified())
 
204
        self.assertTrue(tomboy_modified <=
 
205
                         self._modified_string_to_datetime( \
 
206
                         self.tomboy.GetNoteChangeDate(note)))
 
207
        self.assertEqual(task.get_title(), UPDATED_TOMBOY_TITLE)
 
208
        self.assertEqual(self.tomboy.GetNoteTitle(note), UPDATED_TOMBOY_TITLE)
 
209
 
 
210
    def TEST_processing_tomboy_notes_with_tags(self):
 
211
        self.backend.set_attached_tags(['@a'])
 
212
        #adding a not syncable note
 
213
        note = self.tomboy.CreateNamedNote("title" + str(uuid.uuid4()))
 
214
        self.backend._process_tomboy_note(note)
 
215
        self.assertEqual(len(self.datastore.get_all_tasks()), 0)
 
216
        #re-adding that (should not change anything)
 
217
        self.backend._process_tomboy_note(note)
 
218
        self.assertEqual(len(self.datastore.get_all_tasks()), 0)
 
219
        #adding a tag to that note
 
220
        self.tomboy.SetNoteContents(note, "something with @a")
 
221
        self.backend._process_tomboy_note(note)
 
222
        self.assertEqual(len(self.datastore.get_all_tasks()), 1)
 
223
        #removing the tag and resyncing
 
224
        self.tomboy.SetNoteContents(note, "something with no tags")
 
225
        self.backend._process_tomboy_note(note)
 
226
        self.assertEqual(len(self.datastore.get_all_tasks()), 0)
 
227
        #adding a syncable note
 
228
        note = self.tomboy.CreateNamedNote("title @a" + str(uuid.uuid4()))
 
229
        self.backend._process_tomboy_note(note)
 
230
        self.assertEqual(len(self.datastore.get_all_tasks()), 1)
 
231
        tid = self.backend.sync_engine.sync_memes.get_local_id(note)
 
232
        task = self.datastore.get_task(tid)
 
233
        #re-adding that (should not change anything)
 
234
        self.backend._process_tomboy_note(note)
 
235
        self.assertEqual(len(self.datastore.get_all_tasks()), 1)
 
236
        self.assertEqual( \
 
237
                self.backend.sync_engine.sync_memes.get_local_id(note), tid)
 
238
        #removing the note and updating gtg
 
239
        self.tomboy.DeleteNote(note)
 
240
        self.backend.set_task(task)
 
241
        self.assertEqual(len(self.datastore.get_all_tasks()), 0)
 
242
 
 
243
    def TEST_set_task_with_tags(self):
 
244
        self.backend.set_attached_tags(['@a'])
 
245
        #adding a not syncable task
 
246
        task = self.datastore.requester.new_task()
 
247
        task.set_title("title")
 
248
        self.backend.set_task(task)
 
249
        self.assertEqual(len(self.tomboy.ListAllNotes()), 0)
 
250
        #making that task  syncable 
 
251
        task.set_title("something else")
 
252
        task.add_tag("@a")
 
253
        self.backend.set_task(task)
 
254
        self.assertEqual(len(self.tomboy.ListAllNotes()),  1)
 
255
        note = self.tomboy.ListAllNotes()[0]
 
256
        self.assertEqual(str(self.tomboy.GetNoteTitle(note)), task.get_title())
 
257
        #re-adding that (should not change anything)
 
258
        self.backend.set_task(task)
 
259
        self.assertEqual(len(self.tomboy.ListAllNotes()), 1)
 
260
        self.assertEqual(note, self.tomboy.ListAllNotes()[0])
 
261
        #removing the syncable property and updating tomboy
 
262
        task.remove_tag("@a")
 
263
        self.backend.set_task(task)
 
264
        self.assertEqual(len(self.tomboy.ListAllNotes()), 0)
 
265
 
 
266
    def TEST_multiple_task_same_title(self):
 
267
        self.backend.set_attached_tags(['@a'])
 
268
        how_many_tasks = int(math.ceil(20 * random.random()))
 
269
        for iteration in xrange(0, how_many_tasks):
 
270
            task = self.datastore.requester.new_task()
 
271
            task.set_title("title")
 
272
            task.add_tag('@a')
 
273
            self.backend.set_task(task)
 
274
        self.assertEqual(len(self.tomboy.ListAllNotes()), how_many_tasks)
 
275
 
 
276
    def _modified_string_to_datetime(self, modified_string):
 
277
        return datetime.fromtimestamp(modified_string)
 
278
 
 
279
 
 
280
def test_suite():
 
281
    return unittest.TestLoader().loadTestsFromTestCase(TestBackendTomboy)
 
282
 
 
283
 
 
284
 
 
285
class FakeTomboy(dbus.service.Object):
 
286
    """
 
287
    D-Bus service object that mimics TOMBOY
 
288
    """
 
289
 
 
290
 
 
291
    #We don't directly use the tomboy dbus path to avoid conflicts
 
292
    # if tomboy is running during the test
 
293
 
 
294
    BUS_NAME = "Fake.Tomboy"
 
295
    BUS_PATH = "/Fake/Tomboy"
 
296
    BUS_INTERFACE = "Fake.Tomboy.RemoteControl"
 
297
 
 
298
    def __init__(self):
 
299
        # Attach the object to D-Bus
 
300
        DBusGMainLoop(set_as_default=True)
 
301
        self.bus = dbus.SessionBus() 
 
302
        bus_name = dbus.service.BusName(self.BUS_NAME, bus = self.bus)
 
303
        dbus.service.Object.__init__(self, bus_name, self.BUS_PATH)
 
304
        self.notes = {}
 
305
        threading.Thread(target = self.fake_main_loop).start()
 
306
 
 
307
    @dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
 
308
    def GetNoteContents(self, note):
 
309
        return self.notes[note]['content']
 
310
 
 
311
    @dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="b")
 
312
    def NoteExists(self, note):
 
313
        return self.notes.has_key(note)
 
314
 
 
315
    @dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="d")
 
316
    def GetNoteChangeDate(self, note):
 
317
        return self.notes[note]['changed']
 
318
 
 
319
    @dbus.service.method(BUS_INTERFACE, in_signature="ss")
 
320
    def SetNoteContents(self, note, text):
 
321
        self.fake_update_note(note)
 
322
        self.notes[note]['content'] = text
 
323
 
 
324
    @dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
 
325
    def GetNoteTitle(self, note):
 
326
        return self._GetNoteTitle(note)
 
327
 
 
328
    def _GetNoteTitle(self, note):
 
329
        content = self.notes[note]['content']
 
330
        try:
 
331
            end_of_title = content.index('\n')
 
332
        except ValueError:
 
333
            return content
 
334
        return content[:end_of_title]
 
335
 
 
336
    @dbus.service.method(BUS_INTERFACE, in_signature="s")
 
337
    def DeleteNote(self, note):
 
338
        del self.notes[note]
 
339
 
 
340
    @dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
 
341
    def CreateNamedNote(self, title):
 
342
        #this is to mimic the way tomboy handles title clashes
 
343
        if self._FindNote(title) != '':
 
344
            return ''
 
345
        note = str(uuid.uuid4())
 
346
        self.notes[note] = {'content': title}
 
347
        self.fake_update_note(note)
 
348
        return note
 
349
 
 
350
    @dbus.service.method(BUS_INTERFACE, in_signature="s", out_signature="s")
 
351
    def FindNote(self, title):
 
352
        return self._FindNote(title)
 
353
 
 
354
    def _FindNote(self, title):
 
355
        for note in self.notes:
 
356
            if self._GetNoteTitle(note) == title:
 
357
                return note
 
358
        return ''
 
359
 
 
360
    @dbus.service.method(BUS_INTERFACE, out_signature = "as")
 
361
    def ListAllNotes(self):
 
362
        return list(self.notes)
 
363
    
 
364
    @dbus.service.signal(BUS_INTERFACE, signature='s')
 
365
    def NoteSaved(self, note):
 
366
        pass
 
367
 
 
368
    @dbus.service.signal(BUS_INTERFACE, signature='s')
 
369
    def NoteDeleted(self, note):
 
370
        pass
 
371
 
 
372
###############################################################################
 
373
### Function with the fake_ prefix are here to assist in testing, they do not
 
374
### need to be present in the real class
 
375
###############################################################################
 
376
 
 
377
    def fake_update_note(self, note):
 
378
        self.notes[note]['changed'] = time.mktime(datetime.now().timetuple())
 
379
 
 
380
    def fake_main_loop(self):
 
381
        gobject.threads_init()
 
382
        dbus.glib.init_threads()
 
383
        self.main_loop = gobject.MainLoop()
 
384
        self.main_loop.run()
 
385
 
 
386
    @dbus.service.method(BUS_INTERFACE)
 
387
    def Reset(self):
 
388
        self.notes = {}
 
389
 
 
390
    @dbus.service.method(BUS_INTERFACE)
 
391
    def FakeQuit(self):
 
392
        threading.Timer(0.2, self._fake_quit).start()
 
393
 
 
394
    def _fake_quit(self):
 
395
        self.main_loop.quit()
 
396
        sys.exit(0)
 
397