Differential D3342 Diff 11853 swh/storage/tests/test_storage.py

Changeset View

Standalone View

swh/storage/tests/test_storage.py

# Copyright (C) 2015-2020 The Software Heritage developers		# Copyright (C) 2015-2020 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 attr
import copy		import copy
from contextlib import contextmanager		from contextlib import contextmanager
import datetime		import datetime
import inspect		import inspect
import itertools		import itertools
import math		import math
import queue		import queue
import random		import random
▲ Show 20 Lines • Show All 1,425 Lines • ▼ Show 20 Lines	def test_origin_visit_get_random(self, swh_storage):
visits = self._generate_random_visits()		visits = self._generate_random_visits()
visit_type = "git"		visit_type = "git"

# Add visits to those origins		# Add visits to those origins
for origin in data.origins:		for origin in data.origins:
origin_url = origin["url"]		origin_url = origin["url"]
for date_visit in visits:		for date_visit in visits:
visit = swh_storage.origin_visit_add(		visit = swh_storage.origin_visit_add(
[		[OriginVisit(origin=origin_url, date=date_visit, type=visit_type,)]
OriginVisit(
origin=origin_url,
date=date_visit,
type=visit_type,
status="ongoing",
snapshot=None,
)
]
)[0]		)[0]
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin_url,
visit=visit.visit,		visit=visit.visit,
date=now(),		date=now(),
status="full",		status="full",
Show All 19 Lines	def test_origin_visit_get_random_nothing_found(self, swh_storage):
visit_type = "hg"		visit_type = "hg"
# Add some visits outside of the random generation selection so nothing		# Add some visits outside of the random generation selection so nothing
# will be found by the random selection		# will be found by the random selection
visits = self._generate_random_visits(nb_visits=3, start=13, end=24)		visits = self._generate_random_visits(nb_visits=3, start=13, end=24)
for origin in data.origins:		for origin in data.origins:
origin_url = origin["url"]		origin_url = origin["url"]
for date_visit in visits:		for date_visit in visits:
visit = swh_storage.origin_visit_add(		visit = swh_storage.origin_visit_add(
[		[OriginVisit(origin=origin_url, date=date_visit, type=visit_type,)]
OriginVisit(
origin=origin_url,
date=date_visit,
type=visit_type,
status="ongoing",
snapshot=None,
)
]
)[0]		)[0]
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin_url,
visit=visit.visit,		visit=visit.visit,
date=now(),		date=now(),
status="full",		status="full",
▲ Show 20 Lines • Show All 146 Lines • ▼ Show 20 Lines	def test_origin_visit_add(self, swh_storage):

date_visit = now()		date_visit = now()
date_visit2 = date_visit + datetime.timedelta(minutes=1)		date_visit2 = date_visit + datetime.timedelta(minutes=1)

date_visit = round_to_milliseconds(date_visit)		date_visit = round_to_milliseconds(date_visit)
date_visit2 = round_to_milliseconds(date_visit2)		date_visit2 = round_to_milliseconds(date_visit2)

visit1 = OriginVisit(		visit1 = OriginVisit(
origin=origin1.url,		origin=origin1.url, date=date_visit, type=data.type_visit1,
date=date_visit,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)		)
visit2 = OriginVisit(		visit2 = OriginVisit(
origin=origin1.url,		origin=origin1.url, date=date_visit2, type=data.type_visit2,
date=date_visit2,
type=data.type_visit2,
status="partial",
snapshot=None,
)		)

# add once		# add once
ov1, ov2 = swh_storage.origin_visit_add([visit1, visit2])		ov1, ov2 = swh_storage.origin_visit_add([visit1, visit2])
# then again (will be ignored as they already exist)		# then again (will be ignored as they already exist)
origin_visit1, origin_visit2 = swh_storage.origin_visit_add([ov1, ov2])		origin_visit1, origin_visit2 = swh_storage.origin_visit_add([ov1, ov2])
assert ov1 == origin_visit1		assert ov1 == origin_visit1
assert ov2 == origin_visit2		assert ov2 == origin_visit2

		ovs1 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov1.visit,
		date=date_visit,
		status="created",
		snapshot=None,
		)
		ovs2 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov2.visit,
		date=date_visit2,
		status="created",
		snapshot=None,
		)

actual_origin_visits = list(swh_storage.origin_visit_get(origin1.url))		actual_origin_visits = list(swh_storage.origin_visit_get(origin1.url))
expected_visits = [		expected_visits = [
attr.evolve(visit1, visit=ov1.visit),		{**ovs1.to_dict(), "type": ov1.type},
attr.evolve(visit2, visit=ov2.visit),		{**ovs2.to_dict(), "type": ov2.type},
]		]

expected_visit_statuses = []
for visit in expected_visits:
visit_status = visit.to_dict()
visit_status.pop("type")
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status))

assert len(expected_visits) == len(actual_origin_visits)		assert len(expected_visits) == len(actual_origin_visits)

