utun2/tests/test_db_sync.c

// test_db_sync.c — всестороннее тестирование модуля db_sync
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include <time.h>
#include <sys/stat.h>
#include "../lib/platform_compat.h"
#include "test_utils.h"
#ifdef _WIN32
#include <windows.h>
#include <direct.h>
#else
#include <unistd.h>
#endif

#include "../src/etcp.h"
#include "../src/etcp_connections.h"
#include "../src/config_parser.h"
#include "../src/config_updater.h"
#include "../src/utun_instance.h"
#include "../src/routing.h"
#include "../src/tun_if.h"
#include "../src/secure_channel.h"
#include "../src/db_sync.h"
#include "../lib/u_async.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"

#define TEST_TIMEOUT_TB  300000  // 30s total
#define PHASE_TIMEOUT_TB 150000  // 15s per phase
#define POLL_INTERVAL_MS 5

#define NODE_ID_A 0xAAAAAAAAAAAAAAAAULL
#define NODE_ID_B 0xBBBBBBBBBBBBBBBBULL

static struct UTUN_INSTANCE* inst_a = NULL;
static struct UTUN_INSTANCE* inst_b = NULL;
static struct DB_SYNC_INSTANCE* si_a = NULL;
static struct DB_SYNC_INSTANCE* si_b = NULL;
static struct UASYNC* ua = NULL;
static int test_phase = 0;  // 0=running, 1=success, 2=failure
static void* timeout_id = NULL;

static char temp_dir[] = "/tmp/utun_dbsync_XXXXXX";
static char config_a[256], config_b[256];
static int port_a_srv, port_b_srv;

static int write_file(const char* path, const char* fmt, ...) {
    va_list ap;
    FILE* f = fopen(path, "w");
    if (!f) return -1;
    va_start(ap, fmt); vfprintf(f, fmt, ap); va_end(ap);
    fclose(f); return 0;
}
static char* get_pubkey(const char* path) {
    struct utun_config* cfg = parse_config(path);
    if (!cfg) return NULL;
    char* pub = strdup(cfg->global.my_public_key_hex);
    free_config(cfg); return pub;
}

static int create_temp_configs(void) {
    if (test_mkdtemp(temp_dir) != 0) { fprintf(stderr, "mkdtemp failed\n"); return -1; }
    int base = 42000 + (getpid() % 15000);
    port_a_srv = base; port_b_srv = base + 1;
    snprintf(config_a, sizeof(config_a), "%s/a.conf", temp_dir);
    snprintf(config_b, sizeof(config_b), "%s/b.conf", temp_dir);

    // Create SQLite db directories (parent dir for sync file)
    char db_path[320];
    snprintf(db_path, sizeof(db_path), "%s/db_a", temp_dir); utun_mkdir(db_path, 0755);
    snprintf(db_path, sizeof(db_path), "%s/db_b", temp_dir); utun_mkdir(db_path, 0755);

    if (write_file(config_a,
        "[global]\n"
        "my_node_id=0xAAAAAAAAAAAAAAAA\n"
        "tun_ip=10.200.0.1/24\n"
        "tun_ifname=tun200\n"
        "keepalive_adaptive=0\n"
        "db_path=%s/db_a\n"
        "db_sync_enabled=1\n"
        "\n"
        "[server: srv_a]\n"
        "addr=127.0.0.1:%d\n"
        "type=public\n"
        "\n"
        "[allowed_keys]\n"
        "allow_all=1\n",
        temp_dir, port_a_srv) != 0) return -1;
    if (config_ensure_keys_and_node_id(config_a) != 0) return -1;
    char* pub_a = get_pubkey(config_a);
    if (!pub_a) return -1;

    if (write_file(config_b,
        "[global]\n"
        "my_node_id=0xBBBBBBBBBBBBBBBB\n"
        "tun_ip=10.200.0.2/24\n"
        "tun_ifname=tun201\n"
        "keepalive_adaptive=0\n"
        "db_path=%s/db_b\n"
        "db_sync_enabled=1\n"
        "\n"
        "[server: srv_b]\n"
        "addr=127.0.0.1:%d\n"
        "type=public\n"
        "\n"
        "[client: to_a]\n"
        "keepalive=1\n"
        "peer_public_key=%s\n"
        "link=srv_b:127.0.0.1:%d\n"
        "\n"
        "[allowed_keys]\n"
        "allow_all=1\n",
        temp_dir, port_b_srv, pub_a, port_a_srv) != 0) { free(pub_a); return -1; }
    free(pub_a);
    if (config_ensure_keys_and_node_id(config_b) != 0) return -1;
    return 0;
}

