diff --git a/README.md b/README.md index 92a13f84..f7a794e3 100644 --- a/README.md +++ b/README.md @@ -397,8 +397,6 @@ 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 @@ -533,7 +531,7 @@ local_file = pathlib.Path('/path/to/local/real/file.txt') container = scenario.Container( name="foo", can_connect=True, - mounts={'local': scenario.Mount(location='/local/share/config.yaml', source=local_file)} + mounts={'local': scenario.Mount('/local/share/config.yaml', local_file)} ) state = scenario.State(containers=[container]) ``` @@ -570,7 +568,7 @@ def test_pebble_push(): container = scenario,Container( name='foo', can_connect=True, - mounts={'local': Mount(location='/local/share/config.yaml', source=local_file.name)} + mounts={'local': Mount('/local/share/config.yaml', local_file.name)} ) state_in = State(containers=[container]) ctx = Context( @@ -669,29 +667,32 @@ 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["my-container"].pebble_custom_notice, self._on_notice) + framework.observe(self.on["cont"].pebble_custom_notice, self._on_notice) def _on_notice(self, event): event.notice.key # == "example.com/c" - for notice in self.unit.get_container("my-container").get_notices(): + for notice in self.unit.get_container("cont").get_notices(): ... ctx = scenario.Context(MyCharm, meta={"name": "foo", "containers": {"my-container": {}}}) notices = [ - scenario.Notice(key="example.com/a", occurrences=10), + scenario.Notice(key="example.com/a", occurences=10), scenario.Notice(key="example.com/b", last_data={"bar": "baz"}), scenario.Notice(key="example.com/c"), ] -container = scenario.Container("my-container", notices=notices) -ctx.run(container.get_notice("example.com/c").event, scenario.State(containers=[container])) +cont = scenario.Container(notices=notices) +ctx.run(container.get_notice("example.com/c").event, scenario.State(containers=[cont])) ``` ## Storage @@ -765,14 +766,15 @@ 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.TCPPort(42)])) +ctx.run(ctx.on.start(), scenario.State(opened_ports=[scenario.Port("tcp", 42)])) ``` - 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.TCPPort(42)] +assert state1.opened_ports == [scenario.Port("tcp", 42)] state2 = ctx.run(ctx.on.stop(), state1) assert state2.opened_ports == [] @@ -786,8 +788,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'}} ) ] ) @@ -815,8 +817,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 @@ -831,8 +833,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 ) diff --git a/scenario/__init__.py b/scenario/__init__.py index a73570a6..93059ebf 100644 --- a/scenario/__init__.py +++ b/scenario/__init__.py @@ -11,12 +11,12 @@ Container, DeferredEvent, ExecOutput, - ICMPPort, Model, Mount, Network, Notice, PeerRelation, + Port, Relation, Secret, State, @@ -24,8 +24,6 @@ Storage, StoredState, SubordinateRelation, - TCPPort, - UDPPort, deferred, ) @@ -49,9 +47,7 @@ "Address", "BindAddress", "Network", - "ICMPPort", - "TCPPort", - "UDPPort", + "Port", "Storage", "StoredState", "State", diff --git a/scenario/context.py b/scenario/context.py index caac79c4..6d60541b 100644 --- a/scenario/context.py +++ b/scenario/context.py @@ -19,7 +19,6 @@ Storage, _CharmSpec, _Event, - _max_posargs, ) if TYPE_CHECKING: # pragma: no cover @@ -35,8 +34,8 @@ DEFAULT_JUJU_VERSION = "3.4" -@dataclasses.dataclass(frozen=True) -class ActionOutput(_max_posargs(0)): +@dataclasses.dataclass +class ActionOutput: """Wraps the results of running an action event with `run_action`.""" state: "State" @@ -44,7 +43,7 @@ class ActionOutput(_max_posargs(0)): 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]] = None + results: Optional[Dict[str, Any]] """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 @@ -317,7 +316,6 @@ 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, @@ -384,7 +382,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.: @@ -555,10 +553,10 @@ def run_action(self, action: "Action", state: "State") -> ActionOutput: def _finalize_action(self, state_out: "State"): ao = ActionOutput( - state=state_out, - logs=self._action_logs, - results=self._action_results, - failure=self._action_failure, + state_out, + self._action_logs, + self._action_results, + self._action_failure, ) # reset all action-related state diff --git a/scenario/mocking.py b/scenario/mocking.py index b17627d3..8885e420 100644 --- a/scenario/mocking.py +++ b/scenario/mocking.py @@ -20,11 +20,8 @@ cast, ) -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 import CloudSpec, JujuVersion, pebble +from ops.model import ModelError, RelationNotFoundError from ops.model import Secret as Secret_Ops # lol from ops.model import ( SecretInfo, @@ -42,8 +39,8 @@ Mount, Network, PeerRelation, + Port, Storage, - _port_cls_by_protocol, _RawPortProtocolLiteral, _RawStatusLiteral, ) @@ -115,11 +112,8 @@ def __init__( self._context = context self._charm_spec = charm_spec - def opened_ports(self) -> Set[Port_Ops]: - return { - Port_Ops(protocol=port.protocol, port=port.port) - for port in self._state.opened_ports - } + def opened_ports(self) -> Set[Port]: + return set(self._state.opened_ports) def open_port( self, @@ -128,7 +122,7 @@ def open_port( ): # fixme: the charm will get hit with a StateValidationError # here, not the expected ModelError... - port_ = _port_cls_by_protocol[protocol](port=port) + port_ = Port(protocol, port) ports = self._state.opened_ports if port_ not in ports: ports.append(port_) @@ -138,7 +132,7 @@ def close_port( protocol: "_RawPortProtocolLiteral", port: Optional[int] = None, ): - _port = _port_cls_by_protocol[protocol](port=port) + _port = Port(protocol, port) ports = self._state.opened_ports if _port in ports: ports.remove(_port) @@ -635,7 +629,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_Ops: + def credential_get(self) -> CloudSpec: if not self._context.app_trusted: raise ModelError( "ERROR charm is not trusted, initialise Context with `app_trusted=True`", @@ -675,7 +669,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.source) + mounting_dir.symlink_to(mount.src) self._root = container_root diff --git a/scenario/state.py b/scenario/state.py index d89838dd..b2c6fe76 100644 --- a/scenario/state.py +++ b/scenario/state.py @@ -14,7 +14,6 @@ Any, Callable, Dict, - Final, Generic, List, Literal, @@ -28,11 +27,10 @@ ) from uuid import uuid4 +import ops import yaml from ops import pebble from ops.charm import CharmBase, CharmEvents -from ops.model import CloudCredential as CloudCredential_Ops -from ops.model import CloudSpec as CloudSpec_Ops from ops.model import SecretRotate, StatusBase from scenario.logger import logger as scenario_logger @@ -122,77 +120,8 @@ class MetadataNotFoundError(RuntimeError): """Raised when Scenario can't find a metadata.yaml file in the provided charm root.""" -# This can be replaced with the KW_ONLY dataclasses functionality in Python 3.10+. -def _max_posargs(n: int): - class _MaxPositionalArgs: - """Raises TypeError when instantiating objects if arguments are not passed as keywords. - - Looks for a `_max_positional_args` class attribute, which should be an int - indicating the maximum number of positional arguments that can be passed to - `__init__` (excluding `self`). - """ - - _max_positional_args = n - - def __new__(cls, *args, **kwargs): - # inspect.signature guarantees the order of parameters is as - # declared, which aligns with dataclasses. Simpler ways of - # getting the arguments (like __annotations__) do not have that - # guarantee, although in practice it is the case. - parameters = inspect.signature(cls).parameters - required_args = [ - name - for name in tuple(parameters) - if parameters[name].default is inspect.Parameter.empty - and name not in kwargs - ] - n_posargs = len(args) - max_n_posargs = cls._max_positional_args - kw_only = { - name - for name in tuple(parameters)[max_n_posargs:] - if not name.startswith("_") - } - if n_posargs > max_n_posargs: - raise TypeError( - f"{cls.__name__} takes {max_n_posargs} positional " - f"argument{'' if max_n_posargs == 1 else 's'} but " - f"{n_posargs} {'was' if n_posargs == 1 else 'were'} " - f"given. The following arguments are keyword-only: " - f"{', '.join(kw_only)}", - ) from None - # Also check if there are just not enough arguments at all, because - # the default TypeError message will incorrectly describe some of - # the arguments as positional. - elif n_posargs < len(required_args): - required_pos = [ - f"'{arg}'" - for arg in required_args[n_posargs:] - if arg not in kw_only - ] - required_kw = { - f"'{arg}'" for arg in required_args[n_posargs:] if arg in kw_only - } - if required_pos and required_kw: - details = f"positional: {', '.join(required_pos)} and keyword: {', '.join(required_kw)} arguments" - elif required_pos: - details = f"positional argument{'' if len(required_pos) == 1 else 's'}: {', '.join(required_pos)}" - else: - details = f"keyword argument{'' if len(required_kw) == 1 else 's'}: {', '.join(required_kw)}" - raise TypeError(f"{cls.__name__} missing required {details}") from None - return super().__new__(cls) - - def __reduce__(self): - # The default __reduce__ doesn't understand that some arguments have - # to be passed as keywords, so using the copy module fails. - attrs = cast(Dict[str, Any], super().__reduce__()[2]) - return (lambda: self.__class__(**attrs), ()) - - return _MaxPositionalArgs - - @dataclasses.dataclass(frozen=True) -class CloudCredential(_max_posargs(0)): +class CloudCredential: auth_type: str """Authentication type.""" @@ -206,8 +135,8 @@ class CloudCredential(_max_posargs(0)): redacted: List[str] = dataclasses.field(default_factory=list) """A list of redacted generic cloud API secrets.""" - def _to_ops(self) -> CloudCredential_Ops: - return CloudCredential_Ops( + def _to_ops(self) -> ops.CloudCredential: + return ops.CloudCredential( auth_type=self.auth_type, attributes=self.attributes, redacted=self.redacted, @@ -215,7 +144,7 @@ def _to_ops(self) -> CloudCredential_Ops: @dataclasses.dataclass(frozen=True) -class CloudSpec(_max_posargs(1)): +class CloudSpec: type: str """Type of the cloud.""" @@ -246,8 +175,8 @@ class CloudSpec(_max_posargs(1)): is_controller_cloud: bool = False """If this is the cloud used by the controller.""" - def _to_ops(self) -> CloudSpec_Ops: - return CloudSpec_Ops( + def _to_ops(self) -> ops.CloudSpec: + return ops.CloudSpec( type=self.type, name=self.name, region=self.region, @@ -262,16 +191,16 @@ def _to_ops(self) -> CloudSpec_Ops: @dataclasses.dataclass(frozen=True) -class Secret(_max_posargs(1)): - # mapping from revision IDs to each revision's contents - contents: Dict[int, "RawSecretRevisionContents"] - +class Secret: id: str # CAUTION: ops-created Secrets (via .add_secret()) will have a canonicalized # secret id (`secret:` prefix) # but user-created ones will not. Using post-init to patch it in feels bad, but requiring the user to # add the prefix manually every time seems painful as well. + # mapping from revision IDs to each revision's contents + contents: Dict[int, "RawSecretRevisionContents"] + # indicates if the secret is owned by THIS unit, THIS app or some other app/unit. # if None, the implication is that the secret has been granted to this unit. owner: Literal["unit", "app", None] = None @@ -325,17 +254,17 @@ def normalize_name(s: str): @dataclasses.dataclass(frozen=True) -class Address(_max_posargs(1)): +class Address: + hostname: str value: str - hostname: str = "" - cidr: str = "" + cidr: str address: str = "" # legacy @dataclasses.dataclass(frozen=True) -class BindAddress(_max_posargs(1)): +class BindAddress: + interface_name: str addresses: List[Address] - interface_name: str = "" mac_address: Optional[str] = None def hook_tool_output_fmt(self): @@ -351,7 +280,7 @@ def hook_tool_output_fmt(self): @dataclasses.dataclass(frozen=True) -class Network(_max_posargs(0)): +class Network: bind_addresses: List[BindAddress] ingress_addresses: List[str] egress_subnets: List[str] @@ -394,7 +323,7 @@ def default( _next_relation_id_counter = 1 -def next_relation_id(*, update=True): +def next_relation_id(update=True): global _next_relation_id_counter cur = _next_relation_id_counter if update: @@ -403,7 +332,7 @@ def next_relation_id(*, update=True): @dataclasses.dataclass(frozen=True) -class _RelationBase(_max_posargs(2)): +class _RelationBase: endpoint: str """Relation endpoint name. Must match some endpoint name defined in metadata.yaml.""" @@ -579,9 +508,8 @@ def _random_model_name(): @dataclasses.dataclass(frozen=True) -class Model(_max_posargs(1)): +class Model: name: str = dataclasses.field(default_factory=_random_model_name) - uuid: str = dataclasses.field(default_factory=lambda: str(uuid4())) # whatever juju models --format=json | jq '.models[].type' gives back. @@ -610,7 +538,7 @@ def _generate_new_change_id(): @dataclasses.dataclass(frozen=True) -class ExecOutput(_max_posargs(0)): +class ExecOutput: return_code: int = 0 stdout: str = "" stderr: str = "" @@ -626,9 +554,9 @@ def _run(self) -> int: @dataclasses.dataclass(frozen=True) -class Mount(_max_posargs(0)): +class Mount: location: Union[str, PurePosixPath] - source: Union[str, Path] + src: Union[str, Path] def _now_utc(): @@ -638,7 +566,7 @@ def _now_utc(): _next_notice_id_counter = 1 -def next_notice_id(*, update=True): +def next_notice_id(update=True): global _next_notice_id_counter cur = _next_notice_id_counter if update: @@ -647,7 +575,7 @@ def next_notice_id(*, update=True): @dataclasses.dataclass(frozen=True) -class Notice(_max_posargs(1)): +class Notice: key: str """The notice key, a string that differentiates notices of this type. @@ -706,7 +634,7 @@ def _to_ops(self) -> pebble.Notice: @dataclasses.dataclass(frozen=True) -class _BoundNotice(_max_posargs(0)): +class _BoundNotice: notice: Notice container: "Container" @@ -722,9 +650,8 @@ def event(self): @dataclasses.dataclass(frozen=True) -class Container(_max_posargs(1)): +class Container: name: str - can_connect: bool = False # This is the base plan. On top of it, one can add layers. @@ -749,8 +676,8 @@ class Container(_max_posargs(1)): # # this becomes: # mounts = { - # 'foo': Mount(location='/home/foo/', source=Path('/path/to/local/dir/containing/bar/py/')) - # 'bin': Mount(location='/bin/', source=Path('/path/to/local/dir/containing/bash/and/baz/')) + # 'foo': Mount('/home/foo/', Path('/path/to/local/dir/containing/bar/py/')) + # 'bin': Mount('/bin/', Path('/path/to/local/dir/containing/bash/and/baz/')) # } # when the charm runs `pebble.pull`, it will return .open() from one of those paths. # when the charm pushes, it will either overwrite one of those paths (careful!) or it will @@ -830,7 +757,7 @@ def get_notice( """ for notice in self.notices: if notice.key == key and notice.type == notice_type: - return _BoundNotice(notice=notice, container=self) + return _BoundNotice(notice, self) raise KeyError( f"{self.name} does not have a notice with key {key} and type {notice_type}", ) @@ -884,13 +811,12 @@ class _MyClass(_EntityStatus, statusbase_subclass): @dataclasses.dataclass(frozen=True) -class StoredState(_max_posargs(1)): - name: str = "_stored" - +class StoredState: # /-separated Object names. E.g. MyCharm/MyCharmLib. # if None, this StoredState instance is owned by the Framework. - owner_path: Optional[str] = None + owner_path: Optional[str] + name: str = "_stored" # Ideally, the type here would be only marshallable types, rather than Any. # However, it's complex to describe those types, since it's a recursive # definition - even in TypeShed the _Marshallable type includes containers @@ -908,78 +834,35 @@ def handle_path(self): @dataclasses.dataclass(frozen=True) -class _Port(_max_posargs(1)): +class Port: """Represents a port on the charm host.""" + protocol: _RawPortProtocolLiteral port: Optional[int] = None """The port to open. Required for TCP and UDP; not allowed for ICMP.""" - protocol: _RawPortProtocolLiteral = "tcp" - - def __post_init__(self): - if type(self) is _Port: - raise RuntimeError( - "_Port cannot be instantiated directly; " - "please use TCPPort, UDPPort, or ICMPPort", - ) - - -@dataclasses.dataclass(frozen=True) -class TCPPort(_Port): - """Represents a TCP port on the charm host.""" - - port: int - """The port to open.""" - protocol: _RawPortProtocolLiteral = "tcp" def __post_init__(self): - super().__post_init__() - if not (1 <= self.port <= 65535): - raise StateValidationError( - f"`port` outside bounds [1:65535], got {self.port}", - ) - - -@dataclasses.dataclass(frozen=True) -class UDPPort(_Port): - """Represents a UDP port on the charm host.""" - - port: int - """The port to open.""" - protocol: _RawPortProtocolLiteral = "udp" - - def __post_init__(self): - super().__post_init__() - if not (1 <= self.port <= 65535): + 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` outside bounds [1:65535], got {self.port}", + f"`port` arg required with `{self.protocol}` protocol", ) -@dataclasses.dataclass(frozen=True) -class ICMPPort(_Port): - """Represents an ICMP port on the charm host.""" - - protocol: _RawPortProtocolLiteral = "icmp" - - _max_positional_args: Final = 0 - - def __post_init__(self): - super().__post_init__() - if self.port is not None: - raise StateValidationError("`port` cannot be set for `ICMPPort`") - - -_port_cls_by_protocol = { - "tcp": TCPPort, - "udp": UDPPort, - "icmp": ICMPPort, -} - - _next_storage_index_counter = 0 # storage indices start at 0 -def next_storage_index(*, update=True): +def next_storage_index(update=True): """Get the index (used to be called ID) the next Storage to be created will get. Pass update=False if you're only inspecting it. @@ -993,7 +876,7 @@ def next_storage_index(*, update=True): @dataclasses.dataclass(frozen=True) -class Storage(_max_posargs(1)): +class Storage: """Represents an (attached!) storage made available to the charm container.""" name: str @@ -1007,7 +890,7 @@ def get_filesystem(self, ctx: "Context") -> Path: @dataclasses.dataclass(frozen=True) -class State(_max_posargs(0)): +class State: """Represents the juju-owned portion of a unit's state. Roughly speaking, it wraps all hook-tool- and pebble-mediated data a charm can access in its @@ -1036,7 +919,7 @@ class State(_max_posargs(0)): If a storage is not attached, omit it from this listing.""" # we don't use sets to make json serialization easier - opened_ports: List[_Port] = dataclasses.field(default_factory=list) + opened_ports: List[Port] = dataclasses.field(default_factory=list) """Ports opened by juju on this charm.""" leader: bool = False """Whether this charm has leadership.""" @@ -1501,7 +1384,7 @@ def deferred(self, handler: Callable, event_id: int = 1) -> DeferredEvent: _next_action_id_counter = 1 -def next_action_id(*, update=True): +def next_action_id(update=True): global _next_action_id_counter cur = _next_action_id_counter if update: @@ -1512,7 +1395,7 @@ def next_action_id(*, update=True): @dataclasses.dataclass(frozen=True) -class Action(_max_posargs(1)): +class Action: name: str params: Dict[str, "AnyJson"] = dataclasses.field(default_factory=dict) diff --git a/tests/test_consistency_checker.py b/tests/test_consistency_checker.py index 82321558..6a955be7 100644 --- a/tests/test_consistency_checker.py +++ b/tests/test_consistency_checker.py @@ -3,6 +3,7 @@ import pytest from ops.charm import CharmBase +from scenario import Model from scenario.consistency_checker import check_consistency from scenario.runtime import InconsistentScenarioError from scenario.state import ( @@ -11,7 +12,6 @@ CloudCredential, CloudSpec, Container, - Model, Network, Notice, PeerRelation, @@ -285,7 +285,7 @@ def test_secrets_jujuv_bad(bad_v): @pytest.mark.parametrize("good_v", ("3.0", "3.1", "3", "3.33", "4", "100")) def test_secrets_jujuv_bad(good_v): assert_consistent( - State(secrets=[Secret(id="secret:foo", contents={0: {"a": "b"}})]), + State(secrets=[Secret("secret:foo", {0: {"a": "b"}})]), _Event("bar"), _CharmSpec(MyCharm, {}), good_v, @@ -293,7 +293,7 @@ def test_secrets_jujuv_bad(good_v): def test_secret_not_in_state(): - secret = Secret(id="secret:foo", contents={"a": "b"}) + secret = Secret("secret:foo", {"a": "b"}) assert_inconsistent( State(), _Event("secret_changed", secret=secret), @@ -673,10 +673,10 @@ def test_storedstate_consistency(): assert_consistent( State( stored_state=[ - StoredState(content={"foo": "bar"}), - StoredState(name="my_stored_state", content={"foo": 1}), - StoredState(owner_path="MyCharmLib", content={"foo": None}), - StoredState(owner_path="OtherCharmLib", content={"foo": (1, 2, 3)}), + StoredState(None, content={"foo": "bar"}), + StoredState(None, "my_stored_state", content={"foo": 1}), + StoredState("MyCharmLib", content={"foo": None}), + StoredState("OtherCharmLib", content={"foo": (1, 2, 3)}), ] ), _Event("start"), @@ -690,8 +690,8 @@ def test_storedstate_consistency(): assert_inconsistent( State( stored_state=[ - StoredState(owner_path=None, content={"foo": "bar"}), - StoredState(owner_path=None, name="_stored", content={"foo": "bar"}), + StoredState(None, content={"foo": "bar"}), + StoredState(None, "_stored", content={"foo": "bar"}), ] ), _Event("start"), @@ -703,13 +703,7 @@ def test_storedstate_consistency(): ), ) assert_inconsistent( - State( - stored_state=[ - StoredState( - owner_path=None, content={"secret": Secret(id="foo", contents={})} - ) - ] - ), + State(stored_state=[StoredState(None, content={"secret": Secret("foo", {})})]), _Event("start"), _CharmSpec( MyCharm, diff --git a/tests/test_context.py b/tests/test_context.py index aed14159..d6995efc 100644 --- a/tests/test_context.py +++ b/tests/test_context.py @@ -3,7 +3,7 @@ import pytest from ops import CharmBase -from scenario import Action, ActionOutput, Context, State +from scenario import Action, Context, State from scenario.state import _Event, next_action_id @@ -59,23 +59,3 @@ def test_app_name(app_name, unit_id): with ctx.manager(ctx.on.start(), State()) as mgr: assert mgr.charm.app.name == app_name assert mgr.charm.unit.name == f"{app_name}/{unit_id}" - - -def test_action_output_no_positional_arguments(): - with pytest.raises(TypeError): - ActionOutput(None, None) - - -def test_action_output_no_results(): - class MyCharm(CharmBase): - def __init__(self, framework): - super().__init__(framework) - framework.observe(self.on.act_action, self._on_act_action) - - def _on_act_action(self, _): - pass - - ctx = Context(MyCharm, meta={"name": "foo"}, actions={"act": {}}) - out = ctx.run_action(Action("act"), State()) - assert out.results is None - assert out.failure is None diff --git a/tests/test_context_on.py b/tests/test_context_on.py index d9609d2e..be8c70b5 100644 --- a/tests/test_context_on.py +++ b/tests/test_context_on.py @@ -81,9 +81,7 @@ def test_simple_events(event_name, event_kind): ) def test_simple_secret_events(as_kwarg, event_name, event_kind, owner): ctx = scenario.Context(ContextCharm, meta=META, actions=ACTIONS) - secret = scenario.Secret( - id="secret:123", contents={0: {"password": "xxxx"}}, owner=owner - ) + secret = scenario.Secret("secret:123", {0: {"password": "xxxx"}}, owner=owner) state_in = scenario.State(secrets=[secret]) # These look like: # ctx.run(ctx.on.secret_changed(secret=secret), state) @@ -114,8 +112,8 @@ def test_simple_secret_events(as_kwarg, event_name, event_kind, owner): def test_revision_secret_events(event_name, event_kind): ctx = scenario.Context(ContextCharm, meta=META, actions=ACTIONS) secret = scenario.Secret( - id="secret:123", - contents={42: {"password": "yyyy"}, 43: {"password": "xxxx"}}, + "secret:123", + {42: {"password": "yyyy"}, 43: {"password": "xxxx"}}, owner="app", ) state_in = scenario.State(secrets=[secret]) @@ -137,9 +135,7 @@ def test_revision_secret_events(event_name, event_kind): def test_revision_secret_events_as_positional_arg(event_name): ctx = scenario.Context(ContextCharm, meta=META, actions=ACTIONS) secret = scenario.Secret( - id="secret:123", - contents={42: {"password": "yyyy"}, 43: {"password": "xxxx"}}, - owner=None, + "secret:123", {42: {"password": "yyyy"}, 43: {"password": "xxxx"}}, owner=None ) state_in = scenario.State(secrets=[secret]) with pytest.raises(TypeError): @@ -184,7 +180,7 @@ def test_action_event_no_params(): def test_action_event_with_params(): ctx = scenario.Context(ContextCharm, meta=META, actions=ACTIONS) - action = scenario.Action("act", params={"param": "hello"}) + action = scenario.Action("act", {"param": "hello"}) # These look like: # ctx.run_action(ctx.on.action(action=action), state) # So that any parameters can be included and the ID can be customised. diff --git a/tests/test_e2e/test_actions.py b/tests/test_e2e/test_actions.py index 39a057e6..6256885c 100644 --- a/tests/test_e2e/test_actions.py +++ b/tests/test_e2e/test_actions.py @@ -5,7 +5,7 @@ from scenario import Context from scenario.context import InvalidEventError -from scenario.state import Action, State, _Event, next_action_id +from scenario.state import Action, State, _Event @pytest.fixture(scope="function") @@ -154,17 +154,3 @@ def handle_evt(charm: CharmBase, evt: ActionEvent): action = Action("foo", id=uuid) ctx = Context(mycharm, meta={"name": "foo"}, actions={"foo": {}}) ctx.run_action(action, State()) - - -def test_positional_arguments(): - with pytest.raises(TypeError): - Action("foo", {}) - - -def test_default_arguments(): - expected_id = next_action_id(update=False) - name = "foo" - action = Action(name) - assert action.name == name - assert action.params == {} - assert action.id == expected_id diff --git a/tests/test_e2e/test_pebble.py b/tests/test_e2e/test_pebble.py index 7dfbba67..a9223120 100644 --- a/tests/test_e2e/test_pebble.py +++ b/tests/test_e2e/test_pebble.py @@ -10,7 +10,7 @@ from ops.pebble import ExecError, ServiceStartup, ServiceStatus from scenario import Context -from scenario.state import Container, ExecOutput, Mount, Notice, State +from scenario.state import Container, ExecOutput, Mount, Notice, Port, State from tests.helpers import jsonpatch_delta, trigger @@ -86,7 +86,7 @@ def callback(self: CharmBase): Container( name="foo", can_connect=True, - mounts={"bar": Mount(location="/bar/baz.txt", source=pth)}, + mounts={"bar": Mount("/bar/baz.txt", pth)}, ) ] ), @@ -97,6 +97,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" @@ -118,9 +122,7 @@ def callback(self: CharmBase): td = tempfile.TemporaryDirectory() container = Container( - name="foo", - can_connect=True, - mounts={"foo": Mount(location="/foo", source=td.name)}, + name="foo", can_connect=True, mounts={"foo": Mount("/foo", td.name)} ) state = State(containers=[container]) @@ -133,15 +135,14 @@ def callback(self: CharmBase): callback(mgr.charm) if make_dirs: - # file = (out.get_container("foo").mounts["foo"].source + "bar/baz.txt").open("/foo/bar/baz.txt") + # file = (out.get_container("foo").mounts["foo"].src + "bar/baz.txt").open("/foo/bar/baz.txt") # this is one way to retrieve the file file = Path(td.name + "/bar/baz.txt") # another is: assert ( - file - == Path(out.get_container("foo").mounts["foo"].source) / "bar" / "baz.txt" + file == Path(out.get_container("foo").mounts["foo"].src) / "bar" / "baz.txt" ) # but that is actually a symlink to the context's root tmp folder: diff --git a/tests/test_e2e/test_ports.py b/tests/test_e2e/test_ports.py index 3a19148f..13502971 100644 --- a/tests/test_e2e/test_ports.py +++ b/tests/test_e2e/test_ports.py @@ -2,7 +2,7 @@ from ops import CharmBase, Framework, StartEvent, StopEvent from scenario import Context, State -from scenario.state import StateValidationError, TCPPort, UDPPort, _Port +from scenario.state import Port class MyCharm(CharmBase): @@ -35,18 +35,5 @@ def test_open_port(ctx): def test_close_port(ctx): - out = ctx.run(ctx.on.stop(), State(opened_ports=[TCPPort(42)])) + out = ctx.run(ctx.on.stop(), State(opened_ports=[Port("tcp", 42)])) assert not out.opened_ports - - -def test_port_no_arguments(): - with pytest.raises(RuntimeError): - _Port() - - -@pytest.mark.parametrize("klass", (TCPPort, UDPPort)) -def test_port_port(klass): - with pytest.raises(StateValidationError): - klass(port=0) - with pytest.raises(StateValidationError): - klass(port=65536) diff --git a/tests/test_e2e/test_relations.py b/tests/test_e2e/test_relations.py index 853c7ba5..e72f754c 100644 --- a/tests/test_e2e/test_relations.py +++ b/tests/test_e2e/test_relations.py @@ -21,7 +21,6 @@ StateValidationError, SubordinateRelation, _RelationBase, - next_relation_id, ) from tests.helpers import trigger @@ -422,54 +421,3 @@ def test_broken_relation_not_in_model_relations(mycharm): assert charm.model.get_relation("foo") is None assert charm.model.relations["foo"] == [] - - -@pytest.mark.parametrize("klass", (Relation, PeerRelation, SubordinateRelation)) -def test_relation_positional_arguments(klass): - with pytest.raises(TypeError): - klass("foo", "bar", None) - - -def test_relation_default_values(): - expected_id = next_relation_id(update=False) - endpoint = "database" - interface = "postgresql" - relation = Relation(endpoint, interface) - assert relation.id == expected_id - assert relation.endpoint == endpoint - assert relation.interface == interface - assert relation.local_app_data == {} - assert relation.local_unit_data == DEFAULT_JUJU_DATABAG - assert relation.remote_app_name == "remote" - assert relation.limit == 1 - assert relation.remote_app_data == {} - assert relation.remote_units_data == {0: DEFAULT_JUJU_DATABAG} - - -def test_subordinate_relation_default_values(): - expected_id = next_relation_id(update=False) - endpoint = "database" - interface = "postgresql" - relation = SubordinateRelation(endpoint, interface) - assert relation.id == expected_id - assert relation.endpoint == endpoint - assert relation.interface == interface - assert relation.local_app_data == {} - assert relation.local_unit_data == DEFAULT_JUJU_DATABAG - assert relation.remote_app_name == "remote" - assert relation.remote_unit_id == 0 - assert relation.remote_app_data == {} - assert relation.remote_unit_data == DEFAULT_JUJU_DATABAG - - -def test_peer_relation_default_values(): - expected_id = next_relation_id(update=False) - endpoint = "peers" - interface = "shared" - relation = PeerRelation(endpoint, interface) - assert relation.id == expected_id - assert relation.endpoint == endpoint - assert relation.interface == interface - assert relation.local_app_data == {} - assert relation.local_unit_data == DEFAULT_JUJU_DATABAG - assert relation.peers_data == {0: DEFAULT_JUJU_DATABAG} diff --git a/tests/test_e2e/test_secrets.py b/tests/test_e2e/test_secrets.py index 7229bd9f..97e1f3b2 100644 --- a/tests/test_e2e/test_secrets.py +++ b/tests/test_e2e/test_secrets.py @@ -538,23 +538,3 @@ def __init__(self, *args): secret.remove_all_revisions() assert not mgr.output.secrets[0].contents # secret wiped - - -def test_no_additional_positional_arguments(): - with pytest.raises(TypeError): - Secret({}, None) - - -def test_default_values(): - contents = {"foo": "bar"} - id = "secret:1" - secret = Secret(contents, id=id) - assert secret.contents == contents - assert secret.id == id - assert secret.label is None - assert secret.revision == 0 - assert secret.description is None - assert secret.owner is None - assert secret.rotate is None - assert secret.expire is None - assert secret.remote_grants == {} diff --git a/tests/test_e2e/test_state.py b/tests/test_e2e/test_state.py index 3f119909..0c79da86 100644 --- a/tests/test_e2e/test_state.py +++ b/tests/test_e2e/test_state.py @@ -1,4 +1,3 @@ -import copy from dataclasses import asdict, replace from typing import Type @@ -7,16 +6,7 @@ from ops.framework import EventBase, Framework from ops.model import ActiveStatus, UnknownStatus, WaitingStatus -from scenario.state import ( - DEFAULT_JUJU_DATABAG, - Address, - BindAddress, - Container, - Model, - Network, - Relation, - State, -) +from scenario.state import DEFAULT_JUJU_DATABAG, Container, Relation, State from tests.helpers import jsonpatch_delta, sort_patch, trigger CUSTOM_EVT_SUFFIXES = { @@ -241,78 +231,3 @@ def pre_event(charm: CharmBase): assert out.relations[0].local_app_data == {"a": "b"} assert out.relations[0].local_unit_data == {"c": "d", **DEFAULT_JUJU_DATABAG} - - -@pytest.mark.parametrize( - "klass,num_args", - [ - (State, (1,)), - (Address, (0, 2)), - (BindAddress, (0, 2)), - (Network, (0, 2)), - ], -) -def test_positional_arguments(klass, num_args): - for num in num_args: - args = (None,) * num - with pytest.raises(TypeError): - klass(*args) - - -def test_model_positional_arguments(): - with pytest.raises(TypeError): - Model("", "") - - -def test_container_positional_arguments(): - with pytest.raises(TypeError): - Container("", "") - - -def test_container_default_values(): - name = "foo" - container = Container(name) - assert container.name == name - assert container.can_connect is False - assert container.layers == {} - assert container.service_status == {} - assert container.mounts == {} - assert container.exec_mock == {} - assert container.layers == {} - assert container._base_plan == {} - - -def test_state_default_values(): - state = State() - assert state.config == {} - assert state.relations == [] - assert state.networks == {} - assert state.containers == [] - assert state.storage == [] - assert state.opened_ports == [] - assert state.secrets == [] - assert state.resources == {} - assert state.deferred == [] - assert isinstance(state.model, Model) - assert state.leader is False - assert state.planned_units == 1 - assert state.app_status == UnknownStatus() - assert state.unit_status == UnknownStatus() - assert state.workload_version == "" - - -def test_deepcopy_state(): - containers = [Container("foo"), Container("bar")] - state = State(containers=containers) - state_copy = copy.deepcopy(state) - for container in state.containers: - copied_container = state_copy.get_container(container.name) - assert container.name == copied_container.name - - -def test_replace_state(): - containers = [Container("foo"), Container("bar")] - state = State(containers=containers, leader=True) - state2 = replace(state, leader=False) - assert state.leader != state2.leader - assert state.containers == state2.containers diff --git a/tests/test_e2e/test_stored_state.py b/tests/test_e2e/test_stored_state.py index 38c38efd..22a6235e 100644 --- a/tests/test_e2e/test_stored_state.py +++ b/tests/test_e2e/test_stored_state.py @@ -39,9 +39,7 @@ def test_stored_state_initialized(mycharm): out = trigger( State( stored_state=[ - StoredState( - owner_path="MyCharm", name="_stored", content={"foo": "FOOX"} - ), + StoredState("MyCharm", name="_stored", content={"foo": "FOOX"}), ] ), "start", @@ -51,16 +49,3 @@ def test_stored_state_initialized(mycharm): # todo: ordering is messy? assert out.stored_state[1].content == {"foo": "FOOX", "baz": {12: 142}} assert out.stored_state[0].content == {"foo": "bar", "baz": {12: 142}} - - -def test_positional_arguments(): - with pytest.raises(TypeError): - StoredState("_stored", "") - - -def test_default_arguments(): - s = StoredState() - assert s.name == "_stored" - assert s.owner_path == None - assert s.content == {} - assert s._data_type_name == "StoredStateData"