Differential D2089 Diff 7012 swh/storage/tests/test_storage.py

Changeset 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
from contextlib import contextmanager		from contextlib import contextmanager
import datetime		import datetime
import itertools		import itertools
import queue		import queue
import random		import random
import threading		import threading
import unittest		import unittest
from collections import defaultdict		from collections import defaultdict
from unittest.mock import Mock, patch		from unittest.mock import Mock

import psycopg2		import psycopg2
import pytest		import pytest

from hypothesis import given, strategies, settings, HealthCheck		from hypothesis import given, strategies, settings, HealthCheck

from typing import ClassVar, Optional		from typing import ClassVar, Optional

▲ Show 20 Lines • Show All 536 Lines • ▼ Show 20 Lines	class CommonTestStorage(TestStorageData):
test_remote_storage.py.		test_remote_storage.py.

We need to have the two classes inherit from this base class		We need to have the two classes inherit from this base class
separately to avoid nosetests running the tests from the base		separately to avoid nosetests running the tests from the base
class twice.		class twice.

"""		"""
maxDiff = None # type: ClassVar[Optional[int]]		maxDiff = None # type: ClassVar[Optional[int]]
_test_origin_ids = True

@staticmethod		@staticmethod
def normalize_entity(entity):		def normalize_entity(entity):
entity = copy.deepcopy(entity)		entity = copy.deepcopy(entity)
for key in ('date', 'committer_date'):		for key in ('date', 'committer_date'):
if key in entity:		if key in entity:
entity[key] = identifiers.normalize_timestamp(entity[key])		entity[key] = identifiers.normalize_timestamp(entity[key])

▲ Show 20 Lines • Show All 906 Lines • ▼ Show 20 Lines	class CommonTestStorage(TestStorageData):

def test_origin_add_one(self):		def test_origin_add_one(self):
origin0 = self.storage.origin_get(self.origin)		origin0 = self.storage.origin_get(self.origin)
self.assertIsNone(origin0)		self.assertIsNone(origin0)

id = self.storage.origin_add_one(self.origin)		id = self.storage.origin_add_one(self.origin)

actual_origin = self.storage.origin_get({'url': self.origin['url']})		actual_origin = self.storage.origin_get({'url': self.origin['url']})
if self._test_origin_ids:
self.assertEqual(actual_origin['id'], id)
self.assertEqual(actual_origin['url'], self.origin['url'])		self.assertEqual(actual_origin['url'], self.origin['url'])

id2 = self.storage.origin_add_one(self.origin)		id2 = self.storage.origin_add_one(self.origin)

self.assertEqual(id, id2)		self.assertEqual(id, id2)

def test_origin_add(self):		def test_origin_add(self):
origin0 = self.storage.origin_get([self.origin])[0]		origin0 = self.storage.origin_get([self.origin])[0]
self.assertIsNone(origin0)		self.assertIsNone(origin0)

origin1, origin2 = self.storage.origin_add([self.origin, self.origin2])		origin1, origin2 = self.storage.origin_add([self.origin, self.origin2])

actual_origin = self.storage.origin_get([{		actual_origin = self.storage.origin_get([{
'url': self.origin['url'],		'url': self.origin['url'],
}])[0]		}])[0]
if self._test_origin_ids:
self.assertEqual(actual_origin['id'], origin1['id'])
self.assertEqual(actual_origin['url'], origin1['url'])		self.assertEqual(actual_origin['url'], origin1['url'])

actual_origin2 = self.storage.origin_get([{		actual_origin2 = self.storage.origin_get([{
'url': self.origin2['url'],		'url': self.origin2['url'],
}])[0]		}])[0]
if self._test_origin_ids:
self.assertEqual(actual_origin2['id'], origin2['id'])
self.assertEqual(actual_origin2['url'], origin2['url'])		self.assertEqual(actual_origin2['url'], origin2['url'])

if 'id' in actual_origin:		if 'id' in actual_origin:
del actual_origin['id']		del actual_origin['id']
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),
Show All 15 Lines	def test_origin_add_twice(self):
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))
id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)

# lookup per url (returns id)		# lookup per url
actual_origin0 = self.storage.origin_get(		actual_origin0 = self.storage.origin_get(
{'url': self.origin['url']})		{'url': self.origin['url']})
if self._test_origin_ids:
self.assertEqual(actual_origin0['id'], id)
self.assertEqual(actual_origin0['url'], self.origin['url'])		self.assertEqual(actual_origin0['url'], self.origin['url'])

# lookup per id (returns dict)
if self._test_origin_ids:
actual_origin1 = self.storage.origin_get({'id': id})

self.assertEqual(actual_origin1, {'id': id,
'type': self.origin['type'],
'url': self.origin['url']})

def test_origin_get(self):		def test_origin_get(self):
self.assertIsNone(self.storage.origin_get(self.origin))		self.assertIsNone(self.storage.origin_get(self.origin))
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)

# lookup per url (returns id)		# lookup per url
actual_origin0 = self.storage.origin_get(		actual_origin0 = self.storage.origin_get(
[{'url': self.origin['url']}])		[{'url': self.origin['url']}])
self.assertEqual(len(actual_origin0), 1, actual_origin0)		self.assertEqual(len(actual_origin0), 1, actual_origin0)
if self._test_origin_ids:
self.assertEqual(actual_origin0[0]['id'], origin_id)
self.assertEqual(actual_origin0[0]['url'], self.origin['url'])		self.assertEqual(actual_origin0[0]['url'], self.origin['url'])

if self._test_origin_ids:
# lookup per id (returns dict)
actual_origin1 = self.storage.origin_get([{'id': origin_id}])

self.assertEqual(len(actual_origin1), 1, actual_origin1)
self.assertEqual(actual_origin1[0], {'id': origin_id,
'type': self.origin['type'],
'url': self.origin['url']})

def test_origin_get_consistency(self):		def test_origin_get_consistency(self):
self.assertIsNone(self.storage.origin_get(self.origin))		self.assertIsNone(self.storage.origin_get(self.origin))
id = self.storage.origin_add_one(self.origin)		id = self.storage.origin_add_one(self.origin)

with self.assertRaises(ValueError):		with self.assertRaises(ValueError):
self.storage.origin_get([		self.storage.origin_get([
{'url': self.origin['url']},		{'url': self.origin['url']},
{'id': id}])		{'id': id}])
▲ Show 20 Lines • Show All 123 Lines • ▼ Show 20 Lines	def test_origin_search_regexp_fullstring(self):
self.assertIn(found_origins1[0], [origin, origin2])		self.assertIn(found_origins1[0], [origin, origin2])

# check both origins were returned		# check both origins were returned

self.assertCountEqual(		self.assertCountEqual(
found_origins0 + found_origins1,		found_origins0 + found_origins1,
[origin, origin2])		[origin, origin2])

@given(strategies.booleans())		def test_origin_visit_add(self):
def test_origin_visit_add(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

# given		# given
self.assertIsNone(self.storage.origin_get([self.origin2])[0])		self.assertIsNone(self.storage.origin_get([self.origin2])[0])

origin_id = self.storage.origin_add_one(self.origin2)		origin_url = self.storage.origin_add_one(self.origin2)
self.assertIsNotNone(origin_id)		self.assertIsNotNone(origin_url)

origin_id_or_url = self.origin2['url'] if use_url else origin_id		origin_url = self.origin2['url']

# when		# when
origin_visit1 = self.storage.origin_visit_add(		origin_visit1 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
type='git',		type='git',
date=self.date_visit2)		date=self.date_visit2)

actual_origin_visits = list(self.storage.origin_visit_get(		actual_origin_visits = list(self.storage.origin_visit_get(
origin_id_or_url))		origin_url))
self.assertEqual(actual_origin_visits,		self.assertEqual(actual_origin_visits,
[{		[{
'origin': origin_id,		'origin': origin_url,
'date': self.date_visit2,		'date': self.date_visit2,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': 'git',		'type': 'git',
'status': 'ongoing',		'status': 'ongoing',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
}])		}])

expected_origin = self.origin2.copy()		expected_origin = self.origin2.copy()
data = {		data = {
'origin': expected_origin,		'origin': expected_origin,
'date': self.date_visit2,		'date': self.date_visit2,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': 'git',		'type': 'git',
'status': 'ongoing',		'status': 'ongoing',
'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', data)])		('origin_visit', data)])

def test_origin_visit_get__unknown_origin(self):		def test_origin_visit_get__unknown_origin(self):
self.assertEqual([], list(self.storage.origin_visit_get('foo')))		self.assertEqual([], list(self.storage.origin_visit_get('foo')))
if self._test_origin_ids:
self.assertEqual([], list(self.storage.origin_visit_get(10)))

