feedc0de/platformio-core
1
0
Fork 0
forked from platformio/platformio-core
Code Pull Requests Activity

Port package management "print_message" to the Python logging system

Browse Source
This commit is contained in:
Ivan Kravets
2022-02-18 12:57:30 +02:00
parent e8051838a3
commit f34e6e9c4c
15 changed files with 202 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
platformio/commands/lib/command.py
View File

@ -15,6 +15,7 @@
# pylint: disable=too-many-branches, too-many-locals
import json
import logging
import os
import time
@ -152,16 +153,16 @@ def lib_install( # pylint: disable=too-many-arguments,unused-argument
if not silent and (libraries or storage_dir in storage_libdeps):
print_storage_header(storage_dirs, storage_dir)
lm = LibraryPackageManager(storage_dir)
lm.set_log_level(logging.WARN if silent else logging.DEBUG)
if libraries:
installed_pkgs = {
library: lm.install(library, silent=silent, force=force)
for library in libraries
library: lm.install(library, force=force) for library in libraries
}
elif storage_dir in storage_libdeps:
for library in storage_libdeps[storage_dir]:
lm.install(library, silent=silent, force=force)
lm.install(library, force=force)
if save and installed_pkgs:
_save_deps(ctx, installed_pkgs)
@ -212,9 +213,8 @@ def lib_uninstall(ctx, libraries, save, silent):
for storage_dir in storage_dirs:
print_storage_header(storage_dirs, storage_dir)
lm = LibraryPackageManager(storage_dir)
uninstalled_pkgs = {
library: lm.uninstall(library, silent=silent) for library in libraries
}
lm.set_log_level(logging.WARN if silent else logging.DEBUG)
uninstalled_pkgs = {library: lm.uninstall(library) for library in libraries}
if save and uninstalled_pkgs:
_save_deps(ctx, uninstalled_pkgs, action="remove")
@ -245,6 +245,7 @@ def lib_update( # pylint: disable=too-many-arguments
print_storage_header(storage_dirs, storage_dir)
lib_deps = ctx.meta.get(CTX_META_STORAGE_LIBDEPS_KEY, {}).get(storage_dir, [])
lm = LibraryPackageManager(storage_dir)
lm.set_log_level(logging.WARN if silent else logging.DEBUG)
_libraries = libraries or lib_deps or lm.get_installed()
if only_check and json_output:
@ -277,9 +278,7 @@ def lib_update( # pylint: disable=too-many-arguments
None if isinstance(library, PackageItem) else PackageSpec(library)
)
try:
lm.update(
library, to_spec=to_spec, only_check=only_check, silent=silent
)
lm.update(library, to_spec=to_spec, only_check=only_check)
except UnknownPackageError as e:
if library not in lib_deps:
raise e
@ -438,7 +437,8 @@ def lib_builtin(storage, json_output):
@click.option("--json-output", is_flag=True)
def lib_show(library, json_output):
lm = LibraryPackageManager()
lib_id = lm.reveal_registry_package_id(library, silent=json_output)
lm.set_log_level(logging.ERROR if json_output else logging.DEBUG)
lib_id = lm.reveal_registry_package_id(library)
regclient = lm.get_registry_client_instance()
lib = regclient.fetch_json_data(
"get", "/v2/lib/info/%d" % lib_id, x_cache_valid="1h"

9
platformio/commands/platform.py
View File

@ -13,6 +13,7 @@
# limitations under the License.
import json
import logging
import os
import click
@ -208,6 +209,7 @@ def _platform_install( # pylint: disable=too-many-arguments
force=False,
):
pm = PlatformPackageManager()
pm.set_log_level(logging.WARN if silent else logging.DEBUG)
for platform in platforms:
pkg = pm.install(
spec=platform,
@ -215,7 +217,6 @@ def _platform_install( # pylint: disable=too-many-arguments
without_packages=without_package or [],
skip_default_package=skip_default_package,
with_all_packages=with_all_packages,
silent=silent,
force=force,
)
if pkg and not silent:
@ -231,6 +232,7 @@ def _platform_install( # pylint: disable=too-many-arguments
@click.argument("platforms", nargs=-1, required=True, metavar="[PLATFORM...]")
def platform_uninstall(platforms):
pm = PlatformPackageManager()
pm.set_log_level(logging.DEBUG)
for platform in platforms:
if pm.uninstall(platform):
click.secho(
@ -259,6 +261,7 @@ def platform_update( # pylint: disable=too-many-locals, too-many-arguments
platforms, only_packages, only_check, dry_run, silent, json_output
):
pm = PlatformPackageManager()
pm.set_log_level(logging.WARN if silent else logging.DEBUG)
platforms = platforms or pm.get_installed()
only_check = dry_run or only_check
@ -304,9 +307,7 @@ def platform_update( # pylint: disable=too-many-locals, too-many-arguments
)
)
click.echo("--------")
pm.update(
platform, only_packages=only_packages, only_check=only_check, silent=silent
)
pm.update(platform, only_packages=only_packages, only_check=only_check)
click.echo()
return True

15
platformio/package/manager/_download.py
View File

@ -13,10 +13,13 @@
# limitations under the License.
import hashlib
import logging
import os
import tempfile
import time
import click
from platformio import app, compat
from platformio.package.download import FileDownloader
from platformio.package.lockfile import LockFile
@ -51,7 +54,8 @@ class PackageManagerDownloadMixin(object):
if os.path.isfile(dl_path):
os.remove(dl_path)
def download(self, url, checksum=None, silent=False):
def download(self, url, checksum=None):
silent = not self.log.isEnabledFor(logging.INFO)
dl_path = self.compute_download_path(url, checksum or "")
if os.path.isfile(dl_path):
self.set_download_utime(dl_path)
@ -75,10 +79,11 @@ class PackageManagerDownloadMixin(object):
except IOError:
raise_error = True
if raise_error:
self.print_message(
"Error: Please read https://bit.ly/package-manager-ioerror",
fg="red",
err=True,
self.log.error(
click.style(
"Error: Please read https://bit.ly/package-manager-ioerror",
fg="red",
)
)
raise e
if checksum:

38
platformio/package/manager/_install.py
View File

@ -42,23 +42,20 @@ class PackageManagerInstallMixin(object):
with FileUnpacker(src) as fu:
return fu.unpack(dst, with_progress=False)
def install(self, spec, silent=False, skip_dependencies=False, force=False):
def install(self, spec, skip_dependencies=False, force=False):
try:
self.lock()
pkg = self._install(
spec, silent=silent, skip_dependencies=skip_dependencies, force=force
)
pkg = self._install(spec, skip_dependencies=skip_dependencies, force=force)
self.memcache_reset()
self.cleanup_expired_downloads()
return pkg
finally:
self.unlock()
def _install( # pylint: disable=too-many-arguments
def _install(
self,
spec,
search_filters=None,
silent=False,
skip_dependencies=False,
force=False,
):
@ -75,28 +72,26 @@ class PackageManagerInstallMixin(object):
# if a forced installation
if pkg and force:
self.uninstall(pkg, silent=silent)
self.uninstall(pkg)
pkg = None
if pkg:
if not silent:
self.print_message(
self.log.debug(
click.style(
"{name} @ {version} is already installed".format(
**pkg.metadata.as_dict()
),
fg="yellow",
)
)
return pkg
if not silent:
self.print_message(
"Installing %s" % click.style(spec.humanize(), fg="cyan")
)
self.log.info("Installing %s" % click.style(spec.humanize(), fg="cyan"))
if spec.external:
pkg = self.install_from_url(spec.url, spec, silent=silent)
pkg = self.install_from_url(spec.url, spec)
else:
pkg = self.install_from_registry(spec, search_filters, silent=silent)
pkg = self.install_from_registry(spec, search_filters)
if not pkg or not pkg.metadata:
raise PackageException(
@ -104,24 +99,25 @@ class PackageManagerInstallMixin(object):
% (spec.humanize(), util.get_systype())
)
if not silent:
self.print_message(
self.log.info(
click.style(
"{name} @ {version} has been installed!".format(
**pkg.metadata.as_dict()
),
fg="green",
)
)
self.memcache_reset()
if not skip_dependencies:
self.install_dependencies(pkg, silent)
self.install_dependencies(pkg)
self._INSTALL_HISTORY[spec] = pkg
return pkg
def install_dependencies(self, pkg, silent=False):
def install_dependencies(self, pkg):
pass
def install_from_url(self, url, spec, checksum=None, silent=False):
def install_from_url(self, url, spec, checksum=None):
spec = self.ensure_spec(spec)
tmp_dir = tempfile.mkdtemp(prefix="pkg-installing-", dir=self.get_tmp_dir())
vcs = None
@ -134,7 +130,7 @@ class PackageManagerInstallMixin(object):
fs.rmtree(tmp_dir)
shutil.copytree(_url, tmp_dir, symlinks=True)
elif url.startswith(("http://", "https://")):
dl_path = self.download(url, checksum, silent=silent)
dl_path = self.download(url, checksum)
assert os.path.isfile(dl_path)
self.unpack(dl_path, tmp_dir)
else:

42
platformio/package/manager/_registry.py
View File

@ -79,7 +79,7 @@ class RegistryFileMirrorIterator(object):
class PackageManageRegistryMixin(object):
def install_from_registry(self, spec, search_filters=None, silent=False):
def install_from_registry(self, spec, search_filters=None):
if spec.owner and spec.name and not search_filters:
package = self.fetch_registry_package(spec)
if not package:
@ -89,7 +89,7 @@ class PackageManageRegistryMixin(object):
packages = self.search_registry_packages(spec, search_filters)
if not packages:
raise UnknownPackageError(spec.humanize())
if len(packages) > 1 and not silent:
if len(packages) > 1:
self.print_multi_package_issue(packages, spec)
package, version = self.find_best_registry_version(packages, spec)
@ -110,11 +110,14 @@ class PackageManageRegistryMixin(object):
name=package["name"],
),
checksum or pkgfile["checksum"]["sha256"],
silent=silent,
)
except Exception as e: # pylint: disable=broad-except
self.print_message("Warning! Package Mirror: %s" % e, fg="yellow")
self.print_message("Looking for another mirror...", fg="yellow")
self.log.warning(
click.style("Warning! Package Mirror: %s" % e, fg="yellow")
)
self.log.warning(
click.style("Looking for another mirror...", fg="yellow")
)
return None
@ -153,36 +156,41 @@ class PackageManageRegistryMixin(object):
raise UnknownPackageError(spec.humanize())
return result
def reveal_registry_package_id(self, spec, silent=False):
def reveal_registry_package_id(self, spec):
spec = self.ensure_spec(spec)
if spec.id:
return spec.id
packages = self.search_registry_packages(spec)
if not packages:
raise UnknownPackageError(spec.humanize())
if len(packages) > 1 and not silent:
if len(packages) > 1:
self.print_multi_package_issue(packages, spec)
click.echo("")
self.log.info("")
return packages[0]["id"]
def print_multi_package_issue(self, packages, spec):
self.print_message(
"Warning! More than one package has been found by ", fg="yellow", nl=False
self.log.warning(
click.style(
"Warning! More than one package has been found by ", fg="yellow"
)
+ click.style(spec.humanize(), fg="cyan")
+ click.style(" requirements:", fg="yellow")
)
click.secho(spec.humanize(), fg="cyan", nl=False)
click.secho(" requirements:", fg="yellow")
for item in packages:
click.echo(
self.log.warning(
" - {owner}/{name} @ {version}".format(
owner=click.style(item["owner"]["username"], fg="cyan"),
name=item["name"],
version=item["version"]["name"],
)
)
self.print_message(
"Please specify detailed REQUIREMENTS using package owner and version "
"(shown above) to avoid name conflicts",
fg="yellow",
self.log.warning(
click.style(
"Please specify detailed REQUIREMENTS using package owner and version "
"(shown above) to avoid name conflicts",
fg="yellow",
)
)
def find_best_registry_version(self, packages, spec):

24
platformio/package/manager/_uninstall.py
View File

@ -23,27 +23,26 @@ from platformio.package.meta import PackageSpec
class PackageManagerUninstallMixin(object):
def uninstall(self, spec, silent=False, skip_dependencies=False):
def uninstall(self, spec, skip_dependencies=False):
try:
self.lock()
return self._uninstall(spec, silent, skip_dependencies)
return self._uninstall(spec, skip_dependencies)
finally:
self.unlock()
def _uninstall(self, spec, silent=False, skip_dependencies=False):
def _uninstall(self, spec, skip_dependencies=False):
pkg = self.get_package(spec)
if not pkg or not pkg.metadata:
raise UnknownPackageError(spec)
if not silent:
self.print_message(
"Removing %s @ %s"
% (click.style(pkg.metadata.name, fg="cyan"), pkg.metadata.version),
)
self.log.info(
"Removing %s @ %s"
% (click.style(pkg.metadata.name, fg="cyan"), pkg.metadata.version)
)
# firstly, remove dependencies
if not skip_dependencies:
self.uninstall_dependencies(pkg, silent)
self.uninstall_dependencies(pkg)
if os.path.islink(pkg.path):
os.unlink(pkg.path)
@ -66,13 +65,14 @@ class PackageManagerUninstallMixin(object):
)
self.memcache_reset()
if not silent:
self.print_message(
self.log.info(
click.style(
"{name} @ {version} has been removed!".format(**pkg.metadata.as_dict()),
fg="green",
)
)
return pkg
def uninstall_dependencies(self, pkg, silent=False):
def uninstall_dependencies(self, pkg):
pass

14
platformio/package/manager/_update.py
View File

@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import os
import click
@ -77,18 +78,18 @@ class PackageManagerUpdateMixin(object):
).version
)
def update( # pylint: disable=too-many-arguments
def update(
self,
from_spec,
to_spec=None,
only_check=False,
silent=False,
show_incompatible=True,
):
pkg = self.get_package(from_spec)
if not pkg or not pkg.metadata:
raise UnknownPackageError(from_spec)
silent = not self.log.isEnabledFor(logging.INFO)
if not silent:
click.echo(
"{} {:<45} {:<35}".format(
@ -114,7 +115,7 @@ class PackageManagerUpdateMixin(object):
try:
self.lock()
return self._update(pkg, outdated, silent=silent)
return self._update(pkg, outdated)
finally:
self.unlock()
@ -156,7 +157,7 @@ class PackageManagerUpdateMixin(object):
)
)
def _update(self, pkg, outdated, silent=False):
def _update(self, pkg, outdated):
if pkg.metadata.spec.external:
vcs = VCSClientFactory.new(pkg.path, pkg.metadata.spec.url)
assert vcs.update()
@ -170,8 +171,7 @@ class PackageManagerUpdateMixin(object):
owner=pkg.metadata.spec.owner,
name=pkg.metadata.spec.name,
requirements=outdated.wanted or outdated.latest,
),
silent=silent,
)
)
if new_pkg:
old_pkg = self.get_package(
@ -183,5 +183,5 @@ class PackageManagerUpdateMixin(object):
)
)
if old_pkg:
self.uninstall(old_pkg, silent=silent, skip_dependencies=True)
self.uninstall(old_pkg, skip_dependencies=True)
return new_pkg

28
platformio/package/manager/base.py
View File

@ -12,7 +12,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import os
import sys
from datetime import datetime
import click
@ -39,7 +41,7 @@ from platformio.package.meta import (
from platformio.project.helpers import get_project_cache_dir
class BasePackageManager( # pylint: disable=too-many-public-methods
class BasePackageManager( # pylint: disable=too-many-public-methods,too-many-instance-attributes
PackageManagerDownloadMixin,
PackageManageRegistryMixin,
PackageManagerInstallMixin,
@ -52,8 +54,9 @@ class BasePackageManager( # pylint: disable=too-many-public-methods
def __init__(self, pkg_type, package_dir):
self.pkg_type = pkg_type
self.package_dir = package_dir
self._MEMORY_CACHE = {}
self.log = self._setup_logger()
self._MEMORY_CACHE = {}
self._lockfile = None
self._download_dir = None
self._tmp_dir = None
@ -65,6 +68,19 @@ class BasePackageManager( # pylint: disable=too-many-public-methods
f"package_dir={self.package_dir}>"
)
def _setup_logger(self):
logger = logging.getLogger(str(self.__class__.__name__).replace("Package", " "))
logger.setLevel(logging.ERROR if PlatformioCLI.in_silence() else logging.INFO)
formatter = logging.Formatter("%(name)s: %(message)s")
sh = logging.StreamHandler(sys.stdout)
sh.setFormatter(formatter)
logger.handlers.clear()
logger.addHandler(sh)
return logger
def set_log_level(self, level):
self.log.setLevel(level)
def lock(self):
if self._lockfile:
return
@ -111,12 +127,6 @@ class BasePackageManager( # pylint: disable=too-many-public-methods
def manifest_names(self):
raise NotImplementedError
def print_message(self, message, **kwargs):
click.echo(
"%s: " % str(self.__class__.__name__).replace("Package", " "), nl=False
)
click.secho(message, **kwargs)
def get_download_dir(self):
if not self._download_dir:
self._download_dir = self.ensure_dir_exists(
@ -171,7 +181,7 @@ class BasePackageManager( # pylint: disable=too-many-public-methods
return result
except ManifestException as e:
if not PlatformioCLI.in_silence():
self.print_message(str(e), fg="yellow")
self.log.warning(click.style(str(e), fg="yellow"))
raise MissingPackageManifestError(", ".join(self.manifest_names))
@staticmethod

36
platformio/package/manager/library.py
View File

@ -15,6 +15,8 @@
import json
import os
import click
from platformio.package.exception import (
MissingPackageManifestError,
UnknownPackageError,
@ -85,7 +87,6 @@ class LibraryPackageManager(BasePackageManager): # pylint: disable=too-many-anc
self,
spec,
search_filters=None,
silent=False,
skip_dependencies=False,
force=False,
):
@ -93,7 +94,6 @@ class LibraryPackageManager(BasePackageManager): # pylint: disable=too-many-anc
return super(LibraryPackageManager, self)._install(
spec,
search_filters=search_filters,
silent=silent,
skip_dependencies=skip_dependencies,
force=force,
)
@ -103,27 +103,28 @@ class LibraryPackageManager(BasePackageManager): # pylint: disable=too-many-anc
spec = self.ensure_spec(spec)
if is_builtin_lib(spec.name):
self.print_message("Already installed, built-in library", fg="yellow")
self.log.info("Already installed, built-in library", fg="yellow")
return True
raise e
def install_dependencies(self, pkg, silent=False):
def install_dependencies(self, pkg):
assert isinstance(pkg, PackageItem)
manifest = self.load_manifest(pkg)
if not manifest.get("dependencies"):
return
if not silent:
self.print_message("Installing dependencies...")
self.log.info("Installing dependencies...")
for dependency in manifest.get("dependencies"):
if not self._install_dependency(dependency, silent) and not silent:
self.print_message(
"Warning! Could not install dependency %s for package '%s'"
% (dependency, pkg.metadata.name),
fg="yellow",
if not self._install_dependency(dependency):
self.log.warning(
click.style(
"Warning! Could not install dependency %s for package '%s'"
% (dependency, pkg.metadata.name),
fg="yellow",
)
)
def _install_dependency(self, dependency, silent=False):
def _install_dependency(self, dependency):
spec = PackageSpec(
owner=dependency.get("owner"),
name=dependency.get("name"),
@ -135,20 +136,17 @@ class LibraryPackageManager(BasePackageManager): # pylint: disable=too-many-anc
if key in ("authors", "platforms", "frameworks")
}
try:
return self._install(
spec, search_filters=search_filters or None, silent=silent
)
return self._install(spec, search_filters=search_filters or None)
except UnknownPackageError:
pass
return None
def uninstall_dependencies(self, pkg, silent=False):
def uninstall_dependencies(self, pkg):
assert isinstance(pkg, PackageItem)
manifest = self.load_manifest(pkg)
if not manifest.get("dependencies"):
return
if not silent:
self.print_message("Removing dependencies...", fg="yellow")
self.log.info(click.style("Removing dependencies...", fg="yellow"))
for dependency in manifest.get("dependencies"):
spec = PackageSpec(
owner=dependency.get("owner"),
@ -158,4 +156,4 @@ class LibraryPackageManager(BasePackageManager): # pylint: disable=too-many-anc
pkg = self.get_package(spec)
if not pkg:
continue
self._uninstall(pkg, silent=silent)
self._uninstall(pkg)

21
platformio/package/manager/platform.py
View File

@ -45,23 +45,23 @@ class PlatformPackageManager(BasePackageManager): # pylint: disable=too-many-an
without_packages=None,
skip_default_package=False,
with_all_packages=False,
silent=False,
force=False,
project_env=None,
):
already_installed = self.get_package(spec)
pkg = super(PlatformPackageManager, self).install(
spec, silent=silent, force=force, skip_dependencies=True
spec, force=force, skip_dependencies=True
)
try:
p = PlatformFactory.new(pkg)
p.ensure_engine_compatible()
except IncompatiblePlatform as e:
super(PlatformPackageManager, self).uninstall(
pkg, silent=silent, skip_dependencies=True
)
super(PlatformPackageManager, self).uninstall(pkg, skip_dependencies=True)
raise e
# set logging level for underlying tool manager
p.pm.set_log_level(self.log.getEffectiveLevel())
if project_env:
p.configure_project_packages(project_env)
@ -72,20 +72,21 @@ class PlatformPackageManager(BasePackageManager): # pylint: disable=too-many-an
with_packages,
without_packages,
skip_default_package,
silent=silent,
force=force,
)
if not already_installed:
p.on_installed()
return pkg
def uninstall(self, spec, silent=False, skip_dependencies=False):
def uninstall(self, spec, skip_dependencies=False):
pkg = self.get_package(spec)
if not pkg or not pkg.metadata:
raise UnknownPackageError(spec)
p = PlatformFactory.new(pkg)
# set logging level for underlying tool manager
p.pm.set_log_level(self.log.getEffectiveLevel())
assert super(PlatformPackageManager, self).uninstall(
pkg, silent=silent, skip_dependencies=True
pkg, skip_dependencies=True
)
if not skip_dependencies:
p.on_uninstalled()
@ -96,7 +97,6 @@ class PlatformPackageManager(BasePackageManager): # pylint: disable=too-many-an
from_spec,
to_spec=None,
only_check=False,
silent=False,
show_incompatible=True,
only_packages=False,
):
@ -104,6 +104,8 @@ class PlatformPackageManager(BasePackageManager): # pylint: disable=too-many-an
if not pkg or not pkg.metadata:
raise UnknownPackageError(from_spec)
p = PlatformFactory.new(pkg)
# set logging level for underlying tool manager
p.pm.set_log_level(self.log.getEffectiveLevel())
pkgs_before = [item.metadata.name for item in p.get_installed_packages()]
new_pkg = None
@ -113,7 +115,6 @@ class PlatformPackageManager(BasePackageManager): # pylint: disable=too-many-an
from_spec,
to_spec,
only_check=only_check,
silent=silent,
show_incompatible=show_incompatible,
)
p = PlatformFactory.new(new_pkg)

3
platformio/platform/_packages.py
View File

@ -79,7 +79,6 @@ class PlatformPackagesMixin(object):
with_packages=None,
without_packages=None,
skip_default_package=False,
silent=False,
force=False,
):
with_packages = set(self._find_pkg_names(with_packages or []))
@ -96,7 +95,7 @@ class PlatformPackagesMixin(object):
if name in with_packages or not (
skip_default_package or options.get("optional", False)
):
self.pm.install(self.get_package_spec(name), silent=silent, force=force)
self.pm.install(self.get_package_spec(name), force=force)
return True

10
tests/commands/test_lib_complex.py
View File

@ -237,7 +237,7 @@ def test_global_lib_update_check(clirunner, validate_cliresult):
) == set(lib["name"] for lib in output)
def test_global_lib_update(clirunner, validate_cliresult):
def test_global_lib_update(clirunner, validate_cliresult, strip_ansi):
# update library using package directory
result = clirunner.invoke(
cmd_lib, ["-g", "update", "NeoPixelBus", "--dry-run", "--json-output"]
@ -248,7 +248,7 @@ def test_global_lib_update(clirunner, validate_cliresult):
assert "__pkg_dir" in oudated[0]
result = clirunner.invoke(cmd_lib, ["-g", "update", oudated[0]["__pkg_dir"]])
validate_cliresult(result)
assert "Removing NeoPixelBus @ 2.2.4" in result.output
assert "Removing NeoPixelBus @ 2.2.4" in strip_ansi(result.output)
# update rest libraries
result = clirunner.invoke(cmd_lib, ["-g", "update"])
@ -262,7 +262,9 @@ def test_global_lib_update(clirunner, validate_cliresult):
assert isinstance(result.exception, UnknownPackageError)
def test_global_lib_uninstall(clirunner, validate_cliresult, isolated_pio_core):
def test_global_lib_uninstall(
clirunner, validate_cliresult, isolated_pio_core, strip_ansi
):
# uninstall using package directory
result = clirunner.invoke(cmd_lib, ["-g", "list", "--json-output"])
validate_cliresult(result)
@ -270,7 +272,7 @@ def test_global_lib_uninstall(clirunner, validate_cliresult, isolated_pio_core):
items = sorted(items, key=lambda item: item["__pkg_dir"])
result = clirunner.invoke(cmd_lib, ["-g", "uninstall", items[0]["__pkg_dir"]])
validate_cliresult(result)
assert ("Removing %s" % items[0]["name"]) in result.output
assert ("Removing %s" % items[0]["name"]) in strip_ansi(result.output)
# uninstall the rest libraries
result = clirunner.invoke(

16
tests/commands/test_platform.py
View File

@ -66,14 +66,17 @@ def test_install_core_3_dev_platform(clirunner, validate_cliresult, isolated_pio
assert result.exit_code == 0
def test_install_known_version(clirunner, validate_cliresult, isolated_pio_core):
def test_install_known_version(
clirunner, validate_cliresult, isolated_pio_core, strip_ansi
):
result = clirunner.invoke(
cli_platform.platform_install,
["atmelavr@2.0.0", "--skip-default-package", "--with-package", "tool-avrdude"],
)
validate_cliresult(result)
assert "atmelavr @ 2.0.0" in result.output
assert "Installing tool-avrdude @" in result.output
output = strip_ansi(result.output)
assert "atmelavr @ 2.0.0" in output
assert "Installing tool-avrdude @" in output
assert len(isolated_pio_core.join("packages").listdir()) == 1
@ -117,11 +120,12 @@ def test_update_check(clirunner, validate_cliresult, isolated_pio_core):
assert len(isolated_pio_core.join("packages").listdir()) == 1
def test_update_raw(clirunner, validate_cliresult, isolated_pio_core):
def test_update_raw(clirunner, validate_cliresult, isolated_pio_core, strip_ansi):
result = clirunner.invoke(cli_platform.platform_update)
validate_cliresult(result)
assert "Removing atmelavr @ 2.0.0" in result.output
assert "Platform Manager: Installing platformio/atmelavr @" in result.output
output = strip_ansi(result.output)
assert "Removing atmelavr @ 2.0.0" in output
assert "Platform Manager: Installing platformio/atmelavr @" in output
assert len(isolated_pio_core.join("packages").listdir()) == 2

9
tests/conftest.py
View File

@ -15,6 +15,7 @@
import email
import imaplib
import os
import re
import time
import pytest
@ -61,6 +62,14 @@ def clirunner(request):
return CliRunner()
@pytest.fixture(scope="session")
def strip_ansi():
def decorator(text):
return re.sub(r"\x1B\[\d+(;\d+){0,2}m", "", text)
return decorator
@pytest.fixture(scope="module")
def isolated_pio_core(request, tmpdir_factory):
core_dir = tmpdir_factory.mktemp(".platformio")

72
tests/package/test_manager.py
View File

@ -14,6 +14,7 @@
# pylint: disable=unused-argument
import logging
import os
import time
@ -37,7 +38,8 @@ def test_download(isolated_pio_core):
url = "https://github.com/platformio/platformio-core/archive/v4.3.4.zip"
checksum = "69d59642cb91e64344f2cdc1d3b98c5cd57679b5f6db7accc7707bd4c5d9664a"
lm = LibraryPackageManager()
archive_path = lm.download(url, checksum, silent=True)
lm.set_log_level(logging.ERROR)
archive_path = lm.download(url, checksum)
assert fs.calculate_file_hashsum("sha256", archive_path) == checksum
lm.cleanup_expired_downloads()
assert os.path.isfile(archive_path)
@ -150,12 +152,13 @@ def test_install_from_url(isolated_pio_core, tmpdir_factory):
tmp_dir = tmpdir_factory.mktemp("tmp")
storage_dir = tmpdir_factory.mktemp("storage")
lm = LibraryPackageManager(str(storage_dir))
lm.set_log_level(logging.ERROR)
# install from local directory
src_dir = tmp_dir.join("local-lib-dir").mkdir()
src_dir.join("main.cpp").write("")
spec = PackageSpec("file://%s" % src_dir)
pkg = lm.install(spec, silent=True)
pkg = lm.install(spec)
assert os.path.isfile(os.path.join(pkg.path, "main.cpp"))
manifest = lm.load_manifest(pkg)
assert manifest["name"] == "local-lib-dir"
@ -171,7 +174,7 @@ def test_install_from_url(isolated_pio_core, tmpdir_factory):
)
tarball_path = PackagePacker(str(src_dir)).pack(str(tmp_dir))
spec = PackageSpec("file://%s" % tarball_path)
pkg = lm.install(spec, silent=True)
pkg = lm.install(spec)
assert os.path.isfile(os.path.join(pkg.path, "src", "main.cpp"))
assert pkg == lm.get_package(spec)
assert spec == pkg.metadata.spec
@ -198,38 +201,41 @@ version = 5.2.7
def test_install_from_registry(isolated_pio_core, tmpdir_factory):
# Libraries
lm = LibraryPackageManager(str(tmpdir_factory.mktemp("lib-storage")))
lm.set_log_level(logging.ERROR)
# library with dependencies
lm.install("AsyncMqttClient-esphome @ 0.8.6", silent=True)
lm.install("AsyncMqttClient-esphome @ 0.8.6")
assert len(lm.get_installed()) == 3
pkg = lm.get_package("AsyncTCP-esphome")
assert pkg.metadata.spec.owner == "esphome"
assert not lm.get_package("non-existing-package")
# mbed library
assert lm.install("wolfSSL", silent=True)
assert lm.install("wolfSSL")
assert len(lm.get_installed()) == 4
# case sensitive author name
assert lm.install("DallasTemperature", silent=True)
assert lm.install("DallasTemperature")
assert lm.get_package("OneWire").metadata.version.major >= 2
assert len(lm.get_installed()) == 6
# test conflicted names
lm = LibraryPackageManager(str(tmpdir_factory.mktemp("conflicted-storage")))
lm.install("z3t0/IRremote@2.6.1", silent=True)
lm.install("mbed-yuhki50/IRremote", silent=True)
lm.set_log_level(logging.ERROR)
lm.install("z3t0/IRremote@2.6.1")
lm.install("mbed-yuhki50/IRremote")
assert len(lm.get_installed()) == 2
# Tools
tm = ToolPackageManager(str(tmpdir_factory.mktemp("tool-storage")))
pkg = tm.install("platformio/tool-stlink @ ~1.10400.0", silent=True)
tm.set_log_level(logging.ERROR)
pkg = tm.install("platformio/tool-stlink @ ~1.10400.0")
manifest = tm.load_manifest(pkg)
assert tm.is_system_compatible(manifest.get("system"))
assert util.get_systype() in manifest.get("system", [])
# Test unknown
with pytest.raises(UnknownPackageError):
tm.install("unknown-package-tool @ 9.1.1", silent=True)
tm.install("unknown-package-tool @ 9.1.1")
with pytest.raises(UnknownPackageError):
tm.install("owner/unknown-package-tool", silent=True)
tm.install("owner/unknown-package-tool")
def test_install_lib_depndencies(isolated_pio_core, tmpdir_factory):
@ -259,7 +265,8 @@ def test_install_lib_depndencies(isolated_pio_core, tmpdir_factory):
)
lm = LibraryPackageManager(str(tmpdir_factory.mktemp("lib-storage")))
lm.install("file://%s" % str(src_dir), silent=True)
lm.set_log_level(logging.ERROR)
lm.install("file://%s" % str(src_dir))
installed = lm.get_installed()
assert len(installed) == 4
assert set(["external-repo", "ArduinoJson", "lib-with-deps", "OneWire"]) == set(
@ -269,15 +276,16 @@ def test_install_lib_depndencies(isolated_pio_core, tmpdir_factory):
def test_install_force(isolated_pio_core, tmpdir_factory):
lm = LibraryPackageManager(str(tmpdir_factory.mktemp("lib-storage")))
lm.set_log_level(logging.ERROR)
# install #64 ArduinoJson
pkg = lm.install("64 @ ^5", silent=True)
pkg = lm.install("64 @ ^5")
assert pkg.metadata.version.major == 5
# try install the latest without specification
pkg = lm.install("64", silent=True)
pkg = lm.install("64")
assert pkg.metadata.version.major == 5
assert len(lm.get_installed()) == 1
# re-install the latest
pkg = lm.install(64, silent=True, force=True)
pkg = lm.install(64, force=True)
assert len(lm.get_installed()) == 1
assert pkg.metadata.version.major > 5
@ -364,6 +372,7 @@ def test_uninstall(isolated_pio_core, tmpdir_factory):
tmp_dir = tmpdir_factory.mktemp("tmp")
storage_dir = tmpdir_factory.mktemp("storage")
lm = LibraryPackageManager(str(storage_dir))
lm.set_log_level(logging.ERROR)
# foo @ 1.0.0
pkg_dir = tmp_dir.join("foo").mkdir()
@ -376,42 +385,44 @@ def test_uninstall(isolated_pio_core, tmpdir_factory):
# bar
pkg_dir = tmp_dir.join("bar").mkdir()
pkg_dir.join("library.json").write('{"name": "bar", "version": "1.0.0"}')
bar_pkg = lm.install("file://%s" % pkg_dir, silent=True)
bar_pkg = lm.install("file://%s" % pkg_dir)
assert len(lm.get_installed()) == 3
assert os.path.isdir(os.path.join(str(storage_dir), "foo"))
assert os.path.isdir(os.path.join(str(storage_dir), "foo@1.0.0"))
# check detaching
assert lm.uninstall("FOO", silent=True)
assert lm.uninstall("FOO")
assert len(lm.get_installed()) == 2
assert os.path.isdir(os.path.join(str(storage_dir), "foo"))
assert not os.path.isdir(os.path.join(str(storage_dir), "foo@1.0.0"))
# uninstall the rest
assert lm.uninstall(foo_1_0_0_pkg.path, silent=True)
assert lm.uninstall(bar_pkg, silent=True)
assert lm.uninstall(foo_1_0_0_pkg.path)
assert lm.uninstall(bar_pkg)
assert not lm.get_installed()
# test uninstall dependencies
assert lm.install("AsyncMqttClient-esphome @ 0.8.4", silent=True)
assert lm.install("AsyncMqttClient-esphome @ 0.8.4")
assert len(lm.get_installed()) == 3
assert lm.uninstall("AsyncMqttClient-esphome", silent=True, skip_dependencies=True)
assert lm.uninstall("AsyncMqttClient-esphome", skip_dependencies=True)
assert len(lm.get_installed()) == 2
lm = LibraryPackageManager(str(storage_dir))
assert lm.install("AsyncMqttClient-esphome @ 0.8.4", silent=True)
assert lm.uninstall("AsyncMqttClient-esphome", silent=True)
lm.set_log_level(logging.ERROR)
assert lm.install("AsyncMqttClient-esphome @ 0.8.4")
assert lm.uninstall("AsyncMqttClient-esphome")
assert not lm.get_installed()
def test_registry(isolated_pio_core):
lm = LibraryPackageManager()
lm.set_log_level(logging.ERROR)
# reveal ID
assert lm.reveal_registry_package_id(PackageSpec(id=13)) == 13
assert lm.reveal_registry_package_id(PackageSpec(name="OneWire"), silent=True) == 1
assert lm.reveal_registry_package_id(PackageSpec(name="OneWire")) == 1
with pytest.raises(UnknownPackageError):
lm.reveal_registry_package_id(PackageSpec(name="/non-existing-package/"))
@ -435,14 +446,15 @@ def test_registry(isolated_pio_core):
def test_update_with_metadata(isolated_pio_core, tmpdir_factory):
storage_dir = tmpdir_factory.mktemp("storage")
lm = LibraryPackageManager(str(storage_dir))
lm.set_log_level(logging.ERROR)
# test non SemVer in registry
pkg = lm.install("adafruit/Adafruit NeoPixel @ <1.9", silent=True)
pkg = lm.install("adafruit/Adafruit NeoPixel @ <1.9")
outdated = lm.outdated(pkg)
assert str(outdated.current) == "1.8.7"
assert outdated.latest > semantic_version.Version("1.10.0")
pkg = lm.install("ArduinoJson @ 5.10.1", silent=True)
pkg = lm.install("ArduinoJson @ 5.10.1")
# tesy latest
outdated = lm.outdated(pkg)
assert str(outdated.current) == "5.10.1"
@ -457,14 +469,15 @@ def test_update_with_metadata(isolated_pio_core, tmpdir_factory):
assert outdated.latest > semantic_version.Version("6.16.0")
# update to the wanted 5.x
new_pkg = lm.update("ArduinoJson@^5", PackageSpec("ArduinoJson@^5"), silent=True)
new_pkg = lm.update("ArduinoJson@^5", PackageSpec("ArduinoJson@^5"))
assert str(new_pkg.metadata.version) == "5.13.4"
# check that old version is removed
assert len(lm.get_installed()) == 2
# update to the latest
lm = LibraryPackageManager(str(storage_dir))
pkg = lm.update("ArduinoJson", silent=True)
lm.set_log_level(logging.ERROR)
pkg = lm.update("ArduinoJson")
assert pkg.metadata.version == outdated.latest
@ -485,6 +498,7 @@ def test_update_without_metadata(isolated_pio_core, tmpdir_factory):
# update
lm = LibraryPackageManager(str(storage_dir))
new_pkg = lm.update(pkg, silent=True)
lm.set_log_level(logging.ERROR)
new_pkg = lm.update(pkg)
assert len(lm.get_installed()) == 4
assert new_pkg.metadata.spec.owner == "ottowinter"