Add tests

internal/controller/controller_test.go

@@ -163,14 +163,17 @@ func TestStartStop(t *testing.T) {
 		require.NoError(t, err)
 	}()
 
-	time.Sleep(time.Millisecond * 100)
-	c.Stop()
+	require.Eventually(t, func() bool {
+		st, _ := c.GetStatus()
+		return st.Status == scheduler.SchedulerStatus_Running
+	}, time.Millisecond*1500, time.Millisecond*100)
 
-	time.Sleep(time.Millisecond * 100)
-	s, err := c.GetLastStatus()
-	require.NoError(t, err)
+	c.Stop()
 
-	require.Equal(t, scheduler.SchedulerStatus_Cancel, s.Status)
+	require.Eventually(t, func() bool {
+		st, _ := c.GetLastStatus()
+		return st.Status == scheduler.SchedulerStatus_Cancel
+	}, time.Millisecond*1500, time.Millisecond*100)
 }
 
 func TestRetry(t *testing.T) {

internal/scheduler/node.go

@@ -104,7 +104,7 @@ func (n *Node) updateStatus(status NodeStatus) {
 
 func (n *Node) signal(sig os.Signal) {
 	status := n.ReadStatus()
-	if status == NodeStatus_Running {
+	if status == NodeStatus_None || status == NodeStatus_Running {
 		n.updateStatus(NodeStatus_Cancel)
 	}
 	if n.cmd != nil {

internal/scheduler/scheduler_test.go

@@ -331,6 +331,30 @@ func TestSchedulerOnExitOnFail(t *testing.T) {
 	assert.Equal(t, scheduler.NodeStatus_Success, sc.HanderNode(constants.OnExit).ReadStatus())
 }
 
+func TestSchedulerOnSignal(t *testing.T) {
+	g, _ := scheduler.NewExecutionGraph(
+		&config.Step{
+			Name:    "1",
+			Command: "sleep",
+			Args:    []string{"10"},
+		},
+	)
+	sc := scheduler.New(&scheduler.Config{})
+
+	go func() {
+		<-time.After(time.Millisecond * 50)
+		sc.Signal(g, syscall.SIGTERM, nil)
+	}()
+
+	err := sc.Schedule(g, nil)
+	require.NoError(t, err)
+
+	nodes := g.Nodes()
+
+	assert.Equal(t, sc.Status(g), scheduler.SchedulerStatus_Cancel)
+	assert.Equal(t, scheduler.NodeStatus_Cancel, nodes[0].Status)
+}
+
 func TestSchedulerOnCancel(t *testing.T) {
 	g, sc := newTestSchedule(t,
 		&scheduler.Config{

internal/sock/address.go

@@ -15,5 +15,5 @@ func GetSockAddr(key string) string {
 	h := md5.New()
 	h.Write([]byte(s))
 	bs := h.Sum(nil)
-	return path.Join(sockDir, fmt.Sprintf("@dagu-%s-%x", name, bs))
+	return path.Join(sockDir, fmt.Sprintf("@dagu-%s-%x.sock", name, bs))
 }

internal/sock/address_test.go

@@ -0,0 +1,12 @@
+package sock
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSockAddr(t *testing.T) {
+	val := GetSockAddr("test")
+	require.Regexp(t, `^/tmp/@dagu-test-[0-9a-f]+\.sock$`, val)
+}

internal/sock/client.go

@@ -8,36 +8,23 @@ import (
 	"net"
 	"net/http"
 	"time"
-
-	"github.com/yohamta/dagu/internal/utils"
 )
 
 var ErrTimeout = fmt.Errorf("unix socket timeout")
 var ErrConnectionRefused = fmt.Errorf("unix socket connection failed")
-var ErrFileNotExist = fmt.Errorf("unix socket file does not exit")
 var timeout = time.Millisecond * 3000
 
 type Client struct {
 	Addr string
 }
 
 func (cl *Client) Request(method, url string) (string, error) {
-	if !utils.FileExists(cl.Addr) {
-		return "", fmt.Errorf("%w: %s", ErrFileNotExist, cl.Addr)
-	}
 	conn, err := net.DialTimeout("unix", cl.Addr, timeout)
 	if err != nil {
-		if err.(net.Error).Timeout() {
-			return "", fmt.Errorf("%s: %w", err, ErrTimeout)
-		} else {
-			return "", fmt.Errorf("%s: %w", err, ErrConnectionRefused)
-		}
+		return "", procError("dial to socket", err)
 	}
 	defer conn.Close()
-	err = conn.SetDeadline((time.Now().Add(timeout)))
-	if err != nil {
-		return "", err
-	}
+	conn.SetDeadline((time.Now().Add(timeout)))
 	request, err := http.NewRequest(method, url, nil)
 	if err != nil {
 		log.Printf("NewRequest %v", err)
@@ -46,19 +33,20 @@ func (cl *Client) Request(method, url string) (string, error) {
 	request.Write(conn)
 	response, err := http.ReadResponse(bufio.NewReader(conn), request)
 	if err != nil {
-		if err.(net.Error).Timeout() {
-			return "", fmt.Errorf("%s: %w", err, ErrTimeout)
-		} else {
-			return "", fmt.Errorf("failed to read: %w addr=%s", err, cl.Addr)
-		}
+		return "", procError("read response", err)
 	}
 	body, err := io.ReadAll(response.Body)
 	if err != nil {
-		if err.(net.Error).Timeout() {
-			return "", fmt.Errorf("%s : %w", err, ErrTimeout)
-		} else {
-			return "", fmt.Errorf("failed to write: %w", err)
-		}
+		return "", procError("read response body", err)
 	}
 	return string(body), nil
 }
+
+func procError(action string, err error) error {
+	if err, ok := err.(net.Error); ok {
+		if err.Timeout() {
+			return fmt.Errorf("%s timeout %w: %s", action, ErrTimeout, err.Error())
+		}
+	}
+	return fmt.Errorf("%s failed: %w", action, err)
+}

internal/sock/client_test.go

@@ -1,35 +1,36 @@
-package sock_test
+package sock
 
 import (
 	"errors"
+	"fmt"
 	"io/ioutil"
+	"net"
 	"net/http"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/yohamta/dagu/internal/sock"
 )
 
 func TestDialFail(t *testing.T) {
 	f, err := ioutil.TempFile("", "sock_client_dial_failure")
 	require.NoError(t, err)
 	defer os.Remove(f.Name())
 
-	client := sock.Client{Addr: f.Name()}
+	client := Client{Addr: f.Name()}
 	_, err = client.Request("GET", "/status")
-	assert.True(t, errors.Is(err, sock.ErrConnectionRefused))
+	require.Error(t, err)
 }
 
 func TestDialTimeout(t *testing.T) {
 	f, err := ioutil.TempFile("", "sock_client_test")
 	require.NoError(t, err)
 	defer os.Remove(f.Name())
 
-	s, err := sock.NewServer(
-		&sock.Config{
+	s, err := NewServer(
+		&Config{
 			Addr: f.Name(),
 			HandlerFunc: func(w http.ResponseWriter, r *http.Request) {
 				time.Sleep(time.Second * 3100)
@@ -46,8 +47,23 @@ func TestDialTimeout(t *testing.T) {
 	time.Sleep(time.Millisecond * 500)
 
 	require.NoError(t, err)
-	client := sock.Client{Addr: f.Name()}
+	client := Client{Addr: f.Name()}
 	_, err = client.Request("GET", "/status")
 	require.Error(t, err)
-	assert.True(t, errors.Is(err, sock.ErrTimeout))
+	assert.True(t, errors.Is(err, ErrTimeout))
+}
+
+func TestProcErr(t *testing.T) {
+	e := procError("test", fmt.Errorf("error"))
+	require.Contains(t, e.Error(), "test failed")
+
+	e = procError("test", errTimeout)
+	require.Contains(t, e.Error(), "test timeout")
 }
+
+type testTimeout struct{ error }
+
+var errTimeout net.Error = &testTimeout{error: fmt.Errorf("timeout")}
+
+func (t *testTimeout) Timeout() bool   { return true }
+func (t *testTimeout) Temporary() bool { return false }

internal/sock/server_test.go

@@ -1,4 +1,4 @@
-package sock_test
+package sock
 
 import (
 	"errors"
@@ -10,7 +10,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/yohamta/dagu/internal/sock"
 )
 
 func TestMain(m *testing.M) {
@@ -29,8 +28,8 @@ func TestStartAndShutdownServer(t *testing.T) {
 	require.NoError(t, err)
 	defer os.Remove(tmpFile.Name())
 
-	unixServer, err := sock.NewServer(
-		&sock.Config{
+	unixServer, err := NewServer(
+		&Config{
 			Addr: tmpFile.Name(),
 			HandlerFunc: func(w http.ResponseWriter, r *http.Request) {
 				w.WriteHeader(http.StatusOK)
@@ -39,7 +38,7 @@ func TestStartAndShutdownServer(t *testing.T) {
 		})
 	require.NoError(t, err)
 
-	client := sock.Client{Addr: tmpFile.Name()}
+	client := Client{Addr: tmpFile.Name()}
 	listen := make(chan error)
 	go func() {
 		for range listen {
@@ -48,22 +47,84 @@ func TestStartAndShutdownServer(t *testing.T) {
 
 	go func() {
 		err = unixServer.Serve(listen)
-		assert.True(t, errors.Is(sock.ErrServerRequestedShutdown, err))
+		assert.True(t, errors.Is(ErrServerRequestedShutdown, err))
 	}()
 
-	time.Sleep(time.Second * 1)
+	time.Sleep(time.Millisecond * 50)
 
 	ret, err := client.Request(http.MethodPost, "/")
-	assert.Equal(t, ret, "OK")
+	assert.Equal(t, "OK", ret)
 
 	unixServer.Shutdown()
 
-	time.Sleep(time.Millisecond * 100)
+	time.Sleep(time.Millisecond * 50)
 	_, err = client.Request(http.MethodPost, "/")
-	assert.True(t, errors.Is(err, sock.ErrFileNotExist))
+	assert.Error(t, err)
+}
+
+func TestNoResponse(t *testing.T) {
+	tmpFile, err := ioutil.TempFile("", "test_error_response")
+	require.NoError(t, err)
+	defer os.Remove(tmpFile.Name())
+
+	unixServer, err := NewServer(
+		&Config{
+			Addr: tmpFile.Name(),
+			HandlerFunc: func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusForbidden)
+			},
+		})
+	require.NoError(t, err)
+
+	client := Client{Addr: tmpFile.Name()}
+	listen := make(chan error)
+	go func() {
+		for range listen {
+		}
+	}()
+
+	go func() {
+		err = unixServer.Serve(listen)
+		defer unixServer.Shutdown()
+	}()
+
+	time.Sleep(time.Millisecond * 50)
+
+	_, err = client.Request(http.MethodGet, "/")
+	require.Error(t, err)
+}
+
+func TestErrorResponse(t *testing.T) {
+	tmpFile, err := ioutil.TempFile("", "test_error_response")
+	require.NoError(t, err)
+	defer os.Remove(tmpFile.Name())
+
+	unixServer, err := NewServer(
+		&Config{
+			Addr:        tmpFile.Name(),
+			HandlerFunc: func(w http.ResponseWriter, r *http.Request) {},
+		})
+	require.NoError(t, err)
+
+	client := Client{Addr: tmpFile.Name()}
+	listen := make(chan error)
+	go func() {
+		for range listen {
+		}
+	}()
+
+	go func() {
+		err = unixServer.Serve(listen)
+		defer unixServer.Shutdown()
+	}()
+
+	time.Sleep(time.Millisecond * 50)
+
+	_, err = client.Request(http.MethodGet, "/")
+	require.Error(t, err)
 }
 
 func TestResponseWriter(t *testing.T) {
-	w := sock.NewHttpResponseWriter(nil)
+	w := NewHttpResponseWriter(nil)
 	require.Equal(t, make(http.Header), w.Header())
 }

tests/testdata/controller_start.yaml

@@ -1,4 +1,4 @@
 name: "test"
 steps:
   - name: "1"
-    command: "sleep 1"
+    command: "sleep 10"