From 495427bd2305f8ea7825973f77feafb394f0bfcb Mon Sep 17 00:00:00 2001 From: Hezko Date: Mon, 21 Apr 2025 17:09:20 +0300 Subject: [PATCH] Revert "mgr/dashboard: Ns create size fix" Signed-off-by: Tomer Haskalovitch --- src/pybind/ceph_argparse.py | 96 +----------- .../mgr/dashboard/controllers/nvmeof.py | 7 +- src/pybind/mgr/dashboard/openapi.yaml | 1 + src/pybind/mgr/tests/test_ceph_argtypes.py | 144 ------------------ src/pybind/mgr/tests/test_cli_command.py | 33 ---- 5 files changed, 5 insertions(+), 276 deletions(-) delete mode 100644 src/pybind/mgr/tests/test_ceph_argtypes.py delete mode 100644 src/pybind/mgr/tests/test_cli_command.py diff --git a/src/pybind/ceph_argparse.py b/src/pybind/ceph_argparse.py index 703e73d3beec0..7377c3143e8c9 100644 --- a/src/pybind/ceph_argparse.py +++ b/src/pybind/ceph_argparse.py @@ -9,11 +9,9 @@ Copyright (C) 2013 Inktank Storage, Inc. LGPL-2.1 or LGPL-3.0. See file COPYING. """ -from abc import ABC, abstractmethod import copy import enum import math -import itertools import json import os import pprint @@ -25,11 +23,7 @@ import threading import uuid from collections import abc -from typing import cast, Any, Callable, Dict, Generic, List, Optional, Sequence, Tuple, Union, \ - Annotated, Generic, TypeVar - -S = TypeVar('S') # Input type -T = TypeVar('T') # Output type +from typing import cast, Any, Callable, Dict, Generic, List, Optional, Sequence, Tuple, Union if sys.version_info >= (3, 8): from typing import get_args, get_origin @@ -119,20 +113,6 @@ class JsonFormat(Exception): pass -class Converter(ABC, Generic[S, T]): - @abstractmethod - def convert(self, value: S) -> T: pass - - -def _get_annotation_metadata(tp): - if get_origin(tp) is Annotated: - annotated_args = get_args(tp) - try: - return annotated_args[1] - except IndexError: - return None - - class CephArgtype(object): """ Base class for all Ceph argument types @@ -213,10 +193,6 @@ class CephArgtype(object): attrs['req'] = 'false' if not positional: attrs['positional'] = 'false' - if annotation := _get_annotation_metadata(tp): - if isinstance(annotation, CephArgtype): - return annotation.argdesc(attrs) - CEPH_ARG_TYPES = { str: CephString, int: CephInt, @@ -256,10 +232,6 @@ class CephArgtype(object): @staticmethod def cast_to(tp, v): - if annotation := _get_annotation_metadata(tp): - if isinstance(annotation, Converter): - return annotation.convert(v) - PYTHON_TYPES = ( str, int, @@ -392,72 +364,6 @@ class CephString(CephArgtype): return super().argdesc(attrs) -class CephSizeBytes(CephArgtype, Converter[str, int]): - """ - Size in bytes. e.g. 1024, 1KB, 100MB, etc.. - """ - MULTIPLES = ['', "K", "M", "G", "T", "P"] - UNITS = { - f"{prefix}{suffix}": 1024 ** mult - for mult, prefix in enumerate(MULTIPLES) - for suffix in ['', 'B', 'iB'] - if not (prefix == '' and suffix == 'iB') - } - - def __init__(self, units_types: set = None): - self.units = set(CephSizeBytes.UNITS.keys()) - if units_types : - self.units &= units_types - self.re_exp = re.compile(f"r'^\d+({'|'.join(self.units)})?$'") - self.number_re_exp = re.compile(r'^\d+') - self.num_and_unit_re_exp = re.compile(r'(\d+)([A-Za-z]*)$') - - - def valid(self, s: str, partial: bool = False) -> None: - if not s: - raise ArgumentValid("Size string not provided.") - - number_str = ''.join(itertools.takewhile(str.isdigit, s)) - if not number_str: - raise ArgumentFormat("Size must start with a positive number.") - - unit = s[len(number_str):] - if unit and not unit.isalpha(): - raise ArgumentFormat("Invalid format. Expected format: [KB|MB|GB] (e.g., 100MB, 10KB, 1000).") - - if unit and unit not in self.units: - raise ArgumentValid(f'{unit} is not a valid size unit. Supported units: {self.units}') - self.val = s - - def __str__(self) -> str: - b = '|'.join(self.units) - return ''.format(b) - - def argdesc(self, attrs): - return super().argdesc(attrs) - - @staticmethod - def _convert_to_bytes(size: Union[int, str], default_unit=None): - if isinstance(size, int): - number = size - size = str(size) - else: - num_str = ''.join(filter(str.isdigit, size)) - number = int(num_str) - unit_str = ''.join(filter(str.isalpha, size)) - if not unit_str: - if not default_unit: - raise ValueError("No size unit was provided") - unit_str = default_unit - - if unit_str in CephSizeBytes.UNITS: - return number * CephSizeBytes.UNITS[unit_str] - raise ValueError(f"Invalid unit: {unit_str}") - - def convert(self, value: str) -> int: - return CephSizeBytes._convert_to_bytes(value, default_unit="B") - - class CephSocketpath(CephArgtype): """ Admin socket path; check that it's readable and S_ISSOCK diff --git a/src/pybind/mgr/dashboard/controllers/nvmeof.py b/src/pybind/mgr/dashboard/controllers/nvmeof.py index 6c3afe2fee202..b75a335bdc6fc 100644 --- a/src/pybind/mgr/dashboard/controllers/nvmeof.py +++ b/src/pybind/mgr/dashboard/controllers/nvmeof.py @@ -1,9 +1,8 @@ # -*- coding: utf-8 -*- import logging -from typing import Annotated, Any, Dict, Optional +from typing import Any, Dict, Optional import cherrypy -from ceph_argparse import CephSizeBytes from orchestrator import OrchestratorError from .. import mgr @@ -386,8 +385,8 @@ else: rbd_image_name: str, rbd_pool: str = "rbd", create_image: Optional[bool] = True, - size: Optional[int] = None, - rbd_image_size: Optional[Annotated[int, CephSizeBytes()]] = None, + size: Optional[int] = 1024, + rbd_image_size: Optional[int] = None, trash_image: Optional[bool] = False, block_size: int = 512, load_balancing_group: Optional[int] = None, diff --git a/src/pybind/mgr/dashboard/openapi.yaml b/src/pybind/mgr/dashboard/openapi.yaml index c35b96bf1f110..df071682e26fe 100755 --- a/src/pybind/mgr/dashboard/openapi.yaml +++ b/src/pybind/mgr/dashboard/openapi.yaml @@ -9111,6 +9111,7 @@ paths: description: RBD pool name type: string size: + default: 1024 description: RBD image size type: integer traddr: diff --git a/src/pybind/mgr/tests/test_ceph_argtypes.py b/src/pybind/mgr/tests/test_ceph_argtypes.py deleted file mode 100644 index 3f4140c582b30..0000000000000 --- a/src/pybind/mgr/tests/test_ceph_argtypes.py +++ /dev/null @@ -1,144 +0,0 @@ -import enum -import pytest -from ceph_argparse import CephSizeBytes, CephArgtype - -class TestCephArgtypeArgdesc: - - def test_to_argdesc_with_default(self): - attrs = {} - result = CephArgtype.to_argdesc(str, attrs, has_default=True) - assert result == "req=false,type=CephString" - - def test_to_argdesc_without_default(self): - attrs = {} - result = CephArgtype.to_argdesc(str, attrs, has_default=False) - assert result == "type=CephString" - - def test_to_argdesc_positional_false(self): - attrs = {} - result = CephArgtype.to_argdesc(str, attrs, positional=False) - assert result == "positional=false,type=CephString" - - def test_to_argdesc_str(self): - attrs = {} - result = CephArgtype.to_argdesc(str, attrs) - assert result == "type=CephString" - - def test_to_argdesc_int(self): - attrs = {} - result = CephArgtype.to_argdesc(int, attrs) - assert result == "type=CephInt" - - def test_to_argdesc_float(self): - attrs = {} - result = CephArgtype.to_argdesc(float, attrs) - assert result == "type=CephFloat" - - def test_to_argdesc_bool(self): - attrs = {} - result = CephArgtype.to_argdesc(bool, attrs) - assert result == "type=CephBool" - - def test_to_argdesc_invalid_type(self): - attrs = {} - # Simulate an invalid type that isn't in CEPH_ARG_TYPES - with pytest.raises(ValueError): - CephArgtype.to_argdesc(object, attrs) - - def test_to_argdesc_with_enum(self): - import enum - class MyEnum(enum.Enum): - A = "one" - B = "two" - - attrs = {} - result = CephArgtype.to_argdesc(MyEnum, attrs) - assert result == "strings=one|two,type=CephChoices" - - -class TestCephArgtypeCastTo: - def test_cast_to_with_str(self): - result = CephArgtype.cast_to(str, 123) - assert result == "123" - - def test_cast_to_with_int(self): - result = CephArgtype.cast_to(int, "123") - assert result == 123 - - def test_cast_to_with_float(self): - result = CephArgtype.cast_to(float, "123.45") - assert result == 123.45 - - def test_cast_to_with_bool(self): - result = CephArgtype.cast_to(bool, "True") - assert result is True - - def test_cast_to_with_enum(self): - class MyEnum(enum.Enum): - A = "one" - B = "two" - - result = CephArgtype.cast_to(MyEnum, "one") - assert result == MyEnum.A - - def test_cast_to_with_unknown_type(self): - class UnknownType: - pass - - with pytest.raises(ValueError): - CephArgtype.cast_to(UnknownType, "value") - - def test_cast_to_invalid_value_for_type(self): - with pytest.raises(ValueError): - CephArgtype.cast_to(int, "invalid_integer") - - -class TestConvertToBytes: - def test_with_kb(self): - assert CephSizeBytes._convert_to_bytes(f"100KB") == 102400 - assert CephSizeBytes._convert_to_bytes(f"100K") == 102400 - - def test_with_mb(self): - assert CephSizeBytes._convert_to_bytes(f"2MB") == 2 * 1024 ** 2 - assert CephSizeBytes._convert_to_bytes(f"2M") == 2 * 1024 ** 2 - - def test_with_gb(self): - assert CephSizeBytes._convert_to_bytes(f"1GB") == 1024 ** 3 - assert CephSizeBytes._convert_to_bytes(f"1G") == 1024 ** 3 - - def test_with_tb(self): - assert CephSizeBytes._convert_to_bytes(f"1TB") == 1024 ** 4 - assert CephSizeBytes._convert_to_bytes(f"1T") == 1024 ** 4 - - def test_with_pb(self): - assert CephSizeBytes._convert_to_bytes(f"1PB") == 1024 ** 5 - assert CephSizeBytes._convert_to_bytes(f"1P") == 1024 ** 5 - - def test_with_integer(self): - assert CephSizeBytes._convert_to_bytes(50, default_unit="B") == 50 - - def test_invalid_unit(self): - with pytest.raises(ValueError): - CephSizeBytes._convert_to_bytes("50XYZ") - - def test_b(self): - assert CephSizeBytes._convert_to_bytes(f"500B") == 500 - - def test_with_large_number(self): - assert CephSizeBytes._convert_to_bytes(f"1000GB") == 1000 * 1024 ** 3 - - def test_no_number(self): - with pytest.raises(ValueError): - CephSizeBytes._convert_to_bytes("GB") - - def test_no_unit_with_default_unit_gb(self): - assert CephSizeBytes._convert_to_bytes("500", default_unit="GB") == 500 * 1024 ** 3 - - def test_no_unit_with_no_default_unit_raises(self): - with pytest.raises(ValueError): - CephSizeBytes._convert_to_bytes("500") - - def test_unit_in_input_overrides_default(self): - assert CephSizeBytes._convert_to_bytes("50", default_unit="KB") == 50 * 1024 - assert CephSizeBytes._convert_to_bytes("50KB", default_unit="KB") == 50 * 1024 - assert CephSizeBytes._convert_to_bytes("50MB", default_unit="KB") == 50 * 1024 ** 2 diff --git a/src/pybind/mgr/tests/test_cli_command.py b/src/pybind/mgr/tests/test_cli_command.py deleted file mode 100644 index 817ecf9516342..0000000000000 --- a/src/pybind/mgr/tests/test_cli_command.py +++ /dev/null @@ -1,33 +0,0 @@ -from unittest.mock import MagicMock -import json -import pytest -from typing import Annotated - -from mgr_module import CLICommand -from ceph_argparse import CephSizeBytes - -@pytest.fixture(scope="class", name="command_with_size_annotation_name") -def fixture_command_with_size_annotation_name(): - return "test annotated size command" - - -@pytest.fixture(scope="class", name="command_with_size_annotation") -def fixture_command_with_size_annotation(command_with_size_annotation_name): - @CLICommand(command_with_size_annotation_name) - def func(_, param: Annotated[int, CephSizeBytes()]): # noqa # pylint: disable=unused-variable - return {'a': '1', 'param': param} - yield func - del CLICommand.COMMANDS[command_with_size_annotation_name] - assert command_with_size_annotation_name not in CLICommand.COMMANDS - - -class TestConvertAnnotatedType: - def test_command_convert_annotated_parameter(self, command_with_size_annotation, command_with_size_annotation_name): - result = CLICommand.COMMANDS[command_with_size_annotation_name].call(MagicMock(), {"param": f"{5 * 1024 ** 2}"}) - assert result['param'] == 5 * 1024 ** 2 - - result = CLICommand.COMMANDS[command_with_size_annotation_name].call(MagicMock(), {"param": f"{5 * 1024}KB"}) - assert result['param'] == 5 * 1024 ** 2 - - result = CLICommand.COMMANDS[command_with_size_annotation_name].call(MagicMock(), {"param": f"5MB"}) - assert result['param'] == 5 * 1024 ** 2 -- 2.47.3