static void cleanup_temp_configs(void) {
    unlink(config_a); unlink(config_b);
    char pa[320];
    snprintf(pa, sizeof(pa), "%s/db_a/sync", temp_dir); unlink(pa);
    snprintf(pa, sizeof(pa), "%s/db_a/sync-wal", temp_dir); unlink(pa);
    snprintf(pa, sizeof(pa), "%s/db_a/sync-shm", temp_dir); unlink(pa);
    snprintf(pa, sizeof(pa), "%s/db_b/sync", temp_dir); unlink(pa);
    snprintf(pa, sizeof(pa), "%s/db_b/sync-wal", temp_dir); unlink(pa);
    snprintf(pa, sizeof(pa), "%s/db_b/sync-shm", temp_dir); unlink(pa);
    snprintf(pa, sizeof(pa), "%s/db_a", temp_dir); test_rmdir(pa);
    snprintf(pa, sizeof(pa), "%s/db_b", temp_dir); test_rmdir(pa);
    test_rmdir(temp_dir);
}

static void test_timeout(void* arg) { (void)arg; test_phase = 2; }

static int wait_for(const char* desc, int (*cond)(void), int timeout_tb) {
    uint64_t start = get_time_tb();
    while (!cond() && (get_time_tb() - start) < (uint64_t)timeout_tb && test_phase == 0)
        uasync_poll(ua, POLL_INTERVAL_MS);
    if (cond()) return 1;
    if (test_phase == 0) { fprintf(stderr, "TIMEOUT: %s\n", desc); test_phase = 2; }
    return 0;
}

static void sleep_tb(int tb) {
    uint64_t end = get_time_tb() + (uint64_t)tb;
    while (get_time_tb() < end && test_phase == 0) uasync_poll(ua, POLL_INTERVAL_MS);
}

// ---- Condition functions ----
static int cond_links_init(void) {
    if (!inst_a || !inst_b) return 0;
    struct ll_entry* entry = inst_a->connections->head;
    while (entry) { struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data;
        struct ETCP_CONN* ca = ce->conn;
        struct ETCP_LINK* l = ca->links; while (l) { if (l->initialized) return 1; l = l->next; }
        entry = entry->next; }
    return 0;
}

static int cond_count_a(uint32_t expected) {
    if (!si_a) return 0;
    return db_sync_count(si_a) == expected;
}
static int cond_count_b(uint32_t expected) {
    if (!si_b) return 0;
    return db_sync_count(si_b) == expected;
}
static uint32_t count_a_target, count_b_target;
static int _cond_count_a(void) { return cond_count_a(count_a_target); }
static int _cond_count_b(void) { return cond_count_b(count_b_target); }

static int insert_many(struct DB_SYNC_INSTANCE* si, int start, int count) {
    char buf[128];
    for (int i = start; i < start + count && test_phase == 0; i++) {
        snprintf(buf, sizeof(buf), "{\"idx\":%d,\"val\":\"data_%d\",\"pad\":\"%s\"}", i, i,
                 "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx");
        int ret = db_sync_insert(si, buf);
        if (ret < 0) { fprintf(stderr, "insert_many failed at %d ret=%d\n", i, ret); return -1; }
    }
    return 0;
}

static int insert_many_batch(struct DB_SYNC_INSTANCE* si, int start, int count) {
    char buf[256];
    for (int i = start; i < start + count && test_phase == 0; i++) {
        snprintf(buf, sizeof(buf), "{\"n\":%d,\"text\":\"record_number_%d_abcdefghijklmnopqrstuvwxyz\"}", i, i);
        db_sync_insert(si, buf);
    }
    return 0;
}

