chore: enable integration test in ci

Author: astromechza

.github/workflows/tests.yml

@@ -10,6 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     env:
       GOPROXY: "https://proxy.golang.org,direct"
+      ENABLE_DOCKER_INTEGRATION_TESTS: TRUE
 
     steps:
       - name: Set up Go

integration_test.go

@@ -14,7 +14,7 @@ import (
 const enableDockerIntegrationTestsFlag = `ENABLE_DOCKER_INTEGRATION_TESTS`
 
 func prepareDockerTest(t *testing.T) (connStr string) {
-	if v, ok := os.LookupEnv(enableDockerIntegrationTestsFlag); ok && strings.ToUpper(v) != "TRUE" {
+	if v, ok := os.LookupEnv(enableDockerIntegrationTestsFlag); !ok || strings.ToUpper(v) != "TRUE" {
 		t.Skipf("integration tests are only run if '%s' is TRUE", enableDockerIntegrationTestsFlag)
 		return
 	}
@@ -27,7 +27,6 @@ func prepareDockerTest(t *testing.T) (connStr string) {
 		t.Fatalf("error launching rabbitmq in docker: %v", err)
 	}
 	t.Cleanup(func() {
-		t.Log("hi")
 		containerId := strings.TrimSpace(string(out))
 		t.Logf("attempting to shutdown container '%s'", containerId)
 		if err := exec.Command("docker", "rm", "--force", containerId).Run(); err != nil {
@@ -37,7 +36,7 @@ func prepareDockerTest(t *testing.T) (connStr string) {
 	return "amqp://guest:guest@localhost:5672/"
 }
 
-func waitForHealthyAmqp(t *testing.T, connStr string) {
+func waitForHealthyAmqp(t *testing.T, connStr string) *Conn {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
 	tkr := time.NewTicker(time.Second)
@@ -47,58 +46,66 @@ func waitForHealthyAmqp(t *testing.T, connStr string) {
 		case <-ctx.Done():
 			t.Fatal("timed out waiting for healthy amqp", ctx.Err())
 		case <-tkr.C:
-			if err := func() error {
-				t.Log("attempting connection")
-				conn, err := NewConn(connStr)
-				if err != nil {
-					return fmt.Errorf("failed to setup connection: %v", err)
-				}
-				defer conn.Close()
-
-				pub, err := NewPublisher(conn)
-				if err != nil {
-					return fmt.Errorf("failed to setup publisher: %v", err)
-				}
-
-				t.Log("attempting publish")
-				return pub.PublishWithContext(ctx, []byte{}, []string{"ping"}, WithPublishOptionsExchange(""))
-			}(); err != nil {
-				t.Log("publish ping failed", err.Error())
+			conn, err := NewConn(connStr)
+			if err != nil {
+				t.Log("failed to connect", err.Error())
 			} else {
-				t.Log("ping successful")
-				return
+				if err := func() error {
+					t.Log("attempting connection")
+
+					pub, err := NewPublisher(conn)
+					if err != nil {
+						return fmt.Errorf("failed to setup publisher: %v", err)
+					}
+
+					t.Log("attempting publish")
+					return pub.PublishWithContext(ctx, []byte{}, []string{"ping"}, WithPublishOptionsExchange(""))
+				}(); err != nil {
+					_ = conn.Close()
+					t.Log("publish ping failed", err.Error())
+				} else {
+					t.Log("ping successful")
+					return conn
+				}
 			}
 		}
 	}
+	return nil
 }
 
 func TestSimplePubSub(t *testing.T) {
 	connStr := prepareDockerTest(t)
-	waitForHealthyAmqp(t, connStr)
-
-	conn, err := NewConn(connStr)
-	if err != nil {
-		t.Fatal("error creating connection", err)
-	}
+	conn := waitForHealthyAmqp(t, connStr)
 	defer conn.Close()
 
 	t.Logf("new consumer")
-	consumer, err := NewConsumer(conn, "my_queue")
+	consumerQueue := "my_queue"
+	consumer, err := NewConsumer(conn, consumerQueue)
 	if err != nil {
 		t.Fatal("error creating consumer", err)
 	}
-	defer consumer.Close()
+	defer consumer.CloseWithContext(context.Background())
+
+	// Setup a consumer which pushes each of its consumed messages over the channel. If the channel is closed or full
+	// it does not block.
+	consumed := make(chan string)
+	defer close(consumed)
 
 	go func() {
 		err = consumer.Run(func(d Delivery) Action {
-			log.Printf("consumed: %v", string(d.Body))
+			log.Printf("consumed")
+			select {
+			case consumed <- string(d.Body):
+			default:
+			}
 			return Ack
 		})
 		if err != nil {
 			t.Log("consumer run failed", err)
 		}
 	}()
 
+	// Setup a publisher with notifications enabled
 	t.Logf("new publisher")
 	publisher, err := NewPublisher(conn)
 	if err != nil {
@@ -107,23 +114,33 @@ func TestSimplePubSub(t *testing.T) {
 	publisher.NotifyPublish(func(p Confirmation) {
 		return
 	})
+	defer publisher.Close()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	// For test stability we cannot rely on the fact that the consumer go routines are up and running before the
+	// publisher starts it's first publish attempt. For this reason we run the publisher in a loop every second and
+	// pass after we see the first message come through.
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
-
-	t.Logf("new publish")
-	confirms, err := publisher.PublishWithDeferredConfirmWithContext(
-		ctx, []byte("example"), []string{"my_queue"},
-		WithPublishOptionsMandatory,
-	)
-	if err != nil {
-		t.Fatal("failed to publish", err)
-	}
-	for _, confirm := range confirms {
-		if _, err := confirm.WaitContext(ctx); err != nil {
-			t.Fatal("failed to wait for publish", err)
+	tkr := time.NewTicker(time.Second)
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatal("timed out waiting for pub sub", ctx.Err())
+		case <-tkr.C:
+			t.Logf("new publish")
+			confirms, err := publisher.PublishWithDeferredConfirmWithContext(ctx, []byte("example"), []string{consumerQueue})
+			if err != nil {
+				// publish should always succeed since we've verified the ping previously
+				t.Fatal("failed to publish", err)
+			}
+			for _, confirm := range confirms {
+				if _, err := confirm.WaitContext(ctx); err != nil {
+					t.Fatal("failed to wait for publish", err)
+				}
+			}
+		case msg := <-consumed:
+			t.Logf("successfully saw message round trip: '%s'", msg)
+			return
 		}
 	}
-	t.Logf("success")
-
 }