« back to all changes in this revision
Viewing changes to tests/syncdaemon/test_fsm.py
- browse files at revision 1.1.36
- compare with another revision
- download diff
- download tarball
- view history from revision 1.1.36
- Committer: Bazaar Package Importer
- Author(s): Rodrigo Moya
- Date: 2010-09-15 22:17:37 UTC
- mto: This revision was merged to the branch mainline in revision 55.
- Revision ID: james.westby@ubuntu.com-20100915221737-l0sy32v8vh0c9coj
Tags: upstream-1.4.1
Import upstream version 1.4.1
- files modified:
- Makefile.am
added
removed
tests/syncdaemon/test_fsm.py
36
36
FileSystemManager,
37
37
InconsistencyError,
38
38
METADATA_VERSION,
39
TrashFileShelf,
39
40
)
40
41
from ubuntuone.syncdaemon.event_queue import EventQueue
41
42
from ubuntuone.syncdaemon import logger
43
from ubuntuone.syncdaemon.interfaces import IMarker
44
from ubuntuone.syncdaemon.marker import MDMarker
42
45
from ubuntuone.syncdaemon.volume_manager import Share, allow_writes
43
46
44
47
TESTS_DIR = os.path.join(os.getcwd(), "tmp")
73
76
self.share_path = self.share.path
74
77
75
78
# add a in-memory logger handler
76
self.memento = MementoHandler()
77
self.memento.setLevel(0)
78
logger.root_logger.addHandler(self.memento)
79
self.handler = MementoHandler()
80
self.handler.setLevel(0)
81
logger.root_logger.addHandler(self.handler)
79
82
80
83
def tearDown(self):
81
84
""" Clean up the tests. """
83
86
self.rmtree(TESTS_DIR)
84
87
85
88
# remove the handler
86
logger.root_logger.removeHandler(self.memento)
89
logger.root_logger.removeHandler(self.handler)
87
90
88
91
@staticmethod
89
92
def rmtree(path):
1453
1456
1454
1457
def test_move_overwriting(self):
1455
1458
"""Test that move refreshes stat when overwrites other file."""
1459
self.fsm.create(self.share_path, self.share.id, is_dir=True)
1456
1460
path1 = os.path.join(self.share.path, "thisfile1")
1457
1461
path2 = os.path.join(self.share.path, "thisfile2")
1458
1462
open(path1, "w").close()
1679
1683
self.fsm.remove_partial("uuid", "share")
1680
1684
1681
1685
def test_ro_share(self):
1682
"""
1683
Test that creating a partial of a ro share leave the partials dir
1684
permissions intact.
1686
"""Test creating a partial of a RO share.
1687
1688
It should leave the partials dir permissions intact.
1685
1689
"""
1686
1690
share = self.create_share('ro_share', 'ro_share_name',
1687
1691
self.fsm, self.shares_dir,
1820
1824
1821
1825
def test_move_file_overwrite(self):
1822
1826
"""Test that a file is moved over other one."""
1827
self.fsm.create(self.share_path, self.share.id,
1828
self.share.node_id, is_dir=True)
1823
1829
testfile1 = os.path.join(self.share_path, "path1")
1824
1830
mdid1 = self.fsm.create(testfile1, "share")
1825
1831
self.fsm.set_node_id(testfile1, "uuid1")
1843
1849
mdobj = self.fsm.get_by_path(testfile2)
1844
1850
self.assertEqual(mdobj.mdid, mdid1)
1845
1851
1846
# check that the info for the overwritten one is gone
1852
# check that the info for the overwritten one is gone to trash
1847
1853
self.assert_no_metadata(mdid2, testfile1, "share", "uuid2")
1854
self.assertEqual(self.fsm.trash[(self.share.id, "uuid2")],
1855
(mdid2, self.share.node_id, testfile2))
1848
1856
1849
1857
def test_move_file_withdir(self):
1850
1858
"""Test that a dir is moved from one point to the other."""
2008
2016
assert not local_file in self.fsm._idx_path
2009
2017
self.assertRaises(OSError, self.fsm.delete_file, local_dir)
2010
2018
2011
infos = [record.message for record in self.memento.records
2019
infos = [record.message for record in self.handler.records
2012
2020
if record.levelname == 'INFO']
2013
2021
self.assertTrue(len(infos) == 1)
2014
2022
self.assertTrue(local_file in infos[0])
2028
2036
assert not local_subdir in self.fsm._idx_path
2029
2037
self.assertRaises(OSError, self.fsm.delete_file, local_dir)
2030
2038
2031
infos = [record.message for record in self.memento.records
2039
infos = [record.message for record in self.handler.records
2032
2040
if record.levelname == 'INFO']
2033
2041
self.assertTrue(len(infos) == 1)
2034
2042
self.assertTrue(local_subdir in infos[0])
2045
2053
self.fsm.create(local_file, "")
2046
2054
self.fsm.set_node_id(local_file, "uuid_file")
2047
2055
2048
previous = self.memento.records
2056
previous = self.handler.records
2049
2057
self.fsm.delete_file(local_dir)
2050
2058
2051
2059
# no logs were added
2052
self.assertEquals(previous, self.memento.records)
2060
self.assertEquals(previous, self.handler.records)
2053
2061
2054
2062
def test_warning_on_log_file_when_failing_delete(self):
2055
2063
"""Test that sucessfully deleted dir does not log OSError."""
2061
2069
# local_dir does not exist on the file system
2062
2070
self.fsm.delete_file(local_dir)
2063
2071
2064
warnings = [record.message for record in self.memento.records
2072
warnings = [record.message for record in self.handler.records
2065
2073
if record.levelname == 'WARNING']
2066
2074
self.assertTrue(len(warnings) == 1)
2067
2075
self.assertTrue('OSError [Errno 2]' in warnings[0])
2122
2130
self.assertTrue(os.path.exists(tdir1))
2123
2131
self.assertTrue(self.fsm.get_by_mdid(mdid1))
2124
2132
2133
2134
class LimboTests(FSMTestCase):
2135
"""Test related to trash and move limbo."""
2136
2137
def test_movelimbo(self):
2138
"""Test that a node is sent to and removed from the move limbo."""
2139
# move to limbo
2140
self.fsm.add_to_move_limbo("share", "uuid", "old_parent",
2141
"new_parent", "new_name")
2142
d = {("share", "uuid"): ("old_parent", "new_parent", "new_name")}
2143
self.assertEqual(self.fsm.move_limbo, d)
2144
r = [("share", "uuid", "old_parent", "new_parent", "new_name")]
2145
self.assertEqual(list(self.fsm.get_iter_move_limbo()), r)
2146
2147
# remove from limbo
2148
self.fsm.remove_from_move_limbo("share", "uuid")
2149
self.assertEqual(self.fsm.move_limbo, {})
2150
self.assertEqual(list(self.fsm.get_iter_move_limbo()), [])
2151
2125
2152
def test_trash_normal(self):
2126
2153
"""Test that a node is sent to and removed from trash."""
2127
2154
testfile = os.path.join(self.share_path, "path")
2131
2158
self.assertTrue(self.fsm.has_metadata(mdid=mdid))
2132
2159
self.assertEqual(self.fsm.trash, {})
2133
2160
self.assertEqual(list(self.fsm.get_iter_trash()), [])
2161
self.assertFalse(self.fsm.node_in_trash("share", "uuid"))
2134
2162
2135
2163
# delete to trash
2136
2164
self.fsm.delete_to_trash(mdid, "parent")
2139
2167
{("share", "uuid"): (mdid, "parent", testfile)})
2140
2168
self.assertEqual(list(self.fsm.get_iter_trash()),
2141
2169
[("share", "uuid", "parent")])
2170
self.assertTrue(self.fsm.node_in_trash("share", "uuid"))
2142
2171
2143
2172
# remove from trash
2144
2173
self.fsm.remove_from_trash("share", "uuid")
2145
2174
self.assertFalse(self.fsm.has_metadata(mdid=mdid))
2146
2175
self.assertEqual(self.fsm.trash, {})
2147
2176
self.assertEqual(list(self.fsm.get_iter_trash()), [])
2177
self.assertFalse(self.fsm.node_in_trash("share", "uuid"))
2148
2178
2149
2179
def test_trash_old(self):
2150
2180
"""Test that get_iter_trash supports old trash."""
2152
2182
self.assertEqual(list(self.fsm.get_iter_trash()),
2153
2183
[("share", "uuid", "parent")])
2154
2184
2155
def test_local_changed_empty_dir(self):
2156
"""Test the recursive changed feature for a node."""
2157
local_dir = os.path.join(self.root_dir, "foo")
2158
os.mkdir(local_dir)
2159
mdid = self.fsm.create(local_dir, "", is_dir=True)
2160
self.fsm.set_node_id(local_dir, "uuid")
2161
2162
# local_hash differs from server_hash for local_dir
2163
self.fsm.set_by_mdid(mdid, local_hash=98765)
2164
2165
assert len(os.listdir(local_dir)) == 0 # local_dir is empty
2166
self.assertTrue(self.fsm.local_changed(path=local_dir))
2167
2168
def test_local_changed_non_empty_dir(self):
2169
"""Test the recursive changed feature for a non empty dir."""
2170
local_dir = os.path.join(self.root_dir, "foo")
2171
os.mkdir(local_dir)
2172
self.fsm.create(local_dir, "", is_dir=True)
2173
self.fsm.set_node_id(local_dir, "uuid")
2174
2175
sub_dir = os.path.join(local_dir, "bar")
2176
os.mkdir(sub_dir)
2177
mdid_subdir = self.fsm.create(sub_dir, "", is_dir=True)
2178
self.fsm.set_node_id(sub_dir, "uuid_subdir")
2179
2180
assert len(os.listdir(local_dir)) > 0 # local_dir is not empty
2181
self.assertFalse(self.fsm.local_changed(path=local_dir))
2182
2183
# local_hash differs from server_hash for sub_dir
2184
self.fsm.set_by_mdid(mdid_subdir, local_hash=98765)
2185
2186
self.assertTrue(self.fsm.local_changed(path=local_dir))
2185
def test_trash_with_node_in_none(self):
2186
"""Test that in trash is saved the marker if node_id is None."""
2187
testfile = os.path.join(self.share_path, "path")
2188
open(testfile, "w").close()
2189
mdid = self.fsm.create(testfile, "share")
2190
2191
# delete to trash and check the marker
2192
self.fsm.delete_to_trash(mdid, "parent")
2193
marker = MDMarker(mdid)
2194
self.assertEqual(self.fsm.trash,
2195
{("share", marker): (mdid, "parent", testfile)})
2196
2197
def test_dereference_ok_limbos_none(self):
2198
"""Limbos' markers ok dereferencing is fine if no marker at all."""
2199
self.fsm.dereference_ok_limbos('nothing', "foo")
2200
2201
def test_dereference_err_limbos_none(self):
2202
"""Limbos' markers err dereferencing is fine if no marker at all."""
2203
self.fsm.dereference_err_limbos('nothing')
2204
2205
def test_dereference_ok_trash_node(self):
2206
"""Dereference possible marker in trash, node."""
2207
# set up
2208
testfile = os.path.join(self.share_path, "path")
2209
mdid = self.fsm.create(testfile, "share")
2210
self.fsm.set_node_id(testfile, "mrkr")
2211
self.fsm.delete_to_trash(mdid, "parent")
2212
2213
# dereference and test
2214
self.fsm.dereference_ok_limbos("mrkr", "final")
2215
self.assertFalse(self.fsm.node_in_trash("share", "mrkr"))
2216
self.assertTrue(self.fsm.node_in_trash("share", "final"))
2217
self.assertTrue(self.handler.check_debug("dereference ok trash",
2218
"marker", "node"))
2219
2220
def test_dereference_ok_trash_parent(self):
2221
"""Dereference possible marker in trash, parent."""
2222
# set up
2223
testfile = os.path.join(self.share_path, "path")
2224
mdid = self.fsm.create(testfile, "share")
2225
self.fsm.set_node_id(testfile, "node")
2226
self.fsm.delete_to_trash(mdid, "mrkr")
2227
2228
# dereference and test
2229
self.fsm.dereference_ok_limbos("mrkr", "final")
2230
self.assertEqual(self.fsm.trash[("share", "node")][1], "final")
2231
self.assertTrue(self.handler.check_debug("dereference ok trash",
2232
"marker", "parent"))
2233
2234
def test_dereference_ok_trash_parent_node(self):
2235
"""An unlinked node can be a parent of other."""
2236
# set up one node
2237
testfile = os.path.join(self.share_path, "path1")
2238
mdid = self.fsm.create(testfile, "share")
2239
self.fsm.set_node_id(testfile, "mrkr")
2240
self.fsm.delete_to_trash(mdid, "parent")
2241
2242
# set up child node
2243
testfile = os.path.join(self.share_path, "path2")
2244
mdid = self.fsm.create(testfile, "share")
2245
self.fsm.set_node_id(testfile, "node")
2246
self.fsm.delete_to_trash(mdid, "mrkr")
2247
2248
# dereference and test
2249
self.fsm.dereference_ok_limbos("mrkr", "final")
2250
self.assertFalse(self.fsm.node_in_trash("share", "mrkr"))
2251
self.assertTrue(self.fsm.node_in_trash("share", "final"))
2252
self.assertEqual(self.fsm.trash[("share", "node")][1], "final")
2253
2254
def test_dereference_err_trash_node(self):
2255
"""Dereference with error possible marker in trash, node."""
2256
# set up
2257
testfile = os.path.join(self.share_path, "path")
2258
mdid = self.fsm.create(testfile, "share")
2259
self.fsm.set_node_id(testfile, "mrkr")
2260
self.fsm.delete_to_trash(mdid, "parent")
2261
2262
# dereference and test
2263
self.fsm.dereference_err_limbos("mrkr")
2264
self.assertFalse(self.fsm.node_in_trash("share", "mrkr"))
2265
self.assertTrue(self.handler.check_debug("dereference err trash", "marker"))
2266
2267
def test_dereference_err_trash_parent(self):
2268
"""Dereference with error possible marker in trash, parent."""
2269
# set up
2270
testfile = os.path.join(self.share_path, "path")
2271
mdid = self.fsm.create(testfile, "share")
2272
self.fsm.set_node_id(testfile, "node")
2273
self.fsm.delete_to_trash(mdid, "mrkr")
2274
2275
# dereference and test
2276
self.fsm.dereference_err_limbos("mrkr")
2277
self.assertFalse(self.fsm.node_in_trash("share", "node"))
2278
self.assertTrue(self.handler.check_debug("dereference err trash", "marker"))
2279
2280
def test_dereference_err_trash_parent_node(self):
2281
"""An unlinked node can be a parent of other, both with failure."""
2282
# set up one node
2283
testfile = os.path.join(self.share_path, "path1")
2284
mdid = self.fsm.create(testfile, "share")
2285
self.fsm.set_node_id(testfile, "mrkr")
2286
self.fsm.delete_to_trash(mdid, "parent")
2287
2288
# set up child node
2289
testfile = os.path.join(self.share_path, "path2")
2290
mdid = self.fsm.create(testfile, "share")
2291
self.fsm.set_node_id(testfile, "node")
2292
self.fsm.delete_to_trash(mdid, "mrkr")
2293
2294
# dereference and test
2295
self.fsm.dereference_err_limbos("mrkr")
2296
self.assertFalse(self.fsm.node_in_trash("share", "mrkr"))
2297
self.assertFalse(self.fsm.node_in_trash("share", "node"))
2298
2299
def test_dereference_ok_movelimbo_node(self):
2300
"""Dereference possible marker in move limbo, node."""
2301
# set up
2302
self.fsm.add_to_move_limbo("sh", "mrkr", "oldparent", "newparent", "x")
2303
2304
# dereference and test
2305
self.fsm.dereference_ok_limbos("mrkr", "final")
2306
self.assertFalse(("sh", "mrkr") in self.fsm.move_limbo)
2307
self.assertTrue(("sh", "final") in self.fsm.move_limbo)
2308
self.assertTrue(self.handler.check_debug("dereference ok move limbo",
2309
"marker", "node"))
2310
2311
def test_dereference_ok_movelimbo_oldparent(self):
2312
"""Dereference possible marker in move limbo, oldparent."""
2313
# set up
2314
self.fsm.add_to_move_limbo("sh", "node", "mrkr", "newparent", "x")
2315
2316
# dereference and test
2317
self.fsm.dereference_ok_limbos("mrkr", "final")
2318
self.assertEqual(self.fsm.move_limbo[("sh", "node")],
2319
("final", "newparent", "x"))
2320
self.assertTrue(self.handler.check_debug("dereference ok move limbo",
2321
"marker", "old_parent"))
2322
2323
def test_dereference_ok_movelimbo_newparent(self):
2324
"""Dereference possible marker in move limbo, newparent."""
2325
# set up
2326
self.fsm.add_to_move_limbo("sh", "node", "oldparent", "mrkr", "x")
2327
2328
# dereference and test
2329
self.fsm.dereference_ok_limbos("mrkr", "final")
2330
self.assertEqual(self.fsm.move_limbo[("sh", "node")],
2331
("oldparent", "final", "x"))
2332
self.assertTrue(self.handler.check_debug("dereference ok move limbo",
2333
"marker", "new_parent"))
2334
2335
def test_dereference_ok_movelimbo_bothparents(self):
2336
"""Dereference possible marker in move limbo, both parents."""
2337
# set up
2338
self.fsm.add_to_move_limbo("sh", "node", "mrkr", "mrkr", "x")
2339
2340
# dereference and test
2341
self.fsm.dereference_ok_limbos("mrkr", "final")
2342
self.assertEqual(self.fsm.move_limbo[("sh", "node")],
2343
("final", "final", "x"))
2344
2345
def test_dereference_err_movelimbo_node(self):
2346
"""Dereference with error possible marker in move limbo, node."""
2347
# set up
2348
self.fsm.add_to_move_limbo("sh", "mrkr", "oldparent", "newparent", "x")
2349
2350
# dereference and test
2351
self.fsm.dereference_err_limbos("mrkr")
2352
self.assertFalse(("sh", "mrkr") in self.fsm.move_limbo)
2353
self.assertTrue(self.handler.check_debug("dereference err move limbo",
2354
"marker"))
2355
2356
def test_dereference_err_movelimbo_oldparent(self):
2357
"""Dereference with error possible marker in move limbo, oldparent."""
2358
# set up
2359
self.fsm.add_to_move_limbo("sh", "node", "mrkr", "newparent", "x")
2360
2361
# dereference and test
2362
self.fsm.dereference_err_limbos("mrkr")
2363
self.assertFalse(("sh", "node") in self.fsm.move_limbo)
2364
self.assertTrue(self.handler.check_debug("dereference err move limbo",
2365
"marker"))
2366
2367
def test_dereference_err_movelimbo_newparent(self):
2368
"""Dereference with error possible marker in move limbo, newparent."""
2369
# set up
2370
self.fsm.add_to_move_limbo("sh", "node", "oldparent", "mrkr", "x")
2371
2372
# dereference and test
2373
self.fsm.dereference_err_limbos("mrkr")
2374
self.assertFalse(("sh", "node") in self.fsm.move_limbo)
2375
self.assertTrue(self.handler.check_debug("dereference err move limbo",
2376
"marker"))
2377
2378
def test_dereference_err_movelimbo_bothparents(self):
2379
"""Dereference with error possible marker in move limbo, both."""
2380
# set up
2381
self.fsm.add_to_move_limbo("sh", "node", "mrkr", "mrkr", "x")
2382
2383
# dereference and test
2384
self.fsm.dereference_err_limbos("mrkr")
2385
self.assertFalse(("sh", "node") in self.fsm.move_limbo)
2187
2386
2188
2387
2189
2388
class SyntheticInfoTests(FSMTestCase):
3233
3432
self.assertTrue(mdid4 in dirty_mdids)
3234
3433
3235
3434
3435
class TrashFileShelfTests(unittest.TestCase):
3436
"""Test the customized file shelf."""
3437
3438
def setUp(self):
3439
"""Set up."""
3440
try:
3441
os.mkdir(TESTS_DIR)
3442
except OSError:
3443
# already there, remove it to clean and create again
3444
shutil.rmtree(TESTS_DIR)
3445
os.mkdir(TESTS_DIR)
3446
3447
self.tfs = TrashFileShelf(os.path.join(TESTS_DIR, "trash"))
3448
3449
def tearDown(self):
3450
"""Tear down."""
3451
shutil.rmtree(TESTS_DIR)
3452
3453
def test_one_value(self):
3454
"""Test the file shelf with one value."""
3455
self.tfs[("foo", "bar")] = 'value'
3456
3457
self.assertEqual(self.tfs[("foo", "bar")], 'value')
3458
self.assertEqual(list(self.tfs.keys()), [("foo", "bar")])
3459
3460
def test_two_values(self):
3461
"""Test the file shelf with two values."""
3462
self.tfs[("foo", "bar")] = 'value1'
3463
self.tfs[("xyz", "hfb")] = 'value2'
3464
3465
self.assertEqual(self.tfs[("foo", "bar")], 'value1')
3466
self.assertEqual(self.tfs[("xyz", "hfb")], 'value2')
3467
self.assertEqual(sorted(self.tfs.keys()),
3468
[("foo", "bar"), ("xyz", "hfb")])
3469
3470
def test_node_id_None(self):
3471
"""node_id can be None."""
3472
self.tfs[("foo", None)] = 'value'
3473
self.assertEqual(self.tfs[("foo", None)], 'value')
3474
self.assertEqual(list(self.tfs.keys()), [("foo", None)])
3475
3476
def test_node_id_marker(self):
3477
"""node_id can be a marker."""
3478
marker = MDMarker("bar")
3479
self.tfs[("foo", marker)] = 'value'
3480
self.assertEqual(self.tfs[("foo", marker)], 'value')
3481
self.assertEqual(list(self.tfs.keys()), [("foo", marker)])
3482
node_id = list(self.tfs.keys())[0][1]
3483
self.assertTrue(IMarker.providedBy(node_id))
3484
3485
def test_share_id_marker(self):
3486
"""share_id can be a marker."""
3487
marker = MDMarker("bar")
3488
self.tfs[(marker, "foo")] = 'value'
3489
self.assertEqual(self.tfs[(marker, "foo")], 'value')
3490
self.assertEqual(list(self.tfs.keys()), [(marker, "foo")])
3491
share_id = list(self.tfs.keys())[0][0]
3492
self.assertTrue(IMarker.providedBy(share_id))
3493
3494
3236
3495
def test_suite():
3237
3496
# pylint: disable-msg=C0111
3238
3497
return unittest.TestLoader().loadTestsFromName(__name__)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1