@given(strategies.booleans())		def test_origin_visit_add_default_type(self):
def test_origin_visit_add_default_type(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

# given		# given
self.assertIsNone(self.storage.origin_get([self.origin2])[0])		self.assertIsNone(self.storage.origin_get([self.origin2])[0])

origin_id = self.storage.origin_add_one(self.origin2)		origin_url = self.storage.origin_add_one(self.origin2)
origin_id_or_url = self.origin2['url'] if use_url else origin_id		self.assertIsNotNone(origin_url)
self.assertIsNotNone(origin_id)

# when		# when
origin_visit1 = self.storage.origin_visit_add(		origin_visit1 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date=self.date_visit2)		date=self.date_visit2)
origin_visit2 = self.storage.origin_visit_add(		origin_visit2 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date='2018-01-01 23:00:00+00')		date='2018-01-01 23:00:00+00')

# then		# then
self.assertEqual(origin_visit1['origin'], origin_id)		self.assertEqual(origin_visit1['origin'], origin_url)
self.assertIsNotNone(origin_visit1['visit'])		self.assertIsNotNone(origin_visit1['visit'])

actual_origin_visits = list(self.storage.origin_visit_get(		actual_origin_visits = list(self.storage.origin_visit_get(
origin_id_or_url))		origin_url))
self.assertEqual(actual_origin_visits, [		self.assertEqual(actual_origin_visits, [
{		{
'origin': origin_id,		'origin': origin_url,
'date': self.date_visit2,		'date': self.date_visit2,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': 'hg',		'type': 'hg',
'status': 'ongoing',		'status': 'ongoing',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
},		},
{		{
'origin': origin_id,		'origin': origin_url,
'date': self.date_visit3,		'date': self.date_visit3,
'visit': origin_visit2['visit'],		'visit': origin_visit2['visit'],
'type': 'hg',		'type': 'hg',
'status': 'ongoing',		'status': 'ongoing',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
},		},
])		])
Show All 18 Lines	def test_origin_visit_add_default_type(self):
'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)])

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_url = self.storage.origin_add_one(self.origin2)

with self.assertRaises((TypeError, psycopg2.ProgrammingError)) as cm:		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_url, date=[b'foo'])

if type(cm.exception) == psycopg2.ProgrammingError:		if type(cm.exception) == psycopg2.ProgrammingError:
self.assertEqual(cm.exception.pgcode,		self.assertEqual(cm.exception.pgcode,
psycopg2.errorcodes.UNDEFINED_FUNCTION)		psycopg2.errorcodes.UNDEFINED_FUNCTION)

@given(strategies.booleans())		def test_origin_visit_update(self):
def test_origin_visit_update(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

# given		# given
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_id2 = self.storage.origin_add_one(self.origin2)		self.storage.origin_add_one(self.origin2)
origin2_id_or_url = self.origin2['url'] if use_url else origin_id2		origin2_url = self.origin2['url']

origin_id_or_url = self.origin['url'] if use_url else origin_id		origin_url = self.origin['url']

origin_visit1 = self.storage.origin_visit_add(		origin_visit1 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date=self.date_visit2)		date=self.date_visit2)

origin_visit2 = self.storage.origin_visit_add(		origin_visit2 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date=self.date_visit3)		date=self.date_visit3)

origin_visit3 = self.storage.origin_visit_add(		origin_visit3 = self.storage.origin_visit_add(
origin2_id_or_url,		origin2_url,
date=self.date_visit3)		date=self.date_visit3)

# when		# when
visit1_metadata = {		visit1_metadata = {
'contents': 42,		'contents': 42,
'directories': 22,		'directories': 22,
}		}
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
origin_visit1['visit'], status='full',		origin_visit1['visit'], status='full',
metadata=visit1_metadata)		metadata=visit1_metadata)
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin2_id_or_url,		origin2_url,
origin_visit3['visit'], status='partial')		origin_visit3['visit'], status='partial')

# then		# then
actual_origin_visits = list(self.storage.origin_visit_get(		actual_origin_visits = list(self.storage.origin_visit_get(
origin_id_or_url))		origin_url))
self.assertEqual(actual_origin_visits, [{		self.assertEqual(actual_origin_visits, [{
'origin': origin_visit2['origin'],		'origin': origin_visit2['origin'],
'date': self.date_visit2,		'date': self.date_visit2,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': self.origin['type'],		'type': self.origin['type'],
'status': 'full',		'status': 'full',
'metadata': visit1_metadata,		'metadata': visit1_metadata,
'snapshot': None,		'snapshot': None,
}, {		}, {
'origin': origin_visit2['origin'],		'origin': origin_visit2['origin'],
'date': self.date_visit3,		'date': self.date_visit3,
'visit': origin_visit2['visit'],		'visit': origin_visit2['visit'],
'type': self.origin['type'],		'type': self.origin['type'],
'status': 'ongoing',		'status': 'ongoing',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
}])		}])

actual_origin_visits_bis = list(self.storage.origin_visit_get(		actual_origin_visits_bis = list(self.storage.origin_visit_get(
origin_id_or_url,		origin_url,
limit=1))		limit=1))
self.assertEqual(actual_origin_visits_bis,		self.assertEqual(actual_origin_visits_bis,
[{		[{
'origin': origin_visit2['origin'],		'origin': origin_visit2['origin'],
'date': self.date_visit2,		'date': self.date_visit2,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': self.origin['type'],		'type': self.origin['type'],
'status': 'full',		'status': 'full',
'metadata': visit1_metadata,		'metadata': visit1_metadata,
'snapshot': None,		'snapshot': None,
}])		}])

actual_origin_visits_ter = list(self.storage.origin_visit_get(		actual_origin_visits_ter = list(self.storage.origin_visit_get(
origin_id_or_url,		origin_url,
last_visit=origin_visit1['visit']))		last_visit=origin_visit1['visit']))
self.assertEqual(actual_origin_visits_ter,		self.assertEqual(actual_origin_visits_ter,
[{		[{
'origin': origin_visit2['origin'],		'origin': origin_visit2['origin'],
'date': self.date_visit3,		'date': self.date_visit3,
'visit': origin_visit2['visit'],		'visit': origin_visit2['visit'],
'type': self.origin['type'],		'type': self.origin['type'],
'status': 'ongoing',		'status': 'ongoing',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
}])		}])

actual_origin_visits2 = list(self.storage.origin_visit_get(		actual_origin_visits2 = list(self.storage.origin_visit_get(
origin2_id_or_url))		origin2_url))
self.assertEqual(actual_origin_visits2,		self.assertEqual(actual_origin_visits2,
[{		[{
'origin': origin_visit3['origin'],		'origin': origin_visit3['origin'],
'date': self.date_visit3,		'date': self.date_visit3,
'visit': origin_visit3['visit'],		'visit': origin_visit3['visit'],
'type': self.origin2['type'],		'type': self.origin2['type'],
'status': 'partial',		'status': 'partial',
'metadata': None,		'metadata': None,
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines	def test_origin_visit_update(self):
('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)])

def test_origin_visit_update_validation(self):		def test_origin_visit_update_validation(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
visit = self.storage.origin_visit_add(		visit = self.storage.origin_visit_add(
origin_id,		self.origin['url'],
date=self.date_visit2)		date=self.date_visit2)

with self.assertRaisesRegex(		with self.assertRaisesRegex(
(ValueError, psycopg2.DataError), 'status') as cm:		(ValueError, psycopg2.DataError), 'status') as cm:
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit['visit'], status='foobar')		self.origin['url'], visit['visit'], status='foobar')

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_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)
Show All 18 Lines	def test_origin_visit_find_by_date(self):
# There are two visits at the same date, the latest must be returned		# There are two visits at the same date, the latest must be returned
visit = self.storage.origin_visit_find_by_date(		visit = self.storage.origin_visit_find_by_date(
self.origin['url'], self.date_visit2)		self.origin['url'], self.date_visit2)
self.assertEqual(visit['visit'], origin_visit3['visit'])		self.assertEqual(visit['visit'], origin_visit3['visit'])

def test_origin_visit_find_by_date__unknown_origin(self):		def test_origin_visit_find_by_date__unknown_origin(self):
self.storage.origin_visit_find_by_date('foo', self.date_visit2)		self.storage.origin_visit_find_by_date('foo', self.date_visit2)

@settings(deadline=None)		def test_origin_visit_update_missing_snapshot(self):
@given(strategies.booleans())
def test_origin_visit_update_missing_snapshot(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

# given		# given
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_id_or_url = self.origin['url'] if use_url else origin_id		origin_url = self.origin['url']

origin_visit = self.storage.origin_visit_add(		origin_visit = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date=self.date_visit1)		date=self.date_visit1)

# when		# when
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
origin_visit['visit'],		origin_visit['visit'],
snapshot=self.snapshot['id'])		snapshot=self.snapshot['id'])

# then		# then
actual_origin_visit = self.storage.origin_visit_get_by(		actual_origin_visit = self.storage.origin_visit_get_by(
origin_id_or_url,		origin_url,
origin_visit['visit'])		origin_visit['visit'])
self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id'])		self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id'])

# when		# when
self.storage.snapshot_add([self.snapshot])		self.storage.snapshot_add([self.snapshot])
self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id'])		self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id'])

