fix: redo the local_refresh method signature

Author: dimaqq

juju/application.py

@@ -4,15 +4,15 @@
 import hashlib
 import json
 import logging
-from typing import Dict, List, Union
+from typing import Dict, List, Optional, Union
 from pathlib import Path
 
 from typing_extensions import deprecated
 
 from . import jasyncio, model, tag, utils
 from .annotationhelper import _get_annotations, _set_annotations
 from .bundle import get_charm_series, is_local_charm
-from .client import client
+from .client import client, _definitions
 from .errors import JujuApplicationConfigError, JujuError
 from .origin import Channel
 from .placement import parse as parse_placement
@@ -706,20 +706,28 @@ async def set_constraints(self, constraints):
         return await app_facade.SetConstraints(application=self.name, constraints=constraints)
 
     async def refresh(
-            self, channel=None, force=False, force_series=False, force_units=False,
-            path=None, resources=None, revision=None, switch=None):
+            self,
+            channel: Optional[str] = None,
+            force: bool = False,
+            force_series: bool = False,
+            force_units: bool = False,
+            path: Optional[Union[Path, str]] = None,
+            resources: Optional[Dict[str, str]] = None,
+            revision: Optional[int] = None,
+            switch: Optional[str] = None,
+    ):
         """Refresh the charm for this application.
 
-        :param str channel: Channel to use when getting the charm from the
+        :param str|None channel: Channel to use when getting the charm from the
             charm store, e.g. 'development'
         :param bool force_series: Refresh even if series of deployed
             application is not supported by the new charm
         :param bool force_units: Refresh all units immediately, even if in
             error state
-        :param str path: Refresh to a charm located at path
-        :param dict resources: Dictionary of resource name/filepath pairs
-        :param int revision: Explicit refresh revision
-        :param str switch: Crossgrade charm url
+        :param Path|str|None path: Refresh to a charm located at path
+        :param dict[str,str]|None resources: Dictionary of resource name/filepath pairs
+        :param int|None revision: Explicit refresh revision
+        :param str|None switch: URL of a different charm to cross-grade to
 
         """
         if switch is not None and path is not None:
@@ -743,8 +751,17 @@ async def refresh(
 
         current_origin = charm_url_origin_result.charm_origin
         if path is not None or (switch is not None and is_local_charm(switch)):
-            await self.local_refresh(current_origin, force, force_series,
-                                     force_units, path or switch, resources)
+            local_path = path or switch
+            assert local_path
+
+            await self.local_refresh(
+                charm_origin=current_origin,
+                force=force,
+                force_series=force_series,
+                force_units=force_units,
+                path=local_path,
+                resources=resources,
+            )
             return
 
         origin = _refresh_origin(current_origin, channel, revision)
@@ -870,9 +887,15 @@ async def refresh(
     upgrade_charm = refresh
 
     async def local_refresh(
-            self, charm_origin=None, force=False, force_series=False,
-            force_units=False,
-            path=None, resources=None):
+            self,
+            *,
+            charm_origin: _definitions.CharmOrigin,
+            force: bool,
+            force_series: bool,
+            force_units: bool,
+            path: Union[Path, str],
+            resources: Optional[Dict[str, str]],
+    ):
         """Refresh the charm for this application with a local charm.
 
         :param dict charm_origin: The charm origin of the destination charm
@@ -882,16 +905,16 @@ async def local_refresh(
             application is not supported by the new charm
         :param bool force_units: Refresh all units immediately, even if in
             error state
-        :param str path: Refresh to a charm located at path
+        :param Path|str path: Refresh to a charm located at path
         :param dict resources: Dictionary of resource name/filepath pairs
 
         """
         app_facade = self._facade()
 
         if isinstance(path, str) and path.startswith("local:"):
             path = path[6:]
-        path = Path(path)
-        charm_dir = path.expanduser().resolve()
+        local_path = Path(path)
+        charm_dir = local_path.expanduser().resolve()
         model_config = await self.get_config()
 
         series = (
@@ -907,7 +930,7 @@ async def local_refresh(
             if default_series:
                 series = default_series.value
         charm_url = await self.model.add_local_charm_dir(charm_dir, series)
-        metadata = utils.get_local_charm_metadata(path)
+        metadata = utils.get_local_charm_metadata(local_path)
         if resources is not None:
             resources = await self.model.add_local_resources(self.entity_id,
                                                              charm_url,
@@ -956,14 +979,17 @@ async def get_metrics(self):
         return await self.model.get_metrics(self.tag)
 
 
-def _refresh_origin(current_origin: client.CharmOrigin, channel=None, revision=None) -> client.CharmOrigin:
-    if channel is not None:
-        channel = Channel.parse(channel).normalize()
+def _refresh_origin(
+        current_origin: client.CharmOrigin,
+        channel: Optional[str] = None,
+        revision: Optional[int] = None,
+) -> client.CharmOrigin:
+    chan = None if channel is None else Channel.parse(channel).normalize()
 
     return client.CharmOrigin(
         source=current_origin.source,
-        track=channel.track if channel else current_origin.track,
-        risk=channel.risk if channel else current_origin.risk,
+        track=chan.track if chan else current_origin.track,
+        risk=chan.risk if chan else current_origin.risk,
         revision=revision if revision is not None else current_origin.revision,
         base=current_origin.base,
         architecture=current_origin.get('architecture', DEFAULT_ARCHITECTURE),

juju/bundle.py

@@ -440,7 +440,7 @@ def resolve(self, reference):
         return reference
 
 
-def is_local_charm(charm_url):
+def is_local_charm(charm_url: str):
     return charm_url.startswith('.') or charm_url.startswith('local:') or os.path.isabs(charm_url)
 
 

juju/model.py

@@ -2013,8 +2013,8 @@ async def add_local_resources(self, application, entity_url, metadata, resources
         :param str application: the name of the application
         :param client.CharmURL entity_url: url for the charm that we add resources for
         :param [string]string metadata: metadata for the charm that we add resources for
-        :param [string] resources: the paths for the local files (or oci-images) to be added as
-        local resources
+        :param dict[str, str] resources: the paths for the local files (or oci-images) to
+        be added as local resources
 
         :returns [string]string resource_map that is a map of resources to their assigned
         pendingIDs.

juju/origin.py

@@ -69,7 +69,7 @@ def __init__(self, track=None, risk=None):
         self.risk = risk
 
     @staticmethod
-    def parse(s):
+    def parse(s: str):
         """parse a channel from a given string.
         Parse does not take into account branches.
 

tests/integration/test_application.py

@@ -280,7 +280,14 @@ async def test_local_refresh():
                                     branch="deadbeef", hash_="hash", id_="id", revision=12,
                                     base=client.Base("20.04", "ubuntu"))
 
-        await app.local_refresh(charm_origin=origin, path=charm_path)
+        await app.local_refresh(
+            charm_origin=origin,
+            force=False,
+            force_series=False,
+            force_units=False,
+            path=charm_path,
+            resources=None,
+        )
 
         assert origin == client.CharmOrigin(source="local", revision=0,
                                             base=client.Base("20.04", "ubuntu"))