Differential D897 Diff 2851 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 1,199 Lines • ▼ Show 20 Lines	class CommonTestStorage(TestStorageData):

def test_origin_visit_get_by_no_result(self):		def test_origin_visit_get_by_no_result(self):
# No result		# No result
actual_origin_visit = self.storage.origin_visit_get_by(		actual_origin_visit = self.storage.origin_visit_get_by(
10, 999)		10, 999)

self.assertIsNone(actual_origin_visit)		self.assertIsNone(actual_origin_visit)

		def test_person_get(self):
		# given (person injection through revision for example)
		self.storage.revision_add([self.revision])
		rev = list(self.storage.revision_get([self.revision['id']]))[0]

		id0 = rev['committer']['id']
		person0 = self.revision['committer']

		id1 = rev['author']['id']
		person1 = self.revision['author']

		# when
		actual_persons = self.storage.person_get([id0, id1])

		# then
		self.assertEqual(
		list(actual_persons), [
		{
		'id': id0,
		'fullname': person0['fullname'],
		'name': person0['name'],
		'email': person0['email'],
		},
		{
		'id': id1,
		'fullname': person1['fullname'],
		'name': person1['name'],
		'email': person1['email'],
		}
		])

		def test_person_get_fullname_unicity(self):
		# given (person injection through revisions for example)
		revision = self.revision

		# create a revision with same committer fullname but wo name and email
		revision2 = copy.deepcopy(self.revision2)
		revision2['committer'] = dict(revision['committer'])
		revision2['committer']['email'] = None
		revision2['committer']['name'] = None

		self.storage.revision_add([revision])
		self.storage.revision_add([revision2])

		# when getting added revisions
		revisions = list(
		olasdUnsubmitted Not Done Inline Actions I'd linebreak after `list(` to avoid needing backslashes olasd: I'd linebreak after `list(` to avoid needing backslashes
		self.storage.revision_get([revision['id'], revision2['id']]))

		# then
		# check committers are the same
		self.assertEqual(revisions[0]['committer'],
		revisions[1]['committer'])

		# check person_get return same result
		person0 = list(
		self.storage.person_get([revisions[0]['committer']['id']]))[0]

		person1 = list(
		olasdUnsubmitted Not Done Inline Actions ditto olasd: ditto
		self.storage.person_get([revisions[1]['committer']['id']]))[0]

		self.assertEqual(person0, person1)

def test_snapshot_add_get_empty(self):		def test_snapshot_add_get_empty(self):
origin_id = self.storage.origin_add_one(self.origin)		origin_id = self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_visit1 = self.storage.origin_visit_add(origin_id,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_id = origin_visit1['visit']

self.storage.snapshot_add(origin_id, visit_id, self.empty_snapshot)		self.storage.snapshot_add(origin_id, visit_id, self.empty_snapshot)

▲ Show 20 Lines • Show All 286 Lines • ▼ Show 20 Lines	def test_stat_counters(self):

self.storage.refresh_stat_counters()		self.storage.refresh_stat_counters()
counters = self.storage.stat_counters()		counters = self.storage.stat_counters()
self.assertEqual(counters['content'], 1)		self.assertEqual(counters['content'], 1)
self.assertEqual(counters['directory'], 1)		self.assertEqual(counters['directory'], 1)
self.assertEqual(counters['snapshot'], 1)		self.assertEqual(counters['snapshot'], 1)
self.assertEqual(counters['origin'], 1)		self.assertEqual(counters['origin'], 1)
self.assertEqual(counters['revision'], 1)		self.assertEqual(counters['revision'], 1)
		self.assertEqual(counters['person'], 2)

def test_content_find_with_present_content(self):		def test_content_find_with_present_content(self):
# 1. with something to find		# 1. with something to find
cont = self.cont		cont = self.cont
self.storage.content_add([cont])		self.storage.content_add([cont])

actually_present = self.storage.content_find({'sha1': cont['sha1']})		actually_present = self.storage.content_find({'sha1': cont['sha1']})

▲ Show 20 Lines • Show All 581 Lines • ▼ Show 20 Lines	def test_fetch_history(self):

expected_fetch_history['id'] = fetch_history_id		expected_fetch_history['id'] = fetch_history_id
expected_fetch_history['origin'] = origin		expected_fetch_history['origin'] = origin
expected_fetch_history['date'] = self.fetch_history_date		expected_fetch_history['date'] = self.fetch_history_date
expected_fetch_history['duration'] = self.fetch_history_duration		expected_fetch_history['duration'] = self.fetch_history_duration

self.assertEqual(expected_fetch_history, fetch_history)		self.assertEqual(expected_fetch_history, fetch_history)

# The remote API doesn't expose _person_add
def test_person_get(self):
# given
person0 = {
'fullname': b'bob <alice@bob>',
'name': b'bob',
'email': b'alice@bob',
}
id0 = self.storage._person_add(person0)

person1 = {
'fullname': b'tony <tony@bob>',
'name': b'tony',
'email': b'tony@bob',
}
id1 = self.storage._person_add(person1)

# when
actual_persons = self.storage.person_get([id0, id1])

# given (person injection through release for example)
self.assertEqual(
list(actual_persons), [
{
'id': id0,
'fullname': person0['fullname'],
'name': person0['name'],
'email': person0['email'],
},
{
'id': id1,
'fullname': person1['fullname'],
'name': person1['name'],
'email': person1['email'],
},
])

# This test is only relevant on the local storage, with an actual		# This test is only relevant on the local storage, with an actual
# objstorage raising an exception		# objstorage raising an exception
def test_content_add_objstorage_exception(self):		def test_content_add_objstorage_exception(self):
self.storage.objstorage.add = Mock(		self.storage.objstorage.add = Mock(
side_effect=Exception('mocked broken objstorage')		side_effect=Exception('mocked broken objstorage')
)		)

with self.assertRaises(Exception) as e:		with self.assertRaises(Exception) as e:
▲ Show 20 Lines • Show All 73 Lines • Show Last 20 Lines