lixen/logwisp
SHA256
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.19.1 dep update, test helper package

Browse Source
This commit is contained in:
lixen committed 2026-09-29 01:22:14 -04:00
1 parent cb0da23070
commit 81463e305c
11 files changed
+179 -161
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/config/loader_test.go
+10 -32
View File
@@ -2,39 +2,17 @@ package config
import (
"errors"
"os"
"path/filepath"
"reflect"
"strings"
"testing"
"time"
"logwisp/internal/testutil"
lconfig "github.com/lixenwraith/config"
)
func isolateConfig(t *testing.T) {
t.Helper()
for _, entry := range os.Environ() {
key, value, _ := strings.Cut(entry, "=")
if strings.HasPrefix(key, "LOGWISP_") {
t.Setenv(key, value)
if err := os.Unsetenv(key); err != nil {
t.Fatal(err)
}
}
}
dir := t.TempDir()
t.Setenv("HOME", dir)
t.Chdir(dir)
}
func writeConfig(t *testing.T, path, contents string) {
t.Helper()
if err := os.WriteFile(path, []byte(contents), 0600); err != nil {
t.Fatal(err)
}
}
func TestResolveConfigArguments(t *testing.T) {
isolateConfig(t)
for _, args := range [][]string{{"-c", "chosen.toml"}, {"--config", "chosen.toml"}, {"--config=chosen.toml"}, {"-c=chosen.toml"}, {"-c", "old.toml", "--config=chosen.toml"}} {
@@ -63,7 +41,7 @@ func TestResolveConfigArguments(t *testing.T) {
func TestLoadPrecedenceAndDetachedSnapshots(t *testing.T) {
isolateConfig(t)
writeConfig(t, "selected.toml", "config_file = \"ignored.toml\"\n[logging]\nlevel = \"error\"\n")
testutil.WriteFile(t, "selected.toml", "config_file = \"ignored.toml\"\n[logging]\nlevel = \"error\"\n")
t.Setenv("LOGGING_LEVEL", "invalid-bare-variable")
t.Setenv("LOGWISP_LOGGING_LEVEL", "debug")
m, err := Load([]string{"--config", "selected.toml", "--logging.level=warn", "--quiet"})
@@ -116,7 +94,7 @@ func TestMissingFilesDoNotHideInvalidOverrides(t *testing.T) {
func TestReloadReadsDiskValidatesAndPreservesOldSnapshots(t *testing.T) {
isolateConfig(t)
writeConfig(t, "reload.toml", "status_reporter = true\n")
testutil.WriteFile(t, "reload.toml", "status_reporter = true\n")
m, err := Load([]string{"-c", "reload.toml"})
if err != nil {
t.Fatal(err)
@@ -126,24 +104,24 @@ func TestReloadReadsDiskValidatesAndPreservesOldSnapshots(t *testing.T) {
if err != nil {
t.Fatal(err)
}
writeConfig(t, "reload.toml", "status_reporter = false\n")
testutil.WriteFile(t, "reload.toml", "status_reporter = false\n")
next, err := m.Reload()
if err != nil || next.StatusReporter || !old.StatusReporter || next.ConfigFile != old.ConfigFile {
t.Fatalf("disk reload/snapshot ownership: %+v %v", next, err)
}
// Builder validators do not run on LoadFile; Snapshot must run them again.
writeConfig(t, "reload.toml", "pipelines = []\n")
testutil.WriteFile(t, "reload.toml", "pipelines = []\n")
if _, err := m.Reload(); err == nil || !strings.Contains(err.Error(), "no pipelines") {
t.Fatalf("semantically invalid reload accepted: %v", err)
}
if len(next.Pipelines) == 0 || !old.StatusReporter {
t.Fatal("invalid edit changed a published snapshot")
}
writeConfig(t, "reload.toml", "[logging.file]\nmax_size_mb = 1.5\n")
testutil.WriteFile(t, "reload.toml", "[logging.file]\nmax_size_mb = 1.5\n")
if _, err := m.Reload(); err == nil {
t.Fatal("fractional integer accepted")
}
writeConfig(t, "reload.toml", "# remove the status override\n")
testutil.WriteFile(t, "reload.toml", "# remove the status override\n")
recovered, err := m.Reload()
if err != nil || !recovered.StatusReporter {
t.Fatalf("override removal/recovery: %+v %v", recovered, err)
@@ -152,7 +130,7 @@ func TestReloadReadsDiskValidatesAndPreservesOldSnapshots(t *testing.T) {
func TestWatchStartsOnSubscriptionAndCloses(t *testing.T) {
isolateConfig(t)
writeConfig(t, "watched.toml", "auto_reload = true\n")
testutil.WriteFile(t, "watched.toml", "auto_reload = true\n")
m, err := Load([]string{"-c", "watched.toml"})
if err != nil {
t.Fatal(err)
@@ -178,7 +156,7 @@ func TestWatchStartsOnSubscriptionAndCloses(t *testing.T) {
func TestNestedPluginOptionsKeepDefaultsAndNumericTypes(t *testing.T) {
isolateConfig(t)
writeConfig(t, "plugins.toml", `
testutil.WriteFile(t, "plugins.toml", `
[[pipelines]]
name = "nested"
[[pipelines.plugin_sources]]
internal/config/test_helpers_test.go
+15
View File
@@ -0,0 +1,15 @@
package config
import (
"testing"
"logwisp/internal/testutil"
)
func isolateConfig(t *testing.T) {
t.Helper()
testutil.ClearEnvPrefix(t, "LOGWISP_")
dir := t.TempDir()
t.Setenv("HOME", dir)
t.Chdir(dir)
}
internal/sink/http/http.go
+5
View File
@@ -202,7 +202,12 @@ func (h *HTTPSink) Start(ctx context.Context) error {
if err != nil {
return fmt.Errorf("http sink bind %s: %w", h.addr, err)
}
return h.serve(ctx, ln)
}
// serve owns an already-bound listener, allowing tests to reserve an ephemeral
// port and exercise the same routing and worker lifecycle as Start.
func (h *HTTPSink) serve(ctx context.Context, ln net.Listener) error {
mux := http.NewServeMux()
// Method-scoped patterns: mux answers 405 with Allow header on non-GET
mux.HandleFunc(http.MethodGet+" "+h.config.StreamPath, h.handleStream)
internal/sink/http/http_test.go
+30 -76
View File
@@ -1,42 +1,22 @@
package http
import (
"context"
"bufio"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"logwisp/internal/session"
"logwisp/internal/sink"
"github.com/lixenwraith/log"
)
func TestStatusReportsQueueAndConnectionBounds(t *testing.T) {
manager := session.NewManager(time.Hour)
defer manager.Stop()
created, err := NewHTTPSinkPlugin(
"stream",
map[string]any{
"host": "127.0.0.1",
"port": int64(8081),
"buffer_size": int64(4096),
"client_buffer_size": int64(512),
"max_connections": int64(32),
"write_timeout_ms": int64(5000),
},
log.NewLogger(),
session.NewProxy(manager, "stream"),
)
if err != nil {
t.Fatal(err)
}
httpSink, ok := created.(*HTTPSink)
if !ok {
t.Fatalf("sink type = %T", created)
}
httpSink, _ := newTestHTTPSink(t, map[string]any{
"buffer_size": int64(4096),
"client_buffer_size": int64(512),
"max_connections": int64(32),
})
recorder := httptest.NewRecorder()
httpSink.handleStatus(recorder, httptest.NewRequest("GET", "/status", nil))
@@ -80,32 +60,14 @@ func TestStatusReportsQueueAndConnectionBounds(t *testing.T) {
// bumps activity otherwise, so a quiet source would idle-expire a healthy client
// and the broker would evict it on the next entry.
func TestQuietStreamRefreshesItsSession(t *testing.T) {
manager := session.NewManager(time.Hour)
defer manager.Stop()
created, err := NewHTTPSinkPlugin(
"stream",
map[string]any{"host": "127.0.0.1", "port": int64(18191), "write_timeout_ms": int64(5000)},
log.NewLogger(),
session.NewProxy(manager, "stream"),
)
if err != nil {
t.Fatal(err)
}
httpSink := created.(*HTTPSink)
httpSink, manager := newTestHTTPSink(t, nil)
httpSink.keepalive = 100 * time.Millisecond
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
if err := httpSink.Start(ctx); err != nil {
t.Fatal(err)
}
defer httpSink.Stop()
resp, err := http.Get("http://127.0.0.1:18191/stream")
client, baseURL := serveTestHTTPSink(t, httpSink)
resp, err := client.Get(baseURL + "/stream")
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
t.Cleanup(func() { _ = resp.Body.Close() })
activity := func() time.Time {
for _, s := range manager.GetActiveSessions() {
@@ -115,42 +77,34 @@ func TestQuietStreamRefreshesItsSession(t *testing.T) {
return time.Time{}
}
deadline := time.Now().Add(2 * time.Second)
for activity().IsZero() && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
before := activity()
// No events are sent for several keepalive periods.
time.Sleep(350 * time.Millisecond)
if before.IsZero() {
t.Fatal("stream session was not registered")
}
scanner := bufio.NewScanner(resp.Body)
heartbeats := 0
for heartbeats < 2 && scanner.Scan() {
if scanner.Text() == ":" {
heartbeats++
}
}
if err := scanner.Err(); err != nil {
t.Fatal(err)
}
if heartbeats != 2 {
t.Fatal("stream ended before two keepalives")
}
if after := activity(); !after.After(before) {
t.Fatalf("last activity %v did not advance on a silent stream", after)
t.Fatalf("last activity %v did not advance after keepalives from %v", after, before)
}
}
// HEAD on the stream path is refused rather than served from the GET pattern:
// its body writes are discarded, so the client it would register never reads.
func TestHeadOnStreamPathIsRefused(t *testing.T) {
manager := session.NewManager(time.Hour)
defer manager.Stop()
created, err := NewHTTPSinkPlugin(
"stream",
map[string]any{"host": "127.0.0.1", "port": int64(18192)},
log.NewLogger(),
session.NewProxy(manager, "stream"),
)
if err != nil {
t.Fatal(err)
}
httpSink := created.(*HTTPSink)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
if err := httpSink.Start(ctx); err != nil {
t.Fatal(err)
}
defer httpSink.Stop()
resp, err := http.Head("http://127.0.0.1:18192/stream")
httpSink, manager := newTestHTTPSink(t, nil)
client, baseURL := serveTestHTTPSink(t, httpSink)
resp, err := client.Head(baseURL + "/stream")
if err != nil {
t.Fatal(err)
}
internal/sink/http/test_helpers_test.go
+48
View File
@@ -0,0 +1,48 @@
package http
import (
"context"
"maps"
"net"
"net/http"
"testing"
"time"
"logwisp/internal/session"
"github.com/lixenwraith/log"
)
func newTestHTTPSink(t *testing.T, overrides map[string]any) (*HTTPSink, *session.Manager) {
t.Helper()
manager := session.NewManager(time.Hour)
t.Cleanup(manager.Stop)
opts := map[string]any{"host": "127.0.0.1", "port": int64(8081), "write_timeout_ms": int64(5000)}
maps.Copy(opts, overrides)
created, err := NewHTTPSinkPlugin("stream", opts, log.NewLogger(), session.NewProxy(manager, "stream"))
if err != nil {
t.Fatal(err)
}
h, ok := created.(*HTTPSink)
if !ok {
t.Fatalf("sink type = %T", created)
}
return h, manager
}
func serveTestHTTPSink(t *testing.T, h *HTTPSink) (*http.Client, string) {
t.Helper()
listener, err := net.Listen("tcp4", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = listener.Close() })
ctx, cancel := context.WithCancel(t.Context())
t.Cleanup(func() { cancel(); h.Stop() })
if err := h.serve(ctx, listener); err != nil {
t.Fatal(err)
}
client := &http.Client{Timeout: 3 * time.Second}
t.Cleanup(client.CloseIdleConnections)
return client, "http://" + listener.Addr().String()
}
internal/source/file/file_test.go
+3 -6
View File
@@ -9,13 +9,12 @@ import (
"time"
"logwisp/internal/core"
"logwisp/internal/testutil"
)
func TestStoppedWatcherReturnsNormally(t *testing.T) {
path := filepath.Join(t.TempDir(), "session.jsonl")
if err := os.WriteFile(path, nil, 0o600); err != nil {
t.Fatal(err)
}
testutil.WriteFile(t, path, "")
watcher := newFileWatcher(path, true, true, func(_ core.LogEntry) {}, nil)
watcher.stop()
@@ -52,9 +51,7 @@ func TestRemoveWatcherPreservesReplacement(t *testing.T) {
func TestRotatedFileResumesInsteadOfReplaying(t *testing.T) {
dir := t.TempDir()
active := filepath.Join(dir, "session.jsonl")
if err := os.WriteFile(active, []byte("one\ntwo\n"), 0o600); err != nil {
t.Fatal(err)
}
testutil.WriteFile(t, active, "one\ntwo\n")
info, err := os.Stat(active)
if err != nil {
t.Fatal(err)
internal/testutil/fixtures.go
+31
View File
@@ -0,0 +1,31 @@
// Package testutil provides standard-library fixtures shared by logwisp tests.
// Application packages must not import it.
package testutil
import (
"os"
"strings"
"testing"
)
// ClearEnvPrefix isolates overrides while preserving the caller's environment.
// Like testing.Setenv, it must not be used by parallel tests.
func ClearEnvPrefix(t testing.TB, prefix string) {
t.Helper()
for _, entry := range os.Environ() {
key, value, _ := strings.Cut(entry, "=")
if strings.HasPrefix(key, prefix) {
t.Setenv(key, value) // Register restoration before unsetting it.
if err := os.Unsetenv(key); err != nil {
t.Fatalf("unset fixture variable %s: %v", key, err)
}
}
}
}
func WriteFile(t testing.TB, path, contents string) {
t.Helper()
if err := os.WriteFile(path, []byte(contents), 0600); err != nil {
t.Fatalf("write fixture %q: %v", path, err)
}
}