depot/packages/networking/ipfs-cluster/api/rest/client/methods_test.go

package client

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	types "github.com/ipfs-cluster/ipfs-cluster/api"
	rest "github.com/ipfs-cluster/ipfs-cluster/api/rest"
	test "github.com/ipfs-cluster/ipfs-cluster/test"

	rpc "github.com/libp2p/go-libp2p-gorpc"
	peer "github.com/libp2p/go-libp2p/core/peer"
	ma "github.com/multiformats/go-multiaddr"
)

func testClients(t *testing.T, api *rest.API, f func(*testing.T, Client)) {
	t.Run("in-parallel", func(t *testing.T) {
		t.Run("libp2p", func(t *testing.T) {
			t.Parallel()
			f(t, testClientLibp2p(t, api))
		})
		t.Run("http", func(t *testing.T) {
			t.Parallel()
			f(t, testClientHTTP(t, api))
		})
	})
}

func TestVersion(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		v, err := c.Version(ctx)
		if err != nil || v.Version == "" {
			t.Logf("%+v", v)
			t.Log(err)
			t.Error("expected something in version")
		}
	}

	testClients(t, api, testF)
}

func TestID(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		id, err := c.ID(ctx)
		if err != nil {
			t.Fatal(err)
		}
		if id.ID == "" {
			t.Error("bad id")
		}
	}

	testClients(t, api, testF)
}

func TestPeers(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		out := make(chan types.ID, 10)
		err := c.Peers(ctx, out)
		if err != nil {
			t.Fatal(err)
		}
		if len(out) == 0 {
			t.Error("expected some peers")
		}
	}

	testClients(t, api, testF)
}

func TestPeersWithError(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		addr, _ := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/44444")
		var _ = c
		c, _ = NewDefaultClient(&Config{APIAddr: addr, DisableKeepAlives: true})
		out := make(chan types.ID, 10)
		err := c.Peers(ctx, out)
		if err == nil {
			t.Fatal("expected error")
		}
		if len(out) > 0 {
			t.Fatal("expected no ids")
		}
	}

	testClients(t, api, testF)
}

func TestPeerAdd(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		id, err := c.PeerAdd(ctx, test.PeerID1)
		if err != nil {
			t.Fatal(err)
		}
		if id.ID != test.PeerID1 {
			t.Error("bad peer")
		}
	}

	testClients(t, api, testF)
}

func TestPeerRm(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		err := c.PeerRm(ctx, test.PeerID1)
		if err != nil {
			t.Fatal(err)
		}
	}

	testClients(t, api, testF)
}

func TestPin(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		opts := types.PinOptions{
			ReplicationFactorMin: 6,
			ReplicationFactorMax: 7,
			Name:                 "hello there",
		}
		_, err := c.Pin(ctx, test.Cid1, opts)
		if err != nil {
			t.Fatal(err)
		}
	}

	testClients(t, api, testF)
}

func TestUnpin(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		_, err := c.Unpin(ctx, test.Cid1)
		if err != nil {
			t.Fatal(err)
		}
	}

	testClients(t, api, testF)
}

type pathCase struct {
	path        string
	wantErr     bool
	expectedCid string
}

var pathTestCases = []pathCase{
	{
		test.CidResolved.String(),
		false,
		test.CidResolved.String(),
	},
	{
		test.PathIPFS1,
		false,
		"QmaNJ5acV31sx8jq626qTpAWW4DXKw34aGhx53dECLvXbY",
	},
	{
		test.PathIPFS2,
		false,
		test.CidResolved.String(),
	},
	{
		test.PathIPNS1,
		false,
		test.CidResolved.String(),
	},
	{
		test.PathIPLD1,
		false,
		"QmaNJ5acV31sx8jq626qTpAWW4DXKw34aGhx53dECLvXbY",
	},
	{
		test.InvalidPath1,
		true,
		"",
	},
}

