func TestPssSend(t *testing.T) {
	var (
		logger = logging.New(io.Discard, 0)
		mtx             sync.Mutex
		receivedTopic   pss.Topic
		receivedBytes   []byte
		receivedTargets pss.Targets
		done            bool
		privk, _       = crypto.GenerateSecp256k1Key()
		publicKeyBytes = (*btcec.PublicKey)(&privk.PublicKey).SerializeCompressed()
		sendFn = func(ctx context.Context, targets pss.Targets, chunk swarm.Chunk) error {
			mtx.Lock()
			topic, msg, err := pss.Unwrap(ctx, privk, chunk, []pss.Topic{topic})
			receivedTopic = topic
			receivedBytes = msg
			receivedTargets = targets
			done = true
			mtx.Unlock()
			return err
		}
		mp           = mockpost.New(mockpost.WithIssuer(postage.NewStampIssuer("", "", batchOk, big.NewInt(3), 11, 10, 1000, true)))
		p            = newMockPss(sendFn)
		client, _, _ = newTestServer(t, testServerOptions{
			Pss:    p,
			Storer: mock.NewStorer(),
			Logger: logger,
			Post:   mp,
		})
		recipient = hex.EncodeToString(publicKeyBytes)
		targets   = fmt.Sprintf("[[%d]]", 0x12)
		topic     = "testtopic"
		hasher    = swarm.NewHasher()
		_, err    = hasher.Write([]byte(topic))
		topicHash = hasher.Sum(nil)
	)
	if err != nil {
		t.Fatal(err)
	}
	t.Run("err - bad targets", func(t *testing.T) {
		jsonhttptest.Request(t, client, http.MethodPost, "/pss/send/to/badtarget?recipient="+recipient, http.StatusBadRequest,
			jsonhttptest.WithRequestBody(bytes.NewReader(payload)),
			jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{
				Message: "target is not valid hex string",
				Code:    http.StatusBadRequest,
			}),
		)
		jsonhttptest.Request(t, client, http.MethodPost, "/pss/send/to/123456789abcdf?recipient="+recipient, http.StatusBadRequest,
			jsonhttptest.WithRequestBody(bytes.NewReader(payload)),
			jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{
				Message: "hex string target exceeds max length of 6",
				Code:    http.StatusBadRequest,
			}),
		)
	})
	t.Run("err - bad batch", func(t *testing.T) {
		hexbatch := hex.EncodeToString(batchInvalid)
		jsonhttptest.Request(t, client, http.MethodPost, "/pss/send/to/12", http.StatusBadRequest,
			jsonhttptest.WithRequestHeader(api.SwarmPostageBatchIdHeader, hexbatch),
			jsonhttptest.WithRequestBody(bytes.NewReader(payload)),
			jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{
				Message: "invalid postage batch id",
				Code:    http.StatusBadRequest,
			}),
		)
	})
	t.Run("ok batch", func(t *testing.T) {
		hexbatch := hex.EncodeToString(batchOk)
		jsonhttptest.Request(t, client, http.MethodPost, "/pss/send/to/12", http.StatusCreated,
			jsonhttptest.WithRequestHeader(api.SwarmPostageBatchIdHeader, hexbatch),
			jsonhttptest.WithRequestBody(bytes.NewReader(payload)),
		)
	})
	t.Run("bad request - batch empty", func(t *testing.T) {
		hexbatch := hex.EncodeToString(batchEmpty)
		jsonhttptest.Request(t, client, http.MethodPost, "/pss/send/to/12", http.StatusBadRequest,
			jsonhttptest.WithRequestHeader(api.SwarmPostageBatchIdHeader, hexbatch),
			jsonhttptest.WithRequestBody(bytes.NewReader(payload)),
		)
	})
	t.Run("ok", func(t *testing.T) {
		jsonhttptest.Request(t, client, http.MethodPost, "/pss/send/testtopic/12?recipient="+recipient, http.StatusCreated,
			jsonhttptest.WithRequestHeader(api.SwarmPostageBatchIdHeader, batchOkStr),
			jsonhttptest.WithRequestBody(bytes.NewReader(payload)),
			jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{
				Message: "Created",
				Code:    http.StatusCreated,
			}),
		)
		waitDone(t, &mtx, &done)
		if !bytes.Equal(receivedBytes, payload) {
			t.Fatalf("payload mismatch. want %v got %v", payload, receivedBytes)
		}
		if targets != fmt.Sprint(receivedTargets) {
			t.Fatalf("targets mismatch. want %v got %v", targets, receivedTargets)
		}
		if string(topicHash) != string(receivedTopic[:]) {
			t.Fatalf("topic mismatch. want %v got %v", topic, string(receivedTopic[:]))
		}
	})
	t.Run("without recipient", func(t *testing.T) {
		jsonhttptest.Request(t, client, http.MethodPost, "/pss/send/testtopic/12", http.StatusCreated,
			jsonhttptest.WithRequestHeader(api.SwarmPostageBatchIdHeader, batchOkStr),
			jsonhttptest.WithRequestBody(bytes.NewReader(payload)),
			jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{
				Message: "Created",
				Code:    http.StatusCreated,
			}),
		)
		waitDone(t, &mtx, &done)
		if !bytes.Equal(receivedBytes, payload) {
			t.Fatalf("payload mismatch. want %v got %v", payload, receivedBytes)
		}
		if targets != fmt.Sprint(receivedTargets) {
			t.Fatalf("targets mismatch. want %v got %v", targets, receivedTargets)
		}
		if string(topicHash) != string(receivedTopic[:]) {
			t.Fatalf("topic mismatch. want %v got %v", topic, string(receivedTopic[:]))
		}
	})
}
TestPssSend tests that the pss message sending over http works correctly.
package api_test
import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"encoding/hex"
	"fmt"
	"io"
	"math/big"
	"net/http"
	"net/url"
	"sync"
	"testing"
	"time"
	"github.com/btcsuite/btcd/btcec"
	"github.com/ethersphere/bee/pkg/api"
	"github.com/ethersphere/bee/pkg/crypto"
	"github.com/ethersphere/bee/pkg/jsonhttp"
	"github.com/ethersphere/bee/pkg/jsonhttp/jsonhttptest"
	"github.com/ethersphere/bee/pkg/logging"
	"github.com/ethersphere/bee/pkg/postage"
	mockpost "github.com/ethersphere/bee/pkg/postage/mock"
	"github.com/ethersphere/bee/pkg/pss"
	"github.com/ethersphere/bee/pkg/pushsync"
	"github.com/ethersphere/bee/pkg/storage/mock"
	"github.com/ethersphere/bee/pkg/swarm"
	"github.com/gorilla/websocket"
)
var (
	target  = pss.Target([]byte{1})
	targets = pss.Targets([]pss.Target{target})
	payload = []byte("testdata")
	topic   = pss.NewTopic("testtopic")
	mTimeout    = 10 * time.Second
	rTimeout    = 15 * time.Second
	longTimeout = 30 * time.Second
)
func TestPssWebsocketSingleHandler(t *testing.T) {
	var (
		p, publicKey, cl, _ = newPssTest(t, opts{})
		msgContent = make([]byte, len(payload))
		tc         swarm.Chunk
		mtx        sync.Mutex
		done       = make(chan struct{})
	)
	err := cl.SetReadDeadline(time.Now().Add(longTimeout))
	if err != nil {
		t.Fatal(err)
	}
	cl.SetReadLimit(swarm.ChunkSize)
	defer close(done)
	go waitReadMessage(t, &mtx, cl, msgContent, done)
	tc, err = pss.Wrap(context.Background(), topic, payload, publicKey, targets)
	if err != nil {
		t.Fatal(err)
	}
	p.TryUnwrap(tc)
	waitMessage(t, msgContent, payload, &mtx)
}
func TestPssWebsocketSingleHandlerDeregister(t *testing.T) {
	var (
		p, publicKey, cl, _ = newPssTest(t, opts{})
		msgContent = make([]byte, len(payload))
		tc         swarm.Chunk
		mtx        sync.Mutex
		done       = make(chan struct{})
	)
	err := cl.SetReadDeadline(time.Now().Add(longTimeout))
	if err != nil {
		t.Fatal(err)
	}
	cl.SetReadLimit(swarm.ChunkSize)
	defer close(done)
	go waitReadMessage(t, &mtx, cl, msgContent, done)
	tc, err = pss.Wrap(context.Background(), topic, payload, publicKey, targets)
	if err != nil {
		t.Fatal(err)
	}
	err = cl.WriteMessage(websocket.CloseMessage, []byte{})
	if err != nil {
		t.Fatal(err)
	}
	p.TryUnwrap(tc)
	waitMessage(t, msgContent, nil, &mtx)
}
func TestPssWebsocketMultiHandler(t *testing.T) {
	var (
		p, publicKey, cl, listener = newPssTest(t, opts{})
		u           = url.URL{Scheme: "ws", Host: listener, Path: "/pss/subscribe/testtopic"}
		cl2, _, err = websocket.DefaultDialer.Dial(u.String(), nil)
		msgContent  = make([]byte, len(payload))
		msgContent2 = make([]byte, len(payload))
		tc          swarm.Chunk
		mtx         sync.Mutex
		done        = make(chan struct{})
	)
	if err != nil {
		t.Fatalf("dial: %v. url %v", err, u.String())
	}
	err = cl.SetReadDeadline(time.Now().Add(longTimeout))
	if err != nil {
		t.Fatal(err)
	}
	cl.SetReadLimit(swarm.ChunkSize)
	defer close(done)
	go waitReadMessage(t, &mtx, cl, msgContent, done)
	go waitReadMessage(t, &mtx, cl2, msgContent2, done)
	tc, err = pss.Wrap(context.Background(), topic, payload, publicKey, targets)
	if err != nil {
		t.Fatal(err)
	}
	err = cl.WriteMessage(websocket.CloseMessage, []byte{})
	if err != nil {
		t.Fatal(err)
	}
	p.TryUnwrap(tc)
	waitMessage(t, msgContent, nil, &mtx)
	waitMessage(t, msgContent2, nil, &mtx)
}