1
# -*- coding: utf-8 -*-
3
# Author: Alejandro J. Cura <alecu@canonical.com>
5
# Copyright 2010 Canonical Ltd.
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.
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.
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."""
26
from distutils.spawn import find_executable
28
from twisted.internet.defer import Deferred
29
from zeitgeist.client import ZeitgeistClient
30
from zeitgeist.datamodel import Event, Subject, Interpretation, Manifestation
32
from tests.platform.linux.test_dbus import DBusTwistedTestCase
33
from ubuntuone.eventlog.zglog import ZeitgeistLogger
35
SRCDIR = os.environ.get('SRCDIR', os.getcwd())
38
class NotFoundError(Exception):
39
"""Not found error."""
42
class ZeitgeistNotStartedTests(DBusTwistedTestCase):
43
"""Tests for the zeitgeist logging module."""
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",
56
sample_event = Event.new_for_values(
58
interpretation=Interpretation.ACCESS_EVENT,
59
manifestation=Manifestation.USER_ACTIVITY,
60
actor="mailto:sample_subject",
64
zg = ZeitgeistLogger()
65
d = zg.log(sample_event)
66
self.assertEqual(zg.client, None)
69
"""Stored result is the empty list, because zg not available."""
70
self.assertEqual(result, [])
77
def wait_zeitgeist_started(seconds=10):
78
"""Wait a few seconds until zg is started, or fail if it can't."""
84
client = ZeitgeistClient()
87
if count > seconds*10:
92
class ZeitgeistTestCase(DBusTwistedTestCase):
93
"""Tests for the zeitgeist logging module."""
96
super(ZeitgeistTestCase, self).setUp()
97
zgdaemon = find_executable("zeitgeist-daemon")
99
raise NotFoundError("zeitgeist-daemon was not found.")
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,
108
"""Wait for the process to finish."""
111
del(os.environ["ZEITGEIST_DATA_PATH"])
112
shutil.rmtree(tempfolder)
114
wait_zeitgeist_started()
115
self.addCleanup(cleanup)
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",
128
sample_event = Event.new_for_values(
130
interpretation=Interpretation.ACCESS_EVENT,
131
manifestation=Manifestation.USER_ACTIVITY,
132
actor="mailto:sample_subject",
136
sample_template = Event.new_for_values()
138
zg = ZeitgeistLogger()
139
self.assertNotEqual(zg.client, None)
144
"""The event was logged to zeitgeist."""
146
def events_found(event_list):
147
"""zg returned the list of events."""
148
self.assertEqual(event_list[0].id, id_list[0])
151
zg.client.find_events_for_template(sample_template, events_found)
153
d = zg.log(sample_event)
154
d.addCallbacks(logged, d2.errback)