@settings(deadline=None)		def test_origin_visit_get_by(self):
@given(strategies.booleans())
def test_origin_visit_get_by(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_id2 = self.storage.origin_add_one(self.origin2)		self.storage.origin_add_one(self.origin2)

origin_id_or_url = self.origin['url'] if use_url else origin_id		origin_url = self.origin['url']
origin2_id_or_url = self.origin2['url'] if use_url else origin_id2		origin2_url = self.origin2['url']

origin_visit1 = self.storage.origin_visit_add(		origin_visit1 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date=self.date_visit2)		date=self.date_visit2)

self.storage.snapshot_add([self.snapshot])		self.storage.snapshot_add([self.snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
origin_visit1['visit'],		origin_visit1['visit'],
snapshot=self.snapshot['id'])		snapshot=self.snapshot['id'])

# Add some other {origin, visit} entries		# Add some other {origin, visit} entries
self.storage.origin_visit_add(		self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date=self.date_visit3)		date=self.date_visit3)
self.storage.origin_visit_add(		self.storage.origin_visit_add(
origin2_id_or_url,		origin2_url,
date=self.date_visit3)		date=self.date_visit3)

# when		# when
visit1_metadata = {		visit1_metadata = {
'contents': 42,		'contents': 42,
'directories': 22,		'directories': 22,
}		}

self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
origin_visit1['visit'], status='full',		origin_visit1['visit'], status='full',
metadata=visit1_metadata)		metadata=visit1_metadata)

expected_origin_visit = origin_visit1.copy()		expected_origin_visit = origin_visit1.copy()
expected_origin_visit.update({		expected_origin_visit.update({
'origin': origin_id,		'origin': origin_url,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'date': self.date_visit2,		'date': self.date_visit2,
'type': self.origin['type'],		'type': self.origin['type'],
'metadata': visit1_metadata,		'metadata': visit1_metadata,
'status': 'full',		'status': 'full',
'snapshot': self.snapshot['id'],		'snapshot': self.snapshot['id'],
})		})

# when		# when
actual_origin_visit1 = self.storage.origin_visit_get_by(		actual_origin_visit1 = self.storage.origin_visit_get_by(
origin_id_or_url,		origin_url,
origin_visit1['visit'])		origin_visit1['visit'])

# then		# then
self.assertEqual(actual_origin_visit1, expected_origin_visit)		self.assertEqual(actual_origin_visit1, expected_origin_visit)

def test_origin_visit_get_by__unknown_origin(self):		def test_origin_visit_get_by__unknown_origin(self):
if self._test_origin_ids:
self.assertIsNone(self.storage.origin_visit_get_by(2, 10))
self.assertIsNone(self.storage.origin_visit_get_by('foo', 10))		self.assertIsNone(self.storage.origin_visit_get_by('foo', 10))

@given(strategies.booleans())		def test_origin_visit_upsert_new(self):
def test_origin_visit_upsert_new(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

# given		# given
self.assertIsNone(self.storage.origin_get([self.origin2])[0])		self.assertIsNone(self.storage.origin_get([self.origin2])[0])

origin_id = self.storage.origin_add_one(self.origin2)		self.storage.origin_add_one(self.origin2)
origin_id_or_url = self.origin2['url'] if use_url else origin_id		origin_url = self.origin2['url']
self.assertIsNotNone(origin_id)		self.assertIsNotNone(origin_url)

# when		# when
self.storage.origin_visit_upsert([		self.storage.origin_visit_upsert([
{		{
'origin': self.origin2,		'origin': self.origin2,
'date': self.date_visit2,		'date': self.date_visit2,
'visit': 123,		'visit': 123,
'type': self.origin2['type'],		'type': self.origin2['type'],
Show All 9 Lines	def test_origin_visit_upsert_new(self):
'status': 'full',		'status': 'full',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
},		},
])		])

# then		# then
actual_origin_visits = list(self.storage.origin_visit_get(		actual_origin_visits = list(self.storage.origin_visit_get(
origin_id_or_url))		origin_url))
self.assertEqual(actual_origin_visits, [		self.assertEqual(actual_origin_visits, [
{		{
'origin': origin_id,		'origin': origin_url,
'date': self.date_visit2,		'date': self.date_visit2,
'visit': 123,		'visit': 123,
'type': self.origin2['type'],		'type': self.origin2['type'],
'status': 'full',		'status': 'full',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
},		},
{		{
'origin': origin_id,		'origin': origin_url,
'date': self.date_visit3,		'date': self.date_visit3,
'visit': 1234,		'visit': 1234,
'type': self.origin2['type'],		'type': self.origin2['type'],
'status': 'full',		'status': 'full',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
},		},
])		])
Show All 17 Lines	def test_origin_visit_upsert_new(self):
'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)])

@settings(deadline=None)		def test_origin_visit_upsert_existing(self):
@given(strategies.booleans())
def test_origin_visit_upsert_existing(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

# given		# given
self.assertIsNone(self.storage.origin_get([self.origin2])[0])		self.assertIsNone(self.storage.origin_get([self.origin2])[0])

origin_id = self.storage.origin_add_one(self.origin2)		self.storage.origin_add_one(self.origin2)
origin_id_or_url = self.origin2['url'] if use_url else origin_id		origin_url = self.origin2['url']
self.assertIsNotNone(origin_id)		self.assertIsNotNone(origin_url)

# when		# when
origin_visit1 = self.storage.origin_visit_add(		origin_visit1 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
date=self.date_visit2)		date=self.date_visit2)
self.storage.origin_visit_upsert([{		self.storage.origin_visit_upsert([{
'origin': self.origin2,		'origin': self.origin2,
'date': self.date_visit2,		'date': self.date_visit2,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': self.origin2['type'],		'type': self.origin2['type'],
'status': 'full',		'status': 'full',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
}])		}])

# then		# then
self.assertEqual(origin_visit1['origin'], origin_id)		self.assertEqual(origin_visit1['origin'], origin_url)
self.assertIsNotNone(origin_visit1['visit'])		self.assertIsNotNone(origin_visit1['visit'])

actual_origin_visits = list(self.storage.origin_visit_get(		actual_origin_visits = list(self.storage.origin_visit_get(
origin_id_or_url))		origin_url))
self.assertEqual(actual_origin_visits,		self.assertEqual(actual_origin_visits,
[{		[{
'origin': origin_id,		'origin': origin_url,
'date': self.date_visit2,		'date': self.date_visit2,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': self.origin2['type'],		'type': self.origin2['type'],
'status': 'full',		'status': 'full',
'metadata': None,		'metadata': None,
'snapshot': None,		'snapshot': None,
}])		}])

Show All 17 Lines	def test_origin_visit_upsert_existing(self):
'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)])

def test_origin_visit_get_by_no_result(self):		def test_origin_visit_get_by_no_result(self):
if self._test_origin_ids:
actual_origin_visit = self.storage.origin_visit_get_by(
10, 999)
self.assertIsNone(actual_origin_visit)

self.storage.origin_add([self.origin])		self.storage.origin_add([self.origin])
actual_origin_visit = self.storage.origin_visit_get_by(		actual_origin_visit = self.storage.origin_visit_get_by(
self.origin['url'], 999)		self.origin['url'], 999)
self.assertIsNone(actual_origin_visit)		self.assertIsNone(actual_origin_visit)

@settings(deadline=None) # this test is very slow		def test_origin_visit_get_latest(self):
@given(strategies.booleans())
def test_origin_visit_get_latest(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_id_or_url = self.origin['url'] if use_url else origin_id
origin_url = self.origin['url']		origin_url = self.origin['url']
origin_visit1 = self.storage.origin_visit_add(		origin_visit1 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
self.date_visit1)		self.date_visit1)
visit1_id = origin_visit1['visit']		visit1_id = origin_visit1['visit']
origin_visit2 = self.storage.origin_visit_add(		origin_visit2 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
self.date_visit2)		self.date_visit2)
visit2_id = origin_visit2['visit']		visit2_id = origin_visit2['visit']

