text: #-------------------------------------------------------------------------
  # Copyright (c) Microsoft Corporation. All rights reserved.
  # Licensed under the MIT License. See License.txt in the project root for
  # license information.
  #--------------------------------------------------------------------------
  import logging
  import pytest
  import datetime
  
  import msrest
  from azure.servicebus.management import ServiceBusAdministrationClient, TopicProperties
  from utilities import get_logger
  from azure.core.exceptions import HttpResponseError, ResourceExistsError
  
  from devtools_testutils import AzureMgmtTestCase, CachedResourceGroupPreparer
  from servicebus_preparer import (
      CachedServiceBusNamespacePreparer,
      ServiceBusNamespacePreparer
  )
  
  from mgmt_test_utilities import clear_topics
  
  _logger = get_logger(logging.DEBUG)
  
  
  class ServiceBusAdministrationClientTopicTests(AzureMgmtTestCase):
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_create_by_name(self, servicebus_namespace_connection_string, **kwargs):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          topic_name = "topic_testaddf"
  
          try:
              mgmt_service.create_topic(topic_name)
              topic = mgmt_service.get_topic(topic_name)
              assert topic.name == topic_name
              assert topic.availability_status == 'Available'
              assert topic.status == 'Active'
          finally:
              mgmt_service.delete_topic(topic_name)
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_create_with_topic_description(self, servicebus_namespace_connection_string, **kwargs):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          topic_name = "iweidk"
          try:
              mgmt_service.create_topic(
                  topic_name=topic_name,
                  auto_delete_on_idle=datetime.timedelta(minutes=10),
                  default_message_time_to_live=datetime.timedelta(minutes=11),
                  duplicate_detection_history_time_window=datetime.timedelta(minutes=12),
                  enable_batched_operations=True,
                  enable_express=True,
                  enable_partitioning=True,
                  max_size_in_megabytes=3072
              )
              topic = mgmt_service.get_topic(topic_name)
              assert topic.name == topic_name
              assert topic.auto_delete_on_idle == datetime.timedelta(minutes=10)
              assert topic.default_message_time_to_live == datetime.timedelta(minutes=11)
              assert topic.duplicate_detection_history_time_window == datetime.timedelta(minutes=12)
              assert topic.enable_batched_operations
              assert topic.enable_express
              assert topic.enable_partitioning
              assert topic.max_size_in_megabytes % 3072 == 0
          finally:
              mgmt_service.delete_topic(topic_name)
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_create_duplicate(self, servicebus_namespace_connection_string, **kwargs):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          topic_name = "dqkodq"
          try:
              mgmt_service.create_topic(topic_name)
              with pytest.raises(ResourceExistsError):
                  mgmt_service.create_topic(topic_name)
          finally:
              mgmt_service.delete_topic(topic_name)
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_update_success(self, servicebus_namespace_connection_string, **kwargs):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          topic_name = "fjrui"
  
          try:
              topic_description = mgmt_service.create_topic(topic_name)
  
              # Try updating one setting.
              topic_description.default_message_time_to_live = datetime.timedelta(minutes=2)
              mgmt_service.update_topic(topic_description)
              topic_description = mgmt_service.get_topic(topic_name)
              assert topic_description.default_message_time_to_live == datetime.timedelta(minutes=2)
  
              # Now try updating all settings.
              topic_description.auto_delete_on_idle = datetime.timedelta(minutes=10)
              topic_description.default_message_time_to_live = datetime.timedelta(minutes=11)
              topic_description.duplicate_detection_history_time_window = datetime.timedelta(minutes=12)
              topic_description.enable_batched_operations = True
              topic_description.enable_express = True
              # topic_description.enable_partitioning = True # Cannot be changed after creation
              topic_description.max_size_in_megabytes = 3072
              # topic_description.requires_duplicate_detection = True # Read only
              # topic_description.requires_session = True # Cannot be changed after creation
              topic_description.support_ordering = True
  
              mgmt_service.update_topic(topic_description)
              topic_description = mgmt_service.get_topic(topic_name)
  
              assert topic_description.auto_delete_on_idle == datetime.timedelta(minutes=10)
              assert topic_description.default_message_time_to_live == datetime.timedelta(minutes=11)
              assert topic_description.duplicate_detection_history_time_window == datetime.timedelta(minutes=12)
              assert topic_description.enable_batched_operations == True
              assert topic_description.enable_express == True
              # assert topic_description.enable_partitioning == True
              assert topic_description.max_size_in_megabytes == 3072
              # assert topic_description.requires_duplicate_detection == True
              # assert topic_description.requires_session == True
              assert topic_description.support_ordering == True
          finally:
              mgmt_service.delete_topic(topic_name)
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_update_invalid(self, servicebus_namespace_connection_string, **kwargs):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          topic_name = "dfjfj"
          try:
              topic_description = mgmt_service.create_topic(topic_name)
  
              # handle a null update properly.
              with pytest.raises(AttributeError):
                  mgmt_service.update_topic(None)
  
              # handle an invalid type update properly.
              with pytest.raises(AttributeError):
                  mgmt_service.update_topic(Exception("test"))
  
              # change the name to a topic that doesn't exist; should fail.
              topic_description.name = "iewdm"
              with pytest.raises(HttpResponseError):
                  mgmt_service.update_topic(topic_description)
              topic_description.name = topic_name
  
              # change the name to a topic with an invalid name exist; should fail.
              topic_description.name = ''
              with pytest.raises(msrest.exceptions.ValidationError):
                  mgmt_service.update_topic(topic_description)
              topic_description.name = topic_name
  
              # change to a setting with an invalid value; should still fail.
              topic_description.duplicate_detection_history_time_window = datetime.timedelta(days=25)
              with pytest.raises(HttpResponseError):
                  mgmt_service.update_topic(topic_description)
              topic_description.duplicate_detection_history_time_window = datetime.timedelta(minutes=5)
          finally:
              mgmt_service.delete_topic(topic_name)
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_delete(self, servicebus_namespace_connection_string):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          mgmt_service.create_topic('test_topic')
          topics = list(mgmt_service.list_topics())
          assert len(topics) == 1
  
          mgmt_service.create_topic('txt/.-_123')
          topics = list(mgmt_service.list_topics())
          assert len(topics) == 2
  
          description = mgmt_service.get_topic('test_topic')
          mgmt_service.delete_topic(description.name)
  
          topics = list(mgmt_service.list_topics())
          assert len(topics) == 1 and topics[0].name == 'txt/.-_123'
  
          description = mgmt_service.get_topic('txt/.-_123')
          mgmt_service.delete_topic(description.name)
  
          topics = list(mgmt_service.list_topics())
          assert len(topics) == 0
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_list(self, servicebus_namespace_connection_string, **kwargs):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          topics = list(mgmt_service.list_topics())
          assert len(topics) == 0
          mgmt_service.create_topic("test_topic_1")
          mgmt_service.create_topic("test_topic_2")
          topics = list(mgmt_service.list_topics())
          assert len(topics) == 2
          assert topics[0].name == "test_topic_1"
          assert topics[1].name == "test_topic_2"
          mgmt_service.delete_topic("test_topic_1")
          mgmt_service.delete_topic("test_topic_2")
          topics = list(mgmt_service.list_topics())
          assert len(topics) == 0
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_list_runtime_properties(self, servicebus_namespace_connection_string, **kwargs):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          topics = list(mgmt_service.list_topics())
          topics_infos = list(mgmt_service.list_topics_runtime_properties())
  
          assert len(topics) == len(topics_infos) == 0
  
          mgmt_service.create_topic("test_topic")
  
          topics = list(mgmt_service.list_topics())
          topics_infos = list(mgmt_service.list_topics_runtime_properties())
  
          assert len(topics) == 1 and len(topics_infos) == 1
  
          assert topics[0].name == topics_infos[0].name == "test_topic"
  
          info = topics_infos[0]
  
          assert info.accessed_at_utc is not None
          assert info.updated_at_utc is not None
          assert info.subscription_count is 0
  
          assert info.size_in_bytes == 0
          assert info.scheduled_message_count == 0
  
          mgmt_service.delete_topic("test_topic")
          topics_infos = list(mgmt_service.list_topics_runtime_properties())
          assert len(topics_infos) == 0
  
      @CachedResourceGroupPreparer(name_prefix='servicebustest')
      @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
      def test_mgmt_topic_get_runtime_properties_basic(self, servicebus_namespace_connection_string):
          mgmt_service = ServiceBusAdministrationClient.from_connection_string(servicebus_namespace_connection_string)
          clear_topics(mgmt_service)
          mgmt_service.create_topic("test_topic")
          topic_runtime_properties = mgmt_service.get_topic_runtime_properties("test_topic")
  
          assert topic_runtime_properties
          assert topic_runtime_properties.name == "test_topic"
          assert topic_runtime_properties.created_at_utc is not None
          assert topic_runtime_properties.accessed_at_utc is not None
          assert topic_runtime_properties.updated_at_utc is not None
          assert topic_runtime_properties.size_in_bytes == 0
          assert topic_runtime_properties.subscription_count is 0
          assert topic_runtime_properties.scheduled_message_count == 0
          mgmt_service.delete_topic("test_topic")
  
      def test_topic_properties_constructor(self):
          with pytest.raises(TypeError):
              TopicProperties("randomname")
  