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 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 566 Lines • ▼ Show 20 Lines | def test_content_add(self): | ||||
| journal_objects = list(self.journal_writer.objects) | journal_objects = list(self.journal_writer.objects) | ||||
| for (obj_type, obj) in journal_objects: | for (obj_type, obj) in journal_objects: | ||||
| self.assertLessEqual(insertion_start_time, obj['ctime']) | self.assertLessEqual(insertion_start_time, obj['ctime']) | ||||
| self.assertLessEqual(obj['ctime'], insertion_end_time) | self.assertLessEqual(obj['ctime'], insertion_end_time) | ||||
| del obj['ctime'] | del obj['ctime'] | ||||
| self.assertEqual(journal_objects, | self.assertEqual(journal_objects, | ||||
| [('content', expected_cont)]) | [('content', expected_cont)]) | ||||
| @pytest.mark.db | |||||
| def test_content_add_validation(self): | def test_content_add_validation(self): | ||||
| import psycopg2.errors | |||||
| cont = self.cont | cont = self.cont | ||||
| 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}]) | ||||
| ▲ Show 20 Lines • Show All 396 Lines • ▼ Show 20 Lines | def test_directory_add(self): | ||||
| actual_data = list(self.storage.directory_ls(self.dir['id'])) | actual_data = list(self.storage.directory_ls(self.dir['id'])) | ||||
| expected_data = list(self._transform_entries(self.dir)) | expected_data = list(self._transform_entries(self.dir)) | ||||
| self.assertCountEqual(expected_data, actual_data) | self.assertCountEqual(expected_data, actual_data) | ||||
| after_missing = list(self.storage.directory_missing([self.dir['id']])) | after_missing = list(self.storage.directory_missing([self.dir['id']])) | ||||
| self.assertEqual([], after_missing) | self.assertEqual([], after_missing) | ||||
| @pytest.mark.db | |||||
| def test_directory_add_validation(self): | def test_directory_add_validation(self): | ||||
| import psycopg2.errors | |||||
| dir_ = copy.deepcopy(self.dir) | dir_ = copy.deepcopy(self.dir) | ||||
| dir_['entries'][0]['type'] = 'foobar' | dir_['entries'][0]['type'] = 'foobar' | ||||
| 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'] | ||||
| ▲ Show 20 Lines • Show All 149 Lines • ▼ Show 20 Lines | def test_revision_add(self): | ||||
| 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 | # already there so nothing added | ||||
| 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}) | ||||
| @pytest.mark.db | |||||
| def test_revision_add_validation(self): | def test_revision_add_validation(self): | ||||
| import psycopg2.errors | |||||
| 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.errors.NumericValueOutOfRange), | ||||
| 'offset'): | 'offset'): | ||||
| self.storage.revision_add([rev]) | self.storage.revision_add([rev]) | ||||
| ▲ Show 20 Lines • Show All 165 Lines • ▼ Show 20 Lines | def test_release_add_no_author_date(self): | ||||
| self.assertEqual(actual_result, {'release:add': 1}) | self.assertEqual(actual_result, {'release:add': 1}) | ||||
| end_missing = self.storage.release_missing([self.release['id']]) | end_missing = self.storage.release_missing([self.release['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', release)]) | [('release', release)]) | ||||
| @pytest.mark.db | |||||
| def test_release_add_validation(self): | def test_release_add_validation(self): | ||||
| import psycopg2.errors | |||||
| 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.errors.NumericValueOutOfRange), | ||||
| 'offset'): | 'offset'): | ||||
| self.storage.release_add([rel]) | self.storage.release_add([rel]) | ||||
| ▲ Show 20 Lines • Show All 397 Lines • ▼ Show 20 Lines | def test_origin_visit_add_default_type(self, use_url): | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| 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)]) | ||||
| @pytest.mark.db | |||||
| def test_origin_visit_add_validation(self): | def test_origin_visit_add_validation(self): | ||||
| import psycopg2.errors | |||||
| 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.errors.UndefinedFunction)): | ||||
| 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: | ||||
| ▲ Show 20 Lines • Show All 145 Lines • ▼ Show 20 Lines | def test_origin_visit_update(self, use_url): | ||||
| [('origin', expected_origin), | [('origin', expected_origin), | ||||
| ('origin', expected_origin2), | ('origin', expected_origin2), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2), | ('origin_visit', data2), | ||||
| ('origin_visit', data3), | ('origin_visit', data3), | ||||
| ('origin_visit', data4), | ('origin_visit', data4), | ||||
| ('origin_visit', data5)]) | ('origin_visit', data5)]) | ||||
| @pytest.mark.db | |||||
| def test_origin_visit_update_validation(self): | def test_origin_visit_update_validation(self): | ||||
| import psycopg2.errors | |||||
| 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.assertRaisesRegexp( | ||||
| (ValueError, psycopg2.errors.InvalidTextRepresentation), | (ValueError, psycopg2.errors.InvalidTextRepresentation), | ||||
| 'status'): | 'status'): | ||||
| ▲ Show 20 Lines • Show All 2,075 Lines • Show Last 20 Lines | |||||