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

« back to all changes in this revision

Viewing changes to .pc/02_value_io_error_fix.patch/tests/platform/linux/eventlog/test_zglog.py

  • Committer: Package Import Robot
  • Author(s): Rodney Dawes
  • Date: 2011-12-21 15:46:25 UTC
  • mfrom: (1.1.56)
  • Revision ID: package-import@ubuntu.com-20111221154625-ujvunri4frsecj2k
Tags: 2.99.0-0ubuntu1
* New upstream release.
  - Verify timestamp to avoid invalid auth failures (LP: #692597)
  - Files in new UDFs not uploaded due to filtering (LP: #869920)
* debian/patches:
  - Remove upstreamed patches

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# -*- coding: utf-8 -*-
2
 
#
3
 
# Author: Alejandro J. Cura <alecu@canonical.com>
4
 
#
5
 
# Copyright 2010 Canonical Ltd.
6
 
#
7
 
# This program is free software: you can redistribute it and/or modify it
8
 
# under the terms of the GNU General Public License version 3, as published
9
 
# by the Free Software Foundation.
10
 
#
11
 
# This program is distributed in the hope that it will be useful, but
12
 
# WITHOUT ANY WARRANTY; without even the implied warranties of
13
 
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
14
 
# PURPOSE.  See the GNU General Public License for more details.
15
 
#
16
 
# You should have received a copy of the GNU General Public License along
17
 
# with this program.  If not, see <http://www.gnu.org/licenses/>.
18
 
"""Tests for the Zeitgeist logging."""
19
 
 
20
 
import os
21
 
import shutil
22
 
import subprocess
23
 
import tempfile
24
 
import time
25
 
 
26
 
from distutils.spawn import find_executable
27
 
 
28
 
from twisted.internet.defer import Deferred
29
 
from zeitgeist.client import ZeitgeistClient
30
 
from zeitgeist.datamodel import Event, Subject, Interpretation, Manifestation
31
 
 
32
 
from tests.platform.linux.test_dbus import DBusTwistedTestCase
33
 
from ubuntuone.eventlog.zglog import ZeitgeistLogger
34
 
 
35
 
SRCDIR = os.environ.get('SRCDIR', os.getcwd())
36
 
 
37
 
 
38
 
class NotFoundError(Exception):
39
 
    """Not found error."""
40
 
 
41
 
 
42
 
class ZeitgeistNotStartedTests(DBusTwistedTestCase):
43
 
    """Tests for the zeitgeist logging module."""
44
 
 
45
 
    def test_log_does_not_err_when_daemon_not_started(self):
46
 
        """zeitgeist-daemon was not started."""
47
 
        timestamp = int(time.time() * 1000)
48
 
        subject = Subject.new_for_values(
49
 
                uri="file:///tmp/folder1",
50
 
                interpretation=Interpretation.FOLDER,
51
 
                manifestation=Manifestation.FILE_DATA_OBJECT,
52
 
                origin="ubuntuone:uuid",
53
 
                mimetype="inode/directory",
54
 
                text="sample folder"
55
 
        )
56
 
        sample_event = Event.new_for_values(
57
 
            timestamp=timestamp,
58
 
            interpretation=Interpretation.ACCESS_EVENT,
59
 
            manifestation=Manifestation.USER_ACTIVITY,
60
 
            actor="mailto:sample_subject",
61
 
            subjects=[subject]
62
 
        )
63
 
 
64
 
        zg = ZeitgeistLogger()
65
 
        d = zg.log(sample_event)
66
 
        self.assertEqual(zg.client, None)
67
 
 
68
 
        def verify(result):
69
 
            """Stored result is the empty list, because zg not available."""
70
 
            self.assertEqual(result, [])
71
 
            return result
72
 
 
73
 
        d.addCallback(verify)
74
 
        return d
75
 
 
76
 
 
77
 
def wait_zeitgeist_started(seconds=10):
78
 
    """Wait a few seconds until zg is started, or fail if it can't."""
79
 
    client = None
80
 
    count = 0
81
 
    while client is None:
82
 
        count += 1
83
 
        try:
84
 
            client = ZeitgeistClient()
85
 
            break
86
 
        except RuntimeError:
87
 
            if count > seconds*10:
88
 
                raise
89
 
            time.sleep(0.1)
90
 
 
91
 
 
92
 
class ZeitgeistTestCase(DBusTwistedTestCase):
93
 
    """Tests for the zeitgeist logging module."""
94
 
 
95
 
    def setUp(self):
96
 
        super(ZeitgeistTestCase, self).setUp()
97
 
        zgdaemon = find_executable("zeitgeist-daemon")
98
 
        if not zgdaemon:
99
 
            raise NotFoundError("zeitgeist-daemon was not found.")
100
 
 
101
 
        tempfolder = tempfile.mkdtemp(prefix="test-u1-zeitgeist-")
102
 
        tempstdout = tempfile.TemporaryFile(prefix="test-u1-stdout-")
103
 
        tempstderr = tempfile.TemporaryFile(prefix="test-u1-stderr-")
104
 
        os.environ["ZEITGEIST_DATA_PATH"] = tempfolder
105
 
        p = subprocess.Popen([zgdaemon], bufsize=4096, stdout=tempstdout,
106
 
                             stderr=tempstderr)
107
 
        def cleanup():
108
 
            """Wait for the process to finish."""
109
 
            p.terminate()
110
 
            p.wait()
111
 
            del(os.environ["ZEITGEIST_DATA_PATH"])
112
 
            shutil.rmtree(tempfolder)
113
 
 
114
 
        wait_zeitgeist_started()
115
 
        self.addCleanup(cleanup)
116
 
 
117
 
    def test_log_records_the_event(self):
118
 
        """The log method records the event in zg."""
119
 
        timestamp = int(time.time() * 1000)
120
 
        subject = Subject.new_for_values(
121
 
                uri="file:///tmp/folder1",
122
 
                interpretation=Interpretation.FOLDER,
123
 
                manifestation=Manifestation.FILE_DATA_OBJECT,
124
 
                origin="ubuntuone:uuid",
125
 
                mimetype="inode/directory",
126
 
                text="sample folder"
127
 
        )
128
 
        sample_event = Event.new_for_values(
129
 
            timestamp=timestamp,
130
 
            interpretation=Interpretation.ACCESS_EVENT,
131
 
            manifestation=Manifestation.USER_ACTIVITY,
132
 
            actor="mailto:sample_subject",
133
 
            subjects=[subject]
134
 
        )
135
 
 
136
 
        sample_template = Event.new_for_values()
137
 
 
138
 
        zg = ZeitgeistLogger()
139
 
        self.assertNotEqual(zg.client, None)
140
 
 
141
 
        d2 = Deferred()
142
 
 
143
 
        def logged(id_list):
144
 
            """The event was logged to zeitgeist."""
145
 
 
146
 
            def events_found(event_list):
147
 
                """zg returned the list of events."""
148
 
                self.assertEqual(event_list[0].id, id_list[0])
149
 
                d2.callback("ok")
150
 
 
151
 
            zg.client.find_events_for_template(sample_template, events_found)
152
 
 
153
 
        d = zg.log(sample_event)
154
 
        d.addCallbacks(logged, d2.errback)
155
 
 
156
 
        return d2