From 948a977fa543a7f70eebaabaa5f836323f0906cf Mon Sep 17 00:00:00 2001 From: Ivan Kravets Date: Fri, 19 Apr 2019 20:33:31 +0300 Subject: [PATCH] Initial commit of PIO Unit Testing --- platformio/commands/test.py | 67 --------- platformio/commands/test/__init__.py | 15 ++ platformio/commands/test/command.py | 181 +++++++++++++++++++++++ platformio/commands/test/embedded.py | 133 +++++++++++++++++ platformio/commands/test/native.py | 39 +++++ platformio/commands/test/processor.py | 200 ++++++++++++++++++++++++++ platformio/exception.py | 9 ++ 7 files changed, 577 insertions(+), 67 deletions(-) delete mode 100644 platformio/commands/test.py create mode 100644 platformio/commands/test/__init__.py create mode 100644 platformio/commands/test/command.py create mode 100644 platformio/commands/test/embedded.py create mode 100644 platformio/commands/test/native.py create mode 100644 platformio/commands/test/processor.py diff --git a/platformio/commands/test.py b/platformio/commands/test.py deleted file mode 100644 index 4c6414c9..00000000 --- a/platformio/commands/test.py +++ /dev/null @@ -1,67 +0,0 @@ -# Copyright (c) 2014-present PlatformIO -# -# 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="") -@click.option( - "--filter", - "-f", - multiple=True, - metavar="", - help="Filter tests by a pattern") -@click.option( - "--ignore", - "-i", - multiple=True, - metavar="", - 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:]) diff --git a/platformio/commands/test/__init__.py b/platformio/commands/test/__init__.py new file mode 100644 index 00000000..6d4c3e2b --- /dev/null +++ b/platformio/commands/test/__init__.py @@ -0,0 +1,15 @@ +# Copyright (c) 2014-present PlatformIO +# +# 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 diff --git a/platformio/commands/test/command.py b/platformio/commands/test/command.py new file mode 100644 index 00000000..49da7bba --- /dev/null +++ b/platformio/commands/test/command.py @@ -0,0 +1,181 @@ +# Copyright (c) 2014-present PlatformIO +# +# 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="") +@click.option( + "--filter", + "-f", + multiple=True, + metavar="", + help="Filter tests by a pattern") +@click.option( + "--ignore", + "-i", + multiple=True, + metavar="", + 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 diff --git a/platformio/commands/test/embedded.py b/platformio/commands/test/embedded.py new file mode 100644 index 00000000..39e8555d --- /dev/null +++ b/platformio/commands/test/embedded.py @@ -0,0 +1,133 @@ +# Copyright (c) 2014-present PlatformIO +# +# 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 diff --git a/platformio/commands/test/native.py b/platformio/commands/test/native.py new file mode 100644 index 00000000..0945b10f --- /dev/null +++ b/platformio/commands/test/native.py @@ -0,0 +1,39 @@ +# Copyright (c) 2014-present PlatformIO +# +# 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 diff --git a/platformio/commands/test/processor.py b/platformio/commands/test/processor.py new file mode 100644 index 00000000..7c3b84eb --- /dev/null +++ b/platformio/commands/test/processor.py @@ -0,0 +1,200 @@ +# Copyright (c) 2014-present PlatformIO +# +# 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 ", + "object": "", + "putchar": "Serial.write(c)", + "flush": "Serial.flush()", + "begin": "Serial.begin($baudrate)", + "end": "Serial.end()" + }, + "mbed": { + "include": "#include ", + "object": "Serial pc(USBTX, USBRX);", + "putchar": "pc.putc(c)", + "flush": "", + "begin": "pc.baud($baudrate)", + "end": "" + }, + "energia": { + "include": "#include ", + "object": "", + "putchar": "Serial.write(c)", + "flush": "Serial.flush()", + "begin": "Serial.begin($baudrate)", + "end": "Serial.end()" + }, + "espidf": { + "include": "#include ", + "object": "", + "putchar": "putchar(c)", + "flush": "fflush(stdout)", + "begin": "", + "end": "" + }, + "native": { + "include": "#include ", + "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 ", + "", + "$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) diff --git a/platformio/exception.py b/platformio/exception.py index 7829b648..77837d7a 100644 --- a/platformio/exception.py +++ b/platformio/exception.py @@ -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"