Initial commit of PIO Unit Testing

This commit is contained in:
Ivan Kravets
2019-04-19 20:33:31 +03:00
parent c7d8b50474
commit 948a977fa5
7 changed files with 577 additions and 67 deletions

View File

@ -1,67 +0,0 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
from os import getcwd
import click
from platformio.managers.core import pioplus_call
@click.command("test", short_help="Local Unit Testing")
@click.option("--environment", "-e", multiple=True, metavar="<environment>")
@click.option(
"--filter",
"-f",
multiple=True,
metavar="<pattern>",
help="Filter tests by a pattern")
@click.option(
"--ignore",
"-i",
multiple=True,
metavar="<pattern>",
help="Ignore tests by a pattern")
@click.option("--upload-port")
@click.option("--test-port")
@click.option(
"-d",
"--project-dir",
default=getcwd,
type=click.Path(
exists=True,
file_okay=False,
dir_okay=True,
writable=True,
resolve_path=True))
@click.option("--without-building", is_flag=True)
@click.option("--without-uploading", is_flag=True)
@click.option(
"--no-reset",
is_flag=True,
help="Disable software reset via Serial.DTR/RST")
@click.option(
"--monitor-rts",
default=None,
type=click.IntRange(0, 1),
help="Set initial RTS line state for Serial Monitor")
@click.option(
"--monitor-dtr",
default=None,
type=click.IntRange(0, 1),
help="Set initial DTR line state for Serial Monitor")
@click.option("--verbose", "-v", is_flag=True)
def cli(*args, **kwargs): # pylint: disable=unused-argument
pioplus_call(sys.argv[1:])

View File

@ -0,0 +1,15 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from platformio.commands.test.command import cli

View File

@ -0,0 +1,181 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from fnmatch import fnmatch
from os import getcwd, listdir
from os.path import isdir, join
from time import time
import click
from platformio import exception, util
from platformio.commands.run import check_project_envs, print_header
from platformio.commands.test.embedded import EmbeddedTestProcessor
from platformio.commands.test.native import NativeTestProcessor
@click.command("test", short_help="Unit Testing")
@click.option("--environment", "-e", multiple=True, metavar="<environment>")
@click.option(
"--filter",
"-f",
multiple=True,
metavar="<pattern>",
help="Filter tests by a pattern")
@click.option(
"--ignore",
"-i",
multiple=True,
metavar="<pattern>",
help="Ignore tests by a pattern")
@click.option("--upload-port")
@click.option("--test-port")
@click.option(
"-d",
"--project-dir",
default=getcwd,
type=click.Path(
exists=True,
file_okay=False,
dir_okay=True,
writable=True,
resolve_path=True))
@click.option("--without-building", is_flag=True)
@click.option("--without-uploading", is_flag=True)
@click.option("--without-testing", is_flag=True)
@click.option("--no-reset", is_flag=True)
@click.option(
"--monitor-rts",
default=None,
type=click.IntRange(0, 1),
help="Set initial RTS line state for Serial Monitor")
@click.option(
"--monitor-dtr",
default=None,
type=click.IntRange(0, 1),
help="Set initial DTR line state for Serial Monitor")
@click.option("--verbose", "-v", is_flag=True)
@click.pass_context
def cli( # pylint: disable=redefined-builtin
ctx, environment, ignore, filter, upload_port, test_port, project_dir,
without_building, without_uploading, without_testing, no_reset,
monitor_rts, monitor_dtr, verbose):
with util.cd(project_dir):
test_dir = util.get_projecttest_dir()
if not isdir(test_dir):
raise exception.TestDirNotExists(test_dir)
test_names = get_test_names(test_dir)
projectconf = util.load_project_config()
env_default = None
if projectconf.has_option("platformio", "env_default"):
env_default = util.parse_conf_multi_values(
projectconf.get("platformio", "env_default"))
assert check_project_envs(projectconf, environment or env_default)
click.echo("Verbose mode can be enabled via `-v, --verbose` option")
click.echo("Collected %d items" % len(test_names))
start_time = time()
results = []
for testname in test_names:
for section in projectconf.sections():
if not section.startswith("env:"):
continue
# filter and ignore patterns
patterns = dict(filter=list(filter), ignore=list(ignore))
for key in patterns:
if projectconf.has_option(section, "test_%s" % key):
patterns[key].extend([
p.strip()
for p in projectconf.get(section, "test_%s" %
key).split(", ")
if p.strip()
])
envname = section[4:]
skip_conditions = [
environment and envname not in environment,
not environment and env_default
and envname not in env_default,
testname != "*" and patterns['filter'] and
not any([fnmatch(testname, p)
for p in patterns['filter']]),
testname != "*"
and any([fnmatch(testname, p)
for p in patterns['ignore']]),
]
if any(skip_conditions):
results.append((None, testname, envname))
continue
cls = (NativeTestProcessor if projectconf.get(
section, "platform") == "native" else
EmbeddedTestProcessor)
tp = cls(
ctx, testname, envname,
dict(
project_config=projectconf,
project_dir=project_dir,
upload_port=upload_port,
test_port=test_port,
without_building=without_building,
without_uploading=without_uploading,
without_testing=without_testing,
no_reset=no_reset,
monitor_rts=monitor_rts,
monitor_dtr=monitor_dtr,
verbose=verbose))
results.append((tp.process(), testname, envname))
if without_testing:
return
click.echo()
print_header("[%s]" % click.style("TEST SUMMARY"))
passed = True
for result in results:
status, testname, envname = result
status_str = click.style("PASSED", fg="green")
if status is False:
passed = False
status_str = click.style("FAILED", fg="red")
elif status is None:
status_str = click.style("IGNORED", fg="yellow")
click.echo(
"test/%s/env:%s\t[%s]" % (click.style(testname, fg="yellow"),
click.style(envname, fg="cyan"),
status_str),
err=status is False)
print_header(
"[%s] Took %.2f seconds" % (
(click.style("PASSED", fg="green", bold=True) if passed else
click.style("FAILED", fg="red", bold=True)), time() - start_time),
is_error=not passed)
if not passed:
raise exception.ReturnErrorCode(1)
def get_test_names(test_dir):
names = []
for item in sorted(listdir(test_dir)):
if isdir(join(test_dir, item)):
names.append(item)
if not names:
names = ["*"]
return names

