← Back to branch summary

~ubuntu-branches/ubuntu/precise/ubuntuone-client/precise-201112142106

« back to all changes in this revision

Viewing changes to tests/platform/linux/eventlog/test_zg_listener.py

  • Committer: Bazaar Package Importer
  • Author(s): Rodney Dawes
  • Date: 2011-08-09 12:47:56 UTC
  • mfrom: (1.1.53 upstream)
  • Revision ID: james.westby@ubuntu.com-20110809124756-7nzilp3oix0a1yl9
Tags: 1.7.1-0ubuntu1
* New upstream release.
* debian/*:
  - Removed obsolete pycompat file
  - Removed ubuntuone-client-gnome deps and binary packaging, as it was
    moved out to separate project upstream.
  - Updated copyright to remove obsolete file reference
* debian/patches:
  - Removed patches which have been included upstream.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/platform/linux/eventlog/test_zg_listener.py
21
21
 
22
22
import logging
23
23
import os
24
 
import shutil
25
24
import sys
26
 
import unittest
27
25
import uuid
28
26
 
29
27
from os.path import basename
49
47
from contrib.testing.testcase import (
50
48
        FakeMain, FakeMainTestCase, BaseTwistedTestCase)
51
49
from ubuntuone.devtools.handlers import MementoHandler
52
 
from ubuntuone.platform.linux import get_udf_path
53
50
from ubuntuone.storageprotocol import client, delta
54
51
from ubuntuone.storageprotocol.request import ROOT
55
52
from ubuntuone.storageprotocol.sharersp import NotifyShareHolder
70
67
        INTERPRETATION_U1_CONTACT, URI_PROTOCOL_U1,
71
68
        STORAGE_DELETED, STORAGE_NETWORK, STORAGE_LOCAL)
72
69
from ubuntuone.syncdaemon.sync import Sync
73
 
from ubuntuone.syncdaemon.volume_manager import Share, Shared, UDF
 
70
from ubuntuone.syncdaemon.volume_manager import (
 
71
    get_udf_path, Share, Shared, UDF,
 
72
)
74
73
from tests.syncdaemon.test_action_queue import (
75
74
    ConnectedBaseTestCase,
76
75
    FakeSemaphore,
131
130
class ZeitgeistListenerTestCase(FakeMainTestCase):
132
131
    """Tests for ZeitgeistListener."""
133
132
 
 
133
    @defer.inlineCallbacks
134
134
    def setUp(self):
135
135
        """Initialize this instance."""
136
 
        super(ZeitgeistListenerTestCase, self).setUp()
 
136
        yield super(ZeitgeistListenerTestCase, self).setUp()
137
137
        self.patch(zglog, "ZeitgeistLogger", MockLogger)
138
138
        self.listener = ZeitgeistListener(self.fs, self.vm)
139
139
        self.event_q.subscribe(self.listener)
364
364
class ZeitgeistUDFsTestCase(ZeitgeistListenerTestCase):
365
365
    """Tests for all UDFs-related zeitgeist events."""
366
366
 
 
367
    @defer.inlineCallbacks
367
368
    def setUp(self):
368
369
        """Initialize this test instance."""
369
 
        super(ZeitgeistUDFsTestCase, self).setUp()
 
370
        yield super(ZeitgeistUDFsTestCase, self).setUp()
370
371
        self.home_dir = self.mktemp('ubuntuonehacker')
371
372
        self._old_home = os.environ['HOME']
372
373
        os.environ['HOME'] = self.home_dir
373
374
 
 
375
    @defer.inlineCallbacks
374
376
    def tearDown(self):
375
377
        """Finalize this test instance."""
376
 
        self.rmtree(self.home_dir)
377
378
        os.environ['HOME'] = self._old_home
378
 
        super(ZeitgeistUDFsTestCase, self).tearDown()
 
379
        yield super(ZeitgeistUDFsTestCase, self).tearDown()
379
380
 
380
381
    def _create_udf(self, id, node_id, suggested_path, subscribed=True):
381
382
        """Create an UDF and returns it and the volume."""
561
562
class ZeitgeistRemoteFileSyncTestCase(ConnectedBaseTestCase):
562
563
    """File sync events are logged into Zeitgeist."""
563
564
 
 
565
    @defer.inlineCallbacks
564
566
    def setUp(self):
565
567
        """Initialize this test instance."""
566
 
        ConnectedBaseTestCase.setUp(self)
 
568
        yield super(ZeitgeistRemoteFileSyncTestCase, self).setUp()
567
569
        self.rq = request_queue = RequestQueue(action_queue=self.action_queue)
568
570
        self.rq.transfers_semaphore = FakeSemaphore()
569
571
 
585
587
        self.root_id = "roootid"
586
588
        self.sync = Sync(main=self.main)
587
589
 
588
 
    def tearDown(self):
589
 
        """Finalize this test instance."""
590
 
        ConnectedBaseTestCase.tearDown(self)
591
 
 
592
590
    def test_syncdaemon_creates_file_on_server_is_logged(self):
593
591
        """Files created by SyncDaemon on the server are logged."""
594
592
        filename = "filename.mp3"
803
801
    """Zeitgeist events coming from the server."""
804
802
    timeout = 5
805
803
 
 
804
    @defer.inlineCallbacks
806
805
    def setUp(self):
807
806
        """Initialize this instance."""
808
 
        BaseTwistedTestCase.setUp(self)
 
807
        yield super(ZeitgeistLocalFileSyncTestCase, self).setUp()
809
808
        self.root = self.mktemp('root')
810
809
        self.shares = self.mktemp('shares')
811
810
        self.data = self.mktemp('data')
837
836
        self.listener = ZeitgeistListener(self.main.fs, self.main.vm)
838
837
        self.main.event_q.subscribe(self.listener)
839
838
 
 
839
    @defer.inlineCallbacks
840
840
    def tearDown(self):
841
841
        """Clean up this instance."""
842
842
        self._logger.removeHandler(self.handler)
843
843
        self.main.shutdown()
844
844
        FakeMain._sync_class = None
845
 
        shutil.rmtree(self.root)
846
 
        shutil.rmtree(self.shares)
847
 
        shutil.rmtree(self.data)
848
845
        for record in self.handler.records:
849
846
            exc_info = getattr(record, 'exc_info', None)
850
847
            if exc_info is not None:
851
848
                raise exc_info[0], exc_info[1], exc_info[2]
852
 
        BaseTwistedTestCase.tearDown(self)
 
849
        yield super(ZeitgeistLocalFileSyncTestCase, self).tearDown()
853
850
 
854
851
    @defer.inlineCallbacks
855
852
    def test_syncdaemon_creates_file_locally_is_logged(self):
1218
1215
        self.assertEqual(public_file.text, public_url)
1219
1216
        self.assertEqual(public_file.mimetype, "audio/mpeg")
1220
1217
        self.assertEqual(public_file.storage, STORAGE_DELETED)
1221
 
 
1222
 
 
1223
 
def test_suite():
1224
 
    """Collect these tests only on linux."""
1225
 
    import sys
1226
 
    if sys.platform == 'linux2':
1227
 
        tests = unittest.TestLoader().loadTestsFromName(__name__)
1228
 
    else:
1229
 
        tests = []
1230
 
    return tests