Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[devel_transport] Async host test execition support #78

Merged
merged 30 commits into from
Mar 2, 2016
Merged

[devel_transport] Async host test execition support #78

Changes from 1 commit
Commits
Show all changes
30 commits
Select commit Hold shift + click to select a range
ce92ac2
Add KiVi transport protocol support - draft
PrzemekWirkus Feb 5, 2016
63b5bcd
Polishing
PrzemekWirkus Feb 5, 2016
0b1cc56
Removed parallel tests
PrzemekWirkus Feb 5, 2016
d24242c
Merge Test Case reporting functionality - draft
PrzemekWirkus Feb 5, 2016
fca7efc
Remove copy_method from test case report
PrzemekWirkus Feb 5, 2016
71149d3
Ported Test Case reporting functionality
PrzemekWirkus Feb 5, 2016
9de37ed
Polishing
PrzemekWirkus Feb 5, 2016
10d5007
Update unit tests
PrzemekWirkus Feb 8, 2016
470c733
Add solution for extra new-line when dumping app output to file
PrzemekWirkus Feb 9, 2016
dcfd047
Test case report fixes
PrzemekWirkus Feb 10, 2016
7210f00
Force pySerial v3 usage
PrzemekWirkus Feb 11, 2016
c1219b2
v0.2.0
PrzemekWirkus Feb 11, 2016
151734f
Add test utest case summary support (+checks)
PrzemekWirkus Feb 11, 2016
0dd25c7
Add verbose flag to htrun on-screen logging
PrzemekWirkus Feb 11, 2016
836e330
Add alias for '-S' for CLI switch '--skip-build'
PrzemekWirkus Feb 11, 2016
dad9960
Add alias '-l' for CLI switch '--list'
PrzemekWirkus Feb 11, 2016
43abb3c
Add test case reporting to junit report
PrzemekWirkus Feb 15, 2016
c12a3ff
Add utest log parsing for JUnit results
PrzemekWirkus Feb 15, 2016
7d231c4
unit test polishing
PrzemekWirkus Feb 16, 2016
7280188
Remove extra line in Junit stdout log cut
PrzemekWirkus Feb 17, 2016
cd2bd4a
Typo
PrzemekWirkus Feb 22, 2016
65b7236
Add Test Suite -> Test Case mapping in case of no utest present
PrzemekWirkus Feb 23, 2016
3cda486
Refactore test suite -> test case mapping feature to be more generic
PrzemekWirkus Feb 23, 2016
9625991
Add unicode strip from JUnit report text
PrzemekWirkus Feb 23, 2016
e9efa7b
Add default test case report values for __testcase_finish
PrzemekWirkus Feb 23, 2016
6cdc12b
Test suite -> test case feature polishing
PrzemekWirkus Feb 26, 2016
c7c5f54
Add sorted results for test cases report
PrzemekWirkus Feb 26, 2016
8cf5dd0
Add Hooks & GCOV support
PrzemekWirkus Mar 1, 2016
ce8f693
Add htrun v0.2.0 dependency
PrzemekWirkus Mar 2, 2016
91492a0
Merge master
PrzemekWirkus Mar 2, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
281 changes: 54 additions & 227 deletions mbed_greentea/mbed_test_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,8 @@

import re
import sys
from time import time, sleep
from Queue import Queue, Empty
from threading import Thread
from subprocess import call, Popen, PIPE
from time import time
from subprocess import call, Popen, PIPE, STDOUT

from mbed_greentea.mbed_greentea_log import gt_logger

Expand Down Expand Up @@ -66,20 +64,18 @@
"build_failed" : TEST_RESULT_BUILD_FAILED
}

RE_DETECT_TESTCASE_RESULT = re.compile("\\{(" + "|".join(TEST_RESULT_MAPPING.keys()) + ")\\}")

def get_test_result(output):
"""! Parse test 'output' data
@details If test result not found returns by default TEST_RESULT_TIMEOUT value
@return Returns found test result
"""
result = TEST_RESULT_TIMEOUT
RE_DETECT_TESTCASE_RESULT = re.compile("\\{(" + "|".join(TEST_RESULT_MAPPING.keys()) + ")\\}")
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Still valid to use capital letter for this local variable?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will change it to lower case !

for line in "".join(output).splitlines():
search_result = RE_DETECT_TESTCASE_RESULT.search(line)
if search_result and search_result.groups():
result = TEST_RESULT_MAPPING[search_result.groups(0)[0]]
break
return result
return TEST_RESULT_MAPPING[search_result.groups(0)[0]]
return TEST_RESULT_TIMEOUT


