diff --git a/README.md b/README.md index 638276a8..0cc91dc1 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ $ source <(force completion bash) apiversion Display/Set current API version bigobject Manage big objects bulk Load csv file or query data using Bulk API + bulk2 Load csv file or query data using Bulk API 2.0 completion Generate the autocompletion script for the specified shell create Creates a new, empty Apex Class, Trigger, Visualforce page, or Component. datapipe Manage DataPipes @@ -176,6 +177,51 @@ Export allows you to fetch all codes from your org to local machine. This comman Includes notification library, [beeep](https://github.com/gen2brain/beeep), that displays desktop notifications across platforms. Currently, only the `push` and `test` methods are displaying notifications. +### bulk2 +Bulk API 2.0 provides a REST-based interface for data loading and querying with automatic batch management. Unlike Bulk API 1.0, it handles batch splitting automatically and uses CSV format. + + # Insert records + force bulk2 insert Account accounts.csv --wait + + # Update records + force bulk2 update Account updates.csv --wait + + # Upsert records using external ID + force bulk2 upsert -e External_Id__c Account data.csv --wait + + # Delete records + force bulk2 delete Account deletes.csv --wait + + # Hard delete records + force bulk2 hardDelete Account harddeletes.csv --wait + + # Query records + force bulk2 query "SELECT Id, Name FROM Account" --wait + + # Query including deleted and archived records + force bulk2 query "SELECT Id, Name FROM Account" --query-all --wait + + # Check job status + force bulk2 job + + # List all ingest jobs + force bulk2 jobs + + # List all query jobs + force bulk2 jobs --query + + # Get job results + force bulk2 results + + # Get only failed results + force bulk2 results --failed + + # Abort a job + force bulk2 abort + + # Delete a job + force bulk2 delete-job + ### limits Limits will display limits information for your organization. - Max is the limit total for the organization diff --git a/bubbles/bulk2job.go b/bubbles/bulk2job.go new file mode 100644 index 00000000..7b3ab681 --- /dev/null +++ b/bubbles/bulk2job.go @@ -0,0 +1,113 @@ +package bubbles + +import ( + "fmt" + + force "github.com/ForceCLI/force/lib" + "github.com/charmbracelet/bubbles/progress" + tea "github.com/charmbracelet/bubbletea" + "github.com/charmbracelet/lipgloss" +) + +var ( + bulk2InfoStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("#4E4E4E")).TabWidth(lipgloss.NoTabConversion) + bulk2StatusStyle = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("#89D5C9")).TabWidth(lipgloss.NoTabConversion) +) + +type Bulk2JobModel struct { + force.Bulk2IngestJobInfo + progress progress.Model +} + +type NewBulk2JobStatusMsg struct { + force.Bulk2IngestJobInfo +} + +func NewBulk2JobModel() Bulk2JobModel { + return Bulk2JobModel{ + progress: progress.New(progress.WithDefaultGradient()), + } +} + +func (m Bulk2JobModel) Init() tea.Cmd { + return nil +} + +func (m Bulk2JobModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { + switch msg := msg.(type) { + case tea.WindowSizeMsg: + m.progress.Width = min(msg.Width-padding*2-4, maxWidth) + return m, nil + + case NewBulk2JobStatusMsg: + m.Bulk2IngestJobInfo = msg.Bulk2IngestJobInfo + var cmd tea.Cmd + if m.IsTerminal() { + cmd = m.progress.SetPercent(1.0) + } else if m.State == force.Bulk2JobStateInProgress { + cmd = m.progress.SetPercent(0.5) + } else { + cmd = m.progress.SetPercent(0.1) + } + return m, cmd + + case tea.KeyMsg: + switch msg.String() { + case "q", "ctrl+c": + return m, tea.Quit + } + + case progress.FrameMsg: + progressModel, cmd := m.progress.Update(msg) + m.progress = progressModel.(progress.Model) + return m, cmd + + case QuitMsg: + return m, tea.Quit + } + return m, nil +} + +func (m Bulk2JobModel) View() string { + header := headerStyle.Render("Bulk API 2.0 Job Status") + + var infoMsg = ` +Id %s +State %s +Operation %s +Object %s +Api Version %.1f + +Created By Id %s +Created Date %s +Content Type %s +` + + var statusMsg = ` +Number Records Processed %d +Number Records Failed %d +Retries %d + +Total Processing Time %d ms +Api Active Processing Time %d ms +Apex Processing Time %d ms +` + + components := []string{ + lipgloss.JoinVertical(lipgloss.Top, header, + bulk2InfoStyle.Render(fmt.Sprintf(infoMsg, + m.Id, m.State, m.Operation, m.Object, m.ApiVersion, + m.CreatedById, m.CreatedDate, m.ContentType)), + m.progress.View(), + bulk2StatusStyle.Render(fmt.Sprintf(statusMsg, + m.NumberRecordsProcessed, m.NumberRecordsFailed, m.Retries, + m.TotalProcessingTime, m.ApiActiveProcessingTime, m.ApexProcessingTime))), + } + + if m.ErrorMessage != "" { + errorMsg := failureStyle.Render(fmt.Sprintf("Error: %s", m.ErrorMessage)) + components = append(components, errorMsg) + } + + return lipgloss.JoinVertical(lipgloss.Top, components...) +} diff --git a/command/bulk2.go b/command/bulk2.go new file mode 100644 index 00000000..018027c9 --- /dev/null +++ b/command/bulk2.go @@ -0,0 +1,469 @@ +package command + +import ( + "fmt" + "os" + "strings" + "time" + + "github.com/ForceCLI/force/bubbles" + . "github.com/ForceCLI/force/error" + . "github.com/ForceCLI/force/lib" + tea "github.com/charmbracelet/bubbletea" + "github.com/spf13/cobra" +) + +func init() { + // Data loading commands + ingestCmds := []*cobra.Command{bulk2InsertCmd, bulk2UpdateCmd, bulk2UpsertCmd, bulk2DeleteCmd, bulk2HardDeleteCmd} + for _, cmd := range ingestCmds { + cmd.Flags().BoolP("wait", "w", false, "Wait for job to complete") + cmd.Flags().BoolP("interactive", "i", false, "Interactive mode (implies --wait)") + cmd.Flags().String("delimiter", "COMMA", "Column delimiter (COMMA, TAB, PIPE, SEMICOLON, CARET, BACKQUOTE)") + cmd.Flags().String("lineending", "LF", "Line ending (LF or CRLF)") + } + + bulk2UpsertCmd.Flags().StringP("externalid", "e", "", "External ID field for upserting (required)") + bulk2UpsertCmd.MarkFlagRequired("externalid") + + // Query command flags + bulk2QueryCmd.Flags().BoolP("wait", "w", false, "Wait for job to complete") + bulk2QueryCmd.Flags().BoolP("query-all", "A", false, "Include deleted and archived records") + bulk2QueryCmd.Flags().String("delimiter", "COMMA", "Column delimiter for results (COMMA, TAB, PIPE, SEMICOLON, CARET, BACKQUOTE)") + bulk2QueryCmd.Flags().String("lineending", "LF", "Line ending for results (LF or CRLF)") + + // Results command flags + bulk2ResultsCmd.Flags().BoolP("successful", "s", false, "Show successful results only") + bulk2ResultsCmd.Flags().BoolP("failed", "f", false, "Show failed results only") + bulk2ResultsCmd.Flags().BoolP("unprocessed", "u", false, "Show unprocessed records only") + + // Jobs list command flags + bulk2JobsCmd.Flags().BoolP("query", "q", false, "List query jobs instead of ingest jobs") + + // Add subcommands + bulk2Cmd.AddCommand(bulk2InsertCmd) + bulk2Cmd.AddCommand(bulk2UpdateCmd) + bulk2Cmd.AddCommand(bulk2UpsertCmd) + bulk2Cmd.AddCommand(bulk2DeleteCmd) + bulk2Cmd.AddCommand(bulk2HardDeleteCmd) + bulk2Cmd.AddCommand(bulk2QueryCmd) + bulk2Cmd.AddCommand(bulk2JobCmd) + bulk2Cmd.AddCommand(bulk2JobsCmd) + bulk2Cmd.AddCommand(bulk2ResultsCmd) + bulk2Cmd.AddCommand(bulk2AbortCmd) + bulk2Cmd.AddCommand(bulk2DeleteJobCmd) + + RootCmd.AddCommand(bulk2Cmd) +} + +var bulk2Cmd = &cobra.Command{ + Use: "bulk2", + Short: "Use Bulk API 2.0 for data loading and querying", + Long: "Bulk API 2.0 provides a REST-based interface for data loading and querying with automatic batch management.", + Example: ` + force bulk2 insert Account accounts.csv --wait + force bulk2 update Account updates.csv --wait + force bulk2 upsert -e External_Id__c Account data.csv --wait + force bulk2 delete Account deletes.csv --wait + force bulk2 query "SELECT Id, Name FROM Account LIMIT 100" --wait + force bulk2 job + force bulk2 jobs + force bulk2 jobs --query + force bulk2 results + force bulk2 results --failed + force bulk2 abort + force bulk2 delete-job +`, +} + +var bulk2InsertCmd = &cobra.Command{ + Use: "insert ", + Short: "Insert records from CSV file using Bulk API 2.0", + Args: cobra.ExactArgs(2), + Run: runBulk2IngestCmd, +} + +var bulk2UpdateCmd = &cobra.Command{ + Use: "update ", + Short: "Update records from CSV file using Bulk API 2.0", + Args: cobra.ExactArgs(2), + Run: runBulk2IngestCmd, +} + +var bulk2UpsertCmd = &cobra.Command{ + Use: "upsert -e ", + Short: "Upsert records from CSV file using Bulk API 2.0", + Args: cobra.ExactArgs(2), + Run: runBulk2IngestCmd, +} + +var bulk2DeleteCmd = &cobra.Command{ + Use: "delete ", + Short: "Delete records using Bulk API 2.0", + Args: cobra.ExactArgs(2), + Run: runBulk2IngestCmd, +} + +var bulk2HardDeleteCmd = &cobra.Command{ + Use: "hardDelete ", + Short: "Hard delete records using Bulk API 2.0", + Args: cobra.ExactArgs(2), + Run: runBulk2IngestCmd, +} + +var bulk2QueryCmd = &cobra.Command{ + Use: "query ", + Short: "Query records using Bulk API 2.0", + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { + soql := args[0] + wait, _ := cmd.Flags().GetBool("wait") + queryAll, _ := cmd.Flags().GetBool("query-all") + delimiter, _ := cmd.Flags().GetString("delimiter") + lineEnding, _ := cmd.Flags().GetString("lineending") + + operation := Bulk2OperationQuery + if queryAll { + operation = Bulk2OperationQueryAll + } + + request := Bulk2QueryJobRequest{ + Operation: operation, + Query: soql, + ColumnDelimiter: Bulk2ColumnDelimiter(strings.ToUpper(delimiter)), + LineEnding: Bulk2LineEnding(strings.ToUpper(lineEnding)), + } + + jobInfo, err := force.CreateBulk2QueryJob(request) + if err != nil { + ErrorAndExit("Failed to create query job: " + err.Error()) + } + + fmt.Fprintf(os.Stderr, "Query job created: %s\n", jobInfo.Id) + + if !wait { + fmt.Fprintf(os.Stderr, "To check job status use:\n force bulk2 job %s\n", jobInfo.Id) + return + } + + finalJobInfo := waitForBulk2QueryJob(jobInfo.Id) + if finalJobInfo.State == Bulk2JobStateFailed { + ErrorAndExit("Query job failed: " + finalJobInfo.ErrorMessage) + } + if finalJobInfo.State == Bulk2JobStateAborted { + ErrorAndExit("Query job was aborted") + } + + displayBulk2QueryResults(finalJobInfo.Id) + }, +} + +var bulk2JobCmd = &cobra.Command{ + Use: "job ", + Short: "Show bulk job details", + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { + jobId := args[0] + + // Try ingest job first + ingestInfo, err := force.GetBulk2IngestJobInfo(jobId) + if err == nil { + DisplayBulk2IngestJobInfo(ingestInfo, os.Stdout) + return + } + + // Try query job + queryInfo, err := force.GetBulk2QueryJobInfo(jobId) + if err == nil { + DisplayBulk2QueryJobInfo(queryInfo, os.Stdout) + return + } + + ErrorAndExit("Job not found: " + jobId) + }, +} + +var bulk2JobsCmd = &cobra.Command{ + Use: "jobs", + Short: "List bulk jobs", + Run: func(cmd *cobra.Command, args []string) { + isQuery, _ := cmd.Flags().GetBool("query") + + if isQuery { + jobs, err := force.GetBulk2QueryJobs() + if err != nil { + ErrorAndExit("Failed to list query jobs: " + err.Error()) + } + DisplayBulk2QueryJobList(jobs, os.Stdout) + } else { + jobs, err := force.GetBulk2IngestJobs() + if err != nil { + ErrorAndExit("Failed to list ingest jobs: " + err.Error()) + } + DisplayBulk2IngestJobList(jobs, os.Stdout) + } + }, +} + +var bulk2ResultsCmd = &cobra.Command{ + Use: "results ", + Short: "Get job results", + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { + jobId := args[0] + successful, _ := cmd.Flags().GetBool("successful") + failed, _ := cmd.Flags().GetBool("failed") + unprocessed, _ := cmd.Flags().GetBool("unprocessed") + + // If no specific flag set, try to determine job type and show appropriate results + if !successful && !failed && !unprocessed { + // Try ingest job - show all results + _, err := force.GetBulk2IngestJobInfo(jobId) + if err == nil { + showBulk2IngestResults(jobId, true, true, true) + return + } + + // Try query job - show query results + _, err = force.GetBulk2QueryJobInfo(jobId) + if err == nil { + displayBulk2QueryResults(jobId) + return + } + + ErrorAndExit("Job not found: " + jobId) + return + } + + // Show specific result types for ingest jobs + showBulk2IngestResults(jobId, successful, failed, unprocessed) + }, +} + +var bulk2AbortCmd = &cobra.Command{ + Use: "abort ", + Short: "Abort a bulk job", + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { + jobId := args[0] + + // Try ingest job first + ingestInfo, err := force.AbortBulk2IngestJob(jobId) + if err == nil { + fmt.Fprintf(os.Stdout, "Job %s aborted\n", ingestInfo.Id) + return + } + + // Try query job + queryInfo, err := force.AbortBulk2QueryJob(jobId) + if err == nil { + fmt.Fprintf(os.Stdout, "Job %s aborted\n", queryInfo.Id) + return + } + + ErrorAndExit("Failed to abort job: " + err.Error()) + }, +} + +var bulk2DeleteJobCmd = &cobra.Command{ + Use: "delete-job ", + Short: "Delete a bulk job", + Args: cobra.ExactArgs(1), + Run: func(cmd *cobra.Command, args []string) { + jobId := args[0] + + // Try ingest job first + err := force.DeleteBulk2IngestJob(jobId) + if err == nil { + fmt.Fprintf(os.Stdout, "Job %s deleted\n", jobId) + return + } + + // Try query job + err = force.DeleteBulk2QueryJob(jobId) + if err == nil { + fmt.Fprintf(os.Stdout, "Job %s deleted\n", jobId) + return + } + + ErrorAndExit("Failed to delete job: " + err.Error()) + }, +} + +func runBulk2IngestCmd(cmd *cobra.Command, args []string) { + objectType := args[0] + filePath := args[1] + + operation := Bulk2Operation(strings.ToLower(cmd.Name())) + externalId := "" + if operation == Bulk2OperationUpsert { + externalId, _ = cmd.Flags().GetString("externalid") + } + + wait, _ := cmd.Flags().GetBool("wait") + interactive, _ := cmd.Flags().GetBool("interactive") + delimiter, _ := cmd.Flags().GetString("delimiter") + lineEnding, _ := cmd.Flags().GetString("lineending") + + if interactive { + wait = true + } + + // Open CSV file + file, err := os.Open(filePath) + if err != nil { + ErrorAndExit("Failed to open file: " + err.Error()) + } + defer file.Close() + + // Create job + request := Bulk2IngestJobRequest{ + Object: objectType, + Operation: operation, + ExternalIdFieldName: externalId, + ColumnDelimiter: Bulk2ColumnDelimiter(strings.ToUpper(delimiter)), + LineEnding: Bulk2LineEnding(strings.ToUpper(lineEnding)), + } + + jobInfo, err := force.CreateBulk2IngestJob(request) + if err != nil { + ErrorAndExit("Failed to create job: " + err.Error()) + } + fmt.Fprintf(os.Stderr, "Job created: %s\n", jobInfo.Id) + + // Upload data + fmt.Fprintf(os.Stderr, "Uploading data...\n") + err = force.UploadBulk2JobData(jobInfo.Id, file) + if err != nil { + force.AbortBulk2IngestJob(jobInfo.Id) + ErrorAndExit("Failed to upload data: " + err.Error()) + } + + // Close job to start processing + jobInfo, err = force.CloseBulk2IngestJob(jobInfo.Id) + if err != nil { + force.AbortBulk2IngestJob(jobInfo.Id) + ErrorAndExit("Failed to close job: " + err.Error()) + } + fmt.Fprintf(os.Stderr, "Data uploaded. Job submitted for processing.\n") + + if !wait { + fmt.Fprintf(os.Stderr, "To check job status use:\n force bulk2 job %s\n", jobInfo.Id) + fmt.Fprintf(os.Stderr, "To get results use:\n force bulk2 results %s\n", jobInfo.Id) + return + } + + if interactive { + startBulk2BubbleProgram(jobInfo) + } else { + waitForBulk2IngestJob(jobInfo.Id) + } +} + +func startBulk2BubbleProgram(jobInfo Bulk2IngestJobInfo) { + d := bubbles.NewBulk2JobModel() + p := tea.NewProgram(d, tea.WithOutput(os.Stderr)) + go func() { + for { + status, err := force.GetBulk2IngestJobInfo(jobInfo.Id) + if err != nil { + ErrorAndExit("Failed to get bulk job status: " + err.Error()) + } + p.Send(bubbles.NewBulk2JobStatusMsg{Bulk2IngestJobInfo: status}) + if status.IsTerminal() { + time.Sleep(500 * time.Millisecond) + p.Send(bubbles.QuitMsg{}) + return + } + time.Sleep(2 * time.Second) + } + }() + p.Run() +} + +func waitForBulk2IngestJob(jobId string) Bulk2IngestJobInfo { + for { + status, err := force.GetBulk2IngestJobInfo(jobId) + if err != nil { + ErrorAndExit("Failed to get bulk job status: " + err.Error()) + } + DisplayBulk2IngestJobInfo(status, os.Stderr) + if status.IsTerminal() { + return status + } + time.Sleep(2 * time.Second) + } +} + +func waitForBulk2QueryJob(jobId string) Bulk2QueryJobInfo { + for { + status, err := force.GetBulk2QueryJobInfo(jobId) + if err != nil { + ErrorAndExit("Failed to get bulk job status: " + err.Error()) + } + DisplayBulk2QueryJobInfo(status, os.Stderr) + if status.IsTerminal() { + return status + } + time.Sleep(2 * time.Second) + } +} + +func showBulk2IngestResults(jobId string, successful, failed, unprocessed bool) { + if successful { + results, err := force.GetBulk2SuccessfulResults(jobId) + if err != nil { + fmt.Fprintf(os.Stderr, "Failed to get successful results: %s\n", err.Error()) + } else if len(results) > 0 { + fmt.Println("=== Successful Results ===") + fmt.Print(string(results)) + } + } + + if failed { + results, err := force.GetBulk2FailedResults(jobId) + if err != nil { + fmt.Fprintf(os.Stderr, "Failed to get failed results: %s\n", err.Error()) + } else if len(results) > 0 { + fmt.Println("=== Failed Results ===") + fmt.Print(string(results)) + } + } + + if unprocessed { + results, err := force.GetBulk2UnprocessedRecords(jobId) + if err != nil { + fmt.Fprintf(os.Stderr, "Failed to get unprocessed records: %s\n", err.Error()) + } else if len(results) > 0 { + fmt.Println("=== Unprocessed Records ===") + fmt.Print(string(results)) + } + } +} + +func displayBulk2QueryResults(jobId string) { + locator := "" + headerDisplayed := false + for { + results, err := force.GetBulk2QueryResults(jobId, locator, 0) + if err != nil { + ErrorAndExit("Failed to get query results: " + err.Error()) + } + + data := results.Data + if headerDisplayed && len(data) > 0 { + // Skip header row for subsequent pages + data = stripFirstLine(data) + } + + if len(data) > 0 { + fmt.Print(string(data)) + headerDisplayed = true + } + + if results.Locator == "" { + break + } + locator = results.Locator + } +} diff --git a/command/bulk2_test.go b/command/bulk2_test.go new file mode 100644 index 00000000..d59a9bf3 --- /dev/null +++ b/command/bulk2_test.go @@ -0,0 +1,469 @@ +package command_test + +import ( + "os" + + . "github.com/ForceCLI/force/command" + + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" +) + +var _ = Describe("Bulk2", func() { + Describe("Command Registration", func() { + It("should have bulk2 command registered", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("bulk2")) + }) + + It("should have insert subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "insert"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("insert ")) + }) + + It("should have update subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "update"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("update ")) + }) + + It("should have upsert subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "upsert"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(HavePrefix("upsert")) + }) + + It("should have delete subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "delete"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("delete ")) + }) + + It("should have hardDelete subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "hardDelete"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("hardDelete ")) + }) + + It("should have query subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "query"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("query ")) + }) + + It("should have job subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "job"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("job ")) + }) + + It("should have jobs subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "jobs"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("jobs")) + }) + + It("should have results subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "results"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("results ")) + }) + + It("should have abort subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "abort"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("abort ")) + }) + + It("should have delete-job subcommand", func() { + cmd, _, err := RootCmd.Find([]string{"bulk2", "delete-job"}) + Expect(err).To(BeNil()) + Expect(cmd).NotTo(BeNil()) + Expect(cmd.Use).To(Equal("delete-job ")) + }) + }) + + Describe("Insert Command Flags", func() { + It("should have wait flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + flag := cmd.Flags().Lookup("wait") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("w")) + }) + + It("should have interactive flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + flag := cmd.Flags().Lookup("interactive") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("i")) + }) + + It("should have delimiter flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + flag := cmd.Flags().Lookup("delimiter") + Expect(flag).NotTo(BeNil()) + Expect(flag.DefValue).To(Equal("COMMA")) + }) + + It("should have lineending flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + flag := cmd.Flags().Lookup("lineending") + Expect(flag).NotTo(BeNil()) + Expect(flag.DefValue).To(Equal("LF")) + }) + }) + + Describe("Upsert Command Flags", func() { + It("should have externalid flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "upsert"}) + flag := cmd.Flags().Lookup("externalid") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("e")) + }) + + It("should require externalid flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "upsert"}) + flag := cmd.Flags().Lookup("externalid") + Expect(flag).NotTo(BeNil()) + + annotations := flag.Annotations + Expect(annotations).NotTo(BeNil()) + }) + }) + + Describe("Query Command Flags", func() { + It("should have wait flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "query"}) + flag := cmd.Flags().Lookup("wait") + Expect(flag).NotTo(BeNil()) + }) + + It("should have query-all flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "query"}) + flag := cmd.Flags().Lookup("query-all") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("A")) + }) + + It("should have delimiter flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "query"}) + flag := cmd.Flags().Lookup("delimiter") + Expect(flag).NotTo(BeNil()) + }) + + It("should have lineending flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "query"}) + flag := cmd.Flags().Lookup("lineending") + Expect(flag).NotTo(BeNil()) + }) + }) + + Describe("Results Command Flags", func() { + It("should have successful flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "results"}) + flag := cmd.Flags().Lookup("successful") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("s")) + }) + + It("should have failed flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "results"}) + flag := cmd.Flags().Lookup("failed") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("f")) + }) + + It("should have unprocessed flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "results"}) + flag := cmd.Flags().Lookup("unprocessed") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("u")) + }) + }) + + Describe("Jobs Command Flags", func() { + It("should have query flag", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "jobs"}) + flag := cmd.Flags().Lookup("query") + Expect(flag).NotTo(BeNil()) + Expect(flag.Shorthand).To(Equal("q")) + }) + }) + + Describe("Command Arguments", func() { + It("insert should require exactly 2 arguments", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + err := cmd.Args(cmd, []string{"Account"}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"Account", "file.csv"}) + Expect(err).To(BeNil()) + + err = cmd.Args(cmd, []string{"Account", "file.csv", "extra"}) + Expect(err).NotTo(BeNil()) + }) + + It("update should require exactly 2 arguments", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "update"}) + err := cmd.Args(cmd, []string{"Account"}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"Account", "file.csv"}) + Expect(err).To(BeNil()) + }) + + It("upsert should require exactly 2 arguments", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "upsert"}) + err := cmd.Args(cmd, []string{"Account"}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"Account", "file.csv"}) + Expect(err).To(BeNil()) + }) + + It("delete should require exactly 2 arguments", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "delete"}) + err := cmd.Args(cmd, []string{"Account"}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"Account", "file.csv"}) + Expect(err).To(BeNil()) + }) + + It("hardDelete should require exactly 2 arguments", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "hardDelete"}) + err := cmd.Args(cmd, []string{"Account"}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"Account", "file.csv"}) + Expect(err).To(BeNil()) + }) + + It("query should require exactly 1 argument", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "query"}) + err := cmd.Args(cmd, []string{}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"SELECT Id FROM Account"}) + Expect(err).To(BeNil()) + + err = cmd.Args(cmd, []string{"SELECT", "Id"}) + Expect(err).NotTo(BeNil()) + }) + + It("job should require exactly 1 argument", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "job"}) + err := cmd.Args(cmd, []string{}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"750test"}) + Expect(err).To(BeNil()) + }) + + It("jobs should accept no arguments", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "jobs"}) + Expect(cmd).NotTo(BeNil()) + // jobs command doesn't have explicit args validation (accepts any args) + }) + + It("results should require exactly 1 argument", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "results"}) + err := cmd.Args(cmd, []string{}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"750test"}) + Expect(err).To(BeNil()) + }) + + It("abort should require exactly 1 argument", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "abort"}) + err := cmd.Args(cmd, []string{}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"750test"}) + Expect(err).To(BeNil()) + }) + + It("delete-job should require exactly 1 argument", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "delete-job"}) + err := cmd.Args(cmd, []string{}) + Expect(err).NotTo(BeNil()) + + err = cmd.Args(cmd, []string{"750test"}) + Expect(err).To(BeNil()) + }) + }) + + Describe("Command Help Text", func() { + It("bulk2 should have descriptive help", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2"}) + Expect(cmd.Short).To(ContainSubstring("Bulk API 2.0")) + }) + + It("bulk2 should have examples", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2"}) + Expect(cmd.Example).To(ContainSubstring("insert")) + Expect(cmd.Example).To(ContainSubstring("query")) + Expect(cmd.Example).To(ContainSubstring("jobs")) + }) + + It("insert should have descriptive short help", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + Expect(cmd.Short).To(ContainSubstring("Insert")) + Expect(cmd.Short).To(ContainSubstring("CSV")) + }) + + It("query should have descriptive short help", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "query"}) + Expect(cmd.Short).To(ContainSubstring("Query")) + }) + }) + + Describe("CSV File Operations", func() { + var tempDir string + + BeforeEach(func() { + tempDir, _ = os.MkdirTemp("", "bulk2-test") + }) + + AfterEach(func() { + os.RemoveAll(tempDir) + }) + + It("should accept valid CSV file path", func() { + csvFilePath := tempDir + "/test.csv" + csvContents := "Name,Description\nTest Account,A test account" + os.WriteFile(csvFilePath, []byte(csvContents), 0644) + + _, err := os.Stat(csvFilePath) + Expect(err).To(BeNil()) + }) + + It("should handle CSV with special characters", func() { + csvFilePath := tempDir + "/special.csv" + csvContents := "Name,Description\n\"Test, Account\",\"Description with \"\"quotes\"\"\"" + os.WriteFile(csvFilePath, []byte(csvContents), 0644) + + content, err := os.ReadFile(csvFilePath) + Expect(err).To(BeNil()) + Expect(string(content)).To(Equal(csvContents)) + }) + + It("should handle large CSV file", func() { + csvFilePath := tempDir + "/large.csv" + var csvBuilder string + csvBuilder = "Name,Description\n" + for range 1000 { + csvBuilder += "Test Account,A test account description\n" + } + os.WriteFile(csvFilePath, []byte(csvBuilder), 0644) + + info, err := os.Stat(csvFilePath) + Expect(err).To(BeNil()) + Expect(info.Size()).To(BeNumerically(">", 30000)) + }) + }) + + Describe("Delimiter Values", func() { + It("should accept COMMA delimiter", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("delimiter", "COMMA") + val, _ := cmd.Flags().GetString("delimiter") + Expect(val).To(Equal("COMMA")) + }) + + It("should accept TAB delimiter", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("delimiter", "TAB") + val, _ := cmd.Flags().GetString("delimiter") + Expect(val).To(Equal("TAB")) + }) + + It("should accept PIPE delimiter", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("delimiter", "PIPE") + val, _ := cmd.Flags().GetString("delimiter") + Expect(val).To(Equal("PIPE")) + }) + + It("should accept SEMICOLON delimiter", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("delimiter", "SEMICOLON") + val, _ := cmd.Flags().GetString("delimiter") + Expect(val).To(Equal("SEMICOLON")) + }) + + It("should accept CARET delimiter", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("delimiter", "CARET") + val, _ := cmd.Flags().GetString("delimiter") + Expect(val).To(Equal("CARET")) + }) + + It("should accept BACKQUOTE delimiter", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("delimiter", "BACKQUOTE") + val, _ := cmd.Flags().GetString("delimiter") + Expect(val).To(Equal("BACKQUOTE")) + }) + }) + + Describe("Line Ending Values", func() { + It("should accept LF line ending", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("lineending", "LF") + val, _ := cmd.Flags().GetString("lineending") + Expect(val).To(Equal("LF")) + }) + + It("should accept CRLF line ending", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("lineending", "CRLF") + val, _ := cmd.Flags().GetString("lineending") + Expect(val).To(Equal("CRLF")) + }) + }) + + Describe("Flag Combinations", func() { + It("interactive flag should imply wait", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "insert"}) + cmd.Flags().Set("interactive", "true") + + interactive, _ := cmd.Flags().GetBool("interactive") + Expect(interactive).To(BeTrue()) + }) + + It("query-all flag should be boolean", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "query"}) + cmd.Flags().Set("query-all", "true") + + queryAll, _ := cmd.Flags().GetBool("query-all") + Expect(queryAll).To(BeTrue()) + }) + + It("results flags can be combined", func() { + cmd, _, _ := RootCmd.Find([]string{"bulk2", "results"}) + cmd.Flags().Set("successful", "true") + cmd.Flags().Set("failed", "true") + + successful, _ := cmd.Flags().GetBool("successful") + failed, _ := cmd.Flags().GetBool("failed") + + Expect(successful).To(BeTrue()) + Expect(failed).To(BeTrue()) + }) + }) +}) diff --git a/lib/bulk2.go b/lib/bulk2.go new file mode 100644 index 00000000..c8557f20 --- /dev/null +++ b/lib/bulk2.go @@ -0,0 +1,855 @@ +package lib + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/ForceCLI/force/lib/internal" +) + +// Bulk2Operation represents the type of operation for a Bulk API 2.0 job +type Bulk2Operation string + +const ( + Bulk2OperationInsert Bulk2Operation = "insert" + Bulk2OperationUpdate Bulk2Operation = "update" + Bulk2OperationUpsert Bulk2Operation = "upsert" + Bulk2OperationDelete Bulk2Operation = "delete" + Bulk2OperationHardDelete Bulk2Operation = "hardDelete" + Bulk2OperationQuery Bulk2Operation = "query" + Bulk2OperationQueryAll Bulk2Operation = "queryAll" +) + +// Bulk2JobState represents the state of a Bulk API 2.0 job +type Bulk2JobState string + +const ( + Bulk2JobStateOpen Bulk2JobState = "Open" + Bulk2JobStateUploadComplete Bulk2JobState = "UploadComplete" + Bulk2JobStateInProgress Bulk2JobState = "InProgress" + Bulk2JobStateJobComplete Bulk2JobState = "JobComplete" + Bulk2JobStateFailed Bulk2JobState = "Failed" + Bulk2JobStateAborted Bulk2JobState = "Aborted" +) + +// Bulk2ColumnDelimiter represents the column delimiter for CSV data +type Bulk2ColumnDelimiter string + +const ( + Bulk2DelimiterComma Bulk2ColumnDelimiter = "COMMA" + Bulk2DelimiterTab Bulk2ColumnDelimiter = "TAB" + Bulk2DelimiterPipe Bulk2ColumnDelimiter = "PIPE" + Bulk2DelimiterSemicolon Bulk2ColumnDelimiter = "SEMICOLON" + Bulk2DelimiterCaret Bulk2ColumnDelimiter = "CARET" + Bulk2DelimiterBackquote Bulk2ColumnDelimiter = "BACKQUOTE" +) + +// Bulk2LineEnding represents the line ending for CSV data +type Bulk2LineEnding string + +const ( + Bulk2LineEndingLF Bulk2LineEnding = "LF" + Bulk2LineEndingCRLF Bulk2LineEnding = "CRLF" +) + +// Bulk2IngestJobRequest is the request body for creating an ingest job +type Bulk2IngestJobRequest struct { + Object string `json:"object"` + Operation Bulk2Operation `json:"operation"` + ExternalIdFieldName string `json:"externalIdFieldName,omitempty"` + ColumnDelimiter Bulk2ColumnDelimiter `json:"columnDelimiter,omitempty"` + LineEnding Bulk2LineEnding `json:"lineEnding,omitempty"` + ContentType string `json:"contentType,omitempty"` +} + +// Bulk2IngestJobInfo is the response for an ingest job +type Bulk2IngestJobInfo struct { + Id string `json:"id"` + Operation Bulk2Operation `json:"operation"` + Object string `json:"object"` + CreatedById string `json:"createdById"` + CreatedDate string `json:"createdDate"` + SystemModstamp string `json:"systemModstamp"` + State Bulk2JobState `json:"state"` + ExternalIdFieldName string `json:"externalIdFieldName,omitempty"` + ConcurrencyMode string `json:"concurrencyMode"` + ContentType string `json:"contentType"` + ApiVersion float64 `json:"apiVersion"` + ContentUrl string `json:"contentUrl,omitempty"` + LineEnding Bulk2LineEnding `json:"lineEnding"` + ColumnDelimiter Bulk2ColumnDelimiter `json:"columnDelimiter"` + JobType string `json:"jobType,omitempty"` + NumberRecordsProcessed int `json:"numberRecordsProcessed"` + NumberRecordsFailed int `json:"numberRecordsFailed"` + Retries int `json:"retries"` + TotalProcessingTime int `json:"totalProcessingTime"` + ApiActiveProcessingTime int `json:"apiActiveProcessingTime"` + ApexProcessingTime int `json:"apexProcessingTime"` + ErrorMessage string `json:"errorMessage,omitempty"` +} + +// IsTerminal returns true if the job state is a terminal state +func (j *Bulk2IngestJobInfo) IsTerminal() bool { + return j.State == Bulk2JobStateJobComplete || + j.State == Bulk2JobStateFailed || + j.State == Bulk2JobStateAborted +} + +// Bulk2IngestJobList is the response for listing ingest jobs +type Bulk2IngestJobList struct { + Done bool `json:"done"` + Records []Bulk2IngestJobInfo `json:"records"` + NextRecordsUrl string `json:"nextRecordsUrl,omitempty"` +} + +// Bulk2QueryJobRequest is the request body for creating a query job +type Bulk2QueryJobRequest struct { + Operation Bulk2Operation `json:"operation"` + Query string `json:"query"` + ColumnDelimiter Bulk2ColumnDelimiter `json:"columnDelimiter,omitempty"` + LineEnding Bulk2LineEnding `json:"lineEnding,omitempty"` +} + +// Bulk2QueryJobInfo is the response for a query job +type Bulk2QueryJobInfo struct { + Id string `json:"id"` + Operation Bulk2Operation `json:"operation"` + Object string `json:"object"` + CreatedById string `json:"createdById"` + CreatedDate string `json:"createdDate"` + SystemModstamp string `json:"systemModstamp"` + State Bulk2JobState `json:"state"` + ConcurrencyMode string `json:"concurrencyMode"` + ContentType string `json:"contentType"` + ApiVersion float64 `json:"apiVersion"` + LineEnding Bulk2LineEnding `json:"lineEnding"` + ColumnDelimiter Bulk2ColumnDelimiter `json:"columnDelimiter"` + JobType string `json:"jobType,omitempty"` + NumberRecordsProcessed int `json:"numberRecordsProcessed"` + Retries int `json:"retries"` + TotalProcessingTime int `json:"totalProcessingTime"` + ErrorMessage string `json:"errorMessage,omitempty"` +} + +// IsTerminal returns true if the job state is a terminal state +func (j *Bulk2QueryJobInfo) IsTerminal() bool { + return j.State == Bulk2JobStateJobComplete || + j.State == Bulk2JobStateFailed || + j.State == Bulk2JobStateAborted +} + +// Bulk2QueryJobList is the response for listing query jobs +type Bulk2QueryJobList struct { + Done bool `json:"done"` + Records []Bulk2QueryJobInfo `json:"records"` + NextRecordsUrl string `json:"nextRecordsUrl,omitempty"` +} + +// Bulk2QueryResults holds query results with pagination info +type Bulk2QueryResults struct { + Data []byte + Locator string +} + +// bulk2IngestUrl returns the URL for the Bulk API 2.0 ingest endpoint +func (f *Force) bulk2IngestUrl() string { + return fmt.Sprintf("%s/services/data/%s/jobs/ingest", f.Credentials.InstanceUrl, apiVersion) +} + +// bulk2QueryUrl returns the URL for the Bulk API 2.0 query endpoint +func (f *Force) bulk2QueryUrl() string { + return fmt.Sprintf("%s/services/data/%s/jobs/query", f.Credentials.InstanceUrl, apiVersion) +} + +// CreateBulk2IngestJob creates a new Bulk API 2.0 ingest job +func (f *Force) CreateBulk2IngestJob(request Bulk2IngestJobRequest) (Bulk2IngestJobInfo, error) { + if request.ContentType == "" { + request.ContentType = "CSV" + } + url := f.bulk2IngestUrl() + body, err := json.Marshal(request) + if err != nil { + return Bulk2IngestJobInfo{}, fmt.Errorf("failed to marshal request: %w", err) + } + + req := NewRequest("POST"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + WithBody(strings.NewReader(string(body))). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return Bulk2IngestJobInfo{}, err + } + + var result Bulk2IngestJobInfo + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return Bulk2IngestJobInfo{}, err + } + return result, nil +} + +// CreateBulk2IngestJobWithContext creates a new Bulk API 2.0 ingest job with context +func (f *Force) CreateBulk2IngestJobWithContext(ctx context.Context, request Bulk2IngestJobRequest) (Bulk2IngestJobInfo, error) { + done := make(chan struct{}) + var result Bulk2IngestJobInfo + var err error + go func() { + defer close(done) + result, err = f.CreateBulk2IngestJob(request) + }() + select { + case <-ctx.Done(): + return Bulk2IngestJobInfo{}, fmt.Errorf("CreateBulk2IngestJob canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +// UploadBulk2JobData uploads CSV data to a Bulk API 2.0 job +func (f *Force) UploadBulk2JobData(jobId string, csvReader io.Reader) error { + url := fmt.Sprintf("%s/%s/batches", f.bulk2IngestUrl(), jobId) + + req := NewRequest("PUT"). + AbsoluteUrl(url). + WithContent(ContentTypeCsv). + WithBody(csvReader). + ReadResponseBody() + + _, err := f.ExecuteRequest(req) + return err +} + +// UploadBulk2JobDataWithContext uploads CSV data to a Bulk API 2.0 job with context +func (f *Force) UploadBulk2JobDataWithContext(ctx context.Context, jobId string, csvReader io.Reader) error { + done := make(chan struct{}) + var err error + go func() { + defer close(done) + err = f.UploadBulk2JobData(jobId, csvReader) + }() + select { + case <-ctx.Done(): + return fmt.Errorf("UploadBulk2JobData canceled: %w", ctx.Err()) + case <-done: + return err + } +} + +// CloseBulk2IngestJob closes a Bulk API 2.0 ingest job (sets state to UploadComplete) +func (f *Force) CloseBulk2IngestJob(jobId string) (Bulk2IngestJobInfo, error) { + return f.patchBulk2IngestJobState(jobId, Bulk2JobStateUploadComplete) +} + +// CloseBulk2IngestJobWithContext closes a Bulk API 2.0 ingest job with context +func (f *Force) CloseBulk2IngestJobWithContext(ctx context.Context, jobId string) (Bulk2IngestJobInfo, error) { + done := make(chan struct{}) + var result Bulk2IngestJobInfo + var err error + go func() { + defer close(done) + result, err = f.CloseBulk2IngestJob(jobId) + }() + select { + case <-ctx.Done(): + return Bulk2IngestJobInfo{}, fmt.Errorf("CloseBulk2IngestJob canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +// AbortBulk2IngestJob aborts a Bulk API 2.0 ingest job +func (f *Force) AbortBulk2IngestJob(jobId string) (Bulk2IngestJobInfo, error) { + return f.patchBulk2IngestJobState(jobId, Bulk2JobStateAborted) +} + +// AbortBulk2IngestJobWithContext aborts a Bulk API 2.0 ingest job with context +func (f *Force) AbortBulk2IngestJobWithContext(ctx context.Context, jobId string) (Bulk2IngestJobInfo, error) { + done := make(chan struct{}) + var result Bulk2IngestJobInfo + var err error + go func() { + defer close(done) + result, err = f.AbortBulk2IngestJob(jobId) + }() + select { + case <-ctx.Done(): + return Bulk2IngestJobInfo{}, fmt.Errorf("AbortBulk2IngestJob canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +func (f *Force) patchBulk2IngestJobState(jobId string, state Bulk2JobState) (Bulk2IngestJobInfo, error) { + url := fmt.Sprintf("%s/%s", f.bulk2IngestUrl(), jobId) + body := fmt.Sprintf(`{"state": "%s"}`, state) + + req := NewRequest("PATCH"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + WithBody(strings.NewReader(body)). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return Bulk2IngestJobInfo{}, err + } + + var result Bulk2IngestJobInfo + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return Bulk2IngestJobInfo{}, err + } + return result, nil +} + +// GetBulk2IngestJobInfo retrieves information about a Bulk API 2.0 ingest job +func (f *Force) GetBulk2IngestJobInfo(jobId string) (Bulk2IngestJobInfo, error) { + url := fmt.Sprintf("%s/%s", f.bulk2IngestUrl(), jobId) + + req := NewRequest("GET"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return Bulk2IngestJobInfo{}, err + } + + var result Bulk2IngestJobInfo + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return Bulk2IngestJobInfo{}, err + } + return result, nil +} + +// GetBulk2IngestJobInfoWithContext retrieves information about a Bulk API 2.0 ingest job with context +func (f *Force) GetBulk2IngestJobInfoWithContext(ctx context.Context, jobId string) (Bulk2IngestJobInfo, error) { + done := make(chan struct{}) + var result Bulk2IngestJobInfo + var err error + go func() { + defer close(done) + result, err = f.GetBulk2IngestJobInfo(jobId) + }() + select { + case <-ctx.Done(): + return Bulk2IngestJobInfo{}, fmt.Errorf("GetBulk2IngestJobInfo canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +// GetBulk2IngestJobs retrieves all Bulk API 2.0 ingest jobs +func (f *Force) GetBulk2IngestJobs() ([]Bulk2IngestJobInfo, error) { + url := f.bulk2IngestUrl() + + req := NewRequest("GET"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return nil, err + } + + var result Bulk2IngestJobList + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return nil, err + } + + jobs := result.Records + for !result.Done && result.NextRecordsUrl != "" { + nextUrl := fmt.Sprintf("%s%s", f.Credentials.InstanceUrl, result.NextRecordsUrl) + req := NewRequest("GET"). + AbsoluteUrl(nextUrl). + WithContent(ContentTypeJson). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return nil, err + } + + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return nil, err + } + jobs = append(jobs, result.Records...) + } + + return jobs, nil +} + +// DeleteBulk2IngestJob deletes a Bulk API 2.0 ingest job +func (f *Force) DeleteBulk2IngestJob(jobId string) error { + url := fmt.Sprintf("%s/%s", f.bulk2IngestUrl(), jobId) + + req := NewRequest("DELETE"). + AbsoluteUrl(url). + ReadResponseBody() + + _, err := f.ExecuteRequest(req) + return err +} + +// GetBulk2SuccessfulResults retrieves successful results for a Bulk API 2.0 ingest job +func (f *Force) GetBulk2SuccessfulResults(jobId string) ([]byte, error) { + url := fmt.Sprintf("%s/%s/successfulResults", f.bulk2IngestUrl(), jobId) + return f.getBulk2Results(url) +} + +// GetBulk2FailedResults retrieves failed results for a Bulk API 2.0 ingest job +func (f *Force) GetBulk2FailedResults(jobId string) ([]byte, error) { + url := fmt.Sprintf("%s/%s/failedResults", f.bulk2IngestUrl(), jobId) + return f.getBulk2Results(url) +} + +// GetBulk2UnprocessedRecords retrieves unprocessed records for a Bulk API 2.0 ingest job +func (f *Force) GetBulk2UnprocessedRecords(jobId string) ([]byte, error) { + url := fmt.Sprintf("%s/%s/unprocessedrecords", f.bulk2IngestUrl(), jobId) + return f.getBulk2Results(url) +} + +func (f *Force) getBulk2Results(url string) ([]byte, error) { + req := NewRequest("GET"). + AbsoluteUrl(url). + WithHeader("Accept", "text/csv"). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return nil, err + } + return resp.ReadResponseBody, nil +} + +// GetBulk2ResultsWithCallback retrieves results for a Bulk API 2.0 job using a callback +func (f *Force) GetBulk2ResultsWithCallback(url string, callback HttpCallback) error { + req := NewRequest("GET"). + AbsoluteUrl(url). + WithHeader("Accept", "text/csv"). + WithResponseCallback(callback) + + _, err := f.ExecuteRequest(req) + return err +} + +// CreateBulk2QueryJob creates a new Bulk API 2.0 query job +func (f *Force) CreateBulk2QueryJob(request Bulk2QueryJobRequest) (Bulk2QueryJobInfo, error) { + if request.Operation == "" { + request.Operation = Bulk2OperationQuery + } + url := f.bulk2QueryUrl() + body, err := json.Marshal(request) + if err != nil { + return Bulk2QueryJobInfo{}, fmt.Errorf("failed to marshal request: %w", err) + } + + req := NewRequest("POST"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + WithBody(strings.NewReader(string(body))). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return Bulk2QueryJobInfo{}, err + } + + var result Bulk2QueryJobInfo + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return Bulk2QueryJobInfo{}, err + } + return result, nil +} + +// CreateBulk2QueryJobWithContext creates a new Bulk API 2.0 query job with context +func (f *Force) CreateBulk2QueryJobWithContext(ctx context.Context, request Bulk2QueryJobRequest) (Bulk2QueryJobInfo, error) { + done := make(chan struct{}) + var result Bulk2QueryJobInfo + var err error + go func() { + defer close(done) + result, err = f.CreateBulk2QueryJob(request) + }() + select { + case <-ctx.Done(): + return Bulk2QueryJobInfo{}, fmt.Errorf("CreateBulk2QueryJob canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +// GetBulk2QueryJobInfo retrieves information about a Bulk API 2.0 query job +func (f *Force) GetBulk2QueryJobInfo(jobId string) (Bulk2QueryJobInfo, error) { + url := fmt.Sprintf("%s/%s", f.bulk2QueryUrl(), jobId) + + req := NewRequest("GET"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return Bulk2QueryJobInfo{}, err + } + + var result Bulk2QueryJobInfo + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return Bulk2QueryJobInfo{}, err + } + return result, nil +} + +// GetBulk2QueryJobInfoWithContext retrieves information about a Bulk API 2.0 query job with context +func (f *Force) GetBulk2QueryJobInfoWithContext(ctx context.Context, jobId string) (Bulk2QueryJobInfo, error) { + done := make(chan struct{}) + var result Bulk2QueryJobInfo + var err error + go func() { + defer close(done) + result, err = f.GetBulk2QueryJobInfo(jobId) + }() + select { + case <-ctx.Done(): + return Bulk2QueryJobInfo{}, fmt.Errorf("GetBulk2QueryJobInfo canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +// GetBulk2QueryResults retrieves results for a Bulk API 2.0 query job +// locator is used for pagination, pass empty string for the first page +// maxRecords limits the number of records returned (0 for default) +func (f *Force) GetBulk2QueryResults(jobId string, locator string, maxRecords int) (Bulk2QueryResults, error) { + url := fmt.Sprintf("%s/%s/results", f.bulk2QueryUrl(), jobId) + if locator != "" { + url = fmt.Sprintf("%s?locator=%s", url, locator) + if maxRecords > 0 { + url = fmt.Sprintf("%s&maxRecords=%d", url, maxRecords) + } + } else if maxRecords > 0 { + url = fmt.Sprintf("%s?maxRecords=%d", url, maxRecords) + } + + req := NewRequest("GET"). + AbsoluteUrl(url). + WithHeader("Accept", "text/csv"). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return Bulk2QueryResults{}, err + } + + result := Bulk2QueryResults{ + Data: resp.ReadResponseBody, + Locator: resp.HttpResponse.Header.Get("Sforce-Locator"), + } + if result.Locator == "null" { + result.Locator = "" + } + return result, nil +} + +// GetBulk2QueryResultsWithContext retrieves results for a Bulk API 2.0 query job with context +func (f *Force) GetBulk2QueryResultsWithContext(ctx context.Context, jobId string, locator string, maxRecords int) (Bulk2QueryResults, error) { + done := make(chan struct{}) + var result Bulk2QueryResults + var err error + go func() { + defer close(done) + result, err = f.GetBulk2QueryResults(jobId, locator, maxRecords) + }() + select { + case <-ctx.Done(): + return Bulk2QueryResults{}, fmt.Errorf("GetBulk2QueryResults canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +// AbortBulk2QueryJob aborts a Bulk API 2.0 query job +func (f *Force) AbortBulk2QueryJob(jobId string) (Bulk2QueryJobInfo, error) { + url := fmt.Sprintf("%s/%s", f.bulk2QueryUrl(), jobId) + body := fmt.Sprintf(`{"state": "%s"}`, Bulk2JobStateAborted) + + req := NewRequest("PATCH"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + WithBody(strings.NewReader(body)). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return Bulk2QueryJobInfo{}, err + } + + var result Bulk2QueryJobInfo + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return Bulk2QueryJobInfo{}, err + } + return result, nil +} + +// AbortBulk2QueryJobWithContext aborts a Bulk API 2.0 query job with context +func (f *Force) AbortBulk2QueryJobWithContext(ctx context.Context, jobId string) (Bulk2QueryJobInfo, error) { + done := make(chan struct{}) + var result Bulk2QueryJobInfo + var err error + go func() { + defer close(done) + result, err = f.AbortBulk2QueryJob(jobId) + }() + select { + case <-ctx.Done(): + return Bulk2QueryJobInfo{}, fmt.Errorf("AbortBulk2QueryJob canceled: %w", ctx.Err()) + case <-done: + return result, err + } +} + +// DeleteBulk2QueryJob deletes a Bulk API 2.0 query job +func (f *Force) DeleteBulk2QueryJob(jobId string) error { + url := fmt.Sprintf("%s/%s", f.bulk2QueryUrl(), jobId) + + req := NewRequest("DELETE"). + AbsoluteUrl(url). + ReadResponseBody() + + _, err := f.ExecuteRequest(req) + return err +} + +// GetBulk2QueryJobs retrieves all Bulk API 2.0 query jobs +func (f *Force) GetBulk2QueryJobs() ([]Bulk2QueryJobInfo, error) { + url := f.bulk2QueryUrl() + + req := NewRequest("GET"). + AbsoluteUrl(url). + WithContent(ContentTypeJson). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return nil, err + } + + var result Bulk2QueryJobList + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return nil, err + } + + jobs := result.Records + for !result.Done && result.NextRecordsUrl != "" { + nextUrl := fmt.Sprintf("%s%s", f.Credentials.InstanceUrl, result.NextRecordsUrl) + req := NewRequest("GET"). + AbsoluteUrl(nextUrl). + WithContent(ContentTypeJson). + ReadResponseBody() + + resp, err := f.ExecuteRequest(req) + if err != nil { + return nil, err + } + + if err := internal.JsonUnmarshal(resp.ReadResponseBody, &result); err != nil { + return nil, err + } + jobs = append(jobs, result.Records...) + } + + return jobs, nil +} + +// Bulk2JobStatusCallback is a function that receives job status updates +type Bulk2JobStatusCallback func(jobInfo any) + +// WaitForBulk2IngestJob waits for a Bulk API 2.0 ingest job to complete +func (f *Force) WaitForBulk2IngestJob(jobId string, pollInterval time.Duration, callback Bulk2JobStatusCallback) (Bulk2IngestJobInfo, error) { + for { + jobInfo, err := f.GetBulk2IngestJobInfo(jobId) + if err != nil { + return Bulk2IngestJobInfo{}, err + } + if callback != nil { + callback(jobInfo) + } + if jobInfo.IsTerminal() { + return jobInfo, nil + } + time.Sleep(pollInterval) + } +} + +// WaitForBulk2IngestJobWithContext waits for a Bulk API 2.0 ingest job to complete with context +func (f *Force) WaitForBulk2IngestJobWithContext(ctx context.Context, jobId string, pollInterval time.Duration, callback Bulk2JobStatusCallback) (Bulk2IngestJobInfo, error) { + for { + select { + case <-ctx.Done(): + return Bulk2IngestJobInfo{}, fmt.Errorf("WaitForBulk2IngestJob canceled: %w", ctx.Err()) + default: + } + + jobInfo, err := f.GetBulk2IngestJobInfoWithContext(ctx, jobId) + if err != nil { + return Bulk2IngestJobInfo{}, err + } + if callback != nil { + callback(jobInfo) + } + if jobInfo.IsTerminal() { + return jobInfo, nil + } + + select { + case <-ctx.Done(): + return Bulk2IngestJobInfo{}, fmt.Errorf("WaitForBulk2IngestJob canceled: %w", ctx.Err()) + case <-time.After(pollInterval): + } + } +} + +// WaitForBulk2QueryJob waits for a Bulk API 2.0 query job to complete +func (f *Force) WaitForBulk2QueryJob(jobId string, pollInterval time.Duration, callback Bulk2JobStatusCallback) (Bulk2QueryJobInfo, error) { + for { + jobInfo, err := f.GetBulk2QueryJobInfo(jobId) + if err != nil { + return Bulk2QueryJobInfo{}, err + } + if callback != nil { + callback(jobInfo) + } + if jobInfo.IsTerminal() { + return jobInfo, nil + } + time.Sleep(pollInterval) + } +} + +// WaitForBulk2QueryJobWithContext waits for a Bulk API 2.0 query job to complete with context +func (f *Force) WaitForBulk2QueryJobWithContext(ctx context.Context, jobId string, pollInterval time.Duration, callback Bulk2JobStatusCallback) (Bulk2QueryJobInfo, error) { + for { + select { + case <-ctx.Done(): + return Bulk2QueryJobInfo{}, fmt.Errorf("WaitForBulk2QueryJob canceled: %w", ctx.Err()) + default: + } + + jobInfo, err := f.GetBulk2QueryJobInfoWithContext(ctx, jobId) + if err != nil { + return Bulk2QueryJobInfo{}, err + } + if callback != nil { + callback(jobInfo) + } + if jobInfo.IsTerminal() { + return jobInfo, nil + } + + select { + case <-ctx.Done(): + return Bulk2QueryJobInfo{}, fmt.Errorf("WaitForBulk2QueryJob canceled: %w", ctx.Err()) + case <-time.After(pollInterval): + } + } +} + +// DisplayBulk2IngestJobInfo displays information about a Bulk API 2.0 ingest job +func DisplayBulk2IngestJobInfo(jobInfo Bulk2IngestJobInfo, w io.Writer) { + var msg = ` +Id %s +State %s +Operation %s +Object %s +Api Version %.1f + +Created By Id %s +Created Date %s +System Modstamp %s +Content Type %s +Concurrency Mode %s + +Number Records Processed %d +Number Records Failed %d +Retries %d + +Total Processing Time %d +Api Active Processing Time %d +Apex Processing Time %d +` + fmt.Fprintf(w, msg, jobInfo.Id, jobInfo.State, jobInfo.Operation, jobInfo.Object, jobInfo.ApiVersion, + jobInfo.CreatedById, jobInfo.CreatedDate, jobInfo.SystemModstamp, + jobInfo.ContentType, jobInfo.ConcurrencyMode, + jobInfo.NumberRecordsProcessed, jobInfo.NumberRecordsFailed, jobInfo.Retries, + jobInfo.TotalProcessingTime, jobInfo.ApiActiveProcessingTime, jobInfo.ApexProcessingTime) + + if jobInfo.ErrorMessage != "" { + fmt.Fprintf(w, "Error Message:\t\t\t%s\n", jobInfo.ErrorMessage) + } +} + +// DisplayBulk2QueryJobInfo displays information about a Bulk API 2.0 query job +func DisplayBulk2QueryJobInfo(jobInfo Bulk2QueryJobInfo, w io.Writer) { + var msg = ` +Id %s +State %s +Operation %s +Object %s +Api Version %.1f + +Created By Id %s +Created Date %s +System Modstamp %s +Content Type %s +Concurrency Mode %s + +Number Records Processed %d +Retries %d +Total Processing Time %d +` + fmt.Fprintf(w, msg, jobInfo.Id, jobInfo.State, jobInfo.Operation, jobInfo.Object, jobInfo.ApiVersion, + jobInfo.CreatedById, jobInfo.CreatedDate, jobInfo.SystemModstamp, + jobInfo.ContentType, jobInfo.ConcurrencyMode, + jobInfo.NumberRecordsProcessed, jobInfo.Retries, jobInfo.TotalProcessingTime) + + if jobInfo.ErrorMessage != "" { + fmt.Fprintf(w, "Error Message:\t\t\t%s\n", jobInfo.ErrorMessage) + } +} + +// DisplayBulk2IngestJobList displays a list of Bulk API 2.0 ingest jobs +func DisplayBulk2IngestJobList(jobs []Bulk2IngestJobInfo, w io.Writer) { + fmt.Fprintf(w, "%-18s %-12s %-15s %-15s %-12s %-10s\n", "ID", "State", "Operation", "Object", "Processed", "Failed") + fmt.Fprintf(w, "%s\n", strings.Repeat("-", 90)) + for _, job := range jobs { + fmt.Fprintf(w, "%-18s %-12s %-15s %-15s %-12d %-10d\n", + job.Id, job.State, job.Operation, job.Object, job.NumberRecordsProcessed, job.NumberRecordsFailed) + } +} + +// DisplayBulk2QueryJobList displays a list of Bulk API 2.0 query jobs +func DisplayBulk2QueryJobList(jobs []Bulk2QueryJobInfo, w io.Writer) { + fmt.Fprintf(w, "%-18s %-12s %-15s %-15s %-12s\n", "ID", "State", "Operation", "Object", "Processed") + fmt.Fprintf(w, "%s\n", strings.Repeat("-", 75)) + for _, job := range jobs { + fmt.Fprintf(w, "%-18s %-12s %-15s %-15s %-12d\n", + job.Id, job.State, job.Operation, job.Object, job.NumberRecordsProcessed) + } +} + +// ParseBulk2Warnings parses warning headers from an HTTP response +func ParseBulk2Warnings(resp *http.Response) []string { + var warnings []string + for _, w := range resp.Header.Values("Warning") { + warnings = append(warnings, w) + } + return warnings +} diff --git a/lib/bulk2_test.go b/lib/bulk2_test.go new file mode 100644 index 00000000..d8d8d3d4 --- /dev/null +++ b/lib/bulk2_test.go @@ -0,0 +1,1614 @@ +package lib + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +// ============================================================================= +// Job State Tests +// ============================================================================= + +func TestBulk2IngestJobInfo_IsTerminal(t *testing.T) { + tests := []struct { + name string + state Bulk2JobState + expected bool + }{ + {"Open is not terminal", Bulk2JobStateOpen, false}, + {"UploadComplete is not terminal", Bulk2JobStateUploadComplete, false}, + {"InProgress is not terminal", Bulk2JobStateInProgress, false}, + {"JobComplete is terminal", Bulk2JobStateJobComplete, true}, + {"Failed is terminal", Bulk2JobStateFailed, true}, + {"Aborted is terminal", Bulk2JobStateAborted, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + job := &Bulk2IngestJobInfo{State: tt.state} + if got := job.IsTerminal(); got != tt.expected { + t.Errorf("IsTerminal() = %v, want %v", got, tt.expected) + } + }) + } +} + +func TestBulk2QueryJobInfo_IsTerminal(t *testing.T) { + tests := []struct { + name string + state Bulk2JobState + expected bool + }{ + {"Open is not terminal", Bulk2JobStateOpen, false}, + {"InProgress is not terminal", Bulk2JobStateInProgress, false}, + {"JobComplete is terminal", Bulk2JobStateJobComplete, true}, + {"Failed is terminal", Bulk2JobStateFailed, true}, + {"Aborted is terminal", Bulk2JobStateAborted, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + job := &Bulk2QueryJobInfo{State: tt.state} + if got := job.IsTerminal(); got != tt.expected { + t.Errorf("IsTerminal() = %v, want %v", got, tt.expected) + } + }) + } +} + +// ============================================================================= +// Display Function Tests +// ============================================================================= + +func TestDisplayBulk2IngestJobInfo(t *testing.T) { + jobInfo := Bulk2IngestJobInfo{ + Id: "7501234567890ABCDEF", + State: Bulk2JobStateJobComplete, + Operation: Bulk2OperationInsert, + Object: "Account", + ApiVersion: 55.0, + CreatedById: "0051234567890ABCDEF", + CreatedDate: "2024-01-15T10:30:00.000Z", + SystemModstamp: "2024-01-15T10:35:00.000Z", + ContentType: "CSV", + ConcurrencyMode: "Parallel", + NumberRecordsProcessed: 100, + NumberRecordsFailed: 5, + Retries: 0, + TotalProcessingTime: 5000, + ApiActiveProcessingTime: 4500, + ApexProcessingTime: 500, + } + + var buf bytes.Buffer + DisplayBulk2IngestJobInfo(jobInfo, &buf) + output := buf.String() + + expectedParts := []string{ + "7501234567890ABCDEF", + "JobComplete", + "insert", + "Account", + "55.0", + "100", + "5", + } + + for _, part := range expectedParts { + if !strings.Contains(output, part) { + t.Errorf("Output should contain %q, got:\n%s", part, output) + } + } +} + +func TestDisplayBulk2IngestJobInfo_WithErrorMessage(t *testing.T) { + jobInfo := Bulk2IngestJobInfo{ + Id: "7501234567890ABCDEF", + State: Bulk2JobStateFailed, + Operation: Bulk2OperationInsert, + Object: "Account", + ErrorMessage: "InvalidBatch: Field Name not found", + } + + var buf bytes.Buffer + DisplayBulk2IngestJobInfo(jobInfo, &buf) + output := buf.String() + + if !strings.Contains(output, "InvalidBatch: Field Name not found") { + t.Errorf("Output should contain error message, got:\n%s", output) + } +} + +func TestDisplayBulk2QueryJobInfo(t *testing.T) { + jobInfo := Bulk2QueryJobInfo{ + Id: "7501234567890ABCDEF", + State: Bulk2JobStateJobComplete, + Operation: Bulk2OperationQuery, + Object: "Contact", + ApiVersion: 55.0, + CreatedById: "0051234567890ABCDEF", + CreatedDate: "2024-01-15T10:30:00.000Z", + SystemModstamp: "2024-01-15T10:35:00.000Z", + ContentType: "CSV", + ConcurrencyMode: "Parallel", + NumberRecordsProcessed: 500, + Retries: 1, + TotalProcessingTime: 10000, + } + + var buf bytes.Buffer + DisplayBulk2QueryJobInfo(jobInfo, &buf) + output := buf.String() + + expectedParts := []string{ + "7501234567890ABCDEF", + "JobComplete", + "query", + "Contact", + "55.0", + "500", + } + + for _, part := range expectedParts { + if !strings.Contains(output, part) { + t.Errorf("Output should contain %q, got:\n%s", part, output) + } + } +} + +func TestDisplayBulk2QueryJobInfo_WithErrorMessage(t *testing.T) { + jobInfo := Bulk2QueryJobInfo{ + Id: "7501234567890ABCDEF", + State: Bulk2JobStateFailed, + Operation: Bulk2OperationQuery, + Object: "Account", + ErrorMessage: "MALFORMED_QUERY: unexpected token", + } + + var buf bytes.Buffer + DisplayBulk2QueryJobInfo(jobInfo, &buf) + output := buf.String() + + if !strings.Contains(output, "MALFORMED_QUERY: unexpected token") { + t.Errorf("Output should contain error message, got:\n%s", output) + } +} + +func TestDisplayBulk2IngestJobList(t *testing.T) { + jobs := []Bulk2IngestJobInfo{ + { + Id: "750000000000001", + State: Bulk2JobStateJobComplete, + Operation: Bulk2OperationInsert, + Object: "Account", + NumberRecordsProcessed: 100, + NumberRecordsFailed: 5, + }, + { + Id: "750000000000002", + State: Bulk2JobStateInProgress, + Operation: Bulk2OperationUpdate, + Object: "Contact", + NumberRecordsProcessed: 50, + NumberRecordsFailed: 0, + }, + } + + var buf bytes.Buffer + DisplayBulk2IngestJobList(jobs, &buf) + output := buf.String() + + expectedParts := []string{ + "750000000000001", + "750000000000002", + "JobComplete", + "InProgress", + "insert", + "update", + "Account", + "Contact", + } + + for _, part := range expectedParts { + if !strings.Contains(output, part) { + t.Errorf("Output should contain %q, got:\n%s", part, output) + } + } +} + +func TestDisplayBulk2IngestJobList_Empty(t *testing.T) { + var buf bytes.Buffer + DisplayBulk2IngestJobList([]Bulk2IngestJobInfo{}, &buf) + output := buf.String() + + if !strings.Contains(output, "ID") { + t.Errorf("Output should contain header even for empty list, got:\n%s", output) + } +} + +func TestDisplayBulk2QueryJobList(t *testing.T) { + jobs := []Bulk2QueryJobInfo{ + { + Id: "750000000000001", + State: Bulk2JobStateJobComplete, + Operation: Bulk2OperationQuery, + Object: "Account", + NumberRecordsProcessed: 1000, + }, + { + Id: "750000000000002", + State: Bulk2JobStateFailed, + Operation: Bulk2OperationQueryAll, + Object: "Lead", + NumberRecordsProcessed: 0, + }, + } + + var buf bytes.Buffer + DisplayBulk2QueryJobList(jobs, &buf) + output := buf.String() + + expectedParts := []string{ + "750000000000001", + "750000000000002", + "JobComplete", + "Failed", + "query", + "queryAll", + "Account", + "Lead", + } + + for _, part := range expectedParts { + if !strings.Contains(output, part) { + t.Errorf("Output should contain %q, got:\n%s", part, output) + } + } +} + +// ============================================================================= +// Ingest Job Creation Tests +// ============================================================================= + +func TestCreateBulk2IngestJob(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + t.Errorf("Expected POST request, got %s", r.Method) + } + if !strings.HasSuffix(r.URL.Path, "/jobs/ingest") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"object":"Account"`) { + t.Errorf("Request body should contain object field, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "id": "7501234567890ABCDEF", + "operation": "insert", + "object": "Account", + "state": "Open", + "contentType": "CSV", + "apiVersion": 55.0 + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationInsert, + } + + jobInfo, err := force.CreateBulk2IngestJob(request) + if err != nil { + t.Fatalf("CreateBulk2IngestJob failed: %v", err) + } + + if jobInfo.Id != "7501234567890ABCDEF" { + t.Errorf("Expected job ID '7501234567890ABCDEF', got '%s'", jobInfo.Id) + } + if jobInfo.State != Bulk2JobStateOpen { + t.Errorf("Expected state 'Open', got '%s'", jobInfo.State) + } +} + +func TestCreateBulk2IngestJob_AllOperations(t *testing.T) { + operations := []struct { + operation Bulk2Operation + expectedInBody string + }{ + {Bulk2OperationInsert, `"operation":"insert"`}, + {Bulk2OperationUpdate, `"operation":"update"`}, + {Bulk2OperationUpsert, `"operation":"upsert"`}, + {Bulk2OperationDelete, `"operation":"delete"`}, + {Bulk2OperationHardDelete, `"operation":"hardDelete"`}, + } + + for _, op := range operations { + t.Run(string(op.operation), func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), op.expectedInBody) { + t.Errorf("Request body should contain %s, got: %s", op.expectedInBody, body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "Open"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: op.operation, + } + + _, err := force.CreateBulk2IngestJob(request) + if err != nil { + t.Fatalf("CreateBulk2IngestJob failed for %s: %v", op.operation, err) + } + }) + } +} + +func TestCreateBulk2IngestJob_WithExternalId(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"externalIdFieldName":"External_Id__c"`) { + t.Errorf("Request body should contain externalIdFieldName, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "Open", "externalIdFieldName": "External_Id__c"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationUpsert, + ExternalIdFieldName: "External_Id__c", + } + + jobInfo, err := force.CreateBulk2IngestJob(request) + if err != nil { + t.Fatalf("CreateBulk2IngestJob failed: %v", err) + } + + if jobInfo.ExternalIdFieldName != "External_Id__c" { + t.Errorf("Expected externalIdFieldName 'External_Id__c', got '%s'", jobInfo.ExternalIdFieldName) + } +} + +func TestCreateBulk2IngestJob_WithDelimiterAndLineEnding(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"columnDelimiter":"TAB"`) { + t.Errorf("Request body should contain columnDelimiter, got: %s", body) + } + if !strings.Contains(string(body), `"lineEnding":"CRLF"`) { + t.Errorf("Request body should contain lineEnding, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "Open", "columnDelimiter": "TAB", "lineEnding": "CRLF"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationInsert, + ColumnDelimiter: Bulk2DelimiterTab, + LineEnding: Bulk2LineEndingCRLF, + } + + jobInfo, err := force.CreateBulk2IngestJob(request) + if err != nil { + t.Fatalf("CreateBulk2IngestJob failed: %v", err) + } + + if jobInfo.ColumnDelimiter != Bulk2DelimiterTab { + t.Errorf("Expected columnDelimiter 'TAB', got '%s'", jobInfo.ColumnDelimiter) + } + if jobInfo.LineEnding != Bulk2LineEndingCRLF { + t.Errorf("Expected lineEnding 'CRLF', got '%s'", jobInfo.LineEnding) + } +} + +func TestCreateBulk2IngestJob_WithContext_Canceled(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(100 * time.Millisecond) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "Open"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationInsert, + } + + _, err := force.CreateBulk2IngestJobWithContext(ctx, request) + if err == nil { + t.Fatal("Expected error for canceled context") + } + if !strings.Contains(err.Error(), "canceled") { + t.Errorf("Expected canceled error, got: %v", err) + } +} + +// ============================================================================= +// Upload Data Tests +// ============================================================================= + +func TestUploadBulk2JobData(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "PUT" { + t.Errorf("Expected PUT request, got %s", r.Method) + } + if !strings.Contains(r.URL.Path, "/batches") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + if r.Header.Get("Content-Type") != string(ContentTypeCsv) { + t.Errorf("Expected Content-Type %s, got %s", ContentTypeCsv, r.Header.Get("Content-Type")) + } + + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), "Name") { + t.Errorf("Request body should contain CSV header, got: %s", body) + } + + w.WriteHeader(http.StatusCreated) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + csvData := "Name,Description\nTest Account,A test account" + err := force.UploadBulk2JobData("7501234567890ABCDEF", strings.NewReader(csvData)) + if err != nil { + t.Fatalf("UploadBulk2JobData failed: %v", err) + } +} + +func TestUploadBulk2JobData_LargeCSV(t *testing.T) { + var receivedSize int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + receivedSize = int64(len(body)) + w.WriteHeader(http.StatusCreated) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + var csvBuilder strings.Builder + csvBuilder.WriteString("Name,Description\n") + for range 10000 { + csvBuilder.WriteString("Test Account,A test account description that is reasonably long\n") + } + csvData := csvBuilder.String() + + err := force.UploadBulk2JobData("7501234567890ABCDEF", strings.NewReader(csvData)) + if err != nil { + t.Fatalf("UploadBulk2JobData failed: %v", err) + } + + if receivedSize != int64(len(csvData)) { + t.Errorf("Expected to receive %d bytes, got %d", len(csvData), receivedSize) + } +} + +func TestUploadBulk2JobData_WithContext_Canceled(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(100 * time.Millisecond) + w.WriteHeader(http.StatusCreated) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + csvData := "Name,Description\nTest Account,A test account" + err := force.UploadBulk2JobDataWithContext(ctx, "7501234567890ABCDEF", strings.NewReader(csvData)) + if err == nil { + t.Fatal("Expected error for canceled context") + } +} + +// ============================================================================= +// Close/Abort Job Tests +// ============================================================================= + +func TestCloseBulk2IngestJob(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "PATCH" { + t.Errorf("Expected PATCH request, got %s", r.Method) + } + + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"state": "UploadComplete"`) { + t.Errorf("Request body should contain state=UploadComplete, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "id": "7501234567890ABCDEF", + "state": "UploadComplete" + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobInfo, err := force.CloseBulk2IngestJob("7501234567890ABCDEF") + if err != nil { + t.Fatalf("CloseBulk2IngestJob failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateUploadComplete { + t.Errorf("Expected state 'UploadComplete', got '%s'", jobInfo.State) + } +} + +func TestAbortBulk2IngestJob(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "PATCH" { + t.Errorf("Expected PATCH request, got %s", r.Method) + } + + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"state": "Aborted"`) { + t.Errorf("Request body should contain state=Aborted, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "id": "7501234567890ABCDEF", + "state": "Aborted" + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobInfo, err := force.AbortBulk2IngestJob("7501234567890ABCDEF") + if err != nil { + t.Fatalf("AbortBulk2IngestJob failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateAborted { + t.Errorf("Expected state 'Aborted', got '%s'", jobInfo.State) + } +} + +// ============================================================================= +// Get Job Info Tests +// ============================================================================= + +func TestGetBulk2IngestJobInfo(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + if !strings.Contains(r.URL.Path, "/jobs/ingest/") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "id": "7501234567890ABCDEF", + "operation": "insert", + "object": "Account", + "state": "JobComplete", + "numberRecordsProcessed": 100, + "numberRecordsFailed": 5 + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobInfo, err := force.GetBulk2IngestJobInfo("7501234567890ABCDEF") + if err != nil { + t.Fatalf("GetBulk2IngestJobInfo failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateJobComplete { + t.Errorf("Expected state 'JobComplete', got '%s'", jobInfo.State) + } + if jobInfo.NumberRecordsProcessed != 100 { + t.Errorf("Expected 100 records processed, got %d", jobInfo.NumberRecordsProcessed) + } + if jobInfo.NumberRecordsFailed != 5 { + t.Errorf("Expected 5 records failed, got %d", jobInfo.NumberRecordsFailed) + } +} + +func TestGetBulk2IngestJobs(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "done": true, + "records": [ + {"id": "750000000000001", "state": "JobComplete", "operation": "insert"}, + {"id": "750000000000002", "state": "InProgress", "operation": "update"} + ] + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobs, err := force.GetBulk2IngestJobs() + if err != nil { + t.Fatalf("GetBulk2IngestJobs failed: %v", err) + } + + if len(jobs) != 2 { + t.Errorf("Expected 2 jobs, got %d", len(jobs)) + } +} + +func TestGetBulk2IngestJobs_WithPagination(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + + if callCount == 1 { + w.Write([]byte(`{ + "done": false, + "nextRecordsUrl": "/services/data/v55.0/jobs/ingest?locator=abc", + "records": [ + {"id": "750000000000001", "state": "JobComplete"} + ] + }`)) + } else { + w.Write([]byte(`{ + "done": true, + "records": [ + {"id": "750000000000002", "state": "InProgress"} + ] + }`)) + } + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobs, err := force.GetBulk2IngestJobs() + if err != nil { + t.Fatalf("GetBulk2IngestJobs failed: %v", err) + } + + if len(jobs) != 2 { + t.Errorf("Expected 2 jobs after pagination, got %d", len(jobs)) + } + if callCount != 2 { + t.Errorf("Expected 2 API calls for pagination, got %d", callCount) + } +} + +func TestDeleteBulk2IngestJob(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "DELETE" { + t.Errorf("Expected DELETE request, got %s", r.Method) + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + err := force.DeleteBulk2IngestJob("7501234567890ABCDEF") + if err != nil { + t.Fatalf("DeleteBulk2IngestJob failed: %v", err) + } +} + +// ============================================================================= +// Results Tests +// ============================================================================= + +func TestGetBulk2SuccessfulResults(t *testing.T) { + expectedCSV := "sf__Id,sf__Created,Name\n001000000000001AAA,true,Test Account\n" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + if !strings.Contains(r.URL.Path, "/successfulResults") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + + w.Header().Set("Content-Type", "text/csv") + w.WriteHeader(http.StatusOK) + w.Write([]byte(expectedCSV)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + results, err := force.GetBulk2SuccessfulResults("7501234567890ABCDEF") + if err != nil { + t.Fatalf("GetBulk2SuccessfulResults failed: %v", err) + } + + if string(results) != expectedCSV { + t.Errorf("Expected results '%s', got '%s'", expectedCSV, string(results)) + } +} + +func TestGetBulk2FailedResults(t *testing.T) { + expectedCSV := "sf__Id,sf__Error,Name\n,REQUIRED_FIELD_MISSING:Name required,\n" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + if !strings.Contains(r.URL.Path, "/failedResults") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + + w.Header().Set("Content-Type", "text/csv") + w.WriteHeader(http.StatusOK) + w.Write([]byte(expectedCSV)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + results, err := force.GetBulk2FailedResults("7501234567890ABCDEF") + if err != nil { + t.Fatalf("GetBulk2FailedResults failed: %v", err) + } + + if string(results) != expectedCSV { + t.Errorf("Expected results '%s', got '%s'", expectedCSV, string(results)) + } +} + +func TestGetBulk2UnprocessedRecords(t *testing.T) { + expectedCSV := "Name,Description\nUnprocessed Account,Pending\n" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + if !strings.Contains(r.URL.Path, "/unprocessedrecords") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + + w.Header().Set("Content-Type", "text/csv") + w.WriteHeader(http.StatusOK) + w.Write([]byte(expectedCSV)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + results, err := force.GetBulk2UnprocessedRecords("7501234567890ABCDEF") + if err != nil { + t.Fatalf("GetBulk2UnprocessedRecords failed: %v", err) + } + + if string(results) != expectedCSV { + t.Errorf("Expected results '%s', got '%s'", expectedCSV, string(results)) + } +} + +func TestGetBulk2SuccessfulResults_Empty(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/csv") + w.WriteHeader(http.StatusOK) + w.Write([]byte("")) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + results, err := force.GetBulk2SuccessfulResults("7501234567890ABCDEF") + if err != nil { + t.Fatalf("GetBulk2SuccessfulResults failed: %v", err) + } + + if len(results) != 0 { + t.Errorf("Expected empty results, got '%s'", string(results)) + } +} + +// ============================================================================= +// Query Job Tests +// ============================================================================= + +func TestCreateBulk2QueryJob(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + t.Errorf("Expected POST request, got %s", r.Method) + } + if !strings.HasSuffix(r.URL.Path, "/jobs/query") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"query":"SELECT Id, Name FROM Account"`) { + t.Errorf("Request body should contain query field, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "id": "7501234567890QUERY", + "operation": "query", + "object": "Account", + "state": "UploadComplete", + "apiVersion": 55.0 + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2QueryJobRequest{ + Operation: Bulk2OperationQuery, + Query: "SELECT Id, Name FROM Account", + } + + jobInfo, err := force.CreateBulk2QueryJob(request) + if err != nil { + t.Fatalf("CreateBulk2QueryJob failed: %v", err) + } + + if jobInfo.Id != "7501234567890QUERY" { + t.Errorf("Expected job ID '7501234567890QUERY', got '%s'", jobInfo.Id) + } +} + +func TestCreateBulk2QueryJob_QueryAll(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"operation":"queryAll"`) { + t.Errorf("Request body should contain operation:queryAll, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "UploadComplete", "operation": "queryAll"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2QueryJobRequest{ + Operation: Bulk2OperationQueryAll, + Query: "SELECT Id FROM Account", + } + + jobInfo, err := force.CreateBulk2QueryJob(request) + if err != nil { + t.Fatalf("CreateBulk2QueryJob failed: %v", err) + } + + if jobInfo.Operation != Bulk2OperationQueryAll { + t.Errorf("Expected operation 'queryAll', got '%s'", jobInfo.Operation) + } +} + +func TestGetBulk2QueryJobInfo(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "id": "7501234567890QUERY", + "operation": "query", + "object": "Account", + "state": "JobComplete", + "numberRecordsProcessed": 1000 + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobInfo, err := force.GetBulk2QueryJobInfo("7501234567890QUERY") + if err != nil { + t.Fatalf("GetBulk2QueryJobInfo failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateJobComplete { + t.Errorf("Expected state 'JobComplete', got '%s'", jobInfo.State) + } + if jobInfo.NumberRecordsProcessed != 1000 { + t.Errorf("Expected 1000 records processed, got %d", jobInfo.NumberRecordsProcessed) + } +} + +func TestGetBulk2QueryResults(t *testing.T) { + expectedCSV := "Id,Name\n001000000000001AAA,Test Account\n" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + if !strings.Contains(r.URL.Path, "/results") { + t.Errorf("Unexpected URL path: %s", r.URL.Path) + } + + w.Header().Set("Content-Type", "text/csv") + w.Header().Set("Sforce-Locator", "null") + w.WriteHeader(http.StatusOK) + w.Write([]byte(expectedCSV)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + results, err := force.GetBulk2QueryResults("7501234567890QUERY", "", 0) + if err != nil { + t.Fatalf("GetBulk2QueryResults failed: %v", err) + } + + if string(results.Data) != expectedCSV { + t.Errorf("Expected data '%s', got '%s'", expectedCSV, string(results.Data)) + } + if results.Locator != "" { + t.Errorf("Expected empty locator, got '%s'", results.Locator) + } +} + +func TestGetBulk2QueryResults_WithPagination(t *testing.T) { + callCount := 0 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + w.Header().Set("Content-Type", "text/csv") + + if callCount == 1 { + if strings.Contains(r.URL.RawQuery, "locator=") { + t.Errorf("First call should not have locator, got: %s", r.URL.RawQuery) + } + w.Header().Set("Sforce-Locator", "ABC123") + w.Write([]byte("Id,Name\n001000000000001AAA,Test 1\n")) + } else { + if !strings.Contains(r.URL.RawQuery, "locator=ABC123") { + t.Errorf("Second call should have locator=ABC123, got: %s", r.URL.RawQuery) + } + w.Header().Set("Sforce-Locator", "null") + w.Write([]byte("Id,Name\n001000000000002BBB,Test 2\n")) + } + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + results, err := force.GetBulk2QueryResults("7501234567890QUERY", "", 0) + if err != nil { + t.Fatalf("GetBulk2QueryResults failed: %v", err) + } + if results.Locator != "ABC123" { + t.Errorf("Expected locator 'ABC123', got '%s'", results.Locator) + } + + results, err = force.GetBulk2QueryResults("7501234567890QUERY", "ABC123", 0) + if err != nil { + t.Fatalf("GetBulk2QueryResults failed: %v", err) + } + if results.Locator != "" { + t.Errorf("Expected empty locator, got '%s'", results.Locator) + } +} + +func TestGetBulk2QueryResults_WithMaxRecords(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.Contains(r.URL.RawQuery, "maxRecords=100") { + t.Errorf("Expected maxRecords=100, got: %s", r.URL.RawQuery) + } + + w.Header().Set("Content-Type", "text/csv") + w.Header().Set("Sforce-Locator", "null") + w.WriteHeader(http.StatusOK) + w.Write([]byte("Id,Name\n001test,Test\n")) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + _, err := force.GetBulk2QueryResults("7501234567890QUERY", "", 100) + if err != nil { + t.Fatalf("GetBulk2QueryResults failed: %v", err) + } +} + +func TestAbortBulk2QueryJob(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "PATCH" { + t.Errorf("Expected PATCH request, got %s", r.Method) + } + + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"state": "Aborted"`) { + t.Errorf("Request body should contain state=Aborted, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "Aborted"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobInfo, err := force.AbortBulk2QueryJob("750test") + if err != nil { + t.Fatalf("AbortBulk2QueryJob failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateAborted { + t.Errorf("Expected state 'Aborted', got '%s'", jobInfo.State) + } +} + +func TestDeleteBulk2QueryJob(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "DELETE" { + t.Errorf("Expected DELETE request, got %s", r.Method) + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + err := force.DeleteBulk2QueryJob("750test") + if err != nil { + t.Fatalf("DeleteBulk2QueryJob failed: %v", err) + } +} + +func TestGetBulk2QueryJobs(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("Expected GET request, got %s", r.Method) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{ + "done": true, + "records": [ + {"id": "750000000000001", "state": "JobComplete", "operation": "query"}, + {"id": "750000000000002", "state": "InProgress", "operation": "queryAll"} + ] + }`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobs, err := force.GetBulk2QueryJobs() + if err != nil { + t.Fatalf("GetBulk2QueryJobs failed: %v", err) + } + + if len(jobs) != 2 { + t.Errorf("Expected 2 jobs, got %d", len(jobs)) + } +} + +// ============================================================================= +// Wait/Polling Tests +// ============================================================================= + +func TestWaitForBulk2IngestJob(t *testing.T) { + callCount := int32(0) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&callCount, 1) + count := atomic.LoadInt32(&callCount) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + + if count < 3 { + w.Write([]byte(`{"id": "750test", "state": "InProgress", "numberRecordsProcessed": 50}`)) + } else { + w.Write([]byte(`{"id": "750test", "state": "JobComplete", "numberRecordsProcessed": 100}`)) + } + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + var callbackCount int + callback := func(jobInfo any) { + callbackCount++ + } + + jobInfo, err := force.WaitForBulk2IngestJob("750test", 10*time.Millisecond, callback) + if err != nil { + t.Fatalf("WaitForBulk2IngestJob failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateJobComplete { + t.Errorf("Expected state 'JobComplete', got '%s'", jobInfo.State) + } + if callbackCount < 3 { + t.Errorf("Expected at least 3 callback calls, got %d", callbackCount) + } +} + +func TestWaitForBulk2IngestJob_Failed(t *testing.T) { + server := 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(`{"id": "750test", "state": "Failed", "errorMessage": "Invalid data"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobInfo, err := force.WaitForBulk2IngestJob("750test", 10*time.Millisecond, nil) + if err != nil { + t.Fatalf("WaitForBulk2IngestJob failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateFailed { + t.Errorf("Expected state 'Failed', got '%s'", jobInfo.State) + } + if jobInfo.ErrorMessage != "Invalid data" { + t.Errorf("Expected error message 'Invalid data', got '%s'", jobInfo.ErrorMessage) + } +} + +func TestWaitForBulk2IngestJobWithContext_Canceled(t *testing.T) { + server := 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(`{"id": "750test", "state": "InProgress"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + _, err := force.WaitForBulk2IngestJobWithContext(ctx, "750test", 100*time.Millisecond, nil) + if err == nil { + t.Fatal("Expected error for canceled context") + } + if !strings.Contains(err.Error(), "canceled") { + t.Errorf("Expected canceled error, got: %v", err) + } +} + +func TestWaitForBulk2QueryJob(t *testing.T) { + callCount := int32(0) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&callCount, 1) + count := atomic.LoadInt32(&callCount) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + + if count < 2 { + w.Write([]byte(`{"id": "750test", "state": "InProgress"}`)) + } else { + w.Write([]byte(`{"id": "750test", "state": "JobComplete", "numberRecordsProcessed": 500}`)) + } + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + jobInfo, err := force.WaitForBulk2QueryJob("750test", 10*time.Millisecond, nil) + if err != nil { + t.Fatalf("WaitForBulk2QueryJob failed: %v", err) + } + + if jobInfo.State != Bulk2JobStateJobComplete { + t.Errorf("Expected state 'JobComplete', got '%s'", jobInfo.State) + } +} + +// ============================================================================= +// Warning Header Tests +// ============================================================================= + +func TestParseBulk2Warnings(t *testing.T) { + resp := &http.Response{ + Header: http.Header{ + "Warning": []string{ + "199 - Deprecated API version", + "199 - Feature will be removed", + }, + }, + } + + warnings := ParseBulk2Warnings(resp) + if len(warnings) != 2 { + t.Errorf("Expected 2 warnings, got %d", len(warnings)) + } + if warnings[0] != "199 - Deprecated API version" { + t.Errorf("Unexpected warning: %s", warnings[0]) + } +} + +func TestParseBulk2Warnings_NoWarnings(t *testing.T) { + resp := &http.Response{ + Header: http.Header{}, + } + + warnings := ParseBulk2Warnings(resp) + if len(warnings) != 0 { + t.Errorf("Expected 0 warnings, got %d", len(warnings)) + } +} + +// ============================================================================= +// JSON Serialization Tests +// ============================================================================= + +func TestBulk2IngestJobRequest_JSON(t *testing.T) { + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationUpsert, + ExternalIdFieldName: "External_Id__c", + ColumnDelimiter: Bulk2DelimiterTab, + LineEnding: Bulk2LineEndingCRLF, + } + + data, err := json.Marshal(request) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + var parsed map[string]any + json.Unmarshal(data, &parsed) + + if parsed["object"] != "Account" { + t.Errorf("Expected object=Account, got %v", parsed["object"]) + } + if parsed["operation"] != "upsert" { + t.Errorf("Expected operation=upsert, got %v", parsed["operation"]) + } + if parsed["externalIdFieldName"] != "External_Id__c" { + t.Errorf("Expected externalIdFieldName=External_Id__c, got %v", parsed["externalIdFieldName"]) + } +} + +func TestBulk2QueryJobRequest_JSON(t *testing.T) { + request := Bulk2QueryJobRequest{ + Operation: Bulk2OperationQueryAll, + Query: "SELECT Id FROM Account WHERE IsDeleted = true", + ColumnDelimiter: Bulk2DelimiterPipe, + } + + data, err := json.Marshal(request) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + + var parsed map[string]any + json.Unmarshal(data, &parsed) + + if parsed["operation"] != "queryAll" { + t.Errorf("Expected operation=queryAll, got %v", parsed["operation"]) + } + if parsed["columnDelimiter"] != "PIPE" { + t.Errorf("Expected columnDelimiter=PIPE, got %v", parsed["columnDelimiter"]) + } +} + +// ============================================================================= +// Edge Case Tests +// ============================================================================= + +func TestCreateBulk2IngestJob_DefaultContentType(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"contentType":"CSV"`) { + t.Errorf("Request body should contain contentType:CSV, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "Open"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationInsert, + } + + _, err := force.CreateBulk2IngestJob(request) + if err != nil { + t.Fatalf("CreateBulk2IngestJob failed: %v", err) + } +} + +func TestCreateBulk2QueryJob_DefaultOperation(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"operation":"query"`) { + t.Errorf("Request body should contain operation:query by default, got: %s", body) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"id": "750test", "state": "UploadComplete"}`)) + })) + defer server.Close() + + force := &Force{ + Credentials: &ForceSession{ + InstanceUrl: server.URL, + AccessToken: "test-token", + }, + } + + request := Bulk2QueryJobRequest{ + Query: "SELECT Id FROM Account", + } + + _, err := force.CreateBulk2QueryJob(request) + if err != nil { + t.Fatalf("CreateBulk2QueryJob failed: %v", err) + } +} + +func TestAllBulk2Delimiters(t *testing.T) { + delimiters := []Bulk2ColumnDelimiter{ + Bulk2DelimiterComma, + Bulk2DelimiterTab, + Bulk2DelimiterPipe, + Bulk2DelimiterSemicolon, + Bulk2DelimiterCaret, + Bulk2DelimiterBackquote, + } + + for _, d := range delimiters { + t.Run(string(d), func(t *testing.T) { + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationInsert, + ColumnDelimiter: d, + } + data, err := json.Marshal(request) + if err != nil { + t.Fatalf("Failed to marshal with delimiter %s: %v", d, err) + } + if !strings.Contains(string(data), string(d)) { + t.Errorf("JSON should contain delimiter %s, got: %s", d, data) + } + }) + } +} + +func TestAllBulk2LineEndings(t *testing.T) { + lineEndings := []Bulk2LineEnding{ + Bulk2LineEndingLF, + Bulk2LineEndingCRLF, + } + + for _, le := range lineEndings { + t.Run(string(le), func(t *testing.T) { + request := Bulk2IngestJobRequest{ + Object: "Account", + Operation: Bulk2OperationInsert, + LineEnding: le, + } + data, err := json.Marshal(request) + if err != nil { + t.Fatalf("Failed to marshal with line ending %s: %v", le, err) + } + if !strings.Contains(string(data), string(le)) { + t.Errorf("JSON should contain line ending %s, got: %s", le, data) + } + }) + } +}