Source File
	rpc_suite.go

Belonging Package
	/pkg/rpc/test

package test

import (
	"context"
	"fmt"
	"io"
	"net"
	"os"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

	sst "github.com/pancsta/asyncmachine-go/internal/testing/states"
	"github.com/pancsta/asyncmachine-go/internal/testing/utils"
	amhelp "github.com/pancsta/asyncmachine-go/pkg/helpers"
	amhelpt "github.com/pancsta/asyncmachine-go/pkg/helpers/testing"
	am "github.com/pancsta/asyncmachine-go/pkg/machine"
	arpc "github.com/pancsta/asyncmachine-go/pkg/rpc"
	ssrpc "github.com/pancsta/asyncmachine-go/pkg/rpc/states"
	"github.com/pancsta/asyncmachine-go/pkg/telemetry/dbg"
)

var (
	ssC  = ssrpc.ClientStates
	ssCo = ssrpc.ConsumerStates
	ssS  = ssrpc.ServerStates
	ssM  = ssrpc.MuxStates
	Pass = am.Pass
)

func TemplateTestBasic(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(true)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// init worker
	schema := ssrpc.StateSourceSchema.Merge(am.Schema{
		"Foo": {},
		"Bar": {Require: am.S{"Foo"}},
	})
	names := am.SAdd(ssrpc.StateSourceStates.Names(), am.S{"Foo", "Bar"})
	netSrc := am.New(ctx, schema, &am.Opts{Id: "ns-" + t.Name()})
	err := netSrc.VerifyStates(names)
	if err != nil {
		t.Fatal(err)
	}

	amhelpt.MachDebugEnv(t, netSrc)

	// init server and client
	_, _, s, c := newTest(t, ctx, netSrc, nil, 0, false, nil, nil)

	// test
	c.NetMach.Add1("Foo", nil)

	// assert
	assert.True(t, s.Mach.Is1(ssrpc.ServerStates.Ready), "Server ready")
	assert.True(t, c.Mach.Is1(ssrpc.ClientStates.Ready), "Client ready")

	assert.True(t, s.Mach.Not1(am.StateException), "No server errors")
	assert.True(t, c.Mach.Not1(am.StateException), "No client errors")

	assert.True(t, netSrc.Is1("Foo"), "NetworkMachine state set on the server")
	assert.True(t, c.NetMach.Is1("Foo"), "NetworkMachine state set on the client")

	c.Mach.Log("OK")
	s.Mach.Log("OK")

	// shut down
	c.Mach.Remove1(ssrpc.ClientStates.Start, nil)
	s.Mach.Remove1(ssrpc.ServerStates.Start, nil)
}

func TemplateTestTypeSafe(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}

	// read env
	amDbgAddr := os.Getenv(dbg.EnvAmDbgAddr)
	logLvl := am.EnvLogLevel("")

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	worker := utils.NewRelsNetSrc(t, nil)
	amhelpt.MachDebug(t, worker, amDbgAddr, logLvl, true)

	// init server and client
	_, _, s, c := newTest(t, ctx, worker, nil, 0, false, nil, nil)

	// test
	states := am.S{sst.A, sst.C}
	c.NetMach.Add(states, nil)

	// assert
	assert.True(t, s.Mach.Is1(ssrpc.ServerStates.Ready), "Server ready")
	assert.True(t, s.Mach.Is1(ssrpc.ClientStates.Ready), "Client ready")

	assert.True(t, s.Mach.Not1(am.StateException), "No server errors")
	assert.True(t, c.Mach.Not1(am.StateException), "No client errors")

	assert.True(t, worker.Is(states), "NetworkMachine state set on the server")
	assert.True(t, c.NetMach.Is(states),
		"NetworkMachine state set on the client")

	c.Mach.Log("OK")
	s.Mach.Log("OK")

	// shut down
	c.Mach.Remove1(ssrpc.ClientStates.Start, nil)
	s.Mach.Remove1(ssrpc.ServerStates.Start, nil)
}

