Additional tests

tests/CMakeLists.txt

@@ -518,6 +518,8 @@ add_test_case(rrsm_offline_acquire_streaming_pending_clean_up_unsubscribe_overri
 add_test_case(rrsm_acquire_request_success_offline_online_no_session_subscription_ended_can_reacquire)
 add_test_case(rrsm_request_subscription_ended_while_unsubscribing)
 add_test_case(rrsm_streaming_subscription_lost_resubscribe_on_no_session)
+add_test_case(rrsm_request_subscription_purge_events)
+add_test_case(rrsm_streaming_subscription_purge_events)
 
 # "rrc" = request response client
 add_test_case(rrc_mqtt5_create_destroy)
@@ -528,6 +530,7 @@ add_test_case(rrc_submit_request_operation_failure_invalid_response_topic)
 add_test_case(rrc_submit_request_operation_failure_invalid_response_correlation_token_path)
 add_test_case(rrc_submit_request_operation_failure_no_correlation_token)
 add_test_case(rrc_submit_request_operation_failure_invalid_publish_topic)
+add_test_case(rrc_submit_request_operation_failure_invalid_subscription_topic_filter)
 add_test_case(rrc_submit_request_operation_failure_empty_request)
 add_test_case(rrc_submit_streaming_operation_failure_invalid_subscription_topic_filter)
 

tests/request-response/request_response_client_tests.c

@@ -822,6 +822,20 @@ AWS_TEST_CASE(
     rrc_submit_request_operation_failure_invalid_publish_topic,
     s_rrc_submit_request_operation_failure_invalid_publish_topic_fn)
 
+static void s_invalid_subscription_topic_filter_mutator(struct aws_mqtt_request_operation_options *request_options) {
+    request_options->subscription_topic_filter = aws_byte_cursor_from_c_str("a/#/c");
+}
+
+static int s_rrc_submit_request_operation_failure_invalid_subscription_topic_filter_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    return s_rrc_do_submit_request_operation_failure_test(allocator, s_invalid_subscription_topic_filter_mutator);
+}
+
+AWS_TEST_CASE(
+    rrc_submit_request_operation_failure_invalid_subscription_topic_filter,
+    s_rrc_submit_request_operation_failure_invalid_subscription_topic_filter_fn)
+
 static void s_empty_request_mutator(struct aws_mqtt_request_operation_options *request_options) {
     request_options->serialized_request = aws_byte_cursor_from_c_str("");
 }

tests/request-response/subscription_manager_tests.c