func TestPinPath(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	opts := types.PinOptions{
		ReplicationFactorMin: 6,
		ReplicationFactorMax: 7,
		Name:                 "hello there",
		UserAllocations:      []peer.ID{test.PeerID1, test.PeerID2},
	}

	testF := func(t *testing.T, c Client) {
		for _, testCase := range pathTestCases {
			ec, _ := types.DecodeCid(testCase.expectedCid)
			resultantPin := types.PinWithOpts(ec, opts)
			p := testCase.path
			pin, err := c.PinPath(ctx, p, opts)
			if err != nil {
				if testCase.wantErr {
					continue
				}
				t.Fatalf("unexpected error %s: %s", p, err)
			}

			if !pin.Equals(resultantPin) {
				t.Errorf("expected different pin: %s", p)
				t.Errorf("expected: %+v", resultantPin)
				t.Errorf("actual: %+v", pin)
			}

		}
	}

	testClients(t, api, testF)
}

func TestUnpinPath(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		for _, testCase := range pathTestCases {
			p := testCase.path
			pin, err := c.UnpinPath(ctx, p)
			if err != nil {
				if testCase.wantErr {
					continue
				}
				t.Fatalf("unepected error %s: %s", p, err)
			}

			if pin.Cid.String() != testCase.expectedCid {
				t.Errorf("bad resolved Cid: %s, %s", p, pin.Cid)
			}
		}
	}

	testClients(t, api, testF)
}

func TestAllocations(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		pins := make(chan types.Pin)
		n := 0
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			for range pins {
				n++
			}
		}()

		err := c.Allocations(ctx, types.DataType|types.MetaType, pins)
		if err != nil {
			t.Fatal(err)
		}

		wg.Wait()
		if n == 0 {
			t.Error("should be some pins")
		}
	}

	testClients(t, api, testF)
}

func TestAllocation(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		pin, err := c.Allocation(ctx, test.Cid1)
		if err != nil {
			t.Fatal(err)
		}
		if !pin.Cid.Equals(test.Cid1) {
			t.Error("should be same pin")
		}
	}

	testClients(t, api, testF)
}

func TestStatus(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		pin, err := c.Status(ctx, test.Cid1, false)
		if err != nil {
			t.Fatal(err)
		}
		if !pin.Cid.Equals(test.Cid1) {
			t.Error("should be same pin")
		}
	}

	testClients(t, api, testF)
}

func TestStatusCids(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		out := make(chan types.GlobalPinInfo)

		go func() {
			err := c.StatusCids(ctx, []types.Cid{test.Cid1}, false, out)
			if err != nil {
				t.Error(err)
			}
		}()

		pins := collectGlobalPinInfos(t, out)
		if len(pins) != 1 {
			t.Fatal("wrong number of pins returned")
		}
		if !pins[0].Cid.Equals(test.Cid1) {
			t.Error("should be same pin")
		}
	}

	testClients(t, api, testF)
}

func collectGlobalPinInfos(t *testing.T, out <-chan types.GlobalPinInfo) []types.GlobalPinInfo {
	t.Helper()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	var gpis []types.GlobalPinInfo
	for {
		select {
		case <-ctx.Done():
			t.Error(ctx.Err())
			return gpis
		case gpi, ok := <-out:
			if !ok {
				return gpis
			}
			gpis = append(gpis, gpi)
		}
	}
}

