Fix unit tests in ai-video

.github/workflows/build.yaml

@@ -4,7 +4,6 @@ on:
   pull_request:
   push:
     branches:
-      # - master
       - ai-video
     tags:
       - "v*"
@@ -338,7 +337,7 @@ jobs:
           destination: "build.livepeer.live/${{ github.event.repository.name }}/ai-video/stable"
           parent: false
           process_gcloudignore: false
-      
+
       # Update the latest branch manifest
       - name: Upload branch manifest file to Google Cloud stable folder
         id: upload-manifest-latest

.github/workflows/test.yaml

@@ -1,9 +1,10 @@
 name: Trigger test suite
 
 on:
-  # pull_request:
-  #   branches:
-  #     - master
+  pull_request:
+    branches:
+      - master
+      - ai-video
   push:
     branches:
       - master
@@ -91,13 +92,6 @@ jobs:
           go fmt ./...
           git diff --exit-code
 
-      - name: Lint
-        uses: golangci/golangci-lint-action@v4
-        with:
-          version: v1.52.2
-          skip-pkg-cache: true
-          args: '--disable-all --enable=gofmt --enable=vet --enable=golint --deadline=4m pm verification'
-
       - name: Run Revive Action by building from repository
         uses: docker://morphy/revive-action:v2
         with:

common/testutil.go

@@ -89,7 +89,7 @@ func IgnoreRoutines() []goleak.Option {
 		"github.com/livepeer/go-livepeer/server.(*LivepeerServer).StartMediaServer", "github.com/livepeer/go-livepeer/core.(*RemoteTranscoderManager).Manage.func1",
 		"github.com/livepeer/go-livepeer/server.(*LivepeerServer).HandlePush.func1", "github.com/rjeczalik/notify.(*nonrecursiveTree).dispatch",
 		"github.com/rjeczalik/notify.(*nonrecursiveTree).internal", "github.com/livepeer/lpms/stream.NewBasicRTMPVideoStream.func1", "github.com/patrickmn/go-cache.(*janitor).Run",
-		"github.com/golang/glog.(*fileSink).flushDaemon",
+		"github.com/golang/glog.(*fileSink).flushDaemon", "github.com/ipfs/go-log/writer.(*MirrorWriter).logRoutine",
 	}
 
 	res := make([]goleak.Option, 0, len(funcs2ignore))

discovery/discovery_test.go