def run_host_test(image_path,
Expand Down Expand Up @@ -118,119 +114,15 @@ def run_host_test(image_path,
file name was given as switch option)
"""

class StdInObserver(Thread):
""" Process used to read stdin only as console input from MUT
"""
def __init__(self):
Thread.__init__(self)
self.queue = Queue()
self.daemon = True
self.active = True
self.start()

def run(self):
while self.active:
c = sys.stdin.read(1)
self.queue.put(c)

def stop(self):
self.active = False

class FileObserver(Thread):
""" process used to read file content as console input from MUT
"""
def __init__(self, filename):
Thread.__init__(self)
self.filename = filename
self.queue = Queue()
self.daemon = True
self.active = True
self.start()

def run(self):
with open(self.filename) as f:
while self.active:
c = f.read(1)
self.queue.put(c)

def stop(self):
self.active = False

class ProcessObserver(Thread):
""" Default process used to observe stdout of another process as console input from MUT
"""
def __init__(self, proc):
Thread.__init__(self)
self.proc = proc
self.queue = Queue()
self.daemon = True
self.active = True
self.start()

def run(self):
while self.active:
c = self.proc.stdout.read(1)
self.queue.put(c)

def stop(self):
self.active = False

# Try stopping mbed-host-test
try:
self.proc.stdin.close()
finally:
pass

# Give 5 sec for mbedhtrun to exit
ret_code = None
for i in range(5):
ret_code = self.proc.poll()
# A None value indicates that the process hasn't terminated yet.
if ret_code is not None:
break
sleep(1)

if ret_code is None: # Kill it
print 'Terminating mbed-host-test(mbedhtrun) process (PID %s)' % self.proc.pid
try:
self.proc.terminate()
except Exception as e:
print "ProcessObserver.stop(): %s" % str(e)

def get_char_from_queue(obs):
""" Get character from queue safe way
"""
def run_command(cmd):
"""! Runs command and prints proc stdout on screen """
try:
c = obs.queue.get(block=True, timeout=0.5)
# signals to queue job is done
obs.queue.task_done()
except Empty:
c = None
except:
raise
return c

def filter_queue_char(c):
""" Filters out non ASCII characters from serial port
"""
if ord(c) not in range(128):
c = ' '
return c

def get_auto_property_value(property_name, line):
"""! Scans auto detection line from MUT and returns scanned parameter 'property_name'
@details Host test case has to print additional properties for test to be set up
@return Returns string or None if property search failed
"""
result = None
if re.search("HOST: Property '%s'"% property_name, line) is not None:
property = re.search("HOST: Property '%s' = '([\w\d _]+)'"% property_name, line)
if property is not None and len(property.groups()) == 1:
result = property.groups()[0]
return result

# Detect from where input should be taken, if no --digest switch is specified
# normal test execution can be performed
p = Popen(cmd,
stdout=PIPE,
stderr=STDOUT)
except OSError as e:
print "mbedgt: run_command(%s) ret= %d failed: "% (cmd, int(ret)), str(e), e.child_traceback
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what's the diff between using print and gt_logger in this module?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When in exception we would like to use minimal IO to error message. Just to be clear, gt_logger is the way to go (log workflow infos, wrns, erros etc.) but in exception handler hard to say if gt_logger can be used.

Just to print less noise during exception handle trace IMHO.
PS: Note, that if run_command throws we are done anyways :P

return iter(p.stdout.readline, b'')

if verbose:
gt_logger.gt_log("selecting test case observer...")
Expand All @@ -240,124 +132,57 @@ def get_auto_property_value(property_name, line):
# Select who will digest test case serial port data
if digest_source == 'stdin':
# When we want to scan stdin for test results
obs = StdInObserver()
raise NotImplementedError
elif digest_source is not None:
# When we want to open file to scan for test results
obs = FileObserver(digest_source)
else:
# Command executing CLI for host test supervisor (in detect-mode)
cmd = ["mbedhtrun",
'-d', disk,
'-p', port,
'-f', '"%s"'% image_path,
]
raise NotImplementedError

# Command executing CLI for host test supervisor (in detect-mode)
cmd = ["mbedhtrun",
'-d', disk,
'-p', port,
'-f', '"%s"'% image_path,
]

# Add extra parameters to host_test
if program_cycle_s is not None:
cmd += ["-C", str(program_cycle_s)]
if copy_method is not None:
cmd += ["-c", copy_method]
if micro is not None:
cmd += ["-m", micro]
if reset is not None:
cmd += ["-r", reset]
if reset_tout is not None:
cmd += ["-R", str(reset_tout)]
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is reset_tout = reset_timeout ? I would guess as it would be integer thus casting to str(). Then above program_cycle_s is string or _s as seconds?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • reset_tout is a timeout related to reset.
  • program_cycle_s is parameter inherited from mbed 2.0, _s stands for seconds.
    Both were kept for historical reason (people knowing mbed 2.0 test suite would know this value).

if json_test_cfg is not None:
cmd += ["--test-cfg", '"%s"' % str(json_test_cfg)]
if run_app is not None:
cmd += ["--run"] # -f stores binary name!
if enum_host_tests_path:
cmd += ["-e", '"%s"'% enum_host_tests_path]

# Add extra parameters to host_test
if program_cycle_s is not None:
cmd += ["-C", str(program_cycle_s)]
if copy_method is not None:
cmd += ["-c", copy_method]
if micro is not None:
cmd += ["-m", micro]
if reset is not None:
cmd += ["-r", reset]
if reset_tout is not None:
cmd += ["-R", str(reset_tout)]
if json_test_cfg is not None:
cmd += ["--test-cfg", '"%s"' % str(json_test_cfg)]
if run_app is not None:
cmd += ["--run"] # -f stores binary name!
if enum_host_tests_path:
cmd += ["-e", '"%s"'% enum_host_tests_path]

if verbose:
gt_logger.gt_log_tab("calling mbedhtrun: %s"% " ".join(cmd))
gt_logger.gt_log("mbed-host-test-runner: started")
proc = Popen(cmd, stdin=PIPE, stdout=PIPE)
obs = ProcessObserver(proc)
if verbose:
gt_logger.gt_log_tab("calling mbedhtrun: %s"% " ".join(cmd))
gt_logger.gt_log("mbed-host-test-runner: started")

result = None
update_once_flag = {} # Stores flags checking if some auto-parameter was already set
unknown_property_count = 0
total_duration = 20 # This for flashing, reset and other serial port operations
line = ''
output = []
htrun_output = ''
start_time = time()
while (time() - start_time) < (total_duration):
try:
c = get_char_from_queue(obs)
except Exception as e:
output.append('get_char_from_queue(obs): %s'% str(e))
break
if c:
if verbose:
sys.stdout.write(c)
c = filter_queue_char(c)
output.append(c)
# Give the mbed under test a way to communicate the end of the test
if c in ['\n', '\r']:

# Check for unknown property prints
# If there are too many we will stop test execution and assume test is not ported
if "HOST: Unknown property" in line:
unknown_property_count += 1
if unknown_property_count >= max_failed_properties:
output.append('{{error}}')
break

# Checking for auto-detection information from the test about MUT reset moment
if 'reset_target' not in update_once_flag and "HOST: Reset target..." in line:
# We will update this marker only once to prevent multiple time resets
update_once_flag['reset_target'] = True
start_time = time()
total_duration = duration # After reset we gonna use real test case duration

# Checking for auto-detection information from the test about timeout
auto_timeout_val = get_auto_property_value('timeout', line)
if 'timeout' not in update_once_flag and auto_timeout_val is not None:
# We will update this marker only once to prevent multiple time resets
update_once_flag['timeout'] = True
total_duration = int(auto_timeout_val)

# Detect mbed assert:
if 'mbed assertation failed: ' in line:
output.append('{{mbed_assert}}')
break

# Check for test end. Only a '{{end}}' in the start of line indicates a test end.
# A sub string '{{end}}' may also appear in an error message.
if re.search('^\{\{end\}\}', line, re.I):
break
line = ''
else:
line += c
else:
result = TEST_RESULT_TIMEOUT

if not '{end}' in line:
output.append('{{end}}')

c = get_char_from_queue(obs)

if c:
if verbose:
sys.stdout.write(c)
c = filter_queue_char(c)
output.append(c)

# Stop test process
obs.stop()
for line in run_command(cmd):
htrun_output += line
sys.stdout.write(line)
sys.stdout.flush()

end_time = time()
testcase_duration = end_time - start_time # Test case duration from reset to {end}

result = get_test_result(htrun_output)

if verbose:
gt_logger.gt_log("mbed-host-test-runner: stopped")
if not result:
result = get_test_result(output)
if verbose:
gt_logger.gt_log("mbed-host-test-runner: returned '%s'"% result)
return (result, "".join(output), testcase_duration, duration)
return (result, "".join(htrun_output), testcase_duration, duration)

def run_cli_command(cmd, shell=True, verbose=False):
"""! Runs command from command line
Expand All @@ -373,11 +198,13 @@ def run_cli_command(cmd, shell=True, verbose=False):
result = False
if verbose:
print "mbedgt: [ret=%d] Command: %s"% (int(ret), cmd)
except Exception as e:
except OSError as e:
result = False
if verbose:
print "mbedgt: [ret=%d] Command: %s"% (int(ret), cmd)
print str(e)
print "mbedgt: traceback..."
print e.child_traceback
return (result, ret)

def run_cli_process(cmd):
Expand Down