func TestStatusAll(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		out := make(chan types.GlobalPinInfo)
		go func() {
			err := c.StatusAll(ctx, 0, false, out)
			if err != nil {
				t.Error(err)
			}
		}()
		pins := collectGlobalPinInfos(t, out)

		if len(pins) == 0 {
			t.Error("there should be some pins")
		}

		out2 := make(chan types.GlobalPinInfo)
		go func() {
			err := c.StatusAll(ctx, 0, true, out2)
			if err != nil {
				t.Error(err)
			}
		}()
		pins = collectGlobalPinInfos(t, out2)

		if len(pins) != 2 {
			t.Error("there should be two pins")
		}

		out3 := make(chan types.GlobalPinInfo)
		go func() {
			err := c.StatusAll(ctx, types.TrackerStatusPinning, false, out3)
			if err != nil {
				t.Error(err)
			}
		}()
		pins = collectGlobalPinInfos(t, out3)

		if len(pins) != 1 {
			t.Error("there should be one pin")
		}

		out4 := make(chan types.GlobalPinInfo)
		go func() {
			err := c.StatusAll(ctx, types.TrackerStatusPinned|types.TrackerStatusError, false, out4)
			if err != nil {
				t.Error(err)
			}
		}()
		pins = collectGlobalPinInfos(t, out4)

		if len(pins) != 2 {
			t.Error("there should be two pins")
		}

		out5 := make(chan types.GlobalPinInfo, 1)
		err := c.StatusAll(ctx, 1<<25, false, out5)
		if err == nil {
			t.Error("expected an error")
		}
	}

	testClients(t, api, testF)
}

func TestRecover(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		pin, err := c.Recover(ctx, test.Cid1, false)
		if err != nil {
			t.Fatal(err)
		}
		if !pin.Cid.Equals(test.Cid1) {
			t.Error("should be same pin")
		}
	}

	testClients(t, api, testF)
}

func TestRecoverAll(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		out := make(chan types.GlobalPinInfo, 10)
		err := c.RecoverAll(ctx, true, out)
		if err != nil {
			t.Fatal(err)
		}

		out2 := make(chan types.GlobalPinInfo, 10)
		err = c.RecoverAll(ctx, false, out2)
		if err != nil {
			t.Fatal(err)
		}
	}

	testClients(t, api, testF)
}

func TestAlerts(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		alerts, err := c.Alerts(ctx)
		if err != nil {
			t.Fatal(err)
		}
		if len(alerts) != 1 {
			t.Fatal("expected 1 alert")
		}
		pID2 := test.PeerID2.String()
		if alerts[0].Peer != test.PeerID2 {
			t.Errorf("expected an alert from %s", pID2)
		}
	}

	testClients(t, api, testF)
}

func TestGetConnectGraph(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		cg, err := c.GetConnectGraph(ctx)
		if err != nil {
			t.Fatal(err)
		}
		if len(cg.IPFSLinks) != 3 || len(cg.ClusterLinks) != 3 ||
			len(cg.ClustertoIPFS) != 3 {
			t.Fatal("Bad graph")
		}
	}

	testClients(t, api, testF)
}

func TestMetrics(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		m, err := c.Metrics(ctx, "somemetricstype")
		if err != nil {
			t.Fatal(err)
		}

		if len(m) == 0 {
			t.Fatal("No metrics found")
		}
	}

	testClients(t, api, testF)
}

func TestMetricNames(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		m, err := c.MetricNames(ctx)
		if err != nil {
			t.Fatal(err)
		}

		if len(m) == 0 {
			t.Fatal("No metric names found")
		}
	}

	testClients(t, api, testF)
}

type waitService struct {
	l        sync.Mutex
	pinStart time.Time
}

func (wait *waitService) Pin(ctx context.Context, in types.Pin, out *types.Pin) error {
	wait.l.Lock()
	defer wait.l.Unlock()
	wait.pinStart = time.Now()
	*out = in
	return nil
}

