|
| 1 | +package e2e |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "net/http" |
| 6 | + "os" |
| 7 | + "os/exec" |
| 8 | + "testing" |
| 9 | + "time" |
| 10 | + |
| 11 | + "github.com/s2-streamstore/s2-sdk-go/s2" |
| 12 | + "github.com/stretchr/testify/assert" |
| 13 | + "github.com/stretchr/testify/require" |
| 14 | + |
| 15 | + instanceoapi "github.com/kernel/kernel-images/server/lib/oapi" |
| 16 | +) |
| 17 | + |
| 18 | +// TestS2StorageWriter starts a headless container with S2 credentials, runs a |
| 19 | +// capture session, and verifies that events land in the configured S2 stream. |
| 20 | +// |
| 21 | +// Skips automatically when S2_BASIN, S2_ACCESS_TOKEN, or S2_STREAM are unset. |
| 22 | +func TestS2StorageWriter(t *testing.T) { |
| 23 | + basin := os.Getenv("S2_BASIN") |
| 24 | + accessToken := os.Getenv("S2_ACCESS_TOKEN") |
| 25 | + stream := os.Getenv("S2_STREAM") |
| 26 | + if basin == "" || accessToken == "" || stream == "" { |
| 27 | + t.Skip("S2_BASIN, S2_ACCESS_TOKEN, and S2_STREAM must be set to run this test") |
| 28 | + } |
| 29 | + |
| 30 | + if _, err := exec.LookPath("docker"); err != nil { |
| 31 | + t.Skipf("docker not available: %v", err) |
| 32 | + } |
| 33 | + |
| 34 | + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) |
| 35 | + defer cancel() |
| 36 | + |
| 37 | + c := NewTestContainer(t, headlessImage) |
| 38 | + require.NoError(t, c.Start(ctx, ContainerConfig{ |
| 39 | + Env: map[string]string{ |
| 40 | + "S2_BASIN": basin, |
| 41 | + "S2_ACCESS_TOKEN": accessToken, |
| 42 | + "S2_STREAM": stream, |
| 43 | + }, |
| 44 | + }), "failed to start container") |
| 45 | + defer c.Stop(ctx) |
| 46 | + |
| 47 | + require.NoError(t, c.WaitReady(ctx), "api not ready") |
| 48 | + |
| 49 | + client, err := c.APIClient() |
| 50 | + require.NoError(t, err) |
| 51 | + |
| 52 | + // Note the current S2 stream tail seq before we write anything so we only |
| 53 | + // read records produced by this test run. |
| 54 | + s2Client := s2.New(accessToken, nil) |
| 55 | + streamClient := s2Client.Basin(basin).Stream(s2.StreamName(stream)) |
| 56 | + |
| 57 | + checkResp, err := streamClient.CheckTail(ctx) |
| 58 | + require.NoError(t, err, "check tail before test") |
| 59 | + startSeq := checkResp.Tail.SeqNum |
| 60 | + |
| 61 | + // Start a capture session. |
| 62 | + startResp, err := client.StartCaptureSessionWithResponse(ctx, instanceoapi.StartCaptureSessionJSONRequestBody{}) |
| 63 | + require.NoError(t, err) |
| 64 | + require.Equal(t, http.StatusCreated, startResp.StatusCode(), "start capture session: %s", string(startResp.Body)) |
| 65 | + require.NotNil(t, startResp.JSON201) |
| 66 | + sessionID := startResp.JSON201.Id |
| 67 | + t.Logf("capture session started: %s", sessionID) |
| 68 | + |
| 69 | + // Let the session run briefly so at least one event is published (the |
| 70 | + // session_started system event is emitted on session start). |
| 71 | + time.Sleep(500 * time.Millisecond) |
| 72 | + |
| 73 | + // Stop the capture session. |
| 74 | + stopResp, err := client.StopCaptureSessionWithResponse(ctx) |
| 75 | + require.NoError(t, err) |
| 76 | + require.Equal(t, http.StatusOK, stopResp.StatusCode(), "stop capture session: %s", string(stopResp.Body)) |
| 77 | + t.Log("capture session stopped") |
| 78 | + |
| 79 | + // Give the storage writer time to flush to S2 (batcher linger + network). |
| 80 | + time.Sleep(2 * time.Second) |
| 81 | + |
| 82 | + // Read records written after the pre-test tail and verify at least one |
| 83 | + // envelope is present. |
| 84 | + readCtx, readCancel := context.WithTimeout(ctx, 10*time.Second) |
| 85 | + defer readCancel() |
| 86 | + |
| 87 | + readSession, err := streamClient.ReadSession(readCtx, &s2.ReadOptions{ |
| 88 | + SeqNum: s2.Uint64(startSeq), |
| 89 | + }) |
| 90 | + require.NoError(t, err, "open S2 read session") |
| 91 | + defer readSession.Close() |
| 92 | + |
| 93 | + var count int |
| 94 | + for readSession.Next() { |
| 95 | + count++ |
| 96 | + } |
| 97 | + // EOF is expected once we reach the tail — not an error. |
| 98 | + if err := readSession.Err(); err != nil && readCtx.Err() == nil { |
| 99 | + t.Fatalf("S2 read session error: %v", err) |
| 100 | + } |
| 101 | + |
| 102 | + assert.Greater(t, count, 0, "expected at least one event record in S2 stream %q", stream) |
| 103 | + t.Logf("found %d record(s) in S2 stream after seq %d", count, startSeq) |
| 104 | +} |
0 commit comments