Page MenuHomeSoftware Heritage
Differential D7945

indexer_storage.yml: Replace deprecated alias 'local' with 'postgresql'
ClosedPublic
Actions

Authored by vlorentz on Jun 2 2022, 3:11 PM.

Details

Reviewers
ardumont
Group Reviewers
Reviewers
Commits
rDENVa183569f422b: indexer_storage.yml: Replace deprecated alias 'local' with 'postgresql'
Summary

It will be needed to run 'swh db upgrade'

Depends on D7944 being released

Diff Detail

Repository
rDENV Development environment
Branch
master
Lint
No Linters Available
Unit
No Unit Test Coverage
Build Status
Buildable 29773
Build 46532: arc lint + arc unit

Revision Contents
Changeset List

PathSize
docker/
conf/
2 lines

Diff 28691

View Options

docker/conf/indexer_storage.yml

Show First 20 Lines • Show All 54 Lines • ▼ Show 20 Lines):
yield storage, replayer yield storage, replayer
def test_storage_replayer(replayer_storage_and_client, caplog): def test_storage_replayer(replayer_storage_and_client, caplog):
"""Optimal replayer scenario. """Optimal replayer scenario.
This: This:
- writes objects to a source storage - writes objects to a source storage
- replayer consumes objects from the topic and replay them - replayer consumes objects from the topic and replays them
- a destination storage is filled from this - a destination storage is filled from this
In the end, both storages should have the same content. In the end, both storages should have the same content.
""" """
src, replayer = replayer_storage_and_client src, replayer = replayer_storage_and_client
# Fill Kafka using a source storage # Fill Kafka using a source storage
nb_sent = 0 nb_sent = 0
▲ Show 20 Lines • Show All 120 Lines • ▼ Show 20 Lines
def _check_replayed( def _check_replayed(
src: InMemoryStorage, dst: InMemoryStorage, exclude: Optional[Container] = None src: InMemoryStorage, dst: InMemoryStorage, exclude: Optional[Container] = None
): ):
"""Simple utility function to compare the content of 2 in_memory storages """Simple utility function to compare the content of 2 in_memory storages
""" """
expected_persons = set(src._persons) expected_persons = set(src._persons.values())
got_persons = set(dst._persons) got_persons = set(dst._persons.values())
assert got_persons == expected_persons assert got_persons == expected_persons
for attr in ( for attr in (
"contents", "contents",
"skipped_contents", "skipped_contents",
"directories", "directories",
"revisions", "revisions",
"releases", "releases",
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines return [
"status": "absent", "status": "absent",
"reason": "why not", "reason": "why not",
"origin": f"https://somewhere/{i}", "origin": f"https://somewhere/{i}",
"ctime": now, "ctime": now,
}, },
) )
for i in range(n) for i in range(n)
] ]
def test_storage_play_anonymized(
kafka_prefix: str, kafka_consumer_group: str, kafka_server: str
):
"""Optimal replayer scenario.
This:
- writes objects to the topic
ardumontUnsubmitted
Done
Inline Actions

replay`s`

ardumont: replay`s`
- replayer consumes objects from the topic and replay them
"""
writer_config = {
"cls": "kafka",
"brokers": [kafka_server],
"client_id": "kafka_writer",
"prefix": kafka_prefix,
"anonymize": True,
}
src_config = {"cls": "memory", "journal_writer": writer_config}
storage = get_storage(**src_config)
# Fill the src storage
nb_sent = 0
for obj_type, objs in TEST_OBJECTS.items():
if obj_type == "origin_visit":
# these have non-consistent API and are unrelated with what we
# want to test here
continue
method = getattr(storage, obj_type + "_add")
method(objs)
nb_sent += len(objs)
# Fill a destination storage from Kafka **using anonymized topics**
dst_storage = get_storage(cls="memory")
replayer = JournalClient(
brokers=kafka_server,
group_id=kafka_consumer_group,
prefix=kafka_prefix,
stop_after_objects=nb_sent,
privileged=False,
)
worker_fn = functools.partial(process_replay_objects, storage=dst_storage)
nb_inserted = replayer.process(worker_fn)
assert nb_sent == nb_inserted
check_replayed(storage, dst_storage, expected_anonymized=True)
# Fill a destination storage from Kafka **with stock (non-anonymized) topics**
dst_storage = get_storage(cls="memory")
replayer = JournalClient(
brokers=kafka_server,
group_id=kafka_consumer_group,
prefix=kafka_prefix,
stop_after_objects=nb_sent,
privileged=True,
)
worker_fn = functools.partial(process_replay_objects, storage=dst_storage)
nb_inserted = replayer.process(worker_fn)
assert nb_sent == nb_inserted
check_replayed(storage, dst_storage, expected_anonymized=False)
def check_replayed(src, dst, expected_anonymized=False):
"""Simple utility function to compare the content of 2 in_memory storages
If expected_anonymized is True, objects from the source storage are anonymized
before comparing with the destination storage.
"""
def maybe_anonymize(obj):
if expected_anonymized:
return obj.anonymize() or obj
return obj
expected_persons = {maybe_anonymize(person) for person in src._persons.values()}
got_persons = set(dst._persons.values())
assert got_persons == expected_persons
for attr in (
"contents",
"skipped_contents",
"directories",
"revisions",
"releases",
"snapshots",
"origins",
):
expected_objects = [
(id, maybe_anonymize(obj))
for id, obj in sorted(getattr(src, f"_{attr}").items())
]
got_objects = [
(id, obj) for id, obj in sorted(getattr(dst, f"_{attr}").items())
]
assert got_objects == expected_objects, f"Mismatch object list for {attr}"
About · Developer information · Legal