~bigbes/lethe

ref: ad5ee6528aae4f5d70cee77f87bd9b275eb21928 lethe/internal/collector/ingest/sender_test.go -rw-r--r-- 5.1 KiB
ad5ee652 — Eugene Blikh collector: add ingest sender and outbox replay 24 days ago
                                                                                
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
package ingest

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	"sourcecraft.dev/bigbes/lethe/internal/shared/wire"
)

func TestEncodeNDJSON_EmitsOneObjectPerLineWithTrailingNewline(t *testing.T) {
	events := []wire.TurnEvent{
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t1", Seq: 1, Role: "user", Timestamp: 1000, Content: "hello"},
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t2", Seq: 2, Role: "assistant", Timestamp: 1001, Content: "world"},
	}

	data, err := EncodeNDJSON(events)
	if err != nil {
		t.Fatalf("EncodeNDJSON: %v", err)
	}

	lines := strings.Split(string(data), "\n")
	// Trailing newline means last element is empty.
	if len(lines) != 3 {
		t.Fatalf("expected 3 lines (2 events + trailing empty), got %d", len(lines))
	}
	if lines[2] != "" {
		t.Errorf("expected trailing empty line, got %q", lines[2])
	}

	for i, line := range lines[:2] {
		if line == "" {
			t.Fatalf("line %d is empty", i)
		}
		var ev wire.TurnEvent
		if err := json.Unmarshal([]byte(line), &ev); err != nil {
			t.Fatalf("line %d invalid JSON: %v", i, err)
		}
	}
}

func TestEncodeNDJSON_EmptySlice(t *testing.T) {
	data, err := EncodeNDJSON([]wire.TurnEvent{})
	if err != nil {
		t.Fatalf("EncodeNDJSON: %v", err)
	}
	if len(data) != 0 {
		t.Errorf("expected empty bytes for empty slice, got %q", string(data))
	}
}

func TestSender_PostBatch_Success(t *testing.T) {
	var gotBody []byte
	var gotContentType string
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			t.Errorf("expected POST, got %s", r.Method)
		}
		if r.URL.Path != "/api/v1/ingest" {
			t.Errorf("expected path /api/v1/ingest, got %s", r.URL.Path)
		}
		gotContentType = r.Header.Get("Content-Type")
		var err error
		gotBody, err = io.ReadAll(r.Body)
		if err != nil {
			t.Fatalf("read body: %v", err)
		}
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte(`{"accepted":2,"errors":[]}`))
	}))
	defer ts.Close()

	sender := NewSender(ts.URL, http.DefaultClient)
	events := []wire.TurnEvent{
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t1", Seq: 1, Role: "user", Timestamp: 1000, Content: "hello"},
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t2", Seq: 2, Role: "assistant", Timestamp: 1001, Content: "world"},
	}

	result, err := sender.PostBatch(context.Background(), events)
	if err != nil {
		t.Fatalf("PostBatch: %v", err)
	}
	if result.Accepted != 2 {
		t.Errorf("Accepted = %d, want 2", result.Accepted)
	}
	if len(result.Errors) != 0 {
		t.Errorf("Errors = %v, want empty", result.Errors)
	}
	if gotContentType != "application/x-ndjson" {
		t.Errorf("Content-Type = %q, want application/x-ndjson", gotContentType)
	}

	// Verify body is valid NDJSON.
	lines := strings.Split(string(gotBody), "\n")
	if len(lines) != 3 || lines[2] != "" {
		t.Errorf("body does not look like NDJSON with trailing newline: %q", string(gotBody))
	}
}

func TestSender_PostBatch_Non2xx(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusServiceUnavailable)
		_, _ = w.Write([]byte(`busy`))
	}))
	defer ts.Close()

	sender := NewSender(ts.URL, http.DefaultClient)
	_, err := sender.PostBatch(context.Background(), []wire.TurnEvent{
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t1", Seq: 1, Role: "user", Timestamp: 1000, Content: "hello"},
	})
	if err == nil {
		t.Fatal("expected error on 503, got nil")
	}
}

func TestSender_PostBatch_MalformedResponse(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte(`not json`))
	}))
	defer ts.Close()

	sender := NewSender(ts.URL, http.DefaultClient)
	_, err := sender.PostBatch(context.Background(), []wire.TurnEvent{
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t1", Seq: 1, Role: "user", Timestamp: 1000, Content: "hello"},
	})
	if err == nil {
		t.Fatal("expected error on malformed response, got nil")
	}
}

func TestSender_PostBatch_PartialAccept(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte(`{"accepted":1,"errors":["bad row"]}`))
	}))
	defer ts.Close()

	sender := NewSender(ts.URL, http.DefaultClient)
	result, err := sender.PostBatch(context.Background(), []wire.TurnEvent{
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t1", Seq: 1, Role: "user", Timestamp: 1000, Content: "hello"},
		{Tool: "claude-code", Host: "laptop", SessionID: "s1", TurnID: "t2", Seq: 2, Role: "assistant", Timestamp: 1001, Content: "world"},
	})
	if err != nil {
		t.Fatalf("PostBatch: %v", err)
	}
	if result.Accepted != 1 {
		t.Errorf("Accepted = %d, want 1", result.Accepted)
	}
	if len(result.Errors) != 1 || result.Errors[0] != "bad row" {
		t.Errorf("Errors = %v, want [bad row]", result.Errors)
	}
}