velesdb-server 3.12.0

REST API server for VelesDB vector database
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

//! Integration tests for `POST /collections/{name}/stream/enable` (STREAM-5).
//!
//! Enabling streaming spawns the background drain task so a subsequent
//! `POST /collections/{name}/stream/insert` is accepted (202) instead of being
//! rejected with `409 Conflict` ("streaming not configured").

mod common;

use axum::{
    body::Body,
    http::{Request, StatusCode},
};
use common::create_test_app;
use serde_json::{json, Value};
use tempfile::TempDir;
use tower::ServiceExt;

const COLLECTION: &str = "streaming_enable";
const DIM: usize = 4;

async fn post(app: &axum::Router, uri: &str, body: Value) -> axum::response::Response {
    app.clone()
        .oneshot(
            Request::builder()
                .method("POST")
                .uri(uri)
                .header("Content-Type", "application/json")
                .body(Body::from(body.to_string()))
                .expect("test: build request"),
        )
        .await
        .expect("test: request")
}

async fn create_collection(app: &axum::Router) {
    let resp = post(
        app,
        "/collections",
        json!({"name": COLLECTION, "dimension": DIM, "metric": "cosine"}),
    )
    .await;
    assert_eq!(resp.status(), StatusCode::CREATED, "test setup: create");
}

#[tokio::test]
async fn test_rest_enable_streaming_success() {
    let temp = TempDir::new().expect("test: temp dir");
    let app = create_test_app(&temp);
    create_collection(&app).await;

    // Before enabling, stream/insert is rejected with 409 (not configured).
    let before = post(
        &app,
        &format!("/collections/{COLLECTION}/stream/insert"),
        json!({"id": 1, "vector": [0.1, 0.2, 0.3, 0.4]}),
    )
    .await;
    assert_eq!(
        before.status(),
        StatusCode::CONFLICT,
        "stream/insert should be 409 before streaming is enabled"
    );

    // Enable streaming with explicit (small) config.
    let enable = post(
        &app,
        &format!("/collections/{COLLECTION}/stream/enable"),
        json!({"bufferSize": 1024, "batchSize": 32, "flushIntervalMs": 20}),
    )
    .await;
    assert_eq!(
        enable.status(),
        StatusCode::OK,
        "enable_streaming should return 200 OK"
    );

    // After enabling, the same insert is accepted (202).
    let after = post(
        &app,
        &format!("/collections/{COLLECTION}/stream/insert"),
        json!({"id": 1, "vector": [0.1, 0.2, 0.3, 0.4]}),
    )
    .await;
    assert_eq!(
        after.status(),
        StatusCode::ACCEPTED,
        "stream/insert should be 202 once streaming is enabled"
    );
}

#[tokio::test]
async fn test_rest_enable_streaming_defaults_empty_body() {
    let temp = TempDir::new().expect("test: temp dir");
    let app = create_test_app(&temp);
    create_collection(&app).await;

    // Empty body => all fields fall back to engine defaults.
    let enable = post(
        &app,
        &format!("/collections/{COLLECTION}/stream/enable"),
        json!({}),
    )
    .await;
    assert_eq!(
        enable.status(),
        StatusCode::OK,
        "enable_streaming with empty body should succeed via defaults"
    );
}

#[tokio::test]
async fn test_rest_enable_streaming_collection_not_found() {
    let temp = TempDir::new().expect("test: temp dir");
    let app = create_test_app(&temp);

    let resp = post(&app, "/collections/does_not_exist/stream/enable", json!({})).await;
    assert_eq!(
        resp.status(),
        StatusCode::NOT_FOUND,
        "enable_streaming on a missing collection should return 404"
    );
}