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, [])
76
def wait_zeitgeist_started(seconds=10):
77
"""Wait a few seconds until zg is started, or fail if it can't."""
83
client = ZeitgeistClient()
86
if count > seconds*10:
91
class ZeitgeistTestCase(DBusTwistedTestCase):
92
"""Tests for the zeitgeist logging module."""
95
super(ZeitgeistTestCase, self).setUp()
96
zgdaemon = find_executable("zeitgeist-daemon")
98
raise NotFoundError("zeitgeist-daemon was not found.")
100
tempfolder = tempfile.mkdtemp(prefix="test-u1-zeitgeist-")
101
tempstdout = tempfile.TemporaryFile(prefix="test-u1-stdout-")
102
tempstderr = tempfile.TemporaryFile(prefix="test-u1-stderr-")
103
os.environ["ZEITGEIST_DATA_PATH"] = tempfolder
104
p = subprocess.Popen([zgdaemon], bufsize=4096, stdout=tempstdout,
107
"""Wait for the process to finish."""
110
del(os.environ["ZEITGEIST_DATA_PATH"])
111
shutil.rmtree(tempfolder)
113
wait_zeitgeist_started()
114
self.addCleanup(cleanup)
116
def test_log_records_the_event(self):
117
"""The log method records the event in zg."""
118
timestamp = int(time.time() * 1000)
119
subject = Subject.new_for_values(
120
uri="file:///tmp/folder1",
121
interpretation=Interpretation.FOLDER,
122
manifestation=Manifestation.FILE_DATA_OBJECT,
123
origin="ubuntuone:uuid",
124
mimetype="inode/directory",
127
sample_event = Event.new_for_values(
129
interpretation=Interpretation.ACCESS_EVENT,
130
manifestation=Manifestation.USER_ACTIVITY,
131
actor="mailto:sample_subject",
135
sample_template = Event.new_for_values()
137
zg = ZeitgeistLogger()
138
self.assertNotEqual(zg.client, None)
143
"""The event was logged to zeitgeist."""
145
def events_found(event_list):
146
"""zg returned the list of events."""
147
self.assertEqual(event_list[0].id, id_list[0])
150
zg.client.find_events_for_template(sample_template, events_found)
152
d = zg.log(sample_event)
153
d.addCallbacks(logged, d2.errback)
added
removed
tests/platform/linux/eventlog/test_zglog.py
