client_unit_test.go

package httpsqs

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strconv"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func mockHandler(auth string) http.HandlerFunc {
	type mockQueue struct {
		items    []string
		getIdx   int
		putPos   int64
		getPos   int64
		maxQueue int64
	}

	queues := make(map[string]*mockQueue)
	var mu sync.Mutex

	return func(w http.ResponseWriter, r *http.Request) {
		if auth != "" && r.URL.Query().Get("auth") != auth {
			_, _ = w.Write([]byte("HTTPSQS_AUTH_FAILED"))
			return
		}

		opt := r.URL.Query().Get("opt")
		name := r.URL.Query().Get("name")

		mu.Lock()
		defer mu.Unlock()

		q := queues[name]
		if q == nil {
			q = &mockQueue{maxQueue: 1000000}
			queues[name] = q
		}

		switch opt {
		case "put":
			data := r.URL.Query().Get("data")
			if data == "" {
				_, _ = w.Write([]byte("HTTPSQS_PUT_ERROR"))
				return
			}
			q.putPos++
			q.items = append(q.items, data)
			w.Header().Set("pos", strconv.FormatInt(q.putPos, 10))
			_, _ = w.Write([]byte("HTTPSQS_PUT_OK"))
		case "get":
			if q.getIdx >= len(q.items) {
				_, _ = w.Write([]byte("HTTPSQS_GET_END"))
				return
			}
			data := q.items[q.getIdx]
			q.getIdx++
			q.getPos++
			w.Header().Set("pos", strconv.FormatInt(q.getPos, 10))
			_, _ = w.Write([]byte(data))
		case "status_json":
			unread := int64(len(q.items) - q.getIdx)
			_ = json.NewEncoder(w).Encode(Status{
				Name:     name,
				MaxQueue: q.maxQueue,
				PutPos:   q.putPos,
				GetPos:   q.getPos,
				Unread:   unread,
			})
		case "view":
			pos, _ := strconv.ParseInt(r.URL.Query().Get("pos"), 10, 64)
			if pos > 0 && int(pos) <= len(q.items) {
				_, _ = w.Write([]byte(q.items[pos-1]))
			}
		case "reset":
			*q = mockQueue{maxQueue: 1000000}
			_, _ = w.Write([]byte("HTTPSQS_RESET_OK"))
		case "maxqueue":
			num, _ := strconv.ParseInt(r.URL.Query().Get("num"), 10, 64)
			q.maxQueue = num
			_, _ = w.Write([]byte("HTTPSQS_MAXQUEUE_OK"))
		case "synctime":
			_, _ = w.Write([]byte("HTTPSQS_SYNCTIME_OK"))
		}
	}
}

func newTestClient(serverURL string, maxRetries ...int) IClient {
	u, _ := url.Parse(serverURL)
	cfg := &Config{
		Addr:    u.Host,
		Timeout: 5 * time.Second,
	}
	if len(maxRetries) > 0 {
		cfg.MaxRetries = maxRetries[0]
	}
	return NewClient(cfg)
}

func TestUnit_Put(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	pos, err := client.Put(context.Background(), "test_queue", "hello")
	assert.NoError(t, err)
	assert.Equal(t, int64(1), pos)
}

func TestUnit_PutError(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	_, err := client.Put(context.Background(), "test_queue", "")
	assert.Error(t, err)
}

func TestUnit_Get(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	ctx := context.Background()

	_, err := client.Put(ctx, "test_queue", "hello world")
	require.NoError(t, err)

	data, pos, err := client.Get(ctx, "test_queue")
	assert.NoError(t, err)
	assert.Equal(t, "hello world", data)
	assert.Equal(t, int64(1), pos)
}

func TestUnit_GetEnd(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	data, pos, err := client.Get(context.Background(), "empty_queue")
	assert.NoError(t, err)
	assert.Equal(t, "", data)
	assert.Equal(t, int64(-1), pos)
}

func TestUnit_Status(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	ctx := context.Background()

	_, err := client.Put(ctx, "test_queue", "hello")
	require.NoError(t, err)

	status, err := client.Status(ctx, "test_queue")
	assert.NoError(t, err)
	assert.Equal(t, "test_queue", status.Name)
	assert.Equal(t, int64(1), status.PutPos)
	assert.Equal(t, int64(1), status.Unread)
}

