Changeset View
Changeset View
Standalone View
Standalone View
swh/journal/tests/test_publisher_no_kafka.py
# Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 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 | ||||
import pytest | import pytest | ||||
import unittest | import unittest | ||||
from .conftest import ( | from .conftest import ( | ||||
JournalPublisherTest, TEST_CONFIG, | JournalPublisherTest, TEST_PUBLISHER_CONFIG, | ||||
CONTENTS, REVISIONS, RELEASES, ORIGINS | CONTENTS, REVISIONS, RELEASES, ORIGINS | ||||
) | ) | ||||
from swh.journal.publisher import MANDATORY_KEYS | from swh.journal.publisher import MANDATORY_KEYS | ||||
class JournalPublisherNoKafkaInMemoryStorage(JournalPublisherTest): | class JournalPublisherNoKafkaInMemoryStorage(JournalPublisherTest): | ||||
"""A journal publisher with: | """A journal publisher with: | ||||
- no kafka dependency | - no kafka dependency | ||||
Show All 13 Lines | def _prepare_journal(self, config): | ||||
pass | pass | ||||
class TestPublisherNoKafka(unittest.TestCase): | class TestPublisherNoKafka(unittest.TestCase): | ||||
"""This tests only the part not using any kafka instance | """This tests only the part not using any kafka instance | ||||
""" | """ | ||||
def setUp(self): | def setUp(self): | ||||
self.publisher = JournalPublisherNoKafkaInMemoryStorage(TEST_CONFIG) | self.publisher = JournalPublisherNoKafkaInMemoryStorage( | ||||
TEST_PUBLISHER_CONFIG) | |||||
self.contents = [{b'sha1': c['sha1']} for c in CONTENTS] | self.contents = [{b'sha1': c['sha1']} for c in CONTENTS] | ||||
self.revisions = [{b'id': c['id']} for c in REVISIONS] | self.revisions = [{b'id': c['id']} for c in REVISIONS] | ||||
self.releases = [{b'id': c['id']} for c in RELEASES] | self.releases = [{b'id': c['id']} for c in RELEASES] | ||||
# those needs id generation from the storage | # those needs id generation from the storage | ||||
# so initialization is different than other entities | # so initialization is different than other entities | ||||
self.origins = [{b'url': o['url'], | self.origins = [{b'url': o['url'], | ||||
b'type': o['type']} | b'type': o['type']} | ||||
for o in self.publisher.origins] | for o in self.publisher.origins] | ||||
▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Lines | class JournalPublisherCheckTest(JournalPublisherTest): | ||||
""" | """ | ||||
def _prepare_journal(self, config): | def _prepare_journal(self, config): | ||||
"""No journal for now | """No journal for now | ||||
""" | """ | ||||
pass | pass | ||||
def test_check_config_ok(test_config): | def test_check_config_ok(test_publisher_config): | ||||
"""Instantiate a publisher with the right config is fine | """Instantiate a publisher with the right config is fine | ||||
""" | """ | ||||
publisher = JournalPublisherCheckTest(test_config) | publisher = JournalPublisherCheckTest(test_publisher_config) | ||||
assert publisher is not None | assert publisher is not None | ||||
def test_check_config_ko(test_config): | def test_check_config_ko(test_publisher_config): | ||||
"""Instantiate a publisher with the wrong config should raise | """Instantiate a publisher with the wrong config should raise | ||||
""" | """ | ||||
for k in MANDATORY_KEYS: | for k in MANDATORY_KEYS: | ||||
conf = test_config.copy() | conf = test_publisher_config.copy() | ||||
conf.pop(k) | conf.pop(k) | ||||
with pytest.raises(ValueError) as e: | with pytest.raises(ValueError) as e: | ||||
JournalPublisherCheckTest(conf) | JournalPublisherCheckTest(conf) | ||||
error = ('Configuration error: The following keys must be' | error = ('Configuration error: The following keys must be' | ||||
' provided: %s' % (','.join([k]), )) | ' provided: %s' % (','.join([k]), )) | ||||
assert e.value.args[0] == error | assert e.value.args[0] == error |