diff --git a/tests/test_media_delivery_chat.c b/tests/test_media_delivery_chat.c index 2506d543..65ad8e26 100644 --- a/tests/test_media_delivery_chat.c +++ b/tests/test_media_delivery_chat.c @@ -216,16 +216,6 @@ static void ensure_supernode_type(int idx) { sqlite3_exec(g_inst[idx]->topo_sqlite_db, sql, NULL, NULL, NULL); } -/* прямой send media-пакета через CHAT-группу (доставка идёт по группе канала) */ -static int msend(struct UTUN_INSTANCE* inst, uint64_t dst, const uint8_t* data, size_t len) { - struct ll_entry* e = queue_entry_new(0); if (!e) return -1; - e->dgram = u_malloc(len + 1); if (!e->dgram) { queue_entry_free(e); return -1; } - e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1); - int rc = etcp_route_send(inst, g_group_id, dst, e, 1, 0); - if (rc != 0) { u_free(e->dgram); queue_entry_free(e); } - return rc; -} - /* ── завершение загрузки ── */ static void dl_done_cb(void* arg, int err) { @@ -482,20 +472,16 @@ static void setup_chat_group(void) { } TEST("media_delivery supernode s1,s2"); { - media_delivery_set_supernode(g_inst[I_S1], 1); - media_delivery_set_supernode(g_inst[I_S2], 1); + media_delivery_refresh_roles(g_inst[I_S1]); + media_delivery_refresh_roles(g_inst[I_S2]); if (g_inst[I_S1]->md.is_supernode && g_inst[I_S2]->md.is_supernode) OK(); else FAIL(); } - TEST("SUPER_HELLO s1<->s2 (via CHAT)"); { - struct media_pkt_super_hello h; memset(&h, 0, sizeof(h)); - h.subcmd = MEDIA_SUBCMD_SUPER_HELLO; h.group_id = g_group_id; - msend(g_inst[I_S1], g_nid[I_S2], (const uint8_t*)&h, sizeof(h)); - msend(g_inst[I_S2], g_nid[I_S1], (const uint8_t*)&h, sizeof(h)); + TEST("automatic SYNC s1<->s2 (via CHAT)"); { int a = 0, ok = 0; while (a < 1000) { - ok = (g_inst[I_S1]->md.super_peers && g_inst[I_S1]->md.super_peers->head) - && (g_inst[I_S2]->md.super_peers && g_inst[I_S2]->md.super_peers->head); + ok = db_count(g_inst[I_S1]->md.db, "super_sync", "group_id", g_group_id) > 0 + && db_count(g_inst[I_S2]->md.db, "super_sync", "group_id", g_group_id) > 0; if (ok) break; uasync_poll(g_ua, POLL_MS); a++; } @@ -678,8 +664,9 @@ int main(void) { uint64_t start = get_time_tb(); while ((get_time_tb() - start) < (uint64_t)60000) { sqlite3_stmt* st = NULL; uint64_t lr = 0; - sqlite3_prepare_v2(g_inst[I_S2]->topo_sqlite_db, "SELECT last_recv_id FROM super_sync WHERE peer_node_id=?", -1, &st, NULL); - sqlite3_bind_int64(st, 1, (sqlite3_int64)g_nid[I_S1]); + sqlite3_prepare_v2(g_inst[I_S2]->topo_sqlite_db, "SELECT last_recv_id FROM super_sync WHERE group_id=? AND peer_node_id=?", -1, &st, NULL); + sqlite3_bind_int64(st, 1, (sqlite3_int64)g_group_id); + sqlite3_bind_int64(st, 2, (sqlite3_int64)g_nid[I_S1]); if (sqlite3_step(st) == SQLITE_ROW) lr = (uint64_t)sqlite3_column_int64(st, 0); sqlite3_finalize(st); if (lr > 0) { ok = 1; break; } diff --git a/tests/test_media_delivery_integration.c b/tests/test_media_delivery_integration.c index 6638f30f..4365148c 100644 --- a/tests/test_media_delivery_integration.c +++ b/tests/test_media_delivery_integration.c @@ -2,11 +2,11 @@ // // Проверки: // 1. media_delivery_init на реальных инстансах (таблицы созданы) -// 2. media_delivery_set_supernode → super_peers/served_nodes созданы +// 2. Подписанная роль суперузла CHAT-группы → обработка объявлений // 3. SERVE_REG → добавление в served_nodes // 4. HAVE_BLOCK → вставка в block_availability + ACK // 5. QUERY → поиск в block_availability → QUERY_RESP -// 6. SUPER_HELLO → обмен last_recv_id +// 6. Смена подписанной роли → очистка каталога этой группы // 7. SERVE_LEAVE → удаление из served_nodes // 8. Выключение суперузла → чистка block_availability @@ -18,6 +18,9 @@ #include "../config_updater.h" #include "../routing_layer/topo_group.h" #include "../routing_layer/topo_node.h" +#include "../routing_layer/topo_node_sqlite.h" +#include "../chat/member_sync.h" +#include "../transport_layer/secure_channel.h" #include "../lib/u_async.h" #include "../lib/debug_config.h" #include "../lib/mem.h" @@ -36,6 +39,8 @@ static int g_passed = 0, g_failed = 0, g_total = 0; #define TIMEOUT_TB 300000 #define POLL_MS 5 +#define CHAT_GROUP 111 +#define CHAT_CHANNEL "111" static struct UASYNC* g_ua = NULL; static struct UTUN_INSTANCE* g_a = NULL, *g_b = NULL; @@ -45,6 +50,37 @@ static char g_tdir[] = "/tmp/utun_mdl_XXXXXX"; static char g_ca[256], g_cb[256]; static uint8_t g_a_pubkey[32], g_b_pubkey[32]; +/* Меняем роль через подписанный блок владельца, как в рабочем member_sync. */ +static int set_super_role(int enabled) { + static unsigned version; + char tags[96]; uint8_t message[MS_ADM_TAGS_MAX + 8], sig[64]; + snprintf(tags, sizeof(tags), "{\"ver\":\"%u\",\"supernode\":\"%s\"}", ++version, enabled ? "yes" : "no"); + int n = member_sync_build_owner_msg(g_nid_a, tags, message, sizeof(message)); + if (n < 0 || sc_ed25519_sign(g_a->my_ed25519_privkey, message, n, sig) != SC_OK) return -1; + struct UTUN_INSTANCE* nodes[2] = {g_a, g_b}; + for (int i = 0; i < 2; i++) { + if (member_sync_put(nodes[i], CHAT_CHANNEL, g_nid_a, g_a->my_keys.public_key, g_a->my_ed25519_pubkey, + NULL, 0, NULL, 0, "{}", tags, sig, 0, 0, NULL) < 0) return -1; + media_delivery_refresh_roles(nodes[i]); + } + return 0; +} + +static int setup_chat(void) { + struct UTUN_INSTANCE* nodes[2] = {g_a, g_b}; + uint8_t sig[64]; + if (sc_ed25519_sign(g_a->my_ed25519_privkey, (const uint8_t*)CHAT_CHANNEL, strlen(CHAT_CHANNEL), sig) != SC_OK) return -1; + for (int i = 0; i < 2; i++) { + struct UTUN_INSTANCE* inst = nodes[i]; + if (member_sync_init(inst) < 0 || topo_node_sqlite_channel_put(inst->topo_sqlite_db, CHAT_CHANNEL, "media", g_nid_a, + g_a->my_keys.public_key, NULL, g_a->my_ed25519_pubkey, NULL, sig) < 0) return -1; + if (!topo_groups_create_group(inst->topo_groups, CHAT_GROUP, TOPO_GROUP_TYPE_CHAT, CHAT_CHANNEL)) return -1; + for (int j = 0; j < 2; j++) if (member_sync_put(inst, CHAT_CHANNEL, nodes[j]->node_id, + nodes[j]->my_keys.public_key, nodes[j]->my_ed25519_pubkey, NULL, 0, NULL, 0, "{}", NULL, NULL, 0, 0, NULL) < 0) return -1; + } + return set_super_role(1); +} + /* ── helpers ── */ static int wf(const char* p, const char* f, ...) { @@ -117,7 +153,7 @@ static int msend(struct UTUN_INSTANCE* inst, uint64_t dst, const uint8_t* data, e->dgram = u_malloc(len + 1); if (!e->dgram) { queue_entry_free(e); return -1; } e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1); - int rc = etcp_route_send(inst, TOPO_GROUP_UTUN, dst, e, 1, 0); + int rc = etcp_route_send(inst, CHAT_GROUP, dst, e, 1, 0); return rc; } @@ -139,7 +175,7 @@ static void query_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { uint8_t block[16], request[16]; for (int i = 0; i < 16; i++) block[i] = (uint8_t)(0x10 + i); memset(request, 0x91, 16); - query_valid = source == g_nid_a && group == TOPO_GROUP_UTUN && r->status == MEDIA_QUERY_OK && r->num_blocks == 1 + query_valid = source == g_nid_a && group == CHAT_GROUP && r->status == MEDIA_QUERY_OK && r->num_blocks == 1 && !memcmp(r->request_id, request, 16) && !memcmp(r->media_id, block, 16) && !memcmp(e->block_id, block, 16) && e->num_holders == 1 && e->holders[0] == g_nid_a; } @@ -153,7 +189,7 @@ static void phase1_serve_reg(void) { struct media_pkt_serve_reg pkt; memset(&pkt, 0, sizeof(pkt)); pkt.subcmd = MEDIA_SUBCMD_SERVE_REG; - pkt.group_id = TOPO_GROUP_UTUN; + pkt.group_id = CHAT_GROUP; int rc = msend(g_b, g_nid_a, (const uint8_t*)&pkt, sizeof(pkt)); /* BGP routes need time — poll up to 10s */ @@ -171,7 +207,7 @@ static void phase1_serve_reg(void) { static void phase2_have_block(void) { TEST("HAVE_BLOCK B→A inserts into block_availability"); { uint8_t block_uuid[16]; for (int i = 0; i < 16; i++) block_uuid[i] = (uint8_t)(0x10 + i); - uint64_t group_id = TOPO_GROUP_UTUN; + uint64_t group_id = CHAT_GROUP; struct media_pkt_have_block hb; memset(&hb, 0, sizeof(hb)); @@ -205,7 +241,7 @@ static void phase3_query(void) { memcpy(q->media_id, block_uuid, 16); q->num_blocks = 1; memcpy(pkt + MEDIA_QUERY_HDR_SIZE, block_uuid, 16); - media_delivery_send_have_block(g_a, TOPO_GROUP_UTUN, g_nid_a, block_uuid, block_uuid, 0); + media_delivery_send_have_block(g_a, CHAT_GROUP, g_nid_a, block_uuid, block_uuid, 0); saved_media_recv = g_b->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY]; g_b->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY] = query_recv; query_received = query_valid = 0; @@ -222,7 +258,7 @@ static void phase4_serve_leave(void) { struct media_pkt_serve_leave pkt; memset(&pkt, 0, sizeof(pkt)); pkt.subcmd = MEDIA_SUBCMD_SERVE_LEAVE; - pkt.group_id = TOPO_GROUP_UTUN; + pkt.group_id = CHAT_GROUP; msend(g_b, g_nid_a, (const uint8_t*)&pkt, sizeof(pkt)); int attempts = 0; @@ -238,13 +274,13 @@ static void phase4_serve_leave(void) { static void phase5_supernode_stop(void) { TEST("supernode STOP cleans block_availability"); { - media_delivery_set_supernode(g_a, 0); + if (set_super_role(0)) { FAIL("role update failed"); return; } int n = count_rows(g_a->md.db, "block_availability", NULL, 0); if (n == 0) OK(); else FAIL("%d entries remain", n); } TEST("supernode START re-enables"); { - media_delivery_set_supernode(g_a, 1); + if (set_super_role(1)) { FAIL("role update failed"); return; } if (g_a->md.is_supernode) OK(); else FAIL(); } } @@ -312,8 +348,7 @@ int main(void) { utun_instance_init(g_a); utun_instance_init(g_b); media_delivery_init(g_a); media_delivery_init(g_b); - /* make A supernode */ - media_delivery_set_supernode(g_a, 1); + if (setup_chat()) { FAIL("CHAT setup failed"); goto done; } /* wait for ETCP connection */ mon(NULL);