# Add a visit with the same date as the previous one		# Add a visit with the same date as the previous one
origin_visit3 = self.storage.origin_visit_add(		origin_visit3 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
self.date_visit2)		self.date_visit2)
visit3_id = origin_visit3['visit']		visit3_id = origin_visit3['visit']

origin_visit1 = self.storage.origin_visit_get_by(origin_url, visit1_id)		origin_visit1 = self.storage.origin_visit_get_by(origin_url, visit1_id)
origin_visit2 = self.storage.origin_visit_get_by(origin_url, visit2_id)		origin_visit2 = self.storage.origin_visit_get_by(origin_url, visit2_id)
origin_visit3 = self.storage.origin_visit_get_by(origin_url, visit3_id)		origin_visit3 = self.storage.origin_visit_get_by(origin_url, visit3_id)

# Two visits, both with no snapshot		# Two visits, both with no snapshot
self.assertEqual(		self.assertEqual(
origin_visit3,		origin_visit3,
self.storage.origin_visit_get_latest(origin_url))		self.storage.origin_visit_get_latest(origin_url))
self.assertIsNone(		self.assertIsNone(
self.storage.origin_visit_get_latest(origin_url,		self.storage.origin_visit_get_latest(origin_url,
require_snapshot=True))		require_snapshot=True))

# Add snapshot to visit1; require_snapshot=True makes it return		# Add snapshot to visit1; require_snapshot=True makes it return
# visit1 and require_snapshot=False still returns visit2		# visit1 and require_snapshot=False still returns visit2
self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url, visit1_id,		origin_url, visit1_id,
snapshot=self.complete_snapshot['id'])		snapshot=self.complete_snapshot['id'])
self.assertEqual(		self.assertEqual(
{**origin_visit1, 'snapshot': self.complete_snapshot['id']},		{**origin_visit1, 'snapshot': self.complete_snapshot['id']},
self.storage.origin_visit_get_latest(		self.storage.origin_visit_get_latest(
origin_url, require_snapshot=True)		origin_url, require_snapshot=True)
)		)
self.assertEqual(		self.assertEqual(
origin_visit3,		origin_visit3,
self.storage.origin_visit_get_latest(origin_url)		self.storage.origin_visit_get_latest(origin_url)
)		)

# Status filter: all three visits are status=ongoing, so no visit		# Status filter: all three visits are status=ongoing, so no visit
# returned		# returned
self.assertIsNone(		self.assertIsNone(
self.storage.origin_visit_get_latest(		self.storage.origin_visit_get_latest(
origin_url, allowed_statuses=['full'])		origin_url, allowed_statuses=['full'])
)		)

# Mark the first visit as completed and check status filter again		# Mark the first visit as completed and check status filter again
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
visit1_id, status='full')		visit1_id, status='full')
self.assertEqual(		self.assertEqual(
{		{
**origin_visit1,		**origin_visit1,
'snapshot': self.complete_snapshot['id'],		'snapshot': self.complete_snapshot['id'],
'status': 'full'},		'status': 'full'},
self.storage.origin_visit_get_latest(		self.storage.origin_visit_get_latest(
origin_url, allowed_statuses=['full']),		origin_url, allowed_statuses=['full']),
)		)
self.assertEqual(		self.assertEqual(
origin_visit3,		origin_visit3,
self.storage.origin_visit_get_latest(origin_url),		self.storage.origin_visit_get_latest(origin_url),
)		)

# Add snapshot to visit2 and check that the new snapshot is returned		# Add snapshot to visit2 and check that the new snapshot is returned
self.storage.snapshot_add([self.empty_snapshot])		self.storage.snapshot_add([self.empty_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url, visit2_id,		origin_url, visit2_id,
snapshot=self.empty_snapshot['id'])		snapshot=self.empty_snapshot['id'])
self.assertEqual(		self.assertEqual(
{**origin_visit2, 'snapshot': self.empty_snapshot['id']},		{**origin_visit2, 'snapshot': self.empty_snapshot['id']},
self.storage.origin_visit_get_latest(		self.storage.origin_visit_get_latest(
origin_url, require_snapshot=True),		origin_url, require_snapshot=True),
)		)
self.assertEqual(		self.assertEqual(
origin_visit3,		origin_visit3,
self.storage.origin_visit_get_latest(origin_url),		self.storage.origin_visit_get_latest(origin_url),
)		)

# Check that the status filter is still working		# Check that the status filter is still working
self.assertEqual(		self.assertEqual(
{		{
**origin_visit1,		**origin_visit1,
'snapshot': self.complete_snapshot['id'],		'snapshot': self.complete_snapshot['id'],
'status': 'full'},		'status': 'full'},
self.storage.origin_visit_get_latest(		self.storage.origin_visit_get_latest(
origin_url, allowed_statuses=['full']),		origin_url, allowed_statuses=['full']),
)		)

# Add snapshot to visit3 (same date as visit2)		# Add snapshot to visit3 (same date as visit2)
self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url, visit3_id, snapshot=self.complete_snapshot['id'])		origin_url, visit3_id, snapshot=self.complete_snapshot['id'])
self.assertEqual(		self.assertEqual(
{		{
**origin_visit1,		**origin_visit1,
'snapshot': self.complete_snapshot['id'],		'snapshot': self.complete_snapshot['id'],
'status': 'full'},		'status': 'full'},
self.storage.origin_visit_get_latest(		self.storage.origin_visit_get_latest(
origin_url, allowed_statuses=['full']),		origin_url, allowed_statuses=['full']),
)		)
Show All 34 Lines	def test_person_fullname_unicity(self):
self.storage.revision_get([revision['id'], revision2['id']]))		self.storage.revision_get([revision['id'], revision2['id']]))

# then		# then
# check committers are the same		# check committers are the same
self.assertEqual(revisions[0]['committer'],		self.assertEqual(revisions[0]['committer'],
revisions[1]['committer'])		revisions[1]['committer'])

def test_snapshot_add_get_empty(self):		def test_snapshot_add_get_empty(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_id = origin_visit1['visit']

actual_result = self.storage.snapshot_add([self.empty_snapshot])		actual_result = self.storage.snapshot_add([self.empty_snapshot])
self.assertEqual(actual_result, {'snapshot:add': 1})		self.assertEqual(actual_result, {'snapshot:add': 1})

self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit_id, snapshot=self.empty_snapshot['id'])		origin_url, visit_id, snapshot=self.empty_snapshot['id'])

by_id = self.storage.snapshot_get(self.empty_snapshot['id'])		by_id = self.storage.snapshot_get(self.empty_snapshot['id'])
self.assertEqual(by_id, {**self.empty_snapshot, 'next_branch': None})		self.assertEqual(by_id, {**self.empty_snapshot, 'next_branch': None})

by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id)		by_ov = self.storage.snapshot_get_by_origin_visit(origin_url, visit_id)
self.assertEqual(by_ov, {**self.empty_snapshot, 'next_branch': None})		self.assertEqual(by_ov, {**self.empty_snapshot, 'next_branch': None})

expected_origin = self.origin.copy()		expected_origin = self.origin.copy()
data1 = {		data1 = {
'origin': expected_origin,		'origin': expected_origin,
'date': self.date_visit1,		'date': self.date_visit1,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
'type': self.origin['type'],		'type': self.origin['type'],
Show All 12 Lines	def test_snapshot_add_get_empty(self):
}		}
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),
('snapshot', self.empty_snapshot),		('snapshot', self.empty_snapshot),
('origin_visit', data2)])		('origin_visit', data2)])

def test_snapshot_add_get_complete(self):		def test_snapshot_add_get_complete(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_id = origin_visit1['visit']

actual_result = self.storage.snapshot_add([self.complete_snapshot])		actual_result = self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit_id, snapshot=self.complete_snapshot['id'])		origin_url, visit_id, snapshot=self.complete_snapshot['id'])
self.assertEqual(actual_result, {'snapshot:add': 1})		self.assertEqual(actual_result, {'snapshot:add': 1})

by_id = self.storage.snapshot_get(self.complete_snapshot['id'])		by_id = self.storage.snapshot_get(self.complete_snapshot['id'])
self.assertEqual(by_id,		self.assertEqual(by_id,
{**self.complete_snapshot, 'next_branch': None})		{**self.complete_snapshot, 'next_branch': None})

