Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/replayer/tests/test_cli.py
| # Copyright (C) 2019-2022 The Software Heritage developers | # Copyright (C) 2019-2022 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 | ||||
| from collections import Counter | from collections import Counter | ||||
| import copy | import copy | ||||
| import functools | import functools | ||||
| import logging | import logging | ||||
| import re | import re | ||||
| from subprocess import Popen | from subprocess import Popen | ||||
| import tempfile | import tempfile | ||||
| from typing import Tuple | from typing import Any, Dict, Optional, Tuple | ||||
| from unittest.mock import patch | from unittest.mock import patch | ||||
| from click.testing import CliRunner | from click.testing import CliRunner | ||||
| from confluent_kafka import Producer | from confluent_kafka import Producer | ||||
| import msgpack | import msgpack | ||||
| import pytest | import pytest | ||||
| import yaml | import yaml | ||||
| ▲ Show 20 Lines • Show All 219 Lines • ▼ Show 20 Lines | result = invoke( | ||||
| "group_id": kafka_consumer_group, | "group_id": kafka_consumer_group, | ||||
| "prefix": kafka_prefix, | "prefix": kafka_prefix, | ||||
| }, | }, | ||||
| ) | ) | ||||
| expected = r"Done.\n" | expected = r"Done.\n" | ||||
| assert result.exit_code == 0, result.output | assert result.exit_code == 0, result.output | ||||
| assert re.fullmatch(expected, result.output, re.MULTILINE), result.output | assert re.fullmatch(expected, result.output, re.MULTILINE), result.output | ||||
| consumer_settings = None | consumer_settings: Optional[Dict[str, Any]] = None | ||||
| for record in caplog.records: | for record in caplog.records: | ||||
| if "Consumer settings" in record.message: | if "Consumer settings" in record.message: | ||||
| consumer_settings = record.args | consumer_settings = {} | ||||
| break | elif consumer_settings is not None and len(record.args) == 2: | ||||
| key, value = record.args | |||||
| consumer_settings[key] = value | |||||
| assert consumer_settings is not None, ( | assert consumer_settings is not None, ( | ||||
| "Failed to get consumer settings out of the consumer log. " | "Failed to get consumer settings out of the consumer log. " | ||||
| "See log capture for details." | "See log capture for details." | ||||
| ) | ) | ||||
| print(consumer_settings) | |||||
| assert consumer_settings["group.instance.id"] == "static-group-instance-id" | assert consumer_settings["group.instance.id"] == "static-group-instance-id" | ||||
| assert consumer_settings["session.timeout.ms"] == 60 * 10 * 1000 | assert consumer_settings["session.timeout.ms"] == 60 * 10 * 1000 | ||||
| assert consumer_settings["max.poll.interval.ms"] == 90 * 10 * 1000 | assert consumer_settings["max.poll.interval.ms"] == 90 * 10 * 1000 | ||||
| for (sha1, content) in contents.items(): | for (sha1, content) in contents.items(): | ||||
| assert sha1 in objstorages["dst"], sha1 | assert sha1 in objstorages["dst"], sha1 | ||||
| assert objstorages["dst"].get(sha1) == content | assert objstorages["dst"].get(sha1) == content | ||||
| ▲ Show 20 Lines • Show All 431 Lines • Show Last 20 Lines | |||||