We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Internet connectivity check fails if ICMP traffic is restricted
<!--
   Thank you for submitting an issue. Please fill in the template below
   information about the bug you encountered.
-->

#### Summary
<!-- Please explain the bug in a few short sentences -->
When trying to install on macOS I get a failure to connect to the internet, the check is done via a ping (ICMP) traffic which isn't allowed on my corp network. However http/https (which are the connections actually required) are fine. I can't see any other use of ping other than a test.

#### What Should Happen Instead?
<!-- Please explain what the expected behavior is -->
The internet check should check for http(s) connectivity instead of using ping.

#### Reproduction Steps
<!-- Are you able to consistently reproduce the issue? Please add a list of steps that lead to the bug. -->

1. be on network that disallows ICMP traffic
2. microk8s install

#### Introspection Report
<!-- Please run `microk8s inspect` and attach the generated tarball. -->
N/A

#### Can you suggest a fix?
<!-- (This section is optional). How do you propose that the issue be fixed? -->
https://github.com/canonical/microk8s/blob/master/installer/vm_providers/_base_provider.py#L131

`curl --fail --head https://snapcraft.io` instead perhaps?

#### Are you interested in contributing with a fix?
<!-- yes/no, or @mention maintainers. Community contributions are welcome. -->
Yes

<!-- Thank you for making MicroK8s better -->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `installer/vm_providers/_base_provider.py`
Content:
```
1 # -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
2 #
3 # Copyright (C) 2018-2019 Canonical Ltd
4 #
5 # This program is free software: you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License version 3 as
7 # published by the Free Software Foundation.
8 #
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12 # GNU General Public License for more details.
13 #
14 # You should have received a copy of the GNU General Public License
15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
16 
17 import abc
18 import logging
19 import os
20 import pathlib
21 import shlex
22 import sys
23 from typing import Dict
24 from typing import Optional, Sequence
25 
26 from . import errors
27 from ._multipass._instance_info import InstanceInfo
28 
29 logger = logging.getLogger(__name__)
30 
31 
32 class Provider(abc.ABC):
33     def __init__(
34         self,
35         *,
36         echoer,
37         is_ephemeral: bool = False,
38         build_provider_flags: Dict[str, str] = None,
39     ) -> None:
40         self.echoer = echoer
41         self._is_ephemeral = is_ephemeral
42 
43         self.instance_name = "microk8s-vm"
44 
45         if build_provider_flags is None:
46             build_provider_flags = dict()
47         self.build_provider_flags = build_provider_flags.copy()
48 
49         self._cached_home_directory: Optional[pathlib.Path] = None
50 
51     @classmethod
52     def ensure_provider(cls) -> None:
53         """Necessary steps to ensure the provider is correctly setup."""
54 
55     @classmethod
56     def setup_provider(cls, *, echoer) -> None:
57         """Necessary steps to install the provider on the host."""
58 
59     @classmethod
60     def _get_provider_name(cls) -> str:
61         """Return the provider name."""
62 
63     @classmethod
64     def _get_is_snap_injection_capable(cls) -> bool:
65         """Return whether the provider can install snaps from the host."""
66 
67     @abc.abstractmethod
68     def create(self) -> None:
69         """Provider steps needed to create a fully functioning environment."""
70 
71     @abc.abstractmethod
72     def destroy(self) -> None:
73         """Provider steps needed to ensure the instance is destroyed.
74 
75         This method should be safe to call multiple times and do nothing
76         if the instance to destroy is already destroyed.
77         """
78 
79     @abc.abstractmethod
80     def get_instance_info(self) -> InstanceInfo:
81         """Return the instance info."""
82 
83     @abc.abstractmethod
84     def run(self, command: Sequence[str], hide_output: bool = False) -> Optional[bytes]:
85         """Run a command on the instance."""
86 
87     @abc.abstractmethod
88     def _launch(self, specs: Dict):
89         """Launch the instance."""
90 
91     @abc.abstractmethod
92     def _start(self):
93         """Start an existing the instance."""
94 
95     @abc.abstractmethod
96     def _push_file(self, *, source: str, destination: str) -> None:
97         """Push a file into the instance."""
98 
99     @abc.abstractmethod
100     def pull_file(self, name: str, destination: str, delete: bool = False) -> None:
101         """
102         Provider steps needed to retrieve a file from the instance, optionally
103         deleting the source file after a successful retrieval.
104 
105         :param name: the remote filename.
106         :type name: str
107         :param destination: the local filename.
108         :type destination: str
109         :param delete: whether the file should be deleted.
110         :type delete: bool
111         """
112 
113     @abc.abstractmethod
114     def shell(self) -> None:
115         """Provider steps to provide a shell into the instance."""
116 
117     def launch_instance(self, specs: Dict) -> None:
118         try:
119             # An ProviderStartError exception here means we need to create.
120             self._start()
121         except errors.ProviderInstanceNotFoundError:
122             self._launch(specs)
123             self._check_connectivity()
124             # We need to setup MicroK8s and scan for cli commands.
125             self._setup_microk8s(specs)
126             self._copy_kubeconfig_to_kubectl(specs)
127 
128     def _check_connectivity(self) -> None:
129         """Check that the VM can access the internet."""
130         try:
131             self.run("ping -c 1 snapcraft.io".split(), hide_output=True)
132         except errors.ProviderLaunchError:
133             self.destroy()
134             url = None
135             if sys.platform == "win32":
136                 url = "https://multipass.run/docs/troubleshooting-networking-on-windows"
137             elif sys.platform == "darwin":
138                 url = "https://multipass.run/docs/troubleshooting-networking-on-macos"
139 
140             if url:
141                 raise errors.ConnectivityError(
142                     "The VM cannot connect to snapcraft.io, please see {}".format(url)
143                 )
144             else:
145                 raise
146 
147     def _copy_kubeconfig_to_kubectl(self, specs: Dict):
148         kubeconfig_path = specs.get("kubeconfig")
149         kubeconfig = self.run(command=["microk8s", "config"], hide_output=True)
150 
151         if not os.path.isdir(os.path.dirname(kubeconfig_path)):
152             os.mkdir(os.path.dirname(kubeconfig_path))
153 
154         with open(kubeconfig_path, "wb") as f:
155             f.write(kubeconfig)
156 
157     def _setup_microk8s(self, specs: Dict) -> None:
158         self.run("snap install microk8s --classic --channel {}".format(specs["channel"]).split())
159         if sys.platform == "win32":
160             self.run("snap install microk8s-integrator-windows".split())
161         elif sys.platform == "darwin":
162             self.run("snap install microk8s-integrator-macos".split())
163 
164     def _get_env_command(self) -> Sequence[str]:
165         """Get command sequence for `env` with configured flags."""
166 
167         env_list = ["env"]
168 
169         # Pass through configurable environment variables.
170         for key in ["http_proxy", "https_proxy"]:
171             value = self.build_provider_flags.get(key)
172             if not value:
173                 continue
174 
175             # Ensure item is treated as string and append it.
176             value = str(value)
177             env_list.append(f"{key}={value}")
178 
179         return env_list
180 
181     def _get_home_directory(self) -> pathlib.Path:
182         """Get user's home directory path."""
183         if self._cached_home_directory is not None:
184             return self._cached_home_directory
185 
186         command = ["printenv", "HOME"]
187         run_output = self.run(command=command, hide_output=True)
188 
189         # Shouldn't happen, but due to _run()'s return type as being Optional,
190         # we need to check for it anyways for mypy.
191         if not run_output:
192             provider_name = self._get_provider_name()
193             raise errors.ProviderExecError(
194                 provider_name=provider_name, command=command, exit_code=2
195             )
196 
197         cached_home_directory = pathlib.Path(run_output.decode().strip())
198 
199         self._cached_home_directory = cached_home_directory
200         return cached_home_directory
201 
202     def _base_has_changed(self, base: str, provider_base: str) -> bool:
203         # Make it backwards compatible with instances without project info
204         if base == "core18" and provider_base is None:
205             return False
206         elif base != provider_base:
207             return True
208 
209         return False
210 
211     def _log_run(self, command: Sequence[str]) -> None:
212         cmd_string = " ".join([shlex.quote(c) for c in command])
213         logger.debug(f"Running: {cmd_string}")
214 
215     @abc.abstractmethod
216     def stop(self) -> None:
217         pass
218
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/installer/vm_providers/_base_provider.py b/installer/vm_providers/_base_provider.py
--- a/installer/vm_providers/_base_provider.py
+++ b/installer/vm_providers/_base_provider.py
@@ -18,6 +18,7 @@
 import logging
 import os
 import pathlib
+import requests
 import shlex
 import sys
 from typing import Dict
@@ -128,8 +129,8 @@
     def _check_connectivity(self) -> None:
         """Check that the VM can access the internet."""
         try:
-            self.run("ping -c 1 snapcraft.io".split(), hide_output=True)
-        except errors.ProviderLaunchError:
+            requests.get("https://snapcraft.io")
+        except requests.exceptions.RequestException:
             self.destroy()
             url = None
             if sys.platform == "win32":

{"golden_diff": "diff --git a/installer/vm_providers/_base_provider.py b/installer/vm_providers/_base_provider.py\n--- a/installer/vm_providers/_base_provider.py\n+++ b/installer/vm_providers/_base_provider.py\n@@ -18,6 +18,7 @@\n import logging\n import os\n import pathlib\n+import requests\n import shlex\n import sys\n from typing import Dict\n@@ -128,8 +129,8 @@\n     def _check_connectivity(self) -> None:\n         \"\"\"Check that the VM can access the internet.\"\"\"\n         try:\n-            self.run(\"ping -c 1 snapcraft.io\".split(), hide_output=True)\n-        except errors.ProviderLaunchError:\n+            requests.get(\"https://snapcraft.io\")\n+        except requests.exceptions.RequestException:\n             self.destroy()\n             url = None\n             if sys.platform == \"win32\":\n", "issue": "Internet connectivity check fails if ICMP traffic is restricted\n<!--\r\n   Thank you for submitting an issue. Please fill in the template below\r\n   information about the bug you encountered.\r\n-->\r\n\r\n#### Summary\r\n<!-- Please explain the bug in a few short sentences -->\r\nWhen trying to install on macOS I get a failure to connect to the internet, the check is done via a ping (ICMP) traffic which isn't allowed on my corp network. However http/https (which are the connections actually required) are fine. I can't see any other use of ping other than a test.\r\n\r\n#### What Should Happen Instead?\r\n<!-- Please explain what the expected behavior is -->\r\nThe internet check should check for http(s) connectivity instead of using ping.\r\n\r\n#### Reproduction Steps\r\n<!-- Are you able to consistently reproduce the issue? Please add a list of steps that lead to the bug. -->\r\n\r\n1. be on network that disallows ICMP traffic\r\n2. microk8s install\r\n\r\n#### Introspection Report\r\n<!-- Please run `microk8s inspect` and attach the generated tarball. -->\r\nN/A\r\n\r\n#### Can you suggest a fix?\r\n<!-- (This section is optional). How do you propose that the issue be fixed? -->\r\nhttps://github.com/canonical/microk8s/blob/master/installer/vm_providers/_base_provider.py#L131\r\n\r\n`curl --fail --head https://snapcraft.io` instead perhaps?\r\n\r\n#### Are you interested in contributing with a fix?\r\n<!-- yes/no, or @mention maintainers. Community contributions are welcome. -->\r\nYes\r\n\r\n<!-- Thank you for making MicroK8s better -->\r\n\n", "before_files": [{"content": "# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-\n#\n# Copyright (C) 2018-2019 Canonical Ltd\n#\n# This program is free software: you can redistribute it and/or modify\n# it under the terms of the GNU General Public License version 3 as\n# published by the Free Software Foundation.\n#\n# This program is distributed in the hope that it will be useful,\n# but WITHOUT ANY WARRANTY; without even the implied warranty of\n# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\n# GNU General Public License for more details.\n#\n# You should have received a copy of the GNU General Public License\n# along with this program.  If not, see <http://www.gnu.org/licenses/>.\n\nimport abc\nimport logging\nimport os\nimport pathlib\nimport shlex\nimport sys\nfrom typing import Dict\nfrom typing import Optional, Sequence\n\nfrom . import errors\nfrom ._multipass._instance_info import InstanceInfo\n\nlogger = logging.getLogger(__name__)\n\n\nclass Provider(abc.ABC):\n    def __init__(\n        self,\n        *,\n        echoer,\n        is_ephemeral: bool = False,\n        build_provider_flags: Dict[str, str] = None,\n    ) -> None:\n        self.echoer = echoer\n        self._is_ephemeral = is_ephemeral\n\n        self.instance_name = \"microk8s-vm\"\n\n        if build_provider_flags is None:\n            build_provider_flags = dict()\n        self.build_provider_flags = build_provider_flags.copy()\n\n        self._cached_home_directory: Optional[pathlib.Path] = None\n\n    @classmethod\n    def ensure_provider(cls) -> None:\n        \"\"\"Necessary steps to ensure the provider is correctly setup.\"\"\"\n\n    @classmethod\n    def setup_provider(cls, *, echoer) -> None:\n        \"\"\"Necessary steps to install the provider on the host.\"\"\"\n\n    @classmethod\n    def _get_provider_name(cls) -> str:\n        \"\"\"Return the provider name.\"\"\"\n\n    @classmethod\n    def _get_is_snap_injection_capable(cls) -> bool:\n        \"\"\"Return whether the provider can install snaps from the host.\"\"\"\n\n    @abc.abstractmethod\n    def create(self) -> None:\n        \"\"\"Provider steps needed to create a fully functioning environment.\"\"\"\n\n    @abc.abstractmethod\n    def destroy(self) -> None:\n        \"\"\"Provider steps needed to ensure the instance is destroyed.\n\n        This method should be safe to call multiple times and do nothing\n        if the instance to destroy is already destroyed.\n        \"\"\"\n\n    @abc.abstractmethod\n    def get_instance_info(self) -> InstanceInfo:\n        \"\"\"Return the instance info.\"\"\"\n\n    @abc.abstractmethod\n    def run(self, command: Sequence[str], hide_output: bool = False) -> Optional[bytes]:\n        \"\"\"Run a command on the instance.\"\"\"\n\n    @abc.abstractmethod\n    def _launch(self, specs: Dict):\n        \"\"\"Launch the instance.\"\"\"\n\n    @abc.abstractmethod\n    def _start(self):\n        \"\"\"Start an existing the instance.\"\"\"\n\n    @abc.abstractmethod\n    def _push_file(self, *, source: str, destination: str) -> None:\n        \"\"\"Push a file into the instance.\"\"\"\n\n    @abc.abstractmethod\n    def pull_file(self, name: str, destination: str, delete: bool = False) -> None:\n        \"\"\"\n        Provider steps needed to retrieve a file from the instance, optionally\n        deleting the source file after a successful retrieval.\n\n        :param name: the remote filename.\n        :type name: str\n        :param destination: the local filename.\n        :type destination: str\n        :param delete: whether the file should be deleted.\n        :type delete: bool\n        \"\"\"\n\n    @abc.abstractmethod\n    def shell(self) -> None:\n        \"\"\"Provider steps to provide a shell into the instance.\"\"\"\n\n    def launch_instance(self, specs: Dict) -> None:\n        try:\n            # An ProviderStartError exception here means we need to create.\n            self._start()\n        except errors.ProviderInstanceNotFoundError:\n            self._launch(specs)\n            self._check_connectivity()\n            # We need to setup MicroK8s and scan for cli commands.\n            self._setup_microk8s(specs)\n            self._copy_kubeconfig_to_kubectl(specs)\n\n    def _check_connectivity(self) -> None:\n        \"\"\"Check that the VM can access the internet.\"\"\"\n        try:\n            self.run(\"ping -c 1 snapcraft.io\".split(), hide_output=True)\n        except errors.ProviderLaunchError:\n            self.destroy()\n            url = None\n            if sys.platform == \"win32\":\n                url = \"https://multipass.run/docs/troubleshooting-networking-on-windows\"\n            elif sys.platform == \"darwin\":\n                url = \"https://multipass.run/docs/troubleshooting-networking-on-macos\"\n\n            if url:\n                raise errors.ConnectivityError(\n                    \"The VM cannot connect to snapcraft.io, please see {}\".format(url)\n                )\n            else:\n                raise\n\n    def _copy_kubeconfig_to_kubectl(self, specs: Dict):\n        kubeconfig_path = specs.get(\"kubeconfig\")\n        kubeconfig = self.run(command=[\"microk8s\", \"config\"], hide_output=True)\n\n        if not os.path.isdir(os.path.dirname(kubeconfig_path)):\n            os.mkdir(os.path.dirname(kubeconfig_path))\n\n        with open(kubeconfig_path, \"wb\") as f:\n            f.write(kubeconfig)\n\n    def _setup_microk8s(self, specs: Dict) -> None:\n        self.run(\"snap install microk8s --classic --channel {}\".format(specs[\"channel\"]).split())\n        if sys.platform == \"win32\":\n            self.run(\"snap install microk8s-integrator-windows\".split())\n        elif sys.platform == \"darwin\":\n            self.run(\"snap install microk8s-integrator-macos\".split())\n\n    def _get_env_command(self) -> Sequence[str]:\n        \"\"\"Get command sequence for `env` with configured flags.\"\"\"\n\n        env_list = [\"env\"]\n\n        # Pass through configurable environment variables.\n        for key in [\"http_proxy\", \"https_proxy\"]:\n            value = self.build_provider_flags.get(key)\n            if not value:\n                continue\n\n            # Ensure item is treated as string and append it.\n            value = str(value)\n            env_list.append(f\"{key}={value}\")\n\n        return env_list\n\n    def _get_home_directory(self) -> pathlib.Path:\n        \"\"\"Get user's home directory path.\"\"\"\n        if self._cached_home_directory is not None:\n            return self._cached_home_directory\n\n        command = [\"printenv\", \"HOME\"]\n        run_output = self.run(command=command, hide_output=True)\n\n        # Shouldn't happen, but due to _run()'s return type as being Optional,\n        # we need to check for it anyways for mypy.\n        if not run_output:\n            provider_name = self._get_provider_name()\n            raise errors.ProviderExecError(\n                provider_name=provider_name, command=command, exit_code=2\n            )\n\n        cached_home_directory = pathlib.Path(run_output.decode().strip())\n\n        self._cached_home_directory = cached_home_directory\n        return cached_home_directory\n\n    def _base_has_changed(self, base: str, provider_base: str) -> bool:\n        # Make it backwards compatible with instances without project info\n        if base == \"core18\" and provider_base is None:\n            return False\n        elif base != provider_base:\n            return True\n\n        return False\n\n    def _log_run(self, command: Sequence[str]) -> None:\n        cmd_string = \" \".join([shlex.quote(c) for c in command])\n        logger.debug(f\"Running: {cmd_string}\")\n\n    @abc.abstractmethod\n    def stop(self) -> None:\n        pass\n", "path": "installer/vm_providers/_base_provider.py"}], "after_files": [{"content": "# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-\n#\n# Copyright (C) 2018-2019 Canonical Ltd\n#\n# This program is free software: you can redistribute it and/or modify\n# it under the terms of the GNU General Public License version 3 as\n# published by the Free Software Foundation.\n#\n# This program is distributed in the hope that it will be useful,\n# but WITHOUT ANY WARRANTY; without even the implied warranty of\n# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\n# GNU General Public License for more details.\n#\n# You should have received a copy of the GNU General Public License\n# along with this program.  If not, see <http://www.gnu.org/licenses/>.\n\nimport abc\nimport logging\nimport os\nimport pathlib\nimport requests\nimport shlex\nimport sys\nfrom typing import Dict\nfrom typing import Optional, Sequence\n\nfrom . import errors\nfrom ._multipass._instance_info import InstanceInfo\n\nlogger = logging.getLogger(__name__)\n\n\nclass Provider(abc.ABC):\n    def __init__(\n        self,\n        *,\n        echoer,\n        is_ephemeral: bool = False,\n        build_provider_flags: Dict[str, str] = None,\n    ) -> None:\n        self.echoer = echoer\n        self._is_ephemeral = is_ephemeral\n\n        self.instance_name = \"microk8s-vm\"\n\n        if build_provider_flags is None:\n            build_provider_flags = dict()\n        self.build_provider_flags = build_provider_flags.copy()\n\n        self._cached_home_directory: Optional[pathlib.Path] = None\n\n    @classmethod\n    def ensure_provider(cls) -> None:\n        \"\"\"Necessary steps to ensure the provider is correctly setup.\"\"\"\n\n    @classmethod\n    def setup_provider(cls, *, echoer) -> None:\n        \"\"\"Necessary steps to install the provider on the host.\"\"\"\n\n    @classmethod\n    def _get_provider_name(cls) -> str:\n        \"\"\"Return the provider name.\"\"\"\n\n    @classmethod\n    def _get_is_snap_injection_capable(cls) -> bool:\n        \"\"\"Return whether the provider can install snaps from the host.\"\"\"\n\n    @abc.abstractmethod\n    def create(self) -> None:\n        \"\"\"Provider steps needed to create a fully functioning environment.\"\"\"\n\n    @abc.abstractmethod\n    def destroy(self) -> None:\n        \"\"\"Provider steps needed to ensure the instance is destroyed.\n\n        This method should be safe to call multiple times and do nothing\n        if the instance to destroy is already destroyed.\n        \"\"\"\n\n    @abc.abstractmethod\n    def get_instance_info(self) -> InstanceInfo:\n        \"\"\"Return the instance info.\"\"\"\n\n    @abc.abstractmethod\n    def run(self, command: Sequence[str], hide_output: bool = False) -> Optional[bytes]:\n        \"\"\"Run a command on the instance.\"\"\"\n\n    @abc.abstractmethod\n    def _launch(self, specs: Dict):\n        \"\"\"Launch the instance.\"\"\"\n\n    @abc.abstractmethod\n    def _start(self):\n        \"\"\"Start an existing the instance.\"\"\"\n\n    @abc.abstractmethod\n    def _push_file(self, *, source: str, destination: str) -> None:\n        \"\"\"Push a file into the instance.\"\"\"\n\n    @abc.abstractmethod\n    def pull_file(self, name: str, destination: str, delete: bool = False) -> None:\n        \"\"\"\n        Provider steps needed to retrieve a file from the instance, optionally\n        deleting the source file after a successful retrieval.\n\n        :param name: the remote filename.\n        :type name: str\n        :param destination: the local filename.\n        :type destination: str\n        :param delete: whether the file should be deleted.\n        :type delete: bool\n        \"\"\"\n\n    @abc.abstractmethod\n    def shell(self) -> None:\n        \"\"\"Provider steps to provide a shell into the instance.\"\"\"\n\n    def launch_instance(self, specs: Dict) -> None:\n        try:\n            # An ProviderStartError exception here means we need to create.\n            self._start()\n        except errors.ProviderInstanceNotFoundError:\n            self._launch(specs)\n            self._check_connectivity()\n            # We need to setup MicroK8s and scan for cli commands.\n            self._setup_microk8s(specs)\n            self._copy_kubeconfig_to_kubectl(specs)\n\n    def _check_connectivity(self) -> None:\n        \"\"\"Check that the VM can access the internet.\"\"\"\n        try:\n            requests.get(\"https://snapcraft.io\")\n        except requests.exceptions.RequestException:\n            self.destroy()\n            url = None\n            if sys.platform == \"win32\":\n                url = \"https://multipass.run/docs/troubleshooting-networking-on-windows\"\n            elif sys.platform == \"darwin\":\n                url = \"https://multipass.run/docs/troubleshooting-networking-on-macos\"\n\n            if url:\n                raise errors.ConnectivityError(\n                    \"The VM cannot connect to snapcraft.io, please see {}\".format(url)\n                )\n            else:\n                raise\n\n    def _copy_kubeconfig_to_kubectl(self, specs: Dict):\n        kubeconfig_path = specs.get(\"kubeconfig\")\n        kubeconfig = self.run(command=[\"microk8s\", \"config\"], hide_output=True)\n\n        if not os.path.isdir(os.path.dirname(kubeconfig_path)):\n            os.mkdir(os.path.dirname(kubeconfig_path))\n\n        with open(kubeconfig_path, \"wb\") as f:\n            f.write(kubeconfig)\n\n    def _setup_microk8s(self, specs: Dict) -> None:\n        self.run(\"snap install microk8s --classic --channel {}\".format(specs[\"channel\"]).split())\n        if sys.platform == \"win32\":\n            self.run(\"snap install microk8s-integrator-windows\".split())\n        elif sys.platform == \"darwin\":\n            self.run(\"snap install microk8s-integrator-macos\".split())\n\n    def _get_env_command(self) -> Sequence[str]:\n        \"\"\"Get command sequence for `env` with configured flags.\"\"\"\n\n        env_list = [\"env\"]\n\n        # Pass through configurable environment variables.\n        for key in [\"http_proxy\", \"https_proxy\"]:\n            value = self.build_provider_flags.get(key)\n            if not value:\n                continue\n\n            # Ensure item is treated as string and append it.\n            value = str(value)\n            env_list.append(f\"{key}={value}\")\n\n        return env_list\n\n    def _get_home_directory(self) -> pathlib.Path:\n        \"\"\"Get user's home directory path.\"\"\"\n        if self._cached_home_directory is not None:\n            return self._cached_home_directory\n\n        command = [\"printenv\", \"HOME\"]\n        run_output = self.run(command=command, hide_output=True)\n\n        # Shouldn't happen, but due to _run()'s return type as being Optional,\n        # we need to check for it anyways for mypy.\n        if not run_output:\n            provider_name = self._get_provider_name()\n            raise errors.ProviderExecError(\n                provider_name=provider_name, command=command, exit_code=2\n            )\n\n        cached_home_directory = pathlib.Path(run_output.decode().strip())\n\n        self._cached_home_directory = cached_home_directory\n        return cached_home_directory\n\n    def _base_has_changed(self, base: str, provider_base: str) -> bool:\n        # Make it backwards compatible with instances without project info\n        if base == \"core18\" and provider_base is None:\n            return False\n        elif base != provider_base:\n            return True\n\n        return False\n\n    def _log_run(self, command: Sequence[str]) -> None:\n        cmd_string = \" \".join([shlex.quote(c) for c in command])\n        logger.debug(f\"Running: {cmd_string}\")\n\n    @abc.abstractmethod\n    def stop(self) -> None:\n        pass\n", "path": "installer/vm_providers/_base_provider.py"}]}