Address comments

src/middlewared/middlewared/api/v25_04_0/nfs.py

@@ -1,15 +1,14 @@
 import re
-from ipaddress import ip_network
-from typing import Annotated, Literal
+from ipaddress import ip_address, ip_network
+from typing import Annotated, Literal, TypeAlias
 
 from pydantic import (
-    ConfigDict, Field, IPvAnyAddress, IPvAnyNetwork,
-    AfterValidator, field_validator
+    Field, AfterValidator, field_validator
 )
 
 from middlewared.api.base import (
     BaseModel, Excluded, excluded_field, ForUpdateMetaclass, NonEmptyString,
-    single_argument_args, single_argument_result,
+    single_argument_args,
     TcpPort, exclude_tcp_ports
 )
 
@@ -24,6 +23,7 @@
 NFS_protocols = Literal["NFSV3", "NFSV4"]
 NFS_RDMA_DEFAULT_PORT = 20049
 EXCLUDED_PORTS = [NFS_RDMA_DEFAULT_PORT]
+NfsTcpPort: TypeAlias = Annotated[TcpPort | None, AfterValidator(exclude_tcp_ports(EXCLUDED_PORTS))]
 
 
 class NfsEntry(BaseModel):
@@ -37,36 +37,43 @@ class NfsEntry(BaseModel):
     v4_krb: bool
     """ Force Kerberos authentication on NFS shares. """
     v4_domain: str
-    """ Specify a DNS domain (NFSv4 only) """
-    bindip: list[NonEmptyString] = Field(default_factory=[])  # Using IPvAnyAddress creates JSON serialization errors
-    """ Limit the server IP addresses available for NFS """
-    mountd_port: Annotated[TcpPort | None, AfterValidator(exclude_tcp_ports(EXCLUDED_PORTS))]
-    """ Specify the mountd port binding """
-    rpcstatd_port: Annotated[TcpPort | None, AfterValidator(exclude_tcp_ports(EXCLUDED_PORTS))]
-    """ Specify the rpc.statd port binding """
-    rpclockd_port: Annotated[TcpPort | None, AfterValidator(exclude_tcp_ports(EXCLUDED_PORTS))]
-    """ Specify the rpc.lockd port binding """
+    """ Specify a DNS domain (NFSv4 only). """
+    bindip: list[NonEmptyString] = []  # Using IPvAnyAddress creates JSON serialization errors
+    """ Limit the server IP addresses available for NFS. """
+    mountd_port: NfsTcpPort
+    """ Specify the mountd port binding. """
+    rpcstatd_port: NfsTcpPort
+    """ Specify the rpc.statd port binding. """
+    rpclockd_port: NfsTcpPort
+    """ Specify the rpc.lockd port binding. """
     mountd_log: bool
-    """ Enable or disable mountd logging """
+    """ Enable or disable mountd logging. """
     statd_lockd_log: bool
-    """ Enable or disable statd and lockd logging """
+    """ Enable or disable statd and lockd logging. """
     v4_krb_enabled: bool
-    """ Status of NFSv4 authentication requirement (status only) """
+    """ Status of NFSv4 authentication requirement (status only). """
     userd_manage_gids: bool
-    """ Enable to allow server to manage gids """
+    """ Enable to allow server to manage gids. """
     keytab_has_nfs_spn: bool