func (wait *waitService) Status(ctx context.Context, in types.Cid, out *types.GlobalPinInfo) error {
	wait.l.Lock()
	defer wait.l.Unlock()
	if time.Now().After(wait.pinStart.Add(5 * time.Second)) { //pinned
		*out = types.GlobalPinInfo{
			Cid: in,
			PeerMap: map[string]types.PinInfoShort{
				test.PeerID1.String(): {
					Status: types.TrackerStatusPinned,
					TS:     wait.pinStart,
				},
				test.PeerID2.String(): {
					Status: types.TrackerStatusPinned,
					TS:     wait.pinStart,
				},
				test.PeerID3.String(): {
					Status: types.TrackerStatusPinning,
					TS:     wait.pinStart,
				},
				test.PeerID3.String(): {
					Status: types.TrackerStatusRemote,
					TS:     wait.pinStart,
				},
			},
		}
	} else { // pinning
		*out = types.GlobalPinInfo{
			Cid: in,
			PeerMap: map[string]types.PinInfoShort{
				test.PeerID1.String(): {
					Status: types.TrackerStatusPinning,
					TS:     wait.pinStart,
				},
				test.PeerID2.String(): {
					Status: types.TrackerStatusPinned,
					TS:     wait.pinStart,
				},
				test.PeerID3.String(): {
					Status: types.TrackerStatusPinning,
					TS:     wait.pinStart,
				},
				test.PeerID3.String(): {
					Status: types.TrackerStatusRemote,
					TS:     wait.pinStart,
				},
			},
		}
	}

	return nil
}

func (wait *waitService) PinGet(ctx context.Context, in types.Cid, out *types.Pin) error {
	p := types.PinCid(in)
	p.ReplicationFactorMin = 2
	p.ReplicationFactorMax = 3
	*out = p
	return nil
}

type waitServiceUnpin struct {
	l          sync.Mutex
	unpinStart time.Time
}

func (wait *waitServiceUnpin) Unpin(ctx context.Context, in types.Pin, out *types.Pin) error {
	wait.l.Lock()
	defer wait.l.Unlock()
	wait.unpinStart = time.Now()
	return nil
}

func (wait *waitServiceUnpin) Status(ctx context.Context, in types.Cid, out *types.GlobalPinInfo) error {
	wait.l.Lock()
	defer wait.l.Unlock()
	if time.Now().After(wait.unpinStart.Add(5 * time.Second)) { //unpinned
		*out = types.GlobalPinInfo{
			Cid: in,
			PeerMap: map[string]types.PinInfoShort{
				test.PeerID1.String(): {
					Status: types.TrackerStatusUnpinned,
					TS:     wait.unpinStart,
				},
				test.PeerID2.String(): {
					Status: types.TrackerStatusUnpinned,
					TS:     wait.unpinStart,
				},
			},
		}
	} else { // pinning
		*out = types.GlobalPinInfo{
			Cid: in,
			PeerMap: map[string]types.PinInfoShort{
				test.PeerID1.String(): {
					Status: types.TrackerStatusUnpinning,
					TS:     wait.unpinStart,
				},
				test.PeerID2.String(): {
					Status: types.TrackerStatusUnpinning,
					TS:     wait.unpinStart,
				},
			},
		}
	}

	return nil
}

func (wait *waitServiceUnpin) PinGet(ctx context.Context, in types.Cid, out *types.Pin) error {
	return errors.New("not found")
}

func TestWaitForPin(t *testing.T) {
	ctx := context.Background()
	tapi := testAPI(t)
	defer shutdown(tapi)

	rpcS := rpc.NewServer(nil, "wait")
	rpcC := rpc.NewClientWithServer(nil, "wait", rpcS)
	err := rpcS.RegisterName("Cluster", &waitService{})
	if err != nil {
		t.Fatal(err)
	}

	tapi.SetClient(rpcC)

	testF := func(t *testing.T, c Client) {

		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()

			fp := StatusFilterParams{
				Cid:       test.Cid1,
				Local:     false,
				Target:    types.TrackerStatusPinned,
				CheckFreq: time.Second,
			}
			start := time.Now()

			st, err := WaitFor(ctx, c, fp)
			if err != nil {
				t.Error(err)
				return
			}
			if time.Since(start) <= 5*time.Second {
				t.Error("slow pin should have taken at least 5 seconds")
				return
			}

			totalPinned := 0
			for _, pi := range st.PeerMap {
				if pi.Status == types.TrackerStatusPinned {
					totalPinned++
				}
			}
			if totalPinned < 2 { // repl factor min
				t.Error("pin info should show the item is pinnedin two places at least")
			}
		}()
		_, err := c.Pin(ctx, test.Cid1, types.PinOptions{ReplicationFactorMin: 0, ReplicationFactorMax: 0, Name: "test", ShardSize: 0})
		if err != nil {
			t.Fatal(err)
		}
		wg.Wait()
	}

	testClients(t, tapi, testF)
}