View File

@ -0,0 +1,133 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from time import sleep
import click
import serial
from platformio import exception, util
from platformio.commands.test.processor import TestProcessorBase
from platformio.managers.platform import PlatformFactory
class EmbeddedTestProcessor(TestProcessorBase):
SERIAL_TIMEOUT = 600
def process(self):
if not self.options['without_building']:
self.print_progress("Building... (1/3)")
target = ["__test"]
if self.options['without_uploading']:
target.append("checkprogsize")
self.build_or_upload(target)
if not self.options['without_uploading']:
self.print_progress("Uploading... (2/3)")
target = ["upload"]
if self.options['without_building']:
target.append("nobuild")
else:
target.append("__test")
self.build_or_upload(target)
if self.options['without_testing']:
return None
self.print_progress("Testing... (3/3)")
return self.run()
def run(self):
click.echo("If you don't see any output for the first 10 secs, "
"please reset board (press reset button)")
click.echo()
try:
ser = serial.Serial(
baudrate=self.get_baudrate(), timeout=self.SERIAL_TIMEOUT)
ser.port = self.get_test_port()
ser.rts = self.options['monitor_rts']
ser.dtr = self.options['monitor_dtr']
ser.open()
except serial.SerialException as e:
click.secho(str(e), fg="red", err=True)
return False
if not self.options['no_reset']:
ser.flushInput()
ser.setDTR(False)
ser.setRTS(False)
sleep(0.1)
ser.setDTR(True)
ser.setRTS(True)
sleep(0.1)
while True:
line = ser.readline().strip()
# fix non-ascii output from device
for i, c in enumerate(line[::-1]):
if not isinstance(c, int):
c = ord(c)
if c > 127:
line = line[-i:]
break
if not line:
continue
if isinstance(line, bytes):
line = line.decode("utf8")
self.on_run_out(line)
if all([l in line for l in ("Tests", "Failures", "Ignored")]):
break
ser.close()
return not self._run_failed
def get_test_port(self):
# if test port is specified manually or in config
if self.options.get("test_port"):
return self.options.get("test_port")
if self.env_options.get("test_port"):
return self.env_options.get("test_port")
assert set(["platform", "board"]) & set(self.env_options.keys())
p = PlatformFactory.newPlatform(self.env_options['platform'])
board_hwids = p.board_config(self.env_options['board']).get(
"build.hwids", [])
port = None
elapsed = 0
while elapsed < 5 and not port:
for item in util.get_serialports():
port = item['port']
for hwid in board_hwids:
hwid_str = ("%s:%s" % (hwid[0], hwid[1])).replace("0x", "")
if hwid_str in item['hwid']:
return port
# check if port is already configured
try:
serial.Serial(port, timeout=self.SERIAL_TIMEOUT).close()
except serial.SerialException:
port = None
if not port:
sleep(0.25)
elapsed += 0.25
if not port:
raise exception.PlatformioException(
"Please specify `test_port` for environment or use "
"global `--test-port` option.")
return port

View File

@ -0,0 +1,39 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from os.path import join
from platformio import util
from platformio.commands.test.processor import TestProcessorBase
class NativeTestProcessor(TestProcessorBase):
def process(self):
if not self.options['without_building']:
self.print_progress("Building... (1/2)")
self.build_or_upload(["__test"])
if self.options['without_testing']:
return None
self.print_progress("Testing... (2/2)")
return self.run()
def run(self):
with util.cd(self.options['project_dir']):
build_dir = util.get_projectbuild_dir()
result = util.exec_command([join(build_dir, self.env_name, "program")],
stdout=util.AsyncPipe(self.on_run_out),
stderr=util.AsyncPipe(self.on_run_out))
assert "returncode" in result
return result['returncode'] == 0 and not self._run_failed

View File

