224 lines
10 KiB
Python
224 lines
10 KiB
Python
import unittest
|
|
import re
|
|
import os
|
|
from testfixtures import LogCapture
|
|
import time
|
|
|
|
from resync.resource_list_builder import ResourceListBuilder
|
|
from resync.resource_list import ResourceList
|
|
from resync.resource import Resource
|
|
from resync.mapper import Mapper, MapperError
|
|
|
|
|
|
class TestResourceListBuilder(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
# Set timestamps (mtime) for test data. Timestamps on disk are
|
|
# in UTC so no conversion issues.
|
|
# Test case file_a: 1343236426 = 2012-07-25T17:13:46Z
|
|
# Test case file_b: 1000000000 = 2001-09-09T01:46:40Z
|
|
os.utime("tests/testdata/dir1/file_a", (0, 1343236426))
|
|
os.utime("tests/testdata/dir1/file_b", (0, 1000000000))
|
|
|
|
def test01_simple_scan(self):
|
|
rlb = ResourceListBuilder()
|
|
rlb.mapper = Mapper(['http://example.org/t', 'tests/testdata/dir1'])
|
|
rl = rlb.from_disk()
|
|
self.assertEqual(len(rl), 2)
|
|
rli = iter(rl)
|
|
r = next(rli)
|
|
self.assertEqual(r.uri, 'http://example.org/t/file_a')
|
|
self.assertEqual(r.lastmod, '2012-07-25T17:13:46Z')
|
|
self.assertEqual(r.md5, None)
|
|
self.assertEqual(r.length, 20)
|
|
self.assertEqual(r.path, None)
|
|
r = next(rli)
|
|
self.assertEqual(r.uri, 'http://example.org/t/file_b')
|
|
self.assertEqual(r.lastmod, '2001-09-09T01:46:40Z')
|
|
self.assertEqual(r.md5, None)
|
|
self.assertEqual(r.length, 45)
|
|
self.assertEqual(r.path, None)
|
|
# Make sure at and completed were set
|
|
self.assertTrue(rl.md_at is not None)
|
|
self.assertTrue(rl.md_completed is not None)
|
|
|
|
def test02_no_length(self):
|
|
rlb = ResourceListBuilder(set_length=False)
|
|
rlb.mapper = Mapper(['http://example.org/t', 'tests/testdata/dir1'])
|
|
rl = rlb.from_disk()
|
|
self.assertEqual(len(rl), 2)
|
|
rli = iter(rl)
|
|
r = next(rli)
|
|
self.assertEqual(r.uri, 'http://example.org/t/file_a')
|
|
self.assertEqual(r.lastmod, '2012-07-25T17:13:46Z')
|
|
self.assertEqual(r.md5, None)
|
|
self.assertEqual(r.length, None)
|
|
self.assertEqual(r.path, None)
|
|
r = next(rli)
|
|
self.assertEqual(r.uri, 'http://example.org/t/file_b')
|
|
self.assertEqual(r.lastmod, '2001-09-09T01:46:40Z')
|
|
self.assertEqual(r.md5, None)
|
|
self.assertEqual(r.length, None)
|
|
self.assertEqual(r.path, None)
|
|
|
|
def test03_set_hashes(self):
|
|
rlb = ResourceListBuilder(set_hashes=['md5'])
|
|
rlb.mapper = Mapper(['http://example.org/t', 'tests/testdata/dir1'])
|
|
rl = rlb.from_disk()
|
|
self.assertEqual(len(rl), 2)
|
|
rli = iter(rl)
|
|
r = next(rli)
|
|
self.assertEqual(r.uri, 'http://example.org/t/file_a')
|
|
self.assertEqual(r.lastmod, '2012-07-25T17:13:46Z')
|
|
self.assertEqual(r.md5, '6bf26fd66601b528d2e0b47eaa87edfd')
|
|
self.assertEqual(r.length, 20)
|
|
self.assertEqual(r.path, None)
|
|
r = next(rli)
|
|
self.assertEqual(r.uri, 'http://example.org/t/file_b')
|
|
self.assertEqual(r.lastmod, '2001-09-09T01:46:40Z')
|
|
self.assertEqual(r.md5, '452e54bdae1626ac5d6e7be81b39de21')
|
|
self.assertEqual(r.length, 45)
|
|
self.assertEqual(r.path, None)
|
|
|
|
def test04_data(self):
|
|
rlb = ResourceListBuilder(set_path=True, set_hashes=['md5'])
|
|
rlb.mapper = Mapper(['http://example.org/t', 'tests/testdata/dir1'])
|
|
rl = rlb.from_disk()
|
|
self.assertEqual(len(rl), 2)
|
|
r = rl.resources.get('http://example.org/t/file_a')
|
|
self.assertTrue(r is not None)
|
|
self.assertEqual(r.uri, 'http://example.org/t/file_a')
|
|
self.assertEqual(r.lastmod, '2012-07-25T17:13:46Z')
|
|
self.assertEqual(r.md5, '6bf26fd66601b528d2e0b47eaa87edfd')
|
|
self.assertEqual(r.path, 'tests/testdata/dir1/file_a')
|
|
|
|
def test05_from_disk_paths(self):
|
|
rlb = ResourceListBuilder()
|
|
rlb.mapper = Mapper(['http://example.org/t', 'tests/testdata/dir1'])
|
|
# no path, should get no resources
|
|
rl = rlb.from_disk(paths=[])
|
|
self.assertEqual(len(rl), 0)
|
|
# full path, 2 resources
|
|
rl = rlb.from_disk(paths=['tests/testdata/dir1'])
|
|
self.assertEqual(len(rl), 2)
|
|
# new object with mapper covering larger space of disk
|
|
rlb = ResourceListBuilder(set_path=True)
|
|
rlb.mapper = Mapper(['http://example.org/t', 'tests/testdata'])
|
|
# same path with 2 resources
|
|
rl = rlb.from_disk(paths=['tests/testdata/dir1'])
|
|
self.assertEqual(len(rl), 2)
|
|
# same path with 2 resources
|
|
rl = rlb.from_disk(
|
|
paths=['tests/testdata/dir1', 'tests/testdata/dir2'])
|
|
self.assertEqual(len(rl), 3)
|
|
# path that is just a single file
|
|
rl = rlb.from_disk(paths=['tests/testdata/dir1/file_a'])
|
|
self.assertEqual(len(rl), 1)
|
|
rli = iter(rl)
|
|
r = next(rli)
|
|
self.assertTrue(r is not None)
|
|
self.assertEqual(r.uri, 'http://example.org/t/dir1/file_a')
|
|
self.assertEqual(r.lastmod, '2012-07-25T17:13:46Z')
|
|
self.assertEqual(r.md5, None)
|
|
self.assertEqual(r.length, 20)
|
|
self.assertEqual(r.path, 'tests/testdata/dir1/file_a')
|
|
|
|
def test06_odd_file_names(self):
|
|
"""Verfify we can read unicode file names properly."""
|
|
rlb = ResourceListBuilder()
|
|
rlb.mapper = Mapper(['x:', 'tests/testdata/odd_file_names'])
|
|
rl = rlb.from_disk(paths=['tests/testdata/odd_file_names'])
|
|
# Get list of URIs to test
|
|
uris = [x.uri for x in rl]
|
|
self.assertTrue('x:/not_odd.txt' in uris)
|
|
self.assertTrue('x:/with&ersand.txt' in uris)
|
|
self.assertTrue('x:/with spaces.txt' in uris)
|
|
# File names for accented chars represented with combining chars
|
|
self.assertTrue(u'x:/Pi\u006e\u0303a_Colada.txt' in uris)
|
|
self.assertFalse(u'x:/Pi\u00f1a_Colada.txt' in uris)
|
|
self.assertTrue(u'x:/A_\u0041\u0303_tilde.txt' in uris)
|
|
self.assertFalse(u'x:/A_\u00c3_tilde.txt' in uris)
|
|
# Snowman is single char
|
|
self.assertFalse(u'x:snowman_\u2603.txt' in uris)
|
|
|
|
def test10_add_exclude_patterns(self):
|
|
"""Test add_exclude_patterns method."""
|
|
rlb = ResourceListBuilder()
|
|
self.assertEqual(len(rlb.exclude_patterns), 0)
|
|
rlb.add_exclude_patterns(['aaa', 'bbb'])
|
|
self.assertIn('aaa', rlb.exclude_patterns)
|
|
self.assertIn('bbb', rlb.exclude_patterns)
|
|
|
|
def test11_compile_excludes(self):
|
|
"""Test _compile_excludes method."""
|
|
rlb = ResourceListBuilder()
|
|
self.assertEqual(len(rlb.compiled_exclude_patterns), 0)
|
|
rlb.exclude_patterns = [r'aaa\d+', r'bbb']
|
|
rlb._compile_excludes()
|
|
self.assertEqual(len(rlb.compiled_exclude_patterns), 2)
|
|
# Error case
|
|
rlb.exclude_patterns.append('bad regex \\')
|
|
self.assertRaises(ValueError, rlb._compile_excludes)
|
|
|
|
def test12_exclude(self):
|
|
"""Test _exclude method."""
|
|
rlb = ResourceListBuilder()
|
|
rlb.add_exclude_patterns(['.*frog.*'])
|
|
rlb._compile_excludes()
|
|
self.assertTrue(rlb._exclude('a frog'))
|
|
self.assertFalse(rlb._exclude('toad'))
|
|
|
|
def test13_from_disk_add_path(self):
|
|
"""Test from_disk_add_path method."""
|
|
# Check sanity check - must have path, resource_list and mapper
|
|
rlb = ResourceListBuilder(mapper=Mapper())
|
|
self.assertRaises(ValueError, rlb.from_disk_add_path, path='aaa')
|
|
self.assertRaises(ValueError, rlb.from_disk_add_path, resource_list=ResourceList())
|
|
rlb = ResourceListBuilder()
|
|
self.assertRaises(ValueError, rlb.from_disk_add_path, path='aaa', resource_list=ResourceList())
|
|
# Check log message
|
|
rlb = ResourceListBuilder(mapper=Mapper(['http://example.org/', 'tests']))
|
|
rlb.log_count_increment = 2
|
|
rl = ResourceList()
|
|
with LogCapture() as lc:
|
|
rlb.from_disk_add_path(path='tests/testdata/dir1', resource_list=rl)
|
|
self.assertIn('from_disk_add_path: 2 files...', lc.records[-1].msg)
|
|
# text excluding dirs -- just one file under find2 not excluced
|
|
rlb = ResourceListBuilder(mapper=Mapper(['http://example.org/', 'tests']))
|
|
rl = ResourceList()
|
|
rlb.add_exclude_patterns(['find1', 'find3'])
|
|
rlb._compile_excludes()
|
|
rlb.from_disk_add_path(path='tests/testdata/find', resource_list=rl)
|
|
self.assertEqual(len(rl), 1)
|
|
|
|
def test14_add_file(self):
|
|
"""Test add_file method."""
|
|
rlb = ResourceListBuilder(mapper=Mapper(['http://example.org/', 'tests']))
|
|
rl = ResourceList()
|
|
rlb.add_exclude_patterns(['.*ro'])
|
|
with LogCapture() as lc:
|
|
# escluded
|
|
rlb.add_file(resource_list=rl, file='frog')
|
|
self.assertIn("Excluding file 'frog'", lc.records[-1].msg)
|
|
# mapper error
|
|
self.assertRaises(MapperError, rlb.add_file, resource_list=rl, file='i-dont-exist')
|
|
# map OK but doesn't exist
|
|
rlb.add_file(resource_list=rl, file='tests/i-dont-exist')
|
|
self.assertIn("Ignoring file 'tests/i-dont-exist'", lc.records[-1].msg)
|
|
# ignore symlink by default
|
|
rlb.add_file(resource_list=rl, file='tests/testdata/symlink/dir2/a_file.txt')
|
|
self.assertIn("Ignoring symlink 'tests/testdata/symlink/dir2/a_file.txt'", lc.records[-1].msg)
|
|
# ...or not
|
|
rl = ResourceList()
|
|
rlb.include_symlinks = True
|
|
rlb.add_file(resource_list=rl, file='tests/testdata/symlink/dir2/a_file.txt')
|
|
self.assertEqual(len(rl), 1)
|
|
# Check hashing
|
|
rlb = ResourceListBuilder(mapper=Mapper(['http://example.org/', 'tests/testdata/dir1/']),
|
|
set_hashes=['md5', 'sha-1', 'sha-256'])
|
|
rl = ResourceList()
|
|
rlb.add_file(resource_list=rl, file='tests/testdata/dir1/file_a')
|
|
self.assertEqual(rl['http://example.org/file_a'].md5, '6bf26fd66601b528d2e0b47eaa87edfd')
|
|
self.assertEqual(rl['http://example.org/file_a'].sha1, 'c60a598a5d9e489cf50533eeead6d70f15eafcf8')
|
|
self.assertEqual(rl['http://example.org/file_a'].sha256, '1c6291bfac0322752c4632ebd69bf6d81d53985fbf5ee54de5cc1fefba6566b6')
|