@@ -2069,3 +2069,113 @@ static int s_rrsm_streaming_subscription_lost_resubscribe_on_no_session_fn(struc
 AWS_TEST_CASE(
     rrsm_streaming_subscription_lost_resubscribe_on_no_session,
     s_rrsm_streaming_subscription_lost_resubscribe_on_no_session_fn)
+
+static int s_do_purge_test(struct aws_allocator *allocator, enum aws_rr_subscription_type subscription_type) {
+    aws_mqtt_library_init(allocator);
+
+    struct aws_subscription_manager_test_fixture fixture;
+    ASSERT_SUCCESS(s_aws_subscription_manager_test_fixture_init(&fixture, allocator, NULL));
+
+    struct aws_rr_subscription_manager *manager = &fixture.subscription_manager;
+
+    struct aws_rr_acquire_subscription_options acquire1_options = {
+        .type = subscription_type,
+        .topic_filter = aws_byte_cursor_from_c_str("hello/world"),
+        .operation_id = 1,
+    };
+    ASSERT_INT_EQUALS(AASRT_SUBSCRIBING, aws_rr_subscription_manager_acquire_subscription(manager, &acquire1_options));
+
+    struct aws_protocol_adapter_subscription_event successful_subscription_event = {
+        .topic_filter = aws_byte_cursor_from_c_str("hello/world"),
+        .event_type = AWS_PASET_SUBSCRIBE,
+        .error_code = AWS_ERROR_SUCCESS,
+    };
+    aws_rr_subscription_manager_on_protocol_adapter_subscription_event(manager, &successful_subscription_event);
+
+    struct aws_subscription_status_record expected_empty_subscription_events[] = {
+        {
+            .type = ARRSET_SUBSCRIPTION_EMPTY,
+            .topic_filter_cursor = aws_byte_cursor_from_c_str("hello/world"),
+            .operation_id = 0,
+        },
+    };
+
+    struct aws_subscription_status_record expected_unsubscribe_events[] = {
+        {
+            .type = ARRSET_UNSUBSCRIBE_COMPLETE,
+            .topic_filter_cursor = aws_byte_cursor_from_c_str("hello/world"),
+            .operation_id = 0,
+        },
+    };
+
+    ASSERT_FALSE(s_contains_subscription_event_records(&fixture, 1, expected_empty_subscription_events));
+    ASSERT_FALSE(s_contains_subscription_event_records(&fixture, 1, expected_unsubscribe_events));
+
+    // verify no unsubscribes
+    struct aws_protocol_adapter_api_record expected_unsubscribe = {
+        .type = PAAT_UNSUBSCRIBE,
+        .topic_filter_cursor = aws_byte_cursor_from_c_str("hello/world"),
+        .timeout = DEFAULT_SM_TEST_TIMEOUT,
+    };
+    ASSERT_FALSE(s_api_records_contains_record(fixture.mock_protocol_adapter, &expected_unsubscribe));
+
+    // release once, verify no unsubscribe
+    struct aws_rr_release_subscription_options release1_options = {
+        .topic_filter = aws_byte_cursor_from_c_str("hello/world"),
+        .operation_id = 1,
+    };
+    aws_rr_subscription_manager_release_subscription(manager, &release1_options);
+    ASSERT_FALSE(s_api_records_contains_record(fixture.mock_protocol_adapter, &expected_unsubscribe));
+
+    // verify empty event, but no unsubscribe event yet
+    ASSERT_TRUE(s_contains_subscription_event_records(&fixture, 1, expected_empty_subscription_events));
+    ASSERT_FALSE(s_contains_subscription_event_records(&fixture, 1, expected_unsubscribe_events));
+
+    // unsubscribe is lazy, so we need to trigger it by acquiring something else
+    struct aws_rr_acquire_subscription_options acquire3_options = {
+        .type = ARRST_REQUEST_RESPONSE,
+        .topic_filter = aws_byte_cursor_from_c_str("hello/world2"),
+        .operation_id = 3,
+    };
+    ASSERT_INT_EQUALS(AASRT_SUBSCRIBING, aws_rr_subscription_manager_acquire_subscription(manager, &acquire3_options));
+
+    // now the unsubscribe should be present
+    ASSERT_TRUE(s_api_records_contains_record(fixture.mock_protocol_adapter, &expected_unsubscribe));
+
+    // complete the unsubscribe
+    struct aws_protocol_adapter_subscription_event successful_unsubscribe_event = {
+        .topic_filter = aws_byte_cursor_from_c_str("hello/world"),
+        .event_type = AWS_PASET_UNSUBSCRIBE,
+        .error_code = AWS_ERROR_SUCCESS,
+    };
+    aws_rr_subscription_manager_on_protocol_adapter_subscription_event(manager, &successful_unsubscribe_event);
+
+    // verify unsubscribe attempt emission
+    ASSERT_TRUE(s_contains_subscription_event_records(&fixture, 1, expected_unsubscribe_events));
+
+    s_aws_subscription_manager_test_fixture_clean_up(&fixture);
+    aws_mqtt_library_clean_up();
+
+    return AWS_OP_SUCCESS;
+}
+
+static int s_rrsm_request_subscription_purge_events_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    return s_do_purge_test(allocator, ARRST_REQUEST_RESPONSE);
+}
+
+AWS_TEST_CASE(
+    rrsm_request_subscription_purge_events,
+    s_rrsm_request_subscription_purge_events_fn)
+
+static int s_rrsm_streaming_subscription_purge_events_fn(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    return s_do_purge_test(allocator, ARRST_EVENT_STREAM);
+}
+
+AWS_TEST_CASE(
+    rrsm_streaming_subscription_purge_events,
+    s_rrsm_streaming_subscription_purge_events_fn)
+