func TemplateTestWaiting(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	end := make(chan struct{})
	counter, _, s, c := newTest(t, ctx, nil, end, 0, false, nil, nil)

	// test
	whenA := make(chan struct{})
	go func() {
		<-c.NetMach.When1(sst.A, ctx)
		close(whenA)
	}()
	states := am.S{sst.A, sst.C}
	c.NetMach.Add(states, nil)
	<-whenA

	// mark log and counter
	c.Mach.Log("OK")
	s.Mach.Log("OK")
	close(end)

	// assert
	assert.Equal(t, 0, int(s.CallCount),
		"Server piggybacked clock on resp")
	assert.Equal(t, 3, int(c.CallCount),
		"Client called RemoteHello, RemoteHandshake, RemoteAdd")
	bytesCount := <-counter
	assert.LessOrEqual(t, 1_300, int(bytesCount),
		"Bytes transferred (both ways)")
	assert.GreaterOrEqual(t, 20_000, int(bytesCount),
		"Bytes transferred (both ways)")

	disposeTest(t, c, s, true)
}

func TemplateTestAddMany(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	end := make(chan struct{})
	counter, _, s, c := newTest(t, ctx, nil, end, 0, false, nil, nil)

	// test
	whenD := make(chan struct{})
	go func() {
		<-c.NetMach.When1(sst.D, ctx)
		close(whenD)
	}()
	states := am.S{sst.A, sst.C}
	for i := 0; i < 500; i++ {
		c.NetMach.Add(states, nil)
	}
	c.NetMach.Add1(sst.D, nil)
	<-whenD

	// mark log and counter
	c.Mach.Log("OK")
	s.Mach.Log("OK")
	close(end)

	// assert
	assert.Equal(t, 0, int(s.CallCount),
		"Server piggybacked clock on resp")
	bytesCount := <-counter
	assert.LessOrEqual(t, 20_000, int(bytesCount),
		"Client called handshake (2) and A,C (500) and D(1)")
	assert.GreaterOrEqual(t, 200_000, int(bytesCount),
		"Client called handshake (2) and A,C (500) and D(1)")

	disposeTest(t, c, s, true)
}

func TemplateTestAddManyNoSync(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	end := make(chan struct{})
	// disable clock pushes
	counter, _, s, c := newTest(t, ctx, nil, end, 0, false, nil, nil)

	// test
	whenD := make(chan struct{})
	go func() {
		<-c.NetMach.When1(sst.D, ctx)
		close(whenD)
	}()
	states := am.S{sst.A, sst.C}
	for i := 0; i < 500; i++ {
		c.NetMach.AddNS(states, nil)
	}
	c.NetMach.Add1NS(sst.D, nil)

	// wait for the network to settle, as Sync arrives before +D, this can be
	// flaky, and it's better to Add1(sstest.D, nil), but Sync() is being tested
	// here
	time.Sleep(1000 * time.Millisecond)

	// manual sync, should tick D
	c.Sync()
	<-whenD

	// mark log and counter
	c.Mach.Log("OK")
	s.Mach.Log("OK")
	close(end)

	// assert
	bytesCount := <-counter
	assert.LessOrEqual(t, 10_000, int(bytesCount),
		"Client called handshake (2) and A,C (500) and D(1)")
	assert.GreaterOrEqual(t, 100_000, int(bytesCount),
		"Client called handshake (2) and A,C (500) and D(1)")

	disposeTest(t, c, s, true)
}

