1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
# coding: utf-8
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# -------------------------------------------------------------------------
import pytest
from testcase import WebpubsubTest, WebpubsubPowerShellPreparer
from azure.messaging.webpubsubservice._operations._operations import build_send_to_all_request
from azure.core.exceptions import ServiceRequestError
class WebpubsubSmokeTest(WebpubsubTest):
@WebpubsubPowerShellPreparer()
def test_webpubsub_send_to_all(self, webpubsub_endpoint):
client = self.create_client(endpoint=webpubsub_endpoint, hub='hub')
client.send_to_all({'hello': 'test_webpubsub_send_to_all'})
@WebpubsubPowerShellPreparer()
def test_webpubsub_send_to_all_api_management_proxy(self, webpubsub_endpoint, webpubsub_reverse_proxy_endpoint=None):
client = self.create_client(endpoint=webpubsub_endpoint, hub='hub', reverse_proxy_endpoint=webpubsub_reverse_proxy_endpoint)
client.send_to_all({'hello': 'test_webpubsub_send_to_all_api_management_proxy'})
@WebpubsubPowerShellPreparer()
def test_webpubsub_send_request(self, webpubsub_endpoint):
client = self.create_client(endpoint=webpubsub_endpoint, hub='hub')
request = build_send_to_all_request('Hub', content='test_webpubsub_send_request', content_type='text/plain')
response = client.send_request(request)
assert response.status_code == 202
# If reverse_proxy_endpoint is not available, `ServiceRequestError` will be raised
@WebpubsubPowerShellPreparer()
def test_webpubsub_send_to_all_api_management_proxy_counter_test(self, webpubsub_endpoint):
client = self.create_client(endpoint=webpubsub_endpoint, hub='hub', reverse_proxy_endpoint='https://example.azure-api.net')
with pytest.raises(ServiceRequestError):
client.send_to_all({'hello': 'test_webpubsub_send_to_all_api_management_proxy_counter_test'})
@WebpubsubPowerShellPreparer()
def test_get_client_access_token(self, webpubsub_endpoint):
client = self.create_client(endpoint=webpubsub_endpoint, hub='hub')
access_token = client.get_client_access_token()
assert len(access_token) == 3
assert access_token['baseUrl'][:3] == "wss"
assert access_token['token']
assert access_token['url'][:3] == "wss"
@WebpubsubPowerShellPreparer()
def test_hello_world_with_connection_string(self, webpubsub_connection_string):
client = self.create_client(connection_string=webpubsub_connection_string, hub="hub")
client.send_to_all(message="Hello, World!", content_type="text/plain")
@WebpubsubPowerShellPreparer()
def test_hello_world_with_connection_string_json(self, webpubsub_connection_string):
client = self.create_client(connection_string=webpubsub_connection_string, hub="hub")
client.send_to_all(message={"hello": "world!"})
@WebpubsubPowerShellPreparer()
def test_hello_world_with_connection_string_binary(self, webpubsub_connection_string):
client = self.create_client(connection_string=webpubsub_connection_string, hub="hub")
client.send_to_all(message=b"Hello, World!", content_type="application/octet-stream")
@WebpubsubPowerShellPreparer()
def test_no_users_groups(self, webpubsub_connection_string):
client = self.create_client(connection_string=webpubsub_connection_string, hub="hub")
assert not client.user_exists(user_id="fake user")
assert not client.group_exists(group="fake group")
@WebpubsubPowerShellPreparer()
def test_remove_connection_from_all_groups(self, webpubsub_connection_string):
client = self.create_client(connection_string=webpubsub_connection_string, hub="hub")
client.remove_connection_from_all_groups(connection_id="fake connection id")
@WebpubsubPowerShellPreparer()
def test_send_with_filter(self, webpubsub_connection_string):
client = self.create_client(connection_string=webpubsub_connection_string, hub="hub")
client.send_to_all(message={"hello": "world!"}, filter="userId ne 'user1'", content_type="text/plain")
@WebpubsubPowerShellPreparer()
def test_get_client_access_key_with_groups(self, webpubsub_connection_string):
client = self.create_client(connection_string=webpubsub_connection_string, hub="hub")
client.get_client_access_token(user_id="user1", groups=["groups1"])
|
