Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 656 Lines • ▼ Show 20 Lines | def test_content_add_different_input(self): | ||||
| actual_result = self.storage.content_add([cont, cont2]) | actual_result = self.storage.content_add([cont, cont2]) | ||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 2, | 'content:add': 2, | ||||
| 'content:add:bytes': cont['length'] + cont2['length'], | 'content:add:bytes': cont['length'] + cont2['length'], | ||||
| 'skipped_content:add': 0 | 'skipped_content:add': 0 | ||||
| }) | }) | ||||
| def test_content_add_again(self): | def test_content_add_twice(self): | ||||
| actual_result = self.storage.content_add([self.cont]) | actual_result = self.storage.content_add([self.cont]) | ||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| 'content:add:bytes': self.cont['length'], | 'content:add:bytes': self.cont['length'], | ||||
| 'skipped_content:add': 0 | 'skipped_content:add': 0 | ||||
| }) | }) | ||||
| self.assertEqual(len(self.journal_writer.objects), 1) | |||||
| actual_result = self.storage.content_add([self.cont, self.cont2]) | actual_result = self.storage.content_add([self.cont, self.cont2]) | ||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| 'content:add:bytes': self.cont2['length'], | 'content:add:bytes': self.cont2['length'], | ||||
| 'skipped_content:add': 0 | 'skipped_content:add': 0 | ||||
| }) | }) | ||||
| self.assertEqual(len(self.journal_writer.objects), 2) | |||||
| self.assertEqual(len(self.storage.content_find(self.cont)), 1) | self.assertEqual(len(self.storage.content_find(self.cont)), 1) | ||||
| self.assertEqual(len(self.storage.content_find(self.cont2)), 1) | self.assertEqual(len(self.storage.content_find(self.cont2)), 1) | ||||
| def test_content_add_db(self): | def test_content_add_db(self): | ||||
| cont = self.cont | cont = self.cont | ||||
| actual_result = self.storage.content_add([cont]) | actual_result = self.storage.content_add([cont]) | ||||
| ▲ Show 20 Lines • Show All 337 Lines • ▼ Show 20 Lines | def test_directory_add_validation(self): | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (TypeError, psycopg2.IntegrityError), 'target') as cm: | (TypeError, psycopg2.IntegrityError), 'target') as cm: | ||||
| self.storage.directory_add([dir_]) | self.storage.directory_add([dir_]) | ||||
| if type(cm.exception) == psycopg2.IntegrityError: | if type(cm.exception) == psycopg2.IntegrityError: | ||||
| self.assertEqual(cm.exception.pgcode, | self.assertEqual(cm.exception.pgcode, | ||||
| psycopg2.errorcodes.NOT_NULL_VIOLATION) | psycopg2.errorcodes.NOT_NULL_VIOLATION) | ||||
| def test_directory_add_twice(self): | |||||
| actual_result = self.storage.directory_add([self.dir]) | |||||
| self.assertEqual(actual_result, {'directory:add': 1}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('directory', self.dir)]) | |||||
| actual_result = self.storage.directory_add([self.dir]) | |||||
| self.assertEqual(actual_result, {'directory:add': 0}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('directory', self.dir)]) | |||||
| def test_directory_get_recursive(self): | def test_directory_get_recursive(self): | ||||
| init_missing = list(self.storage.directory_missing([self.dir['id']])) | init_missing = list(self.storage.directory_missing([self.dir['id']])) | ||||
| self.assertEqual([self.dir['id']], init_missing) | self.assertEqual([self.dir['id']], init_missing) | ||||
| actual_result = self.storage.directory_add( | actual_result = self.storage.directory_add( | ||||
| [self.dir, self.dir2, self.dir3]) | [self.dir, self.dir2, self.dir3]) | ||||
| self.assertEqual(actual_result, {'directory:add': 3}) | self.assertEqual(actual_result, {'directory:add': 3}) | ||||
| ▲ Show 20 Lines • Show All 132 Lines • ▼ Show 20 Lines | def test_revision_add(self): | ||||
| self.assertEqual(actual_result, {'revision:add': 1}) | self.assertEqual(actual_result, {'revision:add': 1}) | ||||
| end_missing = self.storage.revision_missing([self.revision['id']]) | end_missing = self.storage.revision_missing([self.revision['id']]) | ||||
| self.assertEqual([], list(end_missing)) | self.assertEqual([], list(end_missing)) | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('revision', self.revision)]) | [('revision', self.revision)]) | ||||
| # already there so nothing added | |||||
| actual_result = self.storage.revision_add([self.revision]) | |||||
| self.assertEqual(actual_result, {'revision:add': 0}) | |||||
| def test_revision_add_validation(self): | def test_revision_add_validation(self): | ||||
| rev = copy.deepcopy(self.revision) | rev = copy.deepcopy(self.revision) | ||||
| rev['date']['offset'] = 2**16 | rev['date']['offset'] = 2**16 | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.DataError), 'offset') as cm: | (ValueError, psycopg2.DataError), 'offset') as cm: | ||||
| self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
| Show All 18 Lines | def test_revision_add_validation(self): | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.DataError), '(?i)type') as cm: | (ValueError, psycopg2.DataError), '(?i)type') as cm: | ||||
| self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
| if type(cm.exception) == psycopg2.DataError: | if type(cm.exception) == psycopg2.DataError: | ||||
| self.assertEqual(cm.exception.pgcode, | self.assertEqual(cm.exception.pgcode, | ||||
| psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION) | psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION) | ||||
| def test_revision_add_twice(self): | |||||
| actual_result = self.storage.revision_add([self.revision]) | |||||
| self.assertEqual(actual_result, {'revision:add': 1}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('revision', self.revision)]) | |||||
| actual_result = self.storage.revision_add( | |||||
| [self.revision, self.revision2]) | |||||
| self.assertEqual(actual_result, {'revision:add': 1}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('revision', self.revision), | |||||
| ('revision', self.revision2)]) | |||||
| def test_revision_add_name_clash(self): | def test_revision_add_name_clash(self): | ||||
| revision1 = self.revision.copy() | revision1 = self.revision.copy() | ||||
| revision2 = self.revision2.copy() | revision2 = self.revision2.copy() | ||||
| revision1['author'] = { | revision1['author'] = { | ||||
| 'fullname': b'John Doe <john.doe@example.com>', | 'fullname': b'John Doe <john.doe@example.com>', | ||||
| 'name': b'John Doe', | 'name': b'John Doe', | ||||
| 'email': b'john.doe@example.com' | 'email': b'john.doe@example.com' | ||||
| } | } | ||||
| ▲ Show 20 Lines • Show All 123 Lines • ▼ Show 20 Lines | def test_release_add(self): | ||||
| end_missing = self.storage.release_missing([self.release['id'], | end_missing = self.storage.release_missing([self.release['id'], | ||||
| self.release2['id']]) | self.release2['id']]) | ||||
| self.assertEqual([], list(end_missing)) | self.assertEqual([], list(end_missing)) | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('release', self.release), | [('release', self.release), | ||||
| ('release', self.release2)]) | ('release', self.release2)]) | ||||
| # already present so nothing added | |||||
| actual_result = self.storage.release_add([self.release, self.release2]) | |||||
| self.assertEqual(actual_result, {'release:add': 0}) | |||||
| def test_release_add_no_author_date(self): | def test_release_add_no_author_date(self): | ||||
| release = self.release.copy() | release = self.release.copy() | ||||
| release['author'] = None | release['author'] = None | ||||
| release['date'] = None | release['date'] = None | ||||
| actual_result = self.storage.release_add([release]) | actual_result = self.storage.release_add([release]) | ||||
| self.assertEqual(actual_result, {'release:add': 1}) | self.assertEqual(actual_result, {'release:add': 1}) | ||||
| Show All 21 Lines | def test_release_add_validation(self): | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.IntegrityError), 'date') as cm: | (ValueError, psycopg2.IntegrityError), 'date') as cm: | ||||
| self.storage.release_add([rel]) | self.storage.release_add([rel]) | ||||
| if type(cm.exception) == psycopg2.IntegrityError: | if type(cm.exception) == psycopg2.IntegrityError: | ||||
| self.assertEqual(cm.exception.pgcode, | self.assertEqual(cm.exception.pgcode, | ||||
| psycopg2.errorcodes.CHECK_VIOLATION) | psycopg2.errorcodes.CHECK_VIOLATION) | ||||
| def test_release_add_twice(self): | |||||
| actual_result = self.storage.release_add([self.release]) | |||||
| self.assertEqual(actual_result, {'release:add': 1}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('release', self.release)]) | |||||
| actual_result = self.storage.release_add([self.release, self.release2]) | |||||
| self.assertEqual(actual_result, {'release:add': 1}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('release', self.release), | |||||
| ('release', self.release2)]) | |||||
| def test_release_add_name_clash(self): | def test_release_add_name_clash(self): | ||||
| release1 = self.release.copy() | release1 = self.release.copy() | ||||
| release2 = self.release2.copy() | release2 = self.release2.copy() | ||||
| release1['author'] = { | release1['author'] = { | ||||
| 'fullname': b'John Doe <john.doe@example.com>', | 'fullname': b'John Doe <john.doe@example.com>', | ||||
| 'name': b'John Doe', | 'name': b'John Doe', | ||||
| 'email': b'john.doe@example.com' | 'email': b'john.doe@example.com' | ||||
| } | } | ||||
| ▲ Show 20 Lines • Show All 67 Lines • ▼ Show 20 Lines | def test_origin_add(self): | ||||
| del actual_origin2['id'] | del actual_origin2['id'] | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', actual_origin), | [('origin', actual_origin), | ||||
| ('origin', actual_origin2)]) | ('origin', actual_origin2)]) | ||||
| def test_origin_add_twice(self): | def test_origin_add_twice(self): | ||||
| add1 = self.storage.origin_add([self.origin, self.origin2]) | add1 = self.storage.origin_add([self.origin, self.origin2]) | ||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('origin', self.origin), | |||||
| ('origin', self.origin2)]) | |||||
| add2 = self.storage.origin_add([self.origin, self.origin2]) | add2 = self.storage.origin_add([self.origin, self.origin2]) | ||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('origin', self.origin), | |||||
| ('origin', self.origin2)]) | |||||
| self.assertEqual(add1, add2) | self.assertEqual(add1, add2) | ||||
| def test_origin_add_validation(self): | def test_origin_add_validation(self): | ||||
| with self.assertRaisesRegex((TypeError, KeyError), 'url'): | with self.assertRaisesRegex((TypeError, KeyError), 'url'): | ||||
| self.storage.origin_add([{'type': 'git'}]) | self.storage.origin_add([{'type': 'git'}]) | ||||
| def test_origin_get_legacy(self): | def test_origin_get_legacy(self): | ||||
| self.assertIsNone(self.storage.origin_get(self.origin)) | self.assertIsNone(self.storage.origin_get(self.origin)) | ||||
| ▲ Show 20 Lines • Show All 1,009 Lines • ▼ Show 20 Lines | def test_snapshot_add_many_incremental(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| {**self.complete_snapshot, 'next_branch': None}, | {**self.complete_snapshot, 'next_branch': None}, | ||||
| self.storage.snapshot_get(self.complete_snapshot['id'])) | self.storage.snapshot_get(self.complete_snapshot['id'])) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| {**self.snapshot, 'next_branch': None}, | {**self.snapshot, 'next_branch': None}, | ||||
| self.storage.snapshot_get(self.snapshot['id'])) | self.storage.snapshot_get(self.snapshot['id'])) | ||||
| def test_snapshot_add_twice(self): | |||||
| actual_result = self.storage.snapshot_add([self.empty_snapshot]) | |||||
| self.assertEqual(actual_result, {'snapshot:add': 1}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('snapshot', self.empty_snapshot)]) | |||||
| actual_result = self.storage.snapshot_add([self.snapshot]) | |||||
| self.assertEqual(actual_result, {'snapshot:add': 1}) | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('snapshot', self.empty_snapshot), | |||||
| ('snapshot', self.snapshot)]) | |||||
| def test_snapshot_add_validation(self): | def test_snapshot_add_validation(self): | ||||
| snap = copy.deepcopy(self.snapshot) | snap = copy.deepcopy(self.snapshot) | ||||
| snap['branches'][b'foo'] = {'target_type': 'revision'} | snap['branches'][b'foo'] = {'target_type': 'revision'} | ||||
| with self.assertRaisesRegex(KeyError, 'target'): | with self.assertRaisesRegex(KeyError, 'target'): | ||||
| self.storage.snapshot_add([snap]) | self.storage.snapshot_add([snap]) | ||||
| snap = copy.deepcopy(self.snapshot) | snap = copy.deepcopy(self.snapshot) | ||||
| ▲ Show 20 Lines • Show All 243 Lines • ▼ Show 20 Lines | def test_snapshot_add_nonexistent_visit(self): | ||||
| with self.assertRaises(ValueError): | with self.assertRaises(ValueError): | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=self.snapshot['id']) | origin_id, visit_id, snapshot=self.snapshot['id']) | ||||
| self.assertEqual(list(self.journal_writer.objects), [ | self.assertEqual(list(self.journal_writer.objects), [ | ||||
| ('snapshot', self.snapshot)]) | ('snapshot', self.snapshot)]) | ||||
| def test_snapshot_add_twice(self): | def test_snapshot_add_twice__by_origin_visit(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) | ||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| self.storage.snapshot_add([self.snapshot]) | self.storage.snapshot_add([self.snapshot]) | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit1_id, snapshot=self.snapshot['id']) | origin_id, visit1_id, snapshot=self.snapshot['id']) | ||||
| ▲ Show 20 Lines • Show All 1,269 Lines • Show Last 20 Lines | |||||