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

Refactor HTTP related operations

Browse Source
This commit is contained in:
Ivan Kravets
2020-08-22 17:48:49 +03:00
parent aa186382a8
commit d92c1d3442
29 changed files with 206 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.pylintrc
View File

@ -15,4 +15,8 @@ disable=
useless-object-inheritance,
useless-import-alias,
fixme,
bad-option-value
bad-option-value,
; PY2 Compat
super-with-arguments,
raise-missing-from

7
platformio/app.py
View File

@ -26,7 +26,7 @@ from os import environ, getenv, listdir, remove
from os.path import dirname, isdir, isfile, join, realpath
from time import time
from platformio import __version__, exception, fs, proc, util
from platformio import __version__, exception, fs, proc
from platformio.compat import WINDOWS, dump_json_to_unicode, hashlib_encode_data
from platformio.package.lockfile import LockFile
from platformio.project.helpers import (
@ -394,6 +394,9 @@ def is_disabled_progressbar():
def get_cid():
# pylint: disable=import-outside-toplevel
from platformio.clients.http import fetch_remote_content
cid = get_state_item("cid")
if cid:
return cid
@ -403,7 +406,7 @@ def get_cid():
elif getenv("CHE_API", getenv("CHE_API_ENDPOINT")):
try:
uid = json.loads(
util.fetch_remote_content(
fetch_remote_content(
"{api}/user?token={token}".format(
api=getenv("CHE_API", getenv("CHE_API_ENDPOINT")),
token=getenv("USER_TOKEN"),

14
platformio/clients/account.py
View File

@ -67,7 +67,7 @@ class AccountClient(HTTPClient): # pylint:disable=too-many-public-methods
token = self.fetch_authentication_token()
headers["Authorization"] = "Bearer %s" % token
kwargs["headers"] = headers
return self.request_json_data(*args, **kwargs)
return self.fetch_json_data(*args, **kwargs)
def login(self, username, password):
try:
@ -79,7 +79,7 @@ class AccountClient(HTTPClient): # pylint:disable=too-many-public-methods
app.get_state_item("account", {}).get("email", "")
)
data = self.request_json_data(
data = self.fetch_json_data(
"post", "/v1/login", data={"username": username, "password": password},
)
app.set_state_item("account", data)
@ -95,7 +95,7 @@ class AccountClient(HTTPClient): # pylint:disable=too-many-public-methods
app.get_state_item("account", {}).get("email", "")
)
result = self.request_json_data(
result = self.fetch_json_data(
"post",
"/v1/login/code",
data={"client_id": client_id, "code": code, "redirect_uri": redirect_uri},
@ -107,7 +107,7 @@ class AccountClient(HTTPClient): # pylint:disable=too-many-public-methods
refresh_token = self.get_refresh_token()
self.delete_local_session()
try:
self.request_json_data(
self.fetch_json_data(
"post", "/v1/logout", data={"refresh_token": refresh_token},
)
except AccountError:
@ -133,7 +133,7 @@ class AccountClient(HTTPClient): # pylint:disable=too-many-public-methods
app.get_state_item("account", {}).get("email", "")
)
return self.request_json_data(
return self.fetch_json_data(
"post",
"/v1/registration",
data={
@ -153,7 +153,7 @@ class AccountClient(HTTPClient): # pylint:disable=too-many-public-methods
).get("auth_token")
def forgot_password(self, username):
return self.request_json_data(
return self.fetch_json_data(
"post", "/v1/forgot", data={"username": username},
)
@ -278,7 +278,7 @@ class AccountClient(HTTPClient): # pylint:disable=too-many-public-methods
return auth.get("access_token")
if auth.get("refresh_token"):
try:
data = self.request_json_data(
data = self.fetch_json_data(
"post",
"/v1/login",
headers={

86
platformio/clients/http.py
View File

@ -12,11 +12,24 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import os
import socket
import requests.adapters
from requests.packages.urllib3.util.retry import Retry # pylint:disable=import-error
from platformio import DEFAULT_REQUESTS_TIMEOUT, app, util
from platformio.exception import PlatformioException
from platformio.exception import PlatformioException, UserSideException
PING_REMOTE_HOSTS = [
"140.82.118.3", # Github.com
"35.231.145.151", # Gitlab.com
"88.198.170.159", # platformio.org
"github.com",
"platformio.org",
]
class HTTPClientError(PlatformioException):
@ -29,6 +42,15 @@ class HTTPClientError(PlatformioException):
return self.message
class InternetIsOffline(UserSideException):
MESSAGE = (
"You are not connected to the Internet.\n"
"PlatformIO needs the Internet connection to"
" download dependent packages or to work with PIO Account."
)
class HTTPClient(object):
def __init__(
self, base_url,
@ -57,7 +79,7 @@ class HTTPClient(object):
def send_request(self, method, path, **kwargs):
# check Internet before and resolve issue with 60 seconds timeout
# print(self, method, path, kwargs)
util.internet_on(raise_exception=True)
ensure_internet_on(raise_exception=True)
# set default timeout
if "timeout" not in kwargs:
@ -68,9 +90,18 @@ class HTTPClient(object):
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
raise HTTPClientError(str(e))
def request_json_data(self, *args, **kwargs):
response = self.send_request(*args, **kwargs)
return self.raise_error_from_response(response)
def fetch_json_data(self, *args, **kwargs):
cache_valid = kwargs.pop("cache_valid") if "cache_valid" in kwargs else None
if not cache_valid:
return self.raise_error_from_response(self.send_request(*args, **kwargs))
cache_key = app.ContentCache.key_from_args(*args, kwargs)
with app.ContentCache() as cc:
result = cc.get(cache_key)
if result is not None:
return json.loads(result)
response = self.send_request(*args, **kwargs)
cc.set(cache_key, response.text, cache_valid)
return self.raise_error_from_response(response)
@staticmethod
def raise_error_from_response(response, expected_codes=(200, 201, 202)):
@ -84,3 +115,48 @@ class HTTPClient(object):
except (KeyError, ValueError):
message = response.text
raise HTTPClientError(message, response)
#
# Helpers
#
@util.memoized(expire="10s")
def _internet_on():
timeout = 2
socket.setdefaulttimeout(timeout)
for host in PING_REMOTE_HOSTS:
try:
for var in ("HTTP_PROXY", "HTTPS_PROXY"):
if not os.getenv(var) and not os.getenv(var.lower()):
continue
requests.get("http://%s" % host, allow_redirects=False, timeout=timeout)
return True
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((host, 80))
s.close()
return True
except: # pylint: disable=bare-except
pass
return False
def ensure_internet_on(raise_exception=False):
result = _internet_on()
if raise_exception and not result:
raise InternetIsOffline()
return result
def fetch_remote_content(*args, **kwargs):
kwargs["headers"] = kwargs.get("headers", {})
if "User-Agent" not in kwargs["headers"]:
kwargs["headers"]["User-Agent"] = app.get_user_agent()
if "timeout" not in kwargs:
kwargs["timeout"] = DEFAULT_REQUESTS_TIMEOUT
r = requests.get(*args, **kwargs)
r.raise_for_status()
return r.text

4
platformio/commands/debug/command.py
View File

@ -21,7 +21,7 @@ from os.path import isfile
import click
from platformio import app, exception, fs, proc, util
from platformio import app, exception, fs, proc
from platformio.commands.debug import helpers
from platformio.commands.debug.exception import DebugInvalidOptionsError
from platformio.package.manager.core import inject_contrib_pysite
@ -130,7 +130,7 @@ def cli(ctx, project_dir, project_conf, environment, verbose, interface, __unpro
nl=False,
)
stream = helpers.GDBMIConsoleStream()
with util.capture_std_streams(stream):
with proc.capture_std_streams(stream):
helpers.predebug_project(ctx, project_dir, env_name, preload, verbose)
stream.close()
else:

3
platformio/commands/home/rpc/handlers/os.py
View File

@ -23,6 +23,7 @@ import click
from twisted.internet import defer # pylint: disable=import-error
from platformio import DEFAULT_REQUESTS_TIMEOUT, app, fs, util
from platformio.clients.http import ensure_internet_on
from platformio.commands.home import helpers
from platformio.compat import PY2, get_filesystem_encoding, glob_recursive
@ -47,7 +48,7 @@ class OSRPC(object):
defer.returnValue(result)
# check internet before and resolve issue with 60 seconds timeout
util.internet_on(raise_exception=True)
ensure_internet_on(raise_exception=True)
session = helpers.requests_session()
if data:

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

@ -28,7 +28,7 @@ from platformio.commands.lib.helpers import (
save_project_libdeps,
)
from platformio.compat import dump_json_to_unicode
from platformio.package.exception import UnknownPackageError
from platformio.package.exception import NotGlobalLibDir, UnknownPackageError
from platformio.package.manager.library import LibraryPackageManager
from platformio.package.meta import PackageItem, PackageSpec
from platformio.proc import is_ci
@ -97,7 +97,7 @@ def cli(ctx, **options):
)
if not storage_dirs:
raise exception.NotGlobalLibDir(
raise NotGlobalLibDir(
get_project_dir(), get_project_global_lib_dir(), ctx.invoked_subcommand
)

4
platformio/commands/remote/client/run_or_test.py
View File

@ -20,7 +20,7 @@ from io import BytesIO
from twisted.spread import pb # pylint: disable=import-error
from platformio import util
from platformio import fs
from platformio.commands.remote.client.async_base import AsyncClientBase
from platformio.commands.remote.projectsync import PROJECT_SYNC_STAGE, ProjectSync
from platformio.compat import hashlib_encode_data
@ -64,7 +64,7 @@ class RunOrTestClient(AsyncClientBase):
return "%s-%s" % (os.path.basename(path), h.hexdigest())
def add_project_items(self, psync):
with util.cd(self.options["project_dir"]):
with fs.cd(self.options["project_dir"]):
cfg = ProjectConfig.get_instance(
os.path.join(self.options["project_dir"], "platformio.ini")
)

7
platformio/commands/upgrade.py
View File

@ -19,7 +19,8 @@ from zipfile import ZipFile
import click
from platformio import VERSION, __version__, app, exception, util
from platformio import VERSION, __version__, app, exception
from platformio.clients.http import fetch_remote_content
from platformio.compat import WINDOWS
from platformio.proc import exec_command, get_pythonexe_path
from platformio.project.helpers import get_project_cache_dir
@ -130,7 +131,7 @@ def get_latest_version():
def get_develop_latest_version():
version = None
content = util.fetch_remote_content(
content = fetch_remote_content(
"https://raw.githubusercontent.com/platformio/platformio"
"/develop/platformio/__init__.py"
)
@ -150,5 +151,5 @@ def get_develop_latest_version():
def get_pypi_latest_version():
content = util.fetch_remote_content("https://pypi.org/pypi/platformio/json")
content = fetch_remote_content("https://pypi.org/pypi/platformio/json")
return json.loads(content)["info"]["version"]

73
platformio/exception.py
View File

@ -30,10 +30,6 @@ class ReturnErrorCode(PlatformioException):
MESSAGE = "{0}"
class LockFileTimeoutError(PlatformioException):
pass
class MinitermException(PlatformioException):
pass
@ -47,61 +43,6 @@ class AbortedByUser(UserSideException):
MESSAGE = "Aborted by user"
# Package Manager
class PlatformIOPackageException(PlatformioException):
pass
class UnknownPackage(UserSideException):
MESSAGE = "Detected unknown package '{0}'"
class MissingPackageManifest(PlatformIOPackageException):
MESSAGE = "Could not find one of '{0}' manifest files in the package"
class UndefinedPackageVersion(PlatformIOPackageException):
MESSAGE = (
"Could not find a version that satisfies the requirement '{0}'"
" for your system '{1}'"
)
class PackageInstallError(PlatformIOPackageException):
MESSAGE = (
"Could not install '{0}' with version requirements '{1}' "
"for your system '{2}'.\n\n"
"Please try this solution -> http://bit.ly/faq-package-manager"
)
#
# Library
#
class NotGlobalLibDir(UserSideException):
MESSAGE = (
"The `{0}` is not a PlatformIO project.\n\n"
"To manage libraries in global storage `{1}`,\n"
"please use `platformio lib --global {2}` or specify custom storage "
"`platformio lib --storage-dir /path/to/storage/ {2}`.\n"
"Check `platformio lib --help` for details."
)
class InvalidLibConfURL(UserSideException):
MESSAGE = "Invalid library config URL '{0}'"
#
# UDEV Rules
#
@ -143,20 +84,6 @@ class GetLatestVersionError(PlatformioException):
MESSAGE = "Can not retrieve the latest PlatformIO version"
class APIRequestError(PlatformioException):
MESSAGE = "[API] {0}"
class InternetIsOffline(UserSideException):
MESSAGE = (
"You are not connected to the Internet.\n"
"PlatformIO needs the Internet connection to"
" download dependent packages or to work with PIO Account."
)
class InvalidSettingName(UserSideException):
MESSAGE = "Invalid setting with the name '{0}'"

4
platformio/fs.py
View File

@ -176,6 +176,10 @@ def expanduser(path):
return os.environ["USERPROFILE"] + path[1:]
def change_filemtime(path, mtime):
os.utime(path, (mtime, mtime))
def rmtree(path):
def _onerror(func, path, __):
try:

9
platformio/maintenance.py
View File

@ -20,6 +20,7 @@ import click
import semantic_version
from platformio import __version__, app, exception, fs, telemetry, util
from platformio.clients import http
from platformio.commands import PlatformioCLI
from platformio.commands.lib.command import CTX_META_STORAGE_DIRS_KEY
from platformio.commands.lib.command import lib_update as cmd_lib_update
@ -53,9 +54,9 @@ def on_platformio_end(ctx, result): # pylint: disable=unused-argument
check_internal_updates(ctx, "platforms")
check_internal_updates(ctx, "libraries")
except (
exception.InternetIsOffline,
http.HTTPClientError,
http.InternetIsOffline,
exception.GetLatestVersionError,
exception.APIRequestError,
):
click.secho(
"Failed to check for PlatformIO upgrades. "
@ -221,7 +222,7 @@ def check_platformio_upgrade():
last_check["platformio_upgrade"] = int(time())
app.set_state_item("last_check", last_check)
util.internet_on(raise_exception=True)
http.ensure_internet_on(raise_exception=True)
# Update PlatformIO's Core packages
update_core_packages(silent=True)
@ -268,7 +269,7 @@ def check_internal_updates(ctx, what): # pylint: disable=too-many-branches
last_check[what + "_update"] = int(time())
app.set_state_item("last_check", last_check)
util.internet_on(raise_exception=True)
http.ensure_internet_on(raise_exception=True)
outdated_items = []
pm = PlatformPackageManager() if what == "platforms" else LibraryPackageManager()

4
platformio/package/download.py
View File

@ -21,7 +21,7 @@ from time import mktime
import click
import requests
from platformio import DEFAULT_REQUESTS_TIMEOUT, app, fs, util
from platformio import DEFAULT_REQUESTS_TIMEOUT, app, fs
from platformio.package.exception import PackageException
@ -134,7 +134,7 @@ class FileDownloader(object):
def _preserve_filemtime(self, lmdate):
timedata = parsedate_tz(lmdate)
lmtime = mktime(timedata[:9])
util.change_filemtime(self._destination, lmtime)
fs.change_filemtime(self._destination, lmtime)
def __del__(self):
if self._request:

11
platformio/package/exception.py
View File

@ -58,3 +58,14 @@ class UnknownPackageError(UserSideException):
"Could not find a package with '{0}' requirements for your system '%s'"
% util.get_systype()
)
class NotGlobalLibDir(UserSideException):
MESSAGE = (
"The `{0}` is not a PlatformIO project.\n\n"
"To manage libraries in global storage `{1}`,\n"
"please use `platformio lib --global {2}` or specify custom storage "
"`platformio lib --storage-dir /path/to/storage/ {2}`.\n"
"Check `platformio lib --help` for details."
)

10
platformio/package/lockfile.py
View File

@ -15,7 +15,7 @@
import os
from time import sleep, time
from platformio import exception
from platformio.exception import PlatformioException
LOCKFILE_TIMEOUT = 3600 # in seconds, 1 hour
LOCKFILE_DELAY = 0.2
@ -36,7 +36,11 @@ except ImportError:
LOCKFILE_CURRENT_INTERFACE = None
class LockFileExists(Exception):
class LockFileExists(PlatformioException):
pass
class LockFileTimeoutError(PlatformioException):
pass
@ -88,7 +92,7 @@ class LockFile(object):
sleep(self.delay)
elapsed += self.delay
raise exception.LockFileTimeoutError()
raise LockFileTimeoutError()
def release(self):
self._unlock()

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

@ -16,10 +16,10 @@ import os
import click
from platformio import util
from platformio.package.exception import UnknownPackageError
from platformio.package.meta import PackageItem, PackageOutdatedResult, PackageSpec
from platformio.package.vcsclient import VCSBaseException, VCSClientFactory
from platformio.clients.http import ensure_internet_on
class PackageManagerUpdateMixin(object):
@ -97,7 +97,7 @@ class PackageManagerUpdateMixin(object):
),
nl=False,
)
if not util.internet_on():
if not ensure_internet_on():
if not silent:
click.echo("[%s]" % (click.style("Off-line", fg="yellow")))
return pkg

4
platformio/package/manager/core.py
View File

@ -17,7 +17,7 @@ import os
import subprocess
import sys
from platformio import __core_packages__, exception, util
from platformio import __core_packages__, fs, exception, util
from platformio.compat import PY2
from platformio.package.manager.tool import ToolPackageManager
from platformio.package.meta import PackageSpec
@ -93,7 +93,7 @@ def inject_contrib_pysite(verify_openssl=False):
def build_contrib_pysite_deps(target_dir):
if os.path.isdir(target_dir):
util.rmtree_(target_dir)
fs.rmtree(target_dir)
os.makedirs(target_dir)
with open(os.path.join(target_dir, "package.json"), "w") as fp:
json.dump(

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

@ -13,7 +13,7 @@
# limitations under the License.
from platformio import util
from platformio.exception import APIRequestError, InternetIsOffline
from platformio.clients.http import HTTPClientError, InternetIsOffline
from platformio.package.exception import UnknownPackageError
from platformio.package.manager.base import BasePackageManager
from platformio.package.manager.tool import ToolPackageManager
@ -176,7 +176,7 @@ class PlatformPackageManager(BasePackageManager): # pylint: disable=too-many-an
key = "%s:%s" % (board["platform"], board["id"])
if key not in know_boards:
boards.append(board)
except (APIRequestError, InternetIsOffline):
except (HTTPClientError, InternetIsOffline):
pass
return sorted(boards, key=lambda b: b["name"])

3
platformio/package/manifest/parser.py
View File

@ -20,6 +20,7 @@ import re
import tarfile
from platformio import util
from platformio.clients.http import fetch_remote_content
from platformio.compat import get_object_members, string_types
from platformio.package.exception import ManifestParserError, UnknownManifestError
from platformio.project.helpers import is_platformio_project
@ -106,7 +107,7 @@ class ManifestParserFactory(object):
@staticmethod
def new_from_url(remote_url):
content = util.fetch_remote_content(remote_url)
content = fetch_remote_content(remote_url)
return ManifestParserFactory.new(
content,
ManifestFileType.from_uri(remote_url) or ManifestFileType.LIBRARY_JSON,

4
platformio/package/manifest/schema.py
View File

@ -21,7 +21,7 @@ import requests
import semantic_version
from marshmallow import Schema, ValidationError, fields, validate, validates
from platformio import util
from platformio.clients.http import fetch_remote_content
from platformio.package.exception import ManifestValidationError
from platformio.util import memoized
@ -256,4 +256,4 @@ class ManifestSchema(BaseSchema):
"https://raw.githubusercontent.com/spdx/license-list-data"
"/v%s/json/licenses.json" % version
)
return json.loads(util.fetch_remote_content(spdx_data_url))
return json.loads(fetch_remote_content(spdx_data_url))

4
platformio/package/unpack.py
View File

@ -19,7 +19,7 @@ from zipfile import ZipFile
import click
from platformio import util
from platformio import fs
from platformio.package.exception import PackageException
@ -109,7 +109,7 @@ class ZIPArchiver(BaseArchiver):
@staticmethod
def preserve_mtime(item, dest_dir):
util.change_filemtime(
fs.change_filemtime(
os.path.join(dest_dir, item.filename),
mktime(tuple(item.date_time) + tuple([0, 0, 0])),
)

12
platformio/proc.py
View File

@ -15,6 +15,7 @@
import os
import subprocess
import sys
from contextlib import contextmanager
from threading import Thread
from platformio import exception
@ -137,6 +138,17 @@ def exec_command(*args, **kwargs):
return result
@contextmanager
def capture_std_streams(stdout, stderr=None):
_stdout = sys.stdout
_stderr = sys.stderr
sys.stdout = stdout
sys.stderr = stderr or stdout
yield
sys.stdout = _stdout
sys.stderr = _stderr
def is_ci():
return os.getenv("CI", "").lower() == "true"

2
platformio/telemetry.py
View File

@ -124,7 +124,7 @@ class MeasurementProtocol(TelemetryBase):
caller_id = str(app.get_session_var("caller_id"))
self["cd1"] = util.get_systype()
self["cd4"] = (
1 if (not util.is_ci() and (caller_id or not is_container())) else 0
1 if (not is_ci() and (caller_id or not is_container())) else 0
)
if caller_id:
self["cd5"] = caller_id.lower()

112
platformio/util.py
View File

@ -19,26 +19,17 @@ import math
import os
import platform
import re
import socket
import sys
import time
from contextlib import contextmanager
from functools import wraps
from glob import glob
import click
import requests
from platformio import DEFAULT_REQUESTS_TIMEOUT, __apiurl__, __version__, exception
from platformio.commands import PlatformioCLI
from platformio import __version__, exception, proc
from platformio.compat import PY2, WINDOWS
from platformio.fs import cd # pylint: disable=unused-import
from platformio.fs import load_json # pylint: disable=unused-import
from platformio.fs import rmtree as rmtree_ # pylint: disable=unused-import
from platformio.proc import exec_command # pylint: disable=unused-import
from platformio.proc import is_ci # pylint: disable=unused-import
# KEEP unused imports for backward compatibility with PIO Core 3.0 API
class memoized(object):
@ -97,17 +88,6 @@ def singleton(cls):
return get_instance
@contextmanager
def capture_std_streams(stdout, stderr=None):
_stdout = sys.stdout
_stderr = sys.stderr
sys.stdout = stdout
sys.stderr = stderr or stdout
yield
sys.stdout = _stdout
sys.stderr = _stderr
def get_systype():
type_ = platform.system().lower()
arch = platform.machine().lower()
@ -116,16 +96,6 @@ def get_systype():
return "%s_%s" % (type_, arch) if arch else type_
def pioversion_to_intstr():
vermatch = re.match(r"^([\d\.]+)", __version__)
assert vermatch
return [int(i) for i in vermatch.group(1).split(".")[:3]]
def change_filemtime(path, mtime):
os.utime(path, (mtime, mtime))
def get_serial_ports(filter_hwid=False):
try:
# pylint: disable=import-outside-toplevel
@ -164,7 +134,7 @@ def get_logical_devices():
items = []
if WINDOWS:
try:
result = exec_command(
result = proc.exec_command(
["wmic", "logicaldisk", "get", "name,VolumeName"]
).get("out", "")
devicenamere = re.compile(r"^([A-Z]{1}\:)\s*(\S+)?")
@ -177,12 +147,12 @@ def get_logical_devices():
except WindowsError: # pylint: disable=undefined-variable
pass
# try "fsutil"
result = exec_command(["fsutil", "fsinfo", "drives"]).get("out", "")
result = proc.exec_command(["fsutil", "fsinfo", "drives"]).get("out", "")
for device in re.findall(r"[A-Z]:\\", result):
items.append({"path": device, "name": None})
return items
result = exec_command(["df"]).get("out")
result = proc.exec_command(["df"]).get("out")
devicenamere = re.compile(r"^/.+\d+\%\s+([a-z\d\-_/]+)$", flags=re.I)
for line in result.split("\n"):
match = devicenamere.match(line.strip())
@ -370,60 +340,27 @@ def get_api_result(url, params=None, data=None, auth=None, cache_valid=None):
)
PING_REMOTE_HOSTS = [
"140.82.118.3", # Github.com
"35.231.145.151", # Gitlab.com
"88.198.170.159", # platformio.org
"github.com",
"platformio.org",
]
@memoized(expire="10s")
def _internet_on():
timeout = 2
socket.setdefaulttimeout(timeout)
for host in PING_REMOTE_HOSTS:
try:
for var in ("HTTP_PROXY", "HTTPS_PROXY"):
if not os.getenv(var, var.lower()):
continue
requests.get("http://%s" % host, allow_redirects=False, timeout=timeout)
return True
socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect((host, 80))
return True
except: # pylint: disable=bare-except
pass
return False
def internet_on(raise_exception=False):
result = _internet_on()
if raise_exception and not result:
raise exception.InternetIsOffline()
return result
def fetch_remote_content(*args, **kwargs):
# pylint: disable=import-outside-toplevel
from platformio.app import get_user_agent
kwargs["headers"] = kwargs.get("headers", {})
if "User-Agent" not in kwargs["headers"]:
kwargs["headers"]["User-Agent"] = get_user_agent()
if "timeout" not in kwargs:
kwargs["timeout"] = DEFAULT_REQUESTS_TIMEOUT
r = requests.get(*args, **kwargs)
r.raise_for_status()
return r.text
def pioversion_to_intstr():
vermatch = re.match(r"^([\d\.]+)", __version__)
assert vermatch
return [int(i) for i in vermatch.group(1).split(".")[:3]]
def pepver_to_semver(pepver):
return re.sub(r"(\.\d+)\.?(dev|a|b|rc|post)", r"\1-\2.", pepver, 1)
def get_original_version(version):
if version.count(".") != 2:
return None
_, raw = version.split(".")[:2]
if int(raw) <= 99:
return None
if int(raw) <= 9999:
return "%s.%s" % (raw[:-2], int(raw[-2:]))
return "%s.%s.%s" % (raw[:-4], int(raw[-4:-2]), int(raw[-2:]))
def items_to_list(items):
if isinstance(items, list):
return items
@ -472,14 +409,3 @@ def humanize_duration_time(duration):
tokens.append(int(round(duration) if multiplier == 1 else fraction))
duration -= fraction * multiplier
return "{:02d}:{:02d}:{:02d}.{:03d}".format(*tokens)
def get_original_version(version):
if version.count(".") != 2:
return None
_, raw = version.split(".")[:2]
if int(raw) <= 99:
return None
if int(raw) <= 9999:
return "%s.%s" % (raw[:-2], int(raw[-2:]))
return "%s.%s.%s" % (raw[:-4], int(raw[-4:-2]), int(raw[-2:]))

6
tests/commands/test_check.py
View File

@ -389,7 +389,7 @@ check_tool = pvs-studio
assert style == 0
def test_check_embedded_platform_all_tools(clirunner, tmpdir):
def test_check_embedded_platform_all_tools(clirunner, validate_cliresult, tmpdir):
config = """
[env:test]
platform = ststm32
@ -422,11 +422,9 @@ int main() {
for framework in frameworks:
for tool in ("cppcheck", "clangtidy", "pvs-studio"):
tmpdir.join("platformio.ini").write(config % (framework, tool))
result = clirunner.invoke(cmd_check, ["--project-dir", str(tmpdir)])
validate_cliresult(result)
defects = sum(count_defects(result.output))
assert result.exit_code == 0 and defects > 0, "Failed %s with %s" % (
framework,
tool,

4
tests/commands/test_test.py
View File

@ -16,12 +16,12 @@ import os
import pytest
from platformio import util
from platformio import proc
from platformio.commands.test.command import cli as cmd_test
def test_local_env():
result = util.exec_command(
result = proc.exec_command(
[
"platformio",
"test",

4
tests/conftest.py
View File

@ -20,7 +20,7 @@ import time
import pytest
from click.testing import CliRunner
from platformio import util
from platformio.clients import http
def pytest_configure(config):
@ -74,7 +74,7 @@ def isolated_pio_core(request, tmpdir_factory):
@pytest.fixture(scope="function")
def without_internet(monkeypatch):
monkeypatch.setattr(util, "_internet_on", lambda: False)
monkeypatch.setattr(http, "_internet_on", lambda: False)
@pytest.fixture

8
tests/test_examples.py
View File

@ -19,7 +19,7 @@ from os.path import basename, dirname, getsize, isdir, isfile, join, normpath
import pytest
from platformio import util
from platformio import fs, proc
from platformio.compat import PY2
from platformio.package.manager.platform import PlatformPackageManager
from platformio.platform.factory import PlatformFactory
@ -64,14 +64,14 @@ def pytest_generate_tests(metafunc):
def test_run(pioproject_dir):
with util.cd(pioproject_dir):
with fs.cd(pioproject_dir):
config = ProjectConfig()
build_dir = config.get_optional_dir("build")
if isdir(build_dir):
util.rmtree_(build_dir)
fs.rmtree(build_dir)
env_names = config.envs()
result = util.exec_command(
result = proc.exec_command(
["platformio", "run", "-e", random.choice(env_names)]
)
if result["returncode"] != 0:

22
tests/test_misc.py
View File

@ -17,29 +17,33 @@
import pytest
import requests
from platformio import exception, util
from platformio import proc
from platformio.clients import http
from platformio.clients.registry import RegistryClient
def test_platformio_cli():
result = util.exec_command(["pio", "--help"])
result = proc.exec_command(["pio", "--help"])
assert result["returncode"] == 0
# pylint: disable=unsupported-membership-test
assert "Usage: pio [OPTIONS] COMMAND [ARGS]..." in result["out"]
def test_ping_internet_ips():
for host in util.PING_REMOTE_HOSTS:
for host in http.PING_REMOTE_HOSTS:
requests.get("http://%s" % host, allow_redirects=False, timeout=2)
def test_api_internet_offline(without_internet, isolated_pio_core):
with pytest.raises(exception.InternetIsOffline):
util.get_api_result("/stats")
regclient = RegistryClient()
with pytest.raises(http.InternetIsOffline):
regclient.fetch_json_data("get", "/v2/stats")
def test_api_cache(monkeypatch, isolated_pio_core):
api_kwargs = {"url": "/stats", "cache_valid": "10s"}
result = util.get_api_result(**api_kwargs)
regclient = RegistryClient()
api_kwargs = {"method": "get", "path": "/v2/stats", "cache_valid": "10s"}
result = regclient.fetch_json_data(**api_kwargs)
assert result and "boards" in result
monkeypatch.setattr(util, "_internet_on", lambda: False)
assert util.get_api_result(**api_kwargs) == result
monkeypatch.setattr(http, "_internet_on", lambda: False)
assert regclient.fetch_json_data(**api_kwargs) == result