Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: shutdown event loop if publisher fails to start and set exceptio…
…n on result future (#124) * feat: adding ability to create subscriptions at head * fix: lint errors * fix: remove absl dependency * fix: lint * feat: use default keyword args * fix: rename offset location to backlog location * fix: broken samples * fix: do not crash if pubsublite distribution can not be found when extracting semver * fix: properly shutdown event loop when failing to initialize publisher * fix: ensure proper shutdown on failure * fix: remove unused dep * fix: adding tests and requested changes
- Loading branch information
1 parent
8edef67
commit c2c2b00
Showing
5 changed files
with
101 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
78 changes: 78 additions & 0 deletions
78
tests/unit/pubsublite/cloudpubsub/internal/multiplexed_publisher_client_test.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,78 @@ | ||
# Copyright 2020 Google LLC | ||
# | ||
# Licensed under the Apache License, Version 2.0 (the "License"); | ||
# you may not use this file except in compliance with the License. | ||
# You may obtain a copy of the License at | ||
# | ||
# http://www.apache.org/licenses/LICENSE-2.0 | ||
# | ||
# Unless required by applicable law or agreed to in writing, software | ||
# distributed under the License is distributed on an "AS IS" BASIS, | ||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
# See the License for the specific language governing permissions and | ||
# limitations under the License. | ||
|
||
from asynctest.mock import MagicMock | ||
import pytest | ||
|
||
from google.cloud.pubsublite.cloudpubsub.internal.multiplexed_publisher_client import ( | ||
MultiplexedPublisherClient, | ||
) | ||
from google.cloud.pubsublite.cloudpubsub.internal.single_publisher import ( | ||
SinglePublisher, | ||
) | ||
from google.cloud.pubsublite.types import TopicPath | ||
from google.api_core.exceptions import GoogleAPICallError | ||
|
||
|
||
@pytest.fixture() | ||
def topic1(): | ||
return TopicPath.parse("projects/1/locations/us-central1-a/topics/topic1") | ||
|
||
|
||
@pytest.fixture() | ||
def topic2(): | ||
return TopicPath.parse("projects/1/locations/us-central1-a/topics/topic2") | ||
|
||
|
||
@pytest.fixture() | ||
def topic1_publisher(): | ||
topic1_publisher = MagicMock(spec=SinglePublisher) | ||
return topic1_publisher | ||
|
||
|
||
@pytest.fixture() | ||
def topic2_publisher(): | ||
topic2_publisher = MagicMock(spec=SinglePublisher) | ||
return topic2_publisher | ||
|
||
|
||
@pytest.fixture() | ||
def multiplexed_publisher(topic1, topic1_publisher, topic2_publisher): | ||
return MultiplexedPublisherClient( | ||
lambda topic: topic1_publisher if topic == topic1 else topic2_publisher | ||
) | ||
|
||
|
||
def test_multiplexed_publish( | ||
topic1, topic2, topic1_publisher, topic2_publisher, multiplexed_publisher | ||
): | ||
topic1_publisher.__enter__.return_value = topic1_publisher | ||
topic2_publisher.__enter__.return_value = topic2_publisher | ||
with multiplexed_publisher: | ||
multiplexed_publisher.publish(topic1, data=b"abc") | ||
topic1_publisher.__enter__.assert_called_once() | ||
topic1_publisher.publish.assert_called_once_with(data=b"abc", ordering_key="") | ||
multiplexed_publisher.publish(topic2, data=b"abc") | ||
topic2_publisher.__enter__.assert_called_once() | ||
topic2_publisher.publish.assert_called_once_with(data=b"abc", ordering_key="") | ||
topic1_publisher.__exit__.assert_called_once() | ||
topic2_publisher.__exit__.assert_called_once() | ||
|
||
|
||
def test_publisher_init_failure(topic1, topic1_publisher, multiplexed_publisher): | ||
topic1_publisher.__enter__.side_effect = GoogleAPICallError("error") | ||
with multiplexed_publisher: | ||
future = multiplexed_publisher.publish(topic1, data=b"abc") | ||
with pytest.raises(GoogleAPICallError): | ||
future.result() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters