From b24d4a469e245c32a5e047f92eda0b20cb748903 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Tue, 17 Feb 2026 23:46:00 +1300 Subject: [PATCH 01/12] refactor: use __init__ to distinguish between types possible for creating and types attributes will be --- testing/src/scenario/state.py | 489 ++++++++++++++++++--------- testing/tests/test_e2e/test_state.py | 2 +- 2 files changed, 326 insertions(+), 165 deletions(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 5b46ac878..d8d1fe783 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -150,14 +150,14 @@ class JujuLogLine: """The log message.""" -@dataclasses.dataclass(frozen=True, kw_only=True) +@dataclasses.dataclass(frozen=True, kw_only=True, init=False) class CloudCredential: # noqa: D101 __doc__ = ops.CloudCredential.__doc__ auth_type: str """Authentication type.""" - attributes: Mapping[str, str] = dataclasses.field(default_factory=dict) + attributes: dict[str, str] """A dictionary containing cloud credentials. For example, for AWS, it contains `access-key` and `secret-key`; @@ -165,11 +165,19 @@ class CloudCredential: # noqa: D101 can be found here. """ - redacted: Sequence[str] = dataclasses.field(default_factory=list) + redacted: list[str] """A list of redacted generic cloud API secrets.""" - def __post_init__(self): - _deepcopy_mutable_fields(self) + def __init__( + self, + *, + auth_type: str, + attributes: Mapping[str, str] | None = None, + redacted: Sequence[str] | None = None, + ): + object.__setattr__(self, 'auth_type', auth_type) + object.__setattr__(self, 'attributes', dict(attributes) if attributes else {}) + object.__setattr__(self, 'redacted', list(redacted) if redacted else []) def _to_ops(self) -> CloudCredential_Ops: return CloudCredential_Ops( @@ -179,44 +187,66 @@ def _to_ops(self) -> CloudCredential_Ops: ) -@dataclasses.dataclass(frozen=True) +@dataclasses.dataclass(frozen=True, init=False) class CloudSpec: # noqa: D101 __doc__ = ops.CloudSpec.__doc__ type: str """Type of the cloud.""" - _: dataclasses.KW_ONLY - - name: str = 'localhost' + name: str """Juju cloud name.""" - region: str | None = None + region: str | None """Region of the cloud.""" - endpoint: str | None = None + endpoint: str | None """Endpoint of the cloud.""" - identity_endpoint: str | None = None + identity_endpoint: str | None """Identity endpoint of the cloud.""" - storage_endpoint: str | None = None + storage_endpoint: str | None """Storage endpoint of the cloud.""" - credential: CloudCredential | None = None + credential: CloudCredential | None """Cloud credentials with key-value attributes.""" - ca_certificates: Sequence[str] = dataclasses.field(default_factory=list) + ca_certificates: list[str] """A list of CA certificates.""" - skip_tls_verify: bool = False + skip_tls_verify: bool """Whether to skip TLS verification.""" - is_controller_cloud: bool = False + is_controller_cloud: bool """If this is the cloud used by the controller.""" - def __post_init__(self): - _deepcopy_mutable_fields(self) + def __init__( + self, + type: str, + *, + name: str = 'localhost', + region: str | None = None, + endpoint: str | None = None, + identity_endpoint: str | None = None, + storage_endpoint: str | None = None, + credential: CloudCredential | None = None, + ca_certificates: Sequence[str] | None = None, + skip_tls_verify: bool = False, + is_controller_cloud: bool = False, + ): + object.__setattr__(self, 'type', type) + object.__setattr__(self, 'name', name) + object.__setattr__(self, 'region', region) + object.__setattr__(self, 'endpoint', endpoint) + object.__setattr__(self, 'identity_endpoint', identity_endpoint) + object.__setattr__(self, 'storage_endpoint', storage_endpoint) + object.__setattr__(self, 'credential', credential) + object.__setattr__( + self, 'ca_certificates', list(ca_certificates) if ca_certificates else [] + ) + object.__setattr__(self, 'skip_tls_verify', skip_tls_verify) + object.__setattr__(self, 'is_controller_cloud', is_controller_cloud) def _to_ops(self) -> CloudSpec_Ops: return CloudSpec_Ops( @@ -240,7 +270,7 @@ def _generate_secret_id(): return f'secret:{secret_id}' -@dataclasses.dataclass(frozen=True) +@dataclasses.dataclass(frozen=True, init=False) class Secret: """A Juju secret. @@ -253,21 +283,19 @@ class Secret: This is the content the charm will receive with a :meth:`ops.Secret.get_content` call.""" - _: dataclasses.KW_ONLY - - latest_content: RawSecretRevisionContents | None = None + latest_content: RawSecretRevisionContents | None """The content of the latest revision of the secret. This is the content the charm will receive with a :meth:`ops.Secret.peek_content` call.""" - id: str = dataclasses.field(default_factory=_generate_secret_id) + id: str """The Juju ID of the secret. This is automatically assigned and should not usually need to be explicitly set. """ - owner: Literal['unit', 'app', None] = None + owner: Literal['unit', 'app', None] """Indicates if the secret is owned by *this* unit, *this* application, or another application/unit. @@ -275,37 +303,63 @@ class Secret: to this unit. """ - remote_grants: Mapping[int, set[str]] = dataclasses.field(default_factory=dict) + remote_grants: dict[int, set[str]] """Mapping from relation IDs to remote units and applications to which this secret has been granted.""" - label: str | None = None + label: str | None """A human-readable label the charm can use to retrieve the secret. If this is set, it implies that the charm has previously set the label. """ - description: str | None = None + description: str | None """A human-readable description of the secret.""" - expire: datetime.datetime | None = None + expire: datetime.datetime | None """The time at which the secret will expire.""" - rotate: SecretRotate | None = None + rotate: SecretRotate | None """The rotation policy for the secret.""" # what revision is currently tracked by this charm. Only meaningful if owner=False - _tracked_revision: int = 1 + _tracked_revision: int # what revision is the latest for this secret. - _latest_revision: int = 1 + _latest_revision: int + + def __init__( + self, + tracked_content: RawSecretRevisionContents, + *, + latest_content: RawSecretRevisionContents | None = None, + id: str | None = None, + owner: Literal['unit', 'app', None] = None, + remote_grants: Mapping[int, set[str]] | None = None, + label: str | None = None, + description: str | None = None, + expire: datetime.datetime | None = None, + rotate: SecretRotate | None = None, + _tracked_revision: int = 1, + _latest_revision: int = 1, + ): + object.__setattr__(self, 'tracked_content', tracked_content) + object.__setattr__( + self, + 'latest_content', + latest_content if latest_content is not None else tracked_content, + ) + object.__setattr__(self, 'id', id if id is not None else _generate_secret_id()) + object.__setattr__(self, 'owner', owner) + object.__setattr__(self, 'remote_grants', dict(remote_grants) if remote_grants else {}) + object.__setattr__(self, 'label', label) + object.__setattr__(self, 'description', description) + object.__setattr__(self, 'expire', expire) + object.__setattr__(self, 'rotate', rotate) + object.__setattr__(self, '_tracked_revision', _tracked_revision) + object.__setattr__(self, '_latest_revision', _latest_revision) + _deepcopy_mutable_fields(self) def __hash__(self) -> int: return hash(self.id) - def __post_init__(self): - if self.latest_content is None: - # bypass frozen dataclass - object.__setattr__(self, 'latest_content', self.tracked_content) - _deepcopy_mutable_fields(self) - def _set_label(self, label: str): # bypass frozen dataclass object.__setattr__(self, 'label', label) @@ -400,7 +454,7 @@ def _hook_tool_output_fmt(self): return dct -@dataclasses.dataclass(frozen=True) +@dataclasses.dataclass(frozen=True, init=False) class Network: """A Juju network space. @@ -423,28 +477,44 @@ class Network: binding_name: str """The name of the network space.""" - bind_addresses: Sequence[BindAddress] = dataclasses.field( - default_factory=lambda: [BindAddress([Address('192.0.2.0')])], - ) + bind_addresses: list[BindAddress] """Addresses that the charm's application should bind to.""" - _: dataclasses.KW_ONLY - - ingress_addresses: Sequence[str] = dataclasses.field( - default_factory=lambda: ['192.0.2.0'], - ) + ingress_addresses: list[str] """Addresses other applications should use to connect to the unit.""" - egress_subnets: Sequence[str] = dataclasses.field( - default_factory=lambda: ['192.0.2.0/24'], - ) + egress_subnets: list[str] """Subnets that other units will see the charm connecting from.""" + def __init__( + self, + binding_name: str, + bind_addresses: Sequence[BindAddress] | None = None, + *, + ingress_addresses: Sequence[str] | None = None, + egress_subnets: Sequence[str] | None = None, + ): + object.__setattr__(self, 'binding_name', binding_name) + object.__setattr__( + self, + 'bind_addresses', + list(bind_addresses) + if bind_addresses is not None + else [BindAddress([Address('192.0.2.0')])], + ) + object.__setattr__( + self, + 'ingress_addresses', + list(ingress_addresses) if ingress_addresses is not None else ['192.0.2.0'], + ) + object.__setattr__( + self, + 'egress_subnets', + list(egress_subnets) if egress_subnets is not None else ['192.0.2.0/24'], + ) + def __hash__(self) -> int: return hash(self.binding_name) - def __post_init__(self): - _deepcopy_mutable_fields(self) - def _hook_tool_output_fmt(self): # dumps itself to dict in the same format the hook command would return { @@ -761,26 +831,24 @@ def _generate_new_change_id(): return _CHANGE_IDS -@dataclasses.dataclass(frozen=True) +@dataclasses.dataclass(frozen=True, init=False) class Exec: """Mock data for simulated :meth:`ops.Container.exec` calls.""" - command_prefix: Sequence[str] - - _: dataclasses.KW_ONLY + command_prefix: tuple[str, ...] - return_code: int = 0 + return_code: int """The return code of the process. Use 0 to mock the process ending successfully, and other values for failure. """ - stdout: str = '' + stdout: str """Any content written to stdout by the process. Provide content that the real process would write to stdout, which can be read by the charm. """ - stderr: str = '' + stderr: str """Any content written to stderr by the process. Provide content that the real process would write to stderr, which can be @@ -788,14 +856,24 @@ class Exec: """ # change ID: used internally to keep track of mocked processes - _change_id: int = dataclasses.field(default_factory=_generate_new_change_id) + _change_id: int - def __post_init__(self): - # The command prefix can be any sequence type, and a list is tidier to - # write when there's only one string. However, this object needs to be - # hashable, so can't contain a list. We 'freeze' the sequence to a tuple - # to support that. - object.__setattr__(self, 'command_prefix', tuple(self.command_prefix)) + def __init__( + self, + command_prefix: Sequence[str], + *, + return_code: int = 0, + stdout: str = '', + stderr: str = '', + _change_id: int | None = None, + ): + object.__setattr__(self, 'command_prefix', tuple(command_prefix)) + object.__setattr__(self, 'return_code', return_code) + object.__setattr__(self, 'stdout', stdout) + object.__setattr__(self, 'stderr', stderr) + object.__setattr__( + self, '_change_id', _change_id if _change_id is not None else _generate_new_change_id() + ) def _run(self) -> int: return self._change_id @@ -831,7 +909,7 @@ def _next_notice_id(*, update: bool = True): return str(cur) -@dataclasses.dataclass(frozen=True) +@dataclasses.dataclass(frozen=True, init=False) class Notice: """A Pebble notice.""" @@ -842,44 +920,72 @@ class Notice: ``canonical.com/postgresql/backup`` or ``example.com/mycharm/notice``. """ - _: dataclasses.KW_ONLY - - id: str = dataclasses.field(default_factory=_next_notice_id) + id: str """Unique ID for this notice.""" - user_id: int | None = None + user_id: int | None """UID of the user who may view this notice (None means notice is public).""" - type: pebble.NoticeType | str = pebble.NoticeType.CUSTOM + type: pebble.NoticeType | str """Type of the notice.""" - first_occurred: datetime.datetime = dataclasses.field(default_factory=_now_utc) + first_occurred: datetime.datetime """The first time one of these notices (type and key combination) occurs.""" - last_occurred: datetime.datetime = dataclasses.field(default_factory=_now_utc) + last_occurred: datetime.datetime """The last time one of these notices occurred.""" - last_repeated: datetime.datetime = dataclasses.field(default_factory=_now_utc) + last_repeated: datetime.datetime """The time this notice was last repeated. See Pebble's `Notices documentation `_ for an explanation of what "repeated" means. """ - occurrences: int = 1 + occurrences: int """The number of times one of these notices has occurred.""" - last_data: Mapping[str, str] = dataclasses.field(default_factory=dict) + last_data: dict[str, str] """Additional data captured from the last occurrence of one of these notices.""" - repeat_after: datetime.timedelta | None = None + repeat_after: datetime.timedelta | None """Minimum time after one of these was last repeated before Pebble will repeat it again.""" - expire_after: datetime.timedelta | None = None + expire_after: datetime.timedelta | None """How long since one of these last occurred until Pebble will drop the notice.""" - def __post_init__(self): - _deepcopy_mutable_fields(self) + def __init__( + self, + key: str, + *, + id: str | None = None, + user_id: int | None = None, + type: pebble.NoticeType | str = pebble.NoticeType.CUSTOM, + first_occurred: datetime.datetime | None = None, + last_occurred: datetime.datetime | None = None, + last_repeated: datetime.datetime | None = None, + occurrences: int = 1, + last_data: Mapping[str, str] | None = None, + repeat_after: datetime.timedelta | None = None, + expire_after: datetime.timedelta | None = None, + ): + object.__setattr__(self, 'key', key) + object.__setattr__(self, 'id', id if id is not None else _next_notice_id()) + object.__setattr__(self, 'user_id', user_id) + object.__setattr__(self, 'type', type) + object.__setattr__( + self, 'first_occurred', first_occurred if first_occurred is not None else _now_utc() + ) + object.__setattr__( + self, 'last_occurred', last_occurred if last_occurred is not None else _now_utc() + ) + object.__setattr__( + self, 'last_repeated', last_repeated if last_repeated is not None else _now_utc() + ) + object.__setattr__(self, 'occurrences', occurrences) + object.__setattr__(self, 'last_data', dict(last_data) if last_data else {}) + object.__setattr__(self, 'repeat_after', repeat_after) + object.__setattr__(self, 'expire_after', expire_after) def _to_ops(self) -> pebble.Notice: return pebble.Notice( @@ -987,16 +1093,14 @@ def _to_ops(self) -> pebble.CheckInfo: ) -@dataclasses.dataclass(frozen=True) +@dataclasses.dataclass(frozen=True, init=False) class Container: """A Kubernetes container where a charm's workload runs.""" name: str """Name of the container, as found in the charm metadata.""" - _: dataclasses.KW_ONLY - - can_connect: bool = False + can_connect: bool """When False, all Pebble operations will fail.""" # This is the base plan. On top of it, one can add layers. @@ -1004,10 +1108,10 @@ class Container: # pebble or derive them from the resulting plan (which one CAN get from pebble). # So if we are instantiating Container by fetching info from a 'live' charm, the 'layers' # will be unknown. all that we can know is the resulting plan (the 'computed plan'). - _base_plan: Mapping[str, Any] = dataclasses.field(default_factory=dict) + _base_plan: dict[str, Any] # We expect most of the user-facing testing to be covered by this 'layers' attribute, # as it is all that will be known when unit-testing. - layers: Mapping[str, pebble.Layer] = dataclasses.field(default_factory=dict) + layers: dict[str, pebble.Layer] """All :class:`ops.pebble.Layer` definitions that have already been added to the container. Note that the layers should be added to the dictionary in the order in which they would have @@ -1016,12 +1120,10 @@ class Container: this means adding them in the order of the API calls. """ - service_statuses: Mapping[str, pebble.ServiceStatus] = dataclasses.field( - default_factory=dict, - ) + service_statuses: dict[str, pebble.ServiceStatus] """The current status of each Pebble service running in the container.""" - mounts: Mapping[str, Mount] = dataclasses.field(default_factory=dict) + mounts: dict[str, Mount] """Provides access to the contents of the simulated container filesystem. For example, suppose you want to express that your container has: @@ -1044,7 +1146,7 @@ class Container: be safely modified. """ - execs: Iterable[Exec] = frozenset() + execs: frozenset[Exec] """Simulate executing commands in the container. Specify each command the charm might run in the container and an :class:`Exec` @@ -1065,21 +1167,43 @@ class Container: ) """ - notices: Sequence[Notice] = dataclasses.field(default_factory=list) + notices: list[Notice] """Any Pebble notices that already exist in the container.""" - check_infos: Iterable[CheckInfo] = frozenset() + check_infos: frozenset[CheckInfo] """All Pebble health checks that have been added to the container.""" + def __init__( + self, + name: str, + *, + can_connect: bool = False, + _base_plan: Mapping[str, Any] | None = None, + layers: Mapping[str, pebble.Layer] | None = None, + service_statuses: Mapping[str, pebble.ServiceStatus] | None = None, + mounts: Mapping[str, Mount] | None = None, + execs: Iterable[Exec] | None = None, + notices: Sequence[Notice] | None = None, + check_infos: Iterable[CheckInfo] | None = None, + ): + object.__setattr__(self, 'name', name) + object.__setattr__(self, 'can_connect', can_connect) + object.__setattr__(self, '_base_plan', dict(_base_plan) if _base_plan else {}) + object.__setattr__(self, 'layers', dict(layers) if layers else {}) + object.__setattr__( + self, 'service_statuses', dict(service_statuses) if service_statuses else {} + ) + object.__setattr__(self, 'mounts', dict(mounts) if mounts else {}) + object.__setattr__(self, 'execs', frozenset(execs) if execs is not None else frozenset()) + object.__setattr__(self, 'notices', list(notices) if notices else []) + object.__setattr__( + self, 'check_infos', frozenset(check_infos) if check_infos is not None else frozenset() + ) + _deepcopy_mutable_fields(self) + def __hash__(self) -> int: return hash(self.name) - def __post_init__(self): - if not isinstance(self.execs, frozenset): - # Allow passing a regular set (or other iterable) of Execs. - object.__setattr__(self, 'execs', frozenset(self.execs)) - _deepcopy_mutable_fields(self) - def _render_services(self): services: dict[str, pebble.Service] = {} for layer in self.layers.values(): @@ -1332,11 +1456,11 @@ def __init__(self, message: str = ''): ) -@dataclasses.dataclass(frozen=True) +@dataclasses.dataclass(frozen=True, init=False) class StoredState: """Represents unit-local state that persists across events.""" - name: str = '_stored' + name: str """The attribute in the parent Object where the state is stored. For example, ``_stored`` in this class:: @@ -1346,9 +1470,7 @@ class MyCharm(ops.CharmBase): """ - _: dataclasses.KW_ONLY - - owner_path: str | None = None + owner_path: str | None """The path to the owner of this StoredState instance. If None, the owner is the Framework. Otherwise, /-separated object names, @@ -1359,18 +1481,29 @@ class MyCharm(ops.CharmBase): # However, it's complex to describe those types, since it's a recursive # definition - even in TypeShed the _Marshallable type includes containers # like list[Any], which seems to defeat the point. - content: Mapping[str, Any] = dataclasses.field(default_factory=dict) + content: dict[str, Any] """The content of the :class:`ops.StoredState` instance.""" - _data_type_name: str = 'StoredStateData' + _data_type_name: str + + def __init__( + self, + name: str = '_stored', + *, + owner_path: str | None = None, + content: Mapping[str, Any] | None = None, + _data_type_name: str = 'StoredStateData', + ): + object.__setattr__(self, 'name', name) + object.__setattr__(self, 'owner_path', owner_path) + object.__setattr__(self, 'content', dict(content) if content else {}) + object.__setattr__(self, '_data_type_name', _data_type_name) + _deepcopy_mutable_fields(self) @property def _handle_path(self): return f'{self.owner_path or ""}/{self._data_type_name}[{self.name}]' - def __post_init__(self): - _deepcopy_mutable_fields(self) - def __hash__(self) -> int: return hash(self._handle_path) @@ -1531,7 +1664,7 @@ class Resource: """A local path that will be provided to the charm as the content of the resource.""" -@dataclasses.dataclass(frozen=True, kw_only=True) +@dataclasses.dataclass(frozen=True, kw_only=True, init=False) class State: """Represents the Juju-owned portion of a unit's state. @@ -1540,13 +1673,11 @@ class State: return data from `State.leader`, and so on. """ - config: dict[str, str | int | float | bool] = dataclasses.field( - default_factory=dict, - ) + config: dict[str, str | int | float | bool] """The present configuration of this charm.""" - relations: Iterable[RelationBase] = dataclasses.field(default_factory=frozenset) + relations: frozenset[RelationBase] """All relations that currently exist for this charm.""" - networks: Iterable[Network] = dataclasses.field(default_factory=frozenset) + networks: frozenset[Network] """Manual overrides for any relation and extra bindings currently provisioned for this charm. If a metadata-defined relation endpoint is not explicitly mapped to a Network in this field, it will be defaulted. @@ -1556,28 +1687,28 @@ class State: support it, but use at your own risk. If a metadata-defined extra-binding is left empty, it will be defaulted. """ - containers: Iterable[Container] = dataclasses.field(default_factory=frozenset) + containers: frozenset[Container] """All containers (whether they can connect or not) that this charm is aware of.""" - storages: Iterable[Storage] = dataclasses.field(default_factory=frozenset) + storages: frozenset[Storage] """All **attached** storage instances for this charm. If a storage is not attached, omit it from this listing.""" # we don't use sets to make json serialization easier - opened_ports: Iterable[Port] = dataclasses.field(default_factory=frozenset) + opened_ports: frozenset[Port] """Ports opened by Juju on this charm.""" - leader: bool = False + leader: bool """Whether this charm has leadership.""" - model: Model = dataclasses.field(default_factory=Model) + model: Model """The model this charm lives in.""" - secrets: Iterable[Secret] = dataclasses.field(default_factory=frozenset) + secrets: frozenset[Secret] """The secrets this charm has access to (as an owner, or as a grantee). The presence of a secret in this list entails that the charm can read it. Whether it can manage it or not depends on the individual secret's `owner` flag.""" - resources: Iterable[Resource] = dataclasses.field(default_factory=frozenset) + resources: frozenset[Resource] """All resources that this charm can access.""" - planned_units: int = 1 + planned_units: int """Number of non-dying planned units that are expected to be running this application. Use with caution.""" @@ -1586,45 +1717,72 @@ class State: # dispatched, and represent the events that had been deferred during the previous run. # If the charm defers any events during "this execution", they will be appended # to this list. - deferred: Sequence[DeferredEvent] = dataclasses.field(default_factory=list) + deferred: list[DeferredEvent] """Events that have been deferred on this charm by some previous execution.""" - stored_states: Iterable[StoredState] = dataclasses.field( - default_factory=frozenset, - ) + stored_states: frozenset[StoredState] """Contents of a charm's stored state.""" # the current statuses. - app_status: _EntityStatus = dataclasses.field(default_factory=UnknownStatus) + app_status: _EntityStatus """Status of the application.""" - unit_status: _EntityStatus = dataclasses.field(default_factory=UnknownStatus) + unit_status: _EntityStatus """Status of the unit.""" - workload_version: str = '' + workload_version: str """Workload version.""" - def __post_init__(self): - # Let people pass in the ops classes, and convert them to the appropriate Scenario classes. - for name in ['app_status', 'unit_status']: - val = getattr(self, name) - if isinstance(val, _EntityStatus): - pass + def __init__( + self, + *, + config: dict[str, str | int | float | bool] | None = None, + relations: Iterable[RelationBase] | None = None, + networks: Iterable[Network] | None = None, + containers: Iterable[Container] | None = None, + storages: Iterable[Storage] | None = None, + opened_ports: Iterable[Port] | None = None, + leader: bool = False, + model: Model | None = None, + secrets: Iterable[Secret] | None = None, + resources: Iterable[Resource] | None = None, + planned_units: int = 1, + deferred: Sequence[DeferredEvent] | None = None, + stored_states: Iterable[StoredState] | None = None, + app_status: _EntityStatus | StatusBase | None = None, + unit_status: _EntityStatus | StatusBase | None = None, + workload_version: str = '', + ): + object.__setattr__(self, 'config', config if config is not None else {}) + object.__setattr__(self, 'leader', leader) + object.__setattr__(self, 'model', model if model is not None else Model()) + object.__setattr__(self, 'planned_units', planned_units) + object.__setattr__(self, 'deferred', list(deferred) if deferred else []) + object.__setattr__(self, 'workload_version', workload_version) + + # Handle status conversion from ops types. + for name, val in [('app_status', app_status), ('unit_status', unit_status)]: + if val is None: + object.__setattr__(self, name, UnknownStatus()) + elif isinstance(val, _EntityStatus): + object.__setattr__(self, name, val) elif isinstance(val, StatusBase): object.__setattr__(self, name, _EntityStatus.from_ops(val)) else: raise TypeError(f'Invalid status.{name}: {val!r}') - normalised_ports = [ + + # Normalise ops.Port to scenario Port. + normalised_ports = frozenset( Port(protocol=port.protocol, port=port.port) if isinstance(port, ops.Port) else port - for port in self.opened_ports - ] - if self.opened_ports != normalised_ports: - object.__setattr__(self, 'opened_ports', normalised_ports) - normalised_storage = [ + for port in (opened_ports or ()) + ) + object.__setattr__(self, 'opened_ports', normalised_ports) + + # Normalise ops.Storage to scenario Storage. + normalised_storage = frozenset( Storage(name=storage.name, index=storage.index) if isinstance(storage, ops.Storage) else storage - for storage in self.storages - ] - if self.storages != normalised_storage: - object.__setattr__(self, 'storages', normalised_storage) + for storage in (storages or ()) + ) + object.__setattr__(self, 'storages', normalised_storage) # ops.Container, ops.Model, ops.Relation, ops.Secret should not be instantiated by # charmers. @@ -1632,23 +1790,26 @@ def __post_init__(self): # ops.Resources does not contain the source of the resource, so cannot be converted. # ops.StoredState is not convenient to initialise with data, so not useful here. - # It's convenient to pass a set, but we really want the attributes to be - # frozen sets to increase the immutability of State objects. - for name in [ - 'relations', - 'containers', - 'storages', - 'networks', - 'opened_ports', - 'secrets', - 'resources', + object.__setattr__( + self, 'relations', frozenset(relations) if relations is not None else frozenset() + ) + object.__setattr__( + self, 'networks', frozenset(networks) if networks is not None else frozenset() + ) + object.__setattr__( + self, 'containers', frozenset(containers) if containers is not None else frozenset() + ) + object.__setattr__( + self, 'secrets', frozenset(secrets) if secrets is not None else frozenset() + ) + object.__setattr__( + self, 'resources', frozenset(resources) if resources is not None else frozenset() + ) + object.__setattr__( + self, 'stored_states', - ]: - val = getattr(self, name) - # It's ok to pass any iterable (of hashable objects), but you'll get - # a frozenset as the actual attribute. - if not isinstance(val, frozenset): - object.__setattr__(self, name, frozenset(val)) + frozenset(stored_states) if stored_states is not None else frozenset(), + ) def _update_workload_version(self, new_workload_version: str): """Update the current app version and record the previous one.""" diff --git a/testing/tests/test_e2e/test_state.py b/testing/tests/test_e2e/test_state.py index 500098d02..9ec32f160 100644 --- a/testing/tests/test_e2e/test_state.py +++ b/testing/tests/test_e2e/test_state.py @@ -385,7 +385,6 @@ def test_replace_state(): (Container, 'layers', {'name': 'foo'}), (Container, 'service_statuses', {'name': 'foo'}), (Container, 'mounts', {'name': 'foo'}), - (Container, 'notices', {'name': 'foo'}), (StoredState, 'content', {}), ], ) @@ -415,6 +414,7 @@ def test_immutable_content_dict( (Network, 'bind_addresses', {'binding_name': 'foo'}), (Network, 'ingress_addresses', {'binding_name': 'foo'}), (Network, 'egress_subnets', {'binding_name': 'foo'}), + (Container, 'notices', {'name': 'foo'}), ], ) def test_immutable_content_list( From 3658da3465cf94b6413f08765d7faef2ae484c61 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Wed, 18 Feb 2026 00:05:26 +1300 Subject: [PATCH 02/12] Restore accidentally removed deepcopies. --- testing/src/scenario/state.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index d8d1fe783..2f9b252d1 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -178,6 +178,7 @@ def __init__( object.__setattr__(self, 'auth_type', auth_type) object.__setattr__(self, 'attributes', dict(attributes) if attributes else {}) object.__setattr__(self, 'redacted', list(redacted) if redacted else []) + _deepcopy_mutable_fields(self) def _to_ops(self) -> CloudCredential_Ops: return CloudCredential_Ops( @@ -247,6 +248,7 @@ def __init__( ) object.__setattr__(self, 'skip_tls_verify', skip_tls_verify) object.__setattr__(self, 'is_controller_cloud', is_controller_cloud) + _deepcopy_mutable_fields(self) def _to_ops(self) -> CloudSpec_Ops: return CloudSpec_Ops( @@ -511,6 +513,7 @@ def __init__( 'egress_subnets', list(egress_subnets) if egress_subnets is not None else ['192.0.2.0/24'], ) + _deepcopy_mutable_fields(self) def __hash__(self) -> int: return hash(self.binding_name) @@ -986,6 +989,7 @@ def __init__( object.__setattr__(self, 'last_data', dict(last_data) if last_data else {}) object.__setattr__(self, 'repeat_after', repeat_after) object.__setattr__(self, 'expire_after', expire_after) + _deepcopy_mutable_fields(self) def _to_ops(self) -> pebble.Notice: return pebble.Notice( From 4b1c247b46eb4fb21a9a5310a575ca05f46b2b3d Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Wed, 18 Feb 2026 21:56:46 +1300 Subject: [PATCH 03/12] Address review comments. --- testing/src/scenario/state.py | 136 +++++++++++---------------- testing/tests/test_e2e/test_state.py | 47 +++++++++ 2 files changed, 102 insertions(+), 81 deletions(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 2f9b252d1..6b533ce1f 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -57,7 +57,7 @@ class _StateKwargs(TypedDict, total=False): secrets: Iterable[Secret] resources: Iterable[Resource] planned_units: int - deferred: Sequence[DeferredEvent] + deferred: Iterable[DeferredEvent] stored_states: Iterable[StoredState] app_status: _EntityStatus unit_status: _EntityStatus @@ -172,13 +172,12 @@ def __init__( self, *, auth_type: str, - attributes: Mapping[str, str] | None = None, - redacted: Sequence[str] | None = None, + attributes: Mapping[str, str] = {}, + redacted: Iterable[str] = (), ): object.__setattr__(self, 'auth_type', auth_type) - object.__setattr__(self, 'attributes', dict(attributes) if attributes else {}) - object.__setattr__(self, 'redacted', list(redacted) if redacted else []) - _deepcopy_mutable_fields(self) + object.__setattr__(self, 'attributes', dict(attributes)) + object.__setattr__(self, 'redacted', list(redacted)) def _to_ops(self) -> CloudCredential_Ops: return CloudCredential_Ops( @@ -232,7 +231,7 @@ def __init__( identity_endpoint: str | None = None, storage_endpoint: str | None = None, credential: CloudCredential | None = None, - ca_certificates: Sequence[str] | None = None, + ca_certificates: Iterable[str] = (), skip_tls_verify: bool = False, is_controller_cloud: bool = False, ): @@ -243,12 +242,9 @@ def __init__( object.__setattr__(self, 'identity_endpoint', identity_endpoint) object.__setattr__(self, 'storage_endpoint', storage_endpoint) object.__setattr__(self, 'credential', credential) - object.__setattr__( - self, 'ca_certificates', list(ca_certificates) if ca_certificates else [] - ) + object.__setattr__(self, 'ca_certificates', list(ca_certificates)) object.__setattr__(self, 'skip_tls_verify', skip_tls_verify) object.__setattr__(self, 'is_controller_cloud', is_controller_cloud) - _deepcopy_mutable_fields(self) def _to_ops(self) -> CloudSpec_Ops: return CloudSpec_Ops( @@ -285,7 +281,7 @@ class Secret: This is the content the charm will receive with a :meth:`ops.Secret.get_content` call.""" - latest_content: RawSecretRevisionContents | None + latest_content: RawSecretRevisionContents """The content of the latest revision of the secret. This is the content the charm will receive with a @@ -297,7 +293,7 @@ class Secret: This is automatically assigned and should not usually need to be explicitly set. """ - owner: Literal['unit', 'app', None] + owner: Literal['unit', 'app'] | None """Indicates if the secret is owned by *this* unit, *this* application, or another application/unit. @@ -322,10 +318,10 @@ class Secret: """The rotation policy for the secret.""" # what revision is currently tracked by this charm. Only meaningful if owner=False - _tracked_revision: int + _tracked_revision = 1 # what revision is the latest for this secret. - _latest_revision: int + _latest_revision = 1 def __init__( self, @@ -333,14 +329,12 @@ def __init__( *, latest_content: RawSecretRevisionContents | None = None, id: str | None = None, - owner: Literal['unit', 'app', None] = None, - remote_grants: Mapping[int, set[str]] | None = None, + owner: Literal['unit', 'app'] | None = None, + remote_grants: Mapping[int, set[str]] = {}, label: str | None = None, description: str | None = None, expire: datetime.datetime | None = None, rotate: SecretRotate | None = None, - _tracked_revision: int = 1, - _latest_revision: int = 1, ): object.__setattr__(self, 'tracked_content', tracked_content) object.__setattr__( @@ -350,13 +344,13 @@ def __init__( ) object.__setattr__(self, 'id', id if id is not None else _generate_secret_id()) object.__setattr__(self, 'owner', owner) - object.__setattr__(self, 'remote_grants', dict(remote_grants) if remote_grants else {}) + object.__setattr__(self, 'remote_grants', dict(remote_grants)) object.__setattr__(self, 'label', label) object.__setattr__(self, 'description', description) object.__setattr__(self, 'expire', expire) object.__setattr__(self, 'rotate', rotate) - object.__setattr__(self, '_tracked_revision', _tracked_revision) - object.__setattr__(self, '_latest_revision', _latest_revision) + object.__setattr__(self, '_tracked_revision', 1) + object.__setattr__(self, '_latest_revision', 1) _deepcopy_mutable_fields(self) def __hash__(self) -> int: @@ -490,10 +484,10 @@ class Network: def __init__( self, binding_name: str, - bind_addresses: Sequence[BindAddress] | None = None, + bind_addresses: Iterable[BindAddress] | None = None, *, - ingress_addresses: Sequence[str] | None = None, - egress_subnets: Sequence[str] | None = None, + ingress_addresses: Iterable[str] | None = None, + egress_subnets: Iterable[str] | None = None, ): object.__setattr__(self, 'binding_name', binding_name) object.__setattr__( @@ -513,7 +507,6 @@ def __init__( 'egress_subnets', list(egress_subnets) if egress_subnets is not None else ['192.0.2.0/24'], ) - _deepcopy_mutable_fields(self) def __hash__(self) -> int: return hash(self.binding_name) @@ -968,7 +961,7 @@ def __init__( last_occurred: datetime.datetime | None = None, last_repeated: datetime.datetime | None = None, occurrences: int = 1, - last_data: Mapping[str, str] | None = None, + last_data: Mapping[str, str] = {}, repeat_after: datetime.timedelta | None = None, expire_after: datetime.timedelta | None = None, ): @@ -986,10 +979,9 @@ def __init__( self, 'last_repeated', last_repeated if last_repeated is not None else _now_utc() ) object.__setattr__(self, 'occurrences', occurrences) - object.__setattr__(self, 'last_data', dict(last_data) if last_data else {}) + object.__setattr__(self, 'last_data', dict(last_data)) object.__setattr__(self, 'repeat_after', repeat_after) object.__setattr__(self, 'expire_after', expire_after) - _deepcopy_mutable_fields(self) def _to_ops(self) -> pebble.Notice: return pebble.Notice( @@ -1182,27 +1174,23 @@ def __init__( name: str, *, can_connect: bool = False, - _base_plan: Mapping[str, Any] | None = None, - layers: Mapping[str, pebble.Layer] | None = None, - service_statuses: Mapping[str, pebble.ServiceStatus] | None = None, - mounts: Mapping[str, Mount] | None = None, - execs: Iterable[Exec] | None = None, - notices: Sequence[Notice] | None = None, - check_infos: Iterable[CheckInfo] | None = None, + _base_plan: Mapping[str, Any] = {}, + layers: Mapping[str, pebble.Layer] = {}, + service_statuses: Mapping[str, pebble.ServiceStatus] = {}, + mounts: Mapping[str, Mount] = {}, + execs: Iterable[Exec] = (), + notices: Iterable[Notice] = (), + check_infos: Iterable[CheckInfo] = (), ): object.__setattr__(self, 'name', name) object.__setattr__(self, 'can_connect', can_connect) - object.__setattr__(self, '_base_plan', dict(_base_plan) if _base_plan else {}) - object.__setattr__(self, 'layers', dict(layers) if layers else {}) - object.__setattr__( - self, 'service_statuses', dict(service_statuses) if service_statuses else {} - ) - object.__setattr__(self, 'mounts', dict(mounts) if mounts else {}) - object.__setattr__(self, 'execs', frozenset(execs) if execs is not None else frozenset()) - object.__setattr__(self, 'notices', list(notices) if notices else []) - object.__setattr__( - self, 'check_infos', frozenset(check_infos) if check_infos is not None else frozenset() - ) + object.__setattr__(self, '_base_plan', dict(_base_plan)) + object.__setattr__(self, 'layers', dict(layers)) + object.__setattr__(self, 'service_statuses', dict(service_statuses)) + object.__setattr__(self, 'mounts', dict(mounts)) + object.__setattr__(self, 'execs', frozenset(execs)) + object.__setattr__(self, 'notices', list(notices)) + object.__setattr__(self, 'check_infos', frozenset(check_infos)) _deepcopy_mutable_fields(self) def __hash__(self) -> int: @@ -1495,12 +1483,12 @@ def __init__( name: str = '_stored', *, owner_path: str | None = None, - content: Mapping[str, Any] | None = None, + content: Mapping[str, Any] = {}, _data_type_name: str = 'StoredStateData', ): object.__setattr__(self, 'name', name) object.__setattr__(self, 'owner_path', owner_path) - object.__setattr__(self, 'content', dict(content) if content else {}) + object.__setattr__(self, 'content', dict(content)) object.__setattr__(self, '_data_type_name', _data_type_name) _deepcopy_mutable_fields(self) @@ -1738,18 +1726,18 @@ def __init__( self, *, config: dict[str, str | int | float | bool] | None = None, - relations: Iterable[RelationBase] | None = None, - networks: Iterable[Network] | None = None, - containers: Iterable[Container] | None = None, - storages: Iterable[Storage] | None = None, - opened_ports: Iterable[Port] | None = None, + relations: Iterable[RelationBase] = (), + networks: Iterable[Network] = (), + containers: Iterable[Container] = (), + storages: Iterable[Storage] = (), + opened_ports: Iterable[Port] = (), leader: bool = False, model: Model | None = None, - secrets: Iterable[Secret] | None = None, - resources: Iterable[Resource] | None = None, + secrets: Iterable[Secret] = (), + resources: Iterable[Resource] = (), planned_units: int = 1, - deferred: Sequence[DeferredEvent] | None = None, - stored_states: Iterable[StoredState] | None = None, + deferred: Iterable[DeferredEvent] = (), + stored_states: Iterable[StoredState] = (), app_status: _EntityStatus | StatusBase | None = None, unit_status: _EntityStatus | StatusBase | None = None, workload_version: str = '', @@ -1758,7 +1746,7 @@ def __init__( object.__setattr__(self, 'leader', leader) object.__setattr__(self, 'model', model if model is not None else Model()) object.__setattr__(self, 'planned_units', planned_units) - object.__setattr__(self, 'deferred', list(deferred) if deferred else []) + object.__setattr__(self, 'deferred', list(deferred)) object.__setattr__(self, 'workload_version', workload_version) # Handle status conversion from ops types. @@ -1775,7 +1763,7 @@ def __init__( # Normalise ops.Port to scenario Port. normalised_ports = frozenset( Port(protocol=port.protocol, port=port.port) if isinstance(port, ops.Port) else port - for port in (opened_ports or ()) + for port in opened_ports ) object.__setattr__(self, 'opened_ports', normalised_ports) @@ -1784,7 +1772,7 @@ def __init__( Storage(name=storage.name, index=storage.index) if isinstance(storage, ops.Storage) else storage - for storage in (storages or ()) + for storage in storages ) object.__setattr__(self, 'storages', normalised_storage) @@ -1794,26 +1782,12 @@ def __init__( # ops.Resources does not contain the source of the resource, so cannot be converted. # ops.StoredState is not convenient to initialise with data, so not useful here. - object.__setattr__( - self, 'relations', frozenset(relations) if relations is not None else frozenset() - ) - object.__setattr__( - self, 'networks', frozenset(networks) if networks is not None else frozenset() - ) - object.__setattr__( - self, 'containers', frozenset(containers) if containers is not None else frozenset() - ) - object.__setattr__( - self, 'secrets', frozenset(secrets) if secrets is not None else frozenset() - ) - object.__setattr__( - self, 'resources', frozenset(resources) if resources is not None else frozenset() - ) - object.__setattr__( - self, - 'stored_states', - frozenset(stored_states) if stored_states is not None else frozenset(), - ) + object.__setattr__(self, 'relations', frozenset(relations)) + object.__setattr__(self, 'networks', frozenset(networks)) + object.__setattr__(self, 'containers', frozenset(containers)) + object.__setattr__(self, 'secrets', frozenset(secrets)) + object.__setattr__(self, 'resources', frozenset(resources)) + object.__setattr__(self, 'stored_states', frozenset(stored_states)) def _update_workload_version(self, new_workload_version: str): """Update the current app version and record the previous one.""" diff --git a/testing/tests/test_e2e/test_state.py b/testing/tests/test_e2e/test_state.py index 9ec32f160..f7403105c 100644 --- a/testing/tests/test_e2e/test_state.py +++ b/testing/tests/test_e2e/test_state.py @@ -462,6 +462,53 @@ def test_immutable_content_dict_of_dicts( assert getattr(obj2, attribute) == {0: {'foo': 'bar'}, 1: {'baz': 'qux'}} +@pytest.mark.parametrize( + 'component,attribute,expected_type,input_value,required_args', + [ + # Mapping -> dict + (CloudCredential, 'attributes', dict, {'a': 'b'}, {'auth_type': 'foo'}), + (Secret, 'remote_grants', dict, {1: {'app'}}, {'tracked_content': {'k': 'v'}}), + (Notice, 'last_data', dict, {'k': 'v'}, {'key': 'foo'}), + (Container, 'layers', dict, {}, {'name': 'foo'}), + (Container, 'service_statuses', dict, {}, {'name': 'foo'}), + (Container, 'mounts', dict, {}, {'name': 'foo'}), + (StoredState, 'content', dict, {'k': 'v'}, {}), + # Iterable -> list + (CloudCredential, 'redacted', list, ('a', 'b'), {'auth_type': 'foo'}), + (CloudSpec, 'ca_certificates', list, ('a', 'b'), {'type': 'foo'}), + ( + Network, + 'bind_addresses', + list, + iter([BindAddress([Address('192.0.2.0')])]), + {'binding_name': 'foo'}, + ), + (Network, 'ingress_addresses', list, ('1.2.3.4',), {'binding_name': 'foo'}), + (Network, 'egress_subnets', list, ('1.2.3.0/24',), {'binding_name': 'foo'}), + (Container, 'notices', list, (Notice(key='foo'),), {'name': 'foo'}), + (State, 'deferred', list, (), {}), + # Iterable -> frozenset + (Container, 'execs', frozenset, (), {'name': 'foo'}), + (Container, 'check_infos', frozenset, (), {'name': 'foo'}), + (State, 'relations', frozenset, (Relation(endpoint='foo'),), {}), + (State, 'networks', frozenset, (Network(binding_name='foo'),), {}), + (State, 'containers', frozenset, (Container(name='foo'),), {}), + (State, 'secrets', frozenset, (Secret(tracked_content={'k': 'v'}),), {}), + (State, 'stored_states', frozenset, (), {}), + ], +) +def test_init_converts_to_concrete_type( + component: type[object], + attribute: str, + expected_type: type, + input_value: Any, + required_args: dict[str, Any], +): + """Verify that __init__ converts broader input types to concrete attribute types.""" + obj = component(**required_args, **{attribute: input_value}) + assert isinstance(getattr(obj, attribute), expected_type) + + @pytest.mark.parametrize( 'obj_in,attribute,get_method,key_attr', [ From 2c7c2bedf39b573a0bca34e681e4dd3d57495875 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Fri, 27 Mar 2026 10:15:13 +1300 Subject: [PATCH 04/12] Don't change attribute types to non-frozen ones, because we want to have type checkers alert to mutating the state. --- testing/src/scenario/state.py | 34 +++++++++++++++++----------------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 6b533ce1f..931aef8d7 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -157,7 +157,7 @@ class CloudCredential: # noqa: D101 auth_type: str """Authentication type.""" - attributes: dict[str, str] + attributes: Mapping[str, str] """A dictionary containing cloud credentials. For example, for AWS, it contains `access-key` and `secret-key`; @@ -165,7 +165,7 @@ class CloudCredential: # noqa: D101 can be found here. """ - redacted: list[str] + redacted: Sequence[str] """A list of redacted generic cloud API secrets.""" def __init__( @@ -212,7 +212,7 @@ class CloudSpec: # noqa: D101 credential: CloudCredential | None """Cloud credentials with key-value attributes.""" - ca_certificates: list[str] + ca_certificates: Sequence[str] """A list of CA certificates.""" skip_tls_verify: bool @@ -301,7 +301,7 @@ class Secret: to this unit. """ - remote_grants: dict[int, set[str]] + remote_grants: Mapping[int, set[str]] """Mapping from relation IDs to remote units and applications to which this secret has been granted.""" @@ -473,12 +473,12 @@ class Network: binding_name: str """The name of the network space.""" - bind_addresses: list[BindAddress] + bind_addresses: Sequence[BindAddress] """Addresses that the charm's application should bind to.""" - ingress_addresses: list[str] + ingress_addresses: Sequence[str] """Addresses other applications should use to connect to the unit.""" - egress_subnets: list[str] + egress_subnets: Sequence[str] """Subnets that other units will see the charm connecting from.""" def __init__( @@ -831,7 +831,7 @@ def _generate_new_change_id(): class Exec: """Mock data for simulated :meth:`ops.Container.exec` calls.""" - command_prefix: tuple[str, ...] + command_prefix: Sequence[str, ...] return_code: int """The return code of the process. @@ -941,7 +941,7 @@ class Notice: occurrences: int """The number of times one of these notices has occurred.""" - last_data: dict[str, str] + last_data: Mapping[str, str] """Additional data captured from the last occurrence of one of these notices.""" repeat_after: datetime.timedelta | None @@ -1104,10 +1104,10 @@ class Container: # pebble or derive them from the resulting plan (which one CAN get from pebble). # So if we are instantiating Container by fetching info from a 'live' charm, the 'layers' # will be unknown. all that we can know is the resulting plan (the 'computed plan'). - _base_plan: dict[str, Any] + _base_plan: Mapping[str, Any] # We expect most of the user-facing testing to be covered by this 'layers' attribute, # as it is all that will be known when unit-testing. - layers: dict[str, pebble.Layer] + layers: Mapping[str, pebble.Layer] """All :class:`ops.pebble.Layer` definitions that have already been added to the container. Note that the layers should be added to the dictionary in the order in which they would have @@ -1116,10 +1116,10 @@ class Container: this means adding them in the order of the API calls. """ - service_statuses: dict[str, pebble.ServiceStatus] + service_statuses: Mapping[str, pebble.ServiceStatus] """The current status of each Pebble service running in the container.""" - mounts: dict[str, Mount] + mounts: Mapping[str, Mount] """Provides access to the contents of the simulated container filesystem. For example, suppose you want to express that your container has: @@ -1163,7 +1163,7 @@ class Container: ) """ - notices: list[Notice] + notices: Sequence[Notice] """Any Pebble notices that already exist in the container.""" check_infos: frozenset[CheckInfo] @@ -1473,7 +1473,7 @@ class MyCharm(ops.CharmBase): # However, it's complex to describe those types, since it's a recursive # definition - even in TypeShed the _Marshallable type includes containers # like list[Any], which seems to defeat the point. - content: dict[str, Any] + content: Mapping[str, Any] """The content of the :class:`ops.StoredState` instance.""" _data_type_name: str @@ -1665,7 +1665,7 @@ class State: return data from `State.leader`, and so on. """ - config: dict[str, str | int | float | bool] + config: Mapping[str, str | int | float | bool] """The present configuration of this charm.""" relations: frozenset[RelationBase] """All relations that currently exist for this charm.""" @@ -1709,7 +1709,7 @@ class State: # dispatched, and represent the events that had been deferred during the previous run. # If the charm defers any events during "this execution", they will be appended # to this list. - deferred: list[DeferredEvent] + deferred: Sequence[DeferredEvent] """Events that have been deferred on this charm by some previous execution.""" stored_states: frozenset[StoredState] """Contents of a charm's stored state.""" From 03c83817bdf410afa0a82c039335140752418254 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Fri, 27 Mar 2026 10:21:36 +1300 Subject: [PATCH 05/12] Copy the good bits of #2335. --- testing/src/scenario/state.py | 18 +++++++++--------- testing/tests/test_e2e/test_state.py | 14 +++++++++++++- 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 931aef8d7..a9125122e 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -46,7 +46,7 @@ from . import Context class _StateKwargs(TypedDict, total=False): - config: dict[str, str | int | float | bool] + config: Mapping[str, str | int | float | bool] relations: Iterable[RelationBase] networks: Iterable[Network] containers: Iterable[Container] @@ -65,7 +65,7 @@ class _StateKwargs(TypedDict, total=False): AnyJson = str | bool | dict[str, 'AnyJson'] | int | float | list['AnyJson'] -RawSecretRevisionContents = RawDataBagContents = dict[str, str] +RawSecretRevisionContents = RawDataBagContents = Mapping[str, str] UnitID = int CharmType = TypeVar('CharmType', bound=CharmBase) @@ -644,7 +644,7 @@ class Relation(RelationBase): remote_app_data: RawDataBagContents = dataclasses.field(default_factory=dict) """The current content of the application databag.""" - remote_units_data: dict[UnitID, RawDataBagContents] = dataclasses.field( + remote_units_data: Mapping[UnitID, RawDataBagContents] = dataclasses.field( default_factory=lambda: {0: _DEFAULT_JUJU_DATABAG.copy()}, # dedup ) """The current content of the databag for each unit in the relation.""" @@ -729,7 +729,7 @@ def remote_unit_name(self) -> str: class PeerRelation(RelationBase): """A relation to share data between units of the charm.""" - peers_data: dict[UnitID, RawDataBagContents] = dataclasses.field(default_factory=dict) + peers_data: Mapping[UnitID, RawDataBagContents] = dataclasses.field(default_factory=dict) """Current contents of the peer databags. Note that this does not include data for the unit being tested. Data for @@ -1725,7 +1725,7 @@ class State: def __init__( self, *, - config: dict[str, str | int | float | bool] | None = None, + config: Mapping[str, str | int | float | bool] | None = None, relations: Iterable[RelationBase] = (), networks: Iterable[Network] = (), containers: Iterable[Container] = (), @@ -1903,7 +1903,7 @@ def from_context( ctx: Context[CharmType], *, # If provided, these merge with or replace the generated versions. - config: dict[str, str | int | float | bool] | None = None, + config: Mapping[str, str | int | float | bool] | None = None, relations: Iterable[RelationBase] | None = None, containers: Iterable[Container] | None = None, storages: Iterable[Storage] | None = None, @@ -2004,9 +2004,9 @@ class _CharmSpec(Generic[CharmType]): """Charm spec.""" charm_type: type[CharmBase] - meta: dict[str, Any] - actions: dict[str, Any] | None = None - config: dict[str, Any] | None = None + meta: Mapping[str, Any] + actions: Mapping[str, Any] | None = None + config: Mapping[str, Any] | None = None # autoloaded means: we are running a 'real' charm class, living in some # /src/charm.py, and the metadata files are 'real' metadata files. diff --git a/testing/tests/test_e2e/test_state.py b/testing/tests/test_e2e/test_state.py index f7403105c..94fee4b78 100644 --- a/testing/tests/test_e2e/test_state.py +++ b/testing/tests/test_e2e/test_state.py @@ -371,7 +371,6 @@ def test_replace_state(): (CloudCredential, 'attributes', {'auth_type': 'foo'}), (Secret, 'tracked_content', {}), (Secret, 'latest_content', {'tracked_content': {'password': 'password'}}), - (Secret, 'remote_grants', {'tracked_content': {'password': 'password'}}), (Relation, 'local_app_data', {'endpoint': 'foo'}), (Relation, 'local_unit_data', {'endpoint': 'foo'}), (Relation, 'remote_app_data', {'endpoint': 'foo'}), @@ -405,6 +404,19 @@ def test_immutable_content_dict( assert getattr(obj2, attribute) == {'foo': 'bar'} +def test_immutable_remote_grants(): + content = {0: {'app1', 'app2'}} + obj1 = Secret(tracked_content={'password': 'password'}, remote_grants=content) + obj2 = Secret(tracked_content={'password': 'password'}, remote_grants=content) + assert obj1.remote_grants == obj2.remote_grants == {0: frozenset({'app1', 'app2'})} + assert obj1.remote_grants is not obj2.remote_grants + content[1] = {'app3'} + assert obj1.remote_grants == obj2.remote_grants == {0: frozenset({'app1', 'app2'})} + object.__setattr__(obj1, 'remote_grants', {1: frozenset({'app3'})}) + assert obj1.remote_grants == {1: frozenset({'app3'})} + assert obj2.remote_grants == {0: frozenset({'app1', 'app2'})} + + @pytest.mark.parametrize( 'component,attribute,required_args', [ From cb4c58525f61995660381cbc9d323847730620d9 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Tue, 31 Mar 2026 13:07:20 +1300 Subject: [PATCH 06/12] Apply suggestion from @tonyandrewmeyer --- testing/src/scenario/state.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 59ba99d13..2a47384e0 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -305,7 +305,7 @@ class Secret: to this unit. """ - remote_grants: Mapping[int, set[str]] + remote_grants: Mapping[int, frozenset[str]] """Mapping from relation IDs to remote units and applications to which this secret has been granted.""" From dd1dd96a72d9b61a6e852bf41d8a24479f5a4a9e Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Tue, 31 Mar 2026 13:07:57 +1300 Subject: [PATCH 07/12] Apply suggestion from @james-garner-canonical Co-authored-by: James Garner --- testing/src/scenario/state.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 2a47384e0..47575d6d9 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -334,7 +334,7 @@ def __init__( latest_content: RawSecretRevisionContents | None = None, id: str | None = None, owner: Literal['unit', 'app'] | None = None, - remote_grants: Mapping[int, set[str]] = {}, + remote_grants: Mapping[int, Iterable[str]] = {}, label: str | None = None, description: str | None = None, expire: datetime.datetime | None = None, From 6dd569764943f09ee7d16b7a97f9b9c1ba17ef15 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Tue, 31 Mar 2026 13:09:54 +1300 Subject: [PATCH 08/12] Apply suggestion from @tonyandrewmeyer --- testing/src/scenario/state.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 47575d6d9..c1399f938 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -351,7 +351,7 @@ def __init__( ) object.__setattr__(self, 'id', id if id is not None else _generate_secret_id()) object.__setattr__(self, 'owner', owner) - object.__setattr__(self, 'remote_grants', dict(remote_grants)) + object.__setattr__(self, 'remote_grants', {k: frozenset(v) for k, v in remote_grants.items()}) object.__setattr__(self, 'label', label) object.__setattr__(self, 'description', description) object.__setattr__(self, 'expire', expire) From eb8247056f4b8821482785045af5fd4eb55a797b Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Tue, 31 Mar 2026 13:12:45 +1300 Subject: [PATCH 09/12] Apply suggestion from @tonyandrewmeyer --- testing/src/scenario/state.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index c1399f938..cf94a6ef2 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -856,7 +856,7 @@ def _generate_new_change_id(): class Exec: """Mock data for simulated :meth:`ops.Container.exec` calls.""" - command_prefix: Sequence[str, ...] + command_prefix: Sequence[str] return_code: int """The return code of the process. From d4ee1f798cbd48a4f5408d0ad28b10de9934a36c Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Tue, 14 Apr 2026 15:35:11 +1200 Subject: [PATCH 10/12] Address review comments. --- testing/src/scenario/state.py | 66 ++++++++++------------------ testing/tests/test_e2e/test_state.py | 50 ++++++++++----------- 2 files changed, 47 insertions(+), 69 deletions(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index cf94a6ef2..5ab7afc17 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -127,19 +127,6 @@ class _StateKwargs(TypedDict, total=False): } -def _deepcopy_mutable_fields(obj: object): - # We don't have a frozendict to freeze any dictionaries, and while we - # could freeze a list into a tuple, we would break tests that currently - # assert that they get a list, so we can't do that until 8.0. That means - # we can't actually freeze the content here, but we can at least - # disassociate it from the original object. - for attr, value in obj.__dict__.items(): - if isinstance(value, (dict, list)): - # We expect that the obj is a frozen dataclass, so have to use - # object.__setattr__ to bypass the frozen check. - object.__setattr__(obj, attr, copy.deepcopy(value)) - - # A lot of JujuLogLine objects are created, so we want them to be fast and light. # Dataclasses define __slots__, so are small, and a namedtuple is actually # slower to create than a dataclass. A plain dictionary (or TypedDict) would be @@ -343,22 +330,23 @@ def __init__( self._validate_content(tracked_content, 'tracked_content') if latest_content is not None: self._validate_content(latest_content, 'latest_content') - object.__setattr__(self, 'tracked_content', tracked_content) + object.__setattr__(self, 'tracked_content', dict(tracked_content)) object.__setattr__( self, 'latest_content', - latest_content if latest_content is not None else tracked_content, + dict(latest_content) if latest_content is not None else dict(tracked_content), ) object.__setattr__(self, 'id', id if id is not None else _generate_secret_id()) object.__setattr__(self, 'owner', owner) - object.__setattr__(self, 'remote_grants', {k: frozenset(v) for k, v in remote_grants.items()}) + object.__setattr__( + self, 'remote_grants', {k: frozenset(v) for k, v in remote_grants.items()} + ) object.__setattr__(self, 'label', label) object.__setattr__(self, 'description', description) object.__setattr__(self, 'expire', expire) object.__setattr__(self, 'rotate', rotate) object.__setattr__(self, '_tracked_revision', 1) object.__setattr__(self, '_latest_revision', 1) - _deepcopy_mutable_fields(self) def __hash__(self) -> int: return hash(self.id) @@ -462,7 +450,8 @@ class BindAddress: """The MAC address of the interface.""" def __post_init__(self): - _deepcopy_mutable_fields(self) + # Stored as list for backwards compatibility. + object.__setattr__(self, 'addresses', list(self.addresses)) def _hook_tool_output_fmt(self): """Dumps itself to dict in the same format the hook command would.""" @@ -509,29 +498,15 @@ class Network: def __init__( self, binding_name: str, - bind_addresses: Iterable[BindAddress] | None = None, + bind_addresses: Iterable[BindAddress] = (BindAddress([Address('192.0.2.0')]),), *, - ingress_addresses: Iterable[str] | None = None, - egress_subnets: Iterable[str] | None = None, + ingress_addresses: Iterable[str] = ('192.0.2.0',), + egress_subnets: Iterable[str] = ('192.0.2.0/24',), ): object.__setattr__(self, 'binding_name', binding_name) - object.__setattr__( - self, - 'bind_addresses', - list(bind_addresses) - if bind_addresses is not None - else [BindAddress([Address('192.0.2.0')])], - ) - object.__setattr__( - self, - 'ingress_addresses', - list(ingress_addresses) if ingress_addresses is not None else ['192.0.2.0'], - ) - object.__setattr__( - self, - 'egress_subnets', - list(egress_subnets) if egress_subnets is not None else ['192.0.2.0/24'], - ) + object.__setattr__(self, 'bind_addresses', list(bind_addresses)) + object.__setattr__(self, 'ingress_addresses', list(ingress_addresses)) + object.__setattr__(self, 'egress_subnets', list(egress_subnets)) def __hash__(self) -> int: return hash(self.binding_name) @@ -631,7 +606,10 @@ def __post_init__(self): for databag in self._databags: self._validate_databag(databag) - _deepcopy_mutable_fields(self) + # Deepcopy mutable fields to disassociate from the caller's objects. + for attr, value in self.__dict__.items(): + if isinstance(value, (dict, list)): + object.__setattr__(self, attr, copy.deepcopy(value)) def __hash__(self) -> int: return hash(self.id) @@ -888,6 +866,7 @@ def __init__( stderr: str = '', _change_id: int | None = None, ): + # Stored as tuple (rather than list) for hashability. object.__setattr__(self, 'command_prefix', tuple(command_prefix)) object.__setattr__(self, 'return_code', return_code) object.__setattr__(self, 'stdout', stdout) @@ -1209,14 +1188,14 @@ def __init__( ): object.__setattr__(self, 'name', name) object.__setattr__(self, 'can_connect', can_connect) - object.__setattr__(self, '_base_plan', dict(_base_plan)) - object.__setattr__(self, 'layers', dict(layers)) + object.__setattr__(self, '_base_plan', copy.deepcopy(dict(_base_plan))) + object.__setattr__(self, 'layers', copy.deepcopy(dict(layers))) object.__setattr__(self, 'service_statuses', dict(service_statuses)) object.__setattr__(self, 'mounts', dict(mounts)) object.__setattr__(self, 'execs', frozenset(execs)) + # Stored as list for backwards compatibility. object.__setattr__(self, 'notices', list(notices)) object.__setattr__(self, 'check_infos', frozenset(check_infos)) - _deepcopy_mutable_fields(self) def __hash__(self) -> int: return hash(self.name) @@ -1516,9 +1495,8 @@ def __init__( ): object.__setattr__(self, 'name', name) object.__setattr__(self, 'owner_path', owner_path) - object.__setattr__(self, 'content', dict(content)) + object.__setattr__(self, 'content', copy.deepcopy(content)) object.__setattr__(self, '_data_type_name', _data_type_name) - _deepcopy_mutable_fields(self) @property def _handle_path(self): diff --git a/testing/tests/test_e2e/test_state.py b/testing/tests/test_e2e/test_state.py index 99e2cdfcb..9783d0578 100644 --- a/testing/tests/test_e2e/test_state.py +++ b/testing/tests/test_e2e/test_state.py @@ -477,46 +477,46 @@ def test_immutable_content_dict_of_dicts( @pytest.mark.parametrize( - 'component,attribute,expected_type,input_value,required_args', + 'component,required_args,attribute,input_value,expected_type', [ # Mapping -> dict - (CloudCredential, 'attributes', dict, {'a': 'b'}, {'auth_type': 'foo'}), - (Secret, 'remote_grants', dict, {1: {'app'}}, {'tracked_content': {'k': 'v'}}), - (Notice, 'last_data', dict, {'k': 'v'}, {'key': 'foo'}), - (Container, 'layers', dict, {}, {'name': 'foo'}), - (Container, 'service_statuses', dict, {}, {'name': 'foo'}), - (Container, 'mounts', dict, {}, {'name': 'foo'}), - (StoredState, 'content', dict, {'k': 'v'}, {}), + (CloudCredential, {'auth_type': 'foo'}, 'attributes', {'a': 'b'}, dict), + (Secret, {'tracked_content': {'k': 'v'}}, 'remote_grants', {1: {'app'}}, dict), + (Notice, {'key': 'foo'}, 'last_data', {'k': 'v'}, dict), + (Container, {'name': 'foo'}, 'layers', {}, dict), + (Container, {'name': 'foo'}, 'service_statuses', {}, dict), + (Container, {'name': 'foo'}, 'mounts', {}, dict), + (StoredState, {}, 'content', {'k': 'v'}, dict), # Iterable -> list - (CloudCredential, 'redacted', list, ('a', 'b'), {'auth_type': 'foo'}), - (CloudSpec, 'ca_certificates', list, ('a', 'b'), {'type': 'foo'}), + (CloudCredential, {'auth_type': 'foo'}, 'redacted', ('a', 'b'), list), + (CloudSpec, {'type': 'foo'}, 'ca_certificates', ('a', 'b'), list), ( Network, + {'binding_name': 'foo'}, 'bind_addresses', - list, iter([BindAddress([Address('192.0.2.0')])]), - {'binding_name': 'foo'}, + list, ), - (Network, 'ingress_addresses', list, ('1.2.3.4',), {'binding_name': 'foo'}), - (Network, 'egress_subnets', list, ('1.2.3.0/24',), {'binding_name': 'foo'}), - (Container, 'notices', list, (Notice(key='foo'),), {'name': 'foo'}), - (State, 'deferred', list, (), {}), + (Network, {'binding_name': 'foo'}, 'ingress_addresses', ('1.2.3.4',), list), + (Network, {'binding_name': 'foo'}, 'egress_subnets', ('1.2.3.0/24',), list), + (Container, {'name': 'foo'}, 'notices', (Notice(key='foo'),), list), + (State, {}, 'deferred', (), list), # Iterable -> frozenset - (Container, 'execs', frozenset, (), {'name': 'foo'}), - (Container, 'check_infos', frozenset, (), {'name': 'foo'}), - (State, 'relations', frozenset, (Relation(endpoint='foo'),), {}), - (State, 'networks', frozenset, (Network(binding_name='foo'),), {}), - (State, 'containers', frozenset, (Container(name='foo'),), {}), - (State, 'secrets', frozenset, (Secret(tracked_content={'k': 'v'}),), {}), - (State, 'stored_states', frozenset, (), {}), + (Container, {'name': 'foo'}, 'execs', (), frozenset), + (Container, {'name': 'foo'}, 'check_infos', (), frozenset), + (State, {}, 'relations', (Relation(endpoint='foo'),), frozenset), + (State, {}, 'networks', (Network(binding_name='foo'),), frozenset), + (State, {}, 'containers', (Container(name='foo'),), frozenset), + (State, {}, 'secrets', (Secret(tracked_content={'k': 'v'}),), frozenset), + (State, {}, 'stored_states', (), frozenset), ], ) def test_init_converts_to_concrete_type( component: type[object], + required_args: dict[str, Any], attribute: str, - expected_type: type, input_value: Any, - required_args: dict[str, Any], + expected_type: type, ): """Verify that __init__ converts broader input types to concrete attribute types.""" obj = component(**required_args, **{attribute: input_value}) From 3ec1bd50a4f30f4a64b9c94d4ff7b91907565a87 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Sun, 7 Jun 2026 21:25:49 +1200 Subject: [PATCH 11/12] Reject bare str for iterable-of-str arguments. Add a _list_of_str helper that raises StateValidationError when a bare str is passed where an iterable of strings is expected, and apply it to redacted, ca_certificates, ingress_addresses, egress_subnets, and the values in Secret.remote_grants. --- testing/src/scenario/state.py | 34 ++++++++++++++++++++++++---- testing/tests/test_e2e/test_state.py | 20 ++++++++++++++++ 2 files changed, 49 insertions(+), 5 deletions(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index 5ab7afc17..a233025ad 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -127,6 +127,21 @@ class _StateKwargs(TypedDict, total=False): } +def _list_of_str(value: Iterable[str], name: str) -> list[str]: + """Convert an iterable of strings to a list, rejecting a bare ``str``. + + The type hints say ``Iterable[str]``, but a bare string also satisfies that, + and silently iterating it character-by-character is never what the caller + intended. + """ + if isinstance(value, str): + raise StateValidationError( + f'{name} must be an iterable of strings, not a single string; ' + f'did you mean [{value!r}]?', + ) + return list(value) + + # A lot of JujuLogLine objects are created, so we want them to be fast and light. # Dataclasses define __slots__, so are small, and a namedtuple is actually # slower to create than a dataclass. A plain dictionary (or TypedDict) would be @@ -168,7 +183,7 @@ def __init__( ): object.__setattr__(self, 'auth_type', auth_type) object.__setattr__(self, 'attributes', dict(attributes)) - object.__setattr__(self, 'redacted', list(redacted)) + object.__setattr__(self, 'redacted', _list_of_str(redacted, 'redacted')) def _to_ops(self) -> CloudCredential_Ops: return CloudCredential_Ops( @@ -233,7 +248,9 @@ def __init__( object.__setattr__(self, 'identity_endpoint', identity_endpoint) object.__setattr__(self, 'storage_endpoint', storage_endpoint) object.__setattr__(self, 'credential', credential) - object.__setattr__(self, 'ca_certificates', list(ca_certificates)) + object.__setattr__( + self, 'ca_certificates', _list_of_str(ca_certificates, 'ca_certificates') + ) object.__setattr__(self, 'skip_tls_verify', skip_tls_verify) object.__setattr__(self, 'is_controller_cloud', is_controller_cloud) @@ -339,7 +356,12 @@ def __init__( object.__setattr__(self, 'id', id if id is not None else _generate_secret_id()) object.__setattr__(self, 'owner', owner) object.__setattr__( - self, 'remote_grants', {k: frozenset(v) for k, v in remote_grants.items()} + self, + 'remote_grants', + { + k: frozenset(_list_of_str(v, f'remote_grants[{k!r}]')) + for k, v in remote_grants.items() + }, ) object.__setattr__(self, 'label', label) object.__setattr__(self, 'description', description) @@ -505,8 +527,10 @@ def __init__( ): object.__setattr__(self, 'binding_name', binding_name) object.__setattr__(self, 'bind_addresses', list(bind_addresses)) - object.__setattr__(self, 'ingress_addresses', list(ingress_addresses)) - object.__setattr__(self, 'egress_subnets', list(egress_subnets)) + object.__setattr__( + self, 'ingress_addresses', _list_of_str(ingress_addresses, 'ingress_addresses') + ) + object.__setattr__(self, 'egress_subnets', _list_of_str(egress_subnets, 'egress_subnets')) def __hash__(self) -> int: return hash(self.binding_name) diff --git a/testing/tests/test_e2e/test_state.py b/testing/tests/test_e2e/test_state.py index 9783d0578..32fc991ed 100644 --- a/testing/tests/test_e2e/test_state.py +++ b/testing/tests/test_e2e/test_state.py @@ -12,6 +12,7 @@ import pytest import yaml from scenario.context import Context +from scenario.errors import StateValidationError from scenario.state import ( _DEFAULT_JUJU_DATABAG, Address, @@ -448,6 +449,25 @@ def test_immutable_content_list( assert getattr(obj2, attribute) == ['foo', 'bar'] +@pytest.mark.parametrize( + 'component,required_args,attribute', + [ + (CloudCredential, {'auth_type': 'foo'}, 'redacted'), + (CloudSpec, {'type': 'foo'}, 'ca_certificates'), + (Network, {'binding_name': 'foo'}, 'ingress_addresses'), + (Network, {'binding_name': 'foo'}, 'egress_subnets'), + ], +) +def test_bare_str_rejected(component: type[object], required_args: dict[str, Any], attribute: str): + with pytest.raises(StateValidationError): + component(**required_args, **{attribute: 'oops'}) + + +def test_bare_str_rejected_in_remote_grants(): + with pytest.raises(StateValidationError): + Secret(tracked_content={'k': 'v'}, remote_grants={0: 'app'}) + + @pytest.mark.parametrize( 'component,attribute,required_args', [ From a6ae3e88ed3671a7fcff16a13b8edb08929086f6 Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Sun, 7 Jun 2026 21:37:58 +1200 Subject: [PATCH 12/12] Address review feedback: comments, bytes guard, config copy. - _list_of_str now also rejects bytes (which iterate as ints). - State.config is now copied so the caller's dict isn't aliased. - Comment Secret._tracked_revision/_latest_revision: class attrs, not fields, deliberately excluded from __eq__. - Comment RelationBase: still uses dataclass-generated __init__, asymmetry with the rest of the module is deliberate. - Comment Container's mixed deepcopy/shallow-copy choices. - Comment why Exec._change_id, Container._base_plan, and StoredState._data_type_name remain in __init__ despite being private. --- testing/src/scenario/state.py | 39 +++++++++++++++++++++++++++++------ 1 file changed, 33 insertions(+), 6 deletions(-) diff --git a/testing/src/scenario/state.py b/testing/src/scenario/state.py index a233025ad..48b35030d 100644 --- a/testing/src/scenario/state.py +++ b/testing/src/scenario/state.py @@ -128,13 +128,13 @@ class _StateKwargs(TypedDict, total=False): def _list_of_str(value: Iterable[str], name: str) -> list[str]: - """Convert an iterable of strings to a list, rejecting a bare ``str``. + """Convert an iterable of strings to a list, rejecting a bare ``str`` or ``bytes``. - The type hints say ``Iterable[str]``, but a bare string also satisfies that, - and silently iterating it character-by-character is never what the caller - intended. + The type hints say ``Iterable[str]``, but a bare string also satisfies + that. Iterating a ``str`` character-by-character (or ``bytes`` byte-by-byte) + is never what the caller intended, so reject both up front. """ - if isinstance(value, str): + if isinstance(value, (str, bytes)): raise StateValidationError( f'{name} must be an iterable of strings, not a single string; ' f'did you mean [{value!r}]?', @@ -325,6 +325,11 @@ class Secret: rotate: SecretRotate | None """The rotation policy for the secret.""" + # These are deliberately class attributes (no annotation) rather than + # dataclass fields, so that they're not part of the generated ``__eq__`` and + # users cannot pass them to ``__init__``. They're mutated in place by + # internal code via ``object.__setattr__``. + # what revision is currently tracked by this charm. Only meaningful if owner=False _tracked_revision = 1 @@ -560,6 +565,13 @@ def _next_relation_id(*, update: bool = True): return cur +# Unlike the other state classes in this module, ``RelationBase`` (and its +# subclasses) still uses the dataclass-generated ``__init__``. This means the +# init signature and the attribute type are the same annotation, so we can't +# advertise "accepts ``Mapping``, attribute is ``dict``" the way the other +# classes do. Rewriting these by hand would be a larger change for limited gain, +# so the asymmetry is deliberate; revisit if the maintenance burden of these +# classes ever changes. @dataclasses.dataclass(frozen=True) class RelationBase: """Base class for the various types of relation. @@ -888,6 +900,9 @@ def __init__( return_code: int = 0, stdout: str = '', stderr: str = '', + # ``_change_id`` is private: users are not expected to pass it, but it + # stays in ``__init__`` so internal code can construct an Exec with a + # specific change ID (for example, when round-tripping through pebble). _change_id: int | None = None, ): # Stored as tuple (rather than list) for hashability. @@ -1202,6 +1217,10 @@ def __init__( name: str, *, can_connect: bool = False, + # ``_base_plan`` is private: users should normally configure the + # container through ``layers``. It stays in ``__init__`` so internal + # code can seed a base plan when building a Container from a live + # charm. _base_plan: Mapping[str, Any] = {}, layers: Mapping[str, pebble.Layer] = {}, service_statuses: Mapping[str, pebble.ServiceStatus] = {}, @@ -1212,8 +1231,12 @@ def __init__( ): object.__setattr__(self, 'name', name) object.__setattr__(self, 'can_connect', can_connect) + # _base_plan values are arbitrary JSON-ish data, and pebble.Layer is not + # frozen, so deepcopy to disassociate from the caller's objects. object.__setattr__(self, '_base_plan', copy.deepcopy(dict(_base_plan))) object.__setattr__(self, 'layers', copy.deepcopy(dict(layers))) + # ServiceStatus is an enum and Mount is a frozen dataclass, so a shallow + # copy of the mapping is enough. object.__setattr__(self, 'service_statuses', dict(service_statuses)) object.__setattr__(self, 'mounts', dict(mounts)) object.__setattr__(self, 'execs', frozenset(execs)) @@ -1515,6 +1538,10 @@ def __init__( *, owner_path: str | None = None, content: Mapping[str, Any] = {}, + # ``_data_type_name`` is private: users should leave it at the default, + # which matches the name ops uses for ``StoredStateData``. It stays in + # ``__init__`` so charms that subclass ``StoredStateData`` can pass the + # subclass name. _data_type_name: str = 'StoredStateData', ): object.__setattr__(self, 'name', name) @@ -1772,7 +1799,7 @@ def __init__( unit_status: _EntityStatus | StatusBase | None = None, workload_version: str = '', ): - object.__setattr__(self, 'config', config if config is not None else {}) + object.__setattr__(self, 'config', dict(config) if config is not None else {}) object.__setattr__(self, 'leader', leader) object.__setattr__(self, 'model', model if model is not None else Model()) object.__setattr__(self, 'planned_units', planned_units)