for visit in expected_visits:		for visit in expected_visits:
assert visit.to_dict() in actual_origin_visits		assert visit in actual_origin_visits

actual_objects = set(swh_storage.journal_writer.journal.objects)		actual_objects = set(swh_storage.journal_writer.journal.objects)
# we write to the journal as many times as we call the endpoint		# we write to the journal as many times as we call the endpoint
assert actual_objects == set(		assert actual_objects == set(
[("origin", origin1)]		[("origin", origin1)]
+ [("origin_visit", visit) for visit in expected_visits] * 2		+ [("origin_visit", visit) for visit in [ov1, ov2]] * 2
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses] * 2		+ [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]]
)		)

def test_origin_visit_add_validation(self, swh_storage):		def test_origin_visit_add_validation(self, swh_storage):
"""Unknown origin when adding visits should raise"""		"""Unknown origin when adding visits should raise"""
visit = OriginVisit(		visit = OriginVisit(
origin="something-unknown",		origin="something-unknown", date=now(), type=data.type_visit1,
date=now(),
type=data.type_visit1,
status="ongoing",
snapshot=None,
)		)
with pytest.raises(StorageArgumentException, match="Unknown origin"):		with pytest.raises(StorageArgumentException, match="Unknown origin"):
swh_storage.origin_visit_add([visit])		swh_storage.origin_visit_add([visit])

objects = list(swh_storage.journal_writer.journal.objects)		objects = list(swh_storage.journal_writer.journal.objects)
assert not objects		assert not objects

def test_origin_visit_status_add_validation(self, swh_storage):		def test_origin_visit_status_add_validation(self, swh_storage):
Show All 18 Lines	def test_origin_visit_status_add(self, swh_storage):
"""		"""
origin1 = Origin.from_dict(data.origin2)		origin1 = Origin.from_dict(data.origin2)
origin2 = Origin(url="new-origin")		origin2 = Origin(url="new-origin")
swh_storage.origin_add([origin1, origin2])		swh_storage.origin_add([origin1, origin2])

ov1, ov2 = swh_storage.origin_visit_add(		ov1, ov2 = swh_storage.origin_visit_add(
[		[
OriginVisit(		OriginVisit(
		origin=origin1.url, date=data.date_visit1, type=data.type_visit1,
		),
		OriginVisit(
		origin=origin2.url, date=data.date_visit2, type=data.type_visit2,
		),
		]
		)

		ovs1 = OriginVisitStatus(
origin=origin1.url,		origin=origin1.url,
		visit=ov1.visit,
date=data.date_visit1,		date=data.date_visit1,
type=data.type_visit1,		status="created",
status="ongoing",
snapshot=None,		snapshot=None,
),		)
OriginVisit(		ovs2 = OriginVisitStatus(
origin=origin2.url,		origin=origin2.url,
		visit=ov2.visit,
date=data.date_visit2,		date=data.date_visit2,
type=data.type_visit2,		status="created",
status="ongoing",
snapshot=None,		snapshot=None,
),
]
)		)

snapshot_id = data.snapshot["id"]		snapshot_id = data.snapshot["id"]
date_visit_now = now()		date_visit_now = now()
visit_status1 = OriginVisitStatus(		visit_status1 = OriginVisitStatus(
origin=ov1.origin,		origin=ov1.origin,
visit=ov1.visit,		visit=ov1.visit,
date=date_visit_now,		date=date_visit_now,
Show All 27 Lines	def test_origin_visit_status_add(self, swh_storage):
assert origin_visit2["status"] == "ongoing"		assert origin_visit2["status"] == "ongoing"
assert origin_visit2["snapshot"] is None		assert origin_visit2["snapshot"] is None
assert origin_visit2["metadata"] == {"intrinsic": "something"}		assert origin_visit2["metadata"] == {"intrinsic": "something"}

actual_objects = list(swh_storage.journal_writer.journal.objects)		actual_objects = list(swh_storage.journal_writer.journal.objects)

expected_origins = [origin1, origin2]		expected_origins = [origin1, origin2]
expected_visits = [ov1, ov2]		expected_visits = [ov1, ov2]
expected_visit_statuses = []		expected_visit_statuses = [ovs1, ovs2, visit_status1, visit_status2]
for visit in expected_visits: # out of origin-visit-add calls
visit_status = visit.to_dict()
visit_status.pop("type")
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status))

# out of origin-visit-status add calls
expected_visit_statuses += [visit_status1, visit_status2]
expected_objects = (		expected_objects = (
[("origin", o) for o in expected_origins]		[("origin", o) for o in expected_origins]
+ [("origin_visit", v) for v in expected_visits]		+ [("origin_visit", v) for v in expected_visits]
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses]		+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses]
)		)

assert len(actual_objects) == len(expected_objects)
for obj in expected_objects:		for obj in expected_objects:
assert obj in actual_objects		assert obj in actual_objects