by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id)		by_ov = self.storage.snapshot_get_by_origin_visit(origin_url, visit_id)
self.assertEqual(by_ov,		self.assertEqual(by_ov,
{**self.complete_snapshot, 'next_branch': None})		{**self.complete_snapshot, 'next_branch': None})

def test_snapshot_add_many(self):		def test_snapshot_add_many(self):
actual_result = self.storage.snapshot_add(		actual_result = self.storage.snapshot_add(
[self.snapshot, self.complete_snapshot])		[self.snapshot, self.complete_snapshot])
self.assertEqual(actual_result, {'snapshot:add': 2})		self.assertEqual(actual_result, {'snapshot:add': 2})

▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines	def test_snapshot_add_validation(self):

snap = copy.deepcopy(self.snapshot)		snap = copy.deepcopy(self.snapshot)
snap['branches'][b'foo'] = {'target': b'\x42'*20}		snap['branches'][b'foo'] = {'target': b'\x42'*20}

with self.assertRaisesRegex(KeyError, 'target_type'):		with self.assertRaisesRegex(KeyError, 'target_type'):
self.storage.snapshot_add([snap])		self.storage.snapshot_add([snap])

def test_snapshot_add_count_branches(self):		def test_snapshot_add_count_branches(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_id = origin_visit1['visit']

actual_result = self.storage.snapshot_add([self.complete_snapshot])		actual_result = self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit_id, snapshot=self.complete_snapshot['id'])		origin_url, visit_id, snapshot=self.complete_snapshot['id'])
self.assertEqual(actual_result, {'snapshot:add': 1})		self.assertEqual(actual_result, {'snapshot:add': 1})

snp_id = self.complete_snapshot['id']		snp_id = self.complete_snapshot['id']
snp_size = self.storage.snapshot_count_branches(snp_id)		snp_size = self.storage.snapshot_count_branches(snp_id)

expected_snp_size = {		expected_snp_size = {
'alias': 1,		'alias': 1,
'content': 1,		'content': 1,
'directory': 2,		'directory': 2,
'release': 1,		'release': 1,
'revision': 1,		'revision': 1,
'snapshot': 1,		'snapshot': 1,
None: 1		None: 1
}		}

self.assertEqual(snp_size, expected_snp_size)		self.assertEqual(snp_size, expected_snp_size)

def test_snapshot_add_get_paginated(self):		def test_snapshot_add_get_paginated(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_id = origin_visit1['visit']

self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit_id,		origin_url, visit_id,
snapshot=self.complete_snapshot['id'])		snapshot=self.complete_snapshot['id'])

snp_id = self.complete_snapshot['id']		snp_id = self.complete_snapshot['id']
branches = self.complete_snapshot['branches']		branches = self.complete_snapshot['branches']
branch_names = list(sorted(branches))		branch_names = list(sorted(branches))

# Test branch_from		# Test branch_from

Show All 39 Lines	def test_snapshot_add_get_paginated(self):
for name in branch_names[dir_idx:dir_idx + 3]		for name in branch_names[dir_idx:dir_idx + 3]
},		},
'next_branch': branch_names[dir_idx + 3],		'next_branch': branch_names[dir_idx + 3],
}		}

self.assertEqual(snapshot, expected_snapshot)		self.assertEqual(snapshot, expected_snapshot)

def test_snapshot_add_get_filtered(self):		def test_snapshot_add_get_filtered(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_id = origin_visit1['visit']

self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit_id, snapshot=self.complete_snapshot['id'])		origin_url, visit_id, snapshot=self.complete_snapshot['id'])

snp_id = self.complete_snapshot['id']		snp_id = self.complete_snapshot['id']
branches = self.complete_snapshot['branches']		branches = self.complete_snapshot['branches']

snapshot = self.storage.snapshot_get_branches(		snapshot = self.storage.snapshot_get_branches(
snp_id, target_types=['release', 'revision'])		snp_id, target_types=['release', 'revision'])

expected_snapshot = {		expected_snapshot = {
Show All 19 Lines	def test_snapshot_add_get_filtered(self):
if tgt and tgt['target_type'] == 'alias'		if tgt and tgt['target_type'] == 'alias'
},		},
'next_branch': None,		'next_branch': None,
}		}

self.assertEqual(snapshot, expected_snapshot)		self.assertEqual(snapshot, expected_snapshot)

def test_snapshot_add_get_filtered_and_paginated(self):		def test_snapshot_add_get_filtered_and_paginated(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_id = origin_visit1['visit']

self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit_id, snapshot=self.complete_snapshot['id'])		origin_url, visit_id, snapshot=self.complete_snapshot['id'])

snp_id = self.complete_snapshot['id']		snp_id = self.complete_snapshot['id']
branches = self.complete_snapshot['branches']		branches = self.complete_snapshot['branches']
branch_names = list(sorted(branches))		branch_names = list(sorted(branches))

# Test branch_from		# Test branch_from

snapshot = self.storage.snapshot_get_branches(		snapshot = self.storage.snapshot_get_branches(
▲ Show 20 Lines • Show All 55 Lines • ▼ Show 20 Lines	def test_snapshot_add_get_filtered_and_paginated(self):
branch_names[dir_idx]: branches[branch_names[dir_idx]],		branch_names[dir_idx]: branches[branch_names[dir_idx]],
},		},
'next_branch': b'release',		'next_branch': b'release',
}		}

self.assertEqual(snapshot, expected_snapshot)		self.assertEqual(snapshot, expected_snapshot)

def test_snapshot_add_get(self):		def test_snapshot_add_get(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit_id = origin_visit1['visit']		visit_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, visit_id, snapshot=self.snapshot['id'])		origin_url, visit_id, snapshot=self.snapshot['id'])

by_id = self.storage.snapshot_get(self.snapshot['id'])		by_id = self.storage.snapshot_get(self.snapshot['id'])
self.assertEqual(by_id, {**self.snapshot, 'next_branch': None})		self.assertEqual(by_id, {**self.snapshot, 'next_branch': None})

by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id)		by_ov = self.storage.snapshot_get_by_origin_visit(origin_url, visit_id)
self.assertEqual(by_ov, {**self.snapshot, 'next_branch': None})		self.assertEqual(by_ov, {**self.snapshot, 'next_branch': None})

origin_visit_info = self.storage.origin_visit_get_by(origin_id,		origin_visit_info = self.storage.origin_visit_get_by(origin_url,
visit_id)		visit_id)
self.assertEqual(origin_visit_info['snapshot'], self.snapshot['id'])		self.assertEqual(origin_visit_info['snapshot'], self.snapshot['id'])

def test_snapshot_add_nonexistent_visit(self):		def test_snapshot_add_nonexistent_visit(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
		origin_url = self.origin['url']
visit_id = 54164461156		visit_id = 54164461156

self.journal_writer.objects[:] = []		self.journal_writer.objects[:] = []

self.storage.snapshot_add([self.snapshot])		self.storage.snapshot_add([self.snapshot])

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_url, 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__by_origin_visit(self):		def test_snapshot_add_twice__by_origin_visit(self):
origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_url = self.origin['url']
		origin_visit1 = self.storage.origin_visit_add(origin_url,
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_url, visit1_id, snapshot=self.snapshot['id'])

by_ov1 = self.storage.snapshot_get_by_origin_visit(origin_id,		by_ov1 = self.storage.snapshot_get_by_origin_visit(origin_url,
visit1_id)		visit1_id)
self.assertEqual(by_ov1, {**self.snapshot, 'next_branch': None})		self.assertEqual(by_ov1, {**self.snapshot, 'next_branch': None})

origin_visit2 = self.storage.origin_visit_add(origin_id,		origin_visit2 = self.storage.origin_visit_add(origin_url,
self.date_visit2)		self.date_visit2)
visit2_id = origin_visit2['visit']		visit2_id = origin_visit2['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, visit2_id, snapshot=self.snapshot['id'])		origin_url, visit2_id, snapshot=self.snapshot['id'])

by_ov2 = self.storage.snapshot_get_by_origin_visit(origin_id,		by_ov2 = self.storage.snapshot_get_by_origin_visit(origin_url,
visit2_id)		visit2_id)
self.assertEqual(by_ov2, {**self.snapshot, 'next_branch': None})		self.assertEqual(by_ov2, {**self.snapshot, 'next_branch': None})

expected_origin = self.origin.copy()		expected_origin = self.origin.copy()
data1 = {		data1 = {
'origin': expected_origin,		'origin': expected_origin,
'date': self.date_visit1,		'date': self.date_visit1,
'visit': origin_visit1['visit'],		'visit': origin_visit1['visit'],
Show All 32 Lines	def test_snapshot_add_twice__by_origin_visit(self):
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),
('snapshot', self.snapshot),		('snapshot', self.snapshot),
('origin_visit', data2),		('origin_visit', data2),
('origin_visit', data3),		('origin_visit', data3),
('origin_visit', data4)])		('origin_visit', data4)])

