ARMmbed
/
mbed-os
mirror of https://github.com/ARMmbed/mbed-os.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Tests: USB: Add Serial test for line coding change

pull/9768/head
Filip Jagodzinski 2018-09-05 11:06:03 +02:00 committed by Russ Butler
parent bf3aa29e0b
commit 7b80cb6586
2 changed files with 196 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
TESTS/host_tests/usb_device_serial.py
View File

@ -29,9 +29,11 @@ MSG_KEY_PORT_OPEN_CLOSE = 'port_open_close'
MSG_KEY_SEND_BYTES_SINGLE = 'send_single'
MSG_KEY_SEND_BYTES_MULTIPLE = 'send_multiple'
MSG_KEY_LOOPBACK = 'loopback'
MSG_KEY_CHANGE_LINE_CODING = 'change_lc'
RX_BUFF_SIZE = 32
TERM_REOPEN_DELAY = 0.1
LINE_CODING_STRLEN = 13
def usb_serial_name(serial_number):
@ -70,6 +72,22 @@ def retry_fun_call(fun, num_retries=3, retry_delay=0.0):
class USBSerialTest(mbed_host_tests.BaseHostTest):
"""Host side test for USB CDC & Serial classes."""
_BYTESIZES = {
5: serial.FIVEBITS,
6: serial.SIXBITS,
7: serial.SEVENBITS,
8: serial.EIGHTBITS}
_PARITIES = {
0: serial.PARITY_NONE,
1: serial.PARITY_ODD,
2: serial.PARITY_EVEN,
3: serial.PARITY_MARK,
4: serial.PARITY_SPACE}
_STOPBITS = {
0: serial.STOPBITS_ONE,
1: serial.STOPBITS_ONE_POINT_FIVE,
2: serial.STOPBITS_TWO}
@staticmethod
def get_usb_serial_name(usb_id_str):
"""Get USB serial device name as registered in the system.
@ -196,12 +214,50 @@ class USBSerialTest(mbed_host_tests.BaseHostTest):
time.sleep(0.001)
mbed_serial.close()
def change_line_coding(self, usb_id_str):
"""Open the serial and change serial params according to device request.
New line coding params are read from the device serial data.
"""
mbed_serial = serial.Serial(timeout=0.5)
try:
mbed_serial.port = retry_fun_call(
fun=functools.partial(self.get_usb_serial_name, usb_id_str), # pylint: disable=not-callable
num_retries=20,
retry_delay=0.05)
retry_fun_call(
fun=mbed_serial.open,
num_retries=10,
retry_delay=0.05)
except RetryError as exc:
self.log('TEST ERROR: {}'.format(exc))
self.notify_complete(False)
return
mbed_serial.reset_output_buffer()
try:
payload = mbed_serial.read(LINE_CODING_STRLEN)
while len(payload) == LINE_CODING_STRLEN:
baud, bits, parity, stop = (int(i) for i in payload.split(','))
new_line_coding = {
'baudrate': baud,
'bytesize': self._BYTESIZES[bits],
'parity': self._PARITIES[parity],
'stopbits': self._STOPBITS[stop]}
mbed_serial.apply_settings(new_line_coding)
payload = mbed_serial.read(LINE_CODING_STRLEN)
except serial.SerialException as exc:
self.log('TEST ERROR: {}'.format(exc))
self.notify_complete(False)
return
mbed_serial.close()
def setup(self):
self.register_callback(MSG_KEY_PORT_OPEN_WAIT, self.cb_port_open_wait)
self.register_callback(MSG_KEY_PORT_OPEN_CLOSE, self.cb_port_open_close)
self.register_callback(MSG_KEY_SEND_BYTES_SINGLE, self.cb_send_bytes_single)
self.register_callback(MSG_KEY_SEND_BYTES_MULTIPLE, self.cb_send_bytes_multiple)
self.register_callback(MSG_KEY_LOOPBACK, self.cb_loopback)
self.register_callback(MSG_KEY_CHANGE_LINE_CODING, self.cb_change_line_coding)
def start_bg_task(self, **thread_kwargs):
"""Start a new daemon thread.
@ -248,3 +304,9 @@ class USBSerialTest(mbed_host_tests.BaseHostTest):
self.start_bg_task(
target=self.loopback,
args=(value, ))
def cb_change_line_coding(self, key, value, timestamp):
"""Open the serial and change the line coding."""
self.start_bg_task(
target=self.change_line_coding,
args=(value, ))

136
TESTS/usb_device/serial/main.cpp
View File

