self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f367c4eadd8>
swh_storage = <swh.storage.validate.ValidatingProxyStorage object at 0x7f369c314a90>
def test_origin_visit_status_add(self, swh_storage):
"""Correct origin visit statuses should add a new visit status
"""
origin1 = Origin.from_dict(data.origin2)
origin2 = Origin(url="new-origin")
swh_storage.origin_add([origin1, origin2])
ov1, ov2 = swh_storage.origin_visit_add(
[
OriginVisit(
origin=origin1.url,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
),
OriginVisit(
origin=origin2.url,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
),
]
)
snapshot_id = data.snapshot["id"]
date_visit_now = now()
visit_status1 = OriginVisitStatus(
origin=ov1.origin,
visit=ov1.visit,
date=date_visit_now,
status="full",
snapshot=snapshot_id,
)
date_visit_now = now()
visit_status2 = OriginVisitStatus(
origin=ov2.origin,
visit=ov2.visit,
date=date_visit_now,
status="ongoing",
snapshot=None,
metadata={"intrinsic": "something"},
)
swh_storage.origin_visit_status_add([visit_status1, visit_status2])
origin_visit1 = swh_storage.origin_visit_get_latest(
> origin1.url, require_snapshot=True
)
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_storage.py:1767:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:935: in origin_visit_get_latest
updated_visit = self._origin_visit_apply_last_status(visit)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <swh.storage.cassandra.storage.CassandraStorage object at 0x7f367c4e5518>
visit = {'date': datetime.datetime(2015, 1, 1, 23, 0, tzinfo=datetime.timezone.utc), 'metadata': None, 'origin': 'file:///dev/zero', 'snapshot': None, ...}
def _origin_visit_apply_last_status(self, visit: Dict[str, Any]) -> Dict[str, Any]:
"""Retrieve the latest visit status information for the origin visit.
Then merge it with the visit and return it.
"""
> visit_status = self._cql_runner.origin_visit_status_get_latest(
visit["origin"], visit["visit"]
)
E AttributeError: 'CqlRunner' object has no attribute 'origin_visit_status_get_latest'
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:866: AttributeError
TEST RESULT
TEST RESULT
- Run At
- Jun 17 2020, 10:13 AM