@@ -56,7 +56,7 @@ func TestDeadLock(t *testing.T) {
 	first := true
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { wg.Wait(); serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		defer wg.Done()
 		if first {
@@ -88,7 +88,7 @@ func TestDeadLock_NewOrchestratorPoolWithPred(t *testing.T) {
 	first := true
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { wg.Wait(); serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		defer wg.Done()
 		if first {
@@ -187,7 +187,7 @@ func TestNewDBOrchestorPoolCache_NoEthAddress(t *testing.T) {
 	oldServerGetOrchInfo := serverGetOrchInfo
 	defer func() { serverGetOrchInfo = oldServerGetOrchInfo }()
 	var mu sync.Mutex
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		defer mu.Unlock()
 
@@ -244,7 +244,7 @@ func TestNewDBOrchestratorPoolCache_InvalidPrices(t *testing.T) {
 	oldServerGetOrchInfo := serverGetOrchInfo
 	defer func() { serverGetOrchInfo = oldServerGetOrchInfo }()
 	var mu sync.Mutex
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		defer mu.Unlock()
 
@@ -294,7 +294,7 @@ func TestNewDBOrchestratorPoolCache_GivenListOfOrchs_CreatesPoolCacheCorrectly(t
 	expPricePerPixel, _ := common.PriceToFixed(big.NewRat(999, 1))
 	var mu sync.Mutex
 	first := true
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		if first {
 			time.Sleep(100 * time.Millisecond)
@@ -386,7 +386,7 @@ func TestNewDBOrchestratorPoolCache_TestURLs(t *testing.T) {
 
 	var mu sync.Mutex
 	first := true
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		if first {
 			time.Sleep(100 * time.Millisecond)
@@ -479,7 +479,7 @@ func TestNewDBOrchestorPoolCache_PollOrchestratorInfo(t *testing.T) {
 	wg := sync.WaitGroup{}
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { wg.Wait(); serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		defer mu.Unlock()
 		// slightly unsafe to be adding to the wg counter here
@@ -634,7 +634,7 @@ func TestCachedPool_AllOrchestratorsTooExpensive_ReturnsAllOrchestrators(t *test
 	defer runtime.GOMAXPROCS(gmp)
 	var mu sync.Mutex
 	first := true
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		if first {
 			time.Sleep(100 * time.Millisecond)
@@ -723,7 +723,7 @@ func TestCachedPool_GetOrchestrators_MaxBroadcastPriceNotSet(t *testing.T) {
 	defer runtime.GOMAXPROCS(gmp)
 	var mu sync.Mutex
 	first := true
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		if first {
 			time.Sleep(100 * time.Millisecond)
@@ -829,7 +829,7 @@ func TestCachedPool_N_OrchestratorsGoodPricing_ReturnsNOrchestrators(t *testing.
 	defer runtime.GOMAXPROCS(gmp)
 	var mu sync.Mutex
 	first := true
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		if first {
 			time.Sleep(100 * time.Millisecond)
@@ -932,7 +932,7 @@ func TestCachedPool_GetOrchestrators_TicketParamsValidation(t *testing.T) {
 
 	server.BroadcastCfg.SetMaxPrice(nil)
 
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		return &net.OrchestratorInfo{
 			Address:      pm.RandBytes(20),
 			Transcoder:   "transcoder",
@@ -1006,7 +1006,7 @@ func TestCachedPool_GetOrchestrators_OnlyActiveOrchestrators(t *testing.T) {
 	defer runtime.GOMAXPROCS(gmp)
 	var mu sync.Mutex
 	first := true
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, orchestratorServer *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		if first {
 			time.Sleep(100 * time.Millisecond)
@@ -1113,7 +1113,7 @@ func TestNewWHOrchestratorPoolCache(t *testing.T) {
 	wg := sync.WaitGroup{}
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { wg.Wait(); serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(c context.Context, b common.Broadcaster, s *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(c context.Context, b common.Broadcaster, s *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		defer wg.Done()
 		return &net.OrchestratorInfo{Transcoder: "transcoder"}, nil
 	}
@@ -1276,7 +1276,7 @@ func TestOrchestratorPool_GetOrchestrators(t *testing.T) {
 	orchCb := func() error { return nil }
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { wg.Wait(); serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		defer wg.Done()
 		err := orchCb()
 		return &net.OrchestratorInfo{
@@ -1341,7 +1341,7 @@ func TestOrchestratorPool_GetOrchestrators_SuspendedOrchs(t *testing.T) {
 	orchCb := func() error { return nil }
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { wg.Wait(); serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		defer wg.Done()
 		err := orchCb()
 		return &net.OrchestratorInfo{
@@ -1413,7 +1413,7 @@ func TestOrchestratorPool_ShuffleGetOrchestrators(t *testing.T) {
 
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		ch <- server
 		return &net.OrchestratorInfo{Transcoder: server.String()}, nil
 	}
@@ -1476,7 +1476,7 @@ func TestOrchestratorPool_GetOrchestratorTimeout(t *testing.T) {
 	ch := make(chan struct{})
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		ch <- struct{}{} // this will block if necessary to simulate a timeout
 		return &net.OrchestratorInfo{}, nil
 	}
@@ -1591,7 +1591,7 @@ func TestOrchestratorPool_Capabilities(t *testing.T) {
 	calls := 0
 	oldOrchInfo := serverGetOrchInfo
 	defer func() { serverGetOrchInfo = oldOrchInfo }()
-	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL) (*net.OrchestratorInfo, error) {
+	serverGetOrchInfo = func(ctx context.Context, bcast common.Broadcaster, server *url.URL, caps *net.Capabilities) (*net.OrchestratorInfo, error) {
 		mu.Lock()
 		defer func() {
 			calls = (calls + 1) % len(responses)

discovery/stub.go

@@ -103,3 +103,7 @@ func (s *stubCapabilities) CompatibleWith(caps *net.Capabilities) bool {
 func (s *stubCapabilities) LegacyOnly() bool {
 	return s.isLegacy
 }
+
+func (s *stubCapabilities) ToNetCapabilities() *net.Capabilities {
+	return &net.Capabilities{Bitstring: capCompatString}
+}

go.mod

@@ -27,7 +27,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.14.0
 	github.com/stretchr/testify v1.9.0
-	github.com/testcontainers/testcontainers-go v0.9.0
+	github.com/testcontainers/testcontainers-go v0.26.0
 	github.com/urfave/cli v1.22.12
 	go.opencensus.io v0.24.0
 	go.uber.org/goleak v1.3.0
@@ -42,16 +42,19 @@ require (
 	cloud.google.com/go/compute/metadata v0.3.0 // indirect
 	cloud.google.com/go/iam v1.1.0 // indirect
 	cloud.google.com/go/storage v1.30.1 // indirect
+	dario.cat/mergo v1.0.0 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/DataDog/zstd v1.4.5 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
+	github.com/Microsoft/hcsshim v0.11.1 // indirect
 	github.com/StackExchange/wmi v1.2.1 // indirect
 	github.com/VictoriaMetrics/fastcache v1.12.1 // indirect
 	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
 	github.com/aws/aws-sdk-go v1.44.273 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.7.0 // indirect
 	github.com/btcsuite/btcd/btcec/v2 v2.2.0 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/cespare/cp v1.1.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cockroachdb/errors v1.8.1 // indirect
@@ -62,7 +65,9 @@ require (
 	github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect
 	github.com/consensys/bavard v0.1.13 // indirect
 	github.com/consensys/gnark-crypto v0.12.1 // indirect
-	github.com/containerd/containerd v1.7.0-beta.2 // indirect
+	github.com/containerd/containerd v1.7.7 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/crate-crypto/go-kzg-4844 v0.7.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -152,6 +157,8 @@ require (
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/livepeer/joy4 v0.1.2-0.20191121080656-b2fea45cbded // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
@@ -166,9 +173,11 @@ require (
 	github.com/mmcloughlin/addchain v0.4.0 // indirect
 	github.com/moby/patternmatcher v0.6.0 // indirect
 	github.com/moby/sys/sequential v0.5.0 // indirect
+	github.com/moby/term v0.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect
+	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
 	github.com/multiformats/go-base32 v0.1.0 // indirect
 	github.com/multiformats/go-base36 v0.2.0 // indirect
@@ -184,6 +193,7 @@ require (
 	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/polydawn/refmt v0.89.0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect
 	github.com/prometheus/procfs v0.8.0 // indirect
@@ -196,6 +206,8 @@ require (
 	github.com/rs/xid v1.5.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible // indirect
+	github.com/shirou/gopsutil/v3 v3.23.9 // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/status-im/keycard-go v0.2.0 // indirect
@@ -209,6 +221,7 @@ require (
 	github.com/vincent-petithory/dataurl v1.0.0 // indirect
 	github.com/whyrusleeping/cbor-gen v0.0.0-20230418232409-daab9ece03a0 // indirect
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
+	github.com/yusufpapurcu/wmi v1.2.3 // indirect
 	go.opentelemetry.io/otel v1.16.0 // indirect
 	go.opentelemetry.io/otel/metric v1.16.0 // indirect
 	go.opentelemetry.io/otel/trace v1.16.0 // indirect