Add message dump capability to stasis cache layer
[asterisk/asterisk.git] / tests / test_stasis.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * David M. Lee, II <dlee@digium.com>
7  *
8  * See http://www.asterisk.org for more information about
9  * the Asterisk project. Please do not directly contact
10  * any of the maintainers of this project for assistance;
11  * the project provides a web site, mailing lists and IRC
12  * channels for your use.
13  *
14  * This program is free software, distributed under the terms of
15  * the GNU General Public License Version 2. See the LICENSE file
16  * at the top of the source tree.
17  */
18
19 /*!
20  * \file \brief Test Stasis message bus.
21  *
22  * \author\verbatim David M. Lee, II <dlee@digium.com> \endverbatim
23  *
24  * \ingroup tests
25  */
26
27 /*** MODULEINFO
28         <depend>TEST_FRAMEWORK</depend>
29         <support_level>core</support_level>
30  ***/
31
32 #include "asterisk.h"
33
34 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
35
36 #include "asterisk/astobj2.h"
37 #include "asterisk/module.h"
38 #include "asterisk/stasis.h"
39 #include "asterisk/test.h"
40
41 static const char *test_category = "/stasis/core/";
42
43 AST_TEST_DEFINE(message_type)
44 {
45         RAII_VAR(struct stasis_message_type *, uut, NULL, ao2_cleanup);
46
47         switch (cmd) {
48         case TEST_INIT:
49                 info->name = __func__;
50                 info->category = test_category;
51                 info->summary = "Test basic message_type functions";
52                 info->description = "Test basic message_type functions";
53                 return AST_TEST_NOT_RUN;
54         case TEST_EXECUTE:
55                 break;
56         }
57
58         ast_test_validate(test, NULL == stasis_message_type_create(NULL));
59         uut = stasis_message_type_create("SomeMessage");
60         ast_test_validate(test, 0 == strcmp(stasis_message_type_name(uut), "SomeMessage"));
61
62         return AST_TEST_PASS;
63 }
64
65 AST_TEST_DEFINE(message)
66 {
67         RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
68         RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
69         RAII_VAR(char *, data, NULL, ao2_cleanup);
70         char *expected = "SomeData";
71         struct timeval expected_timestamp;
72         struct timeval time_diff;
73
74         switch (cmd) {
75         case TEST_INIT:
76                 info->name = __func__;
77                 info->category = test_category;
78                 info->summary = "Test basic message functions";
79                 info->description = "Test basic message functions";
80                 return AST_TEST_NOT_RUN;
81         case TEST_EXECUTE:
82                 break;
83         }
84
85
86         type = stasis_message_type_create("SomeMessage");
87
88         ast_test_validate(test, NULL == stasis_message_create(NULL, NULL));
89         ast_test_validate(test, NULL == stasis_message_create(type, NULL));
90
91         data = ao2_alloc(strlen(expected) + 1, NULL);
92         strcpy(data, expected);
93         expected_timestamp = ast_tvnow();
94         uut = stasis_message_create(type, data);
95
96         ast_test_validate(test, NULL != uut);
97         ast_test_validate(test, type == stasis_message_type(uut));
98         ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut)));
99         ast_test_validate(test, 2 == ao2_ref(data, 0)); /* uut has ref to data */
100
101         time_diff = ast_tvsub(*stasis_message_timestamp(uut), expected_timestamp);
102         /* 10ms is certainly long enough for the two calls to complete */
103         ast_test_validate(test, time_diff.tv_sec == 0);
104         ast_test_validate(test, time_diff.tv_usec < 10000);
105
106         ao2_ref(uut, -1);
107         uut = NULL;
108         ast_test_validate(test, 1 == ao2_ref(data, 0)); /* uut unreffed data */
109
110         return AST_TEST_PASS;
111 }
112
113 struct consumer {
114         ast_mutex_t lock;
115         ast_cond_t out;
116         struct stasis_message **messages_rxed;
117         size_t messages_rxed_len;
118         int ignore_subscriptions;
119         int complete;
120 };
121
122 static void consumer_dtor(void *obj) {
123         struct consumer *consumer = obj;
124
125         ast_mutex_destroy(&consumer->lock);
126         ast_cond_destroy(&consumer->out);
127
128         while (consumer->messages_rxed_len > 0) {
129                 ao2_cleanup(consumer->messages_rxed[--consumer->messages_rxed_len]);
130         }
131         ast_free(consumer->messages_rxed);
132         consumer->messages_rxed = NULL;
133 }
134
135 static struct consumer *consumer_create(int ignore_subscriptions) {
136         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
137
138         consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
139
140         if (!consumer) {
141                 return NULL;
142         }
143
144         consumer->ignore_subscriptions = ignore_subscriptions;
145         consumer->messages_rxed = ast_malloc(0);
146         if (!consumer->messages_rxed) {
147                 return NULL;
148         }
149
150         ast_mutex_init(&consumer->lock);
151         ast_cond_init(&consumer->out, NULL);
152
153         ao2_ref(consumer, +1);
154         return consumer;
155 }
156
157 static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
158 {
159         struct consumer *consumer = data;
160         RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
161         SCOPED_MUTEX(lock, &consumer->lock);
162
163         if (!consumer->ignore_subscriptions || stasis_message_type(message) != stasis_subscription_change()) {
164
165                 ++consumer->messages_rxed_len;
166                 consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
167                 ast_assert(consumer->messages_rxed != NULL);
168                 consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
169                 ao2_ref(message, +1);
170         }
171
172         if (stasis_subscription_final_message(sub, message)) {
173                 consumer->complete = 1;
174                 consumer_needs_cleanup = consumer;
175         }
176
177         ast_cond_signal(&consumer->out);
178 }
179
180 static int consumer_wait_for(struct consumer *consumer, size_t expected_len)
181 {
182         struct timeval start = ast_tvnow();
183         struct timespec end = {
184                 .tv_sec = start.tv_sec + 30,
185                 .tv_nsec = start.tv_usec * 1000
186         };
187
188         SCOPED_MUTEX(lock, &consumer->lock);
189
190         while (consumer->messages_rxed_len < expected_len) {
191                 int r = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
192                 if (r == ETIMEDOUT) {
193                         break;
194                 }
195                 ast_assert(r == 0); /* Not expecting any othet types of errors */
196         }
197         return consumer->messages_rxed_len;
198 }
199
200 static int consumer_wait_for_completion(struct consumer *consumer)
201 {
202         struct timeval start = ast_tvnow();
203         struct timespec end = {
204                 .tv_sec = start.tv_sec + 30,
205                 .tv_nsec = start.tv_usec * 1000
206         };
207
208         SCOPED_MUTEX(lock, &consumer->lock);
209
210         while (!consumer->complete) {
211                 int r = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
212                 if (r == ETIMEDOUT) {
213                         break;
214                 }
215                 ast_assert(r == 0); /* Not expecting any othet types of errors */
216         }
217         return consumer->complete;
218 }
219
220 static int consumer_should_stay(struct consumer *consumer, size_t expected_len)
221 {
222         struct timeval start = ast_tvnow();
223         struct timeval diff = {
224                 .tv_sec = 0,
225                 .tv_usec = 100000 /* wait for 100ms */
226         };
227         struct timeval end_tv = ast_tvadd(start, diff);
228         struct timespec end = {
229                 .tv_sec = end_tv.tv_sec,
230                 .tv_nsec = end_tv.tv_usec * 1000
231         };
232
233         SCOPED_MUTEX(lock, &consumer->lock);
234
235         while (consumer->messages_rxed_len == expected_len) {
236                 int r = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
237                 if (r == ETIMEDOUT) {
238                         break;
239                 }
240                 ast_assert(r == 0); /* Not expecting any othet types of errors */
241         }
242         return consumer->messages_rxed_len;
243 }
244
245 AST_TEST_DEFINE(subscription_messages)
246 {
247         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
248         RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
249         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
250         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
251         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
252         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
253         RAII_VAR(char *, expected_uniqueid, NULL, ast_free);
254         int complete;
255         struct stasis_subscription_change *change;
256
257         switch (cmd) {
258         case TEST_INIT:
259                 info->name = __func__;
260                 info->category = test_category;
261                 info->summary = "Test subscribe/unsubscribe messages";
262                 info->description = "Test subscribe/unsubscribe messages";
263                 return AST_TEST_NOT_RUN;
264         case TEST_EXECUTE:
265                 break;
266         }
267
268         topic = stasis_topic_create("TestTopic");
269         ast_test_validate(test, NULL != topic);
270
271         consumer = consumer_create(0);
272         ast_test_validate(test, NULL != consumer);
273
274         uut = stasis_subscribe(topic, consumer_exec, consumer);
275         ast_test_validate(test, NULL != uut);
276         ao2_ref(consumer, +1);
277         expected_uniqueid = ast_strdup(stasis_subscription_uniqueid(uut));
278
279         stasis_unsubscribe(uut);
280         uut = NULL;
281         complete = consumer_wait_for_completion(consumer);
282         ast_test_validate(test, 1 == complete);
283
284         ast_test_validate(test, 2 == consumer->messages_rxed_len);
285         ast_test_validate(test, stasis_subscription_change() == stasis_message_type(consumer->messages_rxed[0]));
286         ast_test_validate(test, stasis_subscription_change() == stasis_message_type(consumer->messages_rxed[1]));
287
288         change = stasis_message_data(consumer->messages_rxed[0]);
289         ast_test_validate(test, topic == change->topic);
290         ast_test_validate(test, 0 == strcmp("Subscribe", change->description));
291         ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
292
293         change = stasis_message_data(consumer->messages_rxed[1]);
294         ast_test_validate(test, topic == change->topic);
295         ast_test_validate(test, 0 == strcmp("Unsubscribe", change->description));
296         ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
297
298         return AST_TEST_PASS;
299 }
300
301 AST_TEST_DEFINE(publish)
302 {
303         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
304         RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
305         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
306         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
307         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
308         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
309         int actual_len;
310         const char *actual;
311
312         switch (cmd) {
313         case TEST_INIT:
314                 info->name = __func__;
315                 info->category = test_category;
316                 info->summary = "Test simple subscriptions";
317                 info->description = "Test simple subscriptions";
318                 return AST_TEST_NOT_RUN;
319         case TEST_EXECUTE:
320                 break;
321         }
322
323         topic = stasis_topic_create("TestTopic");
324         ast_test_validate(test, NULL != topic);
325
326         consumer = consumer_create(1);
327         ast_test_validate(test, NULL != consumer);
328
329         uut = stasis_subscribe(topic, consumer_exec, consumer);
330         ast_test_validate(test, NULL != uut);
331         ao2_ref(consumer, +1);
332
333         test_data = ao2_alloc(1, NULL);
334         ast_test_validate(test, NULL != test_data);
335         test_message_type = stasis_message_type_create("TestMessage");
336         test_message = stasis_message_create(test_message_type, test_data);
337
338         stasis_publish(topic, test_message);
339
340         actual_len = consumer_wait_for(consumer, 1);
341         ast_test_validate(test, 1 == actual_len);
342         actual = stasis_message_data(consumer->messages_rxed[0]);
343         ast_test_validate(test, test_data == actual);
344
345         return AST_TEST_PASS;
346 }
347
348 AST_TEST_DEFINE(unsubscribe_stops_messages)
349 {
350         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
351         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
352         RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
353         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
354         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
355         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
356         int actual_len;
357
358         switch (cmd) {
359         case TEST_INIT:
360                 info->name = __func__;
361                 info->category = test_category;
362                 info->summary = "Test simple subscriptions";
363                 info->description = "Test simple subscriptions";
364                 return AST_TEST_NOT_RUN;
365         case TEST_EXECUTE:
366                 break;
367         }
368
369         topic = stasis_topic_create("TestTopic");
370         ast_test_validate(test, NULL != topic);
371
372         consumer = consumer_create(1);
373         ast_test_validate(test, NULL != consumer);
374
375         uut = stasis_subscribe(topic, consumer_exec, consumer);
376         ast_test_validate(test, NULL != uut);
377         ao2_ref(consumer, +1);
378
379         stasis_unsubscribe(uut);
380         uut = NULL;
381
382         test_data = ao2_alloc(1, NULL);
383         ast_test_validate(test, NULL != test_data);
384         test_message_type = stasis_message_type_create("TestMessage");
385         test_message = stasis_message_create(test_message_type, test_data);
386
387         stasis_publish(topic, test_message);
388
389         actual_len = consumer_should_stay(consumer, 0);
390         ast_test_validate(test, 0 == actual_len);
391
392         return AST_TEST_PASS;
393 }
394
395
396 AST_TEST_DEFINE(forward)
397 {
398         RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
399         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
400
401         RAII_VAR(struct consumer *, parent_consumer, NULL, ao2_cleanup);
402         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
403
404         RAII_VAR(struct stasis_subscription *, forward_sub, NULL, stasis_unsubscribe);
405         RAII_VAR(struct stasis_subscription *, parent_sub, NULL, stasis_unsubscribe);
406         RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
407
408         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
409         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
410         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
411         int actual_len;
412
413         switch (cmd) {
414         case TEST_INIT:
415                 info->name = __func__;
416                 info->category = test_category;
417                 info->summary = "Test sending events to a parent topic";
418                 info->description = "Test sending events to a parent topic.\n"
419                         "This test creates three topics (one parent, two children)\n"
420                         "and publishes a message to one child, and verifies it's\n"
421                         "only seen by that child and the parent";
422                 return AST_TEST_NOT_RUN;
423         case TEST_EXECUTE:
424                 break;
425         }
426
427         parent_topic = stasis_topic_create("ParentTestTopic");
428         ast_test_validate(test, NULL != parent_topic);
429         topic = stasis_topic_create("TestTopic");
430         ast_test_validate(test, NULL != topic);
431
432         forward_sub = stasis_forward_all(topic, parent_topic);
433         ast_test_validate(test, NULL != forward_sub);
434
435         parent_consumer = consumer_create(1);
436         ast_test_validate(test, NULL != parent_consumer);
437         consumer = consumer_create(1);
438         ast_test_validate(test, NULL != consumer);
439
440         parent_sub = stasis_subscribe(parent_topic, consumer_exec, parent_consumer);
441         ast_test_validate(test, NULL != parent_sub);
442         ao2_ref(parent_consumer, +1);
443         sub = stasis_subscribe(topic, consumer_exec, consumer);
444         ast_test_validate(test, NULL != sub);
445         ao2_ref(consumer, +1);
446
447         test_data = ao2_alloc(1, NULL);
448         ast_test_validate(test, NULL != test_data);
449         test_message_type = stasis_message_type_create("TestMessage");
450         test_message = stasis_message_create(test_message_type, test_data);
451
452         stasis_publish(topic, test_message);
453
454         actual_len = consumer_wait_for(consumer, 1);
455         ast_test_validate(test, 1 == actual_len);
456         actual_len = consumer_wait_for(parent_consumer, 1);
457         ast_test_validate(test, 1 == actual_len);
458
459         return AST_TEST_PASS;
460 }
461
462 struct cache_test_data {
463         char *id;
464         char *value;
465 };
466
467 static void cache_test_data_dtor(void *obj)
468 {
469         struct cache_test_data *data = obj;
470         ast_free(data->id);
471         ast_free(data->value);
472 }
473
474 static struct stasis_message *cache_test_message_create(struct stasis_message_type *type, const char *name, const char *value)
475 {
476         RAII_VAR(struct cache_test_data *, data, NULL, ao2_cleanup);
477
478         data = ao2_alloc(sizeof(*data), cache_test_data_dtor);
479         if (data == NULL) {
480                 return NULL;
481         }
482
483         ast_assert(name != NULL);
484         ast_assert(value != NULL);
485
486         data->id = ast_strdup(name);
487         data->value = ast_strdup(value);
488         if (!data->id || !data->value) {
489                 return NULL;
490         }
491
492         return stasis_message_create(type, data);
493 }
494
495 static const char *cache_test_data_id(struct stasis_message *message) {
496         struct cache_test_data *cachable = stasis_message_data(message);
497
498         if (0 != strcmp("Cacheable", stasis_message_type_name(stasis_message_type(message)))) {
499                 return NULL;
500         }
501         return cachable->id;
502 }
503
504 AST_TEST_DEFINE(cache_passthrough)
505 {
506         RAII_VAR(struct stasis_message_type *, non_cache_type, NULL, ao2_cleanup);
507         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
508         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
509         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
510         RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
511         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
512         int actual_len;
513         struct stasis_message_type *actual_type;
514
515         switch (cmd) {
516         case TEST_INIT:
517                 info->name = __func__;
518                 info->category = test_category;
519                 info->summary = "Test passing messages through cache topic unscathed.";
520                 info->description = "Test passing messages through cache topic unscathed.";
521                 return AST_TEST_NOT_RUN;
522         case TEST_EXECUTE:
523                 break;
524         }
525
526         non_cache_type = stasis_message_type_create("NonCacheable");
527         ast_test_validate(test, NULL != non_cache_type);
528         topic = stasis_topic_create("SomeTopic");
529         ast_test_validate(test, NULL != topic);
530         caching_topic = stasis_caching_topic_create(topic, cache_test_data_id);
531         ast_test_validate(test, NULL != caching_topic);
532         consumer = consumer_create(1);
533         ast_test_validate(test, NULL != consumer);
534         sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
535         ast_test_validate(test, NULL != sub);
536         ao2_ref(consumer, +1);
537
538         test_message = cache_test_message_create(non_cache_type, "1", "1");
539         ast_test_validate(test, NULL != test_message);
540
541         stasis_publish(topic, test_message);
542
543         actual_len = consumer_wait_for(consumer, 1);
544         ast_test_validate(test, 1 == actual_len);
545
546         actual_type = stasis_message_type(consumer->messages_rxed[0]);
547         ast_test_validate(test, non_cache_type == actual_type);
548
549         ast_test_validate(test, test_message == consumer->messages_rxed[0]);
550
551         return AST_TEST_PASS;
552 }
553
554 AST_TEST_DEFINE(cache)
555 {
556         RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
557         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
558         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
559         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
560         RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
561         RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
562         RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
563         RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
564         RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
565         int actual_len;
566         struct stasis_cache_update *actual_update;
567         struct ao2_container *cache_dump;
568
569         switch (cmd) {
570         case TEST_INIT:
571                 info->name = __func__;
572                 info->category = test_category;
573                 info->summary = "Test passing messages through cache topic unscathed.";
574                 info->description = "Test passing messages through cache topic unscathed.";
575                 return AST_TEST_NOT_RUN;
576         case TEST_EXECUTE:
577                 break;
578         }
579
580         cache_type = stasis_message_type_create("Cacheable");
581         ast_test_validate(test, NULL != cache_type);
582         topic = stasis_topic_create("SomeTopic");
583         ast_test_validate(test, NULL != topic);
584         caching_topic = stasis_caching_topic_create(topic, cache_test_data_id);
585         ast_test_validate(test, NULL != caching_topic);
586         consumer = consumer_create(1);
587         ast_test_validate(test, NULL != consumer);
588         sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
589         ast_test_validate(test, NULL != sub);
590         ao2_ref(consumer, +1);
591
592         test_message1_1 = cache_test_message_create(cache_type, "1", "1");
593         ast_test_validate(test, NULL != test_message1_1);
594         test_message2_1 = cache_test_message_create(cache_type, "2", "1");
595         ast_test_validate(test, NULL != test_message2_1);
596
597         /* Post a couple of snapshots */
598         stasis_publish(topic, test_message1_1);
599         stasis_publish(topic, test_message2_1);
600         actual_len = consumer_wait_for(consumer, 2);
601         ast_test_validate(test, 2 == actual_len);
602
603         /* Dump the cache to ensure that it has the correct number of items in it */
604         cache_dump = stasis_cache_dump(caching_topic, NULL);
605         ast_test_validate(test, 2 == ao2_container_count(cache_dump));
606         ao2_ref(cache_dump, -1);
607         cache_dump = NULL;
608
609         /* Check for new snapshot messages */
610         ast_test_validate(test, stasis_cache_update() == stasis_message_type(consumer->messages_rxed[0]));
611         actual_update = stasis_message_data(consumer->messages_rxed[0]);
612         ast_test_validate(test, topic == actual_update->topic);
613         ast_test_validate(test, NULL == actual_update->old_snapshot);
614         ast_test_validate(test, test_message1_1 == actual_update->new_snapshot);
615         ast_test_validate(test, test_message1_1 == stasis_cache_get(caching_topic, cache_type, "1"));
616         /* stasis_cache_get returned a ref, so unref test_message1_1 */
617         ao2_ref(test_message1_1, -1);
618
619         ast_test_validate(test, stasis_cache_update() == stasis_message_type(consumer->messages_rxed[1]));
620         actual_update = stasis_message_data(consumer->messages_rxed[1]);
621         ast_test_validate(test, topic == actual_update->topic);
622         ast_test_validate(test, NULL == actual_update->old_snapshot);
623         ast_test_validate(test, test_message2_1 == actual_update->new_snapshot);
624         ast_test_validate(test, test_message2_1 == stasis_cache_get(caching_topic, cache_type, "2"));
625         /* stasis_cache_get returned a ref, so unref test_message2_1 */
626         ao2_ref(test_message2_1, -1);
627
628         /* Update snapshot 2 */
629         test_message2_2 = cache_test_message_create(cache_type, "2", "2");
630         ast_test_validate(test, NULL != test_message2_2);
631         stasis_publish(topic, test_message2_2);
632
633         actual_len = consumer_wait_for(consumer, 3);
634         ast_test_validate(test, 3 == actual_len);
635
636         actual_update = stasis_message_data(consumer->messages_rxed[2]);
637         ast_test_validate(test, topic == actual_update->topic);
638         ast_test_validate(test, test_message2_1 == actual_update->old_snapshot);
639         ast_test_validate(test, test_message2_2 == actual_update->new_snapshot);
640         ast_test_validate(test, test_message2_2 == stasis_cache_get(caching_topic, cache_type, "2"));
641         /* stasis_cache_get returned a ref, so unref test_message2_2 */
642         ao2_ref(test_message2_2, -1);
643
644         /* Dump the cache to ensure that it has the correct number of items in it */
645         cache_dump = stasis_cache_dump(caching_topic, NULL);
646         ast_test_validate(test, 2 == ao2_container_count(cache_dump));
647         ao2_ref(cache_dump, -1);
648         cache_dump = NULL;
649
650         /* Clear snapshot 1 */
651         test_message1_clear = stasis_cache_clear_create(cache_type, "1");
652         ast_test_validate(test, NULL != test_message1_clear);
653         stasis_publish(topic, test_message1_clear);
654
655         actual_len = consumer_wait_for(consumer, 4);
656         ast_test_validate(test, 4 == actual_len);
657
658         actual_update = stasis_message_data(consumer->messages_rxed[3]);
659         ast_test_validate(test, topic == actual_update->topic);
660         ast_test_validate(test, test_message1_1 == actual_update->old_snapshot);
661         ast_test_validate(test, NULL == actual_update->new_snapshot);
662         ast_test_validate(test, NULL == stasis_cache_get(caching_topic, cache_type, "1"));
663
664         /* Dump the cache to ensure that it has the correct number of items in it */
665         cache_dump = stasis_cache_dump(caching_topic, NULL);
666         ast_test_validate(test, 1 == ao2_container_count(cache_dump));
667         ao2_ref(cache_dump, -1);
668         cache_dump = NULL;
669
670         /* Dump the cache to ensure that it has no subscription change items in it since those aren't cached */
671         cache_dump = stasis_cache_dump(caching_topic, stasis_subscription_change());
672         ast_test_validate(test, 0 == ao2_container_count(cache_dump));
673         ao2_ref(cache_dump, -1);
674         cache_dump = NULL;
675
676         return AST_TEST_PASS;
677 }
678
679 static int unload_module(void)
680 {
681         AST_TEST_UNREGISTER(message_type);
682         AST_TEST_UNREGISTER(message);
683         AST_TEST_UNREGISTER(subscription_messages);
684         AST_TEST_UNREGISTER(publish);
685         AST_TEST_UNREGISTER(unsubscribe_stops_messages);
686         AST_TEST_UNREGISTER(forward);
687         AST_TEST_UNREGISTER(cache_passthrough);
688         AST_TEST_UNREGISTER(cache);
689         return 0;
690 }
691
692 static int load_module(void)
693 {
694         AST_TEST_REGISTER(message_type);
695         AST_TEST_REGISTER(message);
696         AST_TEST_REGISTER(subscription_messages);
697         AST_TEST_REGISTER(publish);
698         AST_TEST_REGISTER(unsubscribe_stops_messages);
699         AST_TEST_REGISTER(forward);
700         AST_TEST_REGISTER(cache_passthrough);
701         AST_TEST_REGISTER(cache);
702         return AST_MODULE_LOAD_SUCCESS;
703 }
704
705 AST_MODULE_INFO(ASTERISK_GPL_KEY, 0, "Stasis testing",
706                 .load = load_module,
707                 .unload = unload_module
708         );