/* mbed Microcontroller Library * Copyright (c) 2017 ARM Limited * SPDX-License-Identifier: Apache-2.0 * * 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. */ #if defined(MBED_RTOS_SINGLE_THREAD) || !defined(MBED_CONF_RTOS_PRESENT) #error [NOT_SUPPORTED] Queue test cases require RTOS with multithread to run #else #if !DEVICE_USTICKER #error [NOT_SUPPORTED] UsTicker need to be enabled for this test. #else #include "mbed.h" #include "greentea-client/test_env.h" #include "unity.h" #include "utest.h" #include "rtos.h" using namespace utest::v1; using namespace std::chrono; #define TEST_ASSERT_DURATION_WITHIN(delta, expected, actual) \ do { \ using ct = std::common_type_t<decltype(delta), decltype(expected), decltype(actual)>; \ TEST_ASSERT_INT_WITHIN(ct(delta).count(), ct(expected).count(), ct(actual).count()); \ } while (0) #define THREAD_STACK_SIZE 512 #define TEST_TIMEOUT 50ms static uint32_t msg; static uint32_t msg2; void thread_put_uint_msg(Queue<uint32_t, 1> *q) { ThisThread::sleep_for(TEST_TIMEOUT); bool stat = q->try_put(&msg); TEST_ASSERT_TRUE(stat); } void thread_get_uint_msg(Queue<uint32_t, 1> *q) { ThisThread::sleep_for(TEST_TIMEOUT); uint32_t *v; bool stat = q->try_get_for(Kernel::wait_for_u32_forever, &v); TEST_ASSERT_TRUE(stat) TEST_ASSERT_EQUAL(&msg, v); } /** Test pass msg Given a queue for uint32_t messages with one slot When a uin32_t value is inserted into the queue and a message is extracted from the queue Then the extracted message is the same as previously inserted message */ void test_pass() { Queue<uint32_t, 1> q; bool stat = q.try_put(&msg); TEST_ASSERT_TRUE(stat) uint32_t *v; stat = q.try_get_for(Kernel::wait_for_u32_forever, &v); TEST_ASSERT_TRUE(stat) TEST_ASSERT_EQUAL(&msg, v); } /** Test pass msg twice Given a queue for uint32_t messages with one slot When a uin32_t value is inserted into the queue and a message is extracted from the queue and the procedure is repeated with different message Then the extracted message is the same as previously inserted message for both iterations */ void test_pass_twice() { Queue<uint32_t, 1> q; bool stat = q.try_put(&msg); TEST_ASSERT_TRUE(stat); uint32_t *v; stat = q.try_get_for(Kernel::wait_for_u32_forever, &v); TEST_ASSERT_TRUE(stat); TEST_ASSERT_EQUAL(&msg, v); stat = q.try_put(&msg2); TEST_ASSERT_TRUE(stat); stat = q.try_get_for(Kernel::wait_for_u32_forever, &v); TEST_ASSERT_TRUE(stat); TEST_ASSERT_EQUAL(&msg2, v); } /** Test get from empty queue Given an empty queue for uint32_t values When @a get is called on the queue with timeout of 0 Then queue returns status of false */ void test_get_empty_no_timeout() { Queue<uint32_t, 1> q; uint32_t *v; bool stat = q.try_get(&v); TEST_ASSERT_FALSE(stat); } /** Test get from empty queue with timeout Given an empty queue for uint32_t values When @a get is called on the queue with timeout of 50ms Then queue returns status of osEventTimeout after about 50ms wait */ void test_get_empty_timeout() { Queue<uint32_t, 1> q; Timer timer; timer.start(); uint32_t *v; bool stat = q.try_get_for(50ms, &v); TEST_ASSERT_FALSE(stat); TEST_ASSERT_DURATION_WITHIN(5ms, 50ms, timer.elapsed_time()); } /** Test get empty wait forever Given a two threads A & B and a queue for uint32_t values When thread A calls @a get on an empty queue with osWaitForever Then the thread A waits for a message to appear in the queue When thread B puts a message in the queue Then thread A wakes up and receives it */ void test_get_empty_wait_forever() { Thread t(osPriorityNormal, THREAD_STACK_SIZE); Queue<uint32_t, 1> q; t.start(callback(thread_put_uint_msg, &q)); Timer timer; timer.start(); uint32_t *v; bool stat = q.try_get_for(Kernel::wait_for_u32_forever, &v); TEST_ASSERT_TRUE(stat); TEST_ASSERT_EQUAL(&msg, v); TEST_ASSERT_DURATION_WITHIN(TEST_TIMEOUT / 10, TEST_TIMEOUT, timer.elapsed_time()); } /** Test put full no timeout * * Given a queue with one slot for uint32_t data * When a thread tries to insert two messages * Then first operation succeeds and second fails */ void test_put_full_no_timeout() { Queue<uint32_t, 1> q; bool stat = q.try_put(&msg); TEST_ASSERT_TRUE(stat); stat = q.try_put(&msg); TEST_ASSERT_FALSE(stat); } /** Test put full timeout * * Given a queue with one slot for uint32_t data * When a thread tries to insert two messages with @ TEST_TIMEOUT timeout * Then first operation succeeds and second fails */ void test_put_full_timeout() { Queue<uint32_t, 1> q; bool stat = q.try_put_for(TEST_TIMEOUT, &msg); TEST_ASSERT_TRUE(stat); Timer timer; timer.start(); stat = q.try_put_for(TEST_TIMEOUT, &msg); TEST_ASSERT_FALSE(stat); TEST_ASSERT_DURATION_WITHIN(TEST_TIMEOUT / 10, TEST_TIMEOUT, timer.elapsed_time()); } /** Test put full wait forever * * Given two threads A & B and a queue with one slot for uint32_t data * When thread A puts a message to the queue and tries to put second one with @a Kernel::wait_for_u32_forever timeout * Then thread waits for a slot to become empty in the queue * When thread B takes one message out of the queue * Then thread A successfully inserts message into the queue */ void test_put_full_waitforever() { Thread t(osPriorityNormal, THREAD_STACK_SIZE); Queue<uint32_t, 1> q; t.start(callback(thread_get_uint_msg, &q)); bool stat = q.try_put(&msg); TEST_ASSERT_TRUE(stat); Timer timer; timer.start(); stat = q.try_put_for(Kernel::wait_for_u32_forever, &msg); TEST_ASSERT_TRUE(stat); TEST_ASSERT_DURATION_WITHIN(TEST_TIMEOUT / 10, TEST_TIMEOUT, timer.elapsed_time()); t.join(); } /** Test message ordering Given a queue of uint32_t data When two messages are inserted with equal priority Then messages should be returned in the exact order they were inserted */ void test_msg_order() { Queue<uint32_t, 2> q; bool stat = q.try_put_for(TEST_TIMEOUT, &msg); TEST_ASSERT_TRUE(stat); stat = q.try_put_for(TEST_TIMEOUT, &msg2); TEST_ASSERT_TRUE(stat); uint32_t *v; stat = q.try_get(&v); TEST_ASSERT_TRUE(stat); TEST_ASSERT_EQUAL(&msg, v); stat = q.try_get(&v); TEST_ASSERT_TRUE(stat); TEST_ASSERT_EQUAL(&msg2, v); } /** Test message priority Given a queue of uint32_t data When two messages are inserted with ascending priority Then messages should be returned in descending priority order */ void test_msg_prio() { Queue<uint32_t, 2> q; bool stat = q.try_put_for(TEST_TIMEOUT, &msg, 0); TEST_ASSERT_TRUE(stat); stat = q.try_put_for(TEST_TIMEOUT, &msg2, 1); TEST_ASSERT_TRUE(stat); uint32_t *v; stat = q.try_get(&v); TEST_ASSERT_TRUE(stat); TEST_ASSERT_EQUAL(&msg2, v); stat = q.try_get(&v); TEST_ASSERT_TRUE(stat); TEST_ASSERT_EQUAL(&msg, v); } /** Test queue empty Given a queue of uint32_t data before data is inserted the queue should be empty after data is inserted the queue shouldn't be empty */ void test_queue_empty() { Queue<uint32_t, 1> q; TEST_ASSERT_TRUE(q.empty()); q.try_put_for(TEST_TIMEOUT, &msg, 1); TEST_ASSERT_FALSE(q.empty()); } /** Test queue empty Given a queue of uint32_t data with size of 1 before data is inserted the queue shouldn't be full after data is inserted the queue should be full */ void test_queue_full() { Queue<uint32_t, 1> q; TEST_ASSERT_FALSE(q.full()); q.try_put_for(TEST_TIMEOUT, &msg, 1); TEST_ASSERT_TRUE(q.full()); } utest::v1::status_t test_setup(const size_t number_of_cases) { GREENTEA_SETUP(5, "default_auto"); #ifdef DEVICE_SEMIHOST // Disconnect semihosting now, because otherwise it will get disconnected on the first sleep call and // cause said call to take several milliseconds, leading to a test failure. mbed_interface_disconnect(); #endif return verbose_test_setup_handler(number_of_cases); } Case cases[] = { Case("Test pass msg", test_pass), Case("Test pass msg twice", test_pass_twice), Case("Test get from empty queue no timeout", test_get_empty_no_timeout), Case("Test get from empty queue timeout", test_get_empty_timeout), Case("Test get empty wait forever", test_get_empty_wait_forever), Case("Test put full no timeout", test_put_full_no_timeout), Case("Test put full timeout", test_put_full_timeout), Case("Test put full wait forever", test_put_full_waitforever), Case("Test message ordering", test_msg_order), Case("Test message priority", test_msg_prio), Case("Test queue empty", test_queue_empty), Case("Test queue full", test_queue_full) }; Specification specification(test_setup, cases); int main() { return !Harness::run(specification); } #endif // !DEVICE_USTICKER #endif // defined(MBED_RTOS_SINGLE_THREAD) || !defined(MBED_CONF_RTOS_PRESENT)