[asterisk-commits] kmoore: branch kmoore/stasis-device_state r383392 - /team/kmoore/stasis-devic...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Tue Mar 19 14:35:54 CDT 2013
Author: kmoore
Date: Tue Mar 19 14:35:51 2013
New Revision: 383392
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=383392
Log:
Add unit tests for device state system
This adds tests for device state message publishing, caching, and
aggregation.
Modified:
team/kmoore/stasis-device_state/tests/test_devicestate.c
Modified: team/kmoore/stasis-device_state/tests/test_devicestate.c
URL: http://svnview.digium.com/svn/asterisk/team/kmoore/stasis-device_state/tests/test_devicestate.c?view=diff&rev=383392&r1=383391&r2=383392
==============================================================================
--- team/kmoore/stasis-device_state/tests/test_devicestate.c (original)
+++ team/kmoore/stasis-device_state/tests/test_devicestate.c Tue Mar 19 14:35:51 2013
@@ -39,7 +39,9 @@
#include "asterisk/test.h"
#include "asterisk/devicestate.h"
#include "asterisk/pbx.h"
-
+#include "asterisk/stasis_message_router.h"
+
+#define UNIT_TEST_DEVICE_IDENTIFIER "unit_test_device_identifier"
/* These arrays are the result of the 'core show device2extenstate' output. */
static int combined_results[] = {
@@ -274,14 +276,241 @@
return res;
}
+struct consumer {
+ ast_mutex_t lock;
+ ast_cond_t out;
+ int already_out;
+ enum ast_device_state state;
+ enum ast_device_state aggregate_state;
+ int sig_on_non_aggregate_state;
+};
+
+static void consumer_dtor(void *obj) {
+ struct consumer *consumer = obj;
+
+ ast_mutex_destroy(&consumer->lock);
+ ast_cond_destroy(&consumer->out);
+}
+
+static struct consumer *consumer_create(void) {
+ RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
+
+ consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
+
+ if (!consumer) {
+ return NULL;
+ }
+
+ ast_mutex_init(&consumer->lock);
+ ast_cond_init(&consumer->out, NULL);
+ consumer->sig_on_non_aggregate_state = 0;
+
+ ao2_ref(consumer, +1);
+ return consumer;
+}
+
+static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
+{
+ struct consumer *consumer = data;
+ RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
+ struct stasis_cache_update *cache_update = stasis_message_data(message);
+ struct stasis_device_state *device_state;
+ SCOPED_MUTEX(lock, &consumer->lock);
+
+ if (!cache_update->new_snapshot) {
+ return;
+ }
+
+ device_state = stasis_message_data(cache_update->new_snapshot);
+
+ if (strcmp(device_state->device, UNIT_TEST_DEVICE_IDENTIFIER)) {
+ /* not a device state we're interested in */
+ return;
+ }
+
+ if (device_state->eid) {
+ consumer->state = device_state->state;
+ if (consumer->sig_on_non_aggregate_state) {
+ consumer->sig_on_non_aggregate_state = 0;
+ consumer->already_out = 1;
+ ast_cond_signal(&consumer->out);
+ }
+ } else {
+ consumer->aggregate_state = device_state->state;
+ consumer->already_out = 1;
+ ast_cond_signal(&consumer->out);
+ }
+}
+
+static void consumer_finalize(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
+{
+ struct consumer *consumer = data;
+
+ if (stasis_subscription_final_message(sub, message)) {
+ ao2_cleanup(consumer);
+ }
+}
+
+static void consumer_wait_for(struct consumer *consumer)
+{
+ int res;
+ struct timeval start = ast_tvnow();
+ struct timespec end = {
+ .tv_sec = start.tv_sec + 10,
+ .tv_nsec = start.tv_usec * 1000
+ };
+
+ SCOPED_MUTEX(lock, &consumer->lock);
+
+ if (consumer->already_out) {
+ consumer->already_out = 0;
+ }
+
+ while(1) {
+ res = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
+ if (!res || res == ETIMEDOUT) {
+ break;
+ }
+ }
+ consumer->already_out = 0;
+}
+
+static int remove_device_states_cb(void *obj, void *arg, int flags)
+{
+ RAII_VAR(struct stasis_message *, msg, obj, ao2_cleanup);
+ struct stasis_device_state *device_state = stasis_message_data(msg);
+ if (strcmp(UNIT_TEST_DEVICE_IDENTIFIER, device_state->device)) {
+ msg = NULL;
+ return 0;
+ }
+
+ msg = stasis_cache_clear_create(stasis_device_state_message(), device_state->cache_id);
+ /* topic guaranteed to have been created by this point */
+ stasis_publish(stasis_device_state_topic(device_state->device), msg);
+ return 0;
+}
+
+static void cache_cleanup(int unused)
+{
+ RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup);
+ /* remove all device states created during this test */
+ cache_dump = stasis_cache_dump(stasis_device_state_topic_cached(), NULL);
+ if (!cache_dump) {
+ return;
+ }
+ ao2_callback(cache_dump, 0, remove_device_states_cb, NULL);
+}
+
+AST_TEST_DEFINE(device_state_aggregation_test)
+{
+ RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_router *, device_msg_router, NULL, stasis_message_router_unsubscribe);
+ RAII_VAR(struct ast_eid *, foreign_eid, NULL, ast_free);
+ RAII_VAR(int, cleanup_cache, 0, cache_cleanup);
+ int res;
+ struct stasis_device_state *device_state;
+ struct stasis_message *msg;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = "device_state_aggregation_test";
+ info->category = "/main/devicestate/";
+ info->summary = "Tests message routing and aggregation through the Stasis device state system.";
+ info->description =
+ "Verifies that the device state system passes "
+ "messages appropriately, that the aggregator is "
+ "working properly, that the aggregate results match "
+ "the expected combined devstate, and that the cached "
+ "aggregate devstate is correct.";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ foreign_eid = ast_malloc(sizeof(*foreign_eid));
+ ast_test_validate(test, NULL != foreign_eid);
+ memset(foreign_eid, 0xFF, sizeof(*foreign_eid));
+
+ consumer = consumer_create();
+ ast_test_validate(test, NULL != consumer);
+
+ device_msg_router = stasis_message_router_create(stasis_caching_get_topic(stasis_device_state_topic_cached()));
+ ast_test_validate(test, NULL != device_msg_router);
+
+ ao2_ref(consumer, +1);
+ res = stasis_message_router_add(device_msg_router, stasis_cache_update(), consumer_exec, consumer);
+ ast_test_validate(test, !res);
+
+ res = stasis_message_router_add(device_msg_router, stasis_subscription_change(), consumer_finalize, consumer);
+ ast_test_validate(test, !res);
+
+ /* push local state */
+ stasis_publish_device_state(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE);
+
+ consumer_wait_for(consumer);
+ ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->state);
+ ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->aggregate_state);
+
+ msg = stasis_cache_get(stasis_device_state_topic_cached(), stasis_device_state_message(), UNIT_TEST_DEVICE_IDENTIFIER);
+ device_state = stasis_message_data(msg);
+ ast_test_validate(test, AST_DEVICE_NOT_INUSE == device_state->state);
+ ao2_cleanup(msg);
+ msg = NULL;
+
+ /* push remote state */
+ /* this will not produce a new aggregate state message since the aggregate state does not change */
+ consumer->sig_on_non_aggregate_state = 1;
+ stasis_publish_device_state_full(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE, foreign_eid);
+
+ consumer_wait_for(consumer);
+ ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->state);
+ ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->aggregate_state);
+
+ msg = stasis_cache_get(stasis_device_state_topic_cached(), stasis_device_state_message(), UNIT_TEST_DEVICE_IDENTIFIER);
+ device_state = stasis_message_data(msg);
+ ast_test_validate(test, AST_DEVICE_NOT_INUSE == device_state->state);
+ ao2_cleanup(msg);
+ msg = NULL;
+
+ /* push remote state different from local state */
+ stasis_publish_device_state_full(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_INUSE, AST_DEVSTATE_CACHABLE, foreign_eid);
+
+ consumer_wait_for(consumer);
+ ast_test_validate(test, AST_DEVICE_INUSE == consumer->state);
+ ast_test_validate(test, AST_DEVICE_INUSE == consumer->aggregate_state);
+
+ msg = stasis_cache_get(stasis_device_state_topic_cached(), stasis_device_state_message(), UNIT_TEST_DEVICE_IDENTIFIER);
+ device_state = stasis_message_data(msg);
+ ast_test_validate(test, AST_DEVICE_INUSE == device_state->state);
+ ao2_cleanup(msg);
+ msg = NULL;
+
+ /* push local state that will cause aggregated state different from local non-aggregate state */
+ stasis_publish_device_state(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_RINGING, AST_DEVSTATE_CACHABLE);
+
+ consumer_wait_for(consumer);
+ ast_test_validate(test, AST_DEVICE_RINGING == consumer->state);
+ ast_test_validate(test, AST_DEVICE_RINGINUSE == consumer->aggregate_state);
+
+ msg = stasis_cache_get(stasis_device_state_topic_cached(), stasis_device_state_message(), UNIT_TEST_DEVICE_IDENTIFIER);
+ device_state = stasis_message_data(msg);
+ ast_test_validate(test, AST_DEVICE_RINGINUSE == device_state->state);
+ ao2_cleanup(msg);
+ msg = NULL;
+
+ return AST_TEST_PASS;
+}
+
static int unload_module(void)
{
AST_TEST_UNREGISTER(device2extenstate_test);
+ AST_TEST_UNREGISTER(device_state_aggregation_test);
return 0;
}
static int load_module(void)
{
+ AST_TEST_REGISTER(device_state_aggregation_test);
AST_TEST_REGISTER(device2extenstate_test);
return AST_MODULE_LOAD_SUCCESS;
}
More information about the asterisk-commits
mailing list