From cb9ef9450127330c55f19c405537c5386b8fc94d Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 14:07:19 +0200 Subject: [PATCH] Keep protocol and overload fixtures under explicit manual upstream control --- tests/test_stream_service.c | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 815cc24f..b1240bc5 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -255,7 +255,7 @@ static void test_network(void) { CHECK(stream_end_verify(&origin.descriptor, nodes[3]->my_ed25519_pubkey, end, sizeof(end), &last) == STREAM_ERR_SIGNATURE); for (unsigned i = 1; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); for (unsigned i = 1; i < 3; i++) { - CHECK(stream_subscribe_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK); stream_set_event_callback(nodes[i], on_network, NULL); } network_write(publisher, 0); network_write(publisher, 1); @@ -322,7 +322,7 @@ static void test_network(void) { CHECK(stream_publish_open(nodes[3], GROUP1, ¶meters, ¶llel_publisher) == STREAM_OK); struct stream_info parallel_info; CHECK(stream_get_info(parallel_publisher, ¶llel_info) == STREAM_OK); wait_state(1, 3, GROUP1, parallel_info.descriptor.source_epoch, 0); - CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[3]->node_id, ¶llel_subscriber) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, ¶llel_subscriber) == STREAM_OK); size_t accepted = 0; CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected)); deadline = get_time_tb() + 100000; @@ -382,7 +382,7 @@ static void test_pressure(void) { CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK); struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); wait_state(1, 3, GROUP2, origin.descriptor.source_epoch, 0); - CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &subscriber) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &subscriber) == STREAM_OK); stream_set_event_callback(nodes[1], on_pressure, NULL); for (unsigned i = 1; i <= 3; i += 2) { pressure_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(pressure_handlers[i] != NULL); @@ -508,7 +508,7 @@ static void test_global_pressure(void) { struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); for (unsigned i = 1; i < 3; i++) { wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); - CHECK(stream_subscribe_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK); stream_set_event_callback(nodes[i], on_global_pressure, NULL); } uint8_t payload[1180]; memset(payload, 3, sizeof(payload)); size_t accepted; @@ -684,11 +684,11 @@ int main(void) { struct stream_info info; CHECK(stream_get_info(publisher, &info) == STREAM_OK); uint64_t epoch = info.descriptor.source_epoch; for (unsigned i = 1; i < 4; i++) wait_state(i, 0, GROUP1, epoch, 0); - CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[0]->node_id, &reader1) == STREAM_OK); - CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[0]->node_id, &reader2) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[0]->node_id, &reader1) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[0]->node_id, &reader2) == STREAM_OK); CHECK(stream_get_info(reader1, &info) == STREAM_OK && info.owners == 2 && info.buffer.blocks == 0); - CHECK(stream_subscribe_open(nodes[2], GROUP1, nodes[0]->node_id, &mobile) == STREAM_OK); - CHECK(stream_subscribe_open(nodes[3], GROUP1, nodes[0]->node_id, &desktop) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[2], GROUP1, nodes[0]->node_id, &mobile) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[3], GROUP1, nodes[0]->node_id, &desktop) == STREAM_OK); struct stream_member_state state; CHECK(stream_member_load(nodes[1], GROUP1, nodes[1]->node_id, &state, NULL) == STREAM_OK && state.count == 1); for (unsigned i = 2; i < 4; i++) wait_state(1, i, GROUP1, epoch, 1); @@ -726,7 +726,7 @@ int main(void) { CHECK(topo_node_sqlite_member_get_sign(nodes[0]->topo_sqlite_db, "710002", nodes[0]->node_id, &parent, tree2) == 0); CHECK(member_sync_put(nodes[1], "710002", nodes[0]->node_id, nodes[0]->my_keys.public_key, nodes[0]->my_ed25519_pubkey, join2, join_ts, sig2, version2, profile2, NULL, NULL, 0, parent, tree2) >= 0); - CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK); for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3); struct stream_encoder *encoder = NULL; @@ -766,7 +766,7 @@ int main(void) { CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_MEMBER); run_until(&closed_events[1], 2); stream_close(other_group); other_group = NULL; put_member(1, 1, GROUP2); - CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK); chat_core_delete_channel(nodes[1], "710002"); CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_GROUP); stream_close(other_group);