def test_origin_visit_status_add_twice(self, swh_storage):		def test_origin_visit_status_add_twice(self, swh_storage):
"""Correct origin visit statuses should add a new visit status		"""Correct origin visit statuses should add a new visit status

"""		"""
origin1 = Origin.from_dict(data.origin2)		origin1 = Origin.from_dict(data.origin2)
swh_storage.origin_add([origin1])		swh_storage.origin_add([origin1])
ov1 = swh_storage.origin_visit_add(		ov1 = swh_storage.origin_visit_add(
[		[
OriginVisit(		OriginVisit(
origin=origin1.url,		origin=origin1.url, date=data.date_visit1, type=data.type_visit1,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
),		),
]		]
)[0]		)[0]

		ovs1 = OriginVisitStatus(
		origin=origin1.url,
		visit=ov1.visit,
		date=data.date_visit1,
		status="created",
		snapshot=None,
		)
snapshot_id = data.snapshot["id"]		snapshot_id = data.snapshot["id"]
date_visit_now = now()		date_visit_now = now()
visit_status1 = OriginVisitStatus(		visit_status1 = OriginVisitStatus(
origin=ov1.origin,		origin=ov1.origin,
visit=ov1.visit,		visit=ov1.visit,
date=date_visit_now,		date=date_visit_now,
status="full",		status="full",
snapshot=snapshot_id,		snapshot=snapshot_id,
Show All 10 Lines	def test_origin_visit_status_add_twice(self, swh_storage):
assert origin_visit1		assert origin_visit1
assert origin_visit1["status"] == "full"		assert origin_visit1["status"] == "full"
assert origin_visit1["snapshot"] == snapshot_id		assert origin_visit1["snapshot"] == snapshot_id

actual_objects = list(swh_storage.journal_writer.journal.objects)		actual_objects = list(swh_storage.journal_writer.journal.objects)

expected_origins = [origin1]		expected_origins = [origin1]
expected_visits = [ov1]		expected_visits = [ov1]
expected_visit_statuses = []		expected_visit_statuses = [ovs1, visit_status1, visit_status1]
for visit in expected_visits: # out of origin-visit-add calls
visit_status = visit.to_dict()
visit_status.pop("type")
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status))

# write twice in the journal		# write twice in the journal
expected_visit_statuses += [visit_status1] * 2
expected_objects = (		expected_objects = (
[("origin", o) for o in expected_origins]		[("origin", o) for o in expected_origins]
+ [("origin_visit", v) for v in expected_visits]		+ [("origin_visit", v) for v in expected_visits]
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses]		+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses]
)		)

assert len(actual_objects) == len(expected_objects)
for obj in expected_objects:		for obj in expected_objects:
assert obj in actual_objects		assert obj in actual_objects

def test_origin_visit_find_by_date(self, swh_storage):		def test_origin_visit_find_by_date(self, swh_storage):
# given		# given
origin_url = swh_storage.origin_add_one(data.origin)		origin = Origin.from_dict(data.origin)
		swh_storage.origin_add_one(data.origin)