@settings(deadline=None) # this test is very slow		def test_snapshot_get_latest(self):
@given(strategies.booleans())
def test_snapshot_get_latest(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_id_or_url = self.origin['url'] if use_url else origin_id		origin_url = self.origin['url']
origin_visit1 = self.storage.origin_visit_add(origin_id,		origin_visit1 = self.storage.origin_visit_add(origin_url,
self.date_visit1)		self.date_visit1)
visit1_id = origin_visit1['visit']		visit1_id = origin_visit1['visit']
origin_visit2 = self.storage.origin_visit_add(origin_id,		origin_visit2 = self.storage.origin_visit_add(origin_url,
self.date_visit2)		self.date_visit2)
visit2_id = origin_visit2['visit']		visit2_id = origin_visit2['visit']

# Add a visit with the same date as the previous one		# Add a visit with the same date as the previous one
origin_visit3 = self.storage.origin_visit_add(origin_id,		origin_visit3 = self.storage.origin_visit_add(origin_url,
self.date_visit2)		self.date_visit2)
visit3_id = origin_visit3['visit']		visit3_id = origin_visit3['visit']

# Two visits, both with no snapshot: latest snapshot is None		# Two visits, both with no snapshot: latest snapshot is None
self.assertIsNone(self.storage.snapshot_get_latest(		self.assertIsNone(self.storage.snapshot_get_latest(
origin_id_or_url))		origin_url))

# Add snapshot to visit1, latest snapshot = visit 1 snapshot		# Add snapshot to visit1, latest snapshot = visit 1 snapshot
self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit1_id, snapshot=self.complete_snapshot['id'])		origin_url, visit1_id, snapshot=self.complete_snapshot['id'])
self.assertEqual({**self.complete_snapshot, 'next_branch': None},		self.assertEqual({**self.complete_snapshot, 'next_branch': None},
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url))		origin_url))

# Status filter: all three visits are status=ongoing, so no snapshot		# Status filter: all three visits are status=ongoing, so no snapshot
# returned		# returned
self.assertIsNone(		self.assertIsNone(
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url,		origin_url,
allowed_statuses=['full'])		allowed_statuses=['full'])
)		)

# Mark the first visit as completed and check status filter again		# Mark the first visit as completed and check status filter again
self.storage.origin_visit_update(origin_id, visit1_id, status='full')		self.storage.origin_visit_update(origin_url, visit1_id, status='full')
self.assertEqual(		self.assertEqual(
{**self.complete_snapshot, 'next_branch': None},		{**self.complete_snapshot, 'next_branch': None},
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url,		origin_url,
allowed_statuses=['full']),		allowed_statuses=['full']),
)		)

# Add snapshot to visit2 and check that the new snapshot is returned		# Add snapshot to visit2 and check that the new snapshot is returned
self.storage.snapshot_add([self.empty_snapshot])		self.storage.snapshot_add([self.empty_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit2_id, snapshot=self.empty_snapshot['id'])		origin_url, visit2_id, snapshot=self.empty_snapshot['id'])
self.assertEqual({**self.empty_snapshot, 'next_branch': None},		self.assertEqual({**self.empty_snapshot, 'next_branch': None},
self.storage.snapshot_get_latest(origin_id))		self.storage.snapshot_get_latest(origin_url))

# Check that the status filter is still working		# Check that the status filter is still working
self.assertEqual(		self.assertEqual(
{**self.complete_snapshot, 'next_branch': None},		{**self.complete_snapshot, 'next_branch': None},
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url,		origin_url,
allowed_statuses=['full']),		allowed_statuses=['full']),
)		)

# Add snapshot to visit3 (same date as visit2) and check that		# Add snapshot to visit3 (same date as visit2) and check that
# the new snapshot is returned		# the new snapshot is returned
self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id, visit3_id, snapshot=self.complete_snapshot['id'])		origin_url, visit3_id, snapshot=self.complete_snapshot['id'])
self.assertEqual({**self.complete_snapshot, 'next_branch': None},		self.assertEqual({**self.complete_snapshot, 'next_branch': None},
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url))		origin_url))

@given(strategies.booleans())		def test_snapshot_get_latest__missing_snapshot(self):
def test_snapshot_get_latest__missing_snapshot(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()		self.reset_storage()

# Origin does not exist		# Origin does not exist
self.assertIsNone(self.storage.snapshot_get_latest(		self.assertIsNone(self.storage.snapshot_get_latest(
self.origin['url'] if use_url else 999))		self.origin['url']))

origin_id = self.storage.origin_add_one(self.origin)		self.storage.origin_add_one(self.origin)
origin_id_or_url = self.origin['url'] if use_url else origin_id		origin_url = self.origin['url']
origin_visit1 = self.storage.origin_visit_add(		origin_visit1 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
self.date_visit1)		self.date_visit1)
visit1_id = origin_visit1['visit']		visit1_id = origin_visit1['visit']
origin_visit2 = self.storage.origin_visit_add(		origin_visit2 = self.storage.origin_visit_add(
origin_id_or_url,		origin_url,
self.date_visit2)		self.date_visit2)
visit2_id = origin_visit2['visit']		visit2_id = origin_visit2['visit']

# Two visits, both with no snapshot: latest snapshot is None		# Two visits, both with no snapshot: latest snapshot is None
self.assertIsNone(self.storage.snapshot_get_latest(		self.assertIsNone(self.storage.snapshot_get_latest(
origin_id_or_url))		origin_url))

# Add unknown snapshot to visit1, check that the inconsistency is		# Add unknown snapshot to visit1, check that the inconsistency is
# detected		# detected
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
visit1_id, snapshot=self.complete_snapshot['id'])		visit1_id, snapshot=self.complete_snapshot['id'])
with self.assertRaises(ValueError):		with self.assertRaises(ValueError):
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url)		origin_url)

# Status filter: both visits are status=ongoing, so no snapshot		# Status filter: both visits are status=ongoing, so no snapshot
# returned		# returned
self.assertIsNone(		self.assertIsNone(
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url,		origin_url,
allowed_statuses=['full'])		allowed_statuses=['full'])
)		)

# Mark the first visit as completed and check status filter again		# Mark the first visit as completed and check status filter again
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
visit1_id, status='full')		visit1_id, status='full')
with self.assertRaises(ValueError):		with self.assertRaises(ValueError):
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url,		origin_url,
allowed_statuses=['full']),		allowed_statuses=['full']),

# Actually add the snapshot and check status filter again		# Actually add the snapshot and check status filter again
self.storage.snapshot_add([self.complete_snapshot])		self.storage.snapshot_add([self.complete_snapshot])
self.assertEqual(		self.assertEqual(
{**self.complete_snapshot, 'next_branch': None},		{**self.complete_snapshot, 'next_branch': None},
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url)		origin_url)
)		)

# Add unknown snapshot to visit2 and check that the inconsistency		# Add unknown snapshot to visit2 and check that the inconsistency
# is detected		# is detected
self.storage.origin_visit_update(		self.storage.origin_visit_update(
origin_id_or_url,		origin_url,
visit2_id, snapshot=self.snapshot['id'])		visit2_id, snapshot=self.snapshot['id'])
with self.assertRaises(ValueError):		with self.assertRaises(ValueError):
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url)		origin_url)

# Actually add that snapshot and check that the new one is returned		# Actually add that snapshot and check that the new one is returned
self.storage.snapshot_add([self.snapshot])		self.storage.snapshot_add([self.snapshot])
self.assertEqual(		self.assertEqual(
{**self.snapshot, 'next_branch': None},		{**self.snapshot, 'next_branch': None},
self.storage.snapshot_get_latest(		self.storage.snapshot_get_latest(
origin_id_or_url)		origin_url)
)		)

def test_stat_counters(self):		def test_stat_counters(self):
expected_keys = ['content', 'directory',		expected_keys = ['content', 'directory',
'origin', 'revision']		'origin', 'revision']

# Initially, all counters are 0		# Initially, all counters are 0

▲ Show 20 Lines • Show All 448 Lines • ▼ Show 20 Lines	def test_metadata_provider_get_by(self):

actual_provider = self.storage.metadata_provider_get_by({		actual_provider = self.storage.metadata_provider_get_by({
'provider_name': self.provider['name'],		'provider_name': self.provider['name'],
'provider_url': self.provider['url']		'provider_url': self.provider['url']
})		})
# then		# then
self.assertTrue(provider_id, actual_provider['id'])		self.assertTrue(provider_id, actual_provider['id'])

