self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f367c1a7470>
swh_storage = <swh.storage.validate.ValidatingProxyStorage object at 0x7f367c561438>
def test_origin_visit_get_latest(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)
visit1 = OriginVisit(
origin=origin_url,
date=data.date_visit1,
type=data.type_visit1,
status="ongoing",
snapshot=None,
)
visit2 = OriginVisit(
origin=origin_url,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)
# Add a visit with the same date as the previous one
visit3 = OriginVisit(
origin=origin_url,
date=data.date_visit2,
type=data.type_visit2,
status="ongoing",
snapshot=None,
)
ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3])
> origin_visit1 = swh_storage.origin_visit_get_by(origin_url, ov1.visit)
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_storage.py:2020:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:921: in origin_visit_get_by
return self._origin_visit_apply_last_status(visit_)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <swh.storage.cassandra.storage.CassandraStorage object at 0x7f367c561470>
visit = {'date': datetime.datetime(2015, 1, 1, 23, 0, tzinfo=datetime.timezone.utc), 'metadata': None, 'origin': 'file:///dev/null', '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