visit1 = OriginVisit(		visit1 = OriginVisit(
origin=origin_url,		origin=origin.url, date=data.date_visit2, type=data.type_visit1,
		)
		visit2 = OriginVisit(
		origin=origin.url, date=data.date_visit3, type=data.type_visit2,
		)
		visit3 = OriginVisit(
		origin=origin.url, date=data.date_visit2, type=data.type_visit3,
		)
		ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])

		ovs1 = OriginVisitStatus(
		origin=origin.url,
		visit=ov1.visit,
date=data.date_visit2,		date=data.date_visit2,
type=data.type_visit1,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
visit2 = OriginVisit(		ovs2 = OriginVisitStatus(
origin=origin_url,		origin=origin.url,
		visit=ov2.visit,
date=data.date_visit3,		date=data.date_visit3,
type=data.type_visit2,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
visit3 = OriginVisit(		ovs3 = OriginVisitStatus(
origin=origin_url,		origin=origin.url,
		visit=ov3.visit,
date=data.date_visit2,		date=data.date_visit2,
type=data.type_visit3,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
		swh_storage.origin_visit_status_add([ovs1, ovs2, ovs3])
_, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])

# Simple case		# Simple case
visit = swh_storage.origin_visit_find_by_date(origin_url, data.date_visit3)		visit = swh_storage.origin_visit_find_by_date(origin.url, data.date_visit3)
assert visit["visit"] == ov2.visit		assert visit["visit"] == ov2.visit

# 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 = swh_storage.origin_visit_find_by_date(origin_url, data.date_visit2)		visit = swh_storage.origin_visit_find_by_date(origin.url, data.date_visit2)
assert visit["visit"] == ov3.visit		assert visit["visit"] == ov3.visit

def test_origin_visit_find_by_date__unknown_origin(self, swh_storage):		def test_origin_visit_find_by_date__unknown_origin(self, swh_storage):
swh_storage.origin_visit_find_by_date("foo", data.date_visit2)		swh_storage.origin_visit_find_by_date("foo", data.date_visit2)

def test_origin_visit_get_by(self, swh_storage):		def test_origin_visit_get_by(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)		origin_url = swh_storage.origin_add_one(data.origin)
origin_url2 = swh_storage.origin_add_one(data.origin2)		origin_url2 = swh_storage.origin_add_one(data.origin2)
visit = OriginVisit(		visit = OriginVisit(
origin=origin_url,		origin=origin_url, date=data.date_visit2, type=data.type_visit2,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)		)
origin_visit1 = swh_storage.origin_visit_add([visit])[0]		origin_visit1 = swh_storage.origin_visit_add([visit])[0]

swh_storage.snapshot_add([data.snapshot])		swh_storage.snapshot_add([data.snapshot])
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin_url,
▲ Show 20 Lines • Show All 84 Lines • ▼ Show 20 Lines

def test_origin_visit_get_latest_filter_type(self, swh_storage):		def test_origin_visit_get_latest_filter_type(self, swh_storage):
"""Filtering origin visit get latest with filter type should be ok		"""Filtering origin visit get latest with filter type should be ok

"""		"""
origin = Origin.from_dict(data.origin)		origin = Origin.from_dict(data.origin)
swh_storage.origin_add_one(origin)		swh_storage.origin_add_one(origin)
visit1 = OriginVisit(		visit1 = OriginVisit(
origin=origin.url,		origin=origin.url, date=data.date_visit1, type=data.type_visit1,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)		)
visit2 = OriginVisit(		visit2 = OriginVisit(
origin=origin.url,		origin=origin.url, date=data.date_visit2, type=data.type_visit2,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)		)
# Add a visit with the same date as the previous one		# Add a visit with the same date as the previous one
visit3 = OriginVisit(		visit3 = OriginVisit(
origin=origin.url,		origin=origin.url, date=data.date_visit2, type=data.type_visit2,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)		)
assert data.type_visit1 != data.type_visit2		assert data.type_visit1 != data.type_visit2
assert data.date_visit1 < data.date_visit2		assert data.date_visit1 < data.date_visit2

ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])		ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])

origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit)		origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit)
origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit)		origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit)

assert data.type_visit1 != data.type_visit2		assert data.type_visit1 != data.type_visit2

# Check type filter is ok		# Check type filter is ok
actual_ov1 = swh_storage.origin_visit_get_latest(		actual_ov1 = swh_storage.origin_visit_get_latest(
origin.url, type=data.type_visit1,		origin.url, type=data.type_visit1,
Show All 14 Lines	def test_origin_visit_get_latest_filter_type(self, swh_storage):
)		)
is None		is None
)		)

def test_origin_visit_get_latest(self, swh_storage):		def test_origin_visit_get_latest(self, swh_storage):
origin = Origin.from_dict(data.origin)		origin = Origin.from_dict(data.origin)
swh_storage.origin_add_one(origin)		swh_storage.origin_add_one(origin)
visit1 = OriginVisit(		visit1 = OriginVisit(
origin=origin.url,		origin=origin.url, date=data.date_visit1, type=data.type_visit1,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)		)
visit2 = OriginVisit(		visit2 = OriginVisit(
origin=origin.url,		origin=origin.url, date=data.date_visit2, type=data.type_visit2,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)		)
# Add a visit with the same date as the previous one		# Add a visit with the same date as the previous one
visit3 = OriginVisit(		visit3 = OriginVisit(
origin=origin.url,		origin=origin.url, date=data.date_visit2, type=data.type_visit2,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)		)

ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])		ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])

origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit)		origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit)
origin_visit2 = swh_storage.origin_visit_get_by(origin.url, ov2.visit)		origin_visit2 = swh_storage.origin_visit_get_by(origin.url, ov2.visit)
origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit)		origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit)
# Two visits, both with no snapshot		# Two visits, both with no snapshot
Show All 13 Lines	def test_origin_visit_get_latest(self, swh_storage):
origin=origin.url,		origin=origin.url,
visit=ov1.visit,		visit=ov1.visit,
date=now(),		date=now(),
status="ongoing",		status="ongoing",
snapshot=complete_snapshot.id,		snapshot=complete_snapshot.id,
)		)
]		]
)		)
assert {		actual_visit = swh_storage.origin_visit_get_latest(
		origin.url, require_snapshot=True
		)
		assert actual_visit == {
**origin_visit1,		**origin_visit1,
"snapshot": complete_snapshot.id,		"snapshot": complete_snapshot.id,
} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)		"status": "ongoing", # visit1 has status created now
		}

assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url)		assert origin_visit3 == swh_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
assert (		assert (
swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"])		swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"])
is None		is None
Show All 33 Lines	def test_origin_visit_get_latest(self, swh_storage):
status="ongoing",		status="ongoing",
snapshot=empty_snapshot.id,		snapshot=empty_snapshot.id,
)		)
]		]
)		)
assert {		assert {
**origin_visit2,		**origin_visit2,
"snapshot": empty_snapshot.id,		"snapshot": empty_snapshot.id,
		"status": "ongoing",
} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)		} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)

assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url)		assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url)

# Check that the status filter is still working		# Check that the status filter is still working
assert {		assert {
**origin_visit1,		**origin_visit1,
"snapshot": complete_snapshot.id,		"snapshot": complete_snapshot.id,
Show All 22 Lines	def test_origin_visit_get_latest(self, swh_storage):
"snapshot": complete_snapshot.id,		"snapshot": complete_snapshot.id,
"status": "full",		"status": "full",
} == swh_storage.origin_visit_get_latest(		} == swh_storage.origin_visit_get_latest(
origin.url, allowed_statuses=["full"], require_snapshot=True		origin.url, allowed_statuses=["full"], require_snapshot=True
)		)
assert {		assert {
**origin_visit3,		**origin_visit3,
"snapshot": complete_snapshot.id,		"snapshot": complete_snapshot.id,
		"status": "ongoing",
} == swh_storage.origin_visit_get_latest(origin.url)		} == swh_storage.origin_visit_get_latest(origin.url)

assert {		assert {
**origin_visit3,		**origin_visit3,
"snapshot": complete_snapshot.id,		"snapshot": complete_snapshot.id,
		"status": "ongoing",
} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)		} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True)

def test_origin_visit_status_get_latest(self, swh_storage):		def test_origin_visit_status_get_latest(self, swh_storage):
origin1 = Origin.from_dict(data.origin)		origin1 = Origin.from_dict(data.origin)
swh_storage.origin_add_one(data.origin)		swh_storage.origin_add_one(data.origin)

# to have some reference visits		# to have some reference visits

ov1, ov2 = swh_storage.origin_visit_add(		ov1, ov2 = swh_storage.origin_visit_add(
[		[
OriginVisit(		OriginVisit(
origin=origin1.url,		origin=origin1.url, date=data.date_visit1, type=data.type_visit1,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
),		),
OriginVisit(		OriginVisit(
origin=origin1.url,		origin=origin1.url, date=data.date_visit2, type=data.type_visit2,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
),		),
]		]
)		)

snapshot = Snapshot.from_dict(data.complete_snapshot)		snapshot = Snapshot.from_dict(data.complete_snapshot)
swh_storage.snapshot_add([snapshot])		swh_storage.snapshot_add([snapshot])

date_now = now()		date_now = now()
Show All 13 Lines	def test_origin_visit_status_get_latest(self, swh_storage):
visit=ov1.visit,		visit=ov1.visit,
date=data.date_visit2,		date=data.date_visit2,
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
ovs3 = OriginVisitStatus(		ovs3 = OriginVisitStatus(
origin=origin1.url,		origin=origin1.url,
visit=ov2.visit,		visit=ov2.visit,
date=data.date_visit2,		date=data.date_visit2 + datetime.timedelta(minutes=1), # to not be ignored
		ardumontAuthorUnsubmitted Done Inline Actions at it was initially the same date as the origin-visit-status from the origin-add call. This one got dropped silently `on conflict (origin, visit, date) ignore` style. ardumont: at it was initially the same date as the origin-visit-status from the origin-add call. This one…
status="ongoing",		status="ongoing",
snapshot=None,		snapshot=None,
)		)
ovs4 = OriginVisitStatus(		ovs4 = OriginVisitStatus(
origin=origin1.url,		origin=origin1.url,
visit=ov2.visit,		visit=ov2.visit,
date=date_now,		date=date_now,
status="full",		status="full",
▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Lines	def test_person_fullname_unicity(self, swh_storage):
# check committers are the same		# check committers are the same
assert revisions[0]["committer"] == revisions[1]["committer"]		assert revisions[0]["committer"] == revisions[1]["committer"]

def test_snapshot_add_get_empty(self, swh_storage):		def test_snapshot_add_get_empty(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)		origin_url = swh_storage.origin_add_one(data.origin)
ov1 = swh_storage.origin_visit_add(		ov1 = swh_storage.origin_visit_add(
[		[
OriginVisit(		OriginVisit(
origin=origin_url,		origin=origin_url, date=data.date_visit1, type=data.type_visit1,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)		)
]		]
)[0]		)[0]

actual_result = swh_storage.snapshot_add([data.empty_snapshot])		actual_result = swh_storage.snapshot_add([data.empty_snapshot])
assert actual_result == {"snapshot:add": 1}		assert actual_result == {"snapshot:add": 1}

date_now = now()		date_now = now()
Show All 11 Lines	def test_snapshot_add_get_empty(self, swh_storage):
)		)

