fix remaining broken tests relying on worker queue logic

internal/federation/federatingdb/announce_test.go

@@ -19,6 +19,7 @@ package federatingdb_test
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/activity/streams/vocab"
@@ -42,7 +43,7 @@ func (suite *AnnounceTestSuite) TestNewAnnounce() {
 	suite.NoError(err)
 
 	// should be a message heading to the processor now, which we can intercept here
-	msg := <-suite.fromFederator
+	msg, _ := suite.getFederatorMsg(5 * time.Second)
 	suite.Equal(ap.ActivityAnnounce, msg.APObjectType)
 	suite.Equal(ap.ActivityCreate, msg.APActivityType)
 
@@ -69,7 +70,7 @@ func (suite *AnnounceTestSuite) TestAnnounceTwice() {
 	suite.NoError(err)
 
 	// should be a message heading to the processor now, which we can intercept here
-	msg := <-suite.fromFederator
+	msg, _ := suite.getFederatorMsg(5 * time.Second)
 	suite.Equal(ap.ActivityAnnounce, msg.APObjectType)
 	suite.Equal(ap.ActivityCreate, msg.APActivityType)
 	boost, ok := msg.GTSModel.(*gtsmodel.Status)
@@ -94,7 +95,8 @@ func (suite *AnnounceTestSuite) TestAnnounceTwice() {
 
 	// since this is a repeat announce with the same URI, just delivered to a different inbox,
 	// we should have nothing in the messages channel...
-	suite.Empty(suite.fromFederator)
+	_, ok = suite.getFederatorMsg(time.Second)
+	suite.False(ok)
 }
 
 func TestAnnounceTestSuite(t *testing.T) {

internal/federation/federatingdb/create_test.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/activity/streams"
@@ -48,7 +49,7 @@ func (suite *CreateTestSuite) TestCreateNote() {
 	suite.NoError(err)
 
 	// should be a message heading to the processor now, which we can intercept here
-	msg := <-suite.fromFederator
+	msg, _ := suite.getFederatorMsg(5 * time.Second)
 	suite.Equal(ap.ObjectNote, msg.APObjectType)
 	suite.Equal(ap.ActivityCreate, msg.APActivityType)
 	suite.Equal(note, msg.APObject)

internal/federation/federatingdb/federatingdb_test.go

@@ -19,6 +19,7 @@ package federatingdb_test
 
 import (
 	"context"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
@@ -34,11 +35,10 @@ import (
 
 type FederatingDBTestSuite struct {
 	suite.Suite
-	db            db.DB
-	tc            *typeutils.Converter
-	fromFederator chan messages.FromFediAPI
-	federatingDB  federatingdb.DB
-	state         state.State
+	db           db.DB
+	tc           *typeutils.Converter
+	federatingDB federatingdb.DB
+	state        state.State
 
 	testTokens       map[string]*gtsmodel.Token
 	testClients      map[string]*gtsmodel.Client
@@ -51,6 +51,13 @@ type FederatingDBTestSuite struct {
 	testActivities   map[string]testrig.ActivityWithSignature
 }
 
+func (suite *FederatingDBTestSuite) getFederatorMsg(timeout time.Duration) (*messages.FromFediAPI, bool) {
+	ctx := context.Background()
+	ctx, cncl := context.WithTimeout(ctx, timeout)
+	defer cncl()
+	return suite.state.Workers.Federator.Queue.PopCtx(ctx)
+}
+
 func (suite *FederatingDBTestSuite) SetupSuite() {
 	suite.testTokens = testrig.NewTestTokens()
 	suite.testClients = testrig.NewTestClients()
@@ -69,8 +76,6 @@ func (suite *FederatingDBTestSuite) SetupTest() {
 	suite.state.Caches.Init()
 	testrig.StartNoopWorkers(&suite.state)
 
-	suite.fromFederator = make(chan messages.FromFediAPI, 10)
-
 	suite.db = testrig.NewTestDB(&suite.state)
 
 	suite.testActivities = testrig.NewTestActivities(suite.testAccounts)
@@ -91,13 +96,6 @@ func (suite *FederatingDBTestSuite) SetupTest() {
 func (suite *FederatingDBTestSuite) TearDownTest() {
 	testrig.StandardDBTeardown(suite.db)
 	testrig.StopWorkers(&suite.state)
-	for suite.fromFederator != nil {
-		select {
-		case <-suite.fromFederator:
-		default:
-			return
-		}
-	}
 }
 
 func createTestContext(receivingAccount *gtsmodel.Account, requestingAccount *gtsmodel.Account) context.Context {

internal/federation/federatingdb/move_test.go

@@ -27,7 +27,6 @@ import (
 	"github.com/superseriousbusiness/activity/streams/vocab"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
-	"github.com/superseriousbusiness/gotosocial/internal/messages"
 )
 
 type MoveTestSuite struct {
@@ -78,13 +77,7 @@ func (suite *MoveTestSuite) TestMove() {
 	suite.move(receivingAcct, requestingAcct, moveStr1)
 
 	// Should be a message heading to the processor.
-	var msg messages.FromFediAPI
-	select {
-	case msg = <-suite.fromFederator:
-		// Fine.
-	case <-time.After(5 * time.Second):
-		suite.FailNow("", "timeout waiting for suite.fromFederator")
-	}
+	msg, _ := suite.getFederatorMsg(5 * time.Second)
 	suite.Equal(ap.ObjectProfile, msg.APObjectType)
 	suite.Equal(ap.ActivityMove, msg.APActivityType)
 
@@ -101,12 +94,7 @@ func (suite *MoveTestSuite) TestMove() {
 
 	// Should be a message heading to the processor
 	// since this is just a straight up retry.
-	select {
-	case msg = <-suite.fromFederator:
-		// Fine.
-	case <-time.After(5 * time.Second):
-		suite.FailNow("", "timeout waiting for suite.fromFederator")
-	}
+	msg, _ = suite.getFederatorMsg(5 * time.Second)
 	suite.Equal(ap.ObjectProfile, msg.APObjectType)
 	suite.Equal(ap.ActivityMove, msg.APActivityType)
 
@@ -126,12 +114,7 @@ func (suite *MoveTestSuite) TestMove() {
 
 	// Should be a message heading to the processor
 	// since this is just a retry with a different ID.
-	select {
-	case msg = <-suite.fromFederator:
-		// Fine.
-	case <-time.After(5 * time.Second):
-		suite.FailNow("", "timeout waiting for suite.fromFederator")
-	}
+	msg, _ = suite.getFederatorMsg(5 * time.Second)
 	suite.Equal(ap.ObjectProfile, msg.APObjectType)
 	suite.Equal(ap.ActivityMove, msg.APActivityType)
 }

internal/federation/federatingdb/reject_test.go

@@ -81,7 +81,8 @@ func (suite *RejectTestSuite) TestRejectFollowRequest() {
 	suite.NoError(err)
 
 	// there should be nothing in the federator channel since nothing needs to be passed
-	suite.Empty(suite.fromFederator)
+	_, ok := suite.getFederatorMsg(time.Second)
+	suite.False(ok)
 
 	// the follow request should not be in the database anymore -- it's been rejected
 	err = suite.db.GetByID(ctx, fr.ID, &gtsmodel.FollowRequest{})