// ---- Main test ----
int main(void) {
    printf("=== test_db_sync ===\n");

    debug_config_init();
    debug_set_level(DEBUG_LEVEL_ERROR);  // quiet

    if (create_temp_configs() != 0) { fprintf(stderr, "config creation failed\n"); return 1; }

    utun_instance_set_tun_init_enabled(0);
    ua = uasync_create();
    if (!ua) { fprintf(stderr, "uasync_create failed\n"); cleanup_temp_configs(); return 1; }

    inst_a = utun_instance_create(ua, config_a);
    inst_b = utun_instance_create(ua, config_b);
    if (!inst_a || !inst_b) { fprintf(stderr, "instance create failed\n"); cleanup_temp_configs(); return 1; }

    if (utun_instance_init(inst_a) != 0 || utun_instance_init(inst_b) != 0) {
        fprintf(stderr, "instance init failed\n"); cleanup_temp_configs(); return 1;
    }

    si_a = db_sync_instance_add(inst_a, "test", 1);
    si_b = db_sync_instance_add(inst_b, "test", 1);
    if (!si_a || !si_b) { fprintf(stderr, "db_sync_instance_add failed\n"); return 1; }

    timeout_id = uasync_set_timeout(ua, TEST_TIMEOUT_TB, NULL, test_timeout, "test_timeout");

    // ===== Phase 1: базовый CRUD =====
    printf("Phase 1: basic CRUD...\n");
    if (db_sync_count(si_a) != 0) { fprintf(stderr, "FAIL: initial count not 0 (got %u)\n", db_sync_count(si_a)); test_phase=2; }

    if (db_sync_insert(si_a, "{\"key\":\"val1\"}") != 0) { fprintf(stderr,"FAIL: insert 1\n"); test_phase=2; }
    if (db_sync_insert(si_a, "{\"key\":\"val2\"}") != 0) { fprintf(stderr,"FAIL: insert 2\n"); test_phase=2; }
    if (db_sync_insert(si_a, "{\"key\":\"val3\"}") != 0) { fprintf(stderr,"FAIL: insert 3\n"); test_phase=2; }
    if (db_sync_count(si_a) != 3) { fprintf(stderr,"FAIL: count not 3 (got %u)\n", db_sync_count(si_a)); test_phase=2; }
    // Dedup by (timestamp,datahash): same content at different time = new record
    if (db_sync_insert(si_a, "{\"key\":\"val4\"}") != 0) { fprintf(stderr,"FAIL: insert 4\n"); test_phase=2; }
    if (db_sync_count(si_a) != 4) { fprintf(stderr,"FAIL: count not 4 (got %u)\n", db_sync_count(si_a)); test_phase=2; }
    if (test_phase == 0) printf("Phase 1: PASS (count=4)\n");

    // ===== Phase 2: initial sync A↔B =====
    printf("Phase 2: initial sync...\n");
    if (test_phase == 0) { count_b_target = 4; if (!wait_for("B count=4", _cond_count_b, PHASE_TIMEOUT_TB)) test_phase=2; }
    if (test_phase == 0) printf("Phase 2: PASS (B synced %u records)\n", db_sync_count(si_b));

    // ===== Cleanup =====
    if (timeout_id && ua) { uasync_cancel_timeout(ua, timeout_id); timeout_id = NULL; }
    if (si_a) { db_sync_instance_remove(si_a); si_a = NULL; }
    if (si_b) { db_sync_instance_remove(si_b); si_b = NULL; }
    if (inst_a) { inst_a->running = 0; utun_instance_destroy(inst_a); inst_a = NULL; }
    if (inst_b) { inst_b->running = 0; utun_instance_destroy(inst_b); inst_b = NULL; }
    if (ua) { uasync_destroy(ua, 0); ua = NULL; }
    cleanup_temp_configs();

    if (test_phase == 0) test_phase = 1;
    printf("=== %s ===\n", test_phase == 1 ? "PASS" : "FAIL");
    return test_phase == 1 ? 0 : 1;
}