Allow switch kwarg in refresh to switch to local charms

juju/application.py

@@ -4,15 +4,15 @@
 import hashlib
 import json
 import logging
-import pathlib
+from pathlib import Path
 
 from . import model, tag, utils, jasyncio
 from .url import URL
 from .status import derive_status
 from .annotationhelper import _get_annotations, _set_annotations
 from .client import client
 from .errors import JujuError, JujuApplicationConfigError
-from .bundle import get_charm_series
+from .bundle import get_charm_series, is_local_charm
 from .placement import parse as parse_placement
 from .origin import Channel, Source
 
@@ -661,6 +661,8 @@ async def refresh(
         :param str switch: Crossgrade charm url
 
         """
+        if switch is not None and path is not None:
+            raise ValueError("switch and path are mutually exclusive")
 
         if switch is not None and revision is not None:
             raise ValueError("switch and revision are mutually exclusive")
@@ -677,17 +679,18 @@ async def refresh(
         if charm_url_origin_result.error is not None:
             err = charm_url_origin_result.error
             raise JujuError(f'{err.code} : {err.message}')
-        charm_url = switch or charm_url_origin_result.url
         origin = charm_url_origin_result.charm_origin
 
-        if path is not None:
+        if path is not None or (switch is not None and is_local_charm(switch)):
             await self.local_refresh(origin, force, force_series,
-                                     force_units, path, resources)
+                                     force_units, path or switch, resources)
             return
 
         if resources is not None:
             raise NotImplementedError("resources option is not implemented")
 
+        # If switch is not None at this point, that means it's a switch to a store charm
+        charm_url = switch or charm_url_origin_result.url
         parsed_url = URL.parse(charm_url)
         charm_name = parsed_url.name
 
@@ -733,8 +736,6 @@ async def refresh(
             err = charm_origin_result.error
             raise JujuError(f'{err.code} : {err.message}')
 
-        # Now take care of the resources:
-
         # Already prepped the charm_resources
         # Now get the existing resources from the ResourcesFacade
         request_data = [client.Entity(self.tag)]
@@ -808,22 +809,22 @@ async def local_refresh(
             path=None, resources=None):
         """Refresh the charm for this application with a local charm.
 
-        :param str channel: Channel to use when getting the charm from the
-            charm store, e.g. 'development'
+        :param dict charm_origin: The charm origin of the destination charm
+            we're refreshing to
+        :param bool force: Refresh even if validation checks fail
         :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
 
         """
         app_facade = self._facade()
 
-        if not isinstance(path, pathlib.Path):
-            path = pathlib.Path(path)
+        if isinstance(path, str) and path.startswith("local:"):
+            path = path[6:]
+        path = Path(path)
         charm_dir = path.expanduser().resolve()
         model_config = await self.get_config()
 

tests/integration/test_application.py

@@ -219,6 +219,21 @@ async def test_upgrade_charm_resource_same_rev_no_update(event_loop):
 
 
 @base.bootstrapped
+@pytest.mark.asyncio
+async def test_refresh_charmhub_to_local(event_loop):
+    charm_path = INTEGRATION_TEST_DIR / 'charm'
+    async with base.CleanModel() as model:
+        app = await model.deploy('ubuntu', application_name='ubu-path')
+        await app.refresh(path=str(charm_path))
+        assert app.data['charm-url'].startswith('local:')
+
+        app = await model.deploy('ubuntu', application_name='ubu-switch')
+        await app.refresh(switch=str(charm_path))
+        assert app.data['charm-url'].startswith('local:')
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
 async def test_trusted(event_loop):
     async with base.CleanModel() as model:
         await model.deploy('ubuntu', trust=True)

tests/unit/test_application.py

@@ -166,3 +166,14 @@ async def test_unexpose_endpoints_on_29_controller(self, mock_conn):
             application="panther",
             exposed_endpoints=["alpha", "beta"]
         )
+
+
+class TestRefreshApplication(unittest.IsolatedAsyncioTestCase):
+    @mock.patch("juju.model.Model.connection")
+    async def test_refresh_mutually_exclusive_kwargs(self, mock_conn):
+        app = Application(entity_id="app-id", model=Model())
+        with self.assertRaises(ValueError):
+            await app.refresh(switch="charm1", revision=10)
+
+        with self.assertRaises(ValueError):
+            await app.refresh(switch="charm1", path="/path/to/charm2")