func TemplateTestAddManyInstantClock(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	end := make(chan struct{})
	// disable clock optimization (instant pushes)
	interval := 1 * time.Nanosecond
	counter, _, s, c := newTest(t, ctx, nil, end, interval, false, nil, nil)

	// test
	whenD := make(chan struct{})
	go func() {
		<-c.NetMach.When1(sst.D, ctx)
		close(whenD)
	}()
	states := am.S{sst.A, sst.C}
	for i := 0; i < 500; i++ {
		c.NetMach.Add(states, nil)
	}
	c.NetMach.Add1(sst.D, nil)
	<-whenD

	// mark log and counter
	c.Mach.Log("OK")
	s.Mach.Log("OK")
	close(end)

	// assert
	bytesCount := <-counter
	assert.LessOrEqual(t, 20_000, int(bytesCount),
		"Bytes transferred (both ways)")
	// 549_751
	assert.GreaterOrEqual(t, 200_000, int(bytesCount),
		"Bytes transferred (both ways)")

	disposeTest(t, c, s, true)
}

func TemplateTestManyStates(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}

	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	end := make(chan struct{})

	// reuse the net src and add many rand states
	schema := ssrpc.StateSourceSchema.Merge(sst.States)
	names := am.SAdd(ssrpc.StateSourceStates.Names(), sst.Names)
	randAmount := 100
	for i := 0; i < randAmount; i++ {
		n := fmt.Sprintf("State%d", i)
		names = append(names, n)
		schema[n] = am.State{}
	}
	netSrc, err := am.NewCommon(context.Background(), "ns-"+t.Name(), schema,
		names, nil, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	counter, _, s, c := newTest(t, ctx, netSrc, end, 0, false, nil, nil)

	// test
	whenD := make(chan struct{})
	go func() {
		<-c.NetMach.When1(sst.D, ctx)
		close(whenD)
	}()
	c.NetMach.Add1(sst.C, nil)
	for i := 5; i < randAmount-5; i++ {
		c.NetMach.Remove1(names[i-3], nil)
		c.NetMach.Add1(names[i], nil)
	}
	c.NetMach.Add1(sst.D, nil)
	<-whenD

	// mark log and counter
	c.Mach.Log("OK")
	s.Mach.Log("OK")
	close(end)

	// assert
	assert.Equal(t, 184, int(c.CallCount),
		"Client called handshake (2) and mutations (181)")
	bytesCount := <-counter
	assert.LessOrEqual(t, 10_000, int(bytesCount),
		"Bytes transferred (both ways)")
	assert.GreaterOrEqual(t, 100_000, int(bytesCount),
		"Bytes transferred (both ways)")

	disposeTest(t, c, s, true)
}

func TemplateTestHighInstantClocks(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	end := make(chan struct{})

	// reuse the worker and bump the clocks high
	worker := utils.NewRelsNetSrc(t, nil)
	clock := worker.Clock(nil)
	clock[sst.A] = 1_000_000
	clock[sst.C] = 1_000_000
	am.TestMockClock(worker, clock)
	// disable clock optimization
	counter, _, s, c := newTest(t, ctx, worker, end, 0, false, nil, nil)

	// test
	assert.GreaterOrEqual(t, int(worker.Tick(sst.A)), 1_000_000,
		"Tick count should be greater or equal to 1M")
	whenD := make(chan struct{})
	go func() {
		<-c.NetMach.When1(sst.D, ctx)
		close(whenD)
	}()
	states := am.S{sst.A, sst.C}
	for i := 0; i < 500; i++ {
		c.NetMach.Add(states, nil)
		c.NetMach.Remove(states, nil)
	}
	c.NetMach.Add1(sst.D, nil)
	<-whenD

	// mark log and counter
	c.Mach.Log("OK")
	s.Mach.Log("OK")
	close(end)

	// assert
	// byte count should be the same as in TestAddManyInstantClock
	bytesCount := <-counter
	assert.LessOrEqual(t, 48_000, int(bytesCount),
		"Bytes transferred (both ways)")
	assert.GreaterOrEqual(t, 350_000, int(bytesCount),
		"Bytes transferred (both ways)")

	disposeTest(t, c, s, true)
}

// TestRetryCall

func TemplateTestClockPush(t *testing.T, newTest NewTestFactory) {
	// TODO TestClockPush
	t.Skip("test server-side mutations push their clock")
}

