mgr/smb: fix and test load_text function

The load_text function was a bit buggy but mainly worked until I started
to try and write a proper test for it. The error handling here is a bit
of a hack, but we can refine it later if people notice/complain.

Signed-off-by: John Mulligan <jmulligan@redhat.com>

diff --git a/src/pybind/mgr/smb/resources.py b/src/pybind/mgr/smb/resources.py
index 842b4387855120265bbc6d6ba6b7847dd4745ca5..bbda46b2256636f4bb9304e76adf173226099899 100644 (file)
--- a/src/pybind/mgr/smb/resources.py
+++ b/src/pybind/mgr/smb/resources.py
@@ -1,10 +1,12 @@
-from typing import Dict, List, Optional, Union, cast
+from typing import Dict, List, Optional, Tuple, Union, cast
 
+import errno
 import json
 
 import yaml
 
 from ceph.deployment.service_spec import PlacementSpec
+from object_format import ErrorResponseBase
 
 from . import resourcelib, validation
 from .enums import (
@@ -49,6 +51,26 @@ class InvalidResourceError(ValueError):
         return err
 
 
+class InvalidInputError(ValueError, ErrorResponseBase):
+    summary_max = 1024
+
+    def __init__(self, msg: str, content: str) -> None:
+        super().__init__(msg)
+        self.content = content
+
+    def to_simplified(self) -> Simplified:
+        return {
+            'input': self.content[: self.summary_max],
+            'truncated_input': len(self.content) > self.summary_max,
+            'msg': str(self),
+            'success': False,
+        }
+
+    def format_response(self) -> Tuple[int, str, str]:
+        data = json.dumps(self.to_simplified())
+        return -errno.EINVAL, data, "Invalid input"
+
+
 class _RBase:
     # mypy doesn't currently (well?) support class decorators adding methods
     # so we use a base class to add this method to all our resource classes.
@@ -425,19 +447,27 @@ SMBResource = Union[
 ]
 
 
-def load_text(blob: str) -> List[SMBResource]:
+def load_text(
+    blob: str, *, input_sample_max: int = 1024
+) -> List[SMBResource]:
     """Given JSON or YAML return a list of SMBResource objects deserialized
     from the input.
     """
+    json_err = None
     try:
-        data = yaml.safe_load(blob)
-    except ValueError:
-        pass
-    try:
+        # apparently JSON is not always as strict subset of YAML
+        # therefore trying to parse as JSON first is not a waste:
+        # https://john-millikin.com/json-is-not-a-yaml-subset
         data = json.loads(blob)
-    except ValueError:
-        pass
-    return load(data)
+    except ValueError as err:
+        json_err = err
+    try:
+        data = yaml.safe_load(blob) if json_err else data
+    except (ValueError, yaml.parser.ParserError) as err:
+        raise InvalidInputError(str(err), blob) from err
+    if not isinstance(data, (list, dict)):
+        raise InvalidInputError("input must be an object or list", blob)
+    return load(cast(Simplified, data))
 
 
 def load(data: Simplified) -> List[SMBResource]:

diff --git a/src/pybind/mgr/smb/tests/test_resources.py b/src/pybind/mgr/smb/tests/test_resources.py
index f1e85e585248cf338459ff050305ada4d0d5d9ad..d8edfafe5d400a94c85589609f2bf29d8c7c2dad 100644 (file)
--- a/src/pybind/mgr/smb/tests/test_resources.py
+++ b/src/pybind/mgr/smb/tests/test_resources.py
@@ -748,3 +748,93 @@ login_control:
     assert share.login_control[3].name == 'delbard'
     assert share.login_control[3].category == enums.LoginCategory.USER
     assert share.login_control[3].access == enums.LoginAccess.NONE
+
+
+@pytest.mark.parametrize(
+    "params",
+    [
+        # single share json
+        {
+            "txt": """
+{
+    "resource_type": "ceph.smb.share",
+    "cluster_id": "foo",
+    "share_id": "bar",
+    "cephfs": {"volume": "zippy", "path": "/"}
+}
+""",
+            'simplified': [
+                {
+                    'resource_type': 'ceph.smb.share',
+                    'cluster_id': 'foo',
+                    'share_id': 'bar',
+                    'intent': 'present',
+                    'name': 'bar',
+                    'cephfs': {
+                        'volume': 'zippy',
+                        'path': '/',
+                        'provider': 'samba-vfs',
+                    },
+                    'browseable': True,
+                    'readonly': False,
+                }
+            ],
+        },
+        # single share yaml
+        {
+            "txt": """
+resource_type: ceph.smb.share
+cluster_id: foo
+share_id: bar
+cephfs: {volume: zippy, path: /}
+""",
+            'simplified': [
+                {
+                    'resource_type': 'ceph.smb.share',
+                    'cluster_id': 'foo',
+                    'share_id': 'bar',
+                    'intent': 'present',
+                    'name': 'bar',
+                    'cephfs': {
+                        'volume': 'zippy',
+                        'path': '/',
+                        'provider': 'samba-vfs',
+                    },
+                    'browseable': True,
+                    'readonly': False,
+                }
+            ],
+        },
+        # invalid share yaml
+        {
+            "txt": """
+resource_type: ceph.smb.share
+""",
+            'exc_type': ValueError,
+            'error': 'missing',
+        },
+        # invalid input
+        {
+            "txt": """
+:
+""",
+            'exc_type': ValueError,
+            'error': 'parsing',
+        },
+        # invalid json, but useless yaml
+        {
+            "txt": """
+slithy
+""",
+            'exc_type': ValueError,
+            'error': 'input',
+        },
+    ],
+)
+def test_load_text(params):
+    if 'simplified' in params:
+        loaded = smb.resources.load_text(params['txt'])
+        assert params['simplified'] == [r.to_simplified() for r in loaded]
+    else:
+        with pytest.raises(params['exc_type'], match=params['error']):
+            smb.resources.load_text(params['txt'])