Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat!: require keyword arguments for most dataclasses #137

Merged
merged 5 commits into from
Jul 8, 2024
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 15 additions & 17 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -397,6 +397,8 @@ meta = {
}
ctx = scenario.Context(ops.CharmBase, meta=meta, unit_id=1)
ctx.run(ctx.on.start(), state_in) # invalid: this unit's id cannot be the ID of a peer.


```

### SubordinateRelation
Expand Down Expand Up @@ -531,7 +533,7 @@ local_file = pathlib.Path('/path/to/local/real/file.txt')
container = scenario.Container(
name="foo",
can_connect=True,
mounts={'local': scenario.Mount('/local/share/config.yaml', local_file)}
mounts={'local': scenario.Mount(location='/local/share/config.yaml', source=local_file)}
)
state = scenario.State(containers=[container])
```
Expand Down Expand Up @@ -568,7 +570,7 @@ def test_pebble_push():
container = scenario,Container(
name='foo',
can_connect=True,
mounts={'local': Mount('/local/share/config.yaml', local_file.name)}
mounts={'local': Mount(location='/local/share/config.yaml', source=local_file.name)}
)
state_in = State(containers=[container])
ctx = Context(
Expand Down Expand Up @@ -667,32 +669,29 @@ Pebble can generate notices, which Juju will detect, and wake up the charm to
let it know that something has happened in the container. The most common
use-case is Pebble custom notices, which is a mechanism for the workload
application to trigger a charm event.

-
When the charm is notified, there might be a queue of existing notices, or just
the one that has triggered the event:

```python
import ops
import scenario

class MyCharm(ops.CharmBase):
def __init__(self, framework):
super().__init__(framework)
framework.observe(self.on["cont"].pebble_custom_notice, self._on_notice)
framework.observe(self.on["my-container"].pebble_custom_notice, self._on_notice)

def _on_notice(self, event):
event.notice.key # == "example.com/c"
for notice in self.unit.get_container("cont").get_notices():
for notice in self.unit.get_container("my-container").get_notices():
...

ctx = scenario.Context(MyCharm, meta={"name": "foo", "containers": {"my-container": {}}})
notices = [
scenario.Notice(key="example.com/a", occurences=10),
scenario.Notice(key="example.com/a", occurrences=10),
scenario.Notice(key="example.com/b", last_data={"bar": "baz"}),
scenario.Notice(key="example.com/c"),
]
cont = scenario.Container(notices=notices)
ctx.run(container.get_notice("example.com/c").event, scenario.State(containers=[cont]))
container = scenario.Container("my-container", notices=notices)
ctx.run(container.get_notice("example.com/c").event, scenario.State(containers=[container]))
```

## Storage
Expand Down Expand Up @@ -766,15 +765,14 @@ ctx.run(ctx.on.storage_attached(foo_1), scenario.State(storage=[foo_0, foo_1]))
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
ctx = scenario.Context(MyCharm, meta=MyCharm.META)
ctx.run(ctx.on.start(), scenario.State(opened_ports=[scenario.Port("tcp", 42)]))
ctx.run(ctx.on.start(), scenario.State(opened_ports=[scenario.TCPPort(port=42)]))
tonyandrewmeyer marked this conversation as resolved.
Show resolved Hide resolved
```
- assert that a charm has called `open-port` or `close-port`:
```python
ctx = scenario.Context(PortCharm, meta=MyCharm.META)
state1 = ctx.run(ctx.on.start(), scenario.State())
assert state1.opened_ports == [scenario.Port("tcp", 42)]
assert state1.opened_ports == [scenario.TCPPort(port=42)]

state2 = ctx.run(ctx.on.stop(), state1)
assert state2.opened_ports == []
tonyandrewmeyer marked this conversation as resolved.
Show resolved Hide resolved
Expand All @@ -788,8 +786,8 @@ Scenario has secrets. Here's how you use them.
state = scenario.State(
secrets=[
scenario.Secret(
{0: {'key': 'public'}},
id='foo',
contents={0: {'key': 'public'}}
)
]
)
Expand Down Expand Up @@ -817,8 +815,8 @@ To specify a secret owned by this unit (or app):
state = scenario.State(
secrets=[
scenario.Secret(
{0: {'key': 'private'}},
id='foo',
contents={0: {'key': 'private'}},
owner='unit', # or 'app'
remote_grants={0: {"remote"}}
# the secret owner has granted access to the "remote" app over some relation with ID 0
Expand All @@ -833,8 +831,8 @@ To specify a secret owned by some other application and give this unit (or app)
state = scenario.State(
secrets=[
scenario.Secret(
{0: {'key': 'public'}},
id='foo',
contents={0: {'key': 'public'}},
# owner=None, which is the default
revision=0, # the revision that this unit (or app) is currently tracking
)
Expand Down
8 changes: 6 additions & 2 deletions scenario/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,19 +11,21 @@
Container,
DeferredEvent,
ExecOutput,
ICMPPort,
Model,
Mount,
Network,
Notice,
PeerRelation,
Port,
Relation,
Secret,
State,
StateValidationError,
Storage,
StoredState,
SubordinateRelation,
TCPPort,
UDPPort,
deferred,
)

Expand All @@ -47,7 +49,9 @@
"Address",
"BindAddress",
"Network",
"Port",
"ICMPPort",
"TCPPort",
"UDPPort",
"Storage",
"StoredState",
"State",
Expand Down
22 changes: 13 additions & 9 deletions scenario/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import tempfile
from contextlib import contextmanager
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type, Union, cast
from typing import TYPE_CHECKING, Any, Dict, Final, List, Optional, Type, Union, cast

from ops import CharmBase, EventBase

Expand All @@ -19,6 +19,7 @@
Storage,
_CharmSpec,
_Event,
_MaxPositionalArgs,
)

if TYPE_CHECKING: # pragma: no cover
Expand All @@ -34,21 +35,23 @@
DEFAULT_JUJU_VERSION = "3.4"


@dataclasses.dataclass
class ActionOutput:
@dataclasses.dataclass(frozen=True)
class ActionOutput(_MaxPositionalArgs):
"""Wraps the results of running an action event with `run_action`."""

state: "State"
"""The charm state after the action has been handled.
In most cases, actions are not expected to be affecting it."""
logs: List[str]
"""Any logs associated with the action output, set by the charm."""
results: Optional[Dict[str, Any]]
results: Optional[Dict[str, Any]] = None
"""Key-value mapping assigned by the charm as a result of the action.
Will be None if the charm never calls action-set."""
failure: Optional[str] = None
"""If the action is not a success: the message the charm set when failing the action."""

_max_positional_args: Final = 0

@property
def success(self) -> bool:
"""Return whether this action was a success."""
Expand Down Expand Up @@ -316,6 +319,7 @@ def __init__(
self,
charm_type: Type["CharmType"],
meta: Optional[Dict[str, Any]] = None,
*,
actions: Optional[Dict[str, Any]] = None,
config: Optional[Dict[str, Any]] = None,
charm_root: Optional["PathLike"] = None,
Expand Down Expand Up @@ -382,7 +386,7 @@ def __init__(
defined in metadata.yaml.
:arg unit_id: Unit ID that this charm is deployed as. Defaults to 0.
:arg app_trusted: whether the charm has Juju trust (deployed with ``--trust`` or added with
``juju trust``). Defaults to False
``juju trust``). Defaults to False.
:arg charm_root: virtual charm root the charm will be executed with.
If the charm, say, expects a `./src/foo/bar.yaml` file present relative to the
execution cwd, you need to use this. E.g.:
Expand Down Expand Up @@ -553,10 +557,10 @@ def run_action(self, action: "Action", state: "State") -> ActionOutput:

def _finalize_action(self, state_out: "State"):
ao = ActionOutput(
state_out,
self._action_logs,
self._action_results,
self._action_failure,
state=state_out,
logs=self._action_logs,
results=self._action_results,
failure=self._action_failure,
)

# reset all action-related state
Expand Down
24 changes: 15 additions & 9 deletions scenario/mocking.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,11 @@
cast,
)

from ops import CloudSpec, JujuVersion, pebble
from ops.model import ModelError, RelationNotFoundError
from ops import JujuVersion, pebble
from ops.model import CloudSpec as CloudSpec_Ops
from ops.model import ModelError
from ops.model import Port as Port_Ops
from ops.model import RelationNotFoundError
from ops.model import Secret as Secret_Ops # lol
from ops.model import (
SecretInfo,
Expand All @@ -39,8 +42,8 @@
Mount,
Network,
PeerRelation,
Port,
Storage,
_port_cls_by_protocol,
_RawPortProtocolLiteral,
_RawStatusLiteral,
)
Expand Down Expand Up @@ -112,8 +115,11 @@ def __init__(
self._context = context
self._charm_spec = charm_spec

def opened_ports(self) -> Set[Port]:
return set(self._state.opened_ports)
def opened_ports(self) -> Set[Port_Ops]:
return {
Port_Ops(protocol=port.protocol, port=port.port)
for port in self._state.opened_ports
}

def open_port(
self,
Expand All @@ -122,7 +128,7 @@ def open_port(
):
# fixme: the charm will get hit with a StateValidationError
# here, not the expected ModelError...
port_ = Port(protocol, port)
port_ = _port_cls_by_protocol[protocol](port=port)
ports = self._state.opened_ports
if port_ not in ports:
ports.append(port_)
Expand All @@ -132,7 +138,7 @@ def close_port(
protocol: "_RawPortProtocolLiteral",
port: Optional[int] = None,
):
_port = Port(protocol, port)
_port = _port_cls_by_protocol[protocol](port=port)
ports = self._state.opened_ports
if _port in ports:
ports.remove(_port)
Expand Down Expand Up @@ -629,7 +635,7 @@ def resource_get(self, resource_name: str) -> str:
f"resource {resource_name} not found in State. please pass it.",
)

def credential_get(self) -> CloudSpec:
def credential_get(self) -> CloudSpec_Ops:
if not self._context.app_trusted:
raise ModelError(
"ERROR charm is not trusted, initialise Context with `app_trusted=True`",
Expand Down Expand Up @@ -669,7 +675,7 @@ def __init__(
path = Path(mount.location).parts
mounting_dir = container_root.joinpath(*path[1:])
mounting_dir.parent.mkdir(parents=True, exist_ok=True)
mounting_dir.symlink_to(mount.src)
mounting_dir.symlink_to(mount.source)

self._root = container_root

Expand Down
Loading
Loading