[WIP] bring up docker env in unit test

.circleci/config.yml

@@ -4,6 +4,7 @@ jobs:
     working_directory: /go/src/github.com/graphite-ng/carbon-relay-ng
     docker:
       - image: circleci/golang:1.10.3
+      - image: rabbitmq
     steps:
       - checkout
       - run: go test -v -race ./...

input/amqp_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/graphite-ng/carbon-relay-ng/cfg"
 	"github.com/streadway/amqp"
 )
 
@@ -30,10 +31,14 @@ func getMockConnector() (chan amqp.Delivery, *MockClosable, *MockClosable, amqpC
 
 type mockDispatcher struct {
 	dispatchDuration time.Duration
+	receivedData     []byte
 }
 
-func (m *mockDispatcher) Dispatch(buf []byte) {}
-func (m *mockDispatcher) IncNumInvalid()      {}
+func (m *mockDispatcher) Dispatch(buf []byte) {
+	m.receivedData = append(m.receivedData, buf...)
+	time.Sleep(m.dispatchDuration)
+}
+func (m *mockDispatcher) IncNumInvalid() {}
 
 func TestMain(m *testing.M) {
 	_shutdownTimeout := shutdownTimeout
@@ -96,3 +101,21 @@ func TestAmqpFailingShutdown(t *testing.T) {
 		t.Fatalf("Expected channel and connection to be closed, but they were not")
 	}
 }
+
+// this test assumes that a rabbitmq is available on localhost
+func TestAmqpConsumeRabbit(t *testing.T) {
+	config := cfg.Config{
+		Amqp: cfg.Amqp{
+			Amqp_host:     "localhost",
+			Amqp_port:     5672,
+			Amqp_user:     "guest",
+			Amqp_password: "guest",
+			Amqp_vhost:    "guest_vhost",
+			Amqp_exchange: "guest_exchange",
+		},
+	}
+	dispatcher := &mockDispatcher{}
+
+	a := NewAMQP(config, dispatcher, AMQPConnector)
+	a.Start()
+}

stacktest/amqp/amqp_test.go

@@ -0,0 +1,65 @@
+package amqp
+
+import (
+	"os/exec"
+	"testing"
+)
+
+// relativePath takes a relative path inside this repository and returns
+// the full path to the given destination
+func relativePath(dst string) string {
+	gopath := os.Getenv("GOPATH")
+	if gopath == "" {
+		var err error
+		gopath, err = homedir.Expand("~/go")
+		if err != nil {
+			panic(err)
+		}
+	}
+	firstPath := strings.Split(gopath, ":")[0]
+	return p.Join(firstPath, "src/github.com/graphite-ng/carbon-relay-ng", dst)
+}
+
+func TestMain(m *testing.M) {
+	log.Println("launching docker-dev stack...")
+	version := exec.Command("docker-compose", "version")
+	output, err := version.CombinedOutput()
+	if err != nil {
+		log.Fatal(err.Error())
+	}
+	log.Println(string(output))
+
+	cmd := exec.Command("docker-compose", "up", "--force-recreate")
+	cmd.Dir = relativePath("stacktest/amqp")
+
+	readPipe := func(in io.ReadCloser) {
+		scanner := bufio.NewScanner(in)
+		for scanner.Scan() {
+			// consume pipes, even if we don't look at the output
+		}
+	}
+
+	go readPipe(cmd.StdoutPipe())
+	go readPipe(cmd.StderrPipe())
+
+	err = cmd.Start()
+	if err != nil {
+		log.Fatal(err.Error())
+	}
+
+	retcode := m.Run()
+
+	log.Println("stopping docker-compose stack...")
+	cmd.Process.Signal(syscall.SIGINT)
+	err = cmd.Wait()
+
+	// 130 means ctrl-C (interrupt) which is what we want
+	if err != nil && err.Error() != "exit status 130" {
+		log.Printf("ERROR: could not cleanly shutdown running docker-compose command: %s", err)
+		retcode = 1
+	} else {
+		log.Println("docker-compose stack is shut down")
+	}
+
+	os.Exit(retcode)
+}