type TestRetryCallHandlers struct {
	blocked bool
}

func (h *TestRetryCallHandlers) DState(e *am.Event) {
	if h.blocked {
		return
	}

	e.Machine().Log("Blocking for 1s")
	time.Sleep(1 * time.Second)
	h.blocked = true
}

func TemplateTestRetryCall(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_, w, s, c := newTest(t, ctx, nil, nil, 0, false, nil, nil)
	handlers := &TestRetryCallHandlers{}
	_, err := w.HandlersBind(handlers)
	require.NoError(t, err)

	// inject a fake error
	c.TmpTestErr = fmt.Errorf("IGNORE MOCK ERR")
	whenRetrying := c.Mach.When1(ssrpc.ClientStates.RetryingCall, nil)
	c.NetMach.Add1(sst.A, nil)
	amhelpt.WaitForAll(t, "RetryingCall", ctx, 2*time.Second, whenRetrying)

	// .TODO amtest
	assert.True(t, s.Mach.Is1(ssrpc.ServerStates.Ready), "Server ready")
	assert.True(t, s.Mach.Is1(ssrpc.ClientStates.Ready), "Client ready")

	c.Mach.Log("Generic err retried")

	// extend the timeout to cause a network one (handler blocks for 1s)
	// TODO avoid races with the queue in iroh, remove in v0.20
	<-w.WhenQueueEnds()
	time.Sleep(100*time.Millisecond)
	w.HandlerTimeout = 5 * time.Second
	c.CallTimeout = 500 * time.Millisecond
	wg := sync.WaitGroup{}
	wg.Add(2)

	go func() {
		// this will block and retry
		c.NetMach.Add1(sst.D, nil)
		wg.Done()
	}()
	go func() {
		<-c.Mach.When1(ssrpc.ClientStates.RetryingCall, ctx)
		c.Mach.Log("Timeout err retried")
		wg.Done()
	}()

	wg.Wait()

	// TODO amtest asserts
	assert.True(t, w.Is1(sst.D), "NetworkMachine state set")
	assert.True(t, s.Mach.Is1(ssrpc.ServerStates.Ready), "Server ready")
	assert.True(t, s.Mach.Is1(ssrpc.ClientStates.Ready), "Client ready")
	assert.True(t, handlers.blocked, "Handlers should block")

	disposeTest(t, c, s, false)
}

func TemplateTestRetryConn(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// EnableDebuggingRpc(true)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// test
	_, _, s, c := newTest(t, ctx, nil, nil, 0, true, nil, nil)
	lis := *s.Listener.Load()
	addr := lis.Addr()
	_ = lis.Close()
	s.Addr = addr.String()

	go func() {
		// wait for client to reconnect and then start the server
		<-c.Mach.WhenTime1(ssC.Connecting, 3, nil)
		s.Start(nil)
	}()

	// client ready
	c.Start(nil)
	amhelpt.WaitForAll(t, "client-server Ready", ctx, 5*time.Second,
		c.Mach.When1(ssC.Ready, ctx),
		s.Mach.When1(ssS.Ready, ctx))

	c.NetMach.Add1(sst.A, nil)

	// assert
	amhelpt.AssertIs1(t, c.Mach, ssrpc.ClientStates.Ready)
	amhelpt.AssertIs1(t, s.Mach, ssrpc.ServerStates.Ready)

	amhelpt.AssertNot1(t, c.Mach, ssrpc.ClientStates.RetryingCall)
	amhelpt.AssertNot1(t, c.Mach, ssrpc.ClientStates.RetryingConn)

	c.Mach.Log("Network err retried")

	disposeTest(t, c, s, false)
}

// TestRetryErrNetworkTimeout

type TestRetryErrNetworkTimeoutHandlers struct {
	blocked     bool
	shouldBlock bool
}

func (h *TestRetryErrNetworkTimeoutHandlers) DState(e *am.Event) {
	if !h.shouldBlock {
		return
	}
	e.Machine().Log("Blocking for 1s")
	time.Sleep(1 * time.Second)
	h.blocked = true
}

