self = <swh.storage.tests.test_in_memory.TestInMemoryStorage object at 0x7fb79c58b780> swh_storage = <swh.storage.in_memory.InMemoryStorage object at 0x7fb7a7d30518> sample_data = <swh.storage.tests.storage_data.StorageData object at 0x7fb764591860> def test_stat_counters(self, swh_storage, sample_data): origin = sample_data.origin snapshot = sample_data.snapshot revision = sample_data.revision release = sample_data.release directory = sample_data.directory content = sample_data.content expected_keys = ["content", "directory", "origin", "revision"] # Initially, all counters are 0 swh_storage.refresh_stat_counters() counters = swh_storage.stat_counters() assert set(expected_keys) <= set(counters) for key in expected_keys: assert counters[key] == 0 # Add a content. Only the content counter should increase. swh_storage.content_add([content]) swh_storage.refresh_stat_counters() counters = swh_storage.stat_counters() assert set(expected_keys) <= set(counters) for key in expected_keys: if key != "content": assert counters[key] == 0 assert counters["content"] == 1 # Add other objects. Check their counter increased as well. swh_storage.origin_add([origin]) visit = OriginVisit( origin=origin.url, date=sample_data.date_visit2, type=sample_data.type_visit2, ) origin_visit1 = swh_storage.origin_visit_add([visit])[0] swh_storage.snapshot_add([snapshot]) swh_storage.origin_visit_status_add( [ OriginVisitStatus( origin=origin.url, visit=origin_visit1.visit, date=now(), status="ongoing", snapshot=snapshot.id, ) ] ) > swh_storage.directory_add([directory]) .tox/py3/lib/python3.7/site-packages/swh/storage/tests/storage_tests.py:3678: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <swh.storage.in_memory.InMemoryStorage object at 0x7fb7a7d30518> directories = [Directory(entries=(DirectoryEntry(name=b'foo', type='file', target=b'\xd8\x1c\xc0q\x0e\xb6\xcf\x9e\xfd[\x92\n\x84S\xe...x04', perms=<DentryPerms.directory: 16384>)), id=b'RV\xe8V\xa0\xa0\x89\x89f\xd6\xba\x14\xfe\xb48\x8b\x8b\x82\xd3\x02')] def directory_add(self, directories: List[Directory]) -> Dict[str, int]: to_add = {d.id: d for d in directories}.values() if not self._allow_overwrite: # Filter out directories that are already inserted. missing = self.directory_missing([dir_.id for dir_ in to_add]) directories = [dir_ for dir_ in directories if dir_.id in missing] self.journal_writer.directory_add(directories) for directory in directories: # Add directory entries to the 'directory_entry' table rows = [ DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) for entry in directory.entries ] > if self._directory_entries_insert_algo == "one-by-one": E AttributeError: 'InMemoryStorage' object has no attribute '_directory_entries_insert_algo' .tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:451: AttributeError