Luca Invernizzi has proposed merging lp:~gtg-user/gtg/backends-utils into lp:gtg with lp:~gtg-user/gtg/backends-window as a prerequisite.
Requested reviews: Gtg developers (gtg) This branch contains all the common utils used by the backends that are not the default one. Mainly, it contains code for: - telling if a remote task is new, has to be updated or removed (it's a standalone library) - getting remote tasks in polling - a watchdog for stalling functions The code contained in this merge isn't used by "Trunk" GTG, but it will be as backends are merged: this is why you won't see any difference in GTG's behavior now. Tests and documentation for these parts is here too. (lp:~gtg-user/gtg/backends-window should be merged before this one. Some file needed by both are just there to review). -- https://code.launchpad.net/~gtg-user/gtg/backends-utils/+merge/32278 Your team Gtg developers is requested to review the proposed merge of lp:~gtg-user/gtg/backends-utils into lp:gtg.
=== added file 'GTG/backends/periodicimportbackend.py' --- GTG/backends/periodicimportbackend.py 1970-01-01 00:00:00 +0000 +++ GTG/backends/periodicimportbackend.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,90 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +Contains PeriodicImportBackend, a GenericBackend specialized for checking the +remote backend in polling. +''' + +import threading + +from GTG.backends.genericbackend import GenericBackend +from GTG.backends.backendsignals import BackendSignals +from GTG.tools.interruptible import interruptible + + + +class PeriodicImportBackend(GenericBackend): + ''' + This class can be used in place of GenericBackend when a periodic import is + necessary, as the remote service providing tasks does not signals the + changes. + To use this, only two things are necessary: + - using do_periodic_import instead of start_get_tasks + - having in _static_parameters a "period" key, as in + "period": { \ + GenericBackend.PARAM_TYPE: GenericBackend.TYPE_INT, \ + GenericBackend.PARAM_DEFAULT_VALUE: 2, }, + This specifies the time that must pass between consecutive imports + (in minutes) + ''' + + @interruptible + def start_get_tasks(self): + ''' + This function launches the first periodic import, and schedules the + next ones. + ''' + try: + if self.import_timer: + self.import_timer.cancel() + except: + pass + self._start_get_tasks() + self.cancellation_point() + if self.is_enabled() == False: + return + self.import_timer = threading.Timer( \ + self._parameters['period'] * 60.0, \ + self.start_get_tasks) + self.import_timer.start() + + def _start_get_tasks(self): + ''' + This function executes an imports and schedules the next + ''' + self.cancellation_point() + BackendSignals().backend_sync_started(self.get_id()) + self.do_periodic_import() + BackendSignals().backend_sync_ended(self.get_id()) + + def quit(self, disable = False): + ''' + Called when GTG quits or disconnects the backend. + ''' + super(PeriodicImportBackend, self).quit(disable) + try: + self.import_timer.cancel() + except Exception: + pass + try: + self.import_timer.join() + except Exception: + pass + === added file 'GTG/backends/syncengine.py' --- GTG/backends/syncengine.py 1970-01-01 00:00:00 +0000 +++ GTG/backends/syncengine.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,268 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +This library deals with synchronizing two sets of objects. +It works like this: + - We have two sets of generic objects (local and remote) + - We present one object of either one of the sets and ask the library what's + the state of its synchronization + - the library will tell us if we need to add a clone object in the other set, + update it or, if the other one has been removed, remove also this one +''' +from GTG.tools.twokeydict import TwoKeyDict + + +TYPE_LOCAL = "local" +TYPE_REMOTE = "remote" + + + +class SyncMeme(object): + ''' + A SyncMeme is the object storing the data needed to keep track of the state + of two objects synchronization. + This basic version, that can be expanded as needed by the code using the + SyncEngine, just stores the modified date and time of the last + synchronization for both objects (local and remote) + ''' + #NOTE: Checking objects CRCs would make this check nicer, as we could know + # if the object was really changed, or it has just updated its + # modified time (invernizzi) + + + def set_local_last_modified(self, modified_datetime): + ''' + Setter function for the local object modified datetime. + + @param modified_datetime: the local object modified datetime + ''' + self.local_last_modified = modified_datetime + + def get_local_last_modified(self): + ''' + Getter function for the local object modified datetime. + ''' + return self.local_last_modified + + def set_remote_last_modified(self, modified_datetime): + ''' + Setter function for the remote object modified datetime. + + @param modified_datetime: the remote object modified datetime + ''' + self.remote_last_modified = modified_datetime + + def get_remote_last_modified(self): + ''' + Getter function for the remote object modified datetime. + ''' + return self.remote_last_modified + + def which_is_newest(self, local_modified, remote_modified): + ''' + Given the updated modified time for both the local and the remote + objects, it checks them against the stored modified times and + then against each other. + + @returns string: "local"- if the local object has been modified and its + the newest + "remote" - the same for the remote object + None - if no object modified time is newer than the + stored one (the objects have not been modified) + ''' + if local_modified <= self.local_last_modified and \ + remote_modified <= self.remote_last_modified: + return None + if local_modified > remote_modified: + return "local" + else: + return "remote" + + def get_origin(self): + ''' + Returns the name of the source that firstly presented the object + ''' + return self.origin + + def set_origin(self, origin): + ''' + Sets the source that presented the object for the first time. This + source holds the original object, while the other holds the copy. + This can be useful in the case of "lost syncability" (see the SyncEngine + for an explaination). + + @param origin: object representing the source + ''' + self.origin = origin + + + +class SyncMemes(TwoKeyDict): + ''' + A TwoKeyDict, with just the names changed to be better understandable. + The meaning of these names is explained in the SyncEngine class description. + It's used to store a set of SyncMeme objects, each one keeping storing all + the data needed to keep track of a single relationship. + ''' + + + get_remote_id = TwoKeyDict._get_secondary_key + get_local_id = TwoKeyDict._get_primary_key + remove_local_id = TwoKeyDict._remove_by_primary + remove_remote_id = TwoKeyDict._remove_by_secondary + get_meme_from_local_id = TwoKeyDict._get_by_primary + get_meme_from_remote_id = TwoKeyDict._get_by_secondary + get_all_local = TwoKeyDict._get_all_primary_keys + get_all_remote = TwoKeyDict._get_all_secondary_keys + + + +class SyncEngine(object): + ''' + The SyncEngine is an object useful in keeping two sets of objects + synchronized. + One set is called the Local set, the other is the Remote one. + It stores the state of the synchronization and the latest state of each + object. + When asked, it can tell if a couple of related objects are up to date in the + sync and, if not, which one must be updated. + + It stores the state of each relationship in a series of SyncMeme. + ''' + + + UPDATE = "update" + REMOVE = "remove" + ADD = "add" + LOST_SYNCABILITY = "lost syncability" + + def __init__(self): + ''' + Initializes the storage of object relationships. + ''' + self.sync_memes = SyncMemes() + + def _analyze_element(self, + element_id, + is_local, + has_local, + has_remote, + is_syncable = True): + ''' + Given an object that should be synced with another one, + it finds out about the related object, and decides whether: + - the other object hasn't been created yet (thus must be added) + - the other object has been deleted (thus this one must be deleted) + - the other object is present, but either one has been changed + + A particular case happens if the other object is present, but the + "is_syncable" parameter (which tells that we intend to keep these two + objects in sync) is set to False. In this case, this function returns + that the Syncability property has been lost. This case is interesting if + we want to delete one of the two objects (the one that has been cloned + from the original). + + @param element_id: the id of the element we're analysing. + @param is_local: True if the element analysed is the local one (not the + remote) + @param has_local: function that accepts an id of the local set and + returns True if the element is present + @param has_remote: function that accepts an id of the remote set and + returns True if the element is present + @param is_syncable: explained above + @returns string: one of self.UPDATE, self.ADD, self.REMOVE, + self.LOST_SYNCABILITY + ''' + if is_local: + get_other_id = self.sync_memes.get_remote_id + is_task_present = has_remote + else: + get_other_id = self.sync_memes.get_local_id + is_task_present = has_local + + try: + other_id = get_other_id(element_id) + if is_task_present(other_id): + if is_syncable: + return self.UPDATE, other_id + else: + return self.LOST_SYNCABILITY, other_id + else: + return self.REMOVE, None + except KeyError: + if is_syncable: + return self.ADD, None + return None, None + + def analyze_local_id(self, element_id, *other_args): + ''' + Shortcut to call _analyze_element for a local element + ''' + return self._analyze_element(element_id, True, *other_args) + + def analyze_remote_id(self, element_id, *other_args): + ''' + Shortcut to call _analyze_element for a remote element + ''' + return self._analyze_element(element_id, False, *other_args) + + def record_relationship(self, local_id, remote_id, meme): + ''' + Records that an object from the local set is related with one a remote + set. + + @param local_id: the id of the local task + @param remote_id: the id of the remote task + @param meme: the SyncMeme that keeps track of the relationship + ''' + triplet = (local_id, remote_id, meme) + self.sync_memes.add(triplet) + + def break_relationship(self, local_id = None, remote_id = None): + ''' + breaks a relationship between two objects. + Only one of the two parameters is necessary to identify the + relationship. + + @param local_id: the id of the local task + @param remote_id: the id of the remote task + ''' + if local_id: + self.sync_memes.remove_local_id(local_id) + elif remote_id: + self.sync_memes.remove_remote_id(remote_id) + + def __getattr__(self, attr): + ''' + The functions listed here are passed directly to the SyncMeme object + + @param attr: a function name among the ones listed here + @returns object: the function return object. + ''' + if attr in ['get_remote_id', + 'get_local_id', + 'get_meme_from_local_id', + 'get_meme_from_remote_id', + 'get_all_local', + 'get_all_remote']: + return getattr(self.sync_memes, attr) + else: + raise AttributeError + === added file 'GTG/tests/test_bidict.py' --- GTG/tests/test_bidict.py 1970-01-01 00:00:00 +0000 +++ GTG/tests/test_bidict.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +Tests for the diDict class +''' + +import unittest +import uuid + +from GTG.tools.bidict import BiDict + + + +class TestBiDict(unittest.TestCase): + ''' + Tests for the BiDict object. + ''' + + + def test_add_and_gets(self): + ''' + Test for the __init__, _get_by_first, _get_by_second function + ''' + pairs = [(uuid.uuid4(), uuid.uuid4()) for a in xrange(10)] + bidict = BiDict(*pairs) + for pair in pairs: + self.assertEqual(bidict._get_by_first(pair[0]), pair[1]) + self.assertEqual(bidict._get_by_second(pair[1]), pair[0]) + + def test_remove_by_first_or_second(self): + ''' + Tests for removing elements from the biDict + ''' + pair_first = (1, 'one') + pair_second = (2, 'two') + bidict = BiDict(pair_first, pair_second) + bidict._remove_by_first(pair_first[0]) + bidict._remove_by_second(pair_second[1]) + missing_first = 0 + missing_second = 0 + try: + bidict._get_by_first(pair_first[0]) + except KeyError: + missing_first += 1 + try: + bidict._get_by_first(pair_second[0]) + except KeyError: + missing_first += 1 + try: + bidict._get_by_second(pair_first[1]) + except KeyError: + missing_second += 1 + try: + bidict._get_by_second(pair_second[1]) + except KeyError: + missing_second += 1 + self.assertEqual(missing_first, 2) + self.assertEqual(missing_second, 2) + +def test_suite(): + return unittest.TestLoader().loadTestsFromTestCase(TestBiDict) + === added file 'GTG/tests/test_dates.py' --- GTG/tests/test_dates.py 1970-01-01 00:00:00 +0000 +++ GTG/tests/test_dates.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,43 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +Tests for the various Date classes +''' + +import unittest + +from GTG.tools.dates import get_canonical_date + +class TestDates(unittest.TestCase): + ''' + Tests for the various Date classes + ''' + + def test_get_canonical_date(self): + ''' + Tests for "get_canonical_date" + ''' + for str in ["1985-03-29", "now", "soon", "later", ""]: + date = get_canonical_date(str) + self.assertEqual(date.__str__(), str) + +def test_suite(): + return unittest.TestLoader().loadTestsFromTestCase(TestDates) + === added file 'GTG/tests/test_syncengine.py' --- GTG/tests/test_syncengine.py 1970-01-01 00:00:00 +0000 +++ GTG/tests/test_syncengine.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,189 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +Tests for the SyncEngine class +''' + +import unittest +import uuid + +from GTG.backends.syncengine import SyncEngine + + + +class TestSyncEngine(unittest.TestCase): + ''' + Tests for the SyncEngine object. + ''' + + def setUp(self): + self.ftp_local = FakeTaskProvider() + self.ftp_remote = FakeTaskProvider() + self.sync_engine = SyncEngine() + + def test_analyze_element_and_record_and_break_relationship(self): + ''' + Test for the _analyze_element, analyze_remote_id, analyze_local_id, + record_relationship, break_relationship + ''' + #adding a new local task + local_id = uuid.uuid4() + self.ftp_local.fake_add_task(local_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + #creating the related remote task + remote_id = uuid.uuid4() + self.ftp_remote.fake_add_task(remote_id) + #informing the sync_engine about that + self.sync_engine.record_relationship(local_id, remote_id, object()) + #verifying that it understood that + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.UPDATE, remote_id)) + self.assertEqual(self.sync_engine.analyze_remote_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.UPDATE, local_id)) + #and not the reverse + self.assertEqual(self.sync_engine.analyze_remote_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + self.assertEqual(self.sync_engine.analyze_local_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + #now we remove the remote task + self.ftp_remote.fake_remove_task(remote_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.REMOVE, None)) + self.sync_engine.break_relationship(local_id = local_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + self.assertEqual(self.sync_engine.analyze_remote_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + #we add them back and remove giving the remote id as key to find what to + #delete + self.ftp_local.fake_add_task(local_id) + self.ftp_remote.fake_add_task(remote_id) + self.ftp_remote.fake_remove_task(remote_id) + self.sync_engine.record_relationship(local_id, remote_id, object) + self.sync_engine.break_relationship(remote_id = remote_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + self.assertEqual(self.sync_engine.analyze_remote_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + + def test_syncability(self): + ''' + Test for the _analyze_element, analyze_remote_id, analyze_local_id. + Checks that the is_syncable parameter is used correctly + ''' + #adding a new local task unsyncable + local_id = uuid.uuid4() + self.ftp_local.fake_add_task(local_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + False), \ + (None, None)) + #adding a new local task, syncable + local_id = uuid.uuid4() + self.ftp_local.fake_add_task(local_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task), \ + (SyncEngine.ADD, None)) + #creating the related remote task + remote_id = uuid.uuid4() + self.ftp_remote.fake_add_task(remote_id) + #informing the sync_engine about that + self.sync_engine.record_relationship(local_id, remote_id, object()) + #checking that it behaves correctly with established relationships + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + True), \ + (SyncEngine.UPDATE, remote_id)) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + False), \ + (SyncEngine.LOST_SYNCABILITY, remote_id)) + self.assertEqual(self.sync_engine.analyze_remote_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + True), \ + (SyncEngine.UPDATE, local_id)) + self.assertEqual(self.sync_engine.analyze_remote_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + False), \ + (SyncEngine.LOST_SYNCABILITY, local_id)) + #now we remove the remote task + self.ftp_remote.fake_remove_task(remote_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + True), \ + (SyncEngine.REMOVE, None)) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + False), \ + (SyncEngine.REMOVE, None)) + self.sync_engine.break_relationship(local_id = local_id) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + True), \ + (SyncEngine.ADD, None)) + self.assertEqual(self.sync_engine.analyze_local_id(local_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + False), \ + (None, None)) + self.assertEqual(self.sync_engine.analyze_remote_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + True), \ + (SyncEngine.ADD, None)) + self.assertEqual(self.sync_engine.analyze_remote_id(remote_id, \ + self.ftp_local.has_task, self.ftp_remote.has_task, + False), \ + (None, None)) + +def test_suite(): + return unittest.TestLoader().loadTestsFromTestCase(TestSyncEngine) + + +class FakeTaskProvider(object): + + def __init__(self): + self.dic = {} + + def has_task(self, tid): + return self.dic.has_key(tid) + +############################################################################### +### Function with the fake_ prefix are here to assist in testing, they do not +### need to be present in the real class +############################################################################### + + def fake_add_task(self, tid): + self.dic[tid] = "something" + + def fake_get_task(self, tid): + return self.dic[tid] + + def fake_remove_task(self, tid): + del self.dic[tid] === added file 'GTG/tests/test_syncmeme.py' --- GTG/tests/test_syncmeme.py 1970-01-01 00:00:00 +0000 +++ GTG/tests/test_syncmeme.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +Tests for the SyncMeme class +''' + +import unittest +import datetime + +from GTG.backends.syncengine import SyncMeme + + + +class TestSyncMeme(unittest.TestCase): + ''' + Tests for the SyncEngine object. + ''' + + def test_which_is_newest(self): + ''' + test the which_is_newest function + + ''' + meme = SyncMeme() + #tasks have not changed + local_modified = datetime.datetime.now() + remote_modified = datetime.datetime.now() + meme.set_local_last_modified(local_modified) + meme.set_remote_last_modified(remote_modified) + self.assertEqual(meme.which_is_newest(local_modified, \ + remote_modified), None) + #we update the local + local_modified = datetime.datetime.now() + self.assertEqual(meme.which_is_newest(local_modified, \ + remote_modified), 'local') + #we update the remote + remote_modified = datetime.datetime.now() + self.assertEqual(meme.which_is_newest(local_modified, \ + remote_modified), 'remote') +def test_suite(): + return unittest.TestLoader().loadTestsFromTestCase(TestSyncMeme) + === added file 'GTG/tests/test_twokeydict.py' --- GTG/tests/test_twokeydict.py 1970-01-01 00:00:00 +0000 +++ GTG/tests/test_twokeydict.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +Tests for the TwoKeyDict class +''' + +import unittest +import uuid + +from GTG.tools.twokeydict import TwoKeyDict + + + +class TestTwoKeyDict(unittest.TestCase): + ''' + Tests for the TwoKeyDict object. + ''' + + + def test_add_and_gets(self): + ''' + Test for the __init__, _get_by_first, _get_by_second function + ''' + triplets = [(uuid.uuid4(), uuid.uuid4(), uuid.uuid4()) \ + for a in xrange(10)] + tw_dict = TwoKeyDict(*triplets) + for triplet in triplets: + self.assertEqual(tw_dict._get_by_primary(triplet[0]), triplet[2]) + self.assertEqual(tw_dict._get_by_secondary(triplet[1]), triplet[2]) + + def test_remove_by_first_or_second(self): + ''' + Test for removing triplets form the TwoKeyDict + ''' + triplet_first = (1, 'I', 'one') + triplet_second = (2, 'II', 'two') + tw_dict = TwoKeyDict(triplet_first, triplet_second) + tw_dict._remove_by_primary(triplet_first[0]) + tw_dict._remove_by_secondary(triplet_second[1]) + missing_first = 0 + missing_second = 0 + try: + tw_dict._get_by_primary(triplet_first[0]) + except KeyError: + missing_first += 1 + try: + tw_dict._get_by_secondary(triplet_second[0]) + except KeyError: + missing_first += 1 + try: + tw_dict._get_by_secondary(triplet_first[1]) + except KeyError: + missing_second += 1 + try: + tw_dict._get_by_secondary(triplet_second[1]) + except KeyError: + missing_second += 1 + self.assertEqual(missing_first, 2) + self.assertEqual(missing_second, 2) + #check for memory leaks + dict_len = 0 + for key in tw_dict._primary_to_value.iterkeys(): + dict_len += 1 + self.assertEqual(dict_len, 0) + + def test_get_primary_and_secondary_key(self): + ''' + Test for fetching the objects stored in the TwoKeyDict + ''' + triplets = [(uuid.uuid4(), uuid.uuid4(), uuid.uuid4()) \ + for a in xrange(10)] + tw_dict = TwoKeyDict(*triplets) + for triplet in triplets: + self.assertEqual(tw_dict._get_secondary_key(triplet[0]), \ + triplet[1]) + self.assertEqual(tw_dict._get_primary_key(triplet[1]), \ + triplet[0]) + +def test_suite(): + return unittest.TestLoader().loadTestsFromTestCase(TestTwoKeyDict) + === added file 'GTG/tools/bidict.py' --- GTG/tools/bidict.py 1970-01-01 00:00:00 +0000 +++ GTG/tools/bidict.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,112 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Getting Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + + + +class BiDict(object): + ''' + Bidirectional dictionary: the pairs stored can be accessed using either the + first or the second element as key (named key1 and key2). + You don't need this if there is no clash between the domains of the first + and second element of the pairs. + ''' + + def __init__(self, *pairs): + ''' + Initialization of the bidirectional dictionary + + @param pairs: optional. A list of pairs to add to the dictionary + ''' + super(BiDict, self).__init__() + self._first_to_second = {} + self._second_to_first = {} + for pair in pairs: + self.add(pair) + + def add(self, pair): + ''' + Adds a pair (key1, key2) to the dictionary + + @param pair: the pair formatted as (key1, key2) + ''' + self._first_to_second[pair[0]] = pair[1] + self._second_to_first[pair[1]] = pair[0] + + def _get_by_first(self, key): + ''' + Gets the key2 given key1 + + @param key: the first key + ''' + return self._first_to_second[key] + + def _get_by_second(self, key): + ''' + Gets the key1 given key2 + + @param key: the second key + ''' + return self._second_to_first[key] + + def _remove_by_first(self, first): + ''' + Removes a pair given the first key + + @param key: the first key + ''' + second = self._first_to_second[first] + del self._second_to_first[second] + del self._first_to_second[first] + + def _remove_by_second(self, second): + ''' + Removes a pair given the second key + + @param key: the second key + ''' + first = self._second_to_first[second] + del self._first_to_second[first] + del self._second_to_first[second] + + def _get_all_first(self): + ''' + Returns the list of all first keys + + @returns list + ''' + return list(self._first_to_second) + + def _get_all_second(self): + ''' + Returns the list of all second keys + + @returns list + ''' + return list(self._second_to_first) + + def __str__(self): + ''' + returns a string representing the content of this BiDict + + @returns string + ''' + return reduce(lambda text, keys: \ + str(text) + str(keys), + self._first_to_second.iteritems()) + === added file 'GTG/tools/twokeydict.py' --- GTG/tools/twokeydict.py 1970-01-01 00:00:00 +0000 +++ GTG/tools/twokeydict.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,135 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- + +''' +Contains TwoKeyDict, a Dictionary which also has a secondary key +''' + +from GTG.tools.bidict import BiDict + + + +class TwoKeyDict(object): + ''' + It's a standard Dictionary with a secondary key. + For example, you can add an element ('2', 'II', two'), where the + first two arguments are keys and the third is the stored object, and access + it as: + twokey['2'] ==> 'two' + twokey['II'] ==> 'two' + You can also request the other key, given one. + Function calls start with _ because you'll probably want to rename them when + you use this dictionary, for the sake of clarity. + ''' + + + def __init__(self, *triplets): + ''' + Creates the TwoKeyDict and optionally populates it with some data + + @oaram triplets: tuples for populating the TwoKeyDict. Format: + ((key1, key2, data_to_store), ...) + ''' + super(TwoKeyDict, self).__init__() + self._key_to_key_bidict = BiDict() + self._primary_to_value = {} + for triplet in triplets: + self.add(triplet) + + def add(self, triplet): + ''' + Adds a new triplet to the TwoKeyDict + + @param triplet: a tuple formatted like this: + (key1, key2, data_to_store) + ''' + self._key_to_key_bidict.add((triplet[0], triplet[1])) + self._primary_to_value[triplet[0]] = triplet[2] + + def _get_by_primary(self, primary): + ''' + Gets the stored data given the primary key + + @param primary: the primary key + @returns object: the stored object + ''' + return self._primary_to_value[primary] + + def _get_by_secondary(self, secondary): + ''' + Gets the stored data given the secondary key + + @param secondary: the primary key + @returns object: the stored object + ''' + primary = self._key_to_key_bidict._get_by_second(secondary) + return self._get_by_primary(primary) + + def _remove_by_primary(self, primary): + ''' + Removes a triplet given the rpimary key. + + @param primary: the primary key + ''' + del self._primary_to_value[primary] + self._key_to_key_bidict._remove_by_first(primary) + + def _remove_by_secondary(self, secondary): + ''' + Removes a triplet given the rpimary key. + + @param secondary: the primary key + ''' + primary = self._key_to_key_bidict._get_by_second(secondary) + self._remove_by_primary(primary) + + def _get_secondary_key(self, primary): + ''' + Gets the secondary key given the primary + + @param primary: the primary key + @returns object: the secondary key + ''' + return self._key_to_key_bidict._get_by_first(primary) + + def _get_primary_key(self, secondary): + ''' + Gets the primary key given the secondary + + @param secondary: the secondary key + @returns object: the primary key + ''' + return self._key_to_key_bidict._get_by_second(secondary) + + def _get_all_primary_keys(self): + ''' + Returns all primary keys + + @returns list: list of all primary keys + ''' + return self._key_to_key_bidict._get_all_first() + + def _get_all_secondary_keys(self): + ''' + Returns all secondary keys + + @returns list: list of all secondary keys + ''' + return self._key_to_key_bidict._get_all_second() + === added file 'GTG/tools/watchdog.py' --- GTG/tools/watchdog.py 1970-01-01 00:00:00 +0000 +++ GTG/tools/watchdog.py 2010-08-10 23:56:03 +0000 @@ -0,0 +1,53 @@ +# -*- coding: utf-8 -*- +# ----------------------------------------------------------------------------- +# Gettings Things Gnome! - a personal organizer for the GNOME desktop +# Copyright (c) 2008-2009 - Lionel Dricot & Bertrand Rousseau +# +# 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 3 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, see <http://www.gnu.org/licenses/>. +# ----------------------------------------------------------------------------- +import threading + +class Watchdog(object): + ''' + a simple thread-safe watchdog. + usage: + with Watchdod(timeout, error_function): + #do something + ''' + + def __init__(self, timeout, error_function): + ''' + Just sets the timeout and the function to execute when an error occours + + @param timeout: timeout in seconds + @param error_function: what to execute in case the watchdog timer + triggers + ''' + self.timeout = timeout + self.error_function = error_function + + def __enter__(self): + '''Starts the countdown''' + self.timer = threading.Timer(self.timeout, self.error_function) + self.timer.start() + + def __exit__(self, type, value, traceback): + '''Aborts the countdown''' + try: + self.timer.cancel() + except: + pass + if value == None: + return True + return False
_______________________________________________ Mailing list: https://launchpad.net/~gtg Post to : [email protected] Unsubscribe : https://launchpad.net/~gtg More help : https://help.launchpad.net/ListHelp
