opensearch_integration_test.go

// SPDX-License-Identifier: Apache-2.0
//
// The OpenSearch Contributors require contributions made to
// this file be licensed under the Apache-2.0 license or a
// compatible open source license.
//
// Modifications Copyright OpenSearch Contributors. See
// GitHub history for details.

// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//go:build integration && core && !multinode

package opensearch_test

import (
	"bytes"
	"context"
	"crypto/tls"
	"errors"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/opensearch-project/opensearch-go/v5"
	"github.com/opensearch-project/opensearch-go/v5/internal/build"
	"github.com/opensearch-project/opensearch-go/v5/opensearchapi"
	"github.com/opensearch-project/opensearch-go/v5/opensearchapi/testutil"
	"github.com/opensearch-project/opensearch-go/v5/opensearchtransport"
	"github.com/opensearch-project/opensearch-go/v5/opensearchtransport/testutil/mockhttp"
)

func TestClientTransport(t *testing.T) {
	/*
		t.Run("Persistent", func(t *testing.T) {
			client, err := testutil.NewClient(t)
			if err != nil {
				t.Fatalf("Error creating the client: %s", err)
			}

			var total int

			for i := 0; i < 101; i++ {
				var curTotal int

				res, err := client.Nodes.Stats(nil, &opensearchapi.NodesStatsReq{Metric: []string{"http"}})
				if err != nil {
					t.Fatalf("Unexpected error: %s", err)
				}

				for _, v := range res.Nodes {
					curTotal = v.HTTP.TotalOpened
					break
				}

				if curTotal < 1 {
					t.Errorf("Unexpected total_opened: %d", curTotal)
				}
				if total == 0 {
					total = curTotal
				}

				if total != curTotal {
					t.Errorf("Expected total_opened=%d, got: %d", total, curTotal)
				}
			}

			log.Printf("total_opened: %d", total)
		})
	*/

	t.Run("Concurrent", func(t *testing.T) {
		var wg sync.WaitGroup

		client, err := testutil.InitClient(t)
		require.NoError(t, err)

		for range 101 {
			time.Sleep(10 * time.Millisecond)

			wg.Go(func() {
				_, err := client.Info(t.Context(), nil)
				require.NoError(t, err)
			})
		}
		wg.Wait()
	})

	t.Run("WithContext", func(t *testing.T) {
		client, err := testutil.InitClient(t)
		require.NoError(t, err)

		ctx, cancel := context.WithTimeout(context.Background(), time.Nanosecond)
		defer cancel()

		_, err = client.Info(ctx, nil)
		require.Error(t, err, "Expected context deadline exceeded error")
	})

	t.Run("Configured", func(t *testing.T) {
		tp := http.DefaultTransport.(*http.Transport).Clone()
		tp.MaxIdleConnsPerHost = 10
		tp.ResponseHeaderTimeout = time.Second
		tp.DialContext = (&net.Dialer{Timeout: time.Nanosecond}).DialContext
		tp.TLSClientConfig.MinVersion = tls.VersionTLS11
		tp.TLSClientConfig.InsecureSkipVerify = true

		cfg := opensearchapi.Config{
			Client: opensearch.Config{
				Context:   t.Context(),
				Transport: tp,
			},
		}

		client, err := opensearchapi.NewClient(cfg)
		require.NoError(t, err)

		_, err = client.Info(t.Context(), nil)
		require.Error(t, err)
		opError := &net.OpError{}
		if !errors.As(err, &opError) {
			t.Fatalf("Expected net.OpError, but got: %T", err)
		}
	})
}

type CustomTransport struct {
	client *http.Client
	logger func(format string, v ...any)
}

func (t *CustomTransport) RoundTrip(req *http.Request) (*http.Response, error) {
	req.Header.Set("X-Foo", "bar")
	if t.logger != nil {
		t.logger("> %s %q %q", req.Method, req.URL.String(), req.Header)
	}
	return t.client.Do(req)
}

