diff --git a/.github/update-published-charms-tests-workflow.py b/.github/update-published-charms-tests-workflow.py index f7b76edc9..c6b1a8e4d 100755 --- a/.github/update-published-charms-tests-workflow.py +++ b/.github/update-published-charms-tests-workflow.py @@ -105,7 +105,7 @@ def get_packages_from_charmhub(): """Get the list of published charms from Charmhub.""" logger.info('Fetching the list of published charms') url = 'https://charmhub.io/packages.json' - with urllib.request.urlopen(url, timeout=120) as response: # noqa: S310 (unsafe URL) + with urllib.request.urlopen(url, timeout=120) as response: data = response.read().decode() packages = json.loads(data)['packages'] return packages diff --git a/AGENTS.md b/AGENTS.md index 6cefba86f..2225d7583 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -59,10 +59,12 @@ import ops import subprocess from typing import Generator # typing is an exception + class MyCharm(ops.CharmBase): def handler(self, event: ops.PebbleReadyEvent): subprocess.run(['echo', 'hello']) + # DON'T: Import objects directly from ops import CharmBase, PebbleReadyEvent # Avoid this ``` diff --git a/README.md b/README.md index ff1590062..b8e206bb7 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,7 @@ Things to note: ```python import ops + class OpsExampleCharm(ops.CharmBase): """Charm the service.""" @@ -78,7 +79,7 @@ class OpsExampleCharm(ops.CharmBase): # Get a reference the container attribute on the PebbleReadyEvent container = event.workload # Add initial Pebble config layer using the Pebble API - container.add_layer("httpbin", self._pebble_layer, combine=True) + container.add_layer('httpbin', self._pebble_layer, combine=True) # Make Pebble reevaluate its plan, ensuring any services are started if enabled. container.replan() # Learn more about statuses at @@ -100,7 +101,7 @@ from charm import OpsExampleCharm def test_httpbin_pebble_ready(): # Arrange: ctx = testing.Context(OpsExampleCharm) - container = testing.Container("httpbin", can_connect=True) + container = testing.Container('httpbin', can_connect=True) state_in = testing.State(containers={container}) # Act: @@ -109,19 +110,19 @@ def test_httpbin_pebble_ready(): # Assert: updated_plan = state_out.get_container(container.name).plan expected_plan = { - "services": { - "httpbin": { - "override": "replace", - "summary": "httpbin", - "command": "gunicorn -b 0.0.0.0:80 httpbin:app -k gevent", - "startup": "enabled", - "environment": {"GUNICORN_CMD_ARGS": "--log-level info"}, + 'services': { + 'httpbin': { + 'override': 'replace', + 'summary': 'httpbin', + 'command': 'gunicorn -b 0.0.0.0:80 httpbin:app -k gevent', + 'startup': 'enabled', + 'environment': {'GUNICORN_CMD_ARGS': '--log-level info'}, } }, } assert expected_plan == updated_plan assert ( - state_out.get_container(container.name).service_statuses["httpbin"] + state_out.get_container(container.name).service_statuses['httpbin'] == ops.pebble.ServiceStatus.ACTIVE ) assert state_out.unit_status == testing.ActiveStatus() diff --git a/STYLE.md b/STYLE.md index 395374436..1dfc3b358 100644 --- a/STYLE.md +++ b/STYLE.md @@ -31,6 +31,7 @@ An exception is names from `typing` -- type annotations get too verbose if these from ops import CharmBase, PebbleReadyEvent from subprocess import run + class MyCharm(CharmBase): def _pebble_ready(self, event: PebbleReadyEvent): run(['echo', 'foo']) @@ -42,12 +43,15 @@ class MyCharm(CharmBase): import ops import subprocess + class MyCharm(ops.CharmBase): def _pebble_ready(self, event: ops.PebbleReadyEvent): subprocess.run(['echo', 'foo']) + # However, "from typing import Foo" is okay to avoid verbosity from typing import Optional, Tuple + counts: Optional[Tuple[str, int]] ``` @@ -136,15 +140,18 @@ Function and method signatures should include a type annotation for the returned ```python def method1(arg1: type1, arg2: type2) -> None: - ... + return + def method2() -> str: ... return 'Hello world!' + class C: def __init__(self, x: type1): - ... + pass + def test_method1(): assert ... diff --git a/docs/explanation/holistic-vs-delta-charms.md b/docs/explanation/holistic-vs-delta-charms.md index 846d7f80a..0e900a94d 100644 --- a/docs/explanation/holistic-vs-delta-charms.md +++ b/docs/explanation/holistic-vs-delta-charms.md @@ -96,12 +96,13 @@ def __init__(self, framework: ops.Framework): self.on.config_changed, self.on['foo-relation'].relation_changed, self.on['bar-relation'].relation_changed, - ... + ..., ] for event in events: framework.observe(event, self._reconcile) + def _reconcile(self, _: ops.EventBase): # Early checks workload_ready = self.workload.is_ready @@ -158,31 +159,35 @@ def __init__(self, framework: ops.Framework): self.framework.observe(self.on.start, self._on_start) hostname = socket.getfqdn() - self.foo_requirer = FooRequirer(self, "foo-relation", address=hostname) + self.foo_requirer = FooRequirer(self, 'foo-relation', address=hostname) self.framework.observe( self.foo_requirer.on.data_available, self._on_data_available, ) - self.bar_provider = BarProvider(self, "bar-relation") + self.bar_provider = BarProvider(self, 'bar-relation') self.framework.observe( self.bar_provider.on.create_bar, self._on_create_bar, ) + def _on_install(self, event: ops.InstallEvent): self.workload.install_binaries() + def _on_start(self, event: ops.StartEvent): self.workload.start_service() # Peer relation is now usable if self.unit.is_leader(): ... + def _on_data_available(self, event: DataAvailableEvent): # Update the workload with event's data self.workload.reconfigure(some_key=event.some_value) + def _on_create_bar(self, event: CreateBarEvent): # Provision a `Bar` resource in the workload self.workload.create_bar(event.some_field) diff --git a/docs/explanation/state-transition-testing.md b/docs/explanation/state-transition-testing.md index 73fba674b..c56f8a490 100644 --- a/docs/explanation/state-transition-testing.md +++ b/docs/explanation/state-transition-testing.md @@ -135,7 +135,9 @@ class MyCharm(ops.CharmBase): def test_status_leader(leader): ctx = testing.Context(MyCharm, meta={'name': 'foo'}) state_out = ctx.run(ctx.on.start(), testing.State(leader=leader)) - assert state_out.unit_status == testing.ActiveStatus('I rule' if leader else 'I am ruled') + assert state_out.unit_status == testing.ActiveStatus( + 'I rule' if leader else 'I am ruled' + ) ``` By defining the right state we can programmatically define what answers will the @@ -266,7 +268,7 @@ td = tempfile.TemporaryDirectory() ctx = testing.Context( charm_type=MyCharmType, meta={'name': 'my-charm-name'}, - charm_root=td.name + charm_root=td.name, ) state = ctx.run(ctx.on.start(), testing.State()) ``` @@ -289,7 +291,9 @@ to manually specify all the metadata the extension adds: ctx = testing.Context(MyFlaskCharm) # The 'ingress' relation is provided by the flask-framework extension. -state = ctx.run(ctx.on.start(), testing.State(relations={testing.Relation('ingress')})) +state = ctx.run( + ctx.on.start(), testing.State(relations={testing.Relation('ingress')}) +) ``` If your `charmcraft.yaml` defines keys that overlap with what the extension diff --git a/docs/howto/debug-your-charm.md b/docs/howto/debug-your-charm.md index e93123bc9..5064d1e0d 100644 --- a/docs/howto/debug-your-charm.md +++ b/docs/howto/debug-your-charm.md @@ -272,7 +272,8 @@ In your charm code, call [](ops.Framework.breakpoint) to define breakpoints that ```python class MyCharm(ops.CharmBase): def _on_config_changed(self, event: ops.ConfigChangedEvent): - self.framework.breakpoint('config-start') # 'config-start' is an arbitrary string you use with `--at` + # 'config-start' is an arbitrary string you use with `--at` + self.framework.breakpoint('config-start') new_val = self.config['setting'] # ... process the new value ... self.framework.breakpoint('config-end') diff --git a/docs/howto/log-from-your-charm.md b/docs/howto/log-from-your-charm.md index f8038b278..feca8dcc7 100644 --- a/docs/howto/log-from-your-charm.md +++ b/docs/howto/log-from-your-charm.md @@ -14,17 +14,19 @@ example: ```python import logging + ... logger = logging.getLogger(__name__) + class HelloOperatorCharm(ops.CharmBase): ... def _on_config_changed(self, _): - current = self.config["thing"] + current = self.config['thing'] if current not in self._stored.things: # Note the use of the logger here: - logger.info("Found a new thing: %r", current) + logger.info('Found a new thing: %r', current) self._stored.things.append(current) ``` diff --git a/docs/howto/manage-actions.md b/docs/howto/manage-actions.md index 0bd090577..20f11a412 100644 --- a/docs/howto/manage-actions.md +++ b/docs/howto/manage-actions.md @@ -52,19 +52,23 @@ class CompressionKind(enum.Enum): BZIP = 'bzip2' XZ = 'xz' + class Compression(pydantic.BaseModel): kind: CompressionKind = pydantic.Field(CompressionKind.BZIP) - quality: int = pydantic.Field(5, description='Compression quality.', ge=0, le=9) + quality: int = pydantic.Field( + 5, description='Compression quality.', ge=0, le=9 + ) + class SnapshotAction(pydantic.BaseModel): """Take a snapshot of the database.""" - filename: str = pydantic.Field(description="The name of the snapshot file.") + filename: str = pydantic.Field(description='The name of the snapshot file.') compression: Compression = pydantic.Field( default_factory=Compression, - description="The type of compression to use.", + description='The type of compression to use.', ) ``` @@ -83,11 +87,11 @@ def _on_snapshot_action(self, event: ops.ActionEvent): """Handle the snapshot action.""" # Fetch the parameters. If the user passes something invalid, this will # fail the action with an appropriate message. - params = event.load_params(SnapshotAction, errors="fail") + params = event.load_params(SnapshotAction, errors='fail') # This might take a while, so let the user know we're working on it. # This is sent back to the Juju user in real-time, and appears in the output # of the `juju run` command. - event.log(f"Generating snapshot into {params.filename}") + event.log(f'Generating snapshot into {params.filename}') # Do the snapshot. success = self.do_snapshot( filename=params.filename, @@ -96,13 +100,15 @@ def _on_snapshot_action(self, event: ops.ActionEvent): ) if not success: # Report to the user that the action has failed. - event.fail("Failed to generate snapshot.") # Ideally, include more details than this! + event.fail( + 'Failed to generate snapshot.' + ) # Ideally, include more details than this! # Note that `fail()` doesn't interrupt code, so is typically followed by a `return`. return # Set the results of the action. - msg = f"Stored snapshot in {params.filename}." + msg = f'Stored snapshot in {params.filename}.' # These will be displayed in the `juju run` output. - event.set_results({"result": msg}) + event.set_results({'result': msg}) ``` > See more: [](ops.ActionEvent.load_params), [](ops.ActionEvent.params), [](ops.ActionEvent.fail), [](ops.ActionEvent.set_results), [](ops.ActionEvent.log) @@ -114,7 +120,11 @@ When a unique ID is needed for the action task - for example, for logging or cre ```python def _on_snapshot(self, event: ops.ActionEvent): temp_filename = f'backup-{event.id}.tar.gz' - logger.info("Using %s as the temporary backup filename in task %s", filename, event.id) + logger.info( + 'Using %s as the temporary backup filename in task %s', + filename, + event.id, + ) self.create_backup(temp_filename) ... ``` @@ -131,10 +141,18 @@ For example: ```python from ops import testing + def test_backup_action(): ctx = testing.Context(MyCharm) - ctx.run(ctx.on.action('snapshot', params={'filename': 'db-snapshot.tar.gz'}), testing.State()) - assert ctx.action_logs == ['Starting snapshot', 'Table1 complete', 'Table2 complete'] + ctx.run( + ctx.on.action('snapshot', params={'filename': 'db-snapshot.tar.gz'}), + testing.State(), + ) + assert ctx.action_logs == [ + 'Starting snapshot', + 'Table1 complete', + 'Table2 complete', + ] assert 'snapshot-size' in ctx.action_results ``` @@ -167,7 +185,9 @@ To verify that an action works correctly against a real Juju instance, write an ```python def test_logger(juju: jubilant.Juju): - action = juju.run("your-app/0", "snapshot", {"filename": "db-snapshot.tar.gz"}) - assert action.status == "completed" - assert action.results["snapshot-size"].isdigit() + action = juju.run( + 'your-app/0', 'snapshot', {'filename': 'db-snapshot.tar.gz'} + ) + assert action.status == 'completed' + assert action.results['snapshot-size'].isdigit() ``` diff --git a/docs/howto/manage-configuration.md b/docs/howto/manage-configuration.md index 46925974a..17845b97f 100644 --- a/docs/howto/manage-configuration.md +++ b/docs/howto/manage-configuration.md @@ -45,7 +45,7 @@ class WikiConfig(pydantic.BaseModel): def validate_name(cls, value): if len(value) < 4: raise ValueError('Name must be at least 4 characters long') - if " " in value: + if ' ' in value: raise ValueError('Name must not contain spaces') return value ``` @@ -99,10 +99,13 @@ To verify that the `config-changed` event validates the port, pass the new confi ```python from ops import testing + def test_short_wiki_name(): ctx = testing.Context(MyCharm) - state_out = ctx.run(ctx.on.config_changed(), testing.State(config={'name': 'ww'})) + state_out = ctx.run( + ctx.on.config_changed(), testing.State(config={'name': 'ww'}) + ) assert isinstance(state_out.unit_status, testing.BlockedStatus) ``` diff --git a/docs/howto/manage-containers/manage-files-in-the-workload-container.md b/docs/howto/manage-containers/manage-files-in-the-workload-container.md index e77396a8e..4d894db9d 100644 --- a/docs/howto/manage-containers/manage-files-in-the-workload-container.md +++ b/docs/howto/manage-containers/manage-files-in-the-workload-container.md @@ -37,7 +37,9 @@ class MyCharm(ops.CharmBase): def __init__(self, framework: ops.Framework): super().__init__(framework) self.container = self.unit.get_container('myapp') - self.myapp_root = pathops.ContainerPath('/etc/myapp', container=self.container) + self.myapp_root = pathops.ContainerPath( + '/etc/myapp', container=self.container + ) # ... ``` diff --git a/docs/howto/manage-containers/manage-pebble-custom-notices.md b/docs/howto/manage-containers/manage-pebble-custom-notices.md index 91c733cfd..675cf1283 100644 --- a/docs/howto/manage-containers/manage-pebble-custom-notices.md +++ b/docs/howto/manage-containers/manage-pebble-custom-notices.md @@ -25,17 +25,21 @@ class PostgresCharm(ops.CharmBase): def __init__(self, framework: ops.Framework): super().__init__(framework) # Note that "db" is the workload container's name - framework.observe(self.on["db"].pebble_custom_notice, self._on_pebble_custom_notice) - - def _on_pebble_custom_notice(self, event: ops.PebbleCustomNoticeEvent) -> None: - if event.notice.key == "canonical.com/postgresql/backup-done": - path = event.notice.last_data["path"] - logger.info("Backup finished, copying %s to the cloud", path) + framework.observe( + self.on['db'].pebble_custom_notice, self._on_pebble_custom_notice + ) + + def _on_pebble_custom_notice( + self, event: ops.PebbleCustomNoticeEvent + ) -> None: + if event.notice.key == 'canonical.com/postgresql/backup-done': + path = event.notice.last_data['path'] + logger.info('Backup finished, copying %s to the cloud', path) f = event.workload.pull(path, encoding=None) - s3_bucket.upload_fileobj(f, "db-backup.sql") + s3_bucket.upload_fileobj(f, 'db-backup.sql') - elif event.notice.key == "canonical.com/postgresql/other-thing": - logger.info("Handling other thing") + elif event.notice.key == 'canonical.com/postgresql/other-thing': + logger.info('Handling other thing') ``` All notice events have a [`notice`](ops.PebbleNoticeEvent.notice) property with the details of the notice recorded. That is used in the example above to switch on the notice `key` and look at its `last_data` (to determine the backup's path). @@ -55,6 +59,7 @@ charm tests could do the following: ```python from ops import testing + @patch('charm.s3_bucket.upload_fileobj') def test_backup_done(upload_fileobj): # Arrange: @@ -64,22 +69,28 @@ def test_backup_done(upload_fileobj): 'canonical.com/postgresql/backup-done', last_data={'path': '/tmp/mydb.sql'}, ) - container = testing.Container('db', can_connect=True, notices=[ - testing.Notice(key='example.com/a', occurrences=10), - testing.Notice(key='example.com/b'), - notice, - ]) + container = testing.Container( + 'db', + can_connect=True, + notices=[ + testing.Notice(key='example.com/a', occurrences=10), + testing.Notice(key='example.com/b'), + notice, + ], + ) root = container.get_filesystem() - (root / "tmp").mkdir() - (root / "tmp" / "mydb.sql").write_text("BACKUP") + (root / 'tmp').mkdir() + (root / 'tmp' / 'mydb.sql').write_text('BACKUP') state_in = testing.State(containers={container}) # Act: - state_out = ctx.run(ctx.on.pebble_custom_notice(container, notice), state_in) + state_out = ctx.run( + ctx.on.pebble_custom_notice(container, notice), state_in + ) # Assert: upload_fileobj.assert_called_once() upload_f, upload_key = upload_fileobj.call_args.args - self.assertEqual(upload_f.read(), b"BACKUP") - self.assertEqual(upload_key, "db-backup.sql") + self.assertEqual(upload_f.read(), b'BACKUP') + self.assertEqual(upload_key, 'db-backup.sql') ``` diff --git a/docs/howto/manage-containers/manage-pebble-health-checks.md b/docs/howto/manage-containers/manage-pebble-health-checks.md index 7d4954a79..bbcb027b6 100644 --- a/docs/howto/manage-containers/manage-pebble-health-checks.md +++ b/docs/howto/manage-containers/manage-pebble-health-checks.md @@ -53,24 +53,29 @@ class PostgresCharm(ops.CharmBase): def __init__(self, framework: ops.Framework): super().__init__(framework) # Note that "db" is the workload container's name - framework.observe(self.on["db"].pebble_check_failed, self._on_pebble_check_failed) - framework.observe(self.on["db"].pebble_check_recovered, self._on_pebble_check_recovered) + framework.observe( + self.on['db'].pebble_check_failed, self._on_pebble_check_failed + ) + framework.observe( + self.on['db'].pebble_check_recovered, + self._on_pebble_check_recovered, + ) def _on_pebble_check_failed(self, event: ops.PebbleCheckFailedEvent): - if event.info.name == "http-test": - logger.warning("The http-test has started failing!") - self.unit.status = ops.ActiveStatus("Degraded functionality ...") + if event.info.name == 'http-test': + logger.warning('The http-test has started failing!') + self.unit.status = ops.ActiveStatus('Degraded functionality ...') - elif event.info == "online": - logger.error("The service is no longer online!") + elif event.info == 'online': + logger.error('The service is no longer online!') def _on_pebble_check_recovered(self, event: ops.PebbleCheckRecoveredEvent): - if event.info.name == "http-test": - logger.warning("The http-test has stopped failing!") + if event.info.name == 'http-test': + logger.warning('The http-test has stopped failing!') self.unit.status = ops.ActiveStatus() - elif event.info == "online": - logger.error("The service is online again!") + elif event.info == 'online': + logger.error('The service is online again!') ``` All check events have an `info` property with the details of the check's current status. Note that, by the time that the charm receives the event, the status of the check may have changed (for example, passed again after failing). If the response to the check failing is light (such as changing the status), then it's fine to rely on the status of the check at the time the event was triggered — there will be a subsequent check-recovered event, and the status will quickly flick back to the correct one. If the response is heavier (such as restarting a service with an adjusted configuration), then the two events should share a common handler and check the current status via the `info` property; for example: @@ -80,11 +85,16 @@ class PostgresCharm(ops.CharmBase): def __init__(self, framework: ops.Framework): super().__init__(framework) # Note that "db" is the workload container's name - framework.observe(self.on["db"].pebble_check_failed, self._on_pebble_check_failed) - framework.observe(self.on["db"].pebble_check_recovered, self._on_pebble_check_recovered) + framework.observe( + self.on['db'].pebble_check_failed, self._on_pebble_check_failed + ) + framework.observe( + self.on['db'].pebble_check_recovered, + self._on_pebble_check_recovered, + ) def _on_pebble_check_failed(self, event: ops.PebbleCheckFailedEvent): - if event.info.name != "up": + if event.info.name != 'up': # For now, we ignore the other tests. return if event.info.status == ops.pebble.CheckStatus.DOWN: diff --git a/docs/howto/manage-containers/manage-pebble-metrics.md b/docs/howto/manage-containers/manage-pebble-metrics.md index ff00311ba..25b179930 100644 --- a/docs/howto/manage-containers/manage-pebble-metrics.md +++ b/docs/howto/manage-containers/manage-pebble-metrics.md @@ -52,6 +52,7 @@ After retrieving the contents of the secret, we'll use the [`replace_identities` ```python from passlib.hash import sha512_crypt + class MyCharm(ops.CharmBase): ... @@ -61,27 +62,29 @@ class MyCharm(ops.CharmBase): # - Stored the secret ID in the 'metrics-secret-id' configuration option if not self.config.get('metrics-secret-id'): return - secret_id = str(self.config["metrics-secret-id"]) + secret_id = str(self.config['metrics-secret-id']) secret = self.model.get_secret(id=secret_id) content = secret.get_content() - self._replace_identities(content["username"], content["password"]) + self._replace_identities(content['username'], content['password']) def _on_secret_changed(self, event: ops.SecretChangedEvent) -> None: if not self.config.get('metrics-secret-id'): return if event.secret.id == self.config['metrics-secret-id']: content = event.secret.peek_content() - self._replace_identities(content["username"], content["password"]) + self._replace_identities(content['username'], content['password']) def _replace_identities(self, username: str, password: str) -> None: identities = { username: ops.pebble.Identity( - access="metrics", - basic=ops.pebble.BasicIdentity(password=sha512_crypt.hash(password)) + access='metrics', + basic=ops.pebble.BasicIdentity( + password=sha512_crypt.hash(password) + ), ), } self.container.pebble.replace_identities(identities) - logger.debug("New metrics username: %s", username) + logger.debug('New metrics username: %s', username) ... ``` diff --git a/docs/howto/manage-containers/manage-the-workload-container.md b/docs/howto/manage-containers/manage-the-workload-container.md index a1d62ec3f..fc768320e 100644 --- a/docs/howto/manage-containers/manage-the-workload-container.md +++ b/docs/howto/manage-containers/manage-the-workload-container.md @@ -100,6 +100,7 @@ When using an OCI-image that is not built specifically for use with Pebble, laye import ops # ... + class PauseCharm(ops.CharmBase): # ... def __init__(self, framework): @@ -107,13 +108,15 @@ class PauseCharm(ops.CharmBase): # Set a friendly name for your charm. This can be used with the Operator # framework to reference the container, add layers, or interact with # providers/consumers easily. - self.name = "pause" + self.name = 'pause' # This event is dynamically determined from the service name # in ops.pebble.Layer # # If you set self.name as above and use it in the layer definition following this # example, the event will be _pebble_ready - framework.observe(self.on.pause_pebble_ready, self._on_pause_pebble_ready) + framework.observe( + self.on.pause_pebble_ready, self._on_pause_pebble_ready + ) # ... def _on_pause_pebble_ready(self, event: ops.PebbleReadyEvent) -> None: @@ -132,20 +135,20 @@ class PauseCharm(ops.CharmBase): def _pause_layer(self) -> ops.pebble.Layer: """Returns Pebble configuration layer for google/pause""" - return ops.pebble.Layer( - { - "summary": "pause layer", - "description": "pebble config layer for google/pause", - "services": { - self.name: { - "override": "replace", - "summary": "pause service", - "command": "/pause", - "startup": "enabled", - } - }, - } - ) + return ops.pebble.Layer({ + 'summary': 'pause layer', + 'description': 'pebble config layer for google/pause', + 'services': { + self.name: { + 'override': 'replace', + 'summary': 'pause service', + 'command': '/pause', + 'startup': 'enabled', + } + }, + }) + + # ... ``` @@ -225,7 +228,7 @@ class MyCharm(ops.CharmBase): ... def _on_config_changed(self, event): - container = self.unit.get_container("main") + container = self.unit.get_container('main') container.replan() plan = container.get_plan() for service in plan.services: @@ -255,18 +258,18 @@ class SnappassTestCharm(ops.CharmBase): ... def _start_snappass(self): - container = self.unit.containers["snappass"] + container = self.unit.containers['snappass'] snappass_layer = { - "services": { - "snappass": { - "override": "replace", - "summary": "snappass service", - "command": "snappass", - "startup": "enabled", + 'services': { + 'snappass': { + 'override': 'replace', + 'summary': 'snappass service', + 'command': 'snappass', + 'startup': 'enabled', } }, } - container.add_layer("snappass", snappass_layer, combine=True) + container.add_layer('snappass', snappass_layer, combine=True) container.replan() self.unit.status = ops.ActiveStatus() ``` @@ -391,7 +394,7 @@ To run simple commands and receive their output, call `Container.exec` to start For example, to back up a PostgreSQL database, you might use `pg_dump`: ```python -process = container.exec(['pg_dump', 'mydb'], timeout=5*60) +process = container.exec(['pg_dump', 'mydb'], timeout=5 * 60) sql, warnings = process.wait_output() if warnings: for line in warnings.splitlines(): @@ -470,8 +473,7 @@ process.wait_output() The simplest way of receiving standard output and standard error is by using the [`ExecProcess.wait_output`](ops.pebble.ExecProcess.wait_output) method as shown below. The simplest way of sending standard input to the program is as a string, using the `stdin` parameter to `exec`. For example: ```python -process = container.exec(['tr', 'a-z', 'A-Z'], - stdin='This is\na test\n') +process = container.exec(['tr', 'a-z', 'A-Z'], stdin='This is\na test\n') stdout, _ = process.wait_output() logger.info('Output: %r', stdout) ``` @@ -481,8 +483,7 @@ By default, input is sent and output is received as Unicode using the UTF-8 enco For example, the following will log `Output: b'\x01\x02'`: ```python -process = container.exec(['cat'], stdin=b'\x01\x02', - encoding=None) +process = container.exec(['cat'], stdin=b'\x01\x02', encoding=None) stdout, _ = process.wait_output() logger.info('Output: %r', stdout) ``` @@ -509,6 +510,7 @@ For advanced uses, you can also perform streaming I/O by reading from and writin ```python process = container.exec(['cat']) + # Thread that sends data to process's stdin def stdin_thread(): try: @@ -518,6 +520,8 @@ def stdin_thread(): time.sleep(1) finally: process.stdin.close() + + threading.Thread(target=stdin_thread).start() # Log from stdout stream as output is received @@ -570,6 +574,7 @@ LS_LL = """ drwxrwxr-x - ubuntu ubuntu 18 jan 12:06 -- lib """ + class MyCharm(ops.CharmBase): def _on_start(self, _): foo = self.unit.get_container('foo') @@ -578,6 +583,7 @@ class MyCharm(ops.CharmBase): stdout, _ = proc.wait_output() assert stdout == LS_LL + def test_pebble_exec(): container = testing.Container( name='foo', @@ -620,10 +626,12 @@ containers, so if the charm were to execute `self.unit.containers`, it would get To give the charm access to some containers, you need to pass them to the input state, like so: ```python -state = testing.State(containers={ - testing.Container(name='foo', can_connect=True), - testing.Container(name='bar', can_connect=False), -}) +state = testing.State( + containers={ + testing.Container(name='foo', can_connect=True), + testing.Container(name='bar', can_connect=False), + } +) ``` In this case, `self.unit.get_container('foo').can_connect()` would return `True`, while for 'bar' it diff --git a/docs/howto/manage-interfaces.md b/docs/howto/manage-interfaces.md index 1653e8662..fc2debae1 100644 --- a/docs/howto/manage-interfaces.md +++ b/docs/howto/manage-interfaces.md @@ -66,17 +66,17 @@ import typing class ProviderUnitData(BaseModel): secret_id: str = Field( - description="Secret ID for the key you need in order to query this unit.", - title="Query key secret ID", - examples=["secret:12312323112313123213"], + description='Secret ID for the key you need in order to query this unit.', + title='Query key secret ID', + examples=['secret:12312323112313123213'], ) class ProviderAppData(BaseModel): api_endpoint: AnyHttpUrl = Field( description="URL to the database's endpoint.", - title="Endpoint API address", - examples=["https://example.com/v1/query"], + title='Endpoint API address', + examples=['https://example.com/v1/query'], ) @@ -87,9 +87,9 @@ class ProviderSchema(DataBagSchema): class RequirerAppData(BaseModel): tables: Json[typing.List[str]] = Field( - description="Tables that the requirer application needs.", - title="Requested tables.", - examples=[["users", "passwords"]], + description='Tables that the requirer application needs.', + title='Requested tables.', + examples=[['users', 'passwords']], ) @@ -254,14 +254,14 @@ def test_nothing_happens_if_remote_empty(): leader=True, relations={ Relation( - endpoint="my-fancy-database", # the name doesn't matter - interface="my_fancy_database", + endpoint='my-fancy-database', # the name doesn't matter + interface='my_fancy_database', ) }, ) ) # WHEN the database charm receives a relation-joined event - state_out = t.run("my-fancy-database-relation-joined") + state_out = t.run('my-fancy-database-relation-joined') # THEN no data is published to the (local) databags t.assert_relation_data_empty() ``` @@ -281,21 +281,21 @@ from scenario import State, Relation def test_contract_happy_path(): # GIVEN that the remote end has requested tables in the right format - tables_json = json.dumps(["users", "passwords"]) + tables_json = json.dumps(['users', 'passwords']) t = Tester( State( leader=True, relations=[ Relation( - endpoint="my-fancy-database", # the name doesn't matter - interface="my_fancy_database", - remote_app_data={"tables": tables_json}, + endpoint='my-fancy-database', # the name doesn't matter + interface='my_fancy_database', + remote_app_data={'tables': tables_json}, ) ], ) ) # WHEN the database charm receives a relation-changed event - state_out = t.run("my-fancy-database-relation-changed") + state_out = t.run('my-fancy-database-relation-changed') # THEN the schema is satisfied (the database charm published all required fields) t.assert_schema_valid() ``` diff --git a/docs/howto/manage-leadership-changes.md b/docs/howto/manage-leadership-changes.md index aa63c509f..9c3104640 100644 --- a/docs/howto/manage-leadership-changes.md +++ b/docs/howto/manage-leadership-changes.md @@ -37,8 +37,8 @@ leadership are needed to guard against non-leaders. For example: ```python if self.unit.is_leader(): - secret = self.model.get_secret(label="my-label") - secret.set_content({"username": "user", "password": "pass"}) + secret = self.model.get_secret(label='my-label') + secret.set_content({'username': 'user', 'password': 'pass'}) ``` Note that Juju guarantees leadership for only 30 seconds after a `leader-elected` @@ -66,9 +66,11 @@ class MyCharm(ops.CharmBase): @pytest.mark.parametrize('leader', (True, False)) def test_status_leader(leader): - ctx = testing.Context(MyCharm, meta={"name": "foo"}) + ctx = testing.Context(MyCharm, meta={'name': 'foo'}) out = ctx.run(ctx.on.start(), testing.State(leader=leader)) - assert out.unit_status == testing.ActiveStatus('I rule' if leader else 'I am ruled') + assert out.unit_status == testing.ActiveStatus( + 'I rule' if leader else 'I am ruled' + ) ``` ## Write integration tests @@ -87,7 +89,7 @@ as expected. For example: ```python def get_leader_unit(juju: jubilant.Juju) -> str | None: """Utility method to get the name of the current leader.""" - for unit_name, unit in juju.status().apps["your-app"].units.items(): + for unit_name, unit in juju.status().apps['your-app'].units.items(): if unit.leader: return unit_name # It's possible that no leader has been elected, diff --git a/docs/howto/manage-libraries.md b/docs/howto/manage-libraries.md index 4f150003f..99b49e9b9 100644 --- a/docs/howto/manage-libraries.md +++ b/docs/howto/manage-libraries.md @@ -9,7 +9,10 @@ In your `src/charm.py` file, observe the custom events that the library provides ```python import ops -from charms.charm_with_lib.v0.database_lib import DatabaseReadyEvent, DatabaseRequirer +from charms.charm_with_lib.v0.database_lib import ( + DatabaseReadyEvent, + DatabaseRequirer, +) class MyCharm(ops.CharmBase): @@ -29,12 +32,15 @@ A unit test for the charm that uses the database library looks like: from ops import testing from charms.charm_with_lib.v0.database_lib import DatabaseRequirer + def test_ready_event(): ctx = testing.Context(MyCharm) secret = testing.Secret({'username': 'admin', 'password': 'admin'}) state_in = testing.State(secrets={secret}) - state_out = ctx.run(ctx.on.custom(DatabaseRequirer, credential_secret=secret), state_in) + state_out = ctx.run( + ctx.on.custom(DatabaseRequirer, credential_secret=secret), state_in + ) assert ... ``` @@ -74,11 +80,14 @@ class DatabaseReadyEvent(ops.EventBase): def restore(self, snapshot: dict[str, Any]): super().restore(snapshot) credential_secret_id = snapshot['credential_secret_id'] - self.credential_secret = self.framework.model.get_secret(id=credential_secret_id) + self.credential_secret = self.framework.model.get_secret( + id=credential_secret_id + ) class DatabaseRequirerEvents(ops.ObjectEvents): """Container for Database Requirer events.""" + ready = ops.EventSource(DatabaseReadyEvent) @@ -87,7 +96,9 @@ class DatabaseRequirer(ops.Object): def __init__(self, charm: ops.CharmBase, relation_name: str): super().__init__(charm, relation_name) - self.framework.observe(charm.on['database'].relation_changed, self._on_db_changed) + self.framework.observe( + charm.on['database'].relation_changed, self._on_db_changed + ) def _on_db_changed(self, event: ops.RelationChangedEvent): if remote_data_is_valid(event.relation): @@ -116,17 +127,23 @@ from lib.charms.my_Charm.v0.my_lib import DatabaseRequirer class MyTestCharm(ops.CharmBase): - META = { - "name": "my-charm" - } + META = {'name': 'my-charm'} + def __init__(self, framework: ops.Framework): super().__init__(framework) self.db = DatabaseRequirer(self, 'my-relation') -@pytest.mark.parametrize('event', ( - 'start', 'install', 'stop', 'remove', 'update-status', #... -)) +@pytest.mark.parametrize( + 'event', + ( + 'start', + 'install', + 'stop', + 'remove', + 'update-status', # ... + ), +) def test_charm_runs(event): """Verify that the charm can create the library object, and doesn't see unexpected events.""" ctx = testing.Context(MyTestCharm, meta=MyTestCharm.META) @@ -153,7 +170,7 @@ from ops import testing from lib.charms.my_charm.v0.my_lib import DatabaseRequirer -@pytest.fixture(params=["foo", "bar"]) +@pytest.fixture(params=['foo', 'bar']) def endpoint(request): return request.param @@ -162,9 +179,8 @@ def endpoint(request): def my_charm_type(endpoint: str): class MyTestCharm(ops.CharmBase): META = { - "name": "my-charm", - "requires": - {endpoint: {"interface": "my_interface"}} + 'name': 'my-charm', + 'requires': {endpoint: {'interface': 'my_interface'}}, } def __init__(self, framework: ops.Framework): @@ -248,43 +264,53 @@ with the `tracing` example: ```python class TransportProtocolType(enum.Enum): """Receiver Type.""" - HTTP = "http" - GRPC = "grpc" + + HTTP = 'http' + GRPC = 'grpc' class ProtocolType(pydantic.BaseModel): """Protocol Type.""" + name: str = pydantic.Field( - description="Receiver protocol name. What protocols are supported (and what they are called) " - "may differ per provider.", - examples=["otlp_grpc", "otlp_http", "tempo_http", "jaeger_thrift_compact"], + description='Receiver protocol name. What protocols are supported (and what they are called) ' + 'may differ per provider.', + examples=[ + 'otlp_grpc', + 'otlp_http', + 'tempo_http', + 'jaeger_thrift_compact', + ], ) type: TransportProtocolType = pydantic.Field( - description="The transport protocol used by this receiver.", - examples=["http", "grpc"], + description='The transport protocol used by this receiver.', + examples=['http', 'grpc'], ) class Receiver(pydantic.BaseModel): """Specification of an active receiver.""" - protocol: ProtocolType = pydantic.Field(description="Receiver protocol name and type.") + + protocol: ProtocolType = pydantic.Field( + description='Receiver protocol name and type.' + ) url: str = pydantic.Field( description="""URL at which the receiver is reachable. If there's an ingress, it would be the external URL. Otherwise, it would be the service's fqdn or internal IP. If the protocol type is grpc, the url will not contain a scheme.""", examples=[ - "http://traefik_address:2331", - "https://traefik_address:2331", - "http://tempo_public_ip:2331", - "https://tempo_public_ip:2331", - "tempo_public_ip:2331", + 'http://traefik_address:2331', + 'https://traefik_address:2331', + 'http://tempo_public_ip:2331', + 'https://tempo_public_ip:2331', + 'tempo_public_ip:2331', ], ) class TracingProviderAppData(pydantic.BaseModel): receivers: list[Receiver] = pydantic.Field( - description="A list of enabled receivers in the form of the protocol they use and their resolvable server url.", + description='A list of enabled receivers in the form of the protocol they use and their resolvable server url.', ) ``` @@ -298,14 +324,17 @@ relation: ```python receiver_protocol_to_transport_protocol: dict[str, TransportProtocolType] = { - "zipkin": TransportProtocolType.HTTP, - "otlp_grpc": TransportProtocolType.GRPC, - "otlp_http": TransportProtocolType.HTTP, - "jaeger_thrift_http": TransportProtocolType.HTTP, - "jaeger_grpc": TransportProtocolType.GRPC, + 'zipkin': TransportProtocolType.HTTP, + 'otlp_grpc': TransportProtocolType.GRPC, + 'otlp_http': TransportProtocolType.HTTP, + 'jaeger_thrift_http': TransportProtocolType.HTTP, + 'jaeger_grpc': TransportProtocolType.GRPC, } -def _publish_provider(self, relation: ops.Relation, receivers: Iterable[tuple[str, str]]): + +def _publish_provider( + self, relation: ops.Relation, receivers: Iterable[tuple[str, str]] +): data = TracingProviderAppData( receivers=[ Receiver( diff --git a/docs/howto/manage-opened-ports.md b/docs/howto/manage-opened-ports.md index d2a9fb6e0..358008191 100644 --- a/docs/howto/manage-opened-ports.md +++ b/docs/howto/manage-opened-ports.md @@ -76,17 +76,17 @@ def test_open_ports(juju: jubilant.Juju): Assert blocked status in case of port 22 and active status for others. """ # Get the public address of the app: - address = juju.status().apps["your-app"].units["your-app/0"].public_address + address = juju.status().apps['your-app'].units['your-app/0'].public_address # Validate that initial port is opened: assert is_port_open(address, 8000) # Set the port to 22 and validate the app goes to blocked status with the port not opened: - juju.config("your-app", {"server-port": "22"}) + juju.config('your-app', {'server-port': '22'}) juju.wait(jubilant.all_blocked) assert not is_port_open(address, 22) # Set the port to 6789 and validate the app goes to active status with the port opened. - juju.config("your-app", {"server-port": "6789"}) + juju.config('your-app', {'server-port': '6789'}) juju.wait(jubuilant.all_active) assert is_port_open(address, 6789) ``` diff --git a/docs/howto/manage-relations.md b/docs/howto/manage-relations.md index 52cd4a424..f98cc6977 100644 --- a/docs/howto/manage-relations.md +++ b/docs/howto/manage-relations.md @@ -95,7 +95,9 @@ For example: ```python class DatabaseProviderAppData(pydantic.BaseModel): - credentials: str | None = pydantic.Field(default=None, description="A Juju secret ID") + credentials: str | None = pydantic.Field( + default=None, description='A Juju secret ID' + ) ``` Now, in the body of the charm definition, define the event handler. In this example, if we are the leader unit, then we create a database and pass the credentials to use it to the charm on the other side via the relation data: @@ -126,7 +128,7 @@ For example: ```python class SMTPProviderUnitData(pydantic.BaseMode): - smtp_credentials: str = pydantic.Field(description="A Juju secret ID") + smtp_credentials: str = pydantic.Field(description='A Juju secret ID') ``` Now, in the body of the charm definition, define the event handler. In this example, a `smtp_credentials` key is set in the unit data with the ID of a secret: @@ -185,7 +187,9 @@ def _update_configuration(self, _: ops.Eventbase): if not secret_id: # The credentials haven't been added to the relation by the remote app yet. return - secret_contents = self.model.get_secret(id=secret_id).get_contents(refresh=True) + secret_contents = self.model.get_secret(id=secret_id).get_contents( + refresh=True + ) self.push_configuration( username=secret['username'], password=secret['password'], @@ -199,7 +203,7 @@ To add data to the relation databag, use the [`.data` attribute](ops.Relation.da ```python def _on_config_changed(self, event: ops.ConfigChangedEvent): if relation := self.model.get_relation('ingress'): - relation.data[self.app]["domain"] = self.config["domain"] + relation.data[self.app]['domain'] = self.config['domain'] ``` To read data from the relation databag, again use the `.data` attribute, selecting the appropriate databag, and then using it as if it were a regular dictionary. @@ -209,7 +213,9 @@ The charm can inspect the contents of the remote unit databags: ```python def _on_database_relation_changed(self, event: ops.RelationChangedEvent): remote_units_databags = { - event.relation.data[unit] for unit in event.relation.units if unit.app is not self.app + event.relation.data[unit] + for unit in event.relation.units + if unit.app is not self.app } ``` @@ -218,7 +224,9 @@ Or the peer unit databags: ```python def _on_database_relation_changed(self, e: ops.RelationChangedEvent): peer_units_databags = { - event.relation.data[unit] for unit in event.relation.units if unit.app is self.app + event.relation.data[unit] + for unit in event.relation.units + if unit.app is self.app } ``` @@ -250,7 +258,9 @@ If the charm does not have permission to do an operation (e.g. because it is not To do clean-up work when a unit in the relation is removed (for example, removing per-unit credentials), have your charm observe the `relation-departed` event. In the `src/charm.py` file, in the `__init__` function of your charm, set up `relation-departed` event observers for the relevant relations and pair those with an event handler. For example: ```python -framework.observe(self.on.smtp_relation_departed, self._on_smtp_relation_departed) +framework.observe( + self.on.smtp_relation_departed, self._on_smtp_relation_departed +) ``` Now, in the body of the charm definition, define the event handler. For example: @@ -296,8 +306,13 @@ from ops import testing ctx = testing.Context(MyCharm) relation = testing.Relation(endpoint='smtp', remote_units_data={1: {}}) state_in = testing.State(relations={relation}) -state_out = ctx.run(ctx.on.relation_joined(relation, remote_unit=1), state=state_in) -assert 'smtp_credentials' in state_out.get_relation(relation.id).remote_units_data[1] +state_out = ctx.run( + ctx.on.relation_joined(relation, remote_unit=1), state=state_in +) +assert ( + 'smtp_credentials' + in state_out.get_relation(relation.id).remote_units_data[1] +) ``` > See more: [](ops.testing.RelationBase) @@ -340,7 +355,7 @@ relation = testing.SubordinateRelation( endpoint='peers', remote_unit_data={'foo': 'bar'}, remote_app_name='zookeeper', - remote_unit_id=42 + remote_unit_id=42, ) relation.remote_unit_name # 'zookeeper/42' ``` @@ -355,9 +370,10 @@ To verify that charm behaves correctly when integrated with another in a real Ju # This assumes that your integration tests already include the standard # build and deploy test. + def test_active_with_another_app(juju: jubilant.Juju): - juju.deploy("another-app") - juju.integrate("your-app:endpoint", "another-app:endpoint") + juju.deploy('another-app') + juju.integrate('your-app:endpoint', 'another-app:endpoint') juju.wait(jubilant.all_active) ``` diff --git a/docs/howto/manage-resources.md b/docs/howto/manage-resources.md index fbeeeaff0..66c6370c1 100644 --- a/docs/howto/manage-resources.md +++ b/docs/howto/manage-resources.md @@ -27,32 +27,33 @@ In your charm's `src/charm.py` you can now use [`Model.resources.fetch(` or standard file operations. For example: ```python - # Prepare each storage instance for use by the workload. - for path in cache_paths: - cache_root = pathops.LocalPath(path) - (cache_root / "uploaded-data").mkdir(exist_ok=True) - (cache_root / "processed-data").mkdir(exist_ok=True) +# Prepare each storage instance for use by the workload. +for path in cache_paths: + cache_root = pathops.LocalPath(path) + (cache_root / 'uploaded-data').mkdir(exist_ok=True) + (cache_root / 'processed-data').mkdir(exist_ok=True) ``` ### Request more storage instances @@ -81,7 +81,7 @@ Your charm will receive a storage-attached event as each additional instance bec To request more instances in charm code, use [](ops.StorageMapping.request). For example: ```python - self.model.storages.request("cache", 2) # Request two more instances. +self.model.storages.request('cache', 2) # Request two more instances. ``` The additional instances won't be available immediately after the call. As with `juju add-storage`, your charm will receive a storage-attached event as each additional instance becomes available. @@ -122,7 +122,7 @@ Juju also mounts the storage instance in the workload container's filesystem, at In your charm's `__init__` method, observe the [storage-attached](ops.StorageAttachedEvent) event: ```python - framework.observe(self.on["cache"].storage_attached, self._update_configuration) +framework.observe(self.on['cache'].storage_attached, self._update_configuration) ``` In this example, we use a holistic event handler called `_update_configuration`. Alternatively, you could use a dedicated handler for the storage-attached event. To learn more about the different approaches, see [](#holistic-vs-delta-charms). @@ -132,20 +132,20 @@ Next, in `_update_configuration`, get the storage instance path in the workload ```python def _update_configuration(self, event: ops.EventBase): """Update the workload configuration.""" - cache = self.model.storages["cache"] + cache = self.model.storages['cache'] if not cache: logger.info("No instance available for storage 'cache'.") return - web_cache_path = self.meta.containers["web"].mounts["cache"].location + web_cache_path = self.meta.containers['web'].mounts['cache'].location # Configure the workload to use the storage instance path (assuming that # the workload container image isn't preconfigured to expect storage at # the location specified in charmcraft.yaml). # For example, provide the storage instance path in the Pebble layer. - web_container = self.unit.get_container("web") + web_container = self.unit.get_container('web') try: web_container.add_layer(...) except ops.pebble.ConnectionError: - logger.info("Workload container is not available.") + logger.info('Workload container is not available.') return web_container.replan() ``` @@ -155,11 +155,11 @@ def _update_configuration(self, event: ops.EventBase): To access the storage instance in charm code, use {external+charmlibs:ref}`pathops ` or standard file operations in the charm container. For example: ```python - # Prepare the storage instance for use by the workload. - charm_cache_path = cache[0].location # Always index 0 in a K8s charm. - charm_cache_root = pathops.LocalPath(charm_cache_path) - (charm_cache_root / "uploaded-data").mkdir(exist_ok=True) - (charm_cache_root / "processed-data").mkdir(exist_ok=True) +# Prepare the storage instance for use by the workload. +charm_cache_path = cache[0].location # Always index 0 in a K8s charm. +charm_cache_root = pathops.LocalPath(charm_cache_path) +(charm_cache_root / 'uploaded-data').mkdir(exist_ok=True) +(charm_cache_root / 'processed-data').mkdir(exist_ok=True) ``` Alternatively, use {external+charmlibs:class}`pathops.ContainerPath` to access `web_cache_path` in the workload container. This approach is more appropriate if you need to reference additional data in the workload container. @@ -169,7 +169,9 @@ Alternatively, use {external+charmlibs:class}`pathops.ContainerPath` to access ` In the `src/charm.py` file, in the `__init__` function of your charm, set up an observer for the detaching event associated with your storage and pair that with an event handler. For example: ```python - framework.observe(self.on["cache"].storage_detaching, self._on_storage_detaching) +framework.observe( + self.on['cache'].storage_detaching, self._on_storage_detaching +) ``` > See more: [](ops.StorageDetachingEvent) @@ -179,7 +181,9 @@ Now, in the body of the charm definition, define the event handler, or adjust an ```python def _on_storage_detaching(self, event: ops.StorageDetachingEvent): """Handle the storage being detached.""" - self.unit.status = ops.ActiveStatus("Caching disabled; provide storage to boost performance") + self.unit.status = ops.ActiveStatus( + 'Caching disabled; provide storage to boost performance' + ) ``` > Examples: [MySQL handling cluster management](https://github.com/canonical/mysql-k8s-operator/blob/4c575b478b7ae2a28b09dde9cade2d3370dd4db6/src/charm.py#L823), [MongoDB updating the set before storage is removed](https://github.com/canonical/mongodb-operator/blob/b33d036173f47c68823e08a9f03189dc534d38dc/src/charm.py#L596) @@ -195,27 +199,27 @@ from ops import testing # Some charm with a 'foo' filesystem-type storage defined in its metadata: ctx = testing.Context(MyCharm) -storage = testing.Storage("foo") +storage = testing.Storage('foo') # Set up storage with some content: -(storage.get_filesystem(ctx) / "myfile.txt").write_text("helloworld") +(storage.get_filesystem(ctx) / 'myfile.txt').write_text('helloworld') with ctx(ctx.on.update_status(), testing.State(storages={storage})) as mgr: - foo = mgr.charm.model.storages["foo"][0] + foo = mgr.charm.model.storages['foo'][0] loc = foo.location - path = loc / "myfile.txt" + path = loc / 'myfile.txt' assert path.exists() - assert path.read_text() == "helloworld" + assert path.read_text() == 'helloworld' - myfile = loc / "path.py" - myfile.write_text("helloworlds") + myfile = loc / 'path.py' + myfile.write_text('helloworlds') state_out = mgr.run() # Verify that the contents are as expected afterwards. assert ( - state_out.get_storage(storage.name).get_filesystem(ctx) / "path.py" -).read_text() == "helloworlds" + state_out.get_storage(storage.name).get_filesystem(ctx) / 'path.py' +).read_text() == 'helloworlds' ``` If a charm requests adding more storage instances while handling some event, you @@ -256,7 +260,7 @@ To verify that adding and removing storage works correctly against a real Juju i ```python def test_storage_attaching(juju: jubilant.Juju): # Add two storage units of 2 gigabyte each to unit 0 of the Kafka app. - juju.cli("add-storage", "kafka/0", "data=2G,2", include_model=True) + juju.cli('add-storage', 'kafka/0', 'data=2G,2', include_model=True) juju.wait(jubilant.all_active) # Assert that the storage is being used appropriately. ``` diff --git a/docs/howto/manage-stored-state.md b/docs/howto/manage-stored-state.md index 9d848ee58..f3d6e2a77 100644 --- a/docs/howto/manage-stored-state.md +++ b/docs/howto/manage-stored-state.md @@ -47,7 +47,6 @@ You then need to use `set_default` to set an initial value; for example: ```python class MyCharm(ops.CharmBase): - _stored = ops.StoredState() def __init__(self, framework): @@ -65,10 +64,11 @@ def _on_start(self, event: ops.StartEvent): if self._stored.expensive_value is None: self._stored.expensive_value = self._calculate_expensive_value() + def _on_install(self, event: ops.InstallEvent): # We can use self._stored.expensive_value here, and it will have the value # set in the start event. - logger.info("Current value: %s", self._stored.expensive_value) + logger.info('Current value: %s', self._stored.expensive_value) ``` > Examples: [Kubernetes-Dashboard stores core settings](https://github.com/charmed-kubernetes/kubernetes-dashboard-operator/blob/03bf0f64d943e39176c804cd796a7a9838bf13ab/src/charm.py#L42) @@ -96,21 +96,25 @@ def test_charm_sets_stored_state(): ctx = testing.Context(MyCharm) state_in = testing.State() state_out = ctx.run(ctx.on.start(), state_in) - ss = state_out.get_stored_state("_stored", owner_path="MyCharm") - assert ss.content["expensive_value"] == 42 + ss = state_out.get_stored_state('_stored', owner_path='MyCharm') + assert ss.content['expensive_value'] == 42 + def test_charm_logs_stored_state(): ctx = testing.Context(MyCharm) - state_in = testing.State(stored_states={ - testing.StoredState( - "_stored", - owner_path="MyCharm", - content={ - 'expensive_value': 42, - }) - }) + state_in = testing.State( + stored_states={ + testing.StoredState( + '_stored', + owner_path='MyCharm', + content={ + 'expensive_value': 42, + }, + ) + } + ) state_out = ctx.run(ctx.on.install(), state_in) - assert ctx.juju_log[0].message == "Current value: 42" + assert ctx.juju_log[0].message == 'Current value: 42' ``` ## Storing state for the lifetime of the application @@ -140,6 +144,7 @@ def _on_start(self, event: ops.StartEvent): peer = self.model.get_relation('charm-peer') peer.data[self.app]['expensive-value'] = self._calculate_expensive_value() + def _on_stop(self, event: ops.StopEvent): peer = self.model.get_relation('charm-peer') logger.info('Value at stop is: %s', peer.data[self.app]['expensive-value']) @@ -175,5 +180,5 @@ def test_charm_sets_stored_state(): state_in = testing.State(relations={peer}) state_out = ctx.run(ctx.on.start(), state_in) rel = state_out.get_relation(peer.id) - assert rel.local_app_data["expensive_value"] == "42" + assert rel.local_app_data['expensive_value'] == '42' ``` diff --git a/docs/howto/manage-the-charm-version.md b/docs/howto/manage-the-charm-version.md index ae2fafc24..536a027e1 100644 --- a/docs/howto/manage-the-charm-version.md +++ b/docs/howto/manage-the-charm-version.md @@ -66,7 +66,11 @@ in your `tests/integration/test_charm.py` file, add a new test: ```py def test_charm_version_is_set(juju: jubilant.Juju): """Verify that the charm version has been set.""" - version = juju.status().apps["your-app"].charm_version - expected_version = subprocess.check_output(["git", "rev-parse", "HEAD"]).decode("utf8") + version = juju.status().apps['your-app'].charm_version + expected_version = subprocess.check_output([ + 'git', + 'rev-parse', + 'HEAD', + ]).decode('utf8') assert version == expected_version ``` diff --git a/docs/howto/manage-the-workload-version.md b/docs/howto/manage-the-workload-version.md index 31ec108ac..535f5dfe1 100644 --- a/docs/howto/manage-the-workload-version.md +++ b/docs/howto/manage-the-workload-version.md @@ -43,7 +43,7 @@ example: ```python def _on_start(self, event: ops.StartEvent): # The workload exposes the version via HTTP at /version - version = requests.get("http://localhost:8000/version").text + version = requests.get('http://localhost:8000/version').text self.unit.set_workload_version(version) ``` @@ -62,16 +62,17 @@ new test that verifies the workload version is set. For example: ```python from ops import testing + def test_workload_version_is_set(): ctx = testing.Context(MyCharm) # Suppose that the charm gets the workload version by running the command # `/bin/server --version` in the container. Firstly, we mock that out: container = testing.Container( - "webserver", - execs={testing.Exec(["/bin/server", "--version"], stdout="1.2\n")}, + 'webserver', + execs={testing.Exec(['/bin/server', '--version'], stdout='1.2\n')}, ) out = ctx.run(ctx.on.start(), testing.State(containers={container})) - assert out.workload_version == "1.2" + assert out.workload_version == '1.2' ``` ### Write integration tests @@ -100,12 +101,12 @@ def test_deploy(charm: pathlib.Path, juju: jubilant.Juju): def test_workload_version_is_set(juju: jubilant.Juju): # Verify that the workload version has been set. - version = juju.status().apps["your-app"].version + version = juju.status().apps['your-app'].version # We'll need to update this version every time we upgrade to a new workload # version. If the workload has an API or some other way of getting the # version, the test should get it from there and use that to compare to the # unit setting. - assert version == "3.14" + assert version == '3.14' ``` > See more: [](jubilant.Juju.status) diff --git a/docs/howto/migrate/migrate-from-a-hooks-based-charm.md b/docs/howto/migrate/migrate-from-a-hooks-based-charm.md index 437986b80..3d029bfc8 100644 --- a/docs/howto/migrate/migrate-from-a-hooks-based-charm.md +++ b/docs/howto/migrate/migrate-from-a-hooks-based-charm.md @@ -100,18 +100,31 @@ The minimal-effort solution in this case could be to create a file `/src/charm.p import os import ops + class Microsample(ops.CharmBase): - def __init__(self, *args): - super().__init__(*args) - self.framework.observe(self.on.config_changed, lambda _: os.popen('../hooks/config-changed')) - self.framework.observe(self.on.install, lambda _: os.popen('../hooks/install')) - self.framework.observe(self.on.start, lambda _: os.popen('../hooks/start')) - self.framework.observe(self.on.stop, lambda _: os.popen('../hooks/stop')) - # etc... + def __init__(self, framework): + super().__init__(framework) + framework.observe(self.on.config_changed, self._on_config_changed) + framework.observe(self.on.install, self._on_install) + framework.observe(self.on.start, self._on_start) + framework.observe(self.on.stop, self._on_stop) + # etc... + + def _on_config_changed(self, _event): + os.popen('../hooks/config-changed') + + def _on_install(self, _event): + os.popen('../hooks/install') + + def _on_start(self, _event): + os.popen('../hooks/start') -if __name__ == "__main__": - main(ops.Microsample) + def _on_stop(self, _event): + os.popen('../hooks/stop') + +if __name__ == '__main__': + ops.main(Microsample) ``` Relying on `popen` is _not_ how Ops is supposed to be used. However, this code will work, and it demonstrates the core principle of mapping hook names to handler code. @@ -121,7 +134,6 @@ Relying on `popen` is _not_ how Ops is supposed to be used. However, this code w We need a few preparatory steps:\ • Add a `requirements.txt` file to ensure that the charm's Python environment will install for us the `ops` package.\ • Modify the install hook to install `snap` for us, which is used in the script.\ -• In practice we cannot bind lambdas to `observe`, we need to write dedicated _methods_ for that.\ • We need to figure out the required environment variables for the commands to work, which is not trivial.\ \ A more detailed explanation of this process is worthy of its own how-to guide, so we'll skip to the punchline here: it works. Check out [this branch](https://github.com/PietroPasotti/hooks-to-ops/tree/1-sh-charm) and see for yourself. @@ -155,28 +167,33 @@ Let's begin with `install`. The `/hooks/install` script checks if a snap package is installed; if not, it installs it. We need to still reach out to a shell to grab the `snap` package info and install the package, but we can have the logic and the status management in Python, which is nice. We use `subprocessing.check_call` to reach out to the OS. And yes, there is a better way to do this, we'll get to that later. ```python - def _on_install(self, _event): - snapinfo_cmd = Popen("snap info microsample".split(" "), - stdout=subprocess.PIPE) - output = check_output("grep -c 'installed'".split(" "), - stdin=snapinfo_cmd.stdout) - is_microsample_installed = bool(output.decode("ascii").strip()) - - if not is_microsample_installed: - self.unit.status = ops.MaintenanceStatus("installing microsample") - out = check_call("snap install microsample --edge") - - self.unit.status = ops.ActiveStatus() +def _on_install(self, _event): + snapinfo_cmd = Popen( + 'snap info microsample'.split(' '), stdout=subprocess.PIPE + ) + output = check_output( + "grep -c 'installed'".split(' '), stdin=snapinfo_cmd.stdout + ) + is_microsample_installed = bool(output.decode('ascii').strip()) + + if not is_microsample_installed: + self.unit.status = ops.MaintenanceStatus('installing microsample') + out = check_call('snap install microsample --edge') + + self.unit.status = ops.ActiveStatus() ``` For `on-start` and `on-stop`, which are simple instructions to `systemctl` to start/stop the `microsample` service, we can copy over the commands as they are: ```python - def _on_start(self, _event): # noqa - check_call("systemctl start snap.microsample.microsample.service".split(' ')) +def _on_start(self, _event): # noqa + check_call( + 'systemctl start snap.microsample.microsample.service'.split(' ') + ) + - def _on_stop(self, _event): # noqa - check_call("systemctl stop snap.microsample.microsample.service".split(' ')) +def _on_stop(self, _event): # noqa + check_call('systemctl stop snap.microsample.microsample.service'.split(' ')) ``` In a couple of places in the scripts, `sleep 3` calls ensure that the service has some time to come up; however, this might get the charm stuck in the waiting loop if for whatever reason the service does NOT come up, so it is quite risky and we are not going to do that. Instead, we are going to rely on the fact that if other event handlers were to fail because of the service not being up, they would handle that case appropriately (e.g., defer the event if necessary). @@ -188,19 +205,19 @@ The rest of the translation is pretty straightforward. However, it is still usef We are going to add a helper method: ```python - def _get_website_relation(self) -> ops.Relation: - # WARNING: would return None if called too early, e.g. during install - return self.model.get_relation("website") +def _get_website_relation(self) -> ops.Relation: + # WARNING: would return None if called too early, e.g. during install + return self.model.get_relation('website') ``` That allows us to fetch the Relation wherever we need it and access its contents or mutate them in a natural way: ```python - def _on_website_relation_joined(self, _event): - relation = self._get_website_relation() - relation.data[self.unit].update( - {"hostname": self.private_address, - "port": self.port} - ) +def _on_website_relation_joined(self, _event): + relation = self._get_website_relation() + relation.data[self.unit].update({ + 'hostname': self.private_address, + 'port': self.port, + }) ``` Note how `relation.data` provides an interface to the relation databag (see [](#set-up-a-relation)) and we need to select which part of that bag to access by passing an `ops.Unit` instance. @@ -209,8 +226,8 @@ Note how `relation.data` provides an interface to the relation databag (see [](# Every maintainable charm will have some form of logging integrated; in a few places in the Bash scripts we see calls to a `juju-log` command; we can replace them with simple `logger.log` calls; such as in ```python - def _on_website_relation_departed(self, _event): # noqa - logger.debug("%s departed website relation", self.unit.name) +def _on_website_relation_departed(self, _event): # noqa + logger.debug('%s departed website relation', self.unit.name) ``` Where `logger = logging.getLogger(__name__)`. @@ -219,7 +236,7 @@ Where `logger = logging.getLogger(__name__)`. Some of the Bash scripts read environment variables such as `$JUJU_REMOTE_UNIT`, `$JUJU_UNIT_NAME` ; of course we could do ```python -JUJU_UNIT_NAME = os.environ["JUJU_UNIT_NAME"] +JUJU_UNIT_NAME = os.environ['JUJU_UNIT_NAME'] ``` but `CharmBase` exposes a `.unit` attribute we can read this information from, instead of grabbing it off the environment; this makes for more readable code. @@ -235,33 +252,33 @@ In the `_on_install` method we had translated one-to-one the calls to `snap info Then we can replace all that `Popen` piping with simpler calls into the lib's API; `_on_install `becomes: ```python - def _on_install(self, _event): - microsample_snap = snap.SnapCache()["microsample"] - if not microsample_snap.present: - self.unit.status = ops.MaintenanceStatus("installing microsample") - microsample_snap.ensure(snap.SnapState.Latest, channel="edge") - - self.wait_service_active() - self.unit.status = ops.ActiveStatus() - +def _on_install(self, _event): + microsample_snap = snap.SnapCache()['microsample'] + if not microsample_snap.present: + self.unit.status = ops.MaintenanceStatus('installing microsample') + microsample_snap.ensure(snap.SnapState.Latest, channel='edge') + + self.wait_service_active() + self.unit.status = ops.ActiveStatus() ``` Similarly all that string parsing we were doing to get a hold of the snap version, can be simplified by grabbing the `microsample_snap.channel` (not quite the same, but for the purposes of this charm, it is close enough). ```python - def _get_microsample_version(self): - microsample_snap = snap.SnapCache()["microsample"] - return microsample_snap.channel +def _get_microsample_version(self): + microsample_snap = snap.SnapCache()['microsample'] + return microsample_snap.channel ``` Also, we can interact with the `microsample` service via the `operator_libs_linux.v0` charm library, which wraps `systemd` and allows us to write simply: ```python - def _on_start(self, _event): # noqa - systemd.service_start("snap.microsample.microsample.service") +def _on_start(self, _event): # noqa + systemd.service_start('snap.microsample.microsample.service') + - def _on_stop(self, _event): # noqa - systemd.service_stop("snap.microsample.microsample.service") +def _on_stop(self, _event): # noqa + systemd.service_stop('snap.microsample.microsample.service') ``` ```{note} diff --git a/docs/howto/migrate/migrate-integration-tests-from-pytest-operator.md b/docs/howto/migrate/migrate-integration-tests-from-pytest-operator.md index b4c833bfe..81a7d10ab 100644 --- a/docs/howto/migrate/migrate-integration-tests-from-pytest-operator.md +++ b/docs/howto/migrate/migrate-integration-tests-from-pytest-operator.md @@ -75,11 +75,11 @@ import os import pathlib -@pytest.fixture(scope="session") +@pytest.fixture(scope='session') def charm(): """Return the path of the charm under test.""" # Assume the current working directory is the charm root. - yield get_charm_path(env_var="CHARM_PATH", default_dir=pathlib.Path()) + yield get_charm_path(env_var='CHARM_PATH', default_dir=pathlib.Path()) def get_charm_path(env_var: str, default_dir: pathlib.Path) -> pathlib.Path: @@ -120,6 +120,7 @@ In your tests, use the fixture like this: ```python # tests/integration/test_charm.py + def test_active(juju: jubilant.Juju, charm_path: pathlib.Path): juju.deploy(charm_path) juju.wait(jubilant.all_active) @@ -143,13 +144,12 @@ import pytest import pytest_jubilant -@pytest.mark.fixture(scope="module") +@pytest.mark.fixture(scope='module') def other_model(juju_factory: pytest_jubilant.JujuFactory): - yield juju_factory.get_juju("other") + yield juju_factory.get_juju('other') -def test_cross_model(juju: jubilant.Juju, other_model: jubilant.Juju): - ... +def test_cross_model(juju: jubilant.Juju, other_model: jubilant.Juju): ... ``` (how_to_migrate_an_application_fixture)= @@ -196,9 +196,10 @@ import pathlib import jubilant import pytest + @pytest.fixture(scope='module') def app(juju: jubilant.Juju, charm_path: pathlib.Path): - my_app_name = "mycharm" + my_app_name = 'mycharm' juju.deploy( charm_path, my_app_name, @@ -221,6 +222,7 @@ In your tests, you'll need to specify that the test depends on `juju` as well as ```python # tests/integration/test_charm.py + def test_active(juju: jubilant.Juju, app: str): status = juju.status() assert status.apps[app].is_active @@ -276,6 +278,7 @@ A python-libjuju model is updated in the background using websockets. In Jubilan async def test_active(app: Application): assert app.units[0].workload_status == ActiveStatus.name + # jubilant def test_active(juju: jubilant.Juju, app: str): status = juju.status() @@ -300,6 +303,7 @@ async def test_active(model: Model): await model.deploy('mycharm') await model.wait_for_idle(status='active') # implies raise_on_error=True + # jubilant def test_active(juju: jubilant.Juju): juju.deploy('mycharm') @@ -314,6 +318,7 @@ async def test_idle(model: Model): await model.deploy('mycharm') await model.wait_for_idle() + # jubilant def test_active(juju: jubilant.Juju): juju.deploy('mycharm') @@ -325,8 +330,8 @@ It's common to use a `lambda` function to customize the callable or compose mult ```python juju.wait( lambda status: ( - jubilant.all_active(status, 'mysql', 'redis') and - jubilant.all_blocked(status, 'logger'), + jubilant.all_active(status, 'mysql', 'redis') + and jubilant.all_blocked(status, 'logger'), ), ) ``` @@ -338,8 +343,8 @@ juju.deploy('mycharm') juju.wait( ready=lambda status: jubilant.all_active(status, 'mycharm'), error=jubilant.any_error, - delay=0.2, # poll "juju status" every 200ms (default 1s) - timeout=60, # set overall timeout to 60s (default juju.wait_timeout) + delay=0.2, # poll "juju status" every 200ms (default 1s) + timeout=60, # set overall timeout to 60s (default juju.wait_timeout) successes=7, # require ready to return success 7x in a row (default 3) ) ``` diff --git a/docs/howto/migrate/migrate-unit-tests-from-harness.md b/docs/howto/migrate/migrate-unit-tests-from-harness.md index 86efdc29f..a7da01c5e 100644 --- a/docs/howto/migrate/migrate-unit-tests-from-harness.md +++ b/docs/howto/migrate/migrate-unit-tests-from-harness.md @@ -26,14 +26,16 @@ class DemoCharm(ops.CharmBase): def __init__(self, framework: ops.Framework) -> None: super().__init__(framework) - framework.observe(self.on["get-value"].action, self._on_get_value_action) + framework.observe( + self.on['get-value'].action, self._on_get_value_action + ) def _on_get_value_action(self, event: ops.ActionEvent) -> None: """Handle the get-value action.""" - if event.params["value"] == "please fail": - event.fail("Action failed, as requested") + if event.params['value'] == 'please fail': + event.fail('Action failed, as requested') else: - event.set_results({"out-value": event.params["value"]}) + event.set_results({'out-value': event.params['value']}) ``` Also suppose that we have the following testing code, written using Harness: @@ -47,8 +49,8 @@ from charm import DemoCharm def test_action(): harness = testing.Harness(DemoCharm) harness.begin() - output = harness.run_action("get-value", {"value": "foo"}) - assert output.results == {"out-value": "foo"} + output = harness.run_action('get-value', {'value': 'foo'}) + assert output.results == {'out-value': 'foo'} harness.cleanup() ``` @@ -70,8 +72,8 @@ from charm import DemoCharm def test_get_value_action(): ctx = testing.Context(DemoCharm) state_in = testing.State() - ctx.run(ctx.on.action("get-value", params={"value": "foo"}), state_in) - assert ctx.action_results == {"out-value": "foo"} + ctx.run(ctx.on.action('get-value', params={'value': 'foo'}), state_in) + assert ctx.action_results == {'out-value': 'foo'} ``` The [`ctx.run`](ops.testing.Context.run) call is the part that simulates the event. @@ -89,8 +91,11 @@ def test_get_value_action_failed(): ctx = testing.Context(DemoCharm) state_in = testing.State() with pytest.raises(testing.ActionFailed) as exc_info: - ctx.run(ctx.on.action("get-value", params={"value": "please fail"}), state_in) - assert exc_info.value.message == "Action failed, as requested" + ctx.run( + ctx.on.action('get-value', params={'value': 'please fail'}), + state_in, + ) + assert exc_info.value.message == 'Action failed, as requested' ``` In a more realistic charm, the action will use data from the charm or workload. For an example, see [](#harness-migration-relation). When writing state-transition tests for a real action, we also need to consider collect-status. @@ -111,19 +116,21 @@ class DemoCharm(ops.CharmBase): def __init__(self, framework: ops.Framework) -> None: super().__init__(framework) - self.container = self.unit.get_container("my-container") + self.container = self.unit.get_container('my-container') framework.observe(self.on.collect_unit_status, self._on_collect_status) - framework.observe(self.on["get-value"].action, self._on_get_value_action) + framework.observe( + self.on['get-value'].action, self._on_get_value_action + ) def _on_collect_status(self, event: ops.CollectStatusEvent) -> None: """Report the status of the workload.""" try: - service = self.container.get_service("workload") + service = self.container.get_service('workload') except (ops.ModelError, ops.pebble.ConnectionError): - event.add_status(ops.MaintenanceStatus("waiting for container")) + event.add_status(ops.MaintenanceStatus('waiting for container')) else: if not service.is_running(): - event.add_status(ops.MaintenanceStatus("waiting for workload")) + event.add_status(ops.MaintenanceStatus('waiting for workload')) event.add_status(ops.ActiveStatus()) ... # _on_get_value_action is unchanged. @@ -144,8 +151,8 @@ As a reminder, here's our definition of `test_get_value_action`: def test_get_value_action(): ctx = testing.Context(DemoCharm) state_in = testing.State() - ctx.run(ctx.on.action("get-value", params={"value": "foo"}), state_in) - assert ctx.action_results == {"out-value": "foo"} + ctx.run(ctx.on.action('get-value', params={'value': 'foo'}), state_in) + assert ctx.action_results == {'out-value': 'foo'} ``` To fix the test, we need to add a mock container to the input state: @@ -153,10 +160,10 @@ To fix the test, we need to add a mock container to the input state: ```python def test_get_value_action(): ctx = testing.Context(DemoCharm) - container = testing.Container("my-container", can_connect=True) + container = testing.Container('my-container', can_connect=True) state_in = testing.State(containers={container}) - ctx.run(ctx.on.action("get-value", params={"value": "foo"}), state_in) - assert ctx.action_results == {"out-value": "foo"} + ctx.run(ctx.on.action('get-value', params={'value': 'foo'}), state_in) + assert ctx.action_results == {'out-value': 'foo'} ``` In [](#harness-migration-container), we'll work through a more realistic example that shows: @@ -178,10 +185,18 @@ class DemoCharm(ops.CharmBase): def __init__(self, framework: ops.Framework) -> None: super().__init__(framework) # Use database helpers from charms.data_platform_libs.v0.data_interfaces. - self.database = DatabaseRequires(self, relation_name="database", database_name="my-db") - framework.observe(self.database.on.database_created, self._on_database_available) - framework.observe(self.database.on.endpoints_changed, self._on_database_available) - framework.observe(self.on["get-db-endpoint"].action, self._on_get_db_endpoint_action) + self.database = DatabaseRequires( + self, relation_name='database', database_name='my-db' + ) + framework.observe( + self.database.on.database_created, self._on_database_available + ) + framework.observe( + self.database.on.endpoints_changed, self._on_database_available + ) + framework.observe( + self.on['get-db-endpoint'].action, self._on_get_db_endpoint_action + ) def _on_database_available( self, _: DatabaseCreatedEvent | DatabaseEndpointsChangedEvent @@ -196,16 +211,16 @@ class DemoCharm(ops.CharmBase): """Handle the get-db-endpoint action.""" endpoint = self.get_endpoint_from_relation() if endpoint: - event.set_results({"endpoint": endpoint}) + event.set_results({'endpoint': endpoint}) else: - event.fail("Database endpoint is not available") + event.fail('Database endpoint is not available') def get_endpoint_from_relation(self) -> str | None: """Get the database endpoint from the relation data.""" relations = self.database.fetch_relation_data() for data in relations.values(): if data: - return data["endpoints"] + return data['endpoints'] def write_workload_config(self, config: str) -> None: """Update the workload's configuration.""" @@ -225,30 +240,32 @@ def test_db_endpoint(monkeypatch: pytest.MonkeyPatch): harness = testing.Harness(DemoCharm) # Prepare the charm with initial relation data. - relation_id = harness.add_relation("database", "postgresql") + relation_id = harness.add_relation('database', 'postgresql') harness.update_relation_data( relation_id, - "postgresql", - {"endpoints": "foo.local:1234"}, + 'postgresql', + {'endpoints': 'foo.local:1234'}, ) harness.begin_with_initial_hooks() # Prepare a mock workload object with matching config, assuming we've # defined a MockWorkload class with suitable attributes and methods. - workload = MockWorkload("foo.local:1234") - monkeypatch.setattr("charm.DemoCharm.write_workload_config", workload.write_config) + workload = MockWorkload('foo.local:1234') + monkeypatch.setattr( + 'charm.DemoCharm.write_workload_config', workload.write_config + ) # Update the relation data and check that the charm wrote new workload config. harness.update_relation_data( relation_id, - "postgresql", - {"endpoints": "bar.local:5678"}, + 'postgresql', + {'endpoints': 'bar.local:5678'}, ) - assert workload.config == "bar.local:5678" + assert workload.config == 'bar.local:5678' # Check that the action returns the expected database endpoint. - output = harness.run_action("get-db-endpoint") - assert output.results == {"endpoint": "bar.local:5678"} + output = harness.run_action('get-db-endpoint') + assert output.results == {'endpoint': 'bar.local:5678'} harness.cleanup() ``` @@ -283,15 +300,17 @@ from charm import DemoCharm def test_relation_changed(monkeypatch: pytest.MonkeyPatch): ctx = testing.Context(DemoCharm) - workload = MockWorkload("foo.local:1234") - monkeypatch.setattr("charm.DemoCharm.write_workload_config", workload.write_config) + workload = MockWorkload('foo.local:1234') + monkeypatch.setattr( + 'charm.DemoCharm.write_workload_config', workload.write_config + ) relation = testing.Relation( - endpoint="database", - remote_app_data={"endpoints": "bar.local:5678"}, + endpoint='database', + remote_app_data={'endpoints': 'bar.local:5678'}, ) state_in = testing.State(relations={relation}) ctx.run(ctx.on.relation_changed(relation), state_in) - assert workload.config == "bar.local:5678" + assert workload.config == 'bar.local:5678' ``` ### Test the action @@ -312,12 +331,12 @@ from charm import DemoCharm def test_get_db_endpoint_action(): ctx = testing.Context(DemoCharm) relation = testing.Relation( - endpoint="database", - remote_app_data={"endpoints": "bar.local:5678"}, + endpoint='database', + remote_app_data={'endpoints': 'bar.local:5678'}, ) state_in = testing.State(relations={relation}) - ctx.run(ctx.on.action("get-db-endpoint"), state_in) - assert ctx.action_results == {"endpoint": "bar.local:5678"} + ctx.run(ctx.on.action('get-db-endpoint'), state_in) + assert ctx.action_results == {'endpoint': 'bar.local:5678'} ``` (harness-migration-container)= @@ -333,34 +352,36 @@ class DemoCharm(ops.CharmBase): def __init__(self, framework: ops.Framework) -> None: super().__init__(framework) - self.container = self.unit.get_container("my-container") - framework.observe(self.on["my-container"].pebble_ready, self._on_pebble_ready) + self.container = self.unit.get_container('my-container') + framework.observe( + self.on['my-container'].pebble_ready, self._on_pebble_ready + ) framework.observe(self.on.collect_unit_status, self._on_collect_status) def _on_pebble_ready(self, _: ops.PebbleReadyEvent) -> None: """Use Pebble to configure and start the workload in the container.""" layer: ops.pebble.LayerDict = { - "services": { - "workload": { - "override": "replace", - "command": "run-workload", - "startup": "enabled", + 'services': { + 'workload': { + 'override': 'replace', + 'command': 'run-workload', + 'startup': 'enabled', } } } - self.container.add_layer("base", layer, combine=True) + self.container.add_layer('base', layer, combine=True) self.container.replan() ... # Check that the workload is actually running. def _on_collect_status(self, event: ops.CollectStatusEvent) -> None: """Report the status of the workload.""" try: - service = self.container.get_service("workload") + service = self.container.get_service('workload') except (ops.ModelError, ops.pebble.ConnectionError): - event.add_status(ops.MaintenanceStatus("waiting for container")) + event.add_status(ops.MaintenanceStatus('waiting for container')) else: if not service.is_running(): - event.add_status(ops.MaintenanceStatus("waiting for workload")) + event.add_status(ops.MaintenanceStatus('waiting for workload')) event.add_status(ops.ActiveStatus()) ``` @@ -383,15 +404,15 @@ def test_container(): assert isinstance(harness.charm.unit.status, ops.model.ActiveStatus) # Check the Pebble plan in the workload container. - plan = harness.get_container_pebble_plan("my-container") - assert "workload" in plan.services - assert plan.services["workload"].command == "run-workload" + plan = harness.get_container_pebble_plan('my-container') + assert 'workload' in plan.services + assert plan.services['workload'].command == 'run-workload' # Simulate a dropped connection to the container, then check the charm's status. - harness.set_can_connect("my-container", False) + harness.set_can_connect('my-container', False) harness.evaluate_status() assert isinstance(harness.charm.unit.status, ops.model.MaintenanceStatus) - assert harness.charm.unit.status.message == "waiting for container" + assert harness.charm.unit.status.message == 'waiting for container' harness.cleanup() ``` @@ -420,12 +441,12 @@ from charm import DemoCharm def test_pebble_ready(): ctx = testing.Context(DemoCharm) - container_in = testing.Container("my-container", can_connect=True) + container_in = testing.Container('my-container', can_connect=True) state_in = testing.State(containers={container_in}) state_out = ctx.run(ctx.on.pebble_ready(container_in), state_in) container_out = state_out.get_container(container_in.name) - assert "workload" in container_out.plan.services - assert container_out.plan.services["workload"].command == "run-workload" + assert 'workload' in container_out.plan.services + assert container_out.plan.services['workload'].command == 'run-workload' ``` ```{note} @@ -435,8 +456,11 @@ In state-transition tests, the objects in the `State` are immutable. Calling `ct The `test_pebble_ready` function doesn't fully cover the charm's `_on_pebble_ready` method. In addition to defining a service in the container, `_on_pebble_ready` uses [`replan`](ops.Container.replan) to start the service. To cover this, one option would be to check the service status at the end of `test_pebble_ready`: ```python - ... - assert container_out.service_statuses["workload"] == ops.pebble.ServiceStatus.ACTIVE +... +assert ( + container_out.service_statuses['workload'] + == ops.pebble.ServiceStatus.ACTIVE +) ``` Alternatively, we can take advantage of the charm's status reporting: @@ -451,12 +475,12 @@ This works because the testing framework automatically simulates running `_on_co ```python def test_pebble_ready(): ctx = testing.Context(DemoCharm) - container_in = testing.Container("my-container", can_connect=True) + container_in = testing.Container('my-container', can_connect=True) state_in = testing.State(containers={container_in}) state_out = ctx.run(ctx.on.pebble_ready(container_in), state_in) container_out = state_out.get_container(container_in.name) - assert "workload" in container_out.plan.services - assert container_out.plan.services["workload"].command == "run-workload" + assert 'workload' in container_out.plan.services + assert container_out.plan.services['workload'].command == 'run-workload' assert state_out.unit_status == testing.ActiveStatus() ``` @@ -482,25 +506,23 @@ from ops import pebble, testing from charm import DemoCharm -layer = pebble.Layer( - { - "services": { - "workload": { - "override": "replace", - "command": "mock-command", - "startup": "enabled", - }, +layer = pebble.Layer({ + 'services': { + 'workload': { + 'override': 'replace', + 'command': 'mock-command', + 'startup': 'enabled', }, - } -) + }, +}) def test_status_active(): ctx = testing.Context(DemoCharm) container = testing.Container( - "my-container", - layers={"base": layer}, - service_statuses={"workload": pebble.ServiceStatus.ACTIVE}, + 'my-container', + layers={'base': layer}, + service_statuses={'workload': pebble.ServiceStatus.ACTIVE}, can_connect=True, ) state_in = testing.State(containers={container}) @@ -510,10 +532,12 @@ def test_status_active(): def test_status_container_down(): ctx = testing.Context(DemoCharm) - container = testing.Container("my-container", can_connect=False) + container = testing.Container('my-container', can_connect=False) state_in = testing.State(containers={container}) state_out = ctx.run(ctx.on.update_status(), state_in) - assert state_out.unit_status == testing.MaintenanceStatus("waiting for container") + assert state_out.unit_status == testing.MaintenanceStatus( + 'waiting for container' + ) ``` These tests cover the same situations as our Harness test, but in isolation, not part of a sequence of events. The biggest difference is in `test_status_active`, where we mock a Pebble layer instead of relying on the layer produced by the pebble-ready event handler. diff --git a/docs/howto/run-workloads-with-a-charm-machines.md b/docs/howto/run-workloads-with-a-charm-machines.md index b3592f041..9493a71b8 100644 --- a/docs/howto/run-workloads-with-a-charm-machines.md +++ b/docs/howto/run-workloads-with-a-charm-machines.md @@ -58,9 +58,9 @@ class MyCharm(ops.CharmBase): def _on_collect_status(self, event: ops.CollectStatusEvent) -> None: if not myworkload.is_installed(): - event.add_status(ops.MaintenanceStatus("Installing workload")) + event.add_status(ops.MaintenanceStatus('Installing workload')) if not myworkload.is_running(): - event.add_status(ops.MaintenanceStatus("Starting workload")) + event.add_status(ops.MaintenanceStatus('Starting workload')) event.add_status(ops.ActiveStatus()) ``` @@ -92,13 +92,13 @@ from charmlibs import apt def install() -> None: apt.update() # Pin to a specific version so deployments are reproducible. - apt.add_package("tinyproxy-bin", "1.11.1-3") + apt.add_package('tinyproxy-bin', '1.11.1-3') # On failure, apt raises charmlibs.apt.PackageError, which puts the # charm into error status with a clear message in the Juju logs. def uninstall() -> None: - apt.remove_package("tinyproxy-bin") + apt.remove_package('tinyproxy-bin') ``` ```{admonition} Best practice @@ -125,16 +125,16 @@ from charmlibs import snap def install() -> None: cache = snap.SnapCache() - workload = cache["my-workload"] - workload.ensure(snap.SnapState.Latest, channel="stable") + workload = cache['my-workload'] + workload.ensure(snap.SnapState.Latest, channel='stable') def start() -> None: - snap.SnapCache()["my-workload"].start(enable=True) + snap.SnapCache()['my-workload'].start(enable=True) def stop() -> None: - snap.SnapCache()["my-workload"].stop(disable=True) + snap.SnapCache()['my-workload'].stop(disable=True) ``` (run-workloads-with-a-charm-machines-when-theres-no-library)= @@ -176,7 +176,7 @@ import signal from charmlibs import pathops -PID_FILE = pathops.LocalPath("/var/run/myworkload.pid") +PID_FILE = pathops.LocalPath('/var/run/myworkload.pid') def reload_config() -> None: @@ -238,16 +238,16 @@ class MockWorkload: return self.running def reload_config(self) -> None: - self.signals.append("SIGUSR1") + self.signals.append('SIGUSR1') def get_version(self) -> str: - return "1.0.0" + return '1.0.0' @pytest.fixture def workload(monkeypatch: pytest.MonkeyPatch) -> MockWorkload: mock = MockWorkload() - monkeypatch.setattr("charm.myworkload", mock) + monkeypatch.setattr('charm.myworkload', mock) return mock @@ -258,7 +258,7 @@ def test_install(workload: MockWorkload): state_out = ctx.run(ctx.on.install(), testing.State()) # Assert assert workload.is_installed() - assert state_out.workload_version == "1.0.0" + assert state_out.workload_version == '1.0.0' def test_start(workload: MockWorkload): @@ -293,25 +293,27 @@ from charm import myworkload def test_install_calls_apt(monkeypatch: pytest.MonkeyPatch): calls: list[tuple[str, str]] = [] monkeypatch.setattr( - "charm.myworkload.apt.update", lambda: calls.append(("update", "")), + 'charm.myworkload.apt.update', + lambda: calls.append(('update', '')), ) monkeypatch.setattr( - "charm.myworkload.apt.add_package", + 'charm.myworkload.apt.add_package', lambda name, version: calls.append((name, version)), ) myworkload.install() - assert calls == [("update", ""), ("tinyproxy-bin", "1.11.1-3")] + assert calls == [('update', ''), ('tinyproxy-bin', '1.11.1-3')] def test_reload_config_sends_sigusr1( - monkeypatch: pytest.MonkeyPatch, tmp_path, + monkeypatch: pytest.MonkeyPatch, + tmp_path: pathlib.Path, ): - pid_file = tmp_path / "myworkload.pid" - pid_file.write_text("1234") - monkeypatch.setattr("charm.myworkload.PID_FILE", pid_file) + pid_file = tmp_path / 'myworkload.pid' + pid_file.write_text('1234') + monkeypatch.setattr('charm.myworkload.PID_FILE', pid_file) sent: list[tuple[int, int]] = [] - monkeypatch.setattr("os.kill", lambda pid, sig: sent.append((pid, sig))) + monkeypatch.setattr('os.kill', lambda pid, sig: sent.append((pid, sig))) myworkload.reload_config() assert sent == [(1234, signal.SIGUSR1)] @@ -320,11 +322,11 @@ def test_reload_config_sends_sigusr1( def test_start_runs_subprocess(monkeypatch: pytest.MonkeyPatch): commands: list[list[str]] = [] monkeypatch.setattr( - "subprocess.run", + 'subprocess.run', lambda cmd, **kwargs: commands.append(cmd) or None, ) myworkload.start() - assert commands == [["myworkload"]] + assert commands == [['myworkload']] ``` ### Run the tests @@ -354,17 +356,18 @@ def test_install_and_start(): assert not myworkload.is_installed() myworkload.install() assert myworkload.is_installed() - assert myworkload.get_version() == "1.11.1" + assert myworkload.get_version() == '1.11.1' myworkload.start() assert myworkload.is_running() # The real systemd unit should be active. result = subprocess.run( - ["/usr/bin/systemctl", "is-active", "tinyproxy"], - capture_output=True, text=True, + ['/usr/bin/systemctl', 'is-active', 'tinyproxy'], + capture_output=True, + text=True, ) - assert result.stdout.strip() == "active" + assert result.stdout.strip() == 'active' myworkload.stop() assert not myworkload.is_running() @@ -386,19 +389,21 @@ import pytest @pytest.mark.juju_setup def test_deploy(charm: pathlib.Path, juju: jubilant.Juju): - juju.deploy(charm, app="myworkload") + juju.deploy(charm, app='myworkload') juju.wait(jubilant.all_active, timeout=600) def test_workload_version(juju: jubilant.Juju): - version = juju.status().apps["myworkload"].version - assert version == "1.11.1" # The version we pinned in install(), as reported by the workload. + version = juju.status().apps['myworkload'].version + assert ( + version == '1.11.1' + ) # The version we pinned in install(), as reported by the workload. def test_blocks_on_invalid_config(juju: jubilant.Juju): - juju.config("myworkload", {"slug": "not/valid"}) + juju.config('myworkload', {'slug': 'not/valid'}) juju.wait(jubilant.all_blocked) - juju.config("myworkload", reset="slug") + juju.config('myworkload', reset='slug') ``` The `juju` fixture from `pytest-jubilant` creates a temporary model per test file and tears it down afterwards. You supply a `charm` fixture that locates the packed `.charm` file. For an example, see [`conftest.py` in machine-tinyproxy's integration tests](https://github.com/canonical/operator/blob/main/examples/machine-tinyproxy/tests/integration/conftest.py). diff --git a/docs/howto/trace-your-charm.md b/docs/howto/trace-your-charm.md index 38558885c..1fd9707f7 100644 --- a/docs/howto/trace-your-charm.md +++ b/docs/howto/trace-your-charm.md @@ -104,8 +104,10 @@ import opentelemetry.trace tracer = opentelemetry.trace.get_tracer(__name__) + class Workload: ... + def migrate_db(self): with tracer.start_as_current_span('migrate-db') as span: for attempt in range(3): @@ -113,7 +115,7 @@ class Workload: subprocess.check_output('/path/to/migrate.sh') except subprocess.CalledProcessError: span.add_event('db-migrate-failed', {'attempt': attempt}) - time.sleep(10 ** attempt) + time.sleep(10**attempt) else: break else: @@ -172,11 +174,13 @@ Or that span A is an ancestor of span C, which allows you to validate that an im ```py spans_by_id = {s.context.span_id: s for s in ctx.trace_data} + def ancestors(span: ReadableSpan) -> Generator[ReadableSpan]: while span.parent: span = spans_by_id[span.parent.span_id] yield span + assert span_a in list(ancestors(span_c)) ``` @@ -184,11 +188,11 @@ You can disambiguate spans using their [`instrumentation_scope`](opentelemetry.s ```py # Spans from Ops -ops_span.instrumentation_scope.name == "ops" +ops_span.instrumentation_scope.name == 'ops' ops_span.name == ... # tracer = opentelemetry.trace.get_tracer("my-charm") -my_span.instrumentation_scope.name == "my-charm" +my_span.instrumentation_scope.name == 'my-charm' my_span.name == ... ``` diff --git a/docs/howto/write-and-structure-charm-code.md b/docs/howto/write-and-structure-charm-code.md index e0e141170..117ff0e3b 100644 --- a/docs/howto/write-and-structure-charm-code.md +++ b/docs/howto/write-and-structure-charm-code.md @@ -111,8 +111,10 @@ Arrange the methods of this class in the following order: ```python def __init__(self, framework: ops.Framework): super().__init__(framework) - framework.observe(self.on["workload_container"].pebble_ready, self._on_pebble_ready) - self.container = self.unit.get_container("workload-container") + framework.observe( + self.on['workload_container'].pebble_ready, self._on_pebble_ready + ) + self.container = self.unit.get_container('workload-container') ``` 2. Event handlers, in the order that they're observed in `__init__`. Make the event handlers private. @@ -165,7 +167,7 @@ class DemoServerCharm(ops.CharmBase): def _on_start(self, event: ops.StartEvent): """Handle start event.""" - self.unit.status = ops.MaintenanceStatus("starting server") + self.unit.status = ops.MaintenanceStatus('starting server') demo_server.start() version = demo_server.get_version() self.unit.set_workload_version(version) @@ -175,7 +177,7 @@ class DemoServerCharm(ops.CharmBase): # If a method doesn't depend on Ops, put it in src/demo_server.py instead. -if __name__ == "__main__": # pragma: nocover +if __name__ == '__main__': # pragma: nocover ops.main(DemoServerCharm) ``` @@ -197,17 +199,19 @@ logger = logging.getLogger(__name__) def install() -> None: """Install the server from a snap.""" - subprocess.run(["snap", "install", "demo-server"], capture_output=True, check=True) + subprocess.run( + ['snap', 'install', 'demo-server'], capture_output=True, check=True + ) def start() -> None: """Start the server.""" - subprocess.run(["demo-server", "start"], capture_output=True, check=True) + subprocess.run(['demo-server', 'start'], capture_output=True, check=True) def get_version() -> str: """Get the running version of the server.""" - response = requests.get("http://localhost:5000/version", timeout=5) + response = requests.get('http://localhost:5000/version', timeout=5) return response.text ``` @@ -232,8 +236,8 @@ class DemoServerCharm(ops.CharmBase): # Observe other events... def _on_collect_status(self, event: ops.CollectStatusEvent): - if "port" not in self.config: - event.add_status(ops.BlockedStatus("no port specified")) + if 'port' not in self.config: + event.add_status(ops.BlockedStatus('no port specified')) return event.add_status(ops.ActiveStatus()) ``` @@ -245,11 +249,11 @@ Your handler for `collect_unit_status` won't have access to data about the main To report the unit status while handling an event, set [`self.unit.status`](ops.Unit.status). When your charm code sets `self.unit.status`, Ops immediately sends the unit status to Juju. For example: ```python - def _on_start(self, event: ops.StartEvent): - """Handle start event.""" - self.unit.status = ops.MaintenanceStatus("starting server") - demo_server.start() - # At the end of the handler, Ops triggers collect_unit_status. +def _on_start(self, event: ops.StartEvent): + """Handle start event.""" + self.unit.status = ops.MaintenanceStatus('starting server') + demo_server.start() + # At the end of the handler, Ops triggers collect_unit_status. ``` ### Application status @@ -264,22 +268,30 @@ class DemoServerCharm(ops.CharmBase): def __init__(self, framework: ops.Framework): super().__init__(framework) - framework.observe(self.on.collect_app_status, self._on_collect_app_status) - framework.observe(self.on.collect_unit_status, self._on_collect_unit_status) + framework.observe( + self.on.collect_app_status, self._on_collect_app_status + ) + framework.observe( + self.on.collect_unit_status, self._on_collect_unit_status + ) # Observe other events... def _on_collect_app_status(self, event: ops.CollectStatusEvent): # This is triggered for the leader unit only. num_degraded = ... # Inspect peer unit databags to find degraded units. if num_degraded: - event.add_status(ops.ActiveStatus(f"degraded units: {num_degraded}")) + event.add_status( + ops.ActiveStatus(f'degraded units: {num_degraded}') + ) return event.add_status(ops.ActiveStatus()) def _on_collect_unit_status(self, event: ops.CollectStatusEvent): # This is triggered for each unit. - if self.is_degraded(): # Use a custom helper method to determine status. - event.add_status(ops.ActiveStatus("degraded")) + if ( + self.is_degraded() + ): # Use a custom helper method to determine status. + event.add_status(ops.ActiveStatus('degraded')) return event.add_status(ops.ActiveStatus()) ``` diff --git a/docs/howto/write-integration-tests-for-a-charm.md b/docs/howto/write-integration-tests-for-a-charm.md index da18fa293..1be0b41cb 100644 --- a/docs/howto/write-integration-tests-for-a-charm.md +++ b/docs/howto/write-integration-tests-for-a-charm.md @@ -120,18 +120,20 @@ import pathlib import pytest -@pytest.fixture(scope="session") +@pytest.fixture(scope='session') def charm(): """Return the path of the charm under test.""" - charm = os.environ.get("CHARM_PATH") + charm = os.environ.get('CHARM_PATH') if not charm: - charm_dir = pathlib.Path() # Assume the current working directory is the charm root. - charms = list(charm_dir.glob("*.charm")) - assert charms, f"No charms were found in {charm_dir.absolute()}" - assert len(charms) == 1, f"Found more than one charm {charms}" + charm_dir = ( + pathlib.Path() + ) # Assume the current working directory is the charm root. + charms = list(charm_dir.glob('*.charm')) + assert charms, f'No charms were found in {charm_dir.absolute()}' + assert len(charms) == 1, f'Found more than one charm {charms}' charm = charms[0] path = pathlib.Path(charm).resolve() - assert path.is_file(), f"{path} is not a file" + assert path.is_file(), f'{path} is not a file' return path ``` @@ -208,17 +210,17 @@ After `test_deploy`, add more tests to check that your charm operates correctly. ```python def test_integrate(charm: pathlib.Path, juju: jubilant.Juju): # Deploy some other charm from Charmhub: - juju.deploy("other-app") + juju.deploy('other-app') # Integrate the charms: - juju.integrate("your-app:endpoint1", "other-app:endpoint2") + juju.integrate('your-app:endpoint1', 'other-app:endpoint2') # Ensure that both applications and all units reach a good state: juju.wait(jubilant.all_active) # Run an action on a unit: - result = juju.run("your-app/0", "some-action") - assert result.results["key"] == "value" + result = juju.run('your-app/0', 'some-action') + assert result.results['key'] == 'value' # What this means depends on the workload: assert charm_operates_correctly() @@ -233,10 +235,10 @@ def test_integrate(charm: pathlib.Path, juju: jubilant.Juju): A charm can require `file` or `oci-image` resources to work, which have revision numbers on Charmhub. OCI images can be referenced directly, while file resources are typically built during packing. ```python - ... - resources = {"resource_name": "localhost:32000/image_name:latest"} - juju.deploy(charm, resources=resources) - ... +... +resources = {'resource_name': 'localhost:32000/image_name:latest'} +juju.deploy(charm, resources=resources) +... ``` In `charmcraft.yaml`'s `resources` section, the `upstream-source` is, by convention, a usable resource that can be used in testing, allowing your integration test to look like this: @@ -249,14 +251,14 @@ import pytest import yaml -METADATA = yaml.safe_load(pathlib.Path("./charmcraft.yaml").read_text()) +METADATA = yaml.safe_load(pathlib.Path('./charmcraft.yaml').read_text()) @pytest.mark.juju_setup def test_deploy(charm: pathlib.Path, juju: jubilant.Juju): resources = { - name: res["upstream-source"] - for name, res in METADATA["resources"].items() + name: res['upstream-source'] + for name, res in METADATA['resources'].items() } juju.deploy(charm, resources=resources) @@ -273,7 +275,7 @@ def test_my_integration(charm: pathlib.Path, juju: jubilant.Juju): ... # Both applications have to be deployed at this point. # This could be done above in the current test or in a previous one. - juju.integrate("your-app:endpoint1", "another:relation_name_2") + juju.integrate('your-app:endpoint1', 'another:relation_name_2') juju.wait(jubilant.all_active) # check any assertion here ... @@ -292,7 +294,7 @@ You can set a configuration option in your application and check its results. ```python def test_config_changed(charm: pathlib.Path, juju: jubilant.Juju): ... - juju.config("your-app", {"server_name": "invalid_name"}) + juju.config('your-app', {'server_name': 'invalid_name'}) # In this case, when setting server_name to "invalid_name" # we could for example expect a blocked status. juju.wait(jubilant.all_blocked, timeout=60) @@ -309,9 +311,11 @@ You can execute an action on a unit and get its results. ```python def test_run_action(charm: pathlib.Path, juju: jubilant.Juju): - action_register_user = juju.run("your-app/0", "register-user", {"username": "ubuntu"}) - assert action_register_user.status == "completed" - password = action_register_user.results["user-password"] + action_register_user = juju.run( + 'your-app/0', 'register-user', {'username': 'ubuntu'} + ) + assert action_register_user.status == 'completed' + password = action_register_user.results['user-password'] # We could for example check here that we can login with the new user ``` @@ -326,11 +330,11 @@ You can get information from your application or unit addresses using `juju.stat ```python def test_workload_connectivity(charm: pathlib.Path, juju: jubilant.Juju): status = juju.status() - app_address = status.applications["my_app"].address + app_address = status.applications['my_app'].address # Or you can try to connect to a concrete unit # address = status.apps["my_app"].units["my_app/0"].public_address # address = status.apps["my_app"].units["my_app/0"].address - r = requests.get(f"http://{address}/") + r = requests.get(f'http://{address}/') assert r.status_code == 200 ``` @@ -345,13 +349,13 @@ How you can connect to a private or public address is dependent on your configur Jubilant provides an escape hatch to invoke the Juju CLI. This can be useful for cases where some feature is not covered. Some commands are global and others only make sense within a model scope: ```python - ... - command = ["add-credential", "some-cloud", "-f", "your-creds-file.yaml"] - stdout = juju.cli(*command) - ... - command = ["unexpose", "some-application"] - stdout = juju.cli(*command, include_model=True) - ... +... +command = ['add-credential', 'some-cloud', '-f', 'your-creds-file.yaml'] +stdout = juju.cli(*command) +... +command = ['unexpose', 'some-application'] +stdout = juju.cli(*command, include_model=True) +... ``` > See more: @@ -367,9 +371,9 @@ import pytest import pytest_jubilant -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def other_model(juju_factory: pytest_jubilant.JujuFactory): - return juju_factory.get_juju(suffix="other") + return juju_factory.get_juju(suffix='other') ``` Each call to `get_juju` creates a separate model. You can then use both `juju` and `other_model` in the same test. This is useful for cross-model scenarios. For example integrating machine charms with Kubernetes charms, or integrating with the [Canonical Observability Stack](https://charmhub.io/cos-lite). @@ -403,7 +407,7 @@ relations: """.strip() # Note that Juju from a snap doesn't have access to /tmp. - with NamedTemporaryFile(dir=".") as f: + with NamedTemporaryFile(dir='.') as f: f.write(bundle_yaml) f.flush() juju.deploy(f.name) diff --git a/docs/howto/write-unit-tests-for-a-charm.md b/docs/howto/write-unit-tests-for-a-charm.md index b0ef9a346..f5bca38d3 100644 --- a/docs/howto/write-unit-tests-for-a-charm.md +++ b/docs/howto/write-unit-tests-for-a-charm.md @@ -68,7 +68,7 @@ def test_pebble_ready_writes_config_file(): """Test that on pebble-ready, a config file is written.""" # Arrange: setting up the inputs ctx = testing.Context(MyCharm) - container = testing.Container(name="some-container", can_connect=True) + container = testing.Container(name='some-container', can_connect=True) state_in = testing.State( containers=[container], leader=True, @@ -78,11 +78,10 @@ def test_pebble_ready_writes_config_file(): state_out = ctx.run(ctx.on.pebble_ready(container=container), state_in) # Assert: - container_fs = state_out.get_container("some-container").get_filesystem(ctx) - cfg_file = container_fs / "etc" / "config.yaml" + container_fs = state_out.get_container('some-container').get_filesystem(ctx) + cfg_file = container_fs / 'etc' / 'config.yaml' config = yaml.safe_load(cfg_file.read_text()) - assert config["message"] == "Hello, world!" - + assert config['message'] == 'Hello, world!' ``` ```{note} @@ -158,7 +157,7 @@ from charm import MyCharm @pytest.fixture def my_charm(): - with patch("charm.lightkube.Client"): + with patch('charm.lightkube.Client'): yield MyCharm ``` @@ -189,7 +188,7 @@ relation = state_out.get_relation(...) # A relation we want to modify. # Copy and modify the relation data. new_local_app_data = relation.local_app_data.copy() -new_local_app_data["foo"] = "bar" +new_local_app_data['foo'] = 'bar' # Create a new State. new_relation = dataclasses.replace(relation, local_app_data=new_local_app_data) @@ -203,6 +202,7 @@ If you need to access the charm instance in a test, use the `testing.Context` in ```python # Charm code + class Charm(CharmBase): def workload_is_ready(self): ... # Some business logic. @@ -211,6 +211,7 @@ class Charm(CharmBase): # Testing code + def test_charm_reports_workload_ready(): ctx = testing.Context(Charm) state_in = testing.State(...) # Some state to represent a ready workload. diff --git a/docs/ruff.toml b/docs/ruff.toml new file mode 100644 index 000000000..1bbef1c13 --- /dev/null +++ b/docs/ruff.toml @@ -0,0 +1,2 @@ +extend = "../pyproject.toml" +line-length = 80 diff --git a/ops/framework.py b/ops/framework.py index 8cfdac753..b3e83da8e 100644 --- a/ops/framework.py +++ b/ops/framework.py @@ -16,7 +16,6 @@ from __future__ import annotations -import collections import collections.abc import inspect import keyword @@ -871,10 +870,10 @@ def _emit(self, event: EventBase): self._validate_snapshot_data(event, this_event_data) # TODO Track observers by (parent_path, event_kind) rather than as a list of # all observers. Avoiding linear search through all observers for every event - for observer_path, method_name, _parent_path, _event_kind in self._observers: - if _parent_path != parent_path: + for observer_path, method_name, parent_path_, event_kind_ in self._observers: + if parent_path_ != parent_path: continue - if _event_kind and _event_kind != event_kind: + if event_kind_ and event_kind_ != event_kind: continue if self.skip_duplicate_events and self._event_is_in_storage( observer_path, method_name, event_path, this_event_data @@ -950,10 +949,11 @@ def _event_context(self, event_name: str): old_hook_is_running = backend._hook_is_running backend._hook_is_running = event_name - yield - backend._hook_is_running = old_hook_is_running - - self._event_name = old_event_name + try: + yield + finally: + backend._hook_is_running = old_hook_is_running + self._event_name = old_event_name def _reemit(self, single_event_path: str | None = None): last_event_path = None diff --git a/ops/model.py b/ops/model.py index fa6af007d..6c98512d0 100644 --- a/ops/model.py +++ b/ops/model.py @@ -1456,14 +1456,14 @@ def get_content(self, *, refresh: bool = False) -> dict[str, str]: subsequent calls do not require querying the secret storage again, unless ``refresh=True`` is used, or :meth:`set_content` is called. - Returns: - A copy of the secret's content dictionary. - Args: refresh: If true, fetch the latest revision's content and tell Juju to update to tracking that revision. The default is to get the content of the currently-tracked revision. + Returns: + A copy of the secret's content dictionary. + Raises: SecretNotFoundError: if the secret no longer exists. ModelError: if the charm does not have permission to access the @@ -3112,7 +3112,7 @@ def _build_destpath( # /src --> /dst/src file_path, source_path, dest_dir = Path(file_path), Path(source_path), Path(dest_dir) prefix = str(source_path.parent) - if prefix != '.' and os.path.commonprefix([prefix, str(file_path)]) != prefix: + if prefix != '.' and os.path.commonpath([prefix, str(file_path)]) != prefix: raise RuntimeError(f'file "{file_path}" does not have specified prefix "{prefix}"') path_suffix = os.path.relpath(str(file_path), prefix) return dest_dir / path_suffix diff --git a/pyproject.toml b/pyproject.toml index 1ba1a69a7..a93521ce1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,7 +37,7 @@ harness = [] [dependency-groups] lint = [ - "ruff==0.11.2", + "ruff==0.15.14", "codespell==2.4.1", "zizmor==1.11.0", "ops[testing,tracing]", @@ -128,7 +128,7 @@ aggressive = 3 line-length = 99 exclude = ["examples"] target-version = "py310" -extend-exclude = ["docs/conf.py", "docs/_dev/"] +extend-exclude = ["docs/conf.py", "docs/_dev/", "docs/tutorial/"] # Ruff formatter configuration [tool.ruff.format] @@ -230,7 +230,11 @@ exclude = ["tracing/ops_tracing/vendor/*"] "S106", # "Useless" expression. - "B018" + "B018", + + # Unreliable floating point equality comparison. Tests check round-tripped + # exact-literal config values; pytest.approx isn't appropriate here. + "RUF069", ] "testing/tests/*" = [ # All documentation linting. @@ -246,6 +250,15 @@ exclude = ["tracing/ops_tracing/vendor/*"] "RUF001", # String contains ambiguous `µ` (MICRO SIGN). Did you mean `μ` (GREEK SMALL LETTER MU)? "RUF002", # Docstring contains ambiguous `µ` (MICRO SIGN). Did you mean `μ` (GREEK SMALL LETTER MU)? ] +"ops/__init__.py" = [ + "RUF067", # Wraps main() and imports ops_tracing at the bottom. +] +"ops/_private/__init__.py" = [ + "RUF067", # Exposes a module-level tracer. +] +"ops/lib/__init__.py" = [ + "RUF067", # Legacy library loader. +] "test/test_helpers.py" = [ "S605", # Starting a process with a shell: seems safe, but may be changed in the future; consider rewriting without `shell` "S607", # Starting a process with a partial executable path diff --git a/test/fake_pebble.py b/test/fake_pebble.py index 85f86d9bf..6ca560cb9 100644 --- a/test/fake_pebble.py +++ b/test/fake_pebble.py @@ -108,10 +108,10 @@ def internal_server_error(self, msg: Exception): } self.respond(d, 500) - def do_GET(self): # noqa: N802 ("should be lowercase") + def do_GET(self): self.do_request('GET') - def do_POST(self): # noqa: N802 ("should be lowercase") + def do_POST(self): self.do_request('POST') def do_request(self, request_method: typing.Literal['GET', 'POST']): diff --git a/test/integration/conftest.py b/test/integration/conftest.py index fad328c80..15cb8bb6c 100644 --- a/test/integration/conftest.py +++ b/test/integration/conftest.py @@ -46,8 +46,9 @@ def tracing_juju(juju: jubilant.Juju) -> Generator[jubilant.Juju]: juju.integrate('tempo:tempo-cluster', 'tempo-worker') juju.wait( - lambda status: jubilant.all_active(status, 'minio') - and jubilant.all_blocked(status, 's3-integrator') + lambda status: ( + jubilant.all_active(status, 'minio') and jubilant.all_blocked(status, 's3-integrator') + ) ) address = juju.status().apps['minio'].address diff --git a/test/test_framework.py b/test/test_framework.py index c335ee573..3a76af6f4 100644 --- a/test/test_framework.py +++ b/test/test_framework.py @@ -198,7 +198,8 @@ def restore(self, snapshot: dict[str, int]): framework3 = create_framework(request, tmpdir=tmp_path) framework3.register_type(Foo, None, handle.kind) - pytest.raises(NoSnapshotError, framework3.load_snapshot, handle) + with pytest.raises(NoSnapshotError): + framework3.load_snapshot(handle) def test_simple_event_observer(self, request: pytest.FixtureRequest): framework = create_framework(request) @@ -446,9 +447,12 @@ def on_any(self, event: ops.EventBase): assert ' '.join(obs2.seen) == 'a b c a b c a b c a c a c c' # Now the event objects must all be gone from storage. - pytest.raises(NoSnapshotError, framework.load_snapshot, ev_a_handle) - pytest.raises(NoSnapshotError, framework.load_snapshot, ev_b_handle) - pytest.raises(NoSnapshotError, framework.load_snapshot, ev_c_handle) + with pytest.raises(NoSnapshotError): + framework.load_snapshot(ev_a_handle) + with pytest.raises(NoSnapshotError): + framework.load_snapshot(ev_b_handle) + with pytest.raises(NoSnapshotError): + framework.load_snapshot(ev_c_handle) def test_repeated_defer(self, request: pytest.FixtureRequest): framework = create_framework(request) @@ -883,7 +887,8 @@ def _on_foo(self, event: ops.EventBase): # Register the type and check that the event is gone from storage. framework_copy.register_type(MyEvent, event_handle.parent, event_handle.kind) - pytest.raises(NoSnapshotError, framework_copy.load_snapshot, event_handle) + with pytest.raises(NoSnapshotError): + framework_copy.load_snapshot(event_handle) def test_auto_register_event_types(self, request: pytest.FixtureRequest): framework = create_framework(request) @@ -981,8 +986,10 @@ class NoneEvent(ops.EventBase): assert obs.seen == ['on_foo:MyFoo:foo', 'on_bar:MyBar:bar'] # Definitions remained local to the specific type. - pytest.raises(AttributeError, lambda: pub.on_a.bar) - pytest.raises(AttributeError, lambda: pub.on_b.foo) + with pytest.raises(AttributeError): + pub.on_a.bar + with pytest.raises(AttributeError): + pub.on_b.foo # Try to use an event name which is not a valid python identifier. with pytest.raises(RuntimeError): diff --git a/test/test_jujuversion.py b/test/test_jujuversion.py index 2a2bf8f2c..bc078cea4 100644 --- a/test/test_jujuversion.py +++ b/test/test_jujuversion.py @@ -45,7 +45,7 @@ def test_parsing(vs: str, major: int, minor: int, tag: str, patch: int, build: i assert v.build == build -@unittest.mock.patch('os.environ', new={}) +@unittest.mock.patch.dict(os.environ, clear=True) def test_from_environ(): # JUJU_VERSION is not set with pytest.deprecated_call(): diff --git a/test/test_storage.py b/test/test_storage.py index 45fac4da1..4929d4311 100644 --- a/test/test_storage.py +++ b/test/test_storage.py @@ -293,7 +293,8 @@ def test_permissions_race(self, exists: unittest.mock.MagicMock): # Create an existing file, but the mock will simulate a race condition saying that it # does not exist. open(filename, 'w').close() - pytest.raises(RuntimeError, ops.storage.SQLiteStorage, filename) + with pytest.raises(RuntimeError): + ops.storage.SQLiteStorage(filename) @unittest.mock.patch('os.chmod') def test_permissions_failure(self, chmod: unittest.mock.MagicMock): @@ -301,7 +302,8 @@ def test_permissions_failure(self, chmod: unittest.mock.MagicMock): with tempfile.TemporaryDirectory() as temp_dir: filename = os.path.join(temp_dir, '.unit-state.db') open(filename, 'w').close() - pytest.raises(RuntimeError, ops.storage.SQLiteStorage, filename) + with pytest.raises(RuntimeError): + ops.storage.SQLiteStorage(filename) def setup_juju_backend(fake_script: FakeScript, state_file: pathlib.Path): diff --git a/test/test_testing.py b/test/test_testing.py index aeb186fb2..3df7f84e6 100644 --- a/test/test_testing.py +++ b/test/test_testing.py @@ -412,7 +412,8 @@ def test_remove_relation(self, request: pytest.FixtureRequest): harness.remove_relation(rel_id) # Check relation no longer exists assert backend.relation_ids('db') == [] - pytest.raises(ops.RelationNotFoundError, backend.relation_list, rel_id) + with pytest.raises(ops.RelationNotFoundError): + backend.relation_list(rel_id) # Check relation broken event is raised with correct data changes = harness.charm.get_changes() assert changes[0] == { @@ -470,7 +471,8 @@ def test_remove_specific_relation_id(self, request: pytest.FixtureRequest): harness.remove_relation(rel_id_2) # Check second relation no longer exists but first does assert backend.relation_ids('db') == [rel_id_1] - pytest.raises(ops.RelationNotFoundError, backend.relation_list, rel_id_2) + with pytest.raises(ops.RelationNotFoundError): + backend.relation_list(rel_id_2) # Check relation broken event is raised with correct data changes = harness.charm.get_changes() @@ -603,9 +605,8 @@ def test_removing_relation_removes_remote_app_data(self, request: pytest.Fixture # Check relation and app data are removed assert backend.relation_ids('db') == [] with harness._event_context('foo'): - pytest.raises( - ops.RelationNotFoundError, backend.relation_get, rel_id, remote_app, is_app=True - ) + with pytest.raises(ops.RelationNotFoundError): + backend.relation_get(rel_id, remote_app, is_app=True) def test_removing_relation_refreshes_charm_model(self, request: pytest.FixtureRequest): # language=YAML @@ -703,7 +704,8 @@ def test_removing_relation_unit_removes_data_also(self, request: pytest.FixtureR # Check relation exists but unit and data are removed assert backend.relation_ids('db') == [rel_id] assert backend.relation_list(rel_id) == [] - pytest.raises(KeyError, backend.relation_get, rel_id, 'postgresql/0', is_app=False) + with pytest.raises(KeyError): + backend.relation_get(rel_id, 'postgresql/0', is_app=False) # Check relation departed was raised with correct data assert harness.charm.get_changes()[0] == { 'name': 'relation-departed', diff --git a/testing/README.md b/testing/README.md index c7c2b22e8..70959596f 100644 --- a/testing/README.md +++ b/testing/README.md @@ -19,6 +19,7 @@ from ops import testing # 'src/charm.py' typically contains the charm class. from charm import MyCharm + def test_start(): ctx = testing.Context(MyCharm) state_in = testing.State() @@ -38,6 +39,7 @@ from ops import testing from charm import MyCharm + @pytest.mark.parametrize( 'leader', [pytest.param(True, id='leader'), pytest.param(False, id='non-leader')], diff --git a/testing/UPGRADING.md b/testing/UPGRADING.md index 3ab8d3cc0..dab8941c1 100644 --- a/testing/UPGRADING.md +++ b/testing/UPGRADING.md @@ -35,11 +35,11 @@ The same applies to action events: ```python # Older Scenario code. -action = Action("backup", params={...}) +action = Action('backup', params={...}) ctx.run_action(action, state) # Scenario 7.x -ctx.run(ctx.on.action("backup", params={...}), state) +ctx.run(ctx.on.action('backup', params={...}), state) ``` ### Provide State components as (frozen) sets @@ -79,19 +79,19 @@ object, and if the charm calls `event.fail()`, an exception will be raised. ```python # Older Scenario Code -action = Action("backup", params={...}) +action = Action('backup', params={...}) out = ctx.run_action(action, state) -assert out.logs == ["baz", "qux"] +assert out.logs == ['baz', 'qux'] assert not out.success -assert out.results == {"foo": "bar"} -assert out.failure == "boo-hoo" +assert out.results == {'foo': 'bar'} +assert out.failure == 'boo-hoo' # Scenario 7.x with pytest.raises(ActionFailure) as exc_info: - ctx.run(ctx.on.action("backup", params={...}), State()) + ctx.run(ctx.on.action('backup', params={...}), State()) assert ctx.action_logs == ['baz', 'qux'] -assert ctx.action_results == {"foo": "bar"} -assert exc_info.value.message == "boo-hoo" +assert ctx.action_results == {'foo': 'bar'} +assert exc_info.value.message == 'boo-hoo' ``` ### Use the Context object as a context manager @@ -159,17 +159,14 @@ pass in an ID. # Older Scenario code. state = State( secrets=[ - scenario.Secret( - id='foo', - contents={0: {'certificate': 'xxxx'}} - ), + scenario.Secret(id='foo', contents={0: {'certificate': 'xxxx'}}), scenario.Secret( id='foo', contents={ 0: {'password': '1234'}, 1: {'password': 'abcd'}, 2: {'password': 'admin'}, - } + }, ), ] ) @@ -195,7 +192,7 @@ if you have a charm lib that will emit a `database-created` event on ```python # Older Scenario code. -ctx.run("my_charm_lib.on.database_created", state) +ctx.run('my_charm_lib.on.database_created', state) # Scenario 7.x ctx.run(ctx.on.relation_created(relation=relation), state) @@ -226,10 +223,10 @@ The resources in State objects were previously plain dictionaries, and are now ```python # Older Scenario code -state = State(resources={"/path/to/foo", pathlib.Path("/mock/foo")}) +state = State(resources={'/path/to/foo', pathlib.Path('/mock/foo')}) # Scenario 7.x -resource = Resource(location="/path/to/foo", source=pathlib.Path("/mock/foo")) +resource = Resource(location='/path/to/foo', source=pathlib.Path('/mock/foo')) state = State(resources={resource}) ``` @@ -242,10 +239,10 @@ requires a binding name to be passed in when it is created. ```python # Older Scenario code -state = State(networks={"foo": Network.default()}) +state = State(networks={'foo': Network.default()}) # Scenario 7.x -state = State(networks={Network.default("foo")}) +state = State(networks={Network.default('foo')}) ``` ### Use the .deferred() method to populate State.deferred @@ -260,16 +257,18 @@ and so on). ```python # Older Scenario code deferred_start = scenario.deferred('start', handler=MyCharm._on_start) -deferred_relation_created = Relation('foo').changed_event.deferred(handler=MyCharm._on_foo_relation_changed) +deferred_relation_created = Relation('foo').changed_event.deferred( + handler=MyCharm._on_foo_relation_changed +) deferred_config_changed = DeferredEvent( - handle_path='MyCharm/on/config_changed[1]', - owner='MyCharm', - observer='_on_config_changed' + handle_path='MyCharm/on/config_changed[1]', owner='MyCharm', observer='_on_config_changed' ) # Scenario 7.x deferred_start = ctx.on.start().deferred(handler=MyCharm._on_start) -deferred_relation_changed = ctx.on.relation_changed(Relation('foo')).deferred(handler=MyCharm._on_foo_relation_changed) +deferred_relation_changed = ctx.on.relation_changed(Relation('foo')).deferred( + handler=MyCharm._on_foo_relation_changed +) deferred_config_changed = ctx.on.config_changed().deferred(handler=MyCharm._on_config_changed) ``` diff --git a/testing/tests/test_charm_spec_autoload.py b/testing/tests/test_charm_spec_autoload.py index b087668c4..dbf6a05bd 100644 --- a/testing/tests/test_charm_spec_autoload.py +++ b/testing/tests/test_charm_spec_autoload.py @@ -5,7 +5,7 @@ import importlib import sys -from collections.abc import Iterator +from collections.abc import Generator from contextlib import contextmanager from pathlib import Path from typing import Any @@ -26,14 +26,16 @@ class MyCharm(CharmBase): pass @contextmanager -def import_name(name: str, source: Path) -> Iterator[type[CharmBase]]: +def import_name(name: str, source: Path) -> Generator[type[CharmBase]]: pkg_path = str(source.parent) sys.path.append(pkg_path) charm = importlib.import_module('mycharm') obj: type[CharmBase] = getattr(charm, name) sys.path.remove(pkg_path) - yield obj - del sys.modules['mycharm'] + try: + yield obj + finally: + del sys.modules['mycharm'] @contextmanager @@ -45,7 +47,7 @@ def create_tempcharm( config: dict[str, Any] | None = None, name: str = 'MyCharm', legacy: bool = False, -) -> Iterator[type[CharmBase]]: +) -> Generator[type[CharmBase]]: src = root / 'src' src.mkdir(parents=True) charmpy = src / 'mycharm.py' diff --git a/uv.lock b/uv.lock index 3db5c63ee..c0d1036aa 100644 --- a/uv.lock +++ b/uv.lock @@ -927,7 +927,7 @@ lint = [ { name = "codespell", specifier = "==2.4.1" }, { name = "ops", extras = ["testing", "tracing"], editable = "." }, { name = "pyright", specifier = "~=1.1" }, - { name = "ruff", specifier = "==0.11.2" }, + { name = "ruff", specifier = "==0.15.14" }, { name = "typing-extensions", specifier = "~=4.2" }, { name = "zizmor", specifier = "==1.11.0" }, ] @@ -1334,15 +1334,15 @@ wheels = [ [[package]] name = "pyright" -version = "1.1.406" +version = "1.1.410" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "nodeenv" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/f7/16/6b4fbdd1fef59a0292cbb99f790b44983e390321eccbc5921b4d161da5d1/pyright-1.1.406.tar.gz", hash = "sha256:c4872bc58c9643dac09e8a2e74d472c62036910b3bd37a32813989ef7576ea2c", size = 4113151, upload-time = "2025-10-02T01:04:45.488Z" } +sdist = { url = "https://files.pythonhosted.org/packages/10/53/e4d8ea1391bd4355231be6f91bf239479aa0014260ed3fb5526eeb12a1f2/pyright-1.1.410.tar.gz", hash = "sha256:07a073b8ba6749826773c1269773efa11b93440d9a6aa60419d9a3172d6dc488", size = 4062013, upload-time = "2026-06-01T17:35:48.894Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/f6/a2/e309afbb459f50507103793aaef85ca4348b66814c86bc73908bdeb66d12/pyright-1.1.406-py3-none-any.whl", hash = "sha256:1d81fb43c2407bf566e97e57abb01c811973fdb21b2df8df59f870f688bdca71", size = 5980982, upload-time = "2025-10-02T01:04:43.137Z" }, + { url = "https://files.pythonhosted.org/packages/d7/33/288b5868fa00846dacf249633719d747893e54aebd196b9968ac1878a5d3/pyright-1.1.410-py3-none-any.whl", hash = "sha256:5e961bed37cacf96b3f7cd7b1da39b350a9239aa2e69138d0e88f728cfaf296c", size = 6082448, upload-time = "2026-06-01T17:35:46.387Z" }, ] [[package]] @@ -1483,27 +1483,27 @@ wheels = [ [[package]] name = "ruff" -version = "0.11.2" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/90/61/fb87430f040e4e577e784e325351186976516faef17d6fcd921fe28edfd7/ruff-0.11.2.tar.gz", hash = "sha256:ec47591497d5a1050175bdf4e1a4e6272cddff7da88a2ad595e1e326041d8d94", size = 3857511, upload-time = "2025-03-21T13:31:17.419Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/62/99/102578506f0f5fa29fd7e0df0a273864f79af044757aef73d1cae0afe6ad/ruff-0.11.2-py3-none-linux_armv6l.whl", hash = "sha256:c69e20ea49e973f3afec2c06376eb56045709f0212615c1adb0eda35e8a4e477", size = 10113146, upload-time = "2025-03-21T13:30:26.68Z" }, - { url = "https://files.pythonhosted.org/packages/74/ad/5cd4ba58ab602a579997a8494b96f10f316e874d7c435bcc1a92e6da1b12/ruff-0.11.2-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:2c5424cc1c4eb1d8ecabe6d4f1b70470b4f24a0c0171356290b1953ad8f0e272", size = 10867092, upload-time = "2025-03-21T13:30:37.949Z" }, - { url = "https://files.pythonhosted.org/packages/fc/3e/d3f13619e1d152c7b600a38c1a035e833e794c6625c9a6cea6f63dbf3af4/ruff-0.11.2-py3-none-macosx_11_0_arm64.whl", hash = "sha256:ecf20854cc73f42171eedb66f006a43d0a21bfb98a2523a809931cda569552d9", size = 10224082, upload-time = "2025-03-21T13:30:39.962Z" }, - { url = "https://files.pythonhosted.org/packages/90/06/f77b3d790d24a93f38e3806216f263974909888fd1e826717c3ec956bbcd/ruff-0.11.2-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0c543bf65d5d27240321604cee0633a70c6c25c9a2f2492efa9f6d4b8e4199bb", size = 10394818, upload-time = "2025-03-21T13:30:42.551Z" }, - { url = "https://files.pythonhosted.org/packages/99/7f/78aa431d3ddebfc2418cd95b786642557ba8b3cb578c075239da9ce97ff9/ruff-0.11.2-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:20967168cc21195db5830b9224be0e964cc9c8ecf3b5a9e3ce19876e8d3a96e3", size = 9952251, upload-time = "2025-03-21T13:30:45.196Z" }, - { url = "https://files.pythonhosted.org/packages/30/3e/f11186d1ddfaca438c3bbff73c6a2fdb5b60e6450cc466129c694b0ab7a2/ruff-0.11.2-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:955a9ce63483999d9f0b8f0b4a3ad669e53484232853054cc8b9d51ab4c5de74", size = 11563566, upload-time = "2025-03-21T13:30:47.516Z" }, - { url = "https://files.pythonhosted.org/packages/22/6c/6ca91befbc0a6539ee133d9a9ce60b1a354db12c3c5d11cfdbf77140f851/ruff-0.11.2-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:86b3a27c38b8fce73bcd262b0de32e9a6801b76d52cdb3ae4c914515f0cef608", size = 12208721, upload-time = "2025-03-21T13:30:49.56Z" }, - { url = "https://files.pythonhosted.org/packages/19/b0/24516a3b850d55b17c03fc399b681c6a549d06ce665915721dc5d6458a5c/ruff-0.11.2-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:a3b66a03b248c9fcd9d64d445bafdf1589326bee6fc5c8e92d7562e58883e30f", size = 11662274, upload-time = "2025-03-21T13:30:52.055Z" }, - { url = "https://files.pythonhosted.org/packages/d7/65/76be06d28ecb7c6070280cef2bcb20c98fbf99ff60b1c57d2fb9b8771348/ruff-0.11.2-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:0397c2672db015be5aa3d4dac54c69aa012429097ff219392c018e21f5085147", size = 13792284, upload-time = "2025-03-21T13:30:54.24Z" }, - { url = "https://files.pythonhosted.org/packages/ce/d2/4ceed7147e05852876f3b5f3fdc23f878ce2b7e0b90dd6e698bda3d20787/ruff-0.11.2-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:869bcf3f9abf6457fbe39b5a37333aa4eecc52a3b99c98827ccc371a8e5b6f1b", size = 11327861, upload-time = "2025-03-21T13:30:56.757Z" }, - { url = "https://files.pythonhosted.org/packages/c4/78/4935ecba13706fd60ebe0e3dc50371f2bdc3d9bc80e68adc32ff93914534/ruff-0.11.2-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:2a2b50ca35457ba785cd8c93ebbe529467594087b527a08d487cf0ee7b3087e9", size = 10276560, upload-time = "2025-03-21T13:30:58.881Z" }, - { url = "https://files.pythonhosted.org/packages/81/7f/1b2435c3f5245d410bb5dc80f13ec796454c21fbda12b77d7588d5cf4e29/ruff-0.11.2-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:7c69c74bf53ddcfbc22e6eb2f31211df7f65054bfc1f72288fc71e5f82db3eab", size = 9945091, upload-time = "2025-03-21T13:31:01.45Z" }, - { url = "https://files.pythonhosted.org/packages/39/c4/692284c07e6bf2b31d82bb8c32f8840f9d0627d92983edaac991a2b66c0a/ruff-0.11.2-py3-none-musllinux_1_2_i686.whl", hash = "sha256:6e8fb75e14560f7cf53b15bbc55baf5ecbe373dd5f3aab96ff7aa7777edd7630", size = 10977133, upload-time = "2025-03-21T13:31:04.013Z" }, - { url = "https://files.pythonhosted.org/packages/94/cf/8ab81cb7dd7a3b0a3960c2769825038f3adcd75faf46dd6376086df8b128/ruff-0.11.2-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:842a472d7b4d6f5924e9297aa38149e5dcb1e628773b70e6387ae2c97a63c58f", size = 11378514, upload-time = "2025-03-21T13:31:06.166Z" }, - { url = "https://files.pythonhosted.org/packages/d9/3a/a647fa4f316482dacf2fd68e8a386327a33d6eabd8eb2f9a0c3d291ec549/ruff-0.11.2-py3-none-win32.whl", hash = "sha256:aca01ccd0eb5eb7156b324cfaa088586f06a86d9e5314b0eb330cb48415097cc", size = 10319835, upload-time = "2025-03-21T13:31:10.7Z" }, - { url = "https://files.pythonhosted.org/packages/86/54/3c12d3af58012a5e2cd7ebdbe9983f4834af3f8cbea0e8a8c74fa1e23b2b/ruff-0.11.2-py3-none-win_amd64.whl", hash = "sha256:3170150172a8f994136c0c66f494edf199a0bbea7a409f649e4bc8f4d7084080", size = 11373713, upload-time = "2025-03-21T13:31:13.148Z" }, - { url = "https://files.pythonhosted.org/packages/d6/d4/dd813703af8a1e2ac33bf3feb27e8a5ad514c9f219df80c64d69807e7f71/ruff-0.11.2-py3-none-win_arm64.whl", hash = "sha256:52933095158ff328f4c77af3d74f0379e34fd52f175144cefc1b192e7ccd32b4", size = 10441990, upload-time = "2025-03-21T13:31:15.206Z" }, +version = "0.15.14" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/dc/8a/8bce2894573e9dae6ff4d77fe34ad727d79b9e6238ad288c5638990d90f6/ruff-0.15.14.tar.gz", hash = "sha256:48e866b165be4a9bdbf310f7d3c9a07edef2fe8cd63ffeb4e00bb590506ebf9f", size = 4700910, upload-time = "2026-05-21T14:34:55.177Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b9/c8/74a92c6ff9fcfb4f1f947126d3ebee8389276e161ecc85de5bda7cda51bd/ruff-0.15.14-py3-none-linux_armv6l.whl", hash = "sha256:8dd2db9416e487c8d4b01fa7056bb02c4d05969d4f8d17a08c229c2f4ff3c108", size = 10739177, upload-time = "2026-05-21T14:34:37.332Z" }, + { url = "https://files.pythonhosted.org/packages/45/91/254a35c20acc38a7223c9d2d594af12e794432464f2cdeb52af1dc4a892d/ruff-0.15.14-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:be4ff55af755bd71a00ab3dc6bd7ffc467bd76e0df6881e286c2e3d23e8fb43b", size = 11144969, upload-time = "2026-05-21T14:34:43.978Z" }, + { url = "https://files.pythonhosted.org/packages/56/9e/d13e40f83b8d0a94430e6778ce1d94a43b38cf2efe63278bdd2b4c65abbf/ruff-0.15.14-py3-none-macosx_11_0_arm64.whl", hash = "sha256:48d5909d7d06276ce7dde6d32bfa4b0d4cb2651145cd8ee4b440722cbc77832f", size = 10478207, upload-time = "2026-05-21T14:34:48.378Z" }, + { url = "https://files.pythonhosted.org/packages/8d/f1/b15a7839fa4f332f8acec78e20564f26bb2d866e3d21710b877fd0263000/ruff-0.15.14-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:ca8cbfa94c4f90984a67561978602746d4cd27103568f745fa90eee3f0d4107d", size = 10818459, upload-time = "2026-05-21T14:34:22.318Z" }, + { url = "https://files.pythonhosted.org/packages/45/33/53d651177f84f94b400a0e27f8824eeada3dddc9d5ee8aeb048f4352a520/ruff-0.15.14-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:9a6bbc0333f1ab053423bcbf6226477d266ca7cec7738c4c8e3f55647803f3c4", size = 10541800, upload-time = "2026-05-21T14:34:20.209Z" }, + { url = "https://files.pythonhosted.org/packages/b8/a6/868f87e0bf9786ed24b5d0d0ad8676b8a94fd1912f42cddf9cfc7857818a/ruff-0.15.14-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:8a24a4f7605d7003a6674d4387651effd939dead3fddd0f36561eb77a9a2e542", size = 11342149, upload-time = "2026-05-21T14:34:46.365Z" }, + { url = "https://files.pythonhosted.org/packages/a7/8b/38cd5c19faffdcc05a408d2b78edccc69492ab9720eadb49ea15ef80d768/ruff-0.15.14-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:049b5326e53ed80978f2fc041a280603f69dd6b0c95464342a2bb4572d9d9e2f", size = 12212563, upload-time = "2026-05-21T14:34:28.579Z" }, + { url = "https://files.pythonhosted.org/packages/3e/4d/a3c5b874a556d5731e3e657aaf04311bb76f0a5c3ec220ed43051be6b64b/ruff-0.15.14-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:d4ed42e6696c8dfa5f06728e6441993901f548eb92d73bc472cb5a38d1395fbf", size = 11493299, upload-time = "2026-05-21T14:34:41.836Z" }, + { url = "https://files.pythonhosted.org/packages/1e/c0/56472c251d09858a53e51efbd485b09e1995d8731668b76d52e5dd6ee0f1/ruff-0.15.14-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:715c543cf450c4888251f91c52f1942a800541d9bddd7ac060aa4e6b77ae7cba", size = 11455931, upload-time = "2026-05-21T14:34:57.276Z" }, + { url = "https://files.pythonhosted.org/packages/2c/4a/e2e7b4d8dbf233d4eace59c75bc3435fa6d8bd3bae82d351d4e4300c0fd1/ruff-0.15.14-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:72ebab6013ec887d439d8b7593737a0a4ffb06d45d209d4e4bf2e92813082d3f", size = 11400794, upload-time = "2026-05-21T14:34:39.773Z" }, + { url = "https://files.pythonhosted.org/packages/97/c7/83c0539fe34c3e09136204d1e75d6052492364e0b3cb05e9465423f567d7/ruff-0.15.14-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:49072d36abdbe97a8dd7f480afe9c675699c0c495d4c84076e2c1203c4550581", size = 10804759, upload-time = "2026-05-21T14:34:31.045Z" }, + { url = "https://files.pythonhosted.org/packages/86/a6/18f2bfc095a2ab4a78745644e428205532ce6653a5d0fa8501572891534d/ruff-0.15.14-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:958522aee105068640c2c2ceae08f413ae44d922f52a1374ac13d6a96032fc93", size = 10539517, upload-time = "2026-05-21T14:34:53.064Z" }, + { url = "https://files.pythonhosted.org/packages/54/3a/5a8b3b69c654d4e4bf1d246ac5b49cbcdac6eaab6905925f8915f31e3b80/ruff-0.15.14-py3-none-musllinux_1_2_i686.whl", hash = "sha256:f3707da619a143a2e8830e2abab8224478d69ace2d28cb6c20543ae97c36bf61", size = 11065169, upload-time = "2026-05-21T14:34:24.484Z" }, + { url = "https://files.pythonhosted.org/packages/ed/c5/8864e4e7925b836ea354b31d57641ec03830564e281a8b6f061f8c3e0ec1/ruff-0.15.14-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:bb01d645694e3ec0102105d07ef2d53703970407d59c04e59d3ba0b7a1d53553", size = 11560214, upload-time = "2026-05-21T14:34:50.975Z" }, + { url = "https://files.pythonhosted.org/packages/36/38/012bf76752e1f89ed50b77b99532d90f3a3e287bc7918e1fc0948ac866ac/ruff-0.15.14-py3-none-win32.whl", hash = "sha256:6d0c1ad2a0ab718d39b6d8fd2217981ce4d625cd96a720095f798fb47d8b13e6", size = 10805548, upload-time = "2026-05-21T14:34:33.453Z" }, + { url = "https://files.pythonhosted.org/packages/d1/b7/4ea2c170f10ad760fff2a5250beb18897719dc8b52b53a24cddbb9dd3f19/ruff-0.15.14-py3-none-win_amd64.whl", hash = "sha256:802342981e056db3851a7836e5b070f8f15f67d4a685ae2a6160939d364b2902", size = 11939523, upload-time = "2026-05-21T14:34:18.077Z" }, + { url = "https://files.pythonhosted.org/packages/62/d5/bc97ff895ec35cf3925d4bd60f3b39d822f377a446906ec9bcc87405e59b/ruff-0.15.14-py3-none-win_arm64.whl", hash = "sha256:ff47b90a9ef6a40c9e2f3b479c1fb78531adf055b94c1eba0a7ba04b31951826", size = 11208607, upload-time = "2026-05-21T14:34:26.525Z" }, ] [[package]]