Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ def load_arguments_eh(self, _):
get_location_type, get_three_state_flag, get_resource_name_completion_list
from azure.cli.core.commands.validators import get_default_location_from_resource_group
from azure.cli.command_modules.eventhubs._completers import get_eventhubs_command_completion_list
from azure.cli.command_modules.eventhubs._validator import validate_storageaccount, validate_partner_namespace
from azure.cli.command_modules.eventhubs._validator import validate_storageaccount, validate_partner_namespace, validate_namespace_name
from knack.arguments import CLIArgumentType
from azure.cli.core.profiles import ResourceType
(SkuName, TlsVersion) = self.get_models('SkuName', 'TlsVersion', resource_type=ResourceType.MGMT_EVENTHUB)
Expand Down Expand Up @@ -88,7 +88,8 @@ def load_arguments_eh(self, _):
c.argument('min_compaction_lag_in_mins', type=int, arg_group='Retention-Description', options_list=['--min-lag', '--min-compaction-lag-in-mins'], help="The minimum time a message will remain ineligible for compaction in the log. This value is used when cleanupPolicy is Compact or DeleteOrCompact.")
c.argument('encoding', arg_group='Capture', options_list=['encoding'], help='Enumerates the possible values for the encoding format of capture description. Note: \'AvroDeflate\' will be deprecated in New API Version')
with self.argument_context('eventhubs eventhub list') as c:
c.argument('namespace_name', options_list=['--namespace-name'], id_part=None, help='Name of Namespace')
c.argument('namespace_name', options_list=['--namespace-name'], id_part=None, help='Name of Namespace',
validator=validate_namespace_name)

# Region Geo DR Configuration
with self.argument_context('eventhubs georecovery-alias set') as c:
Expand Down
10 changes: 10 additions & 0 deletions src/azure-cli/azure/cli/command_modules/eventhubs/_validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,16 @@
from azure.cli.core.util import CLIError


def validate_namespace_name(namespace):
import re
name = namespace.namespace_name
if name and not re.fullmatch(r'^[a-zA-Z][a-zA-Z0-9-]{4,48}[a-zA-Z0-9]$', name):
raise CLIError(
'Invalid namespace name "{}". The namespace name must be 6-50 characters, start with a letter, '
'end with a letter or digit, and contain only letters, digits, and hyphens.'.format(name)
)


def validate_storageaccount(cmd, namespace):
from azure.cli.core.commands.client_factory import get_subscription_id
from azure.mgmt.core.tools import is_valid_resource_id, resource_id
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

import unittest
from unittest.mock import MagicMock
from azure.cli.core.util import CLIError


class EventHubCommandValidationTests(unittest.TestCase):

def _make_namespace(self, name):
ns = MagicMock()
ns.namespace_name = name
return ns

def test_eventhub_list_namespace_name_valid_min_length(self):
from azure.cli.command_modules.eventhubs._validator import validate_namespace_name
validate_namespace_name(self._make_namespace('ns0001'))

def test_eventhub_list_namespace_name_valid_max_length(self):
from azure.cli.command_modules.eventhubs._validator import validate_namespace_name
validate_namespace_name(self._make_namespace('n' + '0' * 48 + '1'))

def test_eventhub_list_namespace_name_valid_short(self):
from azure.cli.command_modules.eventhubs._validator import validate_namespace_name
validate_namespace_name(self._make_namespace('ns00001'))

def test_eventhub_list_namespace_name_too_short(self):
from azure.cli.command_modules.eventhubs._validator import validate_namespace_name
with self.assertRaises(CLIError):
validate_namespace_name(self._make_namespace('ns001'))

def test_eventhub_list_namespace_name_too_long(self):
from azure.cli.command_modules.eventhubs._validator import validate_namespace_name
with self.assertRaises(CLIError):
validate_namespace_name(self._make_namespace('n' + '0' * 49 + '1'))

def test_eventhub_list_namespace_name_none_passes(self):
from azure.cli.command_modules.eventhubs._validator import validate_namespace_name
validate_namespace_name(self._make_namespace(None))

Loading