fix tests

sean- · sean- · commit b551cb97d88a · 2026-02-06T12:05:57.000-08:00
Signed-off-by: Sean Chittenden &lt;sean.chittenden@crowdstrike.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -43,6 +43,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ### Fixed
 
 - Fix flaky connection integration test by replacing arbitrary sleep times with proper server readiness polling
+- Fix OpenSearch 2.8.0+ Tasks API compatibility by adding cancellation_time_millis field to TasksListTask struct
 - Fix OpenSearch 3.1.0+ API compatibility by adding phase_results_processors field to nodes API and time_in_execution fields to cluster pending tasks API
 - Fix OpenSearch 3.2.0+ API compatibility by adding max_last_index_request_timestamp and startree query fields across nodes stats, indices stats, and cat APIs, plus settings field to security plugin health API
 - Fix OpenSearch 3.3.0+ API compatibility by adding neural_search breaker, query_failed and startree_query_failed search fields, search pipeline system_generated fields across multiple APIs, plus ingestion_status field to cluster state API and jwks_uri field to security config API
diff --git a/opensearchapi/api_mget_test.go b/opensearchapi/api_mget_test.go
@@ -53,7 +53,7 @@ func TestMGet(t *testing.T) {
 			context.Background(),
 			opensearchapi.MGetReq{
 				Index: testIndex,
-				Body:  strings.NewReader(`{"docs":[{"_id":"1"},{"_id":"2"}]}`),
+				Body:  strings.NewReader(fmt.Sprintf(`{"docs":[{"_id":"%s-1"},{"_id":"%s-2"}]}`, docIDPrefix, docIDPrefix)),
 			},
 		)
 		require.Nil(t, err)
