Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_fetch_filtering_not_active(self):
"""Test whether classified fields are not removed from the items"""
backend = ClassifiedFieldsBackend('http://example.com/', tag='test')
items = [item for item in backend.fetch(category=ClassifiedFieldsBackend.DEFAULT_CATEGORY,
filter_classified=False)]
for x in range(5):
item = items[x]
expected_uuid = uuid('http://example.com/', str(x))
self.assertEqual(item['origin'], 'http://example.com/')
self.assertEqual(item['uuid'], expected_uuid)
self.assertEqual(item['tag'], 'test')
self.assertEqual(item['category'], ClassifiedFieldsBackend.DEFAULT_CATEGORY)
self.assertEqual(item['classified_fields_filtered'], None)
# Fields in CLASSIFIED_FIELDS are not deleted
expected = {
'category': 'mock_item',
'classified': x,
'item': x,
'my': {
'classified': {'field': x},
'field': x,
'list_classified': [{'dict_classified': {'field': x}, 'field': x}]
},
# Count the number of commits before adding some new
expected = [('bc57a9209f096a130dcc5ba7089a8663f758a703', 1344965413.0),
('87783129c3f00d2c81a3a8e585eb86a47e39891a', 1344965535.0),
('7debcf8a2f57f86663809c58b5c07a398be7674c', 1344965607.0),
('c0d66f92a95e31c77be08dc9d0f11a16715d1885', 1344965702.0),
('c6ba8f7a1058db3e6b4bc6f1090e932b107605fb', 1344966351.0),
('589bb080f059834829a2a5955bebfd7c2baa110a', 1344967441.0),
('ce8e0b86a1e9877f42fe9453ede418519115f367', 1392185269.0),
('51a3b654f252210572297f47597b31527c475fb8', 1392185366.0),
('456a68ee1407a77f3e804a30dff245bb6c6b872f', 1392185439.0)]
self.assertEqual(len(commits), len(expected))
for x in range(len(commits)):
expected_uuid = uuid(editable_path, expected[x][0])
commit = commits[x]
self.assertEqual(commit['data']['commit'], expected[x][0])
self.assertEqual(commit['origin'], editable_path)
self.assertEqual(commit['uuid'], expected_uuid)
self.assertEqual(commit['updated_on'], expected[x][1])
self.assertEqual(commit['category'], 'commit')
self.assertEqual(commit['tag'], editable_path)
# Create some new commits
cmd = ['git', 'checkout', '-b', 'mybranch']
subprocess.check_output(cmd, stderr=subprocess.STDOUT,
cwd=editable_path, env={'LANG': 'C'})
with open(new_file, 'w') as f:
f.write("Testing sync method")
self.assertEqual(big.backend.origin, args['origin'])
self.assertEqual(big.backend.tag, args['tag'])
self.assertEqual(len(items), 5)
with BackendItemsGenerator(CommandBackend, args, category,
manager=manager, fetch_archive=True,
archived_after=str_to_datetime('1970-01-01')) as big:
self.assertIsInstance(big, BackendItemsGenerator)
items = [item for item in big.items]
self.assertEqual(len(items), 10)
for x in range(2):
for y in range(5):
item = items[y + (x * 5)]
expected_uuid = uuid('http://example.com/', str(y))
self.assertEqual(item['data']['item'], y)
self.assertEqual(item['data']['archive'], True)
self.assertEqual(item['origin'], 'http://example.com/')
self.assertEqual(item['uuid'], expected_uuid)
self.assertEqual(item['tag'], 'test')
self.assertEqual(item['classified_fields_filtered'], None)
def test_none_value(self):
"""Check whether a UUID cannot be generated when a given value is None"""
self.assertRaises(ValueError, uuid, '1', '2', None, '3')
self.assertRaises(ValueError, uuid, None, '1', '2', '3')
self.assertRaises(ValueError, uuid, '1', '2', '3', None)
commits = [commit for commit in git.fetch()]
expected = [('bc57a9209f096a130dcc5ba7089a8663f758a703', 1344965413.0),
('87783129c3f00d2c81a3a8e585eb86a47e39891a', 1344965535.0),
('7debcf8a2f57f86663809c58b5c07a398be7674c', 1344965607.0),
('c0d66f92a95e31c77be08dc9d0f11a16715d1885', 1344965702.0),
('c6ba8f7a1058db3e6b4bc6f1090e932b107605fb', 1344966351.0),
('589bb080f059834829a2a5955bebfd7c2baa110a', 1344967441.0),
('ce8e0b86a1e9877f42fe9453ede418519115f367', 1392185269.0),
('51a3b654f252210572297f47597b31527c475fb8', 1392185366.0),
('456a68ee1407a77f3e804a30dff245bb6c6b872f', 1392185439.0)]
self.assertEqual(len(commits), len(expected))
for x in range(len(commits)):
expected_uuid = uuid(self.git_detached_path, expected[x][0])
commit = commits[x]
self.assertEqual(commit['data']['commit'], expected[x][0])
self.assertEqual(commit['origin'], self.git_detached_path)
self.assertEqual(commit['uuid'], expected_uuid)
self.assertEqual(commit['updated_on'], expected[x][1])
self.assertEqual(commit['category'], 'commit')
self.assertEqual(commit['tag'], self.git_detached_path)
shutil.rmtree(new_path)
def test_metadata(self):
backend = MockedBackend('test', 'mytag')
before = datetime_utcnow().timestamp()
items = [item for item in backend.fetch()]
after = datetime_utcnow().timestamp()
for x in range(5):
item = items[x]
expected_uuid = uuid('test', str(x))
expected_updated_on = 1451606400.0 + item['data']['item']
self.assertEqual(item['data']['item'], x)
self.assertEqual(item['backend_name'], 'MockedBackend')
self.assertEqual(item['backend_version'], '0.2.0')
self.assertEqual(item['perceval_version'], __version__)
self.assertEqual(item['origin'], 'test')
self.assertEqual(item['uuid'], expected_uuid)
self.assertEqual(item['updated_on'], expected_updated_on)
self.assertEqual(item['category'], 'mock_item')
self.assertEqual(item['classified_fields_filtered'], None)
self.assertEqual(item['tag'], 'mytag')
self.assertGreater(item['timestamp'], before)
self.assertLess(item['timestamp'], after)
before = item['timestamp']
category, str_to_datetime('1970-01-01'))
self.assertEqual(len(filepaths), 2)
to_remove = filepaths[0]
delete_rows(to_remove, 'archive')
# Fetch items from the archive
items = fetch_from_archive(CommandBackend, args, manager,
category, str_to_datetime('1970-01-01'))
items = [item for item in items]
self.assertEqual(len(items), 5)
for x in range(5):
item = items[x]
expected_uuid = uuid('http://example.com/', str(x))
self.assertEqual(item['data']['item'], x)
self.assertEqual(item['data']['archive'], True)
self.assertEqual(item['origin'], 'http://example.com/')
self.assertEqual(item['uuid'], expected_uuid)
self.assertEqual(item['tag'], 'test')
self.assertEqual(item['classified_fields_filtered'], None)
def test_fetch_filtering_empty_list(self):
"""Test whether no data is removed when classified fields list is empty"""
backend = ClassifiedFieldsBackend('http://example.com/', tag='test')
backend.CLASSIFIED_FIELDS = []
items = [item for item in backend.fetch(category=ClassifiedFieldsBackend.DEFAULT_CATEGORY,
filter_classified=True)]
for x in range(5):
item = items[x]
expected_uuid = uuid('http://example.com/', str(x))
self.assertEqual(item['origin'], 'http://example.com/')
self.assertEqual(item['uuid'], expected_uuid)
self.assertEqual(item['tag'], 'test')
self.assertEqual(item['category'], ClassifiedFieldsBackend.DEFAULT_CATEGORY)
self.assertEqual(item['classified_fields_filtered'], [])
expected = {
'category': 'mock_item',
'classified': x,
'item': x,
'my': {
'classified': {'field': x},
'field': x,
'list_classified': [{'dict_classified': {'field': x}, 'field': x}]
},
}
args = ['--archive-path', self.test_path, '--fetch-archive',
'--from-date', '2015-01-01', '--tag', 'test', '--category', 'mock_item',
'--subtype', 'mocksubtype',
'--output', self.fout_path, 'http://example.com/']
cmd = MockedBackendCommand(*args)
cmd.run()
cmd.outfile.close()
items = [item for item in convert_cmd_output_to_json(self.fout_path)]
self.assertEqual(len(items), 5)
for x in range(5):
item = items[x]
expected_uuid = uuid('http://example.com/', str(x))
# ArchiveMockedBackend sets 'archive' value when
# 'fetch-archive' option is set. This helps to know
# the code is really running
self.assertEqual(item['data']['item'], x)
self.assertEqual(item['data']['archive'], True)
self.assertEqual(item['origin'], 'http://example.com/')
self.assertEqual(item['uuid'], expected_uuid)
self.assertEqual(item['tag'], 'test')
self.assertEqual(item['classified_fields_filtered'], None)
'--archive-path', self.test_path, '--category', MockedBackend.DEFAULT_CATEGORY,
'--subtype', 'mocksubtype',
'--from-date', '2015-01-01', '--tag', 'test',
'--output', self.fout_path, 'http://example.com/']
cmd = MockedBackendCommand(*args)
cmd.run()
cmd.outfile.close()
items = [item for item in convert_cmd_output_to_json(self.fout_path)]
self.assertEqual(len(items), 5)
for x in range(5):
item = items[x]
expected_uuid = uuid('http://example.com/', str(x))
self.assertEqual(item['data']['item'], x)
self.assertEqual(item['origin'], 'http://example.com/')
self.assertEqual(item['uuid'], expected_uuid)
self.assertEqual(item['tag'], 'test')
self.assertEqual(item['category'], MockedBackend.DEFAULT_CATEGORY)
self.assertEqual(item['classified_fields_filtered'], None)