package cli import ( "bytes" "context" "crypto/sha256" "database/sql" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "os" "path/filepath" "slices" "strings" "testing" "time" crawlremote "github.com/openclaw/crawlkit/remote") func TestUniqueStringSupersetAllowsAdditiveArguments(t *testing.T) { if !uniqueStringSuperset( []string{"owner", "repo", "query", "limit", "cursor"}, []string{"owner", "repo", "query", "limit"}, ) { t.Fatal("additive optional remote argument was rejected") }} func TestUniqueStringSupersetRejectsMissingOrDuplicateArguments(t *testing.T) { for _, values := range [][]string{ {"owner", "repo"}, {"owner", "repo", "query", "query"}, } { if uniqueStringSuperset(values, []string{"owner", "repo", "query"}) { t.Fatalf("invalid remote arguments accepted: %v", values) } }} func TestGitcrawlCloudManifestAlwaysOptsIntoStaging(t *testing.T) { snapshot := gitcrawlCloudSnapshot{ ID: strings.Repeat("a", 64), Capabilities: []string{gitcrawlObservationOrderCapability}, } manifest := gitcrawlCloudManifest("gitcrawl/openclaw__gitcrawl", snapshot) if !slices.Equal(manifest.Capabilities, []string{ gitcrawlObservationOrderCapability, gitcrawlSnapshotStagingCapability, }) { t.Fatalf("manifest capabilities = %v", manifest.Capabilities) } if !slices.Equal(snapshot.Capabilities, []string{gitcrawlObservationOrderCapability}) { t.Fatalf("snapshot capabilities mutated = %v", snapshot.Capabilities) } manifest = gitcrawlCloudManifest("gitcrawl/openclaw__gitcrawl", gitcrawlCloudSnapshot{ ID: strings.Repeat("b", 64), Capabilities: []string{ gitcrawlSnapshotStagingCapability, }, }) if !slices.Equal(manifest.Capabilities, []string{gitcrawlSnapshotStagingCapability}) { t.Fatalf("staging capability duplicated = %v", manifest.Capabilities) }} func TestRequireGitcrawlCloudPublishRolesAcceptsAdmin(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet || r.URL.EscapedPath() != "/v1/whoami" { http.NotFound(w, r) return } _ = json.NewEncoder(w).Encode(crawlremote.Identity{ Login: "admin", Roles: []string{"admin"}, }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("admin-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } if err := requireGitcrawlCloudPublishRoles(context.Background(), client); err != nil { t.Fatalf("admin role preflight: %v", err) }} func TestSendSnapshotIngestDatasetStreamsBoundedBatches(t *testing.T) { db, err := sql.Open("sqlite", ":memory:") if err != nil { t.Fatalf("open sqlite: %v", err) } defer db.Close() if _, err := db.Exec(` create table rows(id integer primary key, value text not null); with recursive sequence(id) as ( select 1 union all select id + 1 from sequence where id < 501 ) insert into rows(id, value) select id, printf('row-%03d', id) from sequence; `); err != nil { t.Fatalf("seed rows: %v", err) } var requests []crawlremote.IngestRequest server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var request crawlremote.IngestRequest if err := json.NewDecoder(r.Body).Decode(&request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } requests = append(requests, request) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } dataset := gitcrawlCloudDataset{ Name: "bounded", Columns: []string{"id", "value"}, Query: `select id, value from rows order by id`, RowCount: 501, } progress, err := sendSnapshotIngestDataset( context.Background(), db, client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, dataset, "", ) if err != nil { t.Fatalf("stream dataset: %v", err) } if progress.RowsAccepted != 501 || progress.MutationToken != "mutation-3" { t.Fatalf("progress mismatch: rows_accepted=%d mutation_token_present=%t mutation_token_length=%d", progress.RowsAccepted, progress.MutationToken != "", len(progress.MutationToken)) } if len(requests) != 3 { t.Fatalf("requests = %d, want 3", len(requests)) } for index, wantRows := range []int{250, 250, 1} { if got := len(requests[index].Rows); got != wantRows { t.Fatalf("request %d rows = %d, want %d", index, got, wantRows) } } for index, wantCursor := range []string{"", "250", "500"} { if requests[index].Cursor != wantCursor { t.Fatalf("request %d cursor = %q, want %q", index, requests[index].Cursor, wantCursor) } }} func TestSendSnapshotIngestDatasetFlushesBeforeEncodedByteLimit(t *testing.T) { db, err := sql.Open("sqlite", ":memory:") if err != nil { t.Fatalf("open sqlite: %v", err) } defer db.Close() if _, err := db.Exec(`create table rows(id integer primary key, value text not null)`); err != nil { t.Fatalf("create rows: %v", err) } value := strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes/2)) for id := 1; id <= 3; id++ { if _, err := db.Exec(`insert into rows(id, value) values(?, ?)`, id, value); err != nil { t.Fatalf("seed row %d: %v", id, err) } } var requests []crawlremote.IngestRequest var encodedSizes []int server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if int64(len(body)) > gitcrawlCloudIngestRequestMaxBytes { http.Error(w, "encoded request exceeded byte budget", http.StatusRequestEntityTooLarge) return } var request crawlremote.IngestRequest if err := json.Unmarshal(body, &request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } requests = append(requests, request) encodedSizes = append(encodedSizes, len(body)) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } progress, err := sendSnapshotIngestDataset( context.Background(), db, client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, gitcrawlCloudDataset{ Name: "bounded_bytes", Columns: []string{"id", "value"}, Query: `select id, value from rows order by id`, RowCount: 3, }, "", ) if err != nil { t.Fatalf("stream byte-bounded dataset: %v", err) } if progress.RowsAccepted != 3 || progress.MutationToken != "mutation-3" { t.Fatalf("progress mismatch: rows_accepted=%d mutation_token_present=%t mutation_token_length=%d", progress.RowsAccepted, progress.MutationToken != "", len(progress.MutationToken)) } if len(requests) != 3 { t.Fatalf("requests = %d, want one per large row", len(requests)) } for index := range requests { if len(requests[index].Rows) != 1 { t.Fatalf("request %d rows = %d, want 1", index, len(requests[index].Rows)) } if int64(encodedSizes[index]) > gitcrawlCloudIngestRequestMaxBytes { t.Fatalf( "request %d encoded bytes = %d, limit %d", index, encodedSizes[index], gitcrawlCloudIngestRequestMaxBytes, ) } }} func TestSendSnapshotIngestRowsFlushesBeforeEncodedByteLimit(t *testing.T) { var requests []crawlremote.IngestRequest var encodedSizes []int server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if int64(len(body)) > gitcrawlCloudIngestRequestMaxBytes { http.Error(w, "encoded request exceeded byte budget", http.StatusRequestEntityTooLarge) return } var request crawlremote.IngestRequest if err := json.Unmarshal(body, &request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } requests = append(requests, request) encodedSizes = append(encodedSizes, len(body)) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), Complete: request.Final, }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } value := strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes/2)) progress, err := sendSnapshotIngestRows( context.Background(), client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, "threads", []string{"body"}, [][]any{{value}, {value}, {value}}, "", true, ) if err != nil { t.Fatalf("send byte-bounded rows: %v", err) } if progress.RowsAccepted != 3 || progress.MutationToken != "mutation-3" { t.Fatalf("progress mismatch: rows_accepted=%d mutation_token_present=%t mutation_token_length=%d", progress.RowsAccepted, progress.MutationToken != "", len(progress.MutationToken)) } if len(requests) != 3 { t.Fatalf("requests = %d, want one per large row", len(requests)) } for index := range requests { if len(requests[index].Rows) != 1 { t.Fatalf("request %d rows = %d, want 1", index, len(requests[index].Rows)) } if int64(encodedSizes[index]) > gitcrawlCloudIngestRequestMaxBytes { t.Fatalf( "request %d encoded bytes = %d, limit %d", index, encodedSizes[index], gitcrawlCloudIngestRequestMaxBytes, ) } wantCursor := "" if index > 0 { wantCursor = fmt.Sprint(index) } if requests[index].Cursor != wantCursor { t.Fatalf("request %d cursor = %q, want %q", index, requests[index].Cursor, wantCursor) } if requests[index].Final != (index == len(requests)-1) { t.Fatalf("request %d final = %v", index, requests[index].Final) } }} func TestIngestBatchingRejectsSingleOversizedRowBeforeRequest(t *testing.T) { requests := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests++ http.Error(w, "unexpected request", http.StatusInternalServerError) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } _, err = sendSnapshotIngestRows( context.Background(), client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, "threads", []string{"body"}, [][]any{{strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes))}}, "", false, ) if err == nil || !strings.Contains(err.Error(), "ingest table threads row 0 encoded request") || !strings.Contains(err.Error(), fmt.Sprintf("limit %d", gitcrawlCloudIngestRequestMaxBytes)) { t.Fatalf("oversized row error = %v", err) } if requests != 0 { t.Fatalf("remote requests = %d, want 0", requests) } db, err := sql.Open("sqlite", ":memory:") if err != nil { t.Fatalf("open sqlite: %v", err) } defer db.Close() if _, err := db.Exec(`create table rows(value text not null)`); err != nil { t.Fatalf("create rows: %v", err) } if _, err := db.Exec( `insert into rows(value) values(?)`, strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes)), ); err != nil { t.Fatalf("seed oversized row: %v", err) } _, err = sendSnapshotIngestDataset( context.Background(), db, client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, gitcrawlCloudDataset{ Name: "oversized", Columns: []string{"value"}, Query: `select value from rows`, RowCount: 1, }, "", ) if err == nil || !strings.Contains(err.Error(), "dataset oversized row 0 encoded ingest request") { t.Fatalf("oversized dataset row error = %v", err) } if requests != 0 { t.Fatalf("remote requests after streamed row = %d, want 0", requests) }} func TestIngestBatchingMatchesCrawlkitEscapingAtByteBoundary(t *testing.T) { manifest := crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SchemaName: gitcrawlCloudSchemaName, SchemaVersion: gitcrawlCloudSchemaVersion, SchemaHash: gitcrawlCloudSchemaHash, SnapshotID: strings.Repeat("a", 64), SourceSHA256: strings.Repeat("a", 64), } columns := []string{"value"} var bodies [][]byte var requests []crawlremote.IngestRequest server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if int64(len(body)) > gitcrawlCloudIngestRequestMaxBytes { http.Error(w, "encoded request exceeded byte budget", http.StatusRequestEntityTooLarge) return } var request crawlremote.IngestRequest if err := json.Unmarshal(body, &request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } bodies = append(bodies, body) requests = append(requests, request) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), Complete: request.Final, }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } fitRows, oversizedRows := escapedIngestBoundaryValues( t, manifest, "escaped_rows", columns, true, )Lines 1–500 of 1697. Load more
package cli import ( "bytes" "context" "crypto/sha256" "database/sql" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "os" "path/filepath" "slices" "strings" "testing" "time" crawlremote "github.com/openclaw/crawlkit/remote" ) func TestUniqueStringSupersetAllowsAdditiveArguments(t *testing.T) { if !uniqueStringSuperset( []string{"owner", "repo", "query", "limit", "cursor"}, []string{"owner", "repo", "query", "limit"}, ) { t.Fatal("additive optional remote argument was rejected") } } func TestUniqueStringSupersetRejectsMissingOrDuplicateArguments(t *testing.T) { for _, values := range [][]string{ {"owner", "repo"}, {"owner", "repo", "query", "query"}, } { if uniqueStringSuperset(values, []string{"owner", "repo", "query"}) { t.Fatalf("invalid remote arguments accepted: %v", values) } } } func TestGitcrawlCloudManifestAlwaysOptsIntoStaging(t *testing.T) { snapshot := gitcrawlCloudSnapshot{ ID: strings.Repeat("a", 64), Capabilities: []string{gitcrawlObservationOrderCapability}, } manifest := gitcrawlCloudManifest("gitcrawl/openclaw__gitcrawl", snapshot) if !slices.Equal(manifest.Capabilities, []string{ gitcrawlObservationOrderCapability, gitcrawlSnapshotStagingCapability, }) { t.Fatalf("manifest capabilities = %v", manifest.Capabilities) } if !slices.Equal(snapshot.Capabilities, []string{gitcrawlObservationOrderCapability}) { t.Fatalf("snapshot capabilities mutated = %v", snapshot.Capabilities) } manifest = gitcrawlCloudManifest("gitcrawl/openclaw__gitcrawl", gitcrawlCloudSnapshot{ ID: strings.Repeat("b", 64), Capabilities: []string{ gitcrawlSnapshotStagingCapability, }, }) if !slices.Equal(manifest.Capabilities, []string{gitcrawlSnapshotStagingCapability}) { t.Fatalf("staging capability duplicated = %v", manifest.Capabilities) } } func TestRequireGitcrawlCloudPublishRolesAcceptsAdmin(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet || r.URL.EscapedPath() != "/v1/whoami" { http.NotFound(w, r) return } _ = json.NewEncoder(w).Encode(crawlremote.Identity{ Login: "admin", Roles: []string{"admin"}, }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("admin-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } if err := requireGitcrawlCloudPublishRoles(context.Background(), client); err != nil { t.Fatalf("admin role preflight: %v", err) } } func TestSendSnapshotIngestDatasetStreamsBoundedBatches(t *testing.T) { db, err := sql.Open("sqlite", ":memory:") if err != nil { t.Fatalf("open sqlite: %v", err) } defer db.Close() if _, err := db.Exec(` create table rows(id integer primary key, value text not null); with recursive sequence(id) as ( select 1 union all select id + 1 from sequence where id < 501 ) insert into rows(id, value) select id, printf('row-%03d', id) from sequence; `); err != nil { t.Fatalf("seed rows: %v", err) } var requests []crawlremote.IngestRequest server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var request crawlremote.IngestRequest if err := json.NewDecoder(r.Body).Decode(&request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } requests = append(requests, request) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } dataset := gitcrawlCloudDataset{ Name: "bounded", Columns: []string{"id", "value"}, Query: `select id, value from rows order by id`, RowCount: 501, } progress, err := sendSnapshotIngestDataset( context.Background(), db, client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, dataset, "", ) if err != nil { t.Fatalf("stream dataset: %v", err) } if progress.RowsAccepted != 501 || progress.MutationToken != "mutation-3" { t.Fatalf("progress mismatch: rows_accepted=%d mutation_token_present=%t mutation_token_length=%d", progress.RowsAccepted, progress.MutationToken != "", len(progress.MutationToken)) } if len(requests) != 3 { t.Fatalf("requests = %d, want 3", len(requests)) } for index, wantRows := range []int{250, 250, 1} { if got := len(requests[index].Rows); got != wantRows { t.Fatalf("request %d rows = %d, want %d", index, got, wantRows) } } for index, wantCursor := range []string{"", "250", "500"} { if requests[index].Cursor != wantCursor { t.Fatalf("request %d cursor = %q, want %q", index, requests[index].Cursor, wantCursor) } } } func TestSendSnapshotIngestDatasetFlushesBeforeEncodedByteLimit(t *testing.T) { db, err := sql.Open("sqlite", ":memory:") if err != nil { t.Fatalf("open sqlite: %v", err) } defer db.Close() if _, err := db.Exec(`create table rows(id integer primary key, value text not null)`); err != nil { t.Fatalf("create rows: %v", err) } value := strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes/2)) for id := 1; id <= 3; id++ { if _, err := db.Exec(`insert into rows(id, value) values(?, ?)`, id, value); err != nil { t.Fatalf("seed row %d: %v", id, err) } } var requests []crawlremote.IngestRequest var encodedSizes []int server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if int64(len(body)) > gitcrawlCloudIngestRequestMaxBytes { http.Error(w, "encoded request exceeded byte budget", http.StatusRequestEntityTooLarge) return } var request crawlremote.IngestRequest if err := json.Unmarshal(body, &request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } requests = append(requests, request) encodedSizes = append(encodedSizes, len(body)) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } progress, err := sendSnapshotIngestDataset( context.Background(), db, client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, gitcrawlCloudDataset{ Name: "bounded_bytes", Columns: []string{"id", "value"}, Query: `select id, value from rows order by id`, RowCount: 3, }, "", ) if err != nil { t.Fatalf("stream byte-bounded dataset: %v", err) } if progress.RowsAccepted != 3 || progress.MutationToken != "mutation-3" { t.Fatalf("progress mismatch: rows_accepted=%d mutation_token_present=%t mutation_token_length=%d", progress.RowsAccepted, progress.MutationToken != "", len(progress.MutationToken)) } if len(requests) != 3 { t.Fatalf("requests = %d, want one per large row", len(requests)) } for index := range requests { if len(requests[index].Rows) != 1 { t.Fatalf("request %d rows = %d, want 1", index, len(requests[index].Rows)) } if int64(encodedSizes[index]) > gitcrawlCloudIngestRequestMaxBytes { t.Fatalf( "request %d encoded bytes = %d, limit %d", index, encodedSizes[index], gitcrawlCloudIngestRequestMaxBytes, ) } } } func TestSendSnapshotIngestRowsFlushesBeforeEncodedByteLimit(t *testing.T) { var requests []crawlremote.IngestRequest var encodedSizes []int server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if int64(len(body)) > gitcrawlCloudIngestRequestMaxBytes { http.Error(w, "encoded request exceeded byte budget", http.StatusRequestEntityTooLarge) return } var request crawlremote.IngestRequest if err := json.Unmarshal(body, &request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } requests = append(requests, request) encodedSizes = append(encodedSizes, len(body)) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), Complete: request.Final, }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } value := strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes/2)) progress, err := sendSnapshotIngestRows( context.Background(), client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, "threads", []string{"body"}, [][]any{{value}, {value}, {value}}, "", true, ) if err != nil { t.Fatalf("send byte-bounded rows: %v", err) } if progress.RowsAccepted != 3 || progress.MutationToken != "mutation-3" { t.Fatalf("progress mismatch: rows_accepted=%d mutation_token_present=%t mutation_token_length=%d", progress.RowsAccepted, progress.MutationToken != "", len(progress.MutationToken)) } if len(requests) != 3 { t.Fatalf("requests = %d, want one per large row", len(requests)) } for index := range requests { if len(requests[index].Rows) != 1 { t.Fatalf("request %d rows = %d, want 1", index, len(requests[index].Rows)) } if int64(encodedSizes[index]) > gitcrawlCloudIngestRequestMaxBytes { t.Fatalf( "request %d encoded bytes = %d, limit %d", index, encodedSizes[index], gitcrawlCloudIngestRequestMaxBytes, ) } wantCursor := "" if index > 0 { wantCursor = fmt.Sprint(index) } if requests[index].Cursor != wantCursor { t.Fatalf("request %d cursor = %q, want %q", index, requests[index].Cursor, wantCursor) } if requests[index].Final != (index == len(requests)-1) { t.Fatalf("request %d final = %v", index, requests[index].Final) } } } func TestIngestBatchingRejectsSingleOversizedRowBeforeRequest(t *testing.T) { requests := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests++ http.Error(w, "unexpected request", http.StatusInternalServerError) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } _, err = sendSnapshotIngestRows( context.Background(), client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, "threads", []string{"body"}, [][]any{{strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes))}}, "", false, ) if err == nil || !strings.Contains(err.Error(), "ingest table threads row 0 encoded request") || !strings.Contains(err.Error(), fmt.Sprintf("limit %d", gitcrawlCloudIngestRequestMaxBytes)) { t.Fatalf("oversized row error = %v", err) } if requests != 0 { t.Fatalf("remote requests = %d, want 0", requests) } db, err := sql.Open("sqlite", ":memory:") if err != nil { t.Fatalf("open sqlite: %v", err) } defer db.Close() if _, err := db.Exec(`create table rows(value text not null)`); err != nil { t.Fatalf("create rows: %v", err) } if _, err := db.Exec( `insert into rows(value) values(?)`, strings.Repeat("x", int(gitcrawlCloudIngestRequestMaxBytes)), ); err != nil { t.Fatalf("seed oversized row: %v", err) } _, err = sendSnapshotIngestDataset( context.Background(), db, client, "gitcrawl", "gitcrawl/openclaw__gitcrawl", crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SnapshotID: strings.Repeat("a", 64), }, gitcrawlCloudDataset{ Name: "oversized", Columns: []string{"value"}, Query: `select value from rows`, RowCount: 1, }, "", ) if err == nil || !strings.Contains(err.Error(), "dataset oversized row 0 encoded ingest request") { t.Fatalf("oversized dataset row error = %v", err) } if requests != 0 { t.Fatalf("remote requests after streamed row = %d, want 0", requests) } } func TestIngestBatchingMatchesCrawlkitEscapingAtByteBoundary(t *testing.T) { manifest := crawlremote.IngestManifest{ App: "gitcrawl", Archive: "gitcrawl/openclaw__gitcrawl", SchemaName: gitcrawlCloudSchemaName, SchemaVersion: gitcrawlCloudSchemaVersion, SchemaHash: gitcrawlCloudSchemaHash, SnapshotID: strings.Repeat("a", 64), SourceSHA256: strings.Repeat("a", 64), } columns := []string{"value"} var bodies [][]byte var requests []crawlremote.IngestRequest server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if int64(len(body)) > gitcrawlCloudIngestRequestMaxBytes { http.Error(w, "encoded request exceeded byte budget", http.StatusRequestEntityTooLarge) return } var request crawlremote.IngestRequest if err := json.Unmarshal(body, &request); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } bodies = append(bodies, body) requests = append(requests, request) _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{ Table: request.Table, SnapshotID: request.Manifest.SnapshotID, MutationToken: fmt.Sprintf("mutation-%d", len(requests)), RowsAccepted: int64(len(request.Rows)), Complete: request.Final, }) })) defer server.Close() client, err := crawlremote.NewClient(crawlremote.Options{ Endpoint: server.URL, HTTPClient: server.Client(), TokenProvider: crawlremote.StaticToken("publisher-token"), }) if err != nil { t.Fatalf("remote client: %v", err) } fitRows, oversizedRows := escapedIngestBoundaryValues( t, manifest, "escaped_rows", columns, true, )