package hydrant
import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
// settings for a hydrant that never touches the network, so the test runs offline
func offlineSettings(dir string) map[string]string {
return map[string]string{
"HYDRANT_DATABASE_PATH": filepath.Join(dir, "db"),
"HYDRANT_API_BIND": "unix:" + filepath.Join(dir, "api.sock"),
"HYDRANT_RELAY_HOSTS": "",
"HYDRANT_SEED_HOSTS": "",
"HYDRANT_CRAWLER_URLS": "",
"HYDRANT_ENABLE_CRAWLER": "false",
"RUST_LOG": "warn",
}
}
// a short temp dir, because unix socket paths are capped at ~104 bytes on darwin
func shortTempDir(t *testing.T) string {
t.Helper()
dir, err := os.MkdirTemp("/tmp", "hydrant-ffi-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
// starts hydrant and shuts it down when the test ends, before its dir is removed
func startHydrant(t *testing.T, settings map[string]string) *Hydrant {
t.Helper()
h, err := Start(settings)
if err != nil {
t.Fatalf("start: %v", err)
}
t.Cleanup(func() {
if err := h.Shutdown(10 * time.Second); err != nil {
t.Errorf("shutdown: %v", err)
}
})
return h
}
func unixClient(socket string) *http.Client {
return &http.Client{
Timeout: 5 * time.Second,
Transport: &http.Transport{
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
var d net.Dialer
return d.DialContext(ctx, "unix", socket)
},
},
}
}
func waitReady(t *testing.T, h *Hydrant, client *http.Client) {
t.Helper()
deadline := time.Now().Add(30 * time.Second)
for time.Now().Before(deadline) {
select {
case <-h.Done():
t.Fatalf("hydrant stopped before its api came up: %v", h.Err())
default:
}
if resp, err := client.Get("http://hydrant/stats"); err == nil {
resp.Body.Close()
if resp.StatusCode == http.StatusOK {
return
}
}
time.Sleep(100 * time.Millisecond)
}
t.Fatal("hydrant api never answered on the socket")
}
func request(t *testing.T, client *http.Client, method, path, body string) string {
t.Helper()
req, err := http.NewRequest(method, "http://hydrant"+path, strings.NewReader(body))
if err != nil {
t.Fatal(err)
}
if body != "" {
req.Header.Set("Content-Type", "application/json")
}
resp, err := client.Do(req)
if err != nil {
t.Fatalf("%s %s: %v", method, path, err)
}
defer resp.Body.Close()
out, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusOK {
t.Fatalf("%s %s: status %d: %s", method, path, resp.StatusCode, out)
}
return string(out)
}
// opens a /stream websocket and leaves it open until the test ends
func subscribe(t *testing.T, socket string) net.Conn {
t.Helper()
conn, err := net.Dial("unix", socket)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { conn.Close() })
fmt.Fprint(conn, "GET /stream HTTP/1.1\r\nHost: hydrant\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n"+
"Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n")
conn.SetReadDeadline(time.Now().Add(5 * time.Second))
status, err := bufio.NewReader(conn).ReadString('\n')
if err != nil {
t.Fatalf("read upgrade response: %v", err)
}
if !strings.Contains(status, "101") {
t.Errorf("/stream didn't upgrade over the socket: %q", status)
}
return conn
}
func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) {
dir := shortTempDir(t)
settings := offlineSettings(dir)
socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
// the embedded config must come only from the settings map
envDB := filepath.Join(dir, "from-env")
t.Setenv("HYDRANT_DATABASE_PATH", envDB)
h := startHydrant(t, settings)
client := unixClient(socket)
waitReady(t, h, client)
if _, err := os.Stat(settings["HYDRANT_DATABASE_PATH"]); err != nil {
t.Errorf("database not created at the configured path: %v", err)
}
if _, err := os.Stat(envDB); err == nil {
t.Errorf("hydrant read HYDRANT_DATABASE_PATH from the process environment")
}
info, err := os.Stat(socket)
if err != nil {
t.Fatal(err)
}
if mode := info.Mode().Perm(); mode != 0o600 {
t.Errorf("socket mode is %o, want 600", mode)
}
request(t, client, http.MethodPatch, "/filter", `{"mode":"filter","signals":["sh.tangled.repo"]}`)
var filter struct {
Mode string `json:"mode"`
Signals []string `json:"signals"`
}
if err := json.Unmarshal([]byte(request(t, client, http.MethodGet, "/filter", "")), &filter); err != nil {
t.Fatal(err)
}
if filter.Mode != "filter" || len(filter.Signals) != 1 || filter.Signals[0] != "sh.tangled.repo" {
t.Errorf("filter didn't stick: %+v", filter)
}
request(t, client, http.MethodGet, "/stream/head", "")
subscribe(t, socket)
t.Run("a second hydrant on the same database is refused", func(t *testing.T) {
again := offlineSettings(dir)
again["HYDRANT_API_BIND"] = "unix:" + filepath.Join(dir, "other.sock")
_, err := Start(again)
if err == nil || !strings.Contains(err.Error(), "Locked") {
t.Fatalf("got %v, want the database lock to refuse it", err)
}
})
t.Run("a socket that is still served is not taken over", func(t *testing.T) {
other := offlineSettings(shortTempDir(t))
other["HYDRANT_API_BIND"] = settings["HYDRANT_API_BIND"]
h2 := startHydrant(t, other)
select {
case <-h2.Done():
if !strings.Contains(h2.Err().Error(), "listening on this socket") {
t.Errorf("unexpected error: %v", h2.Err())
}
case <-time.After(30 * time.Second):
t.Fatal("second hydrant kept running on a socket that was in use")
}
// the first one must still be serving
request(t, client, http.MethodGet, "/stats", "")
})
}
func TestShutdownLetsTheSameDatabaseStartAgain(t *testing.T) {
dir := shortTempDir(t)
settings := offlineSettings(dir)
socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
client := unixClient(socket)
h, err := Start(settings)
if err != nil {
t.Fatalf("start: %v", err)
}
waitReady(t, h, client)
request(t, client, http.MethodPatch, "/filter", `{"mode":"filter","signals":["sh.tangled.repo"]}`)
// a subscriber holds a stream thread, which has to end too
subscribe(t, socket)
if err := h.Shutdown(10 * time.Second); err != nil {
t.Fatalf("shutdown: %v", err)
}
if !errors.Is(h.Err(), ErrShutdown) {
t.Errorf("Err after Shutdown is %v, want ErrShutdown", h.Err())
}
if err := h.Shutdown(time.Second); err == nil {
t.Error("a second Shutdown should be refused")
}
// the database lock refuses this unless the first hydrant really let go
again := startHydrant(t, settings)
waitReady(t, again, client)
if filter := request(t, client, http.MethodGet, "/filter", ""); !strings.Contains(filter, "sh.tangled.repo") {
t.Errorf("the filter from before the shutdown is gone: %s", filter)
}
}
func TestShutdownAfterTheSubscriberHungUp(t *testing.T) {
dir := shortTempDir(t)
settings := offlineSettings(dir)
socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
h, err := Start(settings)
if err != nil {
t.Fatalf("start: %v", err)
}
waitReady(t, h, unixClient(socket))
// hydrant's goodbye to this socket is a write into a closed pipe, and on
// linux that raises SIGPIPE on a rust thread, which go doesn't handle. the
// sleep lets it get there before the shutdown cuts that task off
subscribe(t, socket).Close()
time.Sleep(200 * time.Millisecond)
if err := h.Shutdown(10 * time.Second); err != nil {
t.Fatalf("shutdown: %v", err)
}
}
func TestSocketModeIsConfigurable(t *testing.T) {
dir := shortTempDir(t)
settings := offlineSettings(dir)
settings["HYDRANT_API_SOCKET_MODE"] = "0660"
socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
h := startHydrant(t, settings)
waitReady(t, h, unixClient(socket))
info, err := os.Stat(socket)
if err != nil {
t.Fatal(err)
}
if mode := info.Mode().Perm(); mode != 0o660 {
t.Fatalf("socket mode is %o, want 660", mode)
}
}
func TestNoApiKeepsRunning(t *testing.T) {
dir := shortTempDir(t)
settings := offlineSettings(dir)
settings["HYDRANT_API_BIND"] = "none"
h := startHydrant(t, settings)
// with nothing to serve, a serve future that returned right away would end hydrant here
select {
case <-h.Done():
t.Fatalf("hydrant stopped without an api: %v", h.Err())
case <-time.After(2 * time.Second):
}
entries, err := os.ReadDir(dir)
if err != nil {
t.Fatal(err)
}
for _, e := range entries {
if e.Type()&os.ModeSocket != 0 {
t.Fatalf("no api should bind nothing, found socket %s", e.Name())
}
}
}
func TestStartRejectsBadSettings(t *testing.T) {
for name, tc := range map[string]struct {
mutate func(map[string]string)
want string
}{
"missing bind": {func(s map[string]string) { delete(s, "HYDRANT_API_BIND") }, "HYDRANT_API_BIND is required"},
"bad bind": {func(s map[string]string) { s["HYDRANT_API_BIND"] = "nope" }, "invalid HYDRANT_API_BIND"},
"bad setting": {func(s map[string]string) { s["HYDRANT_HISTORY_TTL"] = "soon" }, "HYDRANT_HISTORY_TTL"},
"bad mode": {func(s map[string]string) { s["HYDRANT_API_SOCKET_MODE"] = "0800" }, "HYDRANT_API_SOCKET_MODE"},
} {
t.Run(name, func(t *testing.T) {
settings := offlineSettings(shortTempDir(t))
tc.mutate(settings)
_, err := Start(settings)
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("got %v, want an error mentioning %q", err, tc.want)
}
})
}
}