Enforce structure on the open_port and close_port methods.

Add typing overloads to ensure that type checking will raise issues.

Good:
 * open_port('icmp')
 * open_port('tcp', 1)
 * open_port('udp', 2)
 * open_port(port=3)

Bad:
 * open_port()
 * open_port('icmp', 1)
 * open_port('tcp')
 * open_port('udp')
 * open_port(4, 'tcp')

Also check that nothing bad has been done at runtime, raising an error (ModelError to be consistent with ops.testing) if one of the invalid specifications is used.

ops/model.py

@@ -593,8 +593,76 @@ def add_secret(self, content: Dict[str, str], *,
             owner='unit')
         return Secret(self._backend, id=id, label=label, content=content)
 
+    if typing.TYPE_CHECKING:
+        @typing.overload
+        def open_port(self, protocol: typing.Literal['tcp', 'udp'], port: int) -> None:
+            """Open a port with the given protocol for this unit.
+
+            Calling this registers intent with Juju that the application should be
+            accessed on the given port, but the port isn't actually opened
+            externally until the admin runs "juju expose".
+
+            On Kubernetes sidecar charms, the ports opened are not strictly
+            per-unit: Juju will open the union of ports from all units.
+            However, normally charms should make the same open_port() call from
+            every unit.
+
+            Use :func:`set_ports()` for a more declarative approach where all of
+            the ports that should be open are provided in a single call.
+
+            Args:
+                protocol: String representing the protocol; must be one of
+                    'tcp' or 'udp' (lowercase is recommended, but
+                    uppercase is also supported).
+                port: The port to open.
+            """
+
+        @typing.overload
+        def open_port(self, *, port: int, protocol: typing.Literal['tcp'] = 'tcp') -> None:
+            """Open a tcp port for this unit.
+
+            Calling this registers intent with Juju that the application should be
+            accessed on the given port, but the port isn't actually opened
+            externally until the admin runs "juju expose".
+
+            On Kubernetes sidecar charms, the ports opened are not strictly
+            per-unit: Juju will open the union of ports from all units.
+            However, normally charms should make the same open_port() call from
+            every unit.
+
+            Use :func:`set_ports()` for a more declarative approach where all of
+            the ports that should be open are provided in a single call.
+
+            Args:
+                protocol: Must be 'tcp' (lowercase is recommended, but
+                    uppercase is also supported).
+                port: The port to open.
+            """
+
+        @typing.overload
+        def open_port(self, protocol: typing.Literal['icmp'], port: None = None) -> None:
+            """Open a icmp port for this unit.
+
+            Calling this registers intent with Juju that the application should be
+            accessed on the given port, but the port isn't actually opened
+            externally until the admin runs "juju expose".
+
+            On Kubernetes sidecar charms, the ports opened are not strictly
+            per-unit: Juju will open the union of ports from all units.
+            However, normally charms should make the same open_port() call from
+            every unit.
+
+            Use :func:`set_ports()` for a more declarative approach where all of
+            the ports that should be open are provided in a single call.
+
+            Args:
+                protocol: Must be 'icmp' (lowercase is recommended, but
+                    uppercase is also supported).
+                port: Must be None.
+            """
+
     def open_port(self, protocol: typing.Literal['tcp', 'udp', 'icmp'] = 'tcp',
-                  port: Optional[int] = None):
+                  port: Optional[int] = None) -> None:
         """Open a port with the given protocol for this unit.
 
         Calling this registers intent with Juju that the application should be
@@ -615,11 +683,81 @@ def open_port(self, protocol: typing.Literal['tcp', 'udp', 'icmp'] = 'tcp',
                 uppercase is also supported).
             port: The port to open. Required for TCP and UDP; not allowed
                 for ICMP.
-        """
-        self._backend.open_port(protocol.lower(), port)
+
+        Raises:
+            ModelError: If ``port`` is provided when ``protocol`` is 'icmp'
+                or ``port`` is not provided when ``protocol`` is 'tcp' or
+                'udp'.
+        """
+        normalised_protocol : str = protocol.lower()
+        if normalised_protocol == 'icmp' and port is not None:
+            raise ModelError("icmp cannot have a port number specified")
+        elif normalised_protocol in ('tcp', 'udp') and port is None:
+            raise ModelError(f"{normalised_protocol} must have a port number specified")
+        self._backend.open_port(normalised_protocol, port)
+
+    if typing.TYPE_CHECKING:
+        @typing.overload
+        def close_port(self, protocol: typing.Literal['tcp', 'udp'], port: int) -> None:
+            """Close a port with the given protocol for this unit.
+
+            On Kubernetes sidecar charms, Juju will only close the port once the
+            last unit that opened that port has closed it. However, this is
+            usually not an issue; normally charms should make the same
+            close_port() call from every unit.
+
+            Use :func:`set_ports()` for a more declarative approach where all
+            of the ports that should be open are provided in a single call.
+            For example, ``set_ports()`` will close all open ports.
+
+            Args:
+                protocol: String representing the protocol; must be one of
+                    'tcp' or 'udp' (lowercase is recommended, but
+                    uppercase is also supported).
+                port: The port to open.
+            """
+
+        @typing.overload
+        def close_port(self, *, port: int, protocol: typing.Literal['tcp'] = 'tcp') -> None:
+            """Close a port with the given protocol for this unit.
+
+            On Kubernetes sidecar charms, Juju will only close the port once the
+            last unit that opened that port has closed it. However, this is
+            usually not an issue; normally charms should make the same
+            close_port() call from every unit.
+
+            Use :func:`set_ports()` for a more declarative approach where all
+            of the ports that should be open are provided in a single call.
+            For example, ``set_ports()`` will close all open ports.
+
+            Args:
+                protocol: Must be 'tcp' (lowercase is recommended, but
+                    uppercase is also supported).
+                port: The port to open.
+            """
+
+        @typing.overload
+        def close_port(self, protocol: typing.Literal['icmp'], port: None = None) -> None:
+            """Close a port with the given protocol for this unit.
+
+            On Kubernetes sidecar charms, Juju will only close the port once the
+            last unit that opened that port has closed it. However, this is
+            usually not an issue; normally charms should make the same
+            close_port() call from every unit.
+
+            Use :func:`set_ports()` for a more declarative approach where all
+            of the ports that should be open are provided in a single call.
+            For example, ``set_ports()`` will close all open ports.
+
+            Args:
+                protocol: String representing the protocol; must be one of
+                    'tcp', 'udp', or 'icmp' (lowercase is recommended, but
+                    uppercase is also supported).
+                port: Must be None.
+            """
 
     def close_port(self, protocol: typing.Literal['tcp', 'udp', 'icmp'] = 'tcp',
-                   port: Optional[int] = None):
+                   port: Optional[int] = None) -> None:
         """Close a port with the given protocol for this unit.
 
         On Kubernetes sidecar charms, Juju will only close the port once the