func TestClientCustomTransport(t *testing.T) {
	t.Run("Customized", func(t *testing.T) {
		client, err := opensearchapi.NewDefaultClient()
		require.NoError(t, err)

		// Stop background pollers when the test finishes.
		if closer, ok := client.Client.Transport.(io.Closer); ok {
			t.Cleanup(func() { closer.Close() })
		}

		cfg := testutil.ClientConfig(t)

		if cfg != nil {
			customTP := http.DefaultTransport.(*http.Transport).Clone()
			customTP.TLSClientConfig.InsecureSkipVerify = true
			cfg.Client.Transport = &CustomTransport{
				client: &http.Client{
					Transport: customTP,
				},
				logger: func(format string, v ...any) {
					if testutil.IsDebugEnabled(t) {
						t.Logf(format, v...)
					}
				},
			}
			client, err = opensearchapi.NewClient(*cfg)
			require.NoError(t, err)

			// Wait for cluster to be ready before running tests
			testutil.WaitForClusterReady(t, client)
		}

		// Simple readiness wait for manually-constructed client (only uses Info API)
		ctx := t.Context()
		for {
			_, err := client.Info(ctx, nil)
			if err == nil {
				break
			}
			select {
			case <-ctx.Done():
				t.Fatalf("Cluster not ready: %s", ctx.Err())
			case <-time.After(5 * time.Second):
				// Retry
			}
		}
	})

	t.Run("Manual", func(t *testing.T) {
		config := testutil.ClientConfig(t)

		// Use centralized URL construction
		u := mockhttp.GetOpenSearchURL(t)
		tp, _ := opensearchtransport.New(opensearchtransport.Config{
			URLs:      []*url.URL{u},
			Transport: config.Client.Transport,
			Username:  config.Client.Username,
			Password:  config.Client.Password,
			Context:   t.Context(),
		})

		client := opensearchapi.Client{
			Client: &opensearch.Client{
				Transport: tp,
			},
		}

		// Simple readiness wait for manually-constructed client (only uses Info API)
		ctx := t.Context()
		for {
			_, err := client.Info(ctx, nil)
			if err == nil {
				break
			}
			select {
			case <-ctx.Done():
				t.Fatalf("Cluster not ready: %s", ctx.Err())
			case <-time.After(5 * time.Second):
				// Retry
			}
		}
	})
}

type TestTransport struct {
	counter atomic.Uint64
	t       *testing.T
}

func (tr *TestTransport) Stream(req *http.Request) (*http.Response, error) {
	// Use centralized URL construction
	u := mockhttp.GetOpenSearchURL(tr.t)
	req.URL.Scheme = u.Scheme
	req.URL.Host = u.Host

	config := testutil.ClientConfig(tr.t)
	if testutil.IsSecure(tr.t) {
		req.SetBasicAuth(config.Client.Username, config.Client.Password)
	}

	tr.counter.Add(1)
	transport := config.Client.Transport
	if transport == nil {
		transport = http.DefaultTransport
	}
	return transport.RoundTrip(req)
}

func (tr *TestTransport) Count() uint64 {
	return tr.counter.Load()
}

// Request buffers the response body so *TestTransport satisfies
// opensearchtransport.Interface alongside Stream.
func (tr *TestTransport) Request(req *http.Request) (*http.Response, error) {
	res, err := tr.Stream(req)
	if res != nil && res.Body != nil {
		body, rerr := io.ReadAll(res.Body)
		res.Body.Close()
		res.Body = io.NopCloser(bytes.NewReader(body))
		if rerr != nil && err == nil {
			err = rerr
		}
	}
	return res, err
}

func TestClientReplaceTransport(t *testing.T) {
	t.Run("Replaced", func(t *testing.T) {
		const expectedRequests = 10

		tr := &TestTransport{t: t}
		client := opensearchapi.Client{
			Client: &opensearch.Client{
				Transport: tr,
			},
		}

		// Simple readiness wait for manually-constructed client (only uses Info API)
		ctx := t.Context()
		for {
			_, err := client.Info(ctx, nil)
			if err == nil {
				break
			}
			select {
			case <-ctx.Done():
				t.Fatalf("Cluster not ready: %v", ctx.Err())
			case <-time.After(5 * time.Second):
				// Retry
			}
		}

		// Reset counter after readiness check
		initialCount := tr.Count()

		for range expectedRequests {
			_, err := client.Info(t.Context(), nil)
			require.NoError(t, err)
		}

		actualRequests := tr.Count() - initialCount
		if actualRequests > expectedRequests {
			t.Errorf("Expected at most %d requests, got=%d", expectedRequests, actualRequests)
		}
	})
}

func TestClientAPI(t *testing.T) {
	t.Run("Info", func(t *testing.T) {
		client, err := testutil.NewClient(t)
		require.NoError(t, err)

		res, err := client.Info(t.Context(), nil)
		require.NoError(t, err)
		if res.ClusterName == "" {
			log.Fatalf("cluster_name is empty: %s\n", err)
		}
	})
}

