1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
#!/usr/bin/python
import os
import subprocess
import shutil
import configuration as cfg
from gi.repository import GObject
from gi.repository import GLib
import time
import options
from dconf import DConfClient
import helpers
# Add this after fixing the backup/restore and ontology changes tests
#"G_DEBUG" : "fatal_criticals",
TEST_ENV_DIRS = { "XDG_DATA_HOME" : os.path.join (cfg.TEST_TMP_DIR, "data"),
"XDG_CACHE_HOME": os.path.join (cfg.TEST_TMP_DIR, "cache")}
TEST_ENV_VARS = { "LC_COLLATE": "en_GB.utf8" }
EXTRA_DIRS = [os.path.join (cfg.TEST_TMP_DIR, "data", "tracker"),
os.path.join (cfg.TEST_TMP_DIR, "cache", "tracker")]
REASONABLE_TIMEOUT = 5
class UnableToBootException (Exception):
pass
class TrackerSystemAbstraction (object):
def __init__(self, settings=None, ontodir=None):
self.set_up_environment (settings=settings, ontodir=ontodir)
self.store = None
def set_up_environment (self, settings=None, ontodir=None):
"""
Sets up the XDG_*_HOME variables and make sure the directories exist
Settings should be a dict mapping schema names to dicts that hold the
settings that should be changed in those schemas. The contents dicts
should map key->value, where key is a key name and value is a suitable
GLib.Variant instance.
"""
helpers.log ("[Conf] Setting test environment...")
for var, directory in TEST_ENV_DIRS.iteritems ():
helpers.log ("export %s=%s" %(var, directory))
self.__recreate_directory (directory)
os.environ [var] = directory
for directory in EXTRA_DIRS:
self.__recreate_directory (directory)
if ontodir:
helpers.log ("export %s=%s" % ("TRACKER_DB_ONTOLOGIES_DIR", ontodir))
os.environ ["TRACKER_DB_ONTOLOGIES_DIR"] = ontodir
for var, value in TEST_ENV_VARS.iteritems ():
helpers.log ("export %s=%s" %(var, value))
os.environ [var] = value
# Previous loop should have set DCONF_PROFILE to the test location
if settings is not None:
self._apply_settings(settings)
helpers.log ("[Conf] environment ready")
def _apply_settings(self, settings):
for schema_name, contents in settings.iteritems():
dconf = DConfClient(schema_name)
dconf.reset()
for key, value in contents.iteritems():
dconf.write(key, value)
def tracker_store_testing_start (self, confdir=None, ontodir=None):
"""
Stops any previous instance of the store, calls set_up_environment,
and starts a new instances of the store
"""
self.set_up_environment (confdir, ontodir)
self.store = helpers.StoreHelper ()
self.store.start ()
def tracker_store_start (self):
self.store.start ()
def tracker_store_stop_nicely (self):
self.store.stop ()
def tracker_store_stop_brutally (self):
self.store.kill ()
def tracker_store_restart_with_new_ontologies (self, ontodir):
self.store.stop ()
if ontodir:
helpers.log ("[Conf] Setting %s - %s" % ("TRACKER_DB_ONTOLOGIES_DIR", ontodir))
os.environ ["TRACKER_DB_ONTOLOGIES_DIR"] = ontodir
try:
self.store.start ()
except GLib.Error:
raise UnableToBootException ("Unable to boot the store \n(" + str(e) + ")")
def tracker_store_prepare_journal_replay (self):
db_location = os.path.join (TEST_ENV_DIRS ['XDG_CACHE_HOME'], "tracker", "meta.db")
os.unlink (db_location)
lockfile = os.path.join (TEST_ENV_DIRS ['XDG_DATA_HOME'], "tracker", "data", ".ismeta.running")
f = open (lockfile, 'w')
f.write (" ")
f.close ()
def tracker_store_corrupt_dbs (self):
for filename in ["meta.db", "meta.db-wal"]:
db_path = os.path.join (TEST_ENV_DIRS ['XDG_CACHE_HOME'], "tracker", filename)
f = open (db_path, "w")
for i in range (0, 100):
f.write ("Some stupid content... hohohoho, not a sqlite file anymore!\n")
f.close ()
def tracker_store_remove_journal (self):
db_location = os.path.join (TEST_ENV_DIRS ['XDG_DATA_HOME'], "tracker", "data")
shutil.rmtree (db_location)
os.mkdir (db_location)
def tracker_store_remove_dbs (self):
db_location = os.path.join (TEST_ENV_DIRS ['XDG_CACHE_HOME'], "tracker")
shutil.rmtree (db_location)
os.mkdir (db_location)
def tracker_store_testing_stop (self):
"""
Stops a running tracker-store
"""
assert self.store
self.store.stop ()
def tracker_all_testing_start (self, confdir=None):
# This will start all miner-fs, store and writeback
self.tracker_writeback_testing_start (confdir)
def tracker_all_testing_stop (self):
# This will stop all miner-fs, store and writeback
self.tracker_writeback_testing_stop ()
def __recreate_directory (self, directory):
if (os.path.exists (directory)):
shutil.rmtree (directory)
os.makedirs (directory)
|