func TestWaitForUnpin(t *testing.T) {
	ctx := context.Background()
	tapi := testAPI(t)
	defer shutdown(tapi)

	rpcS := rpc.NewServer(nil, "wait")
	rpcC := rpc.NewClientWithServer(nil, "wait", rpcS)
	err := rpcS.RegisterName("Cluster", &waitServiceUnpin{})
	if err != nil {
		t.Fatal(err)
	}

	tapi.SetClient(rpcC)

	testF := func(t *testing.T, c Client) {

		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()

			fp := StatusFilterParams{
				Cid:       test.Cid1,
				Local:     false,
				Target:    types.TrackerStatusUnpinned,
				CheckFreq: time.Second,
			}
			start := time.Now()

			st, err := WaitFor(ctx, c, fp)
			if err != nil {
				t.Error(err)
				return
			}
			if time.Since(start) <= 5*time.Second {
				t.Error("slow unpin should have taken at least 5 seconds")
				return
			}

			for _, pi := range st.PeerMap {
				if pi.Status != types.TrackerStatusUnpinned {
					t.Error("the item should have been unpinned everywhere")
				}
			}
		}()
		_, err := c.Unpin(ctx, test.Cid1)
		if err != nil {
			t.Fatal(err)
		}
		wg.Wait()
	}

	testClients(t, tapi, testF)
}

func TestAddMultiFile(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer api.Shutdown(ctx)

	sth := test.NewShardingTestHelper()
	defer sth.Clean(t)

	testF := func(t *testing.T, c Client) {
		mfr, closer := sth.GetTreeMultiReader(t)
		defer closer.Close()

		p := types.AddParams{
			PinOptions: types.PinOptions{
				ReplicationFactorMin: -1,
				ReplicationFactorMax: -1,
				Name:                 "test something",
				ShardSize:            1024,
			},
			Shard:  false,
			Format: "",
			IPFSAddParams: types.IPFSAddParams{
				Chunker:   "",
				RawLeaves: false,
			},
			Hidden:         false,
			StreamChannels: true,
		}

		out := make(chan types.AddedOutput, 1)
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			for v := range out {
				t.Logf("output: Name: %s. Hash: %s", v.Name, v.Cid)
			}
		}()

		err := c.AddMultiFile(ctx, mfr, p, out)
		if err != nil {
			t.Fatal(err)
		}

		wg.Wait()
	}

	testClients(t, api, testF)
}

func TestRepoGC(t *testing.T) {
	ctx := context.Background()
	api := testAPI(t)
	defer shutdown(api)

	testF := func(t *testing.T, c Client) {
		globalGC, err := c.RepoGC(ctx, false)
		if err != nil {
			t.Fatal(err)
		}

		if globalGC.PeerMap == nil {
			t.Fatal("expected a non-nil peer map")
		}

		for _, gc := range globalGC.PeerMap {
			if gc.Peer == "" {
				t.Error("bad id")
			}
			if gc.Error != "" {
				t.Error("did not expect any error")
			}
			if gc.Keys == nil {
				t.Error("expected a non-nil array of IPFSRepoGC")
			} else {
				if !gc.Keys[0].Key.Equals(test.Cid1) {
					t.Errorf("expected a different cid, expected: %s, found: %s", test.Cid1, gc.Keys[0].Key)
				}
			}
		}
	}

	testClients(t, api, testF)
}