@ -37,6 +37,7 @@
#define MSG_KEY_SEND_BYTES_SINGLE "send_single"
#define MSG_KEY_SEND_BYTES_MULTIPLE "send_multiple"
#define MSG_KEY_LOOPBACK "loopback"
#define MSG_KEY_CHANGE_LINE_CODING "change_lc"
#define TX_BUFF_SIZE 32
#define RX_BUFF_SIZE 32
@ -48,10 +49,70 @@
// to handle the reconnect operation correctly.
#define USB_DISCONNECT_DELAY_MS 1
#define LINE_CODING_STRLEN 13 // 6 + 2 + 1 + 1 + 3 * comma
using utest::v1::Case;
using utest::v1::Specification;
using utest::v1::Harness;
typedef struct LineCoding {
// bits per second
int baud;
// 5, 6, 7, 8 or 16
int bits;
// 0 -- None,
// 1 -- Odd,
// 2 -- Even,
// 3 -- Mark,
// 4 -- Space
int parity;
// 0 -- 1 Stop bit,
// 1 -- 1.5 Stop bits,
// 2 -- 2 Stop bits
int stop;
int get_num_diffs(LineCoding const &other) const
{
int diffs = 0;
if (baud != other.baud) {
diffs++;
}
if (bits != other.bits) {
diffs++;
}
if (parity != other.parity) {
diffs++;
}
if (stop != other.stop) {
diffs++;
}
return diffs;
}
} line_coding_t;
line_coding_t default_lc = { 9600, 8, 0, 0 };
// There is no POSIX support for 1.5 stop bits.
// Do not set stop bits to 1.5 to keep tests compatible with all supported
// host systems.
line_coding_t test_codings[] = {
{ 9600, 5, 0, 2 },
{ 4800, 7, 2, 0 },
{ 19200, 8, 0, 2 },
{ 115200, 8, 0, 0 },
{ 38400, 8, 1, 0 },
{ 1200, 8, 0, 0 },
{ 19200, 8, 0, 0 },
{ 2400, 7, 2, 0 },
{ 9600, 8, 0, 0 },
{ 57600, 8, 0, 0 },
};
Mail<line_coding_t, 8> lc_mail;
#define EF_SEND (1ul << 0)
EventFlags event_flags;
@ -463,7 +524,7 @@ void test_serial_getc()
/** Test Serial printf & scanf
*
* Given the USB Serial device connected to a host
* When the device trensmits a formatted string with a random value
* When the device transmits a formatted string with a random value
* using the printf method
* and the host sends it back to the device
* Then the device can successfully read the value using scanf method
@ -497,9 +558,79 @@ void test_serial_printf_scanf()
usb_serial.disconnect();
}
void line_coding_changed_cb(int baud, int bits, int parity, int stop)
{
line_coding_t *lc = lc_mail.alloc();
lc->baud = baud;
lc->bits = bits;
lc->parity = parity;
lc->stop = stop;
lc_mail.put(lc);
}
/** Test Serial / CDC line coding change
*
* Given the device transmits a set of line coding params to host
* When the host updates serial port settings
* Then line_coding_changed() callback is called
* and the line coding is set as expected
*/
void test_serial_line_coding_change()
{
USBSerial usb_serial(false, USB_SERIAL_VID, USB_SERIAL_PID);
char usb_serial_sn[USB_SN_MAX_LEN] = { };
usb_desc2str(usb_serial.string_iserial_desc(), usb_serial_sn, USB_SN_MAX_LEN);
usb_serial.connect();
greentea_send_kv(MSG_KEY_CHANGE_LINE_CODING, usb_serial_sn);
while (!usb_serial.connected()) {
wait_ms(1);
}
usb_serial.attach(line_coding_changed_cb);
size_t num_line_codings = sizeof test_codings / sizeof test_codings[0];
line_coding_t *lc_prev = &default_lc;
line_coding_t *lc_expected = NULL;
line_coding_t *lc_actual = NULL;
int num_expected_callbacks, rc;
for (size_t i = 0; i < num_line_codings; i++) {
lc_expected = &(test_codings[i]);
num_expected_callbacks = lc_prev->get_num_diffs(*lc_expected);
rc = usb_serial.printf("%06i,%02i,%01i,%01i", lc_expected->baud, lc_expected->bits, lc_expected->parity,
lc_expected->stop);
TEST_ASSERT_EQUAL_INT(LINE_CODING_STRLEN, rc);
// The pyserial Python module does not update all line coding params
// at once. It updates params one by one instead, and since every
// update is followed by port reconfiguration we get multiple
// calls to line_coding_changed callback on the device.
while (num_expected_callbacks > 0) {
num_expected_callbacks--;
osEvent event = lc_mail.get();
TEST_ASSERT_EQUAL_UINT32(osEventMail, event.status);
lc_actual = (line_coding_t *) event.value.p;
if (lc_expected->get_num_diffs(*lc_actual) == 0) {
break;
} else if (num_expected_callbacks > 0) {
// Discard lc_actual only if there is still a chance to get new
// set of params.
lc_mail.free(lc_actual);
}
}
TEST_ASSERT_EQUAL_INT(lc_expected->baud, lc_actual->baud);
TEST_ASSERT_EQUAL_INT(lc_expected->bits, lc_actual->bits);
TEST_ASSERT_EQUAL_INT(lc_expected->parity, lc_actual->parity);
TEST_ASSERT_EQUAL_INT(lc_expected->stop, lc_actual->stop);
lc_mail.free(lc_actual);
lc_prev = lc_expected;
}
// Wait for the host to close its port.
while (usb_serial.ready()) {
wait_ms(1);
}
usb_serial.disconnect();
}
utest::v1::status_t testsuite_setup(const size_t number_of_cases)
{
GREENTEA_SETUP(25, "usb_device_serial");
GREENTEA_SETUP(35, "usb_device_serial");
srand((unsigned) ticker_read_us(get_us_ticker_data()));
return utest::v1::greentea_test_setup_handler(number_of_cases);
}
@ -515,6 +646,7 @@ Case cases[] = {
Case("Serial terminal reopen", test_serial_term_reopen),
Case("Serial getc", test_serial_getc),
Case("Serial printf/scanf", test_serial_printf_scanf),
Case("Serial line coding change", test_serial_line_coding_change),
};
Specification specification(testsuite_setup, cases);