by_id = swh_storage.snapshot_get(data.empty_snapshot["id"])		by_id = swh_storage.snapshot_get(data.empty_snapshot["id"])
assert by_id == {**data.empty_snapshot, "next_branch": None}		assert by_id == {**data.empty_snapshot, "next_branch": None}

by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit)		by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit)
assert by_ov == {**data.empty_snapshot, "next_branch": None}		assert by_ov == {**data.empty_snapshot, "next_branch": None}

data1 = {		ovs1 = OriginVisitStatus.from_dict(
		{
"origin": origin_url,		"origin": origin_url,
"date": data.date_visit1,		"date": data.date_visit1,
"visit": ov1.visit,		"visit": ov1.visit,
"status": "ongoing",		"status": "created",
"metadata": None,
"snapshot": None,		"snapshot": None,
		"metadata": None,
}		}
data2 = {		)
		ovs2 = OriginVisitStatus.from_dict(
		{
"origin": origin_url,		"origin": origin_url,
"date": date_now,		"date": date_now,
"visit": ov1.visit,		"visit": ov1.visit,
"status": "full",		"status": "full",
"metadata": None,		"metadata": None,
"snapshot": data.empty_snapshot["id"],		"snapshot": data.empty_snapshot["id"],
}		}
		)
actual_objects = list(swh_storage.journal_writer.journal.objects)		actual_objects = list(swh_storage.journal_writer.journal.objects)
assert actual_objects == [
		expected_objects = [
("origin", Origin.from_dict(data.origin)),		("origin", Origin.from_dict(data.origin)),
(		("origin_visit", ov1),
"origin_visit",		("origin_visit_status", ovs1,),
OriginVisit.from_dict({**data1, "type": data.type_visit1},),
),
("origin_visit_status", OriginVisitStatus.from_dict(data1)),
("snapshot", Snapshot.from_dict(data.empty_snapshot)),		("snapshot", Snapshot.from_dict(data.empty_snapshot)),
("origin_visit_status", OriginVisitStatus.from_dict(data2),),		("origin_visit_status", ovs2,),
]		]
		for obj in expected_objects:
		assert obj in actual_objects

def test_snapshot_add_get_complete(self, swh_storage):		def test_snapshot_add_get_complete(self, swh_storage):
origin_url = data.origin["url"]		origin_url = data.origin["url"]
origin_url = swh_storage.origin_add_one(data.origin)		origin_url = swh_storage.origin_add_one(data.origin)
visit = OriginVisit(		visit = OriginVisit(
origin=origin_url,		origin=origin_url,
date=data.date_visit1,		date=data.date_visit1,
type=data.type_visit1,		type=data.type_visit1,
▲ Show 20 Lines • Show All 164 Lines • ▼ Show 20 Lines	def test_snapshot_add_get_paginated(self, swh_storage):
assert snapshot == expected_snapshot		assert snapshot == expected_snapshot

def test_snapshot_add_get_filtered(self, swh_storage):		def test_snapshot_add_get_filtered(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)		origin_url = swh_storage.origin_add_one(data.origin)
visit = OriginVisit(		visit = OriginVisit(
origin=origin_url,		origin=origin_url,
date=data.date_visit1,		date=data.date_visit1,
type=data.type_visit1,		type=data.type_visit1,
status="ongoing",		# status="ongoing",
snapshot=None,		# snapshot=None,
)		)
origin_visit1 = swh_storage.origin_visit_add([visit])[0]		origin_visit1 = swh_storage.origin_visit_add([visit])[0]

swh_storage.snapshot_add([data.complete_snapshot])		swh_storage.snapshot_add([data.complete_snapshot])
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin_url,
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Lines	def test_snapshot_add_get_filtered_and_paginated(self, swh_storage):
assert snapshot == expected_snapshot		assert snapshot == expected_snapshot

def test_snapshot_add_get(self, swh_storage):		def test_snapshot_add_get(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)		origin_url = swh_storage.origin_add_one(data.origin)
visit = OriginVisit(		visit = OriginVisit(
origin=origin_url,		origin=origin_url,
date=data.date_visit1,		date=data.date_visit1,
type=data.type_visit1,		type=data.type_visit1,
status="ongoing",		# status="ongoing",
snapshot=None,		# snapshot=None,
)		)
origin_visit1 = swh_storage.origin_visit_add([visit])[0]		origin_visit1 = swh_storage.origin_visit_add([visit])[0]
visit_id = origin_visit1.visit		visit_id = origin_visit1.visit

swh_storage.snapshot_add([data.snapshot])		swh_storage.snapshot_add([data.snapshot])
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
Show All 15 Lines	def test_snapshot_add_get(self, swh_storage):
origin_visit_info = swh_storage.origin_visit_get_by(origin_url, visit_id)		origin_visit_info = swh_storage.origin_visit_get_by(origin_url, visit_id)
assert origin_visit_info["snapshot"] == data.snapshot["id"]		assert origin_visit_info["snapshot"] == data.snapshot["id"]

def test_snapshot_add_twice__by_origin_visit(self, swh_storage):		def test_snapshot_add_twice__by_origin_visit(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)		origin_url = swh_storage.origin_add_one(data.origin)
ov1 = swh_storage.origin_visit_add(		ov1 = swh_storage.origin_visit_add(
[		[
OriginVisit(		OriginVisit(
origin=origin_url,		origin=origin_url, date=data.date_visit1, type=data.type_visit1,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)		)
]		]
)[0]		)[0]
swh_storage.snapshot_add([data.snapshot])		swh_storage.snapshot_add([data.snapshot])
date_now2 = now()		date_now2 = now()

swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin_url,
visit=ov1.visit,		visit=ov1.visit,
date=date_now2,		date=date_now2,
status="ongoing",		status="ongoing",
snapshot=data.snapshot["id"],		snapshot=data.snapshot["id"],
)		)
]		]
)		)

