Merge pull request #51 from canonical/open-port-support

ports api

README.md

@@ -631,6 +631,30 @@ def test_pebble_exec():
  )
 ```
 
+# Ports
+
+Since `ops 2.6.0`, charms can invoke the `open-port`, `close-port`, and `opened-ports` hook tools to manage the ports opened on the host vm/container. Using the `State.opened_ports` api, you can: 
+
+- simulate a charm run with a port opened by some previous execution
+```python
+from scenario import State, Port, Context
+
+ctx = Context(MyCharm)
+ctx.run("start", State(opened_ports=[Port("tcp", 42)]))
+```
+- assert that a charm has called `open-port` or `close-port`:
+```python
+from scenario import State, Port, Context
+
+ctx = Context(MyCharm)
+state1 = ctx.run("start", State())
+assert state1.opened_ports == [Port("tcp", 42)]
+
+state2 = ctx.run("stop", state1)
+assert state2.opened_ports == []
+```
+
+
 # Secrets
 
 Scenario has secrets. Here's how you use them.

pyproject.toml

@@ -8,7 +8,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "ops-scenario"
 
-version = "5.0"
+version = "5.1.0"
 
 authors = [
  { name = "Pietro Pasotti", email = "[email protected]" }
@@ -18,7 +18,7 @@ license.text = "Apache-2.0"
 keywords = ["juju", "test"]
 
 dependencies = [
- "ops>=2.0",
+ "ops>=2.6",
  "PyYAML>=6.0.1",
  "typer==0.7.0",
 ]

scenario/__init__.py

@@ -16,6 +16,7 @@
  Network,
  ParametrizedEvent,
  PeerRelation,
+ Port,
  Relation,
  RelationBase,
  Secret,
@@ -45,6 +46,7 @@
  "Address",
  "BindAddress",
  "Network",
+ "Port",
  "StoredState",
  "State",
  "DeferredEvent",

scenario/context.py

@@ -10,7 +10,7 @@
 
 from scenario.logger import logger as scenario_logger
 from scenario.runtime import Runtime
-from scenario.state import Action, Event, _CharmSpec
+from scenario.state import Action, Event, MetadataNotFoundError, _CharmSpec
 
 if TYPE_CHECKING:
  from ops.testing import CharmType
@@ -32,6 +32,10 @@ class InvalidActionError(InvalidEventError):
  """raised when something is wrong with the action passed to Context.run_action"""
 
 
+class ContextSetupError(RuntimeError):
+ """Raised by Context when setup fails."""
+
+
 class Context:
  """Scenario test execution context."""
 
@@ -70,7 +74,14 @@ def __init__(
 
  if not any((meta, actions, config)):
  logger.debug("Autoloading charmspec...")
- spec = _CharmSpec.autoload(charm_type)
+ try:
+ spec = _CharmSpec.autoload(charm_type)
+ except MetadataNotFoundError as e:
+ raise ContextSetupError(
+ f"Cannot setup scenario with `charm_type`={charm_type}. "
+ f"Did you forget to pass `meta` to this Context?",
+ ) from e
+
  else:
  if not meta:
  meta = {"name": str(charm_type.__name__)}

scenario/mocking.py

@@ -5,7 +5,7 @@
 import random
 from io import StringIO
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional, Set, Tuple, Union
 
 from ops import pebble
 from ops.model import (
@@ -18,7 +18,7 @@
 from ops.testing import _TestingPebbleClient
 
 from scenario.logger import logger as scenario_logger
-from scenario.state import JujuLogLine, Mount, PeerRelation
+from scenario.state import JujuLogLine, Mount, PeerRelation, Port
 
 if TYPE_CHECKING:
  from scenario.context import Context
@@ -75,6 +75,23 @@ def __init__(
  self._context = context
  self._charm_spec = charm_spec
 
+ def opened_ports(self) -> Set[Port]:
+ return set(self._state.opened_ports)
+
+ def open_port(self, protocol: str, port: Optional[int] = None):
+ # fixme: the charm will get hit with a StateValidationError
+ # here, not the expected ModelError...
+ port = Port(protocol, port)
+ ports = self._state.opened_ports
+ if port not in ports:
+ ports.append(port)
+
+ def close_port(self, protocol: str, port: Optional[int] = None):
+ port = Port(protocol, port)
+ ports = self._state.opened_ports
+ if port in ports:
+ ports.remove(port)
+
  def get_pebble(self, socket_path: str) -> "Client":
  container_name = socket_path.split("/")[
  3

scenario/state.py

@@ -79,6 +79,10 @@ class StateValidationError(RuntimeError):
  # **combination** of several parts of the State are.
 
 
+class MetadataNotFoundError(RuntimeError):
+ """Raised when Scenario can't find a metadata.yaml file in the provided charm root."""
+
+
 @dataclasses.dataclass(frozen=True)
 class _DCBase:
  def replace(self, *args, **kwargs):
@@ -755,6 +759,32 @@ def handle_path(self):
  return f"{self.owner_path or ''}/{self.data_type_name}[{self.name}]"
 
 
+@dataclasses.dataclass(frozen=True)
+class Port(_DCBase):
+ """Represents a port on the charm host."""
+
+ protocol: Literal["tcp", "udp", "icmp"]
+ port: Optional[int] = None
+ """The port to open. Required for TCP and UDP; not allowed for ICMP."""
+
+ def __post_init__(self):
+ port = self.port
+ is_icmp = self.protocol == "icmp"
+ if port:
+ if is_icmp:
+ raise StateValidationError(
+ "`port` arg not supported with `icmp` protocol",
+ )
+ if not (1 <= port <= 65535):
+ raise StateValidationError(
+ f"`port` outside bounds [1:65535], got {port}",
+ )
+ elif not is_icmp:
+ raise StateValidationError(
+ f"`port` arg required with `{self.protocol}` protocol",
+ )
+
+
 @dataclasses.dataclass(frozen=True)
 class State(_DCBase):
  """Represents the juju-owned portion of a unit's state.
@@ -770,6 +800,9 @@ class State(_DCBase):
  relations: List["AnyRelation"] = dataclasses.field(default_factory=list)
  networks: List[Network] = dataclasses.field(default_factory=list)
  containers: List[Container] = dataclasses.field(default_factory=list)
+
+ # we don't use sets to make json serialization easier
+ opened_ports: List[Port] = dataclasses.field(default_factory=list)
  leader: bool = False
  model: Model = Model()
  secrets: List[Secret] = dataclasses.field(default_factory=list)
@@ -890,6 +923,11 @@ def autoload(charm_type: Type["CharmType"]):
  charm_root = charm_source_path.parent.parent
 
  metadata_path = charm_root / "metadata.yaml"
+ if not metadata_path.exists():
+ raise MetadataNotFoundError(
+ f"invalid charm root {charm_root!r}; "
+ f"expected to contain at least a `metadata.yaml` file.",
+ )
  meta = yaml.safe_load(metadata_path.open())
 
  actions = config = None

tests/test_e2e/test_pebble.py

@@ -8,7 +8,7 @@
 from ops.pebble import ServiceStartup, ServiceStatus
 
 from scenario import Context
-from scenario.state import Container, ExecOutput, Mount, State
+from scenario.state import Container, ExecOutput, Mount, Port, State
 from tests.helpers import trigger
 
 
@@ -95,6 +95,10 @@ def callback(self: CharmBase):
  )
 
 
+def test_port_equality():
+ assert Port("tcp", 42) == Port("tcp", 42)
+
+
 @pytest.mark.parametrize("make_dirs", (True, False))
 def test_fs_pull(charm_cls, make_dirs):
  text = "lorem ipsum/n alles amat gloriae foo"

tests/test_e2e/test_ports.py

@@ -0,0 +1,34 @@
+import pytest
+from ops import CharmBase
+
+from scenario import Context, State
+from scenario.state import Port
+
+
+class MyCharm(CharmBase):
+ META = {"name": "edgar"}
+
+
+@pytest.fixture
+def ctx():
+ return Context(MyCharm, meta=MyCharm.META)
+
+
+def test_open_port(ctx):
+ def post_event(charm: CharmBase):
+ charm.unit.open_port("tcp", 12)
+
+ out = ctx.run("start", State(), post_event=post_event)
+ port = out.opened_ports.pop()
+
+ assert port.protocol == "tcp"
+ assert port.port == 12
+
+
+def test_close_port(ctx):
+ def post_event(charm: CharmBase):
+ assert charm.unit.opened_ports()
+ charm.unit.close_port("tcp", 42)
+
+ out = ctx.run("start", State(opened_ports={Port("tcp", 42)}), post_event=post_event)
+ assert not out.opened_ports