func TemplateTestRetryErrNetworkTimeout(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_, w, s, c := newTest(t, ctx, nil, nil, 0, false, nil, nil)
	handlers := &TestRetryErrNetworkTimeoutHandlers{
		shouldBlock: true,
	}
	_, err := w.HandlersBind(handlers)
	require.NoError(t, err)

	// test network timeout
	// extend the handler timeout (handler blocks for 1s)
	w.HandlerTimeout = 5 * time.Second
	c.CallTimeout = 500 * time.Millisecond
	wg := sync.WaitGroup{}
	wg.Add(2)

	go func() {
		<-c.Mach.When1(ssrpc.ClientStates.RetryingCall, ctx)
		c.Mach.Log("Timeout err retried")
		wg.Done()
	}()
	go func() {
		// this will block until connections restarts
		c.NetMach.Add1(sst.D, nil)
		wg.Done()
	}()

	wg.Wait()

	// assert
	amhelpt.AssertIs1(t, w, sst.D)

	amhelpt.AssertIs1(t, c.Mach, ssrpc.ClientStates.Ready)
	amhelpt.AssertIs1(t, s.Mach, ssrpc.ServerStates.Ready)

	amhelpt.AssertNot1(t, c.Mach, ssrpc.ClientStates.RetryingCall)
	amhelpt.AssertNot1(t, c.Mach, ssrpc.ClientStates.RetryingConn)

	assert.True(t, handlers.blocked, "Handlers should block")

	disposeTest(t, c, s, false)
}

func TemplateTestRetryClosedListener(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(true)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_, _, s, c := newTest(t, ctx, nil, nil, 0, false, nil, nil)

	// close the listener and try a mutation
	lis := *s.Listener.Load()
	_ = lis.Close()
	time.Sleep(100 * time.Millisecond)
	c.NetMach.Add1(sst.D, nil)

	// wait for D
	amhelpt.WaitForAll(t, t.Name(), ctx, 2*time.Second,
		c.NetMach.When1(sst.D, ctx))

	// assert
	amhelpt.AssertIs1(t, c.NetMach, sst.D)

	amhelpt.AssertIs1(t, c.Mach, ssrpc.ClientStates.Ready)
	amhelpt.AssertIs1(t, s.Mach, ssrpc.ServerStates.Ready)

	amhelpt.AssertNot1(t, c.Mach, ssrpc.ClientStates.RetryingCall)
	amhelpt.AssertNot1(t, c.Mach, ssrpc.ClientStates.RetryingConn)

	amhelpt.AssertTime(t, c.Mach,
		am.S{ssC.RetryingConn, ssC.HandshakeDone, ssC.Disconnected},
		am.Time{2, 3, 2})
	disposeTest(t, c, s, false)
}

// TestPayload

type TestPayloadHandlers struct {
	srv *arpc.Server
}

// CState will trigger SendPayload
func (w *TestPayloadHandlers) CState(e *am.Event) {
	// TODO use v2 state def
	e.Machine().EvRemove1(e, sst.C, nil)
	args := am.ParseArgs[arpc.A](e.Args)

	_ = w.srv.SendPayload(context.Background(), e, &arpc.MsgSrvPayload{
		Data: "Hello",
		Name: args.Name,
	})
}

type TestPayloadConsumer struct {
	t         *testing.T
	delivered bool
}

func (c *TestPayloadConsumer) ServerPayloadState(e *am.Event) {
	e.Machine().Remove1(ssCo.ServerPayload, nil)

	args := am.ParseArgs[arpc.AServerPayload](e.Args)
	assert.Equal(c.t, "TestPayload", args.Name)
	assert.Equal(c.t, "Hello", args.Payload.Data.(string))

	c.delivered = true
}