@given(strategies.booleans())		def test_origin_metadata_add(self):
def test_origin_metadata_add(self, use_url):
self.reset_storage()		self.reset_storage()
# given		# given
origin = self.storage.origin_add([self.origin])[0]		origin = self.storage.origin_add([self.origin])[0]
origin_id = origin['id']		origin_url = origin['url']
if use_url:
origin = origin['url']
else:
origin = origin['id']
origin_metadata0 = list(self.storage.origin_metadata_get_by(		origin_metadata0 = list(self.storage.origin_metadata_get_by(
origin))		origin_url))
self.assertEqual(len(origin_metadata0), 0, origin_metadata0)		self.assertEqual(len(origin_metadata0), 0, origin_metadata0)

tools = self.storage.tool_add([self.metadata_tool])		tools = self.storage.tool_add([self.metadata_tool])
tool = tools[0]		tool = tools[0]

self.storage.metadata_provider_add(		self.storage.metadata_provider_add(
self.provider['name'],		self.provider['name'],
self.provider['type'],		self.provider['type'],
self.provider['url'],		self.provider['url'],
self.provider['metadata'])		self.provider['metadata'])
provider = self.storage.metadata_provider_get_by({		provider = self.storage.metadata_provider_get_by({
'provider_name': self.provider['name'],		'provider_name': self.provider['name'],
'provider_url': self.provider['url']		'provider_url': self.provider['url']
})		})

# when adding for the same origin 2 metadatas		# when adding for the same origin 2 metadatas
self.storage.origin_metadata_add(		self.storage.origin_metadata_add(
origin,		origin_url,
self.origin_metadata['discovery_date'],		self.origin_metadata['discovery_date'],
provider['id'],		provider['id'],
tool['id'],		tool['id'],
self.origin_metadata['metadata'])		self.origin_metadata['metadata'])
self.storage.origin_metadata_add(		self.storage.origin_metadata_add(
origin,		origin_url,
'2015-01-01 23:00:00+00',		'2015-01-01 23:00:00+00',
provider['id'],		provider['id'],
tool['id'],		tool['id'],
self.origin_metadata2['metadata'])		self.origin_metadata2['metadata'])
actual_om = list(self.storage.origin_metadata_get_by(		actual_om = list(self.storage.origin_metadata_get_by(
origin))		origin_url))
# then		# then
self.assertCountEqual(		self.assertCountEqual(
[item['origin_id'] for item in actual_om],		[item['origin_url'] for item in actual_om],
[origin_id, origin_id])		[origin_url, origin_url])

def test_origin_metadata_get(self):		def test_origin_metadata_get(self):
# given		# given
origin_id = self.storage.origin_add([self.origin])[0]['id']		origin_url = self.storage.origin_add([self.origin])[0]['url']
origin_id2 = self.storage.origin_add([self.origin2])[0]['id']		origin_url2 = self.storage.origin_add([self.origin2])[0]['url']

self.storage.metadata_provider_add(self.provider['name'],		self.storage.metadata_provider_add(self.provider['name'],
self.provider['type'],		self.provider['type'],
self.provider['url'],		self.provider['url'],
self.provider['metadata'])		self.provider['metadata'])
provider = self.storage.metadata_provider_get_by({		provider = self.storage.metadata_provider_get_by({
'provider_name': self.provider['name'],		'provider_name': self.provider['name'],
'provider_url': self.provider['url']		'provider_url': self.provider['url']
})		})
tool = self.storage.tool_add([self.metadata_tool])[0]		tool = self.storage.tool_add([self.metadata_tool])[0]
# when adding for the same origin 2 metadatas		# when adding for the same origin 2 metadatas
self.storage.origin_metadata_add(		self.storage.origin_metadata_add(
origin_id,		origin_url,
self.origin_metadata['discovery_date'],		self.origin_metadata['discovery_date'],
provider['id'],		provider['id'],
tool['id'],		tool['id'],
self.origin_metadata['metadata'])		self.origin_metadata['metadata'])
self.storage.origin_metadata_add(		self.storage.origin_metadata_add(
origin_id2,		origin_url2,
self.origin_metadata2['discovery_date'],		self.origin_metadata2['discovery_date'],
provider['id'],		provider['id'],
tool['id'],		tool['id'],
self.origin_metadata2['metadata'])		self.origin_metadata2['metadata'])
self.storage.origin_metadata_add(		self.storage.origin_metadata_add(
origin_id,		origin_url,
self.origin_metadata2['discovery_date'],		self.origin_metadata2['discovery_date'],
provider['id'],		provider['id'],
tool['id'],		tool['id'],
self.origin_metadata2['metadata'])		self.origin_metadata2['metadata'])
all_metadatas = list(self.storage.origin_metadata_get_by(		all_metadatas = list(self.storage.origin_metadata_get_by(
origin_id))		origin_url))
metadatas_for_origin2 = list(self.storage.origin_metadata_get_by(		metadatas_for_origin2 = list(self.storage.origin_metadata_get_by(
origin_id2))		origin_url2))
expected_results = [{		expected_results = [{
'origin_id': origin_id,		'origin_url': origin_url,
'discovery_date': datetime.datetime(		'discovery_date': datetime.datetime(
2017, 1, 1, 23, 0,		2017, 1, 1, 23, 0,
tzinfo=datetime.timezone.utc),		tzinfo=datetime.timezone.utc),
'metadata': {		'metadata': {
'name': 'test_origin_metadata',		'name': 'test_origin_metadata',
'version': '0.0.1'		'version': '0.0.1'
},		},
'provider_id': provider['id'],		'provider_id': provider['id'],
'provider_name': 'hal',		'provider_name': 'hal',
'provider_type': 'deposit-client',		'provider_type': 'deposit-client',
'provider_url': 'http:///hal/inria',		'provider_url': 'http:///hal/inria',
'tool_id': tool['id']		'tool_id': tool['id']
}, {		}, {
'origin_id': origin_id,		'origin_url': origin_url,
'discovery_date': datetime.datetime(		'discovery_date': datetime.datetime(
2015, 1, 1, 23, 0,		2015, 1, 1, 23, 0,
tzinfo=datetime.timezone.utc),		tzinfo=datetime.timezone.utc),
'metadata': {		'metadata': {
'name': 'test_origin_metadata',		'name': 'test_origin_metadata',
'version': '0.0.1'		'version': '0.0.1'
},		},
'provider_id': provider['id'],		'provider_id': provider['id'],
Show All 27 Lines	def test_metadata_provider_add(self):
'provider_url': 'http://www.swmath.org/'		'provider_url': 'http://www.swmath.org/'
}))		}))
self.assertEqual(		self.assertEqual(
provider,		provider,
self.storage.metadata_provider_get(provider_id))		self.storage.metadata_provider_get(provider_id))