-    """ Report status of NFS Principal Name in keytab (status only)"""
+    """ Report status of NFS Principal Name in keytab (status only). """
     managed_nfsd: bool
     """ Report status of 'servers' field.
-    If True the number of nfsd are managed by the server. (status only)"""
+    If True the number of nfsd are managed by the server (status only). """
     rdma: bool
-    """ Enable or disable NFS over RDMA.  Requires RDMA capable NIC """
+    """ Enable or disable NFS over RDMA.  Requires RDMA capable NIC. """
 
     @field_validator('bindip')
     @classmethod
     def check_bind_ip(cls, field_value: list):
         """ Custom validator for IP addresses to avoid JSON serialization errors """
-        all(isinstance(v, IPvAnyAddress) for v in field_value)
+        not_valid = []
+        for v in field_value:
+            try:
+                ip_address(v)
+            except ValueError:
+                not_valid.append(v)
+        if not_valid:
+            raise ValueError(f"The following do not appear to be valid IPv4 or IPv6 addresses: {not_valid}")
         return field_value
 
 
@@ -86,20 +93,19 @@ class NfsBindipChoicesArgs(BaseModel):
     pass
 
 
-@single_argument_result
 class NfsBindipChoicesResult(BaseModel):
     """ Return a dictionary of IP addresses """
-    model_config = ConfigDict(extra='allow')
+    result: dict[str, str]
 
 
 class NfsShareEntry(BaseModel):
     id: int
     path: NonEmptyString
     """ Local path to be exported. """
     aliases: list[NonEmptyString] = []
-    """ IGNORED for now """
+    """ IGNORED for now. """
     comment: str = ""
-    """ User comment associated with share """
+    """ User comment associated with share. """
     networks: list[NonEmptyString] = []
     """ List of authorized networks that are allowed to access the share having format
         "network/mask" CIDR notation. If empty, all networks are allowed. """
@@ -117,30 +123,38 @@ class NfsShareEntry(BaseModel):
     mapall_group: str | None = None
     """ Map all client groups to a specified group. """
     security: list[Literal["SYS", "KRB5", "KRB5I", "KRB5P"]] = []
-    """ Specify the security schema """
+    """ Specify the security schema. """
     enabled: bool = True
-    """ Enable or disable the share """
+    """ Enable or disable the share. """
     locked: bool | None
+    """ Lock state of the dataset (if encrypted). """
 
     @field_validator('networks', 'hosts')
     @classmethod
     def check_unique(cls, field_value: list):
         """ Custom validator to confirm unique entries """
         s = set()
         not_unique = []
-        for i, v in enumerate(field_value):
+        for v in field_value:
             if v in s:
                 not_unique.append(v)
             s.add(v)
-            if not_unique:
-                raise ValueError(f"Entries must be unique, the following are not: {not_unique}")
+        if not_unique:
+            raise ValueError(f"Entries must be unique, the following are not: {not_unique}")
         return field_value
 
     @field_validator('networks')
     @classmethod
     def check_valid_network(cls, field_value: list):
         """ Custom validator for IP networks """
-        all(isinstance(v, IPvAnyNetwork) for v in field_value)
+        not_valid = []
+        for v in field_value:
+            try:
+                ip_network(v, strict=False)
+            except ValueError:
+                not_valid.append(v)
+        if not_valid:
+            raise ValueError(f"The following do not appear to be valid IPv4 or IPv6 networks: {not_valid}")
 
         # Perform the courtesy conversion to a valid network format
         field_value = [str(ip_network(fv, strict=False)) for fv in field_value]
@@ -155,8 +169,8 @@ def no_spaces_or_quotes(cls, field_value: list):
         for v in field_value:
             if v is not None and regex.match(v):
                 not_valid.append(v)
-            if not_valid:
-                raise ValueError(f"Cannot contain spaces or quotes: {not_valid}")
+        if not_valid:
+            raise ValueError(f"Cannot contain spaces or quotes: {not_valid}")
         return field_value
 
 

tests/api2/test_300_nfs.py

@@ -669,8 +669,8 @@ def test_share_update(self, start_nfs, nfs_dataset_and_share):
             pp(["192.168.0.0/24", "192.168.1.0/24"], True, "", id="IPv4 - non-overlap"),
             pp(["192.168.0.0/16", "192.168.1.0/24"], False, "Overlapped", id="IPv4 - overlap wide"),
             pp(["192.168.0.0/24", "192.168.0.211/32"], False, "Overlapped", id="IPv4 - overlap narrow"),
-            pp(["192.168.0.0/64"], False, "does not appear to be an IPv4 or IPv6", id="IPv4 - invalid range"),
-            pp(["bogus_network"], False, "does not appear to be an IPv4 or IPv6", id="IPv4 - invalid format"),
+            pp(["192.168.0.0/64"], False, "do not appear to be valid IPv4 or IPv6", id="IPv4 - invalid range"),
+            pp(["bogus_network"], False, "do not appear to be valid IPv4 or IPv6", id="IPv4 - invalid format"),
             pp(["192.168.27.211"], True, "", id="IPv4 - auto-convert to CIDR"),
             # IPv6
             pp(["2001:0db8:85a3:0000:0000:8a2e::/96", "2001:0db8:85a3:0000:0000:8a2f::/96"],
@@ -680,7 +680,7 @@ def test_share_update(self, start_nfs, nfs_dataset_and_share):
             pp(["2001:0db8:85a3:0000:0000:8a2e::/96", "2001:0db8:85a3:0000:0000:8a2e:0370:7334/128"],
                False, "Overlapped", id="IPv6 - overlap narrow"),
             pp(["2001:0db8:85a3:0000:0000:8a2e:0370:7334/256"],
-               False, "does not appear to be an IPv4 or IPv6 network", id="IPv6 - invalid range"),
+               False, "do not appear to be valid IPv4 or IPv6", id="IPv6 - invalid range"),
             pp(["2001:0db8:85a3:0000:0000:8a2e:0370:7334"],
                True, "", id="IPv6 - auto-convert to CIDR"),
         ],
@@ -1401,9 +1401,9 @@ def test_runtime_debug(self, start_nfs):
     @pytest.mark.parametrize("TestType,param,errmsg", [
         pp("basic", [], None, id="basic settings"),
         pp("NotInChoice", ["1.2.3.4"], "Cannot use", id="Not in choices"),
-        pp("NotValidIP", ["a.b.c.d"], "does not appear", id="Not a valid IP"),
+        pp("NotValidIP", ["a.b.c.d"], "do not appear to be", id="Not a valid IP"),
         pp("HalfValid-1", ["", "8.8.8.8"], "Cannot use", id="2nd entry not in choices"),
-        pp("HalfValid-2", ["", "ixsystems.com"], "does not appear", id="2nd entry not valid IP"),
+        pp("HalfValid-2", ["", "ixsystems.com"], "do not appear to be", id="2nd entry not valid IP"),
         pp("HalfValid-3", ["", None], "Input should be", id="2nd entry is None")
     ])
     def test_bind_ip(self, start_nfs, TestType, param, errmsg):