@@ -637,8 +775,18 @@ def close_port(self, protocol: typing.Literal['tcp', 'udp', 'icmp'] = 'tcp',
                 uppercase is also supported).
             port: The port to open. Required for TCP and UDP; not allowed
                 for ICMP.
-        """
-        self._backend.close_port(protocol.lower(), port)
+
+        Raises:
+            ModelError: If ``port`` is provided when ``protocol`` is 'icmp'
+                or ``port`` is not provided when ``protocol`` is 'tcp' or
+                'udp'.
+        """
+        normalised_protocol : str = protocol.lower()
+        if normalised_protocol == 'icmp' and port is not None:
+            raise ModelError("icmp cannot have a port number specified")
+        elif normalised_protocol in ('tcp', 'udp') and port is None:
+            raise ModelError(f"{normalised_protocol} must have a port number specified")
+        self._backend.close_port(normalised_protocol, port)
 
     def opened_ports(self) -> Set['Port']:
         """Return a list of opened ports for this unit."""

test/test_model.py

@@ -3339,6 +3339,13 @@ def test_open_port_error(self):
             ['open-port', '8080/ftp'],
         ])
 
+        with self.assertRaises(ops.ModelError) as cm:
+            self.unit.open_port()
+        with self.assertRaises(ops.ModelError) as cm:
+            self.unit.open_port('icmp', 8000)
+        with self.assertRaises(ops.ModelError) as cm:
+            self.unit.open_port('udp')
+
     def test_close_port(self):
         fake_script(self, 'close-port', 'exit 0')
 
@@ -3365,6 +3372,13 @@ def test_close_port_error(self):
             ['close-port', '8080/ftp'],
         ])
 
+        with self.assertRaises(ops.ModelError) as cm:
+            self.unit.close_port()
+        with self.assertRaises(ops.ModelError) as cm:
+            self.unit.close_port('icmp', 8000)
+        with self.assertRaises(ops.ModelError) as cm:
+            self.unit.close_port('udp')
+
     def test_opened_ports(self):
         fake_script(self, 'opened-ports', """echo 8080/tcp; echo icmp""")
 
@@ -3407,31 +3421,51 @@ def test_opened_ports_warnings(self):
         ])
 
     def test_set_ports(self):
+        # No existing ports, open new ones.
         fake_script(self, 'open-port', 'exit 0')
         fake_script(self, 'close-port', 'exit 0')
         fake_script(self, 'opened-ports', 'exit 0')
         self.unit.set_ports(8000, 8025)
         calls = fake_script_calls(self, clear=True)
         self.assertEqual(calls.pop(0), ['opened-ports', ''])
-        # We make no guarantee on the order the ports are opened.
-        calls.sort()
+        calls.sort()  # We make no guarantee on the order the ports are opened.
         self.assertEqual(calls, [
             ['open-port', '8000/tcp'],
             ['open-port', '8025/tcp'],
         ])
-        fake_script(self, 'opened-ports', 'echo 8025/tcp')
-        self.unit.set_ports(ops.Port('udp', 8022))
+        # Two open ports, leave one alone and open another one.
+        fake_script(self, 'opened-ports', 'echo 8025/tcp; echo 8028/tcp')
+        self.unit.set_ports(ops.Port('udp', 8022), 8028)
         self.assertEqual(fake_script_calls(self, clear=True), [
             ['opened-ports', ''],
             ['close-port', '8025/tcp'],
             ['open-port', '8022/udp'],
         ])
+        # Completely replace the opened ports.
+        fake_script(self, 'opened-ports', 'echo 8025/tcp; echo 8028/tcp')
+        self.unit.set_ports(8001, 8002)
+        calls = fake_script_calls(self, clear=True)
+        self.assertEqual(calls.pop(0), ['opened-ports', ''])
+        calls.sort()
+        self.assertEqual(calls, [
+            ['close-port', '8025/tcp'],
+            ['close-port', '8028/tcp'],
+            ['open-port', '8001/tcp'],
+            ['open-port', '8002/tcp'],
+        ])
+        # Close everything.
         fake_script(self, 'opened-ports', 'echo 8022/udp')
         self.unit.set_ports()
         self.assertEqual(fake_script_calls(self, clear=True), [
             ['opened-ports', ''],
             ['close-port', '8022/udp'],
         ])
+        # Opening an already open port is a no-op.
+        fake_script(self, 'opened-ports', 'echo 8000/tcp')
+        self.unit.set_ports(ops.Port('tcp', 8000))
+        self.assertEqual(fake_script_calls(self, clear=True), [
+            ['opened-ports', ''],
+        ])
 
 
 if __name__ == "__main__":