func TemplateTestPayload(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(true)
	// EnableDebuggingRpc(true)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	ssCo := ssrpc.ConsumerStates

	// consumer
	consHandlers := &TestPayloadConsumer{t: t}
	consMach, err := am.NewCommon(ctx, "TestPayloadConsumer",
		ssrpc.ConsumerSchema, ssCo.Names(), consHandlers, nil, nil)
	if err != nil {
		t.Fatal(err)
	}

	// source mach
	source := utils.NewNoRelsNetSrc(t, nil, "")
	handlers := &TestPayloadHandlers{}
	_, err = source.HandlersBind(handlers)
	if err != nil {
		t.Fatal(err)
	}

	// init RPC
	_, _, s, c := newTest(t, ctx, source, nil, 0, false, &arpc.ClientOpts{
		Consumer: consMach,
	}, nil)
	handlers.srv = s

	whenDelivered := consMach.When1(ssCo.ServerPayload, nil)
	// Consumer requests a payload from the remote worker
	c.NetMach.Add1(sst.C, Pass(&arpc.A{
		Name: "TestPayload",
	}))
	// Consumer waits for WorkerDelivered
	err = amhelp.WaitForAll(ctx, 2*time.Second, whenDelivered)

	// assert
	assert.NoError(t, err, "Timeout when waiting for the package")
	assert.True(t, consHandlers.delivered, "Consumer got the package")

	disposeTest(t, c, s, true)
}

// TODO test gob errors (although not user-facing)

func TemplateTestRetryingConnState(t *testing.T, newTest NewTestFactory) {
	t.Skip("TODO")
}

func TemplateTestPartial(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// net source mach with non-zero clocks
	source := utils.NewNoRelsNetSrc(t, nil, "")
	source.Add1(sst.C, nil)
	source.Remove1(sst.C, nil)

	// init RPC
	_, _, s, c := newTest(t, ctx, source, nil, 0, false, &arpc.ClientOpts{
		AllowedStates: am.S{sst.A, sst.B, sst.C},
		SkippedStates: am.S{sst.C},
	}, nil)

	// test
	source.Add1(sst.A, nil)
	source.Add(am.S{sst.B, sst.C}, nil)
	source.Add1(sst.D, nil)

	// assert
	amhelpt.WaitForAll(t, "TestPartial(A, B)", ctx, time.Second,
		c.NetMach.When(am.S{sst.A, sst.B}, nil))
	amhelpt.AssertIs(t, c.NetMach, am.S{sst.A, sst.B})

	// TODO schema change
	// TODO full sync

	// dispose
	disposeTest(t, c, s, true)
	if amhelp.IsTelemetry() {
		time.Sleep(1 * time.Second)
	}
}

func TemplateTestPartialInferred(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// net source mach
	source := utils.NewRelsNetSrc(t, nil)

	// init RPC
	_, _, s, c := newTest(t, ctx, source, nil, 0, false, &arpc.ClientOpts{
		AllowedStates: am.S{sst.A},
	}, nil)

	// test (C will add A, and netmach will infer it from the schema)
	source.Add1(sst.C, nil)

	// assert
	amhelpt.WaitForAll(t, "TestPartial(A, C)", ctx, time.Second,
		c.NetMach.When(am.S{sst.A, sst.C}, nil))
	amhelpt.AssertIs(t, c.NetMach, am.S{sst.A, sst.C})

	// TODO schema change
	// TODO full sync

	// dispose
	disposeTest(t, c, s, true)
	if amhelp.IsTelemetry() {
		time.Sleep(1 * time.Second)
	}
}

func TemplateTestPartialNoSchema(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// net source mach
	source := utils.NewRelsNetSrc(t, nil)

	// init RPC
	_, _, s, c := newTest(t, ctx, source, nil, 0, false, &arpc.ClientOpts{
		AllowedStates: am.S{sst.C},
		NoSchema:      true,
	}, nil)

	// test (C will add A, but netmach A wont be inferred)
	source.Add1(sst.C, nil)

	// assert
	expected := am.S{sst.C}
	amhelpt.WaitForAll(t, "TestPartial(A, C)", ctx, time.Second,
		c.NetMach.When(expected, nil))
	amhelpt.AssertIs(t, c.NetMach, expected)

	// TODO schema change
	// TODO full sync

	// dispose
	disposeTest(t, c, s, true)
	if amhelp.IsTelemetry() {
		time.Sleep(1 * time.Second)
	}
}

