Tests for cliet package

Makefile

@@ -1,4 +1,3 @@
-GO=$(shell which go)
 VERSION := $(shell git describe --tag)
 
 .PHONY:
@@ -50,20 +49,20 @@ docs: docs-deps
 check: test fmt-check vet lint staticcheck
 
 test: .PHONY
-	$(GO) test -v ./...
+	go test -v $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
 
 race: .PHONY
-	$(GO) test -race ./...
+	go test -race $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
 
 coverage:
 	mkdir -p build/coverage
-	$(GO) test -race -coverprofile=build/coverage/coverage.txt -covermode=atomic ./...
+	go test -race -coverprofile=build/coverage/coverage.txt -covermode=atomic $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
-	$(GO) tool cover -func build/coverage/coverage.txt
+	go tool cover -func build/coverage/coverage.txt
 
 coverage-html:
 	mkdir -p build/coverage
-	$(GO) test -race -coverprofile=build/coverage/coverage.txt -covermode=atomic ./...
+	go test -race -coverprofile=build/coverage/coverage.txt -covermode=atomic $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
-	$(GO) tool cover -html build/coverage/coverage.txt
+	go tool cover -html build/coverage/coverage.txt
 
 coverage-upload:
 	cd build/coverage && (curl -s https://codecov.io/bash | bash)
@@ -78,17 +77,17 @@ fmt-check:
 	test -z $(shell gofmt -l .)
 
 vet:
-	$(GO) vet ./...
+	go vet ./...
 
 lint:
-	which golint || $(GO) get -u golang.org/x/lint/golint
+	which golint || go get -u golang.org/x/lint/golint
-	$(GO) list ./... | grep -v /vendor/ | xargs -L1 golint -set_exit_status
+	go list ./... | grep -v /vendor/ | xargs -L1 golint -set_exit_status
 
 staticcheck: .PHONY
 	rm -rf build/staticcheck
 	which staticcheck || go install honnef.co/go/tools/cmd/staticcheck@latest
 	mkdir -p build/staticcheck
-	ln -s "$(GO)" build/staticcheck/go
+	ln -s "go" build/staticcheck/go
 	PATH="$(PWD)/build/staticcheck:$(PATH)" staticcheck ./...
 	rm -rf build/staticcheck
 
@@ -108,7 +107,7 @@ build-snapshot: build-deps
 build-simple: clean
 	mkdir -p dist/ntfy_linux_amd64
 	export CGO_ENABLED=1
-	$(GO) build \
+	go build \
 		-o dist/ntfy_linux_amd64/ntfy \
 		-tags sqlite_omit_load_extension,osusergo,netgo \
 		-ldflags \

client/client.go

@@ -35,7 +35,7 @@ type Client struct {
 }
 
 // Message is a struct that represents a ntfy message
-type Message struct {
+type Message struct { // TODO combine with server.message
 	ID       string
 	Event    string
 	Time     int64
@@ -60,7 +60,7 @@ type subscription struct {
 // New creates a new Client using a given Config
 func New(config *Config) *Client {
 	return &Client{
-		Messages:      make(chan *Message),
+		Messages:      make(chan *Message, 50), // Allow reading a few messages
 		config:        config,
 		subscriptions: make(map[string]*subscription),
 	}
@@ -237,8 +237,10 @@ func performSubscribeRequest(ctx context.Context, msgChan chan *Message, topicUR
 		if err != nil {
 			return err
 		}
+		if m.Event == MessageEvent {
 			msgChan <- m
 		}
+	}
 	return nil
 }
 

client/client_test.go

@@ -1,42 +1,78 @@
 package client_test
 
 import (
+	"fmt"
 	"github.com/stretchr/testify/require"
 	"heckel.io/ntfy/client"
-	"heckel.io/ntfy/server"
+	"heckel.io/ntfy/test"
-	"net/http"
 	"testing"
 	"time"
 )
 
-func TestClient_Publish(t *testing.T) {
+func TestClient_Publish_Subscribe(t *testing.T) {
-	s := startTestServer(t)
+	s, port := test.StartServer(t)
-	defer s.Stop()
+	defer test.StopServer(t, s, port)
-	c := client.New(newTestConfig())
+	c := client.New(newTestConfig(port))
 
-	time.Sleep(time.Second) // FIXME Wait for port up
+	subscriptionID := c.Subscribe("mytopic")
+	time.Sleep(time.Second)
 
-	_, err := c.Publish("mytopic", "some message")
+	msg, err := c.Publish("mytopic", "some message")
 	require.Nil(t, err)
+	require.Equal(t, "some message", msg.Message)
+
+	msg, err = c.Publish("mytopic", "some other message",
+		client.WithTitle("some title"),
+		client.WithPriority("high"),
+		client.WithTags([]string{"tag1", "tag 2"}))
+	require.Nil(t, err)
+	require.Equal(t, "some other message", msg.Message)
+	require.Equal(t, "some title", msg.Title)
+	require.Equal(t, []string{"tag1", "tag 2"}, msg.Tags)
+	require.Equal(t, 4, msg.Priority)
+
+	msg, err = c.Publish("mytopic", "some delayed message",
+		client.WithDelay("25 hours"))
+	require.Nil(t, err)
+	require.Equal(t, "some delayed message", msg.Message)
+	require.True(t, time.Now().Add(24*time.Hour).Unix() < msg.Time)
+
+	msg = nextMessage(c)
+	require.NotNil(t, msg)
+	require.Equal(t, "some message", msg.Message)
+
+	msg = nextMessage(c)
+	require.NotNil(t, msg)
+	require.Equal(t, "some other message", msg.Message)
+	require.Equal(t, "some title", msg.Title)
+	require.Equal(t, []string{"tag1", "tag 2"}, msg.Tags)
+	require.Equal(t, 4, msg.Priority)
+
+	msg = nextMessage(c)
+	require.Nil(t, msg)
+
+	c.Unsubscribe(subscriptionID)
+	time.Sleep(200 * time.Millisecond)
+
+	msg, err = c.Publish("mytopic", "a message that won't be received")
+	require.Nil(t, err)
+	require.Equal(t, "a message that won't be received", msg.Message)
+
+	msg = nextMessage(c)
+	require.Nil(t, msg)
 }
 
-func newTestConfig() *client.Config {
+func newTestConfig(port int) *client.Config {
 	c := client.NewConfig()
-	c.DefaultHost = "http://127.0.0.1:12345"
+	c.DefaultHost = fmt.Sprintf("http://127.0.0.1:%d", port)
 	return c
 }
 
-func startTestServer(t *testing.T) *server.Server {
+func nextMessage(c *client.Client) *client.Message {
-	conf := server.NewConfig()
+	select {
-	conf.ListenHTTP = ":12345"
+	case m := <-c.Messages:
-	s, err := server.New(conf)
+		return m
-	if err != nil {
+	default:
-		t.Fatal(err)
+		return nil
 	}
-	go func() {
-		if err := s.Run(); err != nil && err != http.ErrServerClosed {
-			panic(err) // 'go vet' complains about 't.Fatal(err)'
-		}
-	}()
-	return s
 }

cmd/subscribe.go

@@ -158,9 +158,6 @@ func doSubscribe(c *cli.Context, cl *client.Client, conf *client.Config, topic,
 }
 
 func printMessageOrRunCommand(c *cli.Context, m *client.Message, command string) {
-	if m.Event != client.MessageEvent {
-		return
-	}
 	if command != "" {
 		runCommand(c, command, m)
 	} else {

server/server.go

@@ -191,9 +191,6 @@ func createFirebaseSubscriber(conf *Config) (subscriber, error) {
 // Run executes the main server. It listens on HTTP (+ HTTPS, if configured), and starts
 // a manager go routine to print stats and prune messages.
 func (s *Server) Run() error {
-	go s.runManager()
-	go s.runAtSender()
-	go s.runFirebaseKeepliver()
 	listenStr := fmt.Sprintf("%s/http", s.config.ListenHTTP)
 	if s.config.ListenHTTPS != "" {
 		listenStr += fmt.Sprintf(" %s/https", s.config.ListenHTTPS)
@@ -214,6 +211,9 @@ func (s *Server) Run() error {
 		}()
 	}
 	s.mu.Unlock()
+	go s.runManager()
+	go s.runAtSender()
+	go s.runFirebaseKeepliver()
 	return <-errChan
 }
 

test/server.go

@@ -0,0 +1,38 @@
+package test
+
+import (
+	"fmt"
+	"heckel.io/ntfy/server"
+	"math/rand"
+	"net/http"
+	"testing"
+	"time"
+)
+
+func init() {
+	rand.Seed(time.Now().Unix())
+}
+
+// StartServer starts a server.Server with a random port and waits for the server to be up
+func StartServer(t *testing.T) (*server.Server, int) {
+	port := 10000 + rand.Intn(20000)
+	conf := server.NewConfig()
+	conf.ListenHTTP = fmt.Sprintf(":%d", port)
+	s, err := server.New(conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go func() {
+		if err := s.Run(); err != nil && err != http.ErrServerClosed {
+			panic(err) // 'go vet' complains about 't.Fatal(err)'
+		}
+	}()
+	WaitForPortUp(t, port)
+	return s, port
+}
+
+// StopServer stops the test server and waits for the port to be down
+func StopServer(t *testing.T, s *server.Server, port int) {
+	s.Stop()
+	WaitForPortDown(t, port)
+}

test/test.go

@@ -0,0 +1,3 @@
+// Package test provides test helpers for unit and integration tests.
+// This code is not meant to be used outside of tests.
+package test

test/util.go

@@ -0,0 +1,44 @@
+package test
+
+import (
+	"net"
+	"strconv"
+	"testing"
+	"time"
+)
+
+// WaitForPortUp waits up to 7s for a port to come up and fails t if that fails
+func WaitForPortUp(t *testing.T, port int) {
+	success := false
+	for i := 0; i < 500; i++ {
+		startTime := time.Now()
+		conn, _ := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), 10*time.Millisecond)
+		if conn != nil {
+			success = true
+			conn.Close()
+			break
+		}
+		if time.Since(startTime) < 10*time.Millisecond {
+			time.Sleep(10*time.Millisecond - time.Since(startTime))
+		}
+	}
+	if !success {
+		t.Fatalf("Failed waiting for port %d to be UP", port)
+	}
+}
+
+// WaitForPortDown waits up to 5s for a port to come down and fails t if that fails
+func WaitForPortDown(t *testing.T, port int) {
+	success := false
+	for i := 0; i < 100; i++ {
+		conn, _ := net.DialTimeout("tcp", net.JoinHostPort("", strconv.Itoa(port)), 50*time.Millisecond)
+		if conn == nil {
+			success = true
+			break
+		}
+		conn.Close()
+	}
+	if !success {
+		t.Fatalf("Failed waiting for port %d to be DOWN", port)
+	}
+}