func TestClientGetConfigIntegration(t *testing.T) {
	t.Run("GetConfig returns valid configuration", func(t *testing.T) {
		// Get test config
		cfg := testutil.ClientConfig(t)

		// Create a client with specific configuration
		osClient, err := opensearch.NewClient(cfg.Client)
		require.NoError(t, err)

		// Retrieve the config
		retrievedConfig := osClient.GetConfig()

		// Verify the config matches what was originally provided
		require.Equal(t, cfg.Client.Addresses, retrievedConfig.Addresses)
		require.Equal(t, cfg.Client.Username, retrievedConfig.Username)
		require.Equal(t, cfg.Client.Password, retrievedConfig.Password)
	})

	t.Run("GetConfig with live client", func(t *testing.T) {
		// Create a client from test helper
		apiClient, err := testutil.NewClient(t)
		require.NoError(t, err)

		// Get config from the underlying opensearch client
		config := apiClient.Client.GetConfig()

		// Verify config has expected values
		require.NotEmpty(t, config.Addresses, "addresses should not be empty")

		// Verify we can create a new client with the retrieved config
		newClient, err := opensearch.NewClient(*config)
		require.NoError(t, err)
		require.NotNil(t, newClient)

		// Verify the new client works by making a request
		req, err := build.Request(http.MethodGet, "/", nil, nil, nil)
		require.NoError(t, err)

		resp, err := newClient.Stream(req)
		require.NoError(t, err)
		require.NotNil(t, resp)
		defer resp.Body.Close()
	})
}

func TestNewFromClientIntegration(t *testing.T) {
	t.Run("creates working api client from config", func(t *testing.T) {
		// Get test config
		cfg := testutil.ClientConfig(t)

		// Create an opensearchapi.Client from the shared config
		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
		require.NoError(t, err)
		require.NotNil(t, apiClient)

		// Verify the api client can make requests
		resp, err := apiClient.Info(t.Context(), nil)
		require.NoError(t, err)
		require.NotEmpty(t, resp)
		require.NotEmpty(t, resp.ClusterName)
	})

	t.Run("api client shares config-derived transport with a base client", func(t *testing.T) {
		// Get test config
		cfg := testutil.ClientConfig(t)

		// Create a base opensearch.Client and an api client from the same config
		osClient, err := opensearch.NewClient(cfg.Client)
		require.NoError(t, err)

		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
		require.NoError(t, err)
		require.NotNil(t, apiClient.Client.Transport)

		// Verify both clients can make requests successfully
		req, err := build.Request(http.MethodGet, "/", nil, nil, nil)
		require.NoError(t, err)

		resp1, err := osClient.Stream(req)
		require.NoError(t, err)
		require.NotNil(t, resp1)
		defer resp1.Body.Close()

		resp2, err := apiClient.Info(t.Context(), nil)
		require.NoError(t, err)
		require.NotNil(t, resp2)
	})

	t.Run("maintains config through wrapped client", func(t *testing.T) {
		// Get test config
		cfg := testutil.ClientConfig(t)

		// Create an opensearchapi.Client from the shared config
		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
		require.NoError(t, err)

		// Retrieve config through the api client's wrapped opensearch client
		retrievedConfig := apiClient.Client.GetConfig()

		// Verify the config matches the original
		require.Equal(t, cfg.Client.Addresses, retrievedConfig.Addresses)
		require.Equal(t, cfg.Client.Username, retrievedConfig.Username)
		require.Equal(t, cfg.Client.Password, retrievedConfig.Password)
	})

	t.Run("all sub-clients are functional", func(t *testing.T) {
		// Get test config
		cfg := testutil.ClientConfig(t)

		// Create an opensearchapi.Client from the shared config
		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
		require.NoError(t, err)

		// Test a few sub-clients to ensure they're properly initialized
		// Cat client
		catResp, err := apiClient.Cat.Health(t.Context(), nil)
		require.NoError(t, err)
		require.NotNil(t, catResp)

		// Cluster client
		clusterResp, err := apiClient.Cluster.Health(t.Context(), nil)
		require.NoError(t, err)
		require.NotNil(t, clusterResp)

		// Nodes client
		nodesResp, err := apiClient.Nodes.Info(t.Context(), nil)
		require.NoError(t, err)
		require.NotNil(t, nodesResp)
	})
}