by_ov1 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit)		by_ov1 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit)
assert by_ov1 == {**data.snapshot, "next_branch": None}		assert by_ov1 == {**data.snapshot, "next_branch": None}

ov2 = swh_storage.origin_visit_add(		ov2 = swh_storage.origin_visit_add(
[		[
OriginVisit(		OriginVisit(
origin=origin_url,		origin=origin_url, date=data.date_visit2, type=data.type_visit2,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)		)
]		]
)[0]		)[0]

swh_storage.snapshot_add([data.snapshot])		swh_storage.snapshot_add([data.snapshot])
date_now4 = now()		date_now4 = now()
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin_url,
visit=ov2.visit,		visit=ov2.visit,
date=date_now4,		date=date_now4,
status="ongoing",		status="ongoing",
snapshot=data.snapshot["id"],		snapshot=data.snapshot["id"],
)		)
]		]
)		)

by_ov2 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov2.visit)		by_ov2 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov2.visit)
assert by_ov2 == {**data.snapshot, "next_branch": None}		assert by_ov2 == {**data.snapshot, "next_branch": None}

data1 = {		ovs1 = OriginVisitStatus.from_dict(
		{
"origin": origin_url,		"origin": origin_url,
"date": data.date_visit1,		"date": data.date_visit1,
"visit": ov1.visit,		"visit": ov1.visit,
"status": "ongoing",		"status": "created",
"metadata": None,		"metadata": None,
"snapshot": None,		"snapshot": None,
}		}
data2 = {		)
		ovs2 = OriginVisitStatus.from_dict(
		{
"origin": origin_url,		"origin": origin_url,
"date": date_now2,		"date": date_now2,
"visit": ov1.visit,		"visit": ov1.visit,
"status": "ongoing",		"status": "ongoing",
"metadata": None,		"metadata": None,
"snapshot": data.snapshot["id"],		"snapshot": data.snapshot["id"],
}		}
data3 = {		)
		ovs3 = OriginVisitStatus.from_dict(
		{
"origin": origin_url,		"origin": origin_url,
"date": data.date_visit2,		"date": data.date_visit2,
"visit": ov2.visit,		"visit": ov2.visit,
"status": "ongoing",		"status": "created",
"metadata": None,		"metadata": None,
"snapshot": None,		"snapshot": None,
}		}
data4 = {		)
		ovs4 = OriginVisitStatus.from_dict(
		{
"origin": origin_url,		"origin": origin_url,
"date": date_now4,		"date": date_now4,
"visit": ov2.visit,		"visit": ov2.visit,
"status": "ongoing",		"status": "ongoing",
"metadata": None,		"metadata": None,
"snapshot": data.snapshot["id"],		"snapshot": data.snapshot["id"],
}		}
		)
actual_objects = list(swh_storage.journal_writer.journal.objects)		actual_objects = list(swh_storage.journal_writer.journal.objects)
assert actual_objects == [		expected_objects = [
("origin", Origin.from_dict(data.origin)),		("origin", Origin.from_dict(data.origin)),
(		("origin_visit", ov1),
"origin_visit",		("origin_visit_status", ovs1),
OriginVisit.from_dict({**data1, "type": data.type_visit1}),
),
("origin_visit_status", OriginVisitStatus.from_dict(data1)),
("snapshot", Snapshot.from_dict(data.snapshot)),		("snapshot", Snapshot.from_dict(data.snapshot)),
("origin_visit_status", OriginVisitStatus.from_dict(data2),),		("origin_visit_status", ovs2),
(		("origin_visit", ov2),
"origin_visit",		("origin_visit_status", ovs3),
OriginVisit.from_dict({**data3, "type": data.type_visit2}),		("origin_visit_status", ovs4),
),
("origin_visit_status", OriginVisitStatus.from_dict(data3)),
("origin_visit_status", OriginVisitStatus.from_dict(data4),),
]		]
		for obj in expected_objects:
		assert obj in actual_objects

