# 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. # Code generated by Microsoft (R) Python Code Generator. # Changes may cause incorrect behavior and will be lost if the code is regenerated. # -------------------------------------------------------------------------- # import pytest # from devtools_testutils import recorded_by_proxy # from .testpreparer import WebPubSubClientTestBase, WebPubSubPreparer # @pytest.mark.skip("you may need to update the auto-generated test case before run it") # class TestWebPubSub(WebPubSubClientTestBase): # @WebPubSubPreparer() # @recorded_by_proxy # def test_get_service_status(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.get_service_status() # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_add_connections_to_groups(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.add_connections_to_groups( # groups_to_add={"filter": "str", "groups": ["str"]}, # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_close_all_connections(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.close_all_connections() # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_get_client_access_token(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.get_client_access_token() # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_remove_connections_from_groups(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.remove_connections_from_groups( # groups_to_remove={"filter": "str", "groups": ["str"]}, # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_send_to_all(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.send_to_all( # message=bytes("bytes", encoding="utf-8"), # content_type="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_close_connection(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.close_connection( # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_connection_exists(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.connection_exists( # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_send_to_connection(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.send_to_connection( # connection_id="str", # message=bytes("bytes", encoding="utf-8"), # content_type="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_remove_connection_from_all_groups(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.remove_connection_from_all_groups( # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_group_exists(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.group_exists( # group="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_close_group_connections(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.close_group_connections( # group="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_send_to_group(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.send_to_group( # group="str", # message=bytes("bytes", encoding="utf-8"), # content_type="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_list_connections(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.list_connections( # group="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_remove_connection_from_group(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.remove_connection_from_group( # group="str", # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_add_connection_to_group(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.add_connection_to_group( # group="str", # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_revoke_permission(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.revoke_permission( # permission="str", # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_has_permission(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.has_permission( # permission="str", # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_grant_permission(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.grant_permission( # permission="str", # connection_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_user_exists(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.user_exists( # user_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_close_user_connections(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.close_user_connections( # user_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_send_to_user(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.send_to_user( # user_id="str", # message=bytes("bytes", encoding="utf-8"), # content_type="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_remove_user_from_all_groups(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.remove_user_from_all_groups( # user_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_remove_user_from_group(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.remove_user_from_group( # group="str", # user_id="str", # ) # # please add some check logic here by yourself # # ... # @WebPubSubPreparer() # @recorded_by_proxy # def test_add_user_to_group(self, webpubsub_endpoint): # client = self.create_client(endpoint=webpubsub_endpoint) # response = client.add_user_to_group( # group="str", # user_id="str", # ) # # please add some check logic here by yourself # # ...