diff --git a/opensearchapi/api_mtermvectors_test.go b/opensearchapi/api_mtermvectors_test.go
@@ -26,7 +26,7 @@ func TestMTermvectors(t *testing.T) {
 	client, err := ostest.NewClient(t)
 	require.Nil(t, err)
 
-	testIndex := "test-mtermvectors"
+	testIndex := testutil.MustUniqueString(t, "test-mtermvectors")
 	t.Cleanup(func() {
 		client.Indices.Delete(t.Context(), opensearchapi.IndicesDeleteReq{Indices: []string{testIndex}})
 	})
@@ -95,7 +95,7 @@ func TestMTermvectors(t *testing.T) {
 			t.Context(),
 			opensearchapi.MTermvectorsReq{
 				Index: testIndex,
-				Body:  strings.NewReader(`{"ids":[1,2]}`),
+				Body:  strings.NewReader(fmt.Sprintf(`{"ids":["%s-0","%s-1"]}`, docIDPrefix, docIDPrefix)),
 			},
 		)
 		require.Nil(t, err)
diff --git a/opensearchapi/api_tasks-list.go b/opensearchapi/api_tasks-list.go
@@ -55,19 +55,20 @@ type TasksListNodes struct {
 
 // TasksListTask is a sub type of TaskListResp, TaskListNodes containing information about a task
 type TasksListTask struct {
-	Node               string                 `json:"node"`
-	ID                 int                    `json:"id"`
-	Type               string                 `json:"type"`
-	Action             string                 `json:"action"`
-	Description        string                 `json:"description"`
-	StartTimeInMillis  int64                  `json:"start_time_in_millis"`
-	RunningTimeInNanos int64                  `json:"running_time_in_nanos"`
-	Cancellable        bool                   `json:"cancellable"`
-	Cancelled          bool                   `json:"cancelled"`
-	Headers            map[string]string      `json:"headers"`
-	ResourceStats      TasksListResourceStats `json:"resource_stats"`
-	ParentTaskID       string                 `json:"parent_task_id"`
-	Children           []TasksListTask        `json:"children,omitempty"`
+	Node                   string                 `json:"node"`
+	ID                     int                    `json:"id"`
+	Type                   string                 `json:"type"`
+	Action                 string                 `json:"action"`
+	Description            string                 `json:"description"`
+	StartTimeInMillis      int64                  `json:"start_time_in_millis"`
+	RunningTimeInNanos     int64                  `json:"running_time_in_nanos"`
+	Cancellable            bool                   `json:"cancellable"`
+	Cancelled              bool                   `json:"cancelled"`
+	CancellationTimeMillis *int64                 `json:"cancellation_time_millis,omitempty"` // Added in OpenSearch 2.8.0
+	Headers                map[string]string      `json:"headers"`
+	ResourceStats          TasksListResourceStats `json:"resource_stats"`
+	ParentTaskID           string                 `json:"parent_task_id"`
+	Children               []TasksListTask        `json:"children,omitempty"`
 }
 
 // TasksListResourceStats is a sub type of TaskListTask containing information about task stats
diff --git a/opensearchapi/api_tasks_test.go b/opensearchapi/api_tasks_test.go
@@ -152,10 +152,10 @@ func TestTasksClient(t *testing.T) {
 	}
 	for _, value := range testCases {
 		t.Run(value.Name, func(t *testing.T) {
-			t.Parallel()
+			// Do not run subtests in parallel - they depend on the reindex task state
 			for _, testCase := range value.Tests {
 				t.Run(testCase.Name, func(t *testing.T) {
-					t.Parallel()
+					// Do not run in parallel - task may complete before Get/Cancel tests run
 					res, err := testCase.Results()
 					if testCase.Name == "inspect" {
 						assert.NotNil(t, err)
diff --git a/opensearchapi/api_termvectors_test.go b/opensearchapi/api_termvectors_test.go
@@ -27,7 +27,7 @@ func TestTermvectors(t *testing.T) {
 	client, err := ostest.NewClient(t)
 	require.Nil(t, err)
 
-	testIndex := "test-termvectors"
+	testIndex := testutil.MustUniqueString(t, "test-termvectors")
 	t.Cleanup(func() {
 		client.Indices.Delete(t.Context(), opensearchapi.IndicesDeleteReq{Indices: []string{testIndex}})
 	})
@@ -96,7 +96,7 @@ func TestTermvectors(t *testing.T) {
 			context.Background(),
 			opensearchapi.TermvectorsReq{
 				Index:      testIndex,
-				DocumentID: "1",
+				DocumentID: fmt.Sprintf("%s-%d", docIDPrefix, 0),
 				Body: strings.NewReader(`{"fields":["*"],"offsets":true,"payloads":true,"positions":true,` +
 					`"term_statistics":true,"field_statistics":true}`),
 			},
diff --git a/opensearchapi/api_update_test.go b/opensearchapi/api_update_test.go
@@ -54,7 +54,7 @@ func TestUpdate(t *testing.T) {
 			opensearchapi.UpdateReq{
 				Params:     opensearchapi.UpdateParams{Source: true},
 				Index:      testIndex,
-				DocumentID: "1",
+				DocumentID: fmt.Sprintf("%s-%d", docIDPrefix, 1),
 				Body:       strings.NewReader(`{"script":{"source":"ctx._source.counter += params.count","lang":"painless","params":{"count":4}}}`),
 			},
 		)
diff --git a/opensearchtransport/connection_internal_test.go b/opensearchtransport/connection_internal_test.go
@@ -503,7 +503,6 @@ func TestStatusConnectionPoolResurrect(t *testing.T) {
 		s.curr.Store(-1)
 
 		pool := &statusConnectionPool{
-
 			resurrectTimeoutInitial:      0,
 			resurrectTimeoutFactorCutoff: defaultResurrectTimeoutFactorCutoff,
 		}
diff --git a/opensearchtransport/discovery_internal_test.go b/opensearchtransport/discovery_internal_test.go
@@ -137,26 +137,44 @@ func TestDiscovery(t *testing.T) {
 	srvTLS1 := &http.Server{Addr: "127.0.0.1:20001", Handler: tlsMux, ReadTimeout: 1 * time.Second}
 	srvTLS2 := &http.Server{Addr: "localhost:20002", Handler: tlsMux, ReadTimeout: 1 * time.Second}
 
+	// Create listeners first to ensure ports are bound before tests run
+	ln1, err := net.Listen("tcp", srv.Addr)
+	if err != nil {
+		t.Fatalf("Failed to create listener for srv: %s", err)
+	}
+	ln2, err := net.Listen("tcp", srv2.Addr)
+	if err != nil {
+		t.Fatalf("Failed to create listener for srv2: %s", err)
+	}
+	lnTLS1, err := net.Listen("tcp", srvTLS1.Addr)
+	if err != nil {
+		t.Fatalf("Failed to create listener for srvTLS1: %s", err)
+	}
+	lnTLS2, err := net.Listen("tcp", srvTLS2.Addr)
+	if err != nil {
+		t.Fatalf("Failed to create listener for srvTLS2: %s", err)
+	}
+
 	go func() {
-		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+		if err := srv.Serve(ln1); err != nil && err != http.ErrServerClosed {
 			t.Errorf("Unable to start server: %s", err)
 			return
 		}
 	}()
 	go func() {
-		if err := srv2.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+		if err := srv2.Serve(ln2); err != nil && err != http.ErrServerClosed {
 			t.Errorf("Unable to start server2: %s", err)
 			return
 		}
 	}()
 	go func() {
-		if err := srvTLS1.ListenAndServeTLS("testdata/cert.pem", "testdata/key.pem"); err != nil && err != http.ErrServerClosed {
+		if err := srvTLS1.ServeTLS(lnTLS1, "testdata/cert.pem", "testdata/key.pem"); err != nil && err != http.ErrServerClosed {
 			t.Errorf("Unable to start TLS server1: %s", err)
 			return
 		}
 	}()
 	go func() {
-		if err := srvTLS2.ListenAndServeTLS("testdata/cert.pem", "testdata/key.pem"); err != nil && err != http.ErrServerClosed {
+		if err := srvTLS2.ServeTLS(lnTLS2, "testdata/cert.pem", "testdata/key.pem"); err != nil && err != http.ErrServerClosed {
 			t.Errorf("Unable to start TLS server2: %s", err)
 			return
 		}
@@ -661,10 +679,10 @@ func TestDiscovery(t *testing.T) {
 
 				// Add health check handler (catch-all for /{$} and /)
 				testMux.HandleFunc("/{$}", func(w http.ResponseWriter, r *http.Request) {
-					healthResp := map[string]interface{}{
+					healthResp := map[string]any{
 						"name":         "test-node",
 						"cluster_name": "test-cluster",
-						"version": map[string]interface{}{
+						"version": map[string]any{
 							"number": "2.0.0",
 						},
 					}
@@ -684,25 +702,25 @@ func TestDiscovery(t *testing.T) {
 				// Add nodes info handler with this test's data
 				testMux.HandleFunc("/_nodes/http", func(w http.ResponseWriter, r *http.Request) {
 					// Create a simple response structure compatible with the discovery parsing
-					response := map[string]interface{}{
-						"_nodes": map[string]interface{}{
+					response := map[string]any{
+						"_nodes": map[string]any{
 							"total":      len(tt.args.Nodes),
 							"successful": len(tt.args.Nodes),
 							"failed":     0,
 						},
 						"cluster_name": "test-cluster",
-						"nodes":        make(map[string]interface{}),
+						"nodes":        make(map[string]any),
 					}
 
-					nodes := response["nodes"].(map[string]interface{})
+					nodes := response["nodes"].(map[string]any)
 					for name, node := range tt.args.Nodes {
 						// Use the test server address for publish_address so health checks work
-						nodes[name] = map[string]interface{}{
+						nodes[name] = map[string]any{
 							"name":  name,
 							"host":  "127.0.0.1",
 							"ip":    "127.0.0.1",
 							"roles": node.Roles,
-							"http": map[string]interface{}{
+							"http": map[string]any{
 								"publish_address": testServer.Addr, // Point to our test server, not fictional hostnames
 							},
 						}
@@ -1056,10 +1074,10 @@ func TestDiscoverNodesWithNewRoleValidation(t *testing.T) {
 
 			// Health check endpoint - exact root path match
 			mux.HandleFunc("/{$}", func(w http.ResponseWriter, r *http.Request) {
-				healthResp := map[string]interface{}{
+				healthResp := map[string]any{
 					"name":         "test-node",
 					"cluster_name": "test-cluster",
-					"version": map[string]interface{}{
+					"version": map[string]any{
 						"number": "2.0.0",
 					},
 				}
@@ -1194,17 +1212,21 @@ func TestIncludeDedicatedClusterManagersConfiguration(t *testing.T) {
 			testMux := http.NewServeMux()
 
 			// Start a test server first so we have the address
-			testServer := &http.Server{Addr: "127.0.0.1:0", Handler: testMux}
+			testServer := &http.Server{
+				Addr:              "127.0.0.1:0",
+				Handler:           testMux,
+				ReadHeaderTimeout: 5 * time.Second,
+			}
 			listener, err := net.Listen("tcp", testServer.Addr)
 			require.NoError(t, err)
 			testServer.Addr = listener.Addr().String()
 
 			// Health check endpoint (catch-all for /{$} and /)
 			testMux.HandleFunc("/{$}", func(w http.ResponseWriter, r *http.Request) {
-				healthResp := map[string]interface{}{
+				healthResp := map[string]any{
 					"name":         "test-node",
 					"cluster_name": "test-cluster",
-					"version": map[string]interface{}{
+					"version": map[string]any{
 						"number": "2.0.0",
 					},
 				}
@@ -1215,24 +1237,24 @@ func TestIncludeDedicatedClusterManagersConfiguration(t *testing.T) {
 
 			// Nodes info endpoint
 			testMux.HandleFunc("/_nodes/http", func(w http.ResponseWriter, r *http.Request) {
-				response := map[string]interface{}{
-					"_nodes": map[string]interface{}{
+				response := map[string]any{
+					"_nodes": map[string]any{
 						"total":      len(tt.nodes),
 						"successful": len(tt.nodes),
 						"failed":     0,
 					},
 					"cluster_name": "test-cluster",
-					"nodes":        make(map[string]interface{}),
+					"nodes":        make(map[string]any),
 				}
 
-				nodes := response["nodes"].(map[string]interface{})
+				nodes := response["nodes"].(map[string]any)
 				for name, roles := range tt.nodes {
-					nodes[name] = map[string]interface{}{
+					nodes[name] = map[string]any{
 						"name":  name,
 						"host":  "127.0.0.1",
 						"ip":    "127.0.0.1",
 						"roles": roles,
-						"http": map[string]interface{}{
+						"http": map[string]any{
 							"publish_address": testServer.Addr, // Point to our test server
 						},
 					}
diff --git a/opensearchtransport/logger.go b/opensearchtransport/logger.go
@@ -203,8 +203,6 @@ func (l *ColorLogger) RequestBodyEnabled() bool { return l.EnableRequestBody }
 func (l *ColorLogger) ResponseBodyEnabled() bool { return l.EnableResponseBody }
 
 // LogRoundTrip prints the information about request and response.
-//
-//nolint:unparam // error return required by Logger interface
 func (l *CurlLogger) LogRoundTrip(req *http.Request, res *http.Response, _ error, start time.Time, dur time.Duration) error {
 	var b bytes.Buffer
 
diff --git a/opensearchtransport/opensearchtransport_internal_test.go b/opensearchtransport/opensearchtransport_internal_test.go
@@ -1552,7 +1552,7 @@ func TestConnectionPoolPromotionIntegration(t *testing.T) {
 		if router != nil {
 			// The router should be able to handle the new connections
 			// In a real scenario, the router's internal policies would be updated during discovery
-			req, _ := http.NewRequest("GET", "/", nil)
+			req, _ := http.NewRequest(http.MethodGet, "/", nil)
 			_, routeErr := router.Route(t.Context(), req)
 			// Note: This might return an error because policies aren't initialized,
 			// but it shouldn't panic or cause memory issues

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ func TestMGet(t *testing.T) {`
`53`	`53`	`context.Background(),`
`54`	`54`	`opensearchapi.MGetReq{`
`55`	`55`	`Index: testIndex,`
`56`		- Body: strings.NewReader(`{"docs":[{"_id":"1"},{"_id":"2"}]}`),
	`56`	+ Body: strings.NewReader(fmt.Sprintf(`{"docs":[{"_id":"%s-1"},{"_id":"%s-2"}]}`, docIDPrefix, docIDPrefix)),
`57`	`57`	`},`
`58`	`58`	`)`
`59`	`59`	`require.Nil(t, err)`
Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ func TestUpdate(t *testing.T) {`
`54`	`54`	`opensearchapi.UpdateReq{`
`55`	`55`	`Params: opensearchapi.UpdateParams{Source: true},`
`56`	`56`	`Index: testIndex,`
`57`		`- DocumentID: "1",`
	`57`	`+ DocumentID: fmt.Sprintf("%s-%d", docIDPrefix, 1),`
`58`	`58`	Body: strings.NewReader(`{"script":{"source":"ctx._source.counter += params.count","lang":"painless","params":{"count":4}}}`),
`59`	`59`	`},`
`60`	`60`	`)`
Original file line number	Diff line number	Diff line change
`@@ -503,7 +503,6 @@ func TestStatusConnectionPoolResurrect(t *testing.T) {`
`503`	`503`	`s.curr.Store(-1)`
`504`	`504`
`505`	`505`	`pool := &statusConnectionPool{`
`506`		`-`
`507`	`506`	`resurrectTimeoutInitial: 0,`
`508`	`507`	`resurrectTimeoutFactorCutoff: defaultResurrectTimeoutFactorCutoff,`
`509`	`508`	`}`