def test_origin_metadata_get_by_provider_type(self):		def test_origin_metadata_get_by_provider_type(self):
# given		# given
origin_id = self.storage.origin_add([self.origin])[0]['id']		origin_url = self.storage.origin_add([self.origin])[0]['url']
origin_id2 = self.storage.origin_add([self.origin2])[0]['id']		origin_url2 = self.storage.origin_add([self.origin2])[0]['url']
provider1_id = self.storage.metadata_provider_add(		provider1_id = self.storage.metadata_provider_add(
self.provider['name'],		self.provider['name'],
self.provider['type'],		self.provider['type'],
self.provider['url'],		self.provider['url'],
self.provider['metadata'])		self.provider['metadata'])
provider1 = self.storage.metadata_provider_get_by({		provider1 = self.storage.metadata_provider_get_by({
'provider_name': self.provider['name'],		'provider_name': self.provider['name'],
'provider_url': self.provider['url']		'provider_url': self.provider['url']
Show All 15 Lines	def test_origin_metadata_get_by_provider_type(self):
self.storage.metadata_provider_get(provider2_id))		self.storage.metadata_provider_get(provider2_id))

# using the only tool now inserted in the data.sql, but for this		# using the only tool now inserted in the data.sql, but for this
# provider should be a crawler tool (not yet implemented)		# provider should be a crawler tool (not yet implemented)
tool = self.storage.tool_add([self.metadata_tool])[0]		tool = self.storage.tool_add([self.metadata_tool])[0]

# when adding for the same origin 2 metadatas		# when adding for the same origin 2 metadatas
self.storage.origin_metadata_add(		self.storage.origin_metadata_add(
origin_id,		origin_url,
self.origin_metadata['discovery_date'],		self.origin_metadata['discovery_date'],
provider1['id'],		provider1['id'],
tool['id'],		tool['id'],
self.origin_metadata['metadata'])		self.origin_metadata['metadata'])
self.storage.origin_metadata_add(		self.storage.origin_metadata_add(
origin_id2,		origin_url2,
self.origin_metadata2['discovery_date'],		self.origin_metadata2['discovery_date'],
provider2['id'],		provider2['id'],
tool['id'],		tool['id'],
self.origin_metadata2['metadata'])		self.origin_metadata2['metadata'])
provider_type = 'registry'		provider_type = 'registry'
m_by_provider = list(self.storage.origin_metadata_get_by(		m_by_provider = list(self.storage.origin_metadata_get_by(
origin_id2,		origin_url2,
provider_type))		provider_type))
for item in m_by_provider:		for item in m_by_provider:
if 'id' in item:		if 'id' in item:
del item['id']		del item['id']
expected_results = [{		expected_results = [{
'origin_id': origin_id2,		'origin_url': origin_url2,
'discovery_date': datetime.datetime(		'discovery_date': datetime.datetime(
2017, 1, 1, 23, 0,		2017, 1, 1, 23, 0,
tzinfo=datetime.timezone.utc),		tzinfo=datetime.timezone.utc),
'metadata': {		'metadata': {
'name': 'test_origin_metadata',		'name': 'test_origin_metadata',
'version': '0.0.1'		'version': '0.0.1'
},		},
'provider_id': provider2['id'],		'provider_id': provider2['id'],
'provider_name': 'swMATH',		'provider_name': 'swMATH',
'provider_type': provider_type,		'provider_type': provider_type,
'provider_url': 'http://www.swmath.org/',		'provider_url': 'http://www.swmath.org/',
'tool_id': tool['id']		'tool_id': tool['id']
}]		}]
# 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)


class CommonPropTestStorage:		class CommonPropTestStorage:
_test_origin_ids = True

def assert_contents_ok(self, expected_contents, actual_contents,		def assert_contents_ok(self, expected_contents, actual_contents,
keys_to_check={'sha1', 'data'}):		keys_to_check={'sha1', 'data'}):
"""Assert that a given list of contents matches on a given set of keys.		"""Assert that a given list of contents matches on a given set of keys.

"""		"""
for k in keys_to_check:		for k in keys_to_check:
expected_list = sorted([c[k] for c in expected_contents])		expected_list = sorted([c[k] for c in expected_contents])
actual_list = sorted([c[k] for c in actual_contents])		actual_list = sorted([c[k] for c in actual_contents])
▲ Show 20 Lines • Show All 130 Lines • ▼ Show 20 Lines	def test_generate_content_get_range_limit(self, contents):

self.assertEqual(1, len(actual_contents2))		self.assertEqual(1, len(actual_contents2))
self.assertIsNone(actual_next2)		self.assertIsNone(actual_next2)

self.assert_contents_ok([contents_map[actual_next]], actual_contents2,		self.assert_contents_ok([contents_map[actual_next]], actual_contents2,

keys_to_check)		keys_to_check)

def test_origin_get_invalid_id_legacy(self):
if self._test_origin_ids:
invalid_origin_id = 1
origin_info = self.storage.origin_get({'id': invalid_origin_id})
self.assertIsNone(origin_info)

origin_visits = list(self.storage.origin_visit_get(
invalid_origin_id))
self.assertEqual(origin_visits, [])

def test_origin_get_invalid_id(self):
if self._test_origin_ids:
origin_info = self.storage.origin_get([{'id': 1}, {'id': 2}])
self.assertEqual(origin_info, [None, None])

origin_visits = list(self.storage.origin_visit_get(1))
self.assertEqual(origin_visits, [])

@given(strategies.lists(origins().map(lambda x: x.to_dict()),		@given(strategies.lists(origins().map(lambda x: x.to_dict()),
unique_by=lambda x: x['url'],		unique_by=lambda x: x['url'],
min_size=6, max_size=15))		min_size=6, max_size=15))
def test_origin_get_range(self, new_origins):		def test_origin_get_range(self, new_origins):
self.reset_storage()		self.reset_storage()
nb_origins = len(new_origins)		nb_origins = len(new_origins)

self.storage.origin_add(new_origins)		self.storage.origin_add(new_origins)

origin_from = random.randint(1, nb_origins-1)		origin_from = random.randint(1, nb_origins-1)
origin_count = random.randint(1, nb_origins - origin_from)		origin_count = random.randint(1, nb_origins - origin_from)

actual_origins = list(		actual_origins = list(
self.storage.origin_get_range(origin_from=origin_from,		self.storage.origin_get_range(origin_from=origin_from,
origin_count=origin_count))		origin_count=origin_count))

for origin in actual_origins:		for origin in actual_origins:
del origin['id']

for origin in actual_origins:
self.assertIn(origin, new_origins)		self.assertIn(origin, new_origins)

origin_from = -1		origin_from = -1
origin_count = 5		origin_count = 5
origins = list(		origins = list(
self.storage.origin_get_range(origin_from=origin_from,		self.storage.origin_get_range(origin_from=origin_from,
origin_count=origin_count))		origin_count=origin_count))
self.assertEqual(len(origins), origin_count)		self.assertEqual(len(origins), origin_count)
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines	class CommonPropTestStorage:

@settings(suppress_health_check=[HealthCheck.too_slow])		@settings(suppress_health_check=[HealthCheck.too_slow])
@given(strategies.lists(objects(), max_size=2))		@given(strategies.lists(objects(), max_size=2))
def test_add_arbitrary(self, objects):		def test_add_arbitrary(self, objects):
self.reset_storage()		self.reset_storage()
for (obj_type, obj) in objects:		for (obj_type, obj) in objects:
obj = obj.to_dict()		obj = obj.to_dict()
if obj_type == 'origin_visit':		if obj_type == 'origin_visit':
origin_id = self.storage.origin_add_one(obj.pop('origin'))		origin_url = self.storage.origin_add_one(obj.pop('origin'))
if 'visit' in obj:		if 'visit' in obj:
del obj['visit']		del obj['visit']
self.storage.origin_visit_add(		self.storage.origin_visit_add(
origin_id, obj['date'], obj['type'])		origin_url, obj['date'], obj['type'])
else:		else:
method = getattr(self.storage, obj_type + '_add')		method = getattr(self.storage, obj_type + '_add')
try:		try:
method([obj])		method([obj])
except HashCollision:		except HashCollision:
pass		pass


@pytest.mark.db		@pytest.mark.db
class TestLocalStorage(CommonTestStorage, StorageTestDbFixture,		class TestLocalStorage(CommonTestStorage, StorageTestDbFixture,
unittest.TestCase):		unittest.TestCase):
"""Test the local storage"""		"""Test the local storage"""

# Can only be tested with local storage as you can't mock
# datetimes for the remote server
@given(strategies.booleans())
def test_fetch_history(self, use_url):
if not self._test_origin_ids and not use_url:
return
self.reset_storage()

origin_id = self.storage.origin_add_one(self.origin)
origin_id_or_url = self.origin['url'] if use_url else origin_id
with patch('datetime.datetime'):
datetime.datetime.now.return_value = self.fetch_history_date
fetch_history_id = self.storage.fetch_history_start(
origin_id_or_url)
datetime.datetime.now.assert_called_with(tz=datetime.timezone.utc)

with patch('datetime.datetime'):
datetime.datetime.now.return_value = self.fetch_history_end
self.storage.fetch_history_end(fetch_history_id,
self.fetch_history_data)

fetch_history = self.storage.fetch_history_get(fetch_history_id)
expected_fetch_history = self.fetch_history_data.copy()

expected_fetch_history['id'] = fetch_history_id
expected_fetch_history['origin'] = origin_id
expected_fetch_history['date'] = self.fetch_history_date
expected_fetch_history['duration'] = self.fetch_history_duration

self.assertEqual(expected_fetch_history, fetch_history)

# This test is only relevant on the local storage, with an actual		# This test is only relevant on the local storage, with an actual
# objstorage raising an exception		# objstorage raising an exception
def test_content_add_objstorage_exception(self):		def test_content_add_objstorage_exception(self):
self.storage.objstorage.add = Mock(		self.storage.objstorage.add = Mock(
side_effect=Exception('mocked broken objstorage')		side_effect=Exception('mocked broken objstorage')
)		)

with self.assertRaises(Exception) as e:		with self.assertRaises(Exception) as e:
▲ Show 20 Lines • Show All 115 Lines • Show Last 20 Lines