Differential D646 Diff 2060 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

# Copyright (C) 2015-2017 The Software Heritage developers		# Copyright (C) 2015-2018 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 unittest		import unittest
from collections import defaultdict		from collections import defaultdict
from operator import itemgetter		from operator import itemgetter
from unittest.mock import Mock, patch		from unittest.mock import Mock, patch

import psycopg2		import psycopg2
import pytest		import pytest

		from hypothesis import given

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.storage.tests.storage_testing import StorageTestFixture		from swh.storage.tests.storage_testing import StorageTestFixture

		from . import gen_contents


@pytest.mark.db		@pytest.mark.db
class BaseTestStorage(StorageTestFixture):		class BaseTestStorage(StorageTestFixture):
def setUp(self):		def setUp(self):
super().setUp()		super().setUp()

db = self.test_db[self.TEST_DB_NAME]		db = self.test_db[self.TEST_DB_NAME]
self.conn = db.conn		self.conn = db.conn
▲ Show 20 Lines • Show All 1,809 Lines • ▼ Show 20 Lines	def test_origin_metadata_get_by_provider_type(self):
# then		# then

self.assertEqual(len(m_by_provider), 1)		self.assertEqual(len(m_by_provider), 1)
self.assertEqual(m_by_provider, expected_results)		self.assertEqual(m_by_provider, expected_results)
self.assertEqual(m_by_provider[0]['id'], o_m2)		self.assertEqual(m_by_provider[0]['id'], o_m2)
self.assertIsNotNone(o_m1)		self.assertIsNotNone(o_m1)


		@pytest.mark.property_based
		class PropBasedTestStorage(BaseTestStorage, unittest.TestCase):
		def assert_contents_ok(self, expected_contents, actual_contents,
		keys_to_check={'sha1', 'data'}):
		"""Assert that a given list of contents matches on a given set of keys.

		"""
		for k in keys_to_check:
		expected_list = [c[k] for c in expected_contents]
		expected_list.sort()
		actual_list = [c[k] for c in actual_contents]
		vlorentzUnsubmitted Not Done Inline Actions Nitpick: this code is equivalent and simpler: expected_list = sorted(c[k] for c in expected_contents) actual_list = sorted(c[k] for c in actual_contents) vlorentz: Nitpick: this code is equivalent and simpler: ``` expected_list = sorted(c[k] for c in…
		ardumontAuthorUnsubmitted Done Inline Actions Right! Thanks. ardumont: Right! Thanks.
		actual_list.sort()

		self.assertEqual(actual_list, expected_list)

		@given(gen_contents(min_size=1, max_size=4))
		def test_generate_content_get(self, contents):
		# add contents to storage
		self.storage.content_add(contents)

		# input the list of sha1s we want from storage
		get_sha1s = [c['sha1'] for c in contents]

		# retrieve contents
		actual_contents = list(self.storage.content_get(get_sha1s))

		self.assert_contents_ok(contents, actual_contents)

		@given(gen_contents(min_size=1, max_size=4))
		def test_generate_content_get_metadata(self, contents):
		# add contents to storage
		self.storage.content_add(contents)

		# input the list of sha1s we want from storage
		get_sha1s = [c['sha1'] for c in contents]

		# retrieve contents
		actual_contents = list(self.storage.content_get_metadata(get_sha1s))

		ardumontAuthorUnsubmitted Done Inline Actions assertEqual here. ardumont: assertEqual here.
		vlorentzUnsubmitted Not Done Inline Actions ? vlorentz: ?
		ardumontAuthorUnsubmitted Done Inline Actions assertEquals is deprecated ;) It's a warning from pytest. ardumont: assertEquals is deprecated ;) It's a warning from pytest.
		self.assertEquals(len(actual_contents), len(contents))

		# will check that all contents are retrieved correctly
		one_content = contents[0]
		# content_get_metadata does not return data
		keys_to_check = set(one_content.keys()) - {'data'}
		self.assert_contents_ok(contents, actual_contents,
		vlorentzUnsubmitted Not Done Inline Actions I don't think this properly tests `contents`. What if `contents[0]` is an empty dict? vlorentz: I don't think this properly tests `contents`. What if `contents[0]` is an empty dict?
		ardumontAuthorUnsubmitted Done Inline Actions That should not be the case since it passes through the `content_add`, that'd break earlier. ardumont: That should not be the case since it passes through the `content_add`, that'd break earlier.
		keys_to_check=keys_to_check)


class TestLocalStorage(CommonTestStorage, unittest.TestCase):		class TestLocalStorage(CommonTestStorage, unittest.TestCase):
"""Test the local storage"""		"""Test the local storage"""

# Can only be tested with local storage as you can't mock		# Can only be tested with local storage as you can't mock
# datetimes for the remote server		# datetimes for the remote server
def test_fetch_history(self):		def test_fetch_history(self):
origin = self.storage.origin_add_one(self.origin)		origin = self.storage.origin_add_one(self.origin)
with patch('datetime.datetime'):		with patch('datetime.datetime'):
▲ Show 20 Lines • Show All 129 Lines • Show Last 20 Lines