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') as cm: | ||||
"reason"): | |||||
self.storage.content_add([{**cont, 'status': 'absent'}]) | self.storage.content_add([{**cont, 'status': 'absent'}]) | ||||
if type(cm.exception) == psycopg2.IntegrityError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.NOT_NULL_VIOLATION) | |||||
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): | ||||
cont = self.cont | cont = self.cont | ||||
▲ Show 20 Lines • Show All 394 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') as cm: | ||||
self.storage.directory_add([dir_]) | self.storage.directory_add([dir_]) | ||||
if type(cm.exception) == psycopg2.IntegrityError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.NOT_NULL_VIOLATION) | |||||
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 141 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') as cm: | ||||
'offset'): | |||||
self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
if type(cm.exception) == psycopg2.DataError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.NUMERIC_VALUE_OUT_OF_RANGE) | |||||
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') as cm: | ||||
'offset'): | |||||
self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
if type(cm.exception) == psycopg2.DataError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.NUMERIC_VALUE_OUT_OF_RANGE) | |||||
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') as cm: | ||||
'(?i)type'): | |||||
self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
if type(cm.exception) == psycopg2.DataError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION) | |||||
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 146 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') as cm: | ||||
'offset'): | |||||
self.storage.release_add([rel]) | self.storage.release_add([rel]) | ||||
if type(cm.exception) == psycopg2.DataError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.NUMERIC_VALUE_OUT_OF_RANGE) | |||||
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') as cm: | ||||
'date'): | |||||
self.storage.release_add([rel]) | self.storage.release_add([rel]) | ||||
if type(cm.exception) == psycopg2.IntegrityError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.CHECK_VIOLATION) | |||||
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 384 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)) as cm: | ||||
self.storage.origin_visit_add(origin_id_or_url, date=[b'foo']) | self.storage.origin_visit_add(origin_id_or_url, date=[b'foo']) | ||||
if type(cm.exception) == psycopg2.ProgrammingError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.UNDEFINED_FUNCTION) | |||||
@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() | ||||
# given | # given | ||||
origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
▲ Show 20 Lines • Show All 146 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') as cm: | ||||
'status'): | |||||
self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
origin_id, visit['visit'], status='foobar') | origin_id, visit['visit'], status='foobar') | ||||
if type(cm.exception) == psycopg2.DataError: | |||||
self.assertEqual(cm.exception.pgcode, | |||||
psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION) | |||||
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( | ||||
self.origin['url'], | self.origin['url'], | ||||
date=self.date_visit2) | date=self.date_visit2) | ||||
▲ Show 20 Lines • Show All 2,064 Lines • Show Last 20 Lines |