« back to all changes in this revision
Viewing changes to tests/syncdaemon/test_fsm.py
- browse files at revision 1
- compare with another revision
- download diff
- download tarball
- view history from revision 1
- Committer: Bazaar Package Importer
- Author(s): Rodney Dawes
- Date: 2009-06-30 12:00:00 UTC
- Revision ID: james.westby@ubuntu.com-20090630120000-by806ovmw3193qe8
Tags: upstream-0.90.3
Import upstream version 0.90.3
- files added:
- apport
- bin
- contrib
- contrib/testing
- contrib/testing/dbus-session
- data
- data/icons
- data/icons/16x16
- data/icons/16x16/animations
- data/icons/16x16/apps
- data/icons/16x16/emblems
- data/icons/16x16/status
- data/icons/24x24
- data/icons/24x24/animations
- data/icons/24x24/apps
- data/icons/24x24/emblems
- data/icons/24x24/status
- data/icons/32x32
- data/icons/32x32/apps
- data/icons/32x32/emblems
- data/icons/32x32/status
- data/icons/48x48
- data/icons/48x48/animations
- data/icons/48x48/apps
- data/icons/48x48/emblems
- data/icons/48x48/status
- data/oauth_registration.d
- docs
- docs/man
- m4
- nautilus
- po
- tests
- tests/oauthdesktop
- tests/syncdaemon
- tests/syncdaemon/fsm
- tests/u1sync
- ubuntuone
- ubuntuone/oauthdesktop
- ubuntuone/syncdaemon
- ubuntuone/syncdaemon/fsm
- ubuntuone/u1sync
added
removed
tests/syncdaemon/test_fsm.py
1
#
2
# Author: Facundo Batista <facundo@canonical.com>
3
#
4
# Copyright 2009 Canonical Ltd.
5
#
6
# This program is free software: you can redistribute it and/or modify it
7
# under the terms of the GNU General Public License version 3, as published
8
# by the Free Software Foundation.
9
#
10
# This program is distributed in the hope that it will be useful, but
11
# WITHOUT ANY WARRANTY; without even the implied warranties of
12
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
13
# PURPOSE. See the GNU General Public License for more details.
14
#
15
# You should have received a copy of the GNU General Public License along
16
# with this program. If not, see <http://www.gnu.org/licenses/>.
17
18
'''Tests for the File System Manager.'''
19
20
from __future__ import with_statement
21
22
import os
23
import shutil
24
import unittest
25
import time
26
27
from contrib.testing.testcase import (
28
FakeVolumeManager,
29
)
30
31
from ubuntuone.syncdaemon.filesystem_manager import (
32
FileSystemManager, InconsistencyError)
33
from ubuntuone.syncdaemon.volume_manager import Share
34
35
TESTS_DIR = os.path.join(os.getcwd(), "tmp")
36
37
class FSMTestCase(unittest.TestCase):
38
""" Base test case for FSM """
39
40
def setUp(self):
41
""" Setup the test """
42
unittest.TestCase.setUp(self)
43
try:
44
os.mkdir(TESTS_DIR)
45
except OSError:
46
# already there, remove it to clean and create again
47
shutil.rmtree(TESTS_DIR)
48
os.mkdir(TESTS_DIR)
49
self.shares_dir = os.path.join(TESTS_DIR, 'shares')
50
os.makedirs(self.shares_dir)
51
self.root_dir = os.path.join(TESTS_DIR, 'root')
52
os.makedirs(self.root_dir)
53
self.fsmdir = os.path.join(TESTS_DIR, "fsmdir")
54
self.fsm = FileSystemManager(self.fsmdir,
55
FakeVolumeManager(self.root_dir))
56
self.share = self.create_share('share', 'share_name',
57
self.fsm, self.shares_dir)
58
self.share_path = self.share.path
59
60
def tearDown(self):
61
""" Clean up the tests. """
62
shutil.rmtree(TESTS_DIR)
63
64
@staticmethod
65
def create_share(share_id, share_name, fsm, shares_dir,
66
access_level='Modify'):
67
""" creates a share """
68
share_path = os.path.join(shares_dir, share_name)
69
os.makedirs(share_path)
70
share = Share(share_path, share_id=share_id,
71
access_level=access_level)
72
fsm.vm.add_share(share)
73
return share
74
75
76
class StartupTests(unittest.TestCase):
77
'''Test the basic startup behaviour.'''
78
79
def setUp(self):
80
""" Sets up a test. """
81
try:
82
os.mkdir(TESTS_DIR)
83
except OSError:
84
# already there, remove it to clean and create again
85
shutil.rmtree(TESTS_DIR)
86
os.mkdir(TESTS_DIR)
87
88
def tearDown(self):
89
'''Clean up the tests.'''
90
shutil.rmtree(TESTS_DIR)
91
92
def test_basic_startup(self):
93
'''Test the init interface.'''
94
# only one arg
95
self.assertRaises(TypeError, FileSystemManager)
96
self.assertRaises(TypeError, FileSystemManager, 1, 2)
97
98
# that creates the dir
99
fsmdir = os.path.join(TESTS_DIR, "a_fsmdir")
100
fsm = FileSystemManager(fsmdir, FakeVolumeManager(fsmdir))
101
self.assertTrue(os.path.exists(fsmdir))
102
103
def test_complex_startup(self):
104
'''Test startup after having data.'''
105
# open an empty one
106
fsmdir = os.path.join(TESTS_DIR, "fsmdir")
107
fsm = FileSystemManager(fsmdir, FakeVolumeManager(fsmdir))
108
share = FSMTestCase.create_share('share', 'share_name',
109
fsm, fsmdir)
110
self.assertEqual(fsm._idx_path, {})
111
self.assertEqual(fsm._idx_node_id, {})
112
113
# write some data, one with node_id
114
path1 = os.path.join(share.path, 'path1')
115
fsm.create(path1, "share")
116
created_mdid1 = fsm._idx_path[path1]
117
self.assertEqual(fsm._idx_path, {path1:created_mdid1})
118
fsm.set_node_id(path1, "uuid1")
119
self.assertEqual(fsm._idx_node_id, {("share","uuid1"):created_mdid1})
120
121
# ...and one without
122
path2 = os.path.join(share.path, 'path2')
123
fsm.create(path2, "share")
124
created_mdid2 = fsm._idx_path[path2]
125
self.assertEqual(fsm._idx_path,
126
{path1:created_mdid1, path2:created_mdid2})
127
128
# open a second one to see if everything is ok
129
fsm = FileSystemManager(fsmdir, fsm.vm)
130
self.assertEqual(fsm._idx_path,
131
{path1:created_mdid1, path2:created_mdid2})
132
self.assertEqual(fsm._idx_node_id, {("share","uuid1"):created_mdid1})
133
self.assertTrue(fsm.get_by_mdid(created_mdid1))
134
self.assertTrue(fsm.get_by_mdid(created_mdid2))
135
136
137
class CreationTests(FSMTestCase):
138
'''Test the creation behaviour.'''
139
140
def test_simple(self):
141
'''Test simple creation.'''
142
# create, but not twice
143
path = os.path.join(self.share.path, 'path')
144
self.fsm.create(path, "share")
145
self.assertRaises(ValueError, self.fsm.create, path, "share")
146
self.assertRaises(ValueError, self.fsm.create, path, "other")
147
mdobj = self.fsm.get_by_path(path)
148
self.assertEqual(mdobj.path, "path")
149
self.assertEqual(mdobj.share_id, "share")
150
when = mdobj.info.created
151
now = time.time()
152
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
153
154
# set uuid using valid path, but not twice
155
self.fsm.set_node_id(path, "uuid")
156
self.assertRaises(ValueError, self.fsm.set_node_id, path, "whatever")
157
mdobj = self.fsm.get_by_path(path)
158
when = mdobj.info.node_id_assigned
159
now = time.time()
160
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
161
162
def test_invalid_args(self):
163
'''Test using invalid args in set_node_id.'''
164
path = os.path.join(self.share.path, 'path')
165
self.fsm.create(path, "share")
166
167
# set uuid using an invalid path
168
self.assertRaises(KeyError, self.fsm.set_node_id, "no-path", "whatevr")
169
170
# set uuid using an invalid node_id
171
self.assertRaises(ValueError, self.fsm.set_node_id, path, None)
172
173
def test_no_twice(self):
174
'''Test that assignments must be done once, even in different FSMs.'''
175
# using the first FSM
176
path = os.path.join(self.share.path, 'path')
177
self.fsm.create(path, "share")
178
self.fsm.set_node_id(path, "uuid")
179
180
# opening another FSM
181
fsm = FileSystemManager(self.fsmdir, self.fsm.vm)
182
self.assertRaises(ValueError, fsm.create, path, "share")
183
self.assertRaises(ValueError, fsm.set_node_id, path, "whatever")
184
185
def test_fresh_metadata(self):
186
'''Initing with nothing in the metadata, it should leave it in v1.'''
187
fsm = FileSystemManager(self.fsmdir, self.fsm.vm)
188
md_version = open(os.path.join(self.fsmdir, "metadata_version")).read()
189
self.assertEqual(md_version, "2")
190
191
def test_old_metadata_None(self):
192
'''Initing with old metadata, it should leave it in v2.'''
193
# create some stuff
194
path = os.path.join(self.share.path, 'path')
195
open(path, "w").close()
196
mdid = self.fsm.create(path, "share")
197
self.fsm.set_node_id(path, "uuid")
198
199
# break the node on purpose
200
real_mdobj = self.fsm.fs[mdid]
201
del real_mdobj["stat"]
202
real_mdobj["path"] = unicode(real_mdobj["path"])
203
self.fsm.fs[mdid] = real_mdobj
204
205
# delete the version that should have left the previous fsm
206
version_file = os.path.join(self.fsmdir, "metadata_version")
207
os.remove(version_file)
208
209
# start up again, and check
210
newfsm = FileSystemManager(self.fsmdir, self.fsm.vm)
211
md_version = open(version_file).read()
212
self.assertEqual(md_version, "2")
213
newmdobj = newfsm.get_by_path(path)
214
self.assertEqual(newmdobj.mdid, mdid)
215
self.assertEqual(newmdobj.stat, os.stat(path))
216
self.assertTrue(isinstance(newmdobj.path, str))
217
218
def test_old_metadata_1(self):
219
'''Initing with old metadata, it should leave it in v2.'''
220
# create some stuff
221
path1 = os.path.join(self.share.path, 'path1')
222
path2 = os.path.join(self.share.path, 'path2')
223
mdid1 = self.fsm.create(path1, "share")
224
self.fsm.set_node_id(path1, "uuid1")
225
mdid2 = self.fsm.create(path2, "share")
226
self.fsm.set_node_id(path2, "uuid2")
227
228
# break the node on purpose, with unicode valid and not
229
real_mdobj = self.fsm.fs[mdid1]
230
real_mdobj["path"] = unicode(real_mdobj["path"])
231
self.fsm.fs[mdid1] = real_mdobj
232
real_mdobj = self.fsm.fs[mdid2]
233
real_mdobj["path"] = "asdas\x00\xff\xffasd"
234
self.fsm.fs[mdid2] = real_mdobj
235
236
# put the version file in 1
237
version_file = os.path.join(self.fsmdir, "metadata_version")
238
with open(version_file, "w") as fh:
239
fh.write("1")
240
241
# start up again, and check
242
newfsm = FileSystemManager(self.fsmdir, self.fsm.vm)
243
md_version = open(version_file).read()
244
self.assertEqual(md_version, "2")
245
newmdobj = newfsm.get_by_path(path1)
246
self.assertEqual(newmdobj.mdid, mdid1)
247
self.assertTrue(isinstance(newmdobj.path, str))
248
self.assertEqual(1, len(newfsm._idx_node_id))
249
250
251
252
class GetSetTests(FSMTestCase):
253
'''Test the get/set interface.'''
254
255
def test_bad_data(self):
256
'''No such info is allowed as path.'''
257
self.assertRaises(ValueError, self.fsm.create, "", "share")
258
self.assertRaises(ValueError, self.fsm.create, " ", "share")
259
260
def test_basic(self):
261
'''Test basic retrieval.'''
262
# write some data
263
path1 = os.path.join(self.share_path, "path1")
264
newmdid = self.fsm.create(path1, "share")
265
mdid1 = self.fsm._idx_path[path1]
266
self.fsm.set_node_id(path1, "uuid1")
267
self.assertEqual(newmdid, mdid1)
268
269
mdobj = self.fsm.get_by_mdid(mdid1)
270
self.assertEqual(mdobj.node_id, "uuid1")
271
self.assertEqual(mdobj.path, "path1")
272
self.assertEqual(mdobj.share_id, "share")
273
self.assertEqual(mdobj.local_hash, None)
274
self.assertEqual(mdobj.server_hash, None)
275
self.assertEqual(mdobj.info.is_partial, False)
276
self.assertEqual(mdobj.is_dir, False)
277
self.assertEqual(mdobj.mdid, mdid1)
278
self.assertEqual(self.fsm.get_by_node_id("share", "uuid1"), mdobj)
279
self.assertEqual(self.fsm.get_by_path(path1), mdobj)
280
281
# write more data
282
path2 = os.path.join(self.share_path, "path2")
283
newmdid = self.fsm.create(path2, "share", is_dir=True)
284
mdid2 = self.fsm._idx_path[path2]
285
self.fsm.set_node_id(path2, "uuid2")
286
self.assertEqual(newmdid, mdid2)
287
288
# check that is not mixed
289
mdobj = self.fsm.get_by_mdid(mdid1)
290
self.assertEqual(mdobj.node_id, "uuid1")
291
self.assertEqual(mdobj.path, "path1")
292
self.assertEqual(mdobj.share_id, "share")
293
self.assertEqual(mdobj.mdid, mdid1)
294
self.assertEqual(mdobj.is_dir, False)
295
self.assertEqual(self.fsm.get_by_node_id("share", "uuid1"), mdobj)
296
self.assertEqual(self.fsm.get_by_path(path1), mdobj)
297
mdobj = self.fsm.get_by_mdid(mdid2)
298
self.assertEqual(mdobj.node_id, "uuid2")
299
self.assertEqual(mdobj.path, "path2")
300
self.assertEqual(mdobj.share_id, "share")
301
self.assertEqual(mdobj.mdid, mdid2)
302
self.assertEqual(mdobj.is_dir, True)
303
self.assertEqual(self.fsm.get_by_node_id("share", "uuid2"), mdobj)
304
self.assertEqual(self.fsm.get_by_path(path2), mdobj)
305
306
def test_iteration(self):
307
'''Test basic retrieval.'''
308
# create a few objects
309
mdids = []
310
path_names = "path1 path2 path3".split()
311
paths = []
312
for path in path_names:
313
path = os.path.join(self.share.path, path)
314
paths.append(path)
315
mdid = self.fsm.create(path, "share")
316
mdids.append(mdid)
317
mdids.sort()
318
319
# get them
320
retrieved_mdids = []
321
retrieved_paths = []
322
for mdobj in self.fsm.get_mdobjs_by_share_id("share"):
323
retrieved_mdids.append(mdobj.mdid)
324
retrieved_paths.append(mdobj.path)
325
retrieved_mdids.sort()
326
retrieved_paths.sort()
327
328
# check them
329
self.assertEqual(mdids, retrieved_mdids)
330
self.assertEqual(path_names, retrieved_paths)
331
332
def test_getacopy(self):
333
'''Test that we receive only a copy.'''
334
# write some data
335
path = os.path.join(self.share_path, "path")
336
mdid = self.fsm.create(path, "share")
337
self.fsm.set_node_id(path, "uuid")
338
self.fsm.set_by_mdid(mdid, newarg="foo")
339
340
# test getting a copy with mdid
341
d = self.fsm.get_by_mdid(mdid)
342
d.newarg = "bar"
343
d = self.fsm.get_by_mdid(mdid)
344
self.assertEqual(d.newarg, "foo")
345
346
# test getting a copy with uuid
347
d = self.fsm.get_by_node_id("share", "uuid")
348
d.newarg = "bar"
349
d = self.fsm.get_by_node_id("share", "uuid")
350
self.assertEqual(d.newarg, "foo")
351
352
# test getting a copy with path
353
d = self.fsm.get_by_path(path)
354
d.newarg = "bar"
355
d = self.fsm.get_by_path(path)
356
self.assertEqual(d.newarg, "foo")
357
358
def test_get_raises(self):
359
'''Test that we get an exception if the object is not there.'''
360
# write some data
361
path = os.path.join(self.share_path, "path")
362
mdid = self.fsm.create(path, "share")
363
self.fsm.set_node_id(path, "uuid")
364
365
# with mdid, ok and bad
366
self.fsm.get_by_mdid(mdid)
367
self.assertRaises(KeyError, self.fsm.get_by_mdid, "no-such-key")
368
369
# with uuid, ok and bad
370
self.fsm.get_by_node_id("share", "uuid")
371
self.assertRaises(KeyError, self.fsm.get_by_node_id,
372
"share", "no-such-key")
373
self.assertRaises(ValueError, self.fsm.get_by_node_id,
374
"share", None)
375
376
# with path, ok and bad
377
self.fsm.get_by_path(path)
378
self.assertRaises(KeyError, self.fsm.get_by_path, "no-such-key")
379
380
def test_setters_simple(self):
381
'''Test that setters work.'''
382
# create some data
383
path = os.path.join(self.share.path, 'path')
384
mdid = self.fsm.create(path, "share")
385
self.fsm.set_node_id(path, "uuid")
386
387
# test using mdid
388
self.fsm.set_by_mdid(mdid, foo="foo1")
389
self.fsm.set_by_mdid(mdid, bar="bar1", baz="baz1")
390
mdobj = self.fsm.get_by_mdid(mdid)
391
self.assertEqual(mdobj.foo, "foo1")
392
self.assertEqual(mdobj.bar, "bar1")
393
self.assertEqual(mdobj.baz, "baz1")
394
395
# test using uuid
396
self.assertRaises(ValueError, self.fsm.set_by_node_id, None, "sh", f=3)
397
self.fsm.set_by_node_id("uuid", "share", foo="foo2")
398
self.fsm.set_by_node_id("uuid", "share", bar="bar2", baz="baz2")
399
mdobj = self.fsm.get_by_node_id("share", "uuid")
400
self.assertEqual(mdobj.foo, "foo2")
401
self.assertEqual(mdobj.bar, "bar2")
402
self.assertEqual(mdobj.baz, "baz2")
403
404
# test using path
405
self.fsm.set_by_path(path, foo="foo3")
406
self.fsm.set_by_path(path, bar="bar3", baz="baz3")
407
mdobj = self.fsm.get_by_path(path)
408
self.assertEqual(mdobj.foo, "foo3")
409
self.assertEqual(mdobj.bar, "bar3")
410
self.assertEqual(mdobj.baz, "baz3")
411
412
def test_setters_mixed(self):
413
'''Test the setters using different combinations.'''
414
# create some data
415
path = os.path.join(self.share.path, 'path')
416
mdid = self.fsm.create(path, "share")
417
self.fsm.set_node_id(path, "uuid")
418
419
# set with mdid, get with uuid and path
420
self.fsm.set_by_mdid(mdid, foo="foo1")
421
self.assertEqual(self.fsm.get_by_node_id("share", "uuid").foo, "foo1")
422
self.assertEqual(self.fsm.get_by_path(path).foo, "foo1")
423
424
# set with uuid, get with mdid and path
425
self.fsm.set_by_node_id("uuid", "share", foo="foo2")
426
self.assertEqual(self.fsm.get_by_mdid(mdid).foo, "foo2")
427
self.assertEqual(self.fsm.get_by_path(path).foo, "foo2")
428
429
# set with path, get with uuid and mdid
430
self.fsm.set_by_path(path, foo="foo3")
431
self.assertEqual(self.fsm.get_by_node_id("share", "uuid").foo, "foo3")
432
self.assertEqual(self.fsm.get_by_mdid(mdid).foo, "foo3")
433
434
def test_setters_raises(self):
435
'''Test that setters raise ok.'''
436
# create some data
437
path = os.path.join(self.share.path, 'path')
438
mdid = self.fsm.create(path, "share")
439
self.fsm.set_node_id(path, "uuid")
440
441
# test with bad values
442
self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "bad")
443
self.assertRaises(KeyError, self.fsm.get_by_mdid, "bad-value")
444
self.assertRaises(KeyError, self.fsm.get_by_path, "bad-value")
445
446
def test_setting_forbidden_values(self):
447
'''Test trying to set forbidden values.'''
448
# create some data
449
path = os.path.join(self.share.path, 'path')
450
mdid = self.fsm.create(path, "share")
451
self.fsm.set_node_id(path, "uuid")
452
453
# test with forbidden uuid
454
self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, node_id="-")
455
self.assertRaises(ValueError, self.fsm.set_by_path, path, node_id="")
456
457
# test with forbidden path
458
self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
459
path="-")
460
self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, path="-")
461
462
# test with forbidden info
463
self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
464
info="-")
465
self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, info="-")
466
self.assertRaises(ValueError, self.fsm.set_by_path, path, info="-")
467
468
# test with forbidden stat
469
self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
470
stat="-")
471
self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, stat="-")
472
self.assertRaises(ValueError, self.fsm.set_by_path, path, stat="-")
473
474
# test with forbidden share
475
self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, share_id="-")
476
self.assertRaises(ValueError, self.fsm.set_by_path, path, share_id="-")
477
478
# test with forbidden mdid
479
self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
480
mdid="-")
481
self.assertRaises(ValueError, self.fsm.set_by_path, path, mdid="-")
482
483
# test with forbidden is_dir
484
self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, is_dir="-")
485
self.assertRaises(ValueError, self.fsm.set_by_path, path, is_dir="-")
486
self.assertRaises(ValueError,
487
self.fsm.set_by_node_id, "uuid", "share", is_dir="-")
488
489
def test_setting_forbidden_mixed(self):
490
'''Test that when trying with forbidden, nothing happens at all.'''
491
# create some data
492
path = os.path.join(self.share.path, 'path')
493
mdid = self.fsm.create(path, "share")
494
self.fsm.set_node_id(path, "uuid")
495
496
# test with mixed stuff
497
self.assertRaises(TypeError, self.fsm.set_by_node_id, info="n", foo="")
498
self.assertRaises(TypeError, self.fsm.set_by_mdid, path="nop", bar="?")
499
self.assertRaises(TypeError, self.fsm.set_by_path, node_id="n", baz="")
500
501
# see that it still is unchanged
502
mdobj = self.fsm.get_by_mdid(mdid)
503
self.assertEqual(mdobj.path, "path")
504
self.assertEqual(mdobj.node_id, "uuid")
505
506
def test_get_root(self):
507
""" Test that the root of a share is stored properly. """
508
# write some data
509
mdid = self.fsm.create(self.share_path, "share")
510
self.fsm.set_node_id(self.share_path, "uuid")
511
512
# with path, ok and bad
513
self.fsm.get_by_path(self.share_path)
514
self.assertRaises(KeyError, self.fsm.get_by_path, "no-such-key")
515
516
# check the path stored in the mdobj
517
mdobj = self.fsm.get_by_node_id("share", "uuid")
518
self.assertEquals(self.share_path, mdobj.path)
519
520
def test_get_all_by_share(self):
521
""" Test that it returns all the mdids in a share. """
522
# create the shares
523
share1 = self.create_share('share_id1', 'share_name1', self.fsm,
524
self.shares_dir, access_level='View')
525
share2 = self.create_share('share_id2', 'share_name2', self.fsm,
526
self.shares_dir, access_level='View')
527
mdid1 = self.fsm.create(share1.path, "share_id1", is_dir=True)
528
self.fsm.set_node_id(share1.path, "uuid1")
529
mdid2 = self.fsm.create(share2.path, "share_id2", is_dir=True)
530
self.fsm.set_node_id(share2.path, "uuid2")
531
532
# create some nodes in share 1
533
path3 = os.path.join(share1.path, "a")
534
mdid3 = self.fsm.create(path3, "share_id1", is_dir=True)
535
self.fsm.set_node_id(path3, "uuid3")
536
path4 = os.path.join(share1.path, "a", "b")
537
mdid4 = self.fsm.create(path4, "share_id1")
538
self.fsm.set_node_id(path4, "uuid4")
539
path5 = os.path.join(share1.path, "c")
540
mdid5 = self.fsm.create(path5, "share_id1")
541
self.fsm.set_node_id(path5, "uuid5")
542
543
# create some nodes in share 2
544
path6 = os.path.join(share2.path, "a")
545
mdid6 = self.fsm.create(path6, "share_id2", is_dir=True)
546
self.fsm.set_node_id(path6, "uuid6")
547
path7 = os.path.join(share2.path, "c")
548
mdid7 = self.fsm.create(path7, "share_id2")
549
self.fsm.set_node_id(path7, "uuid7")
550
551
# tricky: node without node_id yet
552
path8 = os.path.join(share2.path, "d")
553
mdid8 = self.fsm.create(path8, "share_id2")
554
555
556
# get them
557
all = set()
558
for mdobj in self.fsm.get_mdobjs_by_share_id("share_id1"):
559
all.add(mdobj.mdid)
560
self.assertTrue(mdid3 in all)
561
self.assertTrue(mdid4 in all)
562
self.assertTrue(mdid5 in all)
563
self.assertTrue(mdid6 not in all)
564
self.assertTrue(mdid7 not in all)
565
self.assertTrue(mdid8 not in all)
566
567
all = set()
568
for mdobj in self.fsm.get_mdobjs_by_share_id("share_id2"):
569
all.add(mdobj.mdid)
570
self.assertTrue(mdid3 not in all)
571
self.assertTrue(mdid4 not in all)
572
self.assertTrue(mdid5 not in all)
573
self.assertTrue(mdid6 in all)
574
self.assertTrue(mdid7 in all)
575
self.assertTrue(mdid8 in all)
576
577
def test_get_all_by_share_mixed(self):
578
"""Test that it returns all the mdids in a share with mixed nodes."""
579
# create the shares
580
share = self.create_share('share_id', 'sharetest', self.fsm,
581
self.shares_dir, access_level='View')
582
self.fsm.create(share.path, "share_id", is_dir=True)
583
self.fsm.set_node_id(share.path, "uuid")
584
585
# create one real node...
586
path1 = os.path.join(share.path, "a")
587
mdid1 = self.fsm.create(path1, "share_id", is_dir=True)
588
self.fsm.set_node_id(path1, "uuid1")
589
590
# ...and two without node_id's
591
path2 = os.path.join(share.path, "b")
592
mdid2 = self.fsm.create(path2, "share_id")
593
path3 = os.path.join(share.path, "c")
594
mdid3 = self.fsm.create(path3, "share_id")
595
596
# get them
597
all = set()
598
for mdobj in self.fsm.get_mdobjs_by_share_id("share_id"):
599
all.add(mdobj.mdid)
600
self.assertTrue(mdid1 in all)
601
self.assertTrue(mdid2 in all)
602
self.assertTrue(mdid3 in all)
603
604
605
class StatTests(FSMTestCase):
606
'''Test all the behaviour regarding the stats.'''
607
608
def create_node(self, name):
609
'''Creates a node.'''
610
path = os.path.join(self.share.path, name)
611
mdid = self.fsm.create(path, self.share.id)
612
self.fsm.set_node_id(path, "uuid1")
613
mdobj = self.fsm.get_by_mdid(mdid)
614
return mdobj
615
616
def test_create_nofile(self):
617
'''Test creation when there's no file.'''
618
mdobj = self.create_node("foo")
619
self.assertEqual(mdobj.stat, None)
620
621
def test_create_file(self):
622
'''Test creation when there's a file.'''
623
# file
624
path = os.path.join(self.share.path, "thisfile")
625
open(path, "w").close()
626
mdobj = self.create_node("thisfile")
627
self.assertEqual(mdobj.stat, os.stat(path))
628
629
# dir
630
path = os.path.join(self.share.path, "thisdir")
631
os.mkdir(path)
632
mdobj = self.create_node("thisdir")
633
self.assertEqual(mdobj.stat, os.stat(path))
634
635
def test_refresh(self):
636
'''Test that refresh_stat works.'''
637
path = os.path.join(self.share.path, "thisfile")
638
open(path, "w").close()
639
mdobj = self.create_node("thisfile")
640
mdid = mdobj.mdid
641
oldstat = os.stat(path)
642
self.assertEqual(mdobj.stat, oldstat)
643
644
# touch the file, it's not automagically updated
645
open(path, "w").close()
646
mdobj = self.fsm.get_by_mdid(mdid)
647
self.assertEqual(mdobj.stat, oldstat)
648
649
# it's updated when asked
650
self.fsm.refresh_stat(path)
651
mdobj = self.fsm.get_by_mdid(mdid)
652
self.assertEqual(mdobj.stat, os.stat(path))
653
654
def test_commit_partial(self):
655
'''Test that it's updated in the commit.'''
656
path = os.path.join(self.share.path, "thisfile")
657
open(path, "w").close()
658
mdobj = self.create_node("thisfile")
659
mdid = mdobj.mdid
660
oldstat = os.stat(path)
661
self.assertEqual(mdobj.stat, oldstat)
662
663
# create a partial
664
self.fsm.create_partial(mdobj.node_id, mdobj.share_id)
665
fh = self.fsm.get_partial_for_writing(mdobj.node_id, mdobj.share_id)
666
fh.write("foobar")
667
fh.close()
668
mdobj = self.fsm.get_by_mdid(mdid)
669
self.assertEqual(mdobj.stat, oldstat)
670
671
# commit the partial
672
self.fsm.commit_partial(mdobj.node_id, mdobj.share_id, "localhash")
673
mdobj = self.fsm.get_by_mdid(mdid)
674
self.assertEqual(mdobj.stat, os.stat(path))
675
676
def test_move(self):
677
'''Test that move refreshes stat.'''
678
path1 = os.path.join(self.share.path, "thisfile1")
679
path2 = os.path.join(self.share.path, "thisfile2")
680
open(path1, "w").close()
681
mdobj = self.create_node(path1)
682
self.assertEqual(mdobj.stat, os.stat(path1))
683
684
# move
685
self.fsm.move_file("share", path1, path2)
686
687
# check
688
mdobj = self.fsm.get_by_path(path2)
689
self.assertEqual(mdobj.stat, os.stat(path2))
690
691
def test_move_overwriting(self):
692
'''Test that move refreshes stat when overwrites other file.'''
693
path1 = os.path.join(self.share.path, "thisfile1")
694
path2 = os.path.join(self.share.path, "thisfile2")
695
open(path1, "w").close()
696
open(path2, "w").close()
697
mdobj1 = self.create_node(path1)
698
mdobj2 = self.create_node(path2)
699
self.assertEqual(mdobj1.stat, os.stat(path1))
700
self.assertEqual(mdobj2.stat, os.stat(path2))
701
702
# move
703
self.fsm.move_file("share", path1, path2)
704
705
# check
706
self.assertRaises(KeyError, self.fsm.get_by_path, path1)
707
mdobj2 = self.fsm.get_by_path(path2)
708
self.assertEqual(mdobj2.stat, os.stat(path2))
709
710
def test_update_stat(self):
711
'''Test that update_stat works.'''
712
path = os.path.join(self.share.path, "thisfile")
713
open(path, "w").close()
714
mdobj = self.create_node("thisfile")
715
mdid = mdobj.mdid
716
oldstat = os.stat(path)
717
self.assertEqual(mdobj.stat, oldstat)
718
719
# touch the file, it's not automagically updated
720
open(path, "w").close()
721
mdobj = self.fsm.get_by_mdid(mdid)
722
self.assertEqual(mdobj.stat, oldstat)
723
724
# it's updated when asked, even if it's an old stat
725
self.fsm.update_stat(mdid, oldstat)
726
mdobj = self.fsm.get_by_mdid(mdid)
727
self.assertEqual(mdobj.stat, oldstat)
728
729
730
class PartialTests(FSMTestCase):
731
'''Test all the .partial nitty gritty.'''
732
733
def test_create_file(self):
734
'''Test create .partial for a file.'''
735
testfile = os.path.join(self.share_path, "path")
736
mdid = self.fsm.create(testfile, "share")
737
self.fsm.set_node_id(testfile, "uuid")
738
739
# create partial ok
740
self.fsm.create_partial("uuid", "share")
741
self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
742
self.assertTrue(os.path.exists(testfile + ".partial"))
743
mdobj = self.fsm.get_by_mdid(mdid)
744
when = mdobj.info.last_partial_created
745
now = time.time()
746
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
747
748
# invalid uuid
749
self.assertRaises(KeyError, self.fsm.create_partial, "foo", "share")
750
self.assertRaises(ValueError, self.fsm.create_partial, None, "share")
751
752
# already has a partial!
753
self.assertRaises(ValueError, self.fsm.create_partial, "uuid", "share")
754
755
def test_commit_file(self):
756
'''Test commit the .partial for a file, after a successful download.'''
757
testfile = os.path.join(self.share_path, "path")
758
mdid = self.fsm.create(testfile, "share")
759
self.fsm.set_node_id(testfile, "uuid")
760
761
# create partial
762
self.fsm.create_partial("uuid", "share")
763
self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
764
with open(testfile + ".partial", "w") as fh:
765
fh.write("test info!")
766
767
# commit partial, and check that the file is moved, and metadata is ok
768
self.fsm.commit_partial("uuid", "share", local_hash=9876)
769
self.assertFalse(os.path.exists(testfile + ".partial"))
770
with open(testfile) as fh:
771
in_file = fh.read()
772
self.assertEqual(in_file, "test info!")
773
mdobj = self.fsm.get_by_mdid(mdid)
774
self.assertFalse(mdobj.info.is_partial)
775
self.assertEqual(mdobj.local_hash, 9876)
776
when = mdobj.info.last_downloaded
777
now = time.time()
778
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
779
780
# invalid uuid
781
self.assertRaises(KeyError, self.fsm.commit_partial,
782
"foo", "share", 123)
783
self.assertRaises(ValueError, self.fsm.commit_partial,
784
None, "share", 123)
785
# it has no partial!
786
self.assertRaises(ValueError, self.fsm.commit_partial,
787
"uuid", "share", 1)
788
789
def test_remove_file(self):
790
'''Test removing the .partial for a file, because a bad download.'''
791
testfile = os.path.join(self.share_path, "path")
792
mdid = self.fsm.create(testfile, "share")
793
self.fsm.set_node_id(testfile, "uuid")
794
795
# create partial
796
self.fsm.create_partial("uuid", "share")
797
self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
798
with open(testfile + ".partial", "w") as fh:
799
fh.write("test info!")
800
with open(testfile, "w") as fh:
801
fh.write("previous stuff!")
802
803
# remove partial, and check that the file is gone, and metadata is ok
804
self.fsm.remove_partial("uuid", "share")
805
self.assertFalse(os.path.exists(testfile + ".partial"))
806
with open(testfile) as fh:
807
in_file = fh.read()
808
self.assertEqual(in_file, "previous stuff!")
809
mdobj = self.fsm.get_by_mdid(mdid)
810
self.assertFalse(mdobj.info.is_partial)
811
when = mdobj.info.last_partial_removed
812
now = time.time()
813
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
814
815
# invalid uuid
816
self.assertRaises(KeyError, self.fsm.remove_partial, "foo", "share")
817
self.assertRaises(ValueError, self.fsm.remove_partial, None, "share")
818
819
# it has no partial!
820
self.fsm.remove_partial("uuid", "share")
821
822
823
def test_create_dir_previous(self):
824
'''Test create .partial for a dir when the dir existed.'''
825
testdir = os.path.join(self.share_path, "path")
826
mdid = self.fsm.create(testdir, "share", is_dir=True)
827
self.fsm.set_node_id(testdir, "uuid")
828
os.mkdir(testdir)
829
830
# create partial ok
831
self.fsm.create_partial("uuid", "share")
832
self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
833
self.assertTrue(os.path.exists(os.path.join(testdir, ".partial")))
834
mdobj = self.fsm.get_by_mdid(mdid)
835
when = mdobj.info.last_partial_created
836
now = time.time()
837
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
838
839
# invalid uuid
840
self.assertRaises(KeyError, self.fsm.create_partial, "foo", "share")
841
842
# already has a partial!
843
self.assertRaises(ValueError, self.fsm.create_partial, "uuid", "share")
844
845
def test_create_dir_notprevious(self):
846
'''Test create .partial for a dir when the dir didn't exist.'''
847
testdir = os.path.join(self.share_path, "path")
848
mdid = self.fsm.create(testdir, "share", is_dir=True)
849
self.fsm.set_node_id(testdir, "uuid")
850
851
# create partial ok
852
self.fsm.create_partial("uuid", "share")
853
self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
854
self.assertTrue(os.path.exists(testdir))
855
self.assertTrue(os.path.exists(os.path.join(testdir, ".partial")))
856
mdobj = self.fsm.get_by_mdid(mdid)
857
when = mdobj.info.last_partial_created
858
now = time.time()
859
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
860
861
# invalid uuid
862
self.assertRaises(KeyError, self.fsm.create_partial, "foo", "share")
863
864
# already has a partial!
865
self.assertRaises(ValueError, self.fsm.create_partial, "uuid", "share")
866
867
def test_commit_dir(self):
868
'''Test commit the .partial for a dir, after a successful download.'''
869
testdir = os.path.join(self.share_path, "path")
870
mdid = self.fsm.create(testdir, "share", is_dir=True)
871
self.fsm.set_node_id(testdir, "uuid")
872
873
# create partial
874
self.fsm.create_partial("uuid", "share")
875
self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
876
877
# commit is forbidden for directories
878
self.assertRaises(ValueError, self.fsm.commit_partial, "uuid", "share",
879
local_hash=9876)
880
881
def test_remove_dir(self):
882
'''Test removing the .partial for a dir, because a bad download.'''
883
testdir = os.path.join(self.share_path, "path")
884
mdid = self.fsm.create(testdir, "share", is_dir=True)
885
self.fsm.set_node_id(testdir, "uuid")
886
887
# create partial
888
self.fsm.create_partial("uuid", "share")
889
self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
890
891
# remove partial, and check that the file is gone, and metadata is ok
892
self.fsm.remove_partial("uuid", "share")
893
self.assertFalse(os.path.exists(os.path.join(testdir, ".partial")))
894
mdobj = self.fsm.get_by_mdid(mdid)
895
self.assertFalse(mdobj.info.is_partial)
896
when = mdobj.info.last_partial_removed
897
now = time.time()
898
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
899
900
# invalid uuid
901
self.assertRaises(KeyError, self.fsm.remove_partial, "foo", "share")
902
903
# it has no partial!
904
self.fsm.remove_partial("uuid", "share")
905
906
907
class FileHandlingTests(FSMTestCase):
908
'''Test the file handling services.'''
909
910
def test_move_to_conflict(self):
911
'''Test that the conflict stuff works.'''
912
testfile = os.path.join(self.share_path, "path")
913
mdid = self.fsm.create(testfile, "share")
914
self.fsm.set_node_id(testfile, "uuid")
915
with open(testfile, "w") as fh:
916
fh.write("test!")
917
918
# move first time
919
self.fsm.move_to_conflict(mdid)
920
self.assertFalse(os.path.exists(testfile))
921
with open(testfile + ".conflict") as fh:
922
in_file = fh.read()
923
self.assertEqual(in_file, "test!")
924
mdobj = self.fsm.get_by_mdid(mdid)
925
when = mdobj.info.last_conflicted
926
now = time.time()
927
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
928
929
# move second time, start the .N serie
930
with open(testfile, "w") as fh:
931
fh.write("test 1!")
932
self.fsm.move_to_conflict(mdid)
933
self.assertFalse(os.path.exists(testfile))
934
with open(testfile + ".conflict.1") as fh:
935
in_file = fh.read()
936
self.assertEqual(in_file, "test 1!")
937
938
# create a few more, test a higher one
939
open(testfile + ".conflict.2", "w").close()
940
open(testfile + ".conflict.3", "w").close()
941
open(testfile + ".conflict.4", "w").close()
942
open(testfile + ".conflict.5", "w").close()
943
with open(testfile, "w") as fh:
944
fh.write("test 6!")
945
self.fsm.move_to_conflict(mdid)
946
self.assertFalse(os.path.exists(testfile))
947
with open(testfile + ".conflict.6") as fh:
948
in_file = fh.read()
949
self.assertEqual(in_file, "test 6!")
950
951
# invalid uuid
952
self.assertRaises(KeyError, self.fsm.move_to_conflict, "no-such-mdid")
953
954
def test_upload_finished(self):
955
'''Test upload finished.'''
956
path = os.path.join(self.share_path, "path")
957
mdid = self.fsm.create(path, "share")
958
self.fsm.set_node_id(path, "uuid")
959
960
# finish the upload!
961
self.fsm.upload_finished(mdid, server_hash=1234567890)
962
mdobj = self.fsm.get_by_mdid(mdid)
963
self.assertEqual(mdobj.server_hash, 1234567890)
964
when = mdobj.info.last_uploaded
965
now = time.time()
966
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
967
968
# invalid mdid
969
self.assertRaises(KeyError, self.fsm.upload_finished,
970
"no-such-mdid", 123)
971
972
# bad arguments
973
self.assertRaises(TypeError, self.fsm.upload_finished, mdid)
974
975
def test_move_file_withfile(self):
976
'''Test that a file is moved from one point to the other.'''
977
testfile = os.path.join(self.share_path, "path")
978
mdid = self.fsm.create(testfile, "share")
979
self.fsm.set_node_id(testfile, "uuid")
980
with open(testfile, "w") as fh:
981
fh.write("test!")
982
983
# move the file
984
to_path = os.path.join(self.share_path, "path2")
985
self.fsm.move_file("share", testfile, to_path)
986
self.assertFalse(os.path.exists(testfile))
987
with open(to_path) as fh:
988
in_file = fh.read()
989
self.assertEqual(in_file, "test!")
990
mdobj = self.fsm.get_by_mdid(mdid)
991
self.assertEqual(mdobj.info.last_moved_from, testfile)
992
when = mdobj.info.last_moved_time
993
now = time.time()
994
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
995
996
# move again, to a directory
997
from_path = to_path
998
os.mkdir(os.path.join(self.share_path, "testdir"))
999
to_path = os.path.join(self.share_path, "testdir", "path3")
1000
self.fsm.move_file("share", from_path, to_path)
1001
self.assertFalse(os.path.exists(from_path))
1002
with open(to_path) as fh:
1003
in_file = fh.read()
1004
self.assertEqual(in_file, "test!")
1005
1006
# invalid path
1007
self.assertRaises(KeyError, self.fsm.move_file,
1008
"share", "no-path", "dest")
1009
1010
# other share
1011
self.assertRaises(KeyError, self.fsm.move_file,
1012
"othershare", testfile, to_path)
1013
1014
# invalid args
1015
self.assertRaises(TypeError, self.fsm.move_file, "one-path")
1016
1017
def test_move_file_overwrite(self):
1018
'''Test that a file is moved over other one.'''
1019
testfile1 = os.path.join(self.share_path, "path1")
1020
mdid1 = self.fsm.create(testfile1, "share")
1021
self.fsm.set_node_id(testfile1, "uuid1")
1022
with open(testfile1, "w") as fh:
1023
fh.write("test 1")
1024
1025
testfile2 = os.path.join(self.share_path, "path2")
1026
mdid2 = self.fsm.create(testfile2, "share")
1027
self.fsm.set_node_id(testfile2, "uuid2")
1028
with open(testfile2, "w") as fh:
1029
fh.write("test 2")
1030
1031
# move the file
1032
self.fsm.move_file("share", testfile1, testfile2)
1033
self.assertFalse(os.path.exists(testfile1))
1034
with open(testfile2) as fh:
1035
in_file = fh.read()
1036
self.assertEqual(in_file, "test 1")
1037
mdobj = self.fsm.get_by_mdid(mdid1)
1038
self.assertEqual(mdobj.path, "path2")
1039
mdobj = self.fsm.get_by_path(testfile2)
1040
self.assertEqual(mdobj.mdid, mdid1)
1041
1042
# check that the info for the overwritten one is gone
1043
self.assertRaises(KeyError, self.fsm.get_by_mdid, mdid2)
1044
self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "uuid2")
1045
1046
def test_move_file_withdir(self):
1047
'''Test that a dir is moved from one point to the other.'''
1048
from_path = os.path.join(self.share_path, "path")
1049
mdid = self.fsm.create(from_path, "share", is_dir=True)
1050
self.fsm.set_node_id(from_path, "uuid")
1051
1052
# move the file
1053
os.mkdir(from_path)
1054
to_path = os.path.join(self.share_path, "path2")
1055
self.fsm.move_file("share", from_path, to_path)
1056
self.assertFalse(os.path.exists(from_path))
1057
self.assertTrue(os.path.exists(to_path))
1058
mdobj = self.fsm.get_by_mdid(mdid)
1059
self.assertEqual(mdobj.info.last_moved_from, from_path)
1060
when = mdobj.info.last_moved_time
1061
now = time.time()
1062
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
1063
1064
# move again, to a directory
1065
from_path = to_path
1066
os.mkdir(os.path.join(self.share_path, "testdir"))
1067
to_path = os.path.join(self.share_path, "testdir", "path3")
1068
self.fsm.move_file("share", from_path, to_path)
1069
self.assertFalse(os.path.exists(from_path))
1070
self.assertTrue(os.path.exists(to_path))
1071
mdobj = self.fsm.get_by_mdid(mdid)
1072
self.assertEqual(mdobj.info.last_moved_from, from_path)
1073
when = mdobj.info.last_moved_time
1074
now = time.time()
1075
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
1076
1077
def test_move_file_withfulldir(self):
1078
'''Test that a dir is moved from even having a file inside.'''
1079
# the containing dir
1080
from_path = os.path.join(self.share_path, "path")
1081
mdid = self.fsm.create(from_path, "share", is_dir=True)
1082
self.fsm.set_node_id(from_path, "uuid")
1083
os.mkdir(from_path)
1084
1085
# the file outside, with a similar name just to confuse
1086
otherfile = os.path.join(self.share_path, "pa")
1087
self.fsm.create(otherfile, "share", is_dir=False)
1088
self.fsm.set_node_id(otherfile, "otheruuid")
1089
open(otherfile, "w").close()
1090
1091
# the file inside
1092
filepath = os.path.join(from_path, "file.txt")
1093
fileid = self.fsm.create(filepath, "share", is_dir=False)
1094
self.fsm.set_node_id(filepath, "fileuuid")
1095
open(filepath, "w").close()
1096
1097
# move the dir
1098
to_path = os.path.join(self.share_path, "path2")
1099
self.fsm.move_file("share", from_path, to_path)
1100
self.assertFalse(os.path.exists(from_path))
1101
self.assertTrue(os.path.exists(to_path))
1102
mdobj = self.fsm.get_by_mdid(mdid)
1103
self.assertEqual(mdobj.info.last_moved_from, from_path)
1104
when = mdobj.info.last_moved_time
1105
now = time.time()
1106
self.assertTrue(now-3 <= when <= now) # 3 seconds test range
1107
1108
# check that file inside is ok
1109
newfilepath = os.path.join(to_path, "file.txt")
1110
self.assertFalse(os.path.exists(filepath))
1111
self.assertTrue(os.path.exists(newfilepath))
1112
mdobj = self.fsm.get_by_path(newfilepath)
1113
self.assertEqual(mdobj.mdid, fileid)
1114
self.assertEqual(mdobj.path, "path2/file.txt")
1115
1116
# check the outer file
1117
self.assertTrue(os.path.exists(otherfile))
1118
mdobj = self.fsm.get_by_path(otherfile)
1119
self.assertEqual(mdobj.path, "pa")
1120
1121
def test_delete_file(self):
1122
'''Test that a file is deleted.'''
1123
testfile = os.path.join(self.share_path, "path")
1124
open(testfile, "w").close()
1125
mdid = self.fsm.create(testfile, "share")
1126
self.fsm.set_node_id(testfile, "uuid")
1127
1128
# delete the file
1129
self.fsm.delete_file(testfile)
1130
self.assertFalse(os.path.exists(testfile))
1131
self.assertRaises(KeyError, self.fsm.get_by_mdid, mdid)
1132
self.assertRaises(KeyError, self.fsm.get_by_path, testfile)
1133
self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "uuid")
1134
1135
def test_delete_dir(self):
1136
'''Test that a dir is deleted.'''
1137
testdir = os.path.join(self.share.path, "path")
1138
os.mkdir(testdir)
1139
mdid = self.fsm.create(testdir, "share", is_dir=True)
1140
self.fsm.set_node_id(testdir, "uuid")
1141
1142
# try to delete the dir, but has files on it
1143
open(os.path.join(testdir, "foo"), "w").close()
1144
self.assertRaises(OSError, self.fsm.delete_file, testdir)
1145
self.assertEqual(self.fsm.get_by_mdid(mdid).path, "path")
1146
self.assertEqual(self.fsm.get_by_path(testdir).path, "path")
1147
self.assertEqual(self.fsm.get_by_node_id("share", "uuid").path, "path")
1148
os.remove(os.path.join(testdir, "foo"))
1149
1150
# really delete the dir
1151
self.fsm.delete_file(testdir)
1152
self.assertFalse(os.path.exists(testdir))
1153
self.assertRaises(KeyError, self.fsm.get_by_mdid, mdid)
1154
self.assertRaises(KeyError, self.fsm.get_by_path, testdir)
1155
self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "uuid")
1156
1157
1158
class SyntheticInfoTests(FSMTestCase):
1159
'''Test the methods that generates attributes.'''
1160
1161
def test_has_metadata(self):
1162
'''Test the has_metadata option.'''
1163
# not yet
1164
self.assertFalse(self.fsm.has_metadata(path="path"))
1165
self.assertFalse(self.fsm.has_metadata(node_id="uuid",
1166
share_id="share"))
1167
self.assertFalse(self.fsm.has_metadata(mdid="garbage"))
1168
1169
# path created
1170
path = os.path.join(self.share.path, 'path')
1171
mdid = self.fsm.create(path, "share")
1172
self.assertTrue(self.fsm.has_metadata(path=path))
1173
self.assertFalse(self.fsm.has_metadata(node_id="uuid",
1174
share_id="share"))
1175
self.assertTrue(self.fsm.has_metadata(mdid=mdid))
1176
1177
# uuid set
1178
self.fsm.set_node_id(path, "uuid")
1179
self.assertTrue(self.fsm.has_metadata(path=path))
1180
self.assertTrue(self.fsm.has_metadata(node_id="uuid",
1181
share_id="share"))
1182
self.assertTrue(self.fsm.has_metadata(mdid=mdid))
1183
self.assertRaises(ValueError, self.fsm.has_metadata, node_id=None,
1184
share_id="share")
1185
1186
def test_is_dir(self):
1187
'''Test the is_directory option.'''
1188
# standard file
1189
testfiledir = os.path.join(self.share_path, "path1")
1190
mdid = self.fsm.create(testfiledir, "share", is_dir=False)
1191
self.fsm.set_node_id(testfiledir, "uuid1")
1192
self.assertFalse(self.fsm.is_dir(path=testfiledir))
1193
self.assertFalse(self.fsm.is_dir(node_id="uuid1", share_id="share"))
1194
self.assertFalse(self.fsm.is_dir(mdid=mdid))
1195
1196
# directory
1197
testfiledir = os.path.join(self.share_path, "path2")
1198
mdid = self.fsm.create(testfiledir, "share", is_dir=True)
1199
self.fsm.set_node_id(testfiledir, "uuid2")
1200
self.assertTrue(self.fsm.is_dir(path=testfiledir))
1201
self.assertTrue(self.fsm.is_dir(node_id="uuid2", share_id="share"))
1202
self.assertTrue(self.fsm.is_dir(mdid=mdid))
1203
1204
def test_changed_server(self):
1205
'''Test the changed option when in SERVER state.'''
1206
# SERVER means: local_hash != server_hash and is_partial == False
1207
testfile = os.path.join(self.share_path, "path")
1208
mdid = self.fsm.create(testfile, "share")
1209
self.fsm.set_node_id(testfile, "uuid")
1210
1211
# set conditions and test
1212
self.fsm.set_by_mdid(mdid, server_hash=98765)
1213
# local_hash is None so far
1214
self.assertTrue(self.fsm.changed(mdid=mdid), "SERVER")
1215
self.assertTrue(self.fsm.changed(node_id="uuid", share_id="share"),
1216
"SERVER")
1217
self.assertTrue(self.fsm.changed(path=testfile), "SERVER")
1218
1219
# put a .partial by hand, to see it crash
1220
open(testfile + ".partial", "w").close()
1221
self.assertRaises(InconsistencyError,
1222
self.fsm._check_partial, mdid=mdid)
1223
1224
def test_changed_none(self):
1225
'''Test the changed option when in NONE state.'''
1226
# NONE means: local_hash == server_hash and is_partial == False
1227
testfile = os.path.join(self.share_path, "path")
1228
mdid = self.fsm.create(testfile, "share")
1229
self.fsm.set_node_id(testfile, "uuid")
1230
1231
# all conditions are set: by default, local_hash and server_hash
1232
# are both None
1233
self.assertTrue(self.fsm.changed(mdid=mdid), "NONE")
1234
self.assertTrue(self.fsm.changed(node_id="uuid", share_id="share"),
1235
"NONE")
1236
self.assertTrue(self.fsm.changed(path=testfile), "NONE")
1237
1238
# put a .partial by hand, to see it crash
1239
open(testfile + ".partial", "w").close()
1240
self.assertRaises(InconsistencyError,
1241
self.fsm._check_partial, mdid=mdid)
1242
1243
def test_changed_local(self):
1244
'''Test the changed option when in LOCAL state.'''
1245
# LOCAL means: local_hash != server_hash and is_partial == True
1246
testfile = os.path.join(self.share_path, "path")
1247
mdid = self.fsm.create(testfile, "share")
1248
self.fsm.set_node_id(testfile, "uuid")
1249
1250
# set conditions and test
1251
self.fsm.set_by_mdid(mdid, server_hash=98765)
1252
# local_hash is None so far
1253
self.fsm.create_partial("uuid", "share")
1254
self.assertTrue(self.fsm.changed(mdid=mdid), "LOCAL")
1255
self.assertTrue(self.fsm.changed(node_id="uuid", share_id="share"),
1256
"LOCAL")
1257
self.assertTrue(self.fsm.changed(path=testfile), "LOCAL")
1258
1259
# remove the .partial by hand, to see it crash
1260
os.remove(testfile + ".partial")
1261
self.assertRaises(InconsistencyError,
1262
self.fsm._check_partial, mdid=mdid)
1263
1264
def test_dir_content(self):
1265
'''Test the dir_content method.'''
1266
# create a structure in md
1267
to_create = []
1268
dir1 = os.path.join(self.share_path, "foo")
1269
to_create.append((dir1, True))
1270
to_create.append((os.path.join(dir1, "file2"), False))
1271
to_create.append((os.path.join(dir1, "file1"), False))
1272
1273
dir2 = os.path.join(dir1, "bar")
1274
to_create.append((dir2, True))
1275
to_create.append((os.path.join(dir2, "file3"), False))
1276
to_create.append((os.path.join(dir2, "file5"), False))
1277
to_create.append((os.path.join(dir2, "file4"), False))
1278
to_create.append((os.path.join(dir2, "file6"), False))
1279
1280
dir3 = os.path.join(dir2, "baz")
1281
to_create.append((dir3, True))
1282
to_create.append((os.path.join(dir3, "file7"), False))
1283
to_create.append((os.path.join(dir3, "file9"), False))
1284
to_create.append((os.path.join(dir3, "file8"), False))
1285
1286
dir4 = os.path.join(dir2, "other")
1287
to_create.append((dir4, True))
1288
1289
for i, (path, is_dir) in enumerate(to_create):
1290
self.fsm.create(path, "share", is_dir=is_dir)
1291
self.fsm.set_node_id(path, "uuid" + str(i))
1292
1293
# ask for the info for dir1
1294
should_be = [
1295
("bar", True, "uuid3"),
1296
("file1", False, "uuid2"),
1297
("file2", False, "uuid1"),
1298
]
1299
content = self.fsm.dir_content(dir1)
1300
self.assertEqual(should_be, content)
1301
1302
# ask for the info for dir2
1303
should_be = [
1304
("baz", True, "uuid8"),
1305
("file3", False, "uuid4"),
1306
("file4", False, "uuid6"),
1307
("file5", False, "uuid5"),
1308
("file6", False, "uuid7"),
1309
("other", True, "uuid12"),
1310
]
1311
content = self.fsm.dir_content(dir2)
1312
self.assertEqual(should_be, content)
1313
1314
# ask for the info for dir3
1315
should_be = [
1316
("file7", False, "uuid9"),
1317
("file8", False, "uuid11"),
1318
("file9", False, "uuid10"),
1319
]
1320
content = self.fsm.dir_content(dir3)
1321
self.assertEqual(should_be, content)
1322
1323
# ask for the info for an empty dir
1324
content = self.fsm.dir_content(dir4)
1325
self.assertEqual([], content)
1326
1327
# ask for the info for an inexistant dir
1328
self.assertRaises(KeyError, self.fsm.dir_content, "no-such-dir")
1329
1330
# ask for the info for file
1331
just_a_file = os.path.join(dir3, "file9")
1332
self.assertRaises(ValueError, self.fsm.dir_content, just_a_file)
1333
1334
1335
class SharesTests(FSMTestCase):
1336
""" Tests fsm with ro and rw shares. """
1337
1338
def tearDown(self):
1339
""" cleanup the mess """
1340
for dirpath, dirs, files in os.walk(TESTS_DIR):
1341
for dir in dirs:
1342
os.chmod(os.path.join(dirpath, dir), 0777)
1343
for file in files:
1344
os.chmod(os.path.join(dirpath, file), 0666)
1345
shutil.rmtree(TESTS_DIR)
1346
1347
def test_file_ro_share_fail(self):
1348
""" Test that manual creation of a file, fails on a ro-share. """
1349
share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1350
self.shares_dir, access_level='View')
1351
testfile = os.path.join(share.path, "a_file")
1352
self.assertRaises(IOError, open, testfile, 'w')
1353
1354
def test_dir_ro_share(self):
1355
""" Test that the creation of a file using fsm, works on a ro-share."""
1356
share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1357
self.shares_dir, access_level='View')
1358
testdir = os.path.join(share.path, "path2")
1359
mdid = self.fsm.create(testdir, share.id, is_dir=True)
1360
self.fsm.set_node_id(testdir, "uuid2")
1361
self.fsm.create_partial('uuid2', share.id)
1362
fd = self.fsm.get_partial_for_writing('uuid2', share.id)
1363
fd.flush()
1364
fd.close()
1365
self.assertTrue(os.path.exists(testdir))
1366
1367
def test_file_ro_share(self):
1368
""" Test that the creation of a file using fsm, works on a ro-share."""
1369
self.share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1370
self.shares_dir, access_level='View')
1371
testfile = os.path.join(self.share.path, "a_file")
1372
file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
1373
self.fsm.set_node_id(testfile, "uuid3")
1374
self.fsm.create_partial('uuid3', self.share.id)
1375
fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
1376
fd.flush()
1377
fd.close()
1378
self.fsm.commit_partial('uuid3', self.share.id, None)
1379
self.assertTrue(os.path.exists(testfile))
1380
1381
def test_delete_dir_ro_share(self):
1382
""" Test that fsm is able to delete a dir in a ro.share. """
1383
share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1384
self.shares_dir, access_level='View')
1385
testdir = os.path.join(share.path, "path2")
1386
mdid = self.fsm.create(testdir, share.id, is_dir=True)
1387
self.fsm.set_node_id(testdir, "uuid2")
1388
self.fsm.create_partial('uuid2', share.id)
1389
fd = self.fsm.get_partial_for_writing('uuid2', share.id)
1390
fd.flush()
1391
fd.close()
1392
self.fsm.remove_partial('uuid2', share.id)
1393
self.assertTrue(os.path.exists(testdir))
1394
self.fsm.delete_file(testdir)
1395
self.assertFalse(os.path.exists(testdir))
1396
1397
def test_delete_file_ro_share(self):
1398
""" Test that fsm is able to delete a file in a ro-share. """
1399
self.share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1400
self.shares_dir, access_level='View')
1401
testfile = os.path.join(self.share.path, "a_file")
1402
file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
1403
self.fsm.set_node_id(testfile, "uuid3")
1404
self.fsm.create_partial('uuid3', self.share.id)
1405
fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
1406
fd.flush()
1407
fd.close()
1408
self.fsm.commit_partial('uuid3', self.share.id, None)
1409
self.assertTrue(os.path.exists(testfile))
1410
self.fsm.delete_file(testfile)
1411
self.assertFalse(os.path.exists(testfile))
1412
1413
def test_move_to_conflict_ro_share(self):
1414
""" Test that fsm is able to handle move_to_conflict in a ro-share. """
1415
self.share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1416
self.shares_dir, access_level='View')
1417
testfile = os.path.join(self.share.path, "a_file")
1418
file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
1419
self.fsm.set_node_id(testfile, "uuid3")
1420
self.fsm.create_partial('uuid3', self.share.id)
1421
fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
1422
fd.flush()
1423
fd.close()
1424
self.fsm.commit_partial('uuid3', self.share.id, None)
1425
self.assertTrue(os.path.exists(testfile))
1426
self.fsm.move_to_conflict(file_mdid)
1427
self.assertTrue(os.path.exists(testfile + ".conflict"))
1428
1429
1430
def test_file_rw_share_no_fail(self):
1431
""" Test that manual creation of a file, ona rw-share. """
1432
share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1433
self.shares_dir)
1434
testfile = os.path.join(share.path, "a_file")
1435
open(testfile, 'w').close()
1436
self.assertTrue(os.path.exists(testfile))
1437
1438
def test_dir_rw_share(self):
1439
""" Test that the creation of a file using fsm, works on a rw-share."""
1440
share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1441
self.shares_dir)
1442
testdir = os.path.join(share.path, "path2")
1443
mdid = self.fsm.create(testdir, share.id, is_dir=True)
1444
self.fsm.set_node_id(testdir, "uuid2")
1445
self.fsm.create_partial('uuid2', share.id)
1446
fd = self.fsm.get_partial_for_writing('uuid2', share.id)
1447
fd.flush()
1448
fd.close()
1449
self.assertTrue(os.path.exists(testdir))
1450
1451
def test_file_rw_share(self):
1452
""" Test that the creation of a file using fsm, works on a rw-share."""
1453
self.share = self.create_share('ro_share', 'ro_share_name', self.fsm,
1454
self.shares_dir)
1455
testfile = os.path.join(self.share.path, "a_file")
1456
file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
1457
self.fsm.set_node_id(testfile, "uuid3")
1458
self.fsm.create_partial('uuid3', self.share.id)
1459
fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
1460
fd.flush()
1461
fd.close()
1462
self.fsm.commit_partial('uuid3', self.share.id, None)
1463
self.assertTrue(os.path.exists(testfile))
1464
1465
1466
def test_share_and_root(self):
1467
""" Test the creation of a file with the same relative path in a share
1468
and in the root.
1469
"""
1470
a_dir_root = os.path.join(self.root_dir, "a_dir")
1471
a_dir_share = os.path.join(self.share.path, "a_dir")
1472
root_mdid = self.fsm.create(a_dir_root, "", is_dir=True)
1473
self.fsm.set_node_id(a_dir_root, "uuid1")
1474
share_mdid = self.fsm.create(a_dir_share, self.share.id, is_dir=True)
1475
self.fsm.set_node_id(a_dir_share, "uuid2")
1476
self.fsm.create_partial('uuid1', "")
1477
fd = self.fsm.get_partial_for_writing('uuid1', "")
1478
fd.flush()
1479
fd.close()
1480
self.fsm.create_partial('uuid2', self.share.id)
1481
fd = self.fsm.get_partial_for_writing('uuid2', self.share.id)
1482
fd.flush()
1483
fd.close()
1484
self.assertTrue(os.path.exists(self.fsm.get_abspath("", a_dir_root)))
1485
self.assertTrue(os.path.exists(a_dir_share))
1486
1487
def test_suite():
1488
# pylint: disable-msg=C0111
1489
return unittest.TestLoader().loadTestsFromName(__name__)
1490
1491
if __name__ == "__main__":
1492
unittest.main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.1