// TestSchemaFilteringSync

type TestSchemaFilteringSyncTracer struct {
	*am.TracerNoOp
	amount int
}

func (t *TestSchemaFilteringSyncTracer) MutationQueued(
	machine am.Api, mutation *am.Mutation,
) {
	t.amount++
}

func TemplateTestSchemaFilteringSync(t *testing.T, newTest NewTestFactory) {
	// TODO
	t.Skip("mutation filtering not implemented yet")
	return

	// amhelp.EnableDebugging(false)

	// // config
	// ctx, cancel := context.WithCancel(context.Background())
	// defer cancel()
	// end := make(chan struct{})
	// counter, netSrc, s, c := NewTest(t, ctx, nil, end, time.Second, false,
	// 	nil, nil)
	//
	// qCount := &TestSchemaFilteringSyncTracer{}
	// require.NoError(t, netSrc.BindTracer(qCount))
	//
	// // test
	// // A req C
	// c.NetMach.Add1(sst.A, nil)
	// c.NetMach.Add(am.S{sst.C, sst.A}, nil)
	//
	// // mark log and counter
	// c.Mach.Log("OK")
	// s.Mach.Log("OK")
	// close(end)
	//
	// // assert
	// assert.Len(t, c.NetMach.schema, 1+len(sst.States)+
	//   len(ssrpc.NetSourceSchema), "schema len")
	// assert.Equal(t, 1, qCount.amount,
	// 	"one queued mutation, one filtered out")
	// bytesCount := <-counter
	// assert.LessOrEqual(t, 1_000, int(bytesCount))
	// assert.GreaterOrEqual(t, 2_000, int(bytesCount))
	// amhelpt.AssertIs(t, netSrc, am.S{sst.C, sst.A})
	//
	// disposeTest(t, c, s, true)
}

func TemplateTestShallowSync(t *testing.T, newTest NewTestFactory) {
	t.Skip("TODO")
}

func TemplateTestNoSchema(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// net source mach
	source := utils.NewRelsNetSrc(t, nil)

	// init RPC
	_, _, s, c := newTest(t, ctx, source, nil, 0, false, &arpc.ClientOpts{
		NoSchema: true,
	}, nil)

	// test (C will add A, but netmach A wont be inferred)
	source.Add1(sst.C, nil)

	// assert
	expected := am.S{sst.A, sst.C}
	amhelpt.WaitForAll(t, "TestNoSchema(A, B, C, D)", ctx, time.Second,
		c.NetMach.When(expected, nil))
	amhelpt.AssertIs(t, c.NetMach, expected)

	// TODO schema change
	// TODO full sync

	// dispose
	disposeTest(t, c, s, true)
	if amhelp.IsTelemetry() {
		time.Sleep(1 * time.Second)
	}
}

// TestMutationsSync

type TestMutationsSyncTracer struct {
	*am.TracerNoOp
	amount int
}

func (t *TestMutationsSyncTracer) TransitionEnd(tx *am.Transition) {
	t.amount++

	// TODO assert mut types, called states
}