@ -0,0 +1,200 @@
# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import atexit
from os import remove
from os.path import isdir, isfile, join
from string import Template
import click
from platformio import exception, util
from platformio.commands.run import cli as cmd_run
from platformio.commands.run import print_header
TRANSPORT_OPTIONS = {
"arduino": {
"include": "#include <Arduino.h>",
"object": "",
"putchar": "Serial.write(c)",
"flush": "Serial.flush()",
"begin": "Serial.begin($baudrate)",
"end": "Serial.end()"
},
"mbed": {
"include": "#include <mbed.h>",
"object": "Serial pc(USBTX, USBRX);",
"putchar": "pc.putc(c)",
"flush": "",
"begin": "pc.baud($baudrate)",
"end": ""
},
"energia": {
"include": "#include <Energia.h>",
"object": "",
"putchar": "Serial.write(c)",
"flush": "Serial.flush()",
"begin": "Serial.begin($baudrate)",
"end": "Serial.end()"
},
"espidf": {
"include": "#include <stdio.h>",
"object": "",
"putchar": "putchar(c)",
"flush": "fflush(stdout)",
"begin": "",
"end": ""
},
"native": {
"include": "#include <stdio.h>",
"object": "",
"putchar": "putchar(c)",
"flush": "fflush(stdout)",
"begin": "",
"end": ""
},
"custom": {
"include": '#include "unittest_transport.h"',
"object": "",
"putchar": "unittest_uart_putchar(c)",
"flush": "unittest_uart_flush()",
"begin": "unittest_uart_begin()",
"end": "unittest_uart_end()"
}
}
class TestProcessorBase(object):
DEFAULT_BAUDRATE = 115200
def __init__(self, cmd_ctx, testname, envname, options):
self.cmd_ctx = cmd_ctx
self.cmd_ctx.meta['piotest_processor'] = True
self.test_name = testname
self.options = options
self.env_name = envname
self.env_options = {
k: v
for k, v in options['project_config'].items("env:" + envname)
}
self._run_failed = False
self._outputcpp_generated = False
def get_transport(self):
transport = self.env_options.get("framework")
if self.env_options.get("platform") == "native":
transport = "native"
if "test_transport" in self.env_options:
transport = self.env_options['test_transport']
if transport not in TRANSPORT_OPTIONS:
raise exception.PlatformioException(
"Unknown Unit Test transport `%s`" % transport)
return transport.lower()
def get_baudrate(self):
return int(self.env_options.get("test_speed", self.DEFAULT_BAUDRATE))
def print_progress(self, text, is_error=False):
click.echo()
print_header(
"[test/%s] %s" % (click.style(
self.test_name, fg="yellow", bold=True), text),
is_error=is_error)
def build_or_upload(self, target):
if not self._outputcpp_generated:
self.generate_outputcpp(util.get_projecttest_dir())
self._outputcpp_generated = True
if self.test_name != "*":
self.cmd_ctx.meta['piotest'] = self.test_name
if not self.options['verbose']:
click.echo("Please wait...")
return self.cmd_ctx.invoke(
cmd_run,
project_dir=self.options['project_dir'],
upload_port=self.options['upload_port'],
silent=not self.options['verbose'],
environment=[self.env_name],
disable_auto_clean="nobuild" in target,
target=target)
def process(self):
raise NotImplementedError
def run(self):
raise NotImplementedError
def on_run_out(self, line):
if line.endswith(":PASS"):
click.echo(
"%s\t[%s]" % (line[:-5], click.style("PASSED", fg="green")))
elif ":FAIL" in line:
self._run_failed = True
click.echo("%s\t[%s]" % (line, click.style("FAILED", fg="red")))
else:
click.echo(line)
def generate_outputcpp(self, test_dir):
assert isdir(test_dir)
cpp_tpl = "\n".join([
"$include",
"#include <output_export.h>",
"",
"$object",
"",
"void output_start(unsigned int baudrate)",
"{",
" $begin;",
"}",
"",
"void output_char(int c)",
"{",
" $putchar;",
"}",
"",
"void output_flush(void)",
"{",
" $flush;",
"}",
"",
"void output_complete(void)",
"{",
" $end;",
"}"
]) # yapf: disable
def delete_tmptest_file(file_):
try:
remove(file_)
except: # pylint: disable=bare-except
if isfile(file_):
click.secho(
"Warning: Could not remove temporary file '%s'. "
"Please remove it manually." % file_,
fg="yellow")
tpl = Template(cpp_tpl).substitute(
TRANSPORT_OPTIONS[self.get_transport()])
data = Template(tpl).substitute(baudrate=self.get_baudrate())
tmp_file = join(test_dir, "output_export.cpp")
with open(tmp_file, "w") as f:
f.write(data)
atexit.register(delete_tmptest_file, tmp_file)

View File

@ -292,3 +292,12 @@ class DebugSupportError(PlatformioException):
class DebugInvalidOptions(PlatformioException):
pass
class TestDirNotExists(PlatformioException):
MESSAGE = "A test folder '{0}' does not exist.\nPlease create 'test' "\
"directory in project's root and put a test set.\n"\
"More details about Unit "\
"Testing: http://docs.platformio.org/page/plus/"\
"unit-testing.html"