func TestUnit_View(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	ctx := context.Background()

	_, err := client.Put(ctx, "test_queue", "view data")
	require.NoError(t, err)

	data, err := client.View(ctx, "test_queue", 1)
	assert.NoError(t, err)
	assert.Equal(t, "view data", data)
}

func TestUnit_ViewInvalidPos(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)

	_, err := client.View(context.Background(), "test_queue", 0)
	assert.Error(t, err)

	_, err = client.View(context.Background(), "test_queue", -1)
	assert.Error(t, err)
}

func TestUnit_Reset(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	ctx := context.Background()

	_, err := client.Put(ctx, "test_queue", "hello")
	require.NoError(t, err)

	err = client.Reset(ctx, "test_queue")
	assert.NoError(t, err)

	status, err := client.Status(ctx, "test_queue")
	assert.NoError(t, err)
	assert.Equal(t, int64(0), status.PutPos)
}

func TestUnit_SetMaxQueue(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)
	ctx := context.Background()

	err := client.SetMaxQueue(ctx, "test_queue", 500000)
	assert.NoError(t, err)

	status, err := client.Status(ctx, "test_queue")
	assert.NoError(t, err)
	assert.Equal(t, int64(500000), status.MaxQueue)
}

func TestUnit_SetSyncTime(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)

	err := client.SetSyncTime(context.Background(), "test_queue", 10*time.Second)
	assert.NoError(t, err)
}

func TestUnit_SetSyncTimeInvalid(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	client := newTestClient(server.URL)

	err := client.SetSyncTime(context.Background(), "test_queue", 500*time.Millisecond)
	assert.Error(t, err)
}

func TestUnit_AuthFailed(t *testing.T) {
	server := httptest.NewServer(mockHandler("secret"))
	defer server.Close()

	client := newTestClient(server.URL)
	_, err := client.Put(context.Background(), "test_queue", "hello")
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "auth")
}

func TestUnit_Ordinal(t *testing.T) {
	tests := []struct {
		n   int64
		exp string
	}{
		{1, "1st"}, {2, "2nd"}, {3, "3rd"}, {4, "4th"},
		{11, "11th"}, {12, "12th"}, {13, "13th"},
		{21, "21st"}, {22, "22nd"}, {23, "23rd"},
		{111, "111th"}, {112, "112th"}, {113, "113th"},
	}
	for _, tt := range tests {
		assert.Equal(t, tt.exp, ordinal(tt.n))
	}
}

func TestUnit_StatusString(t *testing.T) {
	s := Status{
		Name:     "test",
		MaxQueue: 1000000,
		PutPos:   100,
		PutLap:   2,
		GetPos:   50,
		GetLap:   1,
		Unread:   50,
	}
	str := s.String()
	assert.Contains(t, str, "2nd lap")
	assert.Contains(t, str, "1st lap")
}

type failRoundTripper struct {
	core       http.RoundTripper
	failBefore int32
	count      atomic.Int32
}

func (rt *failRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
	if rt.count.Add(1) <= rt.failBefore {
		return nil, fmt.Errorf("simulated connection error")
	}
	return rt.core.RoundTrip(req)
}

func TestUnit_RetryOnNetworkError(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	u, _ := url.Parse(server.URL)
	c := NewClient(&Config{
		Addr:       u.Host,
		Timeout:    5 * time.Second,
		MaxRetries: 3,
	}).(*client)

	c.httpClient.Transport = &failRoundTripper{
		core:       http.DefaultTransport,
		failBefore: 2,
	}

	pos, err := c.Put(context.Background(), "test_queue", "hello")
	assert.NoError(t, err)
	assert.Equal(t, int64(1), pos)
}

func TestUnit_RetryExhausted(t *testing.T) {
	server := httptest.NewServer(mockHandler(""))
	defer server.Close()

	u, _ := url.Parse(server.URL)
	c := NewClient(&Config{
		Addr:       u.Host,
		Timeout:    time.Second,
		MaxRetries: 2,
	}).(*client)

	c.httpClient.Transport = &failRoundTripper{
		core:       http.DefaultTransport,
		failBefore: 10,
	}

	_, err := c.Put(context.Background(), "test_queue", "hello")
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "retries")
}