func TemplateTestMutationsSync(t *testing.T, newTest NewTestFactory) {
	if os.Getenv(am.EnvAmTestDbgAddr) == "" {
		t.Parallel()
	}
	// amhelp.EnableDebugging(false)

	// config
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	end := make(chan struct{})
	counter, netSrc, s, c := newTest(t, ctx, nil, end, time.Second, false,
		&arpc.ClientOpts{
			SyncMutations: true,
		}, nil)

	txCount := &TestMutationsSyncTracer{TracerNoOp: &am.TracerNoOp{Id: t.Name()}}
	_, err := c.NetMach.BindTracer(txCount)
	require.NoError(t, err)

	// test
	// add B and cause auto:A
	netSrc.Add1(sst.B, nil)
	netSrc.Add1(sst.D, nil)
	// let the source tracer finish
	time.Sleep(100 * time.Millisecond)
	// force push
	intNano := time.Nanosecond
	s.PushInterval.Store(&intNano)
	s.PushClient()
	amhelpt.WaitForAll(t, "mutations pushed", ctx, time.Second,
		c.NetMach.When1(sst.D, nil))

	// mark log and counter
	c.Mach.Log("OK")
	s.Mach.Log("OK")
	close(end)

	// assert
	assert.Equal(t, 3, txCount.amount,
		"3 mutations came from the server")
	bytesCount := <-counter
	assert.LessOrEqual(t, 1_000, int(bytesCount))
	assert.GreaterOrEqual(t, 2_000, int(bytesCount))

	disposeTest(t, c, s, true)
}

func TemplateTestExport(t *testing.T, newTest NewTestFactory) {
	t.Skip("TODO")

	// TODO assert mach tick
	// TODO schema change
}

// ///// ///// /////

// ///// UTILS

// ///// ///// /////

type NewTestFactory func(t *testing.T, ctx context.Context, netSrc *am.Machine,
	disposeMeter <-chan struct{}, pushInterval time.Duration, skipStart bool,
	clientOpts *arpc.ClientOpts, serverOpts *arpc.ServerOpts,
) (<-chan int64, *am.Machine, *arpc.Server, *arpc.Client)

func disposeTest(t *testing.T, c *arpc.Client, s *arpc.Server, checkErrs bool) {
	if checkErrs {
		amhelpt.AssertNoErrEver(t, c.Mach)
		amhelpt.AssertNoErrEver(t, s.Mach)
	}
	if os.Getenv(dbg.EnvAmDbgAddr) != "" {
		time.Sleep(time.Second)
	}
	c.Stop(context.TODO(), nil, true)
	<-c.Mach.WhenDisposed()
	s.Stop(nil, true)
}

// TCPMeter measures the traffic of a listener and forwards it to a
// destination. Results are sent to the [counter] channel. Useful for testing
// and benchmarking.
func TCPMeter(
	listener net.Listener, fwdTo string, counter chan<- int64,
	end <-chan struct{},
) {
	defer listener.Close()
	// fmt.Println("Listening on " + listenOn)

	// callFailsafe the destination
	destination, err := net.Dial("tcp4", fwdTo)
	if err != nil {
		fmt.Println("Error connecting to destination:", err.Error())
		return
	}
	defer destination.Close()

	// wait for the connection
	conn, err := listener.Accept()
	if err != nil {
		fmt.Println("Error accepting connection:", err.Error())
		return
	}
	defer conn.Close()

	// forward data bidirectionally
	wg := sync.WaitGroup{}
	wg.Add(2)
	bytes := atomic.Int64{}
	go func() {
		c, _ := io.Copy(destination, conn)
		bytes.Add(c)
		wg.Done()
	}()
	go func() {
		c, _ := io.Copy(conn, destination)
		bytes.Add(c)
		wg.Done()
	}()

	// wait for the test and forwarding to finish
	<-end
	// fmt.Printf("Closing counter...\n")
	_ = listener.Close()
	_ = destination.Close()
	_ = conn.Close()
	wg.Wait()

	c := bytes.Load()
	// fmt.Printf("Forwarded %d bytes\n", c)
	counter <- c
}

The pages are generated with Golds v0.8.4. (GOOS=linux GOARCH=amd64)
Golds is a Go 101 project developed by Tapir Liu.
PR and bug reports are welcome and can be submitted to the issue list.
Please follow @zigo_101 (reachable from the left QR code) to get the latest news of Golds.