def test_snapshot_get_random(self, swh_storage):		def test_snapshot_get_random(self, swh_storage):
swh_storage.snapshot_add(		swh_storage.snapshot_add(
[data.snapshot, data.empty_snapshot, data.complete_snapshot]		[data.snapshot, data.empty_snapshot, data.complete_snapshot]
)		)

assert swh_storage.snapshot_get_random() in {		assert swh_storage.snapshot_get_random() in {
data.snapshot["id"],		data.snapshot["id"],
▲ Show 20 Lines • Show All 760 Lines • ▼ Show 20 Lines	def test_origin_count_with_visit_no_visits(self, swh_storage):
assert swh_storage.origin_count(".user1.", regexp=True, with_visit=True) == 0		assert swh_storage.origin_count(".user1.", regexp=True, with_visit=True) == 0
assert swh_storage.origin_count(".user1.", regexp=False, with_visit=True) == 0		assert swh_storage.origin_count(".user1.", regexp=False, with_visit=True) == 0

def test_origin_count_with_visit_with_visits_no_snapshot(self, swh_storage):		def test_origin_count_with_visit_with_visits_no_snapshot(self, swh_storage):
swh_storage.origin_add([{"url": url} for url in self.ORIGINS])		swh_storage.origin_add([{"url": url} for url in self.ORIGINS])

origin_url = "https://github.com/user1/repo1"		origin_url = "https://github.com/user1/repo1"
visit = OriginVisit(		visit = OriginVisit(
origin=origin_url, date=now(), type="git", status="ongoing", snapshot=None		origin=origin_url,
		date=now(),
		type="git",
		# status="ongoing", snapshot=None
)		)
swh_storage.origin_visit_add([visit])		swh_storage.origin_visit_add([visit])

assert swh_storage.origin_count("github", with_visit=False) == 3		assert swh_storage.origin_count("github", with_visit=False) == 3
# it has a visit, but no snapshot, so with_visit=True => 0		# it has a visit, but no snapshot, so with_visit=True => 0
assert swh_storage.origin_count("github", with_visit=True) == 0		assert swh_storage.origin_count("github", with_visit=True) == 0

assert swh_storage.origin_count("gitlab", with_visit=False) == 2		assert swh_storage.origin_count("gitlab", with_visit=False) == 2
Show All 10 Lines	def test_origin_count_with_visit_with_visits_no_snapshot(self, swh_storage):
assert swh_storage.origin_count("github", regexp=True, with_visit=True) == 0		assert swh_storage.origin_count("github", regexp=True, with_visit=True) == 0

def test_origin_count_with_visit_with_visits_and_snapshot(self, swh_storage):		def test_origin_count_with_visit_with_visits_and_snapshot(self, swh_storage):
swh_storage.origin_add([{"url": url} for url in self.ORIGINS])		swh_storage.origin_add([{"url": url} for url in self.ORIGINS])

swh_storage.snapshot_add([data.snapshot])		swh_storage.snapshot_add([data.snapshot])
origin_url = "https://github.com/user1/repo1"		origin_url = "https://github.com/user1/repo1"
visit = OriginVisit(		visit = OriginVisit(
origin=origin_url, date=now(), type="git", status="ongoing", snapshot=None		origin=origin_url,
		date=now(),
		type="git",
		# status="ongoing", snapshot=None
)		)
visit = swh_storage.origin_visit_add([visit])[0]		visit = swh_storage.origin_visit_add([visit])[0]
swh_storage.origin_visit_status_add(		swh_storage.origin_visit_status_add(
[		[
OriginVisitStatus(		OriginVisitStatus(
origin=origin_url,		origin=origin_url,
visit=visit.visit,		visit=visit.visit,
date=now(),		date=now(),
Show All 25 Lines	def test_add_arbitrary(self, swh_storage, objects):
origin_url = obj.pop("origin")		origin_url = obj.pop("origin")
swh_storage.origin_add_one({"url": origin_url})		swh_storage.origin_add_one({"url": origin_url})
if "visit" in obj:		if "visit" in obj:
del obj["visit"]		del obj["visit"]
visit = OriginVisit(		visit = OriginVisit(
origin=origin_url,		origin=origin_url,
date=obj["date"],		date=obj["date"],
type=obj["type"],		type=obj["type"],
status="ongoing",		# status="ongoing",
snapshot=None,		# snapshot=None,
)		)
swh_storage.origin_visit_add([visit])		swh_storage.origin_visit_add([visit])
else:		else:
if obj_type == "content" and obj["status"] == "absent":		if obj_type == "content" and obj["status"] == "absent":
obj_type = "skipped_content"		obj_type = "skipped_content"
method = getattr(swh_storage, obj_type + "_add")		method = getattr(swh_storage, obj_type + "_add")
try:		try:
method([obj])		method([obj])
▲ Show 20 Lines • Show All 240 Lines • Show Last 20 Lines