Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| # Copyright (C) 2015-2019 The Software Heritage developers | # Copyright (C) 2015-2019 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import copy | import copy | ||||
| import datetime | import datetime | ||||
| import itertools | import itertools | ||||
| import random | import random | ||||
| import unittest | import unittest | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| from unittest.mock import Mock, patch | from unittest.mock import Mock, patch | ||||
| import psycopg2.errors | import psycopg2 | ||||
| import pytest | import pytest | ||||
| from hypothesis import given, strategies, settings, HealthCheck | from hypothesis import given, strategies, settings, HealthCheck | ||||
| from swh.model import from_disk, identifiers | from swh.model import from_disk, identifiers | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.hypothesis_strategies import origins, objects | from swh.model.hypothesis_strategies import origins, objects | ||||
| from swh.storage.tests.storage_testing import StorageTestFixture | from swh.storage.tests.storage_testing import StorageTestFixture | ||||
| ▲ Show 20 Lines • Show All 576 Lines • ▼ Show 20 Lines | def test_content_add_validation(self): | ||||
| with self.assertRaisesRegex(ValueError, 'status'): | with self.assertRaisesRegex(ValueError, 'status'): | ||||
| self.storage.content_add([{**cont, 'status': 'foobar'}]) | self.storage.content_add([{**cont, 'status': 'foobar'}]) | ||||
| with self.assertRaisesRegex(ValueError, "(?i)length"): | with self.assertRaisesRegex(ValueError, "(?i)length"): | ||||
| self.storage.content_add([{**cont, 'length': -2}]) | self.storage.content_add([{**cont, 'length': -2}]) | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.errors.NotNullViolation), | (ValueError, psycopg2.IntegrityError), 'reason'): | ||||
| "reason"): | |||||
| self.storage.content_add([{**cont, 'status': 'absent'}]) | self.storage.content_add([{**cont, 'status': 'absent'}]) | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| ValueError, | ValueError, | ||||
| "^Must not provide a reason if content is not absent.$"): | "^Must not provide a reason if content is not absent.$"): | ||||
| self.storage.content_add([{**cont, 'reason': 'foobar'}]) | self.storage.content_add([{**cont, 'reason': 'foobar'}]) | ||||
| def test_content_get_missing(self): | def test_content_get_missing(self): | ||||
| ▲ Show 20 Lines • Show All 396 Lines • ▼ Show 20 Lines | def test_directory_add_validation(self): | ||||
| with self.assertRaisesRegex(ValueError, 'type.*foobar'): | with self.assertRaisesRegex(ValueError, 'type.*foobar'): | ||||
| self.storage.directory_add([dir_]) | self.storage.directory_add([dir_]) | ||||
| dir_ = copy.deepcopy(self.dir) | dir_ = copy.deepcopy(self.dir) | ||||
| del dir_['entries'][0]['target'] | del dir_['entries'][0]['target'] | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (TypeError, psycopg2.errors.NotNullViolation), 'target'): | (TypeError, psycopg2.IntegrityError), 'target'): | ||||
| self.storage.directory_add([dir_]) | self.storage.directory_add([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]) | ||||
| ▲ Show 20 Lines • Show All 143 Lines • ▼ Show 20 Lines | def test_revision_add(self): | ||||
| actual_result = self.storage.revision_add([self.revision]) | actual_result = self.storage.revision_add([self.revision]) | ||||
| self.assertEqual(actual_result, {'revision:add': 0}) | 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.errors.NumericValueOutOfRange), | (ValueError, psycopg2.DataError), 'offset'): | ||||
| 'offset'): | |||||
| self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
| rev = copy.deepcopy(self.revision) | rev = copy.deepcopy(self.revision) | ||||
| rev['committer_date']['offset'] = 2**16 | rev['committer_date']['offset'] = 2**16 | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.errors.NumericValueOutOfRange), | (ValueError, psycopg2.DataError), 'offset'): | ||||
| 'offset'): | |||||
| self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
| rev = copy.deepcopy(self.revision) | rev = copy.deepcopy(self.revision) | ||||
| rev['type'] = 'foobar' | rev['type'] = 'foobar' | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.errors.InvalidTextRepresentation), | (ValueError, psycopg2.DataError), '(?i)type'): | ||||
| '(?i)type'): | |||||
| self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
| 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', | ||||
| ▲ Show 20 Lines • Show All 148 Lines • ▼ Show 20 Lines | def test_release_add_no_author_date(self): | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('release', release)]) | [('release', release)]) | ||||
| def test_release_add_validation(self): | def test_release_add_validation(self): | ||||
| rel = copy.deepcopy(self.release) | rel = copy.deepcopy(self.release) | ||||
| rel['date']['offset'] = 2**16 | rel['date']['offset'] = 2**16 | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.errors.NumericValueOutOfRange), | (ValueError, psycopg2.DataError), 'offset'): | ||||
| 'offset'): | |||||
| self.storage.release_add([rel]) | self.storage.release_add([rel]) | ||||
| rel = copy.deepcopy(self.release) | rel = copy.deepcopy(self.release) | ||||
| rel['author'] = None | rel['author'] = None | ||||
| with self.assertRaisesRegex( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.errors.CheckViolation), | (ValueError, psycopg2.IntegrityError), 'date'): | ||||
| 'date'): | |||||
| self.storage.release_add([rel]) | self.storage.release_add([rel]) | ||||
| 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', | ||||
| ▲ Show 20 Lines • Show All 386 Lines • ▼ Show 20 Lines | def test_origin_visit_add_default_type(self, use_url): | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', expected_origin), | [('origin', expected_origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)]) | ('origin_visit', data2)]) | ||||
| def test_origin_visit_add_validation(self): | def test_origin_visit_add_validation(self): | ||||
| origin_id_or_url = self.storage.origin_add_one(self.origin2) | origin_id_or_url = self.storage.origin_add_one(self.origin2) | ||||
| with self.assertRaises((TypeError, psycopg2.errors.UndefinedFunction)): | with self.assertRaises((TypeError, psycopg2.ProgrammingError)): | ||||
| self.storage.origin_visit_add(origin_id_or_url, date=[b'foo']) | self.storage.origin_visit_add(origin_id_or_url, date=[b'foo']) | ||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_update(self, use_url): | def test_origin_visit_update(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | if not self._test_origin_ids and not use_url: | ||||
| return | return | ||||
| self.reset_storage() | self.reset_storage() | ||||
| ▲ Show 20 Lines • Show All 148 Lines • ▼ Show 20 Lines | def test_origin_visit_update(self, use_url): | ||||
| ('origin_visit', data5)]) | ('origin_visit', data5)]) | ||||
| def test_origin_visit_update_validation(self): | def test_origin_visit_update_validation(self): | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| visit = self.storage.origin_visit_add( | visit = self.storage.origin_visit_add( | ||||
| origin_id, | origin_id, | ||||
| date=self.date_visit2) | date=self.date_visit2) | ||||
| with self.assertRaisesRegexp( | with self.assertRaisesRegex( | ||||
| (ValueError, psycopg2.errors.InvalidTextRepresentation), | (ValueError, psycopg2.DataError), 'status'): | ||||
| 'status'): | |||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit['visit'], status='foobar') | origin_id, visit['visit'], status='foobar') | ||||
| def test_origin_visit_find_by_date(self): | def test_origin_visit_find_by_date(self): | ||||
| # given | # given | ||||
| self.storage.origin_add_one(self.origin) | self.storage.origin_add_one(self.origin) | ||||
| self.storage.origin_visit_add( | self.storage.origin_visit_add( | ||||
| ▲ Show 20 Lines • Show All 2,067 Lines • Show Last 20 Lines | |||||