From 38958d57c37f8290c3d47eb6a345f2a092093e9b Mon Sep 17 00:00:00 2001 From: Doug Steigerwald Date: Mon, 19 Oct 2026 10:20:16 -0400 Subject: [PATCH 1/4] fix: update go and tools Signed-off-by: Doug Steigerwald --- .github/workflows/ci.yaml | 6 +++--- go.mod | 8 ++++---- go.sum | 9 +++++++++ 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 27d0f87..daacec3 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -20,11 +20,11 @@ on: - main - release-* env: - GO_VERSION: 1.26.6 + GO_VERSION: 1.27 ARTIFACT_DIR: ./artifacts LINT_DIR: ./.github/linters COV_DIR: ./artifacts/cov - BETTERLEAKS_VERSION: 1.1.2 + BETTERLEAKS_VERSION: 1.9.0 jobs: lint-other-files: @@ -117,7 +117,7 @@ jobs: - name: golangci-lint uses: golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a with: - version: v2.11 + version: v2.15 args: --config=${{ env.LINT_DIR }}/golangci-config.yaml --allow-parallel-runners go-test: diff --git a/go.mod b/go.mod index 39d703f..24711cf 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,7 @@ require ( go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.46.0 go.opentelemetry.io/otel/sdk v1.46.0 go.opentelemetry.io/otel/trace v1.46.0 - golang.org/x/text v0.42.0 + golang.org/x/text v0.43.0 golang.org/x/time v0.16.0 google.golang.org/grpc v1.83.2 google.golang.org/protobuf v1.36.12 @@ -79,10 +79,10 @@ require ( go.opentelemetry.io/proto/otlp v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect - golang.org/x/net v0.58.0 // indirect + golang.org/x/net v0.61.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sys v0.47.0 // indirect - golang.org/x/term v0.45.0 // indirect + golang.org/x/sys v0.49.0 // indirect + golang.org/x/term v0.47.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260819154853-08b0e4226688 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688 // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect diff --git a/go.sum b/go.sum index 00b2943..d53540a 100644 --- a/go.sum +++ b/go.sum @@ -184,6 +184,8 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw= +golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -192,17 +194,24 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= +golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= +golang.org/x/term v0.47.0 h1:cpyOwQkr2Nzxe2p6KnrTG7Ei2wtzjsuSjhpXXepZ/BM= +golang.org/x/term v0.47.0/go.mod h1:efAfAuG6RKABvg1ArGebRkvX6LXul14SM67o2e4mUCY= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc= +golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s= golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI= golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo= +golang.org/x/tools v0.51.0 h1:k4Xc/1Om9jwkBJBo4NVLMSARBoWtK10mx+W5BnXCeAI= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260819154853-08b0e4226688 h1:ax2KzoSRIZU/M0cIxri3pKxy99vniH1PVxWC6si/eZI= From 25c4bf4e348451dab76772f2e23fe6564fa06b04 Mon Sep 17 00:00:00 2001 From: Dane Hunt Date: Tue, 27 Oct 2026 14:16:46 -0400 Subject: [PATCH 2/4] chore: lint err fixes Signed-off-by: Dane Hunt --- cmd/serve.go | 8 +- cmd/serve_test.go | 18 ++- i18n/message_loader_test.go | 24 +-- i18n/oslocale_test.go | 24 ++- i18n/oslocale_unix_test.go | 25 ++- internal/metrics/metrics.go | 25 +-- internal/metrics/metrics_test.go | 16 +- internal/metrics/prometheus/metrics_test.go | 60 +++++--- .../provider/connectors/amqp091/amqp091.go | 83 +++++----- .../connectors/amqp091/amqp091_test.go | 142 ++++++++++-------- .../connectors/amqp091/amqp091shim_test.go | 12 +- .../connectors/amqp091/compression.go | 4 +- .../connectors/amqp091/stream_test.go | 50 +++--- .../connectors/amqp091/streamhelpers_test.go | 52 +++---- .../provider/connectors/amqp091/streamshim.go | 6 +- .../server/prometheus/interceptors_test.go | 8 +- .../server/ratelimiter/ratelimiter_test.go | 20 +-- internal/server/server_test.go | 73 ++++----- internal/util/cgroups_test.go | 8 +- internal/util/concurrentMap_test.go | 10 +- internal/util/file_logger.go | 8 +- internal/util/k8s_test.go | 41 ++--- internal/util/tracing/tracing_test.go | 25 ++- internal/util/util_test.go | 19 +-- internal/util/zerolog_test.go | 25 +-- pkg/arke/arke_test.go | 29 ++-- test/config/env_test.go | 129 +++++++--------- test/healthz/healthz.go | 17 ++- test/util/util_test.go | 3 +- 29 files changed, 518 insertions(+), 446 deletions(-) diff --git a/cmd/serve.go b/cmd/serve.go index 1d0fb18..afef48b 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -36,7 +36,9 @@ func run(ctx context.Context) { if err != nil { util.Logger.Fatal(i18n.CPUProfileError, err) } - defer f.Close() + defer func() { + _ = f.Close() + }() if err := pprof.StartCPUProfile(f); err != nil { util.Logger.Fatal(i18n.CPUProfileError, err) } @@ -49,7 +51,9 @@ func run(ctx context.Context) { if err != nil { util.Logger.Fatal(i18n.MemProfileError, err) } - defer f.Close() + defer func() { + _ = f.Close() + }() runtime.GC() // get up-to-date statistics if err := pprof.WriteHeapProfile(f); err != nil { util.Logger.Fatal(i18n.MemProfileError, err) diff --git a/cmd/serve_test.go b/cmd/serve_test.go index 10db71c..56485a1 100644 --- a/cmd/serve_test.go +++ b/cmd/serve_test.go @@ -24,7 +24,9 @@ func testHealth(port int) error { if err != nil { log.Fatalf("did not connect: %v", err) } - defer conn.Close() + defer func() { + _ = conn.Close() + }() c := pb.NewHealthzClient(conn) ctx := context.Background() _, err = c.Check(ctx) @@ -39,8 +41,7 @@ func TestMonitorProcessStats(t *testing.T) { assert.Nil(t, err) cancel() }() - os.Setenv(arke.EnvPort, "50058") - defer os.Unsetenv(arke.EnvPort) + t.Setenv(arke.EnvPort, "50058") run(ctx) } @@ -49,11 +50,15 @@ func TestRunWithCPUAndMemProfile(t *testing.T) { cpuFile, err := os.CreateTemp(tmpDir, "cpuprofile-*.prof") assert.Nil(t, err) - cpuFile.Close() + defer func() { + _ = cpuFile.Close() + }() memFile, err := os.CreateTemp(tmpDir, "memprofile-*.prof") assert.Nil(t, err) - memFile.Close() + defer func() { + _ = memFile.Close() + }() *cpuprofile = cpuFile.Name() *memprofile = memFile.Name() @@ -69,8 +74,7 @@ func TestRunWithCPUAndMemProfile(t *testing.T) { assert.Nil(t, err) cancel() }() - os.Setenv(arke.EnvPort, "50061") - defer os.Unsetenv(arke.EnvPort) + t.Setenv(arke.EnvPort, "50061") run(ctx) diff --git a/i18n/message_loader_test.go b/i18n/message_loader_test.go index de940ed..3695355 100644 --- a/i18n/message_loader_test.go +++ b/i18n/message_loader_test.go @@ -17,14 +17,20 @@ type getLocaleFileTestCase struct { expectedFiles []string } -const testValue2 = "value2" +const ( + localeEnUS = "en-US" + testValue2 = "value2" + localeDeDE = "de-DE" + multipleParamsID = "multiple_params" +) func TestGetLocaleFiles(t *testing.T) { filePrefix := messageFilePrefix + cases := []getLocaleFileTestCase{ { filePrefix: filePrefix, - locale: "en-US", + locale: localeEnUS, expectedFiles: []string{ fmt.Sprintf("resources/%s_en_US.properties", filePrefix), fmt.Sprintf("resources/%s_en.properties", filePrefix), @@ -73,7 +79,7 @@ func TestGetLocaleFiles(t *testing.T) { }, { filePrefix: filePrefix, - locale: "de-DE", + locale: localeDeDE, expectedFiles: []string{ fmt.Sprintf("resources/%s_de_DE.properties", filePrefix), fmt.Sprintf("resources/%s_de.properties", filePrefix), @@ -238,10 +244,10 @@ func TestT(t *testing.T) { }() bundle = &propertiesBundle{ messages: map[string]string{ - "no_params": "This is a test message.", - "one_param": "Hello, {0}!", - "repeated_param": "Value: {0}, Again: {0}", - "multiple_params": "First: {0}, Second: {1}, Third: {2}, First again: {0}", + "no_params": "This is a test message.", + "one_param": "Hello, {0}!", + "repeated_param": "Value: {0}, Again: {0}", + multipleParamsID: "First: {0}, Second: {1}, Third: {2}, First again: {0}", }, } tests := []struct { @@ -270,7 +276,7 @@ func TestT(t *testing.T) { }, { name: "multiple parameters", - messageID: "multiple_params", + messageID: multipleParamsID, args: []interface{}{"A", "B", "C"}, want: "First: A, Second: B, Third: C, First again: A", }, @@ -282,7 +288,7 @@ func TestT(t *testing.T) { }, { name: "extra arguments", - messageID: "multiple_params", + messageID: multipleParamsID, args: []interface{}{"A", "B", "C", "D"}, want: "First: A, Second: B, Third: C, First again: A", }, diff --git a/i18n/oslocale_test.go b/i18n/oslocale_test.go index f127dff..07c73df 100644 --- a/i18n/oslocale_test.go +++ b/i18n/oslocale_test.go @@ -29,43 +29,37 @@ func TestArgsInit(t *testing.T) { }, { name: "cli args override", - Lang: "en-US", + Lang: localeEnUS, LcAll: "", args: []string{"a", "--locale", "zh"}, expected: "zh", }, { name: EnvLang, - Lang: "en-US", + Lang: localeEnUS, LcAll: "", args: []string{"a"}, - expected: "en-US", + expected: localeEnUS, }, { name: EnvLcAll, Lang: "", - LcAll: "de-DE", + LcAll: localeDeDE, args: []string{"a"}, - expected: "de-DE", + expected: localeDeDE, }, { name: EnvLcAll + " priority", Lang: "zh-Hans", - LcAll: "de-DE", + LcAll: localeDeDE, args: []string{"a"}, - expected: "de-DE", + expected: localeDeDE, }, } for _, c := range cases { - oldLang := os.Getenv(EnvLang) - oldLCAll := os.Getenv(EnvLcAll) - defer func() { - os.Setenv(EnvLang, oldLang) - os.Setenv(EnvLcAll, oldLCAll) - }() t.Run(c.name, func(t *testing.T) { - os.Setenv(EnvLang, c.Lang) - os.Setenv(EnvLcAll, c.LcAll) + t.Setenv(EnvLang, c.Lang) + t.Setenv(EnvLcAll, c.LcAll) os.Args = c.args l10Messages = nil SystemLocale = "" diff --git a/i18n/oslocale_unix_test.go b/i18n/oslocale_unix_test.go index 30c2fec..e22b0cb 100644 --- a/i18n/oslocale_unix_test.go +++ b/i18n/oslocale_unix_test.go @@ -7,34 +7,33 @@ package i18n import ( - "os" "testing" "github.com/stretchr/testify/assert" ) func TestDetectIETF_LC_ALL(t *testing.T) { - os.Setenv(EnvLcAll, "fr_FR.UTF-8") + t.Setenv(EnvLcAll, "fr_FR.UTF-8") result, _ := DetectIETF() assert.Equal(t, result, "fr-FR") } func TestDetectIETF_LANG(t *testing.T) { - os.Setenv(EnvLang, "fr_FR.UTF-8") + t.Setenv(EnvLang, "fr_FR.UTF-8") result, _ := DetectIETF() assert.Equal(t, result, "fr-FR") } func TestDetectIETF_Blank(t *testing.T) { - os.Setenv(EnvLcAll, "") - os.Setenv(EnvLang, "") + t.Setenv(EnvLcAll, "") + t.Setenv(EnvLang, "") result, _ := DetectIETF() assert.Equal(t, result, "") } func TestStandardInit(t *testing.T) { - os.Unsetenv(EnvLang) - os.Unsetenv(EnvLcAll) + t.Setenv(EnvLang, "") + t.Setenv(EnvLcAll, "") locale, _ := DetectIETF() if locale == "" { locale = defaultLanguage @@ -43,23 +42,21 @@ func TestStandardInit(t *testing.T) { } func TestLangC(t *testing.T) { - os.Unsetenv(EnvLcAll) - os.Setenv(EnvLang, "C") + t.Setenv(EnvLcAll, "") + t.Setenv(EnvLang, "C") locale, err := DetectIETF() assert.NotNil(t, err) assert.Equal(t, "", locale) - os.Setenv(EnvLang, "C.UTF-8") - defer os.Unsetenv(EnvLang) + t.Setenv(EnvLang, "C.UTF-8") locale, err = DetectIETF() assert.NotNil(t, err) assert.Equal(t, "", locale) } func TestInvalidLang(t *testing.T) { - os.Unsetenv(EnvLcAll) - os.Setenv(EnvLang, "invalid") - defer os.Unsetenv(EnvLang) + t.Setenv(EnvLcAll, "") + t.Setenv(EnvLang, "invalid") locale, err := DetectIETF() assert.NotNil(t, err) assert.Equal(t, "", locale) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 4600e74..2efb16c 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -12,17 +12,24 @@ type LabelSet struct { Labels []met.Label } +// Metric name segments reused across multiple metric definitions +const ( + metricNamespace = "arke" + metricClient = "client" + metricTotal = "total" +) + // Stats global Stats variable for access to the sinks var ( - ClientActMessageGauge = []string{"arke", "client", "active", "messages"} - ClientStreamsGauge = []string{"arke", "client", "streams"} - ClientConsumedGauge = []string{"arke", "client", "consumed"} - ClientProducedGauge = []string{"arke", "client", "produced"} - RequestElapsedSummary = []string{"arke", "request", "elapsed"} - RequestTotalCounter = []string{"arke", "request", "total"} - RecvMsgCounter = []string{"arke", "recvmsg", "total"} - SendMsgCounter = []string{"arke", "sendmsg", "total"} - RateLimitEnforcedSummary = []string{"arke", "rate", "limit", "enforced", "summary"} + ClientActMessageGauge = []string{metricNamespace, metricClient, "active", "messages"} + ClientStreamsGauge = []string{metricNamespace, metricClient, "streams"} + ClientConsumedGauge = []string{metricNamespace, metricClient, "consumed"} + ClientProducedGauge = []string{metricNamespace, metricClient, "produced"} + RequestElapsedSummary = []string{metricNamespace, "request", "elapsed"} + RequestTotalCounter = []string{metricNamespace, "request", metricTotal} + RecvMsgCounter = []string{metricNamespace, "recvmsg", metricTotal} + SendMsgCounter = []string{metricNamespace, "sendmsg", metricTotal} + RateLimitEnforcedSummary = []string{metricNamespace, "rate", "limit", "enforced", "summary"} ) // NewLabelSet Create a new label set diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 4b6213b..8dfa831 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -10,14 +10,14 @@ import ( ) func Test_Constants(t *testing.T) { - assert.Equal(t, ClientActMessageGauge, []string{"arke", "client", "active", "messages"}) - assert.Equal(t, ClientStreamsGauge, []string{"arke", "client", "streams"}) - assert.Equal(t, ClientConsumedGauge, []string{"arke", "client", "consumed"}) - assert.Equal(t, ClientProducedGauge, []string{"arke", "client", "produced"}) - assert.Equal(t, RequestElapsedSummary, []string{"arke", "request", "elapsed"}) - assert.Equal(t, RequestTotalCounter, []string{"arke", "request", "total"}) - assert.Equal(t, RecvMsgCounter, []string{"arke", "recvmsg", "total"}) - assert.Equal(t, SendMsgCounter, []string{"arke", "sendmsg", "total"}) + assert.Equal(t, ClientActMessageGauge, []string{metricNamespace, metricClient, "active", "messages"}) + assert.Equal(t, ClientStreamsGauge, []string{metricNamespace, metricClient, "streams"}) + assert.Equal(t, ClientConsumedGauge, []string{metricNamespace, metricClient, "consumed"}) + assert.Equal(t, ClientProducedGauge, []string{metricNamespace, metricClient, "produced"}) + assert.Equal(t, RequestElapsedSummary, []string{metricNamespace, "request", "elapsed"}) + assert.Equal(t, RequestTotalCounter, []string{metricNamespace, "request", metricTotal}) + assert.Equal(t, RecvMsgCounter, []string{metricNamespace, "recvmsg", metricTotal}) + assert.Equal(t, SendMsgCounter, []string{metricNamespace, "sendmsg", metricTotal}) } func Test_LabelSet(t *testing.T) { diff --git a/internal/metrics/prometheus/metrics_test.go b/internal/metrics/prometheus/metrics_test.go index d43879e..6b6d1c4 100644 --- a/internal/metrics/prometheus/metrics_test.go +++ b/internal/metrics/prometheus/metrics_test.go @@ -8,7 +8,6 @@ import ( "io" "net" "net/http" - "os" "strings" "testing" @@ -18,6 +17,13 @@ import ( "github.com/stretchr/testify/assert" ) +const ( + testMetricNamespace = "arke" + testMetricClient = "client" + testMetricTotal = "total" + testGaugeTwoName = "arke_gauge_two" +) + func Test_Metrics(t *testing.T) { var tests = []struct { name string @@ -28,23 +34,23 @@ func Test_Metrics(t *testing.T) { }{ { name: "sperate key - match", - metricName: []string{"arke", "gauge", "one"}, + metricName: []string{testMetricNamespace, "gauge", "one"}, metricDesc: "description for gauge1", key: "arke_gauge_one", expected: true, }, { name: "one key - match", - metricName: []string{"arke_gauge_two"}, + metricName: []string{testGaugeTwoName}, metricDesc: "description for gauge2", - key: "arke_gauge_two", + key: testGaugeTwoName, expected: true, }, { name: "key - not match", - metricName: []string{"arke", "gauge", "three"}, + metricName: []string{testMetricNamespace, "gauge", "three"}, metricDesc: "description for gauge3", - key: "arke_gauge_two", + key: testGaugeTwoName, expected: false, }, } @@ -79,14 +85,14 @@ func Test_Metrics(t *testing.T) { }) } - assert.Equal(t, metrics.ClientActMessageGauge, []string{"arke", "client", "active", "messages"}) - assert.Equal(t, metrics.ClientStreamsGauge, []string{"arke", "client", "streams"}) - assert.Equal(t, metrics.ClientConsumedGauge, []string{"arke", "client", "consumed"}) - assert.Equal(t, metrics.ClientProducedGauge, []string{"arke", "client", "produced"}) - assert.Equal(t, metrics.RequestElapsedSummary, []string{"arke", "request", "elapsed"}) - assert.Equal(t, metrics.RequestTotalCounter, []string{"arke", "request", "total"}) - assert.Equal(t, metrics.RecvMsgCounter, []string{"arke", "recvmsg", "total"}) - assert.Equal(t, metrics.SendMsgCounter, []string{"arke", "sendmsg", "total"}) + assert.Equal(t, metrics.ClientActMessageGauge, []string{testMetricNamespace, testMetricClient, "active", "messages"}) + assert.Equal(t, metrics.ClientStreamsGauge, []string{testMetricNamespace, testMetricClient, "streams"}) + assert.Equal(t, metrics.ClientConsumedGauge, []string{testMetricNamespace, testMetricClient, "consumed"}) + assert.Equal(t, metrics.ClientProducedGauge, []string{testMetricNamespace, testMetricClient, "produced"}) + assert.Equal(t, metrics.RequestElapsedSummary, []string{testMetricNamespace, "request", "elapsed"}) + assert.Equal(t, metrics.RequestTotalCounter, []string{testMetricNamespace, "request", testMetricTotal}) + assert.Equal(t, metrics.RecvMsgCounter, []string{testMetricNamespace, "recvmsg", testMetricTotal}) + assert.Equal(t, metrics.SendMsgCounter, []string{testMetricNamespace, "sendmsg", testMetricTotal}) } func Test_gatherClientStats(t *testing.T) { @@ -101,7 +107,9 @@ func Test_Serve(t *testing.T) { lis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", ":50052") assert.Nil(t, err) - defer lis.Close() + defer func() { + _ = lis.Close() + }() go Serve(ctx, &lis) req, err := http.NewRequestWithContext(ctx, "GET", "http://localhost:50052/metrics", nil) @@ -111,7 +119,9 @@ func Test_Serve(t *testing.T) { res, err := client.Do(req) assert.Nil(t, err) - defer res.Body.Close() + defer func() { + _ = res.Body.Close() + }() assert.Equal(t, 200, res.StatusCode) body, _ := io.ReadAll(res.Body) sbody := string(body) @@ -127,21 +137,25 @@ func Test_Serve(t *testing.T) { res, err = client.Do(req) assert.Nil(t, err) - defer res.Body.Close() + defer func() { + _ = res.Body.Close() + }() assert.Equal(t, 404, res.StatusCode) body, _ = io.ReadAll(res.Body) assert.Contains(t, string(body), "404 page not found") } func Test_ServePprofEnabled(t *testing.T) { - os.Setenv(EnvPprofEnabled, "true") + t.Setenv(EnvPprofEnabled, "true") ctx, cancel := context.WithCancel(context.Background()) defer cancel() lis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", ":50053") assert.Nil(t, err) - defer lis.Close() + defer func() { + _ = lis.Close() + }() go Serve(ctx, &lis) req, err := http.NewRequestWithContext(ctx, "GET", "http://localhost:50053/metrics", nil) @@ -151,7 +165,9 @@ func Test_ServePprofEnabled(t *testing.T) { res, err := client.Do(req) assert.Nil(t, err) - defer res.Body.Close() + defer func() { + _ = res.Body.Close() + }() assert.Equal(t, 200, res.StatusCode) body, _ := io.ReadAll(res.Body) sbody := string(body) @@ -166,7 +182,9 @@ func Test_ServePprofEnabled(t *testing.T) { res, err = client.Do(req) assert.Nil(t, err) - defer res.Body.Close() + defer func() { + _ = res.Body.Close() + }() assert.Equal(t, 200, res.StatusCode) body, _ = io.ReadAll(res.Body) assert.Contains(t, string(body), "full goroutine stack dump") diff --git a/internal/provider/connectors/amqp091/amqp091.go b/internal/provider/connectors/amqp091/amqp091.go index 71b1cff..b5cc2d4 100644 --- a/internal/provider/connectors/amqp091/amqp091.go +++ b/internal/provider/connectors/amqp091/amqp091.go @@ -32,25 +32,34 @@ import ( "go.opentelemetry.io/otel/trace" ) -const providerName string = "amqp091" -const trustedCerts = "ARKE_TRUSTED_CA_CERTIFICATES_PEM_FILE" -const streamOffsetHeaderName = "x-current-offset" -const retryCountHeaderName = "x-retry-count" -const rabbitReceivedTimeHeaderName = "x-opt-rabbitmq-received-time" -const timeStampInMSHeaderName = "timestamp_in_ms" -const transferEncodingHeaderName = "Transfer-Encoding" - -const maxPubChannels = 10 -const maxPubPCChannels = 10 - -const queueTypeClassic = "classic" -const queueTypeQuorum = "quorum" -const xMatchAny = "any" +const ( + providerName = "amqp091" + trustedCerts = "ARKE_TRUSTED_CA_CERTIFICATES_PEM_FILE" + streamOffsetHeaderName = "x-current-offset" + retryCountHeaderName = "x-retry-count" + rabbitReceivedTimeHeaderName = "x-opt-rabbitmq-received-time" + timeStampInMSHeaderName = "timestamp_in_ms" + transferEncodingHeaderName = "Transfer-Encoding" + errMsgConnectionClosed = "connection to broker is closed" + offsetOption = "Offset" + messageTTLOption = "MessageTTL" + deadLetterAddressOption = "DeadLetterAddress" + deadLetterSubjectOption = "DeadLetterSubject" + expiresOption = "Expires" + consumerGroupOption = "ConsumerGroup" + + maxPubChannels = 10 + maxPubPCChannels = 10 + + queueTypeClassic = "classic" + queueTypeQuorum = "quorum" + xMatchAny = "any" +) -var supportedSourceOptionsList = []string{"MessageTTL", "DeadLetterAddress", "DeadLetterSubject", "Expires", "Offset", "ConsumerGroup"} +var supportedSourceOptionsList = []string{messageTTLOption, deadLetterAddressOption, deadLetterSubjectOption, expiresOption, offsetOption, consumerGroupOption} var supportedSourceOptions map[string]bool -var supportedStreamSourceOptions = map[string]bool{"Offset": true, "MessageTTL": true, "ConsumerGroup": true} +var supportedStreamSourceOptions = map[string]bool{offsetOption: true, messageTTLOption: true, consumerGroupOption: true} // NewAmqpConn091 allow overriding the connection for mocking in tests var NewAmqpConn091 = NewAmqp091Connection @@ -149,7 +158,7 @@ func NewAMQP091Provider() provider.Provider { prov.tlsConfig = &tls.Config{} if caBundlePath != "" { - caBundle, err := os.ReadFile(filepath.FromSlash(filepath.Clean("/" + strings.Trim(caBundlePath, "/")))) + caBundle, err := os.ReadFile(filepath.FromSlash(filepath.Clean("/" + strings.Trim(caBundlePath, "/")))) //nolint:gosec if err == nil { prov.tlsConfig.RootCAs = x509.NewCertPool() prov.tlsConfig.RootCAs.AppendCertsFromPEM(caBundle) @@ -471,7 +480,7 @@ func (prov *amqp091provider) Connect(ctx context.Context, cf *pb.ConnectionConfi activeMessages := util.NewConcurrentMap() pubChCtx := context.WithValue(context.Background(), CtxKey{name: "clientIdentifier"}, clientIdentifier) - pubChCtx, pubChCancel := context.WithCancel(pubChCtx) //nolint:gosec + pubChCtx, pubChCancel := context.WithCancel(pubChCtx) // Allocate BrokerDetails as a pointer so closures and the connections map // all share the same intentional heap object. Previously, declaring bd as // a value caused the compiler to detect 5 separate escape flows and emit 5 @@ -848,7 +857,9 @@ func (bd *BrokerDetails) doManagementRequestWithoutMarshal(method, urn string) ( var body []byte var bodyErr error if resp != nil { - defer resp.Body.Close() + defer func() { + _ = resp.Body.Close() + }() body, bodyErr = io.ReadAll(resp.Body) } @@ -1032,7 +1043,7 @@ func (prov *amqp091provider) Subscribe(ctx context.Context, source *pb.Source, m func (prov *amqp091provider) queueSubscribe(ctx context.Context, bd *BrokerDetails, source *pb.Source, messageChannel chan<- *pb.Message) *pb.Error { //nolint:gocognit if bd.Connection.IsClosed() { - return &pb.Error{Message: "connection to broker is closed"} + return &pb.Error{Message: errMsgConnectionClosed} } // AutoDelete queues are temporary, override what @@ -1054,7 +1065,9 @@ func (prov *amqp091provider) queueSubscribe(ctx context.Context, bd *BrokerDetai if err != nil { return &pb.Error{Message: err.Error()} } - defer amqpChannel.Close() + defer func() { + _ = amqpChannel.Close() + }() _ = prov.declareExchange(source.GetAddress(), bd) @@ -1261,16 +1274,16 @@ func (prov *amqp091provider) streamSubscribe(ctx context.Context, bd *BrokerDeta } if bd.StreamConnection.IsClosed() { - return &pb.Error{Message: "connection to broker is closed"} + return &pb.Error{Message: errMsgConnectionClosed} } offset := "" opts := source.GetOptions() - if _, ok := opts["Offset"]; ok { - offset = opts["Offset"] + if _, ok := opts[offsetOption]; ok { + offset = opts[offsetOption] } var ttl int64 - if sTTL, ok := opts["MessageTTL"]; ok { + if sTTL, ok := opts[messageTTLOption]; ok { val, err := strconv.ParseInt(sTTL, 10, 64) if err != nil { return &pb.Error{Message: "value for MessageTTL option must be a quoted integer"} @@ -1370,7 +1383,7 @@ func (prov *amqp091provider) streamSubscribe(ctx context.Context, bd *BrokerDeta bd.incrementStreamCount() defer bd.decrementStreamCount() <-ctx.Done() - consumer.Close() + _ = consumer.Close() return nil } @@ -1410,11 +1423,11 @@ func (prov *amqp091provider) disconnectClientByIdentifier(clientIdentifier strin bd.shutdownChan <- true // shut down the connectionWatcher // close the client if it is still connected if bd.Connection != nil && !bd.Connection.IsClosed() { - bd.Connection.Close() + _ = bd.Connection.Close() } if bd.StreamConnection != nil { - bd.StreamConnection.Close() + _ = bd.StreamConnection.Close() } prov.connections.Delete(clientIdentifier) @@ -1433,14 +1446,16 @@ func (prov *amqp091provider) Publish(ctx context.Context, messageChannel <-chan defer bd.decrementStreamCount() if bd.Connection.IsClosed() { - return &pb.Error{Message: "connection to broker is closed"} + return &pb.Error{Message: errMsgConnectionClosed} } amqpChannel, err := bd.Connection.NewChannel(false) if err != nil { return &pb.Error{Message: err.Error()} } - defer amqpChannel.Close() + defer func() { + _ = amqpChannel.Close() + }() connErrChan := make(chan amqp091Error, 1) connErrChan = bd.Connection.NotifyClose(connErrChan) @@ -1529,7 +1544,7 @@ func (prov *amqp091provider) PublishOne(ctx context.Context, msg *pb.Message) *p func (prov *amqp091provider) publishOneQueue(ctx context.Context, msg *pb.Message, bd *BrokerDetails) *pb.Error { if bd.Connection.IsClosed() { - return &pb.Error{Message: "connection to broker is closed"} + return &pb.Error{Message: errMsgConnectionClosed} } if msg.GetPublishId() > 0 { return &pb.Error{Message: "Message deduplication is not supported by Queue types"} @@ -1549,14 +1564,14 @@ func (prov *amqp091provider) publishOneQueue(ctx context.Context, msg *pb.Messag defer func() { if err := bd.pubPCChannels.Put(amqpChannel); err != nil { util.Logger.Debugf("Failed to return channel to pool: %s", err.Error()) - (*amqpChannel).Close() + _ = (*amqpChannel).Close() } }() } else { defer func() { if err := bd.pubChannels.Put(amqpChannel); err != nil { util.Logger.Debugf("Failed to return channel to pool: %s", err.Error()) - (*amqpChannel).Close() + _ = (*amqpChannel).Close() } }() } @@ -1571,7 +1586,7 @@ func (prov *amqp091provider) publishOneStream(ctx context.Context, msg *pb.Messa } if bd.StreamConnection.IsClosed() { - return &pb.Error{Message: "connection to broker is closed"} + return &pb.Error{Message: errMsgConnectionClosed} } if msg.GetPublishId() > 0 && msg.GetPublisherName() == "" { diff --git a/internal/provider/connectors/amqp091/amqp091_test.go b/internal/provider/connectors/amqp091/amqp091_test.go index b9e8451..f07e36e 100644 --- a/internal/provider/connectors/amqp091/amqp091_test.go +++ b/internal/provider/connectors/amqp091/amqp091_test.go @@ -25,16 +25,37 @@ import ( "github.com/stretchr/testify/mock" ) -var ctx context.Context -var cf *pb.ConnectionConfiguration +var ( + ctx context.Context + cf *pb.ConnectionConfiguration +) -const testTenant = "tenant" -const testQueueTypeClassic = "classic" -const testQueueTypeQuorum = "quorum" -const testDeadLetterAddress = "dla" -const testContentTypeJSON = "application/json" -const testContentEncodingText = "text" -const testXMatchAny = "any" +const ( + testTenant = "tenant" + testQueueTypeClassic = "classic" + testQueueTypeQuorum = "quorum" + testDeadLetterAddress = "dla" + testContentTypeJSON = "application/json" + testContentEncodingText = "text" + testXMatchAny = "any" + testAddress = "address" + testMyName = "myname" + testAddressName = "addressname" + testValue1 = "value1" + testValue2 = "value2" + testValue3 = "value3" + testSrcName = "srcname" + testXAcquiredCount = "x-acquired-count" + testHeader1 = "header1" + testHeader2 = "header2" + testRoutingKey = "routingkey" + testPassword = "password" + testMessageTTLOption = "MessageTTL" + testQueue = "queue" + testUser = "user" + testPass = "pass" + testLocalhost = "localhost" +) // stopWatcher disconnects the client identified by ctx and blocks until its // connectionWatcher goroutine has fully exited. Tests that call prov.Connect @@ -483,7 +504,7 @@ func Test_Ack_AckErr(t *testing.T) { subjects := make([]string, 0) subjects = append(subjects, "#") - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address", Subjects: subjects}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress, Subjects: subjects}} mc := make(chan *pb.Message) defer close(mc) @@ -653,7 +674,7 @@ func Test_Ack(t *testing.T) { subjects := make([]string, 0) subjects = append(subjects, "#") - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address", Subjects: subjects}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress, Subjects: subjects}} mc := make(chan *pb.Message) defer close(mc) @@ -745,7 +766,7 @@ func Test_Nack(t *testing.T) { subjects := make([]string, 0) subjects = append(subjects, "#") - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address", Subjects: subjects}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress, Subjects: subjects}} mc := make(chan *pb.Message) defer close(mc) go func() { @@ -835,7 +856,7 @@ func Test_Nack_NackErr(t *testing.T) { subjects := make([]string, 0) subjects = append(subjects, "#") - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address", Subjects: subjects}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress, Subjects: subjects}} mc := make(chan *pb.Message) defer close(mc) go func() { @@ -929,7 +950,7 @@ func Test_Retry(t *testing.T) { defer stopWatcher(ctx, prov) assert.Nil(t, err) - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address"}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress}} mc := make(chan *pb.Message) defer close(mc) go func() { @@ -1034,7 +1055,7 @@ func Test_RetryAfterReconnectUsesNewConnection(t *testing.T) { NewAmqpConn091 = oldNewAmqpConn091 }() - source := &pb.Source{Address: &pb.Address{Name: "address"}} + source := &pb.Source{Address: &pb.Address{Name: testAddress}} assert.Nil(t, prov.Retry(context.Background(), source, "old-message", 1)) bd.state.Store(provider.DISCONNECTED) @@ -1136,7 +1157,7 @@ func Test_RetryFailure(t *testing.T) { defer stopWatcher(ctx, prov) assert.Nil(t, err) - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address"}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress}} mc := make(chan *pb.Message) defer close(mc) go func() { @@ -1243,7 +1264,7 @@ func Test_RetryFailure_DeclareErrorsStillSuccess(t *testing.T) { opts := make(map[string]string) opts["junkheader"] = "junkvalue" - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address"}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress}} mc := make(chan *pb.Message) defer close(mc) go func() { @@ -1395,11 +1416,11 @@ func Test_DLQ(t *testing.T) { amock.On("NotifyClose").Return(errs) amock.On("NewChannel", false).Return(cmock, nil) - sbmock.On("ExchangeDeclare", "address", "topic", false).Return(nil).Once() + sbmock.On("ExchangeDeclare", testAddress, "topic", false).Return(nil).Once() sbmock.On("ExchangeDeclare", "dla", "topic", false).Return(nil).Once() sbmock.On("QueueDeclare", "queue.quorum", false, false, args).Return(nil).Once() sbmock.On("QueueDeclare", "queue.dlq", false, false, argsDlq).Return(nil).Once() - sbmock.On("QueueBind", "queue.quorum", "routingkey", "address", mock.Anything).Return(nil).Once() + sbmock.On("QueueBind", "queue.quorum", testRoutingKey, testAddress, mock.Anything).Return(nil).Once() sbmock.On("QueueBind", "queue.dlq", "queue.quorum", "dla", mock.Anything).Return(nil).Once() amock.On("StandbyChannel").Return(sbmock, nil) @@ -1430,9 +1451,9 @@ func Test_DLQ(t *testing.T) { assert.Nil(t, err) subjects := make([]string, 0) - subjects = append(subjects, "routingkey") + subjects = append(subjects, testRoutingKey) options := map[string]string{"DeadLetterAddress": testDeadLetterAddress} - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address", Subjects: subjects}, + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress, Subjects: subjects}, Options: options} mc := make(chan *pb.Message) defer close(mc) @@ -1507,23 +1528,23 @@ func Test_Subscribe_NoAddressName(t *testing.T) { func Test_SourceNameNotQuorum(t *testing.T) { address := &pb.Address{} - src := &pb.Source{Name: "myname", AutoDelete: true, Address: address} + src := &pb.Source{Name: testMyName, AutoDelete: true, Address: address} src.Name = sourceName(src) - assert.Equal(t, "myname", src.GetName()) + assert.Equal(t, testMyName, src.GetName()) } func Test_SourceNameQuorum(t *testing.T) { address := &pb.Address{} - src := &pb.Source{Name: "myname", Address: address} + src := &pb.Source{Name: testMyName, Address: address} src.Name = sourceName(src) assert.Equal(t, "myname.quorum", src.GetName()) } func Test_SourceNameNotQuorum_AutoDelete(t *testing.T) { address := &pb.Address{} - src := &pb.Source{Name: "myname", Address: address, AutoDelete: true} + src := &pb.Source{Name: testMyName, Address: address, AutoDelete: true} src.Name = sourceName(src) - assert.Equal(t, "myname", src.GetName()) + assert.Equal(t, testMyName, src.GetName()) } func Test_SourceNameNamedDotQuorum(t *testing.T) { @@ -1548,7 +1569,7 @@ func Test_Subscribe_Options(t *testing.T) { prov := NewAMQP091Provider() options := make(map[string]string) - options["MessageTTL"] = "100" + options[testMessageTTLOption] = "100" options["Expires"] = "100" options["DeadLetterAddress"] = "dla" options["DeadLetterSubject"] = "dls" @@ -1587,18 +1608,18 @@ func Test_Subscribe_Options(t *testing.T) { subjects = append(subjects, "subject1") subjects = append(subjects, "subject2") parent := &pb.Address{Name: "parent", Type: pb.Address_QUEUE} - address := &pb.Address{Name: "addressname", Subjects: subjects, ParentAddress: parent, Type: pb.Address_FILTER} + address := &pb.Address{Name: testAddressName, Subjects: subjects, ParentAddress: parent, Type: pb.Address_FILTER} matches1 := make([]*pb.Match, 0) - matches1 = append(matches1, &pb.Match{Name: "key1", Value: "value1"}) - matches1 = append(matches1, &pb.Match{Name: "key2", Value: "value2"}) + matches1 = append(matches1, &pb.Match{Name: "key1", Value: testValue1}) + matches1 = append(matches1, &pb.Match{Name: "key2", Value: testValue2}) matches2 := make([]*pb.Match, 0) - matches2 = append(matches2, &pb.Match{Name: "key3", Value: "value3"}) + matches2 = append(matches2, &pb.Match{Name: "key3", Value: testValue3}) matches2 = append(matches2, &pb.Match{Name: "key4", Value: "value4"}) filters := make([]*pb.Filter, 0) filters = append(filters, &pb.Filter{Matches: matches1, Type: pb.Filter_ANY}) filters = append(filters, &pb.Filter{Matches: matches2, Type: pb.Filter_ANY}) - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, Options: options, Filters: filters, @@ -1607,12 +1628,12 @@ func Test_Subscribe_Options(t *testing.T) { PrefetchCount: 4} expectedMatchHeaders1 := amqp091Table{} - expectedMatchHeaders1["key1"] = "value1" - expectedMatchHeaders1["key2"] = "value2" + expectedMatchHeaders1["key1"] = testValue1 + expectedMatchHeaders1["key2"] = testValue2 expectedMatchHeaders1["x-match"] = testXMatchAny expectedMatchHeaders2 := amqp091Table{} - expectedMatchHeaders2["key3"] = "value3" + expectedMatchHeaders2["key3"] = testValue3 expectedMatchHeaders2["key4"] = "value4" expectedMatchHeaders2["x-match"] = testXMatchAny @@ -1692,7 +1713,6 @@ func Test_Subscribe_Options(t *testing.T) { delMock.AssertExpectations(t) cmock.AssertExpectations(t) - sbmock.AssertNumberOfCalls(t, "ExchangeDeclare", 3) sbmock.AssertExpectations(t) } @@ -1723,10 +1743,10 @@ func Test_Subscribe_NoSubjectsNoFilters(t *testing.T) { }(&delMock) subjects := make([]string, 0) - address := &pb.Address{Name: "addressname", Subjects: subjects, Type: pb.Address_FILTER} + address := &pb.Address{Name: testAddressName, Subjects: subjects, Type: pb.Address_FILTER} filters := make([]*pb.Filter, 0) - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, Filters: filters, Exclusive: true, @@ -1770,6 +1790,8 @@ func Test_Subscribe_NoSubjectsNoFilters(t *testing.T) { assert.Nil(t, serr) ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cc := &pb.ConnectionConfiguration{} cc.Tenant = testTenant cc.Host = u.Hostname() @@ -1857,7 +1879,7 @@ func Test_Subscribe_UnsupportedOptions(t *testing.T) { defer stopWatcher(ctx, prov) assert.Nil(t, err) - src := &pb.Source{Address: &pb.Address{Name: "addressname"}, Options: options} + src := &pb.Source{Address: &pb.Address{Name: testAddressName}, Options: options} mc := make(chan *pb.Message) defer close(mc) @@ -1889,7 +1911,7 @@ func Test_Subscribe_ExceedsAcquiredCount(t *testing.T) { msg1 := amqp091Message{ DeliveryTag: 1, Headers: amqp091Table{ - "x-acquired-count": int32(25), + testXAcquiredCount: int32(25), }, } msg1.SetDelivery(&delMock1) @@ -1949,7 +1971,7 @@ func Test_Subscribe_ExceedsAcquiredCount(t *testing.T) { defer stopWatcher(ctx, prov) assert.Nil(t, err) - src := &pb.Source{Name: "queue", Address: &pb.Address{Name: "address", Subjects: []string{"#"}}} + src := &pb.Source{Name: testQueue, Address: &pb.Address{Name: testAddress, Subjects: []string{"#"}}} mc := make(chan *pb.Message, 10) go func() { @@ -2007,7 +2029,7 @@ func Test_SupportedSourceOptions(t *testing.T) { opts := prov.SupportedSourceOptions() assert.NotNil(t, opts) expected := make(map[string]bool) - expected["MessageTTL"] = true + expected[testMessageTTLOption] = true expected["DeadLetterAddress"] = true expected["DeadLetterSubject"] = true expected["Expires"] = true @@ -2549,7 +2571,7 @@ func Test_fromAmqpMessage(t *testing.T) { del := amqp.Delivery{} del.Body = []byte("Hello") del.DeliveryMode = uint8(2) - del.Headers = amqp.Table{"h1": "header1"} + del.Headers = amqp.Table{"h1": testHeader1} del.ContentType = testContentEncodingText del.ContentEncoding = "plain" del.DeliveryTag = 1 @@ -2567,7 +2589,7 @@ func Test_toAmqpMessage(t *testing.T) { aMsg := &amqp091Message{} aMsg.Body = []byte("Hello") aMsg.DeliveryMode = 2 - aMsg.Headers = amqp091Table{"h1": "header1"} + aMsg.Headers = amqp091Table{"h1": testHeader1} aMsg.ContentType = testContentEncodingText aMsg.ContentEncoding = "plain" @@ -2811,9 +2833,9 @@ func mockManagementRequestServer() *httptest.Server { func setupCleanupBindingsTest(t *testing.T) (*BrokerDetails, *pb.Source) { t.Helper() bd := &BrokerDetails{} - addr := &pb.Address{Subjects: []string{"routingkey"}, Name: "address"} - src := &pb.Source{Address: addr, Name: "queue"} - creds := &pb.Credentials{Username: "user", Password: "password"} + addr := &pb.Address{Subjects: []string{testRoutingKey}, Name: testAddress} + src := &pb.Source{Address: addr, Name: testQueue} + creds := &pb.Credentials{Username: testUser, Password: testPassword} bd.connectionConfig = &pb.ConnectionConfiguration{Credentials: creds} msrv := mockManagementRequestServer() @@ -2835,7 +2857,7 @@ func Test_cleanupBindings(t *testing.T) { func Test_cleanupBindings_none(t *testing.T) { bd, src := setupCleanupBindingsTest(t) - removed := bd.cleanupBindings(src, []string{"routingkey"}) + removed := bd.cleanupBindings(src, []string{testRoutingKey}) assert.Len(t, removed, 0) } @@ -2856,8 +2878,8 @@ func Test_declareQueueAutoDelete(t *testing.T) { bd := &BrokerDetails{ knownQueues: util.NewConcurrentMap(), } - addr := &pb.Address{Subjects: []string{"routingkey"}, Name: "address"} - src := &pb.Source{Address: addr, Name: "queue", AutoDelete: adt.autoDelete, Exclusive: adt.exclusive} + addr := &pb.Address{Subjects: []string{testRoutingKey}, Name: testAddress} + src := &pb.Source{Address: addr, Name: testQueue, AutoDelete: adt.autoDelete, Exclusive: adt.exclusive} expectedArgs := make(amqp091Table) if adt.expires > 0 { @@ -2901,8 +2923,8 @@ func Test_singleActiveConsumer(t *testing.T) { bd := &BrokerDetails{ knownQueues: util.NewConcurrentMap(), } - addr := &pb.Address{Subjects: []string{"routingkey"}, Name: "address"} - src := &pb.Source{Address: addr, Name: "queue", SingleActiveConsumer: sac.singleActiveConsumer} + addr := &pb.Address{Subjects: []string{testRoutingKey}, Name: testAddress} + src := &pb.Source{Address: addr, Name: testQueue, SingleActiveConsumer: sac.singleActiveConsumer} expectedArgs := make(amqp091Table) expectedArgs["x-queue-type"] = testQueueTypeQuorum @@ -2952,8 +2974,8 @@ func Test_Subscribe_Queue_DeclareOnly(t *testing.T) { func(t *testing.T) { prov := NewAMQP091Provider() - addr := &pb.Address{Subjects: []string{"routingkey"}, Name: "address"} - src := &pb.Source{Address: addr, Name: "queue", Type: pb.Source_QUEUE, DeclareOnly: true} + addr := &pb.Address{Subjects: []string{testRoutingKey}, Name: testAddress} + src := &pb.Source{Address: addr, Name: testQueue, Type: pb.Source_QUEUE, DeclareOnly: true} cmock := &amqpChannelMock{} sbmock := &amqpChannelMock{} @@ -3031,7 +3053,7 @@ func Test_SourceStats(t *testing.T) { defer msrv.Close() u, err := url.Parse(msrv.URL) assert.Nil(t, err) - creds := &pb.Credentials{Username: "user", Password: "password"} + creds := &pb.Credentials{Username: testUser, Password: testPassword} bd.connectionConfig = &pb.ConnectionConfiguration{Credentials: creds} bd.connectionConfig.Host = u.Hostname() bd.connectionConfig.Tenant = testTenant @@ -3104,7 +3126,7 @@ func Test_SourceStats(t *testing.T) { bd.StreamConnection = nil // make sure we call NewStreamConn again addr := &pb.Address{ - Subjects: []string{"routingkey"}, + Subjects: []string{testRoutingKey}, Name: test.addressName, Type: test.addressType, } @@ -3157,7 +3179,7 @@ func Test_SourceStats_errors(t *testing.T) { assert.Equal(t, "address name not defined", stats.GetError().GetMessage()) // no client id so prov.getBrokerDetails fails - src.Address = &pb.Address{Name: "myname"} + src.Address = &pb.Address{Name: testMyName} stats = prov.SourceStats(ctx, src) assert.NotNil(t, stats) assert.Equal(t, "could not retrieve client-id from context", stats.GetError().GetMessage()) @@ -3177,7 +3199,7 @@ func stockMessage(address *pb.Address) *pb.Message { func stockAddress() *pb.Address { subjects := make([]string, 0) subjects = append(subjects, "subject1") - address := &pb.Address{Name: "addressname", Subjects: subjects, Type: pb.Address_FILTER} + address := &pb.Address{Name: testAddressName, Subjects: subjects, Type: pb.Address_FILTER} return address } @@ -3474,8 +3496,8 @@ func Test_QueueSubscribe_NotifyCloseChannelsAreBuffered(t *testing.T) { defer stopWatcher(ctx, prov) src := &pb.Source{ - Name: "queue", - Address: &pb.Address{Name: "address", Subjects: []string{"#"}}, + Name: testQueue, + Address: &pb.Address{Name: testAddress, Subjects: []string{"#"}}, } mc := make(chan *pb.Message) diff --git a/internal/provider/connectors/amqp091/amqp091shim_test.go b/internal/provider/connectors/amqp091/amqp091shim_test.go index ffa50e4..cdf94fa 100644 --- a/internal/provider/connectors/amqp091/amqp091shim_test.go +++ b/internal/provider/connectors/amqp091/amqp091shim_test.go @@ -34,7 +34,7 @@ func Test_Amqp091Message_Nack_AcquiredCount(t *testing.T) { name: "requeue true with int32 acquired count below limit remains true", initialRequeue: true, headers: amqp091Table{ - "x-acquired-count": int32(5), + testXAcquiredCount: int32(5), }, expectedRequeue: true, }, @@ -42,7 +42,7 @@ func Test_Amqp091Message_Nack_AcquiredCount(t *testing.T) { name: "requeue true with int32 acquired count equal to limit remains true", initialRequeue: true, headers: amqp091Table{ - "x-acquired-count": int32(20), + testXAcquiredCount: int32(20), }, expectedRequeue: true, }, @@ -50,7 +50,7 @@ func Test_Amqp091Message_Nack_AcquiredCount(t *testing.T) { name: "requeue true with int32 acquired count above limit becomes false", initialRequeue: true, headers: amqp091Table{ - "x-acquired-count": int32(21), + testXAcquiredCount: int32(21), }, expectedRequeue: false, }, @@ -58,7 +58,7 @@ func Test_Amqp091Message_Nack_AcquiredCount(t *testing.T) { name: "requeue true with int64 acquired count above limit becomes false", initialRequeue: true, headers: amqp091Table{ - "x-acquired-count": int64(25), + testXAcquiredCount: int64(25), }, expectedRequeue: false, }, @@ -66,7 +66,7 @@ func Test_Amqp091Message_Nack_AcquiredCount(t *testing.T) { name: "requeue true with int64 acquired count below limit remains true", initialRequeue: true, headers: amqp091Table{ - "x-acquired-count": int64(10), + testXAcquiredCount: int64(10), }, expectedRequeue: true, }, @@ -74,7 +74,7 @@ func Test_Amqp091Message_Nack_AcquiredCount(t *testing.T) { name: "x-acquired-count is a string(invalid) should be true", initialRequeue: true, headers: amqp091Table{ - "x-acquired-count": "25", + testXAcquiredCount: "25", }, expectedRequeue: true, }, diff --git a/internal/provider/connectors/amqp091/compression.go b/internal/provider/connectors/amqp091/compression.go index 0e3660d..1d7cb9c 100644 --- a/internal/provider/connectors/amqp091/compression.go +++ b/internal/provider/connectors/amqp091/compression.go @@ -92,7 +92,9 @@ func decompressBody(b []byte) ([]byte, error) { if err != nil { return nil, err } - defer r.Close() + defer func() { + _ = r.Close() + }() out, err := io.ReadAll(r) if err != nil { return nil, err diff --git a/internal/provider/connectors/amqp091/stream_test.go b/internal/provider/connectors/amqp091/stream_test.go index e2bee3a..f411968 100644 --- a/internal/provider/connectors/amqp091/stream_test.go +++ b/internal/provider/connectors/amqp091/stream_test.go @@ -205,7 +205,7 @@ func stockStreamConsumerContext() stream.ConsumerContext { func setFieldValue(target any, fieldName string, value any) { rv := reflect.ValueOf(target) - for rv.Kind() == reflect.Ptr && !rv.IsNil() { + for rv.Kind() == reflect.Pointer && !rv.IsNil() { rv = rv.Elem() } if !rv.CanAddr() { @@ -537,9 +537,9 @@ func Test_SubscribeStream(t *testing.T) { subjects = append(subjects, "subject1") address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0", "MessageTTL": "500"}, + Options: map[string]string{offsetOption: "0", testMessageTTLOption: "500"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -646,9 +646,9 @@ func Test_SubscribeStreamBadOpt(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0", "BadOpt": "true"}, + Options: map[string]string{offsetOption: "0", "BadOpt": "true"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -690,9 +690,9 @@ func Test_streamSubscribe(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - source := &pb.Source{Name: "srcname", + source := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0"}, + Options: map[string]string{offsetOption: "0"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -806,9 +806,9 @@ func Test_SubscribeStreamAutoDeleteOrExclusive(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0"}, + Options: map[string]string{offsetOption: "0"}, Exclusive: false, AutoDelete: true, Type: pb.Source_STREAM, @@ -881,9 +881,9 @@ func Test_SubscribeStreamInvalidTTL(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0", "MessageTTL": "true"}, + Options: map[string]string{offsetOption: "0", testMessageTTLOption: "true"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -945,9 +945,9 @@ func Test_SubscribeStreamFailedConn(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0"}, + Options: map[string]string{offsetOption: "0"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -1008,9 +1008,9 @@ func Test_SubscribeStreamNotConn(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0"}, + Options: map[string]string{offsetOption: "0"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -1072,9 +1072,9 @@ func Test_SubscribeStreamFailedDeclare(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0", "MessageTTL": "500"}, + Options: map[string]string{offsetOption: "0", testMessageTTLOption: "500"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -1111,9 +1111,9 @@ func Test_StreamRetry(t *testing.T) { address := stockAddress() address.Type = pb.Address_STREAM - src := &pb.Source{Name: "srcname", + src := &pb.Source{Name: testSrcName, Address: address, - Options: map[string]string{"Offset": "0", "MessageTTL": "500"}, + Options: map[string]string{offsetOption: "0", testMessageTTLOption: "500"}, Type: pb.Source_STREAM, PrefetchCount: 4} @@ -1124,7 +1124,7 @@ func Test_StreamRetry(t *testing.T) { pmock := &streamConsumerMock{} pmock.On("Close").Return(nil) - smock.On("NewConsumer", src.GetName(), src.GetName(), src.GetOptions()["Offset"], mock.Anything, mock.AnythingOfType("bool")).Return(pmock, nil) + smock.On("NewConsumer", src.GetName(), src.GetName(), src.GetOptions()[offsetOption], mock.Anything, mock.AnythingOfType("bool")).Return(pmock, nil) oldNewStreamConn := NewStreamConn NewStreamConn = func(string, string, *tls.Config) streamConnectionShim { return smock @@ -1206,8 +1206,8 @@ func Test_Subscribe_Stream_DeclareOnly(t *testing.T) { func(t *testing.T) { prov := NewAMQP091Provider() - addr := &pb.Address{Subjects: []string{"routingkey"}, Name: "address", Type: pb.Address_STREAM} - src := &pb.Source{Address: addr, Name: "queue", Type: pb.Source_STREAM, DeclareOnly: true} + addr := &pb.Address{Subjects: []string{testRoutingKey}, Name: testAddress, Type: pb.Address_STREAM} + src := &pb.Source{Address: addr, Name: testQueue, Type: pb.Source_STREAM, DeclareOnly: true} smock := &streamConnectionMock{} smock.On("Connect").Return(nil) @@ -1307,7 +1307,7 @@ func Test_streamSubscribe_ExchangeAndBindingDeclaration(t *testing.T) { Name: sourceName, Address: addr, Type: pb.Source_STREAM, - Options: map[string]string{"Offset": "0"}, + Options: map[string]string{offsetOption: "0"}, DeclareOnly: true, // stop after declare; avoids consumer setup } @@ -1447,7 +1447,7 @@ func setupStreamStatsTest(t *testing.T, guard *unconnectedGuardStreamConn) (*amq port, _ := strconv.Atoi(u.Port()) bd := &BrokerDetails{connectionConfig: &pb.ConnectionConfiguration{ - Credentials: &pb.Credentials{Username: "user", Password: "password"}, + Credentials: &pb.Credentials{Username: testUser, Password: testPassword}, Host: u.Hostname(), Tenant: testTenant, AdminPort: int32(port), //nolint:gosec @@ -1458,7 +1458,7 @@ func setupStreamStatsTest(t *testing.T, guard *unconnectedGuardStreamConn) (*amq src := &pb.Source{ Name: "sourceStream", Type: pb.Source_STREAM, - Address: &pb.Address{Name: "addressStream", Type: pb.Address_STREAM, Subjects: []string{"routingkey"}}, + Address: &pb.Address{Name: "addressStream", Type: pb.Address_STREAM, Subjects: []string{testRoutingKey}}, } return prov, bd, src } diff --git a/internal/provider/connectors/amqp091/streamhelpers_test.go b/internal/provider/connectors/amqp091/streamhelpers_test.go index 35bf784..89811c5 100644 --- a/internal/provider/connectors/amqp091/streamhelpers_test.go +++ b/internal/provider/connectors/amqp091/streamhelpers_test.go @@ -52,13 +52,13 @@ func Test_toStreamOffset(t *testing.T) { func Test_toStreamMessage(t *testing.T) { origMsg := streamMessage{ Body: []byte("test body"), - Headers: map[string]string{"header1": "value1", "header2": "value2"}, + Headers: map[string]string{testHeader1: testValue1, testHeader2: testValue2}, ContentEncoding: "utf-8", ContentType: "text/plain", } expectedMsg := amqp.NewMessage(origMsg.Body) - expectedMsg.ApplicationProperties = map[string]interface{}{"header1": "value1", "header2": "value2"} + expectedMsg.ApplicationProperties = map[string]interface{}{testHeader1: testValue1, testHeader2: testValue2} expectedMsg.Properties = &amqp.MessageProperties{ ContentEncoding: "utf-8", ContentType: "text/plain", @@ -73,15 +73,15 @@ func Test_toStreamMessage(t *testing.T) { func Test_toStreamHeaders(t *testing.T) { origHeaders := map[string]string{ - "header1": "value1", - "header2": "value2", - "header3": "value3", + testHeader1: testValue1, + testHeader2: testValue2, + "header3": testValue3, } expectedHeaders := map[string]interface{}{ - "header1": "value1", - "header2": "value2", - "header3": "value3", + testHeader1: testValue1, + testHeader2: testValue2, + "header3": testValue3, } resultHeaders := toStreamHeaders(origHeaders) @@ -99,10 +99,10 @@ func Test_getStreamConnectionString(t *testing.T) { bd: &BrokerDetails{ connectionConfig: &pb.ConnectionConfiguration{ Credentials: &pb.Credentials{ - Username: "user", - Password: "pass", + Username: testUser, + Password: testPass, }, - Host: "localhost", + Host: testLocalhost, }, tlsEnabled: false, }, @@ -113,10 +113,10 @@ func Test_getStreamConnectionString(t *testing.T) { bd: &BrokerDetails{ connectionConfig: &pb.ConnectionConfiguration{ Credentials: &pb.Credentials{ - Username: "user", - Password: "pass", + Username: testUser, + Password: testPass, }, - Host: "localhost", + Host: testLocalhost, Tenant: "tenant1", }, tlsEnabled: false, @@ -128,10 +128,10 @@ func Test_getStreamConnectionString(t *testing.T) { bd: &BrokerDetails{ connectionConfig: &pb.ConnectionConfiguration{ Credentials: &pb.Credentials{ - Username: "user", - Password: "pass", + Username: testUser, + Password: testPass, }, - Host: "localhost", + Host: testLocalhost, }, tlsEnabled: true, }, @@ -142,10 +142,10 @@ func Test_getStreamConnectionString(t *testing.T) { bd: &BrokerDetails{ connectionConfig: &pb.ConnectionConfiguration{ Credentials: &pb.Credentials{ - Username: "user", - Password: "pass", + Username: testUser, + Password: testPass, }, - Host: "localhost", + Host: testLocalhost, Tenant: "tenant1", }, tlsEnabled: true, @@ -157,10 +157,10 @@ func Test_getStreamConnectionString(t *testing.T) { bd: &BrokerDetails{ connectionConfig: &pb.ConnectionConfiguration{ Credentials: &pb.Credentials{ - Username: "user", - Password: "pass", + Username: testUser, + Password: testPass, }, - Host: "localhost", + Host: testLocalhost, Tenant: "/", }, tlsEnabled: false, @@ -172,10 +172,10 @@ func Test_getStreamConnectionString(t *testing.T) { bd: &BrokerDetails{ connectionConfig: &pb.ConnectionConfiguration{ Credentials: &pb.Credentials{ - Username: "user", - Password: "pass", + Username: testUser, + Password: testPass, }, - Host: "localhost", + Host: testLocalhost, Tenant: "/", }, tlsEnabled: true, diff --git a/internal/provider/connectors/amqp091/streamshim.go b/internal/provider/connectors/amqp091/streamshim.go index 71e1df1..eb8be9d 100644 --- a/internal/provider/connectors/amqp091/streamshim.go +++ b/internal/provider/connectors/amqp091/streamshim.go @@ -128,7 +128,7 @@ func (sc *streamConnection) PutPublisher(confirm bool, pub streamPublisherShim) if ok { if err := pool.(*util.BlockingPool).Put(pub); err != nil { util.Logger.Debugf("Failed to return publisher to pool: %s", err.Error()) - pub.Close() + _ = pub.Close() } } } @@ -295,7 +295,9 @@ func (sc *streamConnection) GetStreamOffset(streamName string) (int64, error) { if err != nil { return 0, err } - defer cons.Close() + defer func() { + _ = cons.Close() + }() return sc.GetConsumerOffset(streamName, fakeConsumerName) } diff --git a/internal/server/prometheus/interceptors_test.go b/internal/server/prometheus/interceptors_test.go index fa66d32..ccbcf97 100644 --- a/internal/server/prometheus/interceptors_test.go +++ b/internal/server/prometheus/interceptors_test.go @@ -14,6 +14,8 @@ import ( "google.golang.org/grpc" ) +const testGRPCMethodPublish = "arke.Producer.Publish" + // mockServerStream implements grpc.ServerStream for testing type mockServerStream struct { grpc.ServerStream @@ -206,7 +208,7 @@ func TestWrappedStream_RecvMsg_Success(t *testing.T) { setupTestSink(t) ms := &mockServerStream{recvErr: nil} - ws := &wrappedStream{ServerStream: ms, GRPCMethod: "arke.Producer.Publish"} + ws := &wrappedStream{ServerStream: ms, GRPCMethod: testGRPCMethodPublish} err := ws.RecvMsg(nil) if err != nil { @@ -219,7 +221,7 @@ func TestWrappedStream_RecvMsg_Error(t *testing.T) { expectedErr := errors.New("recv error") ms := &mockServerStream{recvErr: expectedErr} - ws := &wrappedStream{ServerStream: ms, GRPCMethod: "arke.Producer.Publish"} + ws := &wrappedStream{ServerStream: ms, GRPCMethod: testGRPCMethodPublish} err := ws.RecvMsg(nil) if err != expectedErr { @@ -254,7 +256,7 @@ func TestWrappedStream_SendMsg_Error(t *testing.T) { func TestNewWrappedStream_ReturnsWrappedStream(t *testing.T) { ms := &mockServerStream{} - method := "arke.Producer.Publish" + method := testGRPCMethodPublish ws := newWrappedStream(ms, method) wrapped, ok := ws.(*wrappedStream) diff --git a/internal/server/ratelimiter/ratelimiter_test.go b/internal/server/ratelimiter/ratelimiter_test.go index bd00119..0a1da18 100644 --- a/internal/server/ratelimiter/ratelimiter_test.go +++ b/internal/server/ratelimiter/ratelimiter_test.go @@ -6,7 +6,6 @@ package ratelimiter import ( "context" "fmt" - "os" "testing" "time" @@ -18,12 +17,9 @@ import ( "google.golang.org/grpc/peer" ) -func resetRateLimitEnv() { - _ = os.Unsetenv(EnvRateLimitEnforced) - _ = os.Unsetenv(EnvRateLimitBucketSize) - _ = os.Unsetenv(EnvRateLimitRefill) - _ = os.Unsetenv(EnvRateLimitMaxAge) - os.Setenv(util.EnvLogLevel, "DEBUG") +func resetRateLimitEnv(t *testing.T) { + t.Helper() + t.Setenv(util.EnvLogLevel, "DEBUG") } func TestInitializeClientLimitManager(t *testing.T) { @@ -64,7 +60,7 @@ func TestInitializeClientLimitManager(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - resetRateLimitEnv() + resetRateLimitEnv(t) clm, err := NewClientLimitManager(tt.bucketSize, tt.refillInterval, tt.maxAgeStaleClients, tt.enforced) if tt.shouldBeNil { assert.NotNil(t, err) @@ -127,7 +123,7 @@ func TestCullStaleClients(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - resetRateLimitEnv() + resetRateLimitEnv(t) clm, err := NewClientLimitManager(tt.bucketSize, tt.refillInterval, tt.maxAgeStaleClients, tt.enforced) assert.Nil(t, err, "unexpected error: %v", err) for i, lastActive := range tt.clientLastActive { @@ -191,7 +187,7 @@ func TestLimit(t *testing.T) { for testIndex, tt := range tests { t.Run(tt.name, func(t *testing.T) { - resetRateLimitEnv() + resetRateLimitEnv(t) clm, err := NewClientLimitManager(tt.bucketSize, tt.refillInterval, tt.maxAgeStaleClients, tt.enforced) assert.Nil(t, err, "unexpected error: %v", err) ctx := context.Background() @@ -254,7 +250,7 @@ func TestStartClientCull(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - resetRateLimitEnv() + resetRateLimitEnv(t) clm, err := NewClientLimitManager(tt.bucketSize, tt.refillInterval, tt.maxAgeStaleClients, tt.enforced) assert.Nil(t, err, "unexpected error: %v", err) @@ -366,7 +362,7 @@ func TestNewClientLimitManager(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - resetRateLimitEnv() + resetRateLimitEnv(t) clm, err := NewClientLimitManager(tt.bucketSize, tt.refillInterval, tt.maxAgeStaleClients, tt.enforced) if tt.expectedError { diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 6967560..3972715 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -28,6 +28,11 @@ var cf *pb.ConnectionConfiguration var mockp *MockProvider var conSrv *s.ConsumerServer var proSrv *s.ProducerServer + +const testAddressName = "addressname" +const testBreakingError = "breaking" +const testSourceName = "asdf" + var hlthSrv *s.HealthzServer var expectedErrorMessage = "this is my error" var errMsg pb.Error = pb.Error{Message: expectedErrorMessage} @@ -716,14 +721,14 @@ func TestConsumerServerConsume(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["option1"] = "ok" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} messages := make([]*pb.Message, 0) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("one")}) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("two")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("one")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("two")}) mockp.SetMockMessages(messages) @@ -734,7 +739,7 @@ func TestConsumerServerConsume(t *testing.T) { stream.On("Recv").Return(nil, errors.New("stop")).Once().After(500 * time.Millisecond) // We are returning an error after 500 ms as a simple way of exiting the subscribe - mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: "breaking"}).After(250 * time.Millisecond) + mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: testBreakingError}).After(250 * time.Millisecond) mockp.On("Ack", mock.Anything, mock.Anything).Return(nil) conSrv.Connect(ctx, cf) //nolint:errcheck err := conSrv.Consume(stream) @@ -751,14 +756,14 @@ func TestConsumerServerSubscribeReturnNil(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["option1"] = "ok" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} messages := make([]*pb.Message, 0) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("one")}) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("two")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("one")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("two")}) mockp.SetMockMessages(messages) mockp.SubscribeReturnDelay = 50 * time.Millisecond @@ -785,7 +790,7 @@ func TestConsumerServerSubscribeReturnNil(t *testing.T) { } func TestSetSourceDefaultsWithNoOptions(t *testing.T) { - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}} s.SetSourceDefaults(source) opts := source.GetOptions() @@ -805,7 +810,7 @@ func TestSetSourceDefaultsWithNoOptions(t *testing.T) { func TestSetSourceDefaults(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["option1"] = "ok" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} s.SetSourceDefaults(source) opts := source.GetOptions() @@ -836,14 +841,14 @@ func TestConsumerServerConsume_Nack(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["option1"] = "ok" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} messages := make([]*pb.Message, 0) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("one")}) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("two")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("one")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("two")}) mockp.SetMockMessages(messages) @@ -854,7 +859,7 @@ func TestConsumerServerConsume_Nack(t *testing.T) { stream.On("Recv").Return(nil, errors.New("stop")).Once().After(500 * time.Millisecond) // We are returning an error after 500 ms as a simple way of exiting the subscribe - mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: "breaking"}).After(250 * time.Millisecond) + mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: testBreakingError}).After(250 * time.Millisecond) mockp.On("Nack", mock.Anything, mock.Anything).Return(nil) conSrv.Connect(ctx, cf) //nolint:errcheck err := conSrv.Consume(stream) @@ -871,14 +876,14 @@ func TestConsumerServerConsume_DLQ(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["DeadLetterAddress"] = "dlq" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} messages := make([]*pb.Message, 0) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("one")}) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("two")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("one")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("two")}) mockp.SetMockMessages(messages) @@ -889,7 +894,7 @@ func TestConsumerServerConsume_DLQ(t *testing.T) { stream.On("Recv").Return(nil, errors.New("stop")).Once().After(500 * time.Millisecond) // We are returning an error after 500 ms as a simple way of exiting the subscribe - mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: "breaking"}).After(250 * time.Millisecond) + mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: testBreakingError}).After(250 * time.Millisecond) mockp.On("DeadLetter", mock.Anything, source, mock.Anything).Return(nil) conSrv.Connect(ctx, cf) //nolint:errcheck err := conSrv.Consume(stream) @@ -906,14 +911,14 @@ func TestConsumerServerConsume_DLQFail(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["DeadLetterAddress"] = "dlq" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} messages := make([]*pb.Message, 0) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("one")}) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("two")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("one")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("two")}) mockp.SetMockMessages(messages) @@ -924,8 +929,8 @@ func TestConsumerServerConsume_DLQFail(t *testing.T) { stream.On("Recv").Return(nil, errors.New("stop")).Once().After(500 * time.Millisecond) // We are returning an error after 500 ms as a simple way of exiting the subscribe - mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: "breaking"}).After(250 * time.Millisecond) - mockp.On("DeadLetter", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: "breaking"}) + mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: testBreakingError}).After(250 * time.Millisecond) + mockp.On("DeadLetter", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: testBreakingError}) mockp.On("Nack", mock.Anything, mock.Anything).Return(nil) conSrv.Connect(ctx, cf) //nolint:errcheck err := conSrv.Consume(stream) @@ -942,14 +947,14 @@ func TestConsumerServerConsume_Retry(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["option1"] = "ok" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} messages := make([]*pb.Message, 0) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("one")}) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("two")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("one")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("two")}) mockp.SetMockMessages(messages) @@ -960,7 +965,7 @@ func TestConsumerServerConsume_Retry(t *testing.T) { stream.On("Recv").Return(nil, errors.New("stop")).Once().After(500 * time.Millisecond) // We are returning an error after 500 ms as a simple way of exiting the subscribe - mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: "breaking"}).After(250 * time.Millisecond) + mockp.On("Subscribe", mock.Anything, source, mock.Anything).Return(&pb.Error{Message: testBreakingError}).After(250 * time.Millisecond) mockp.On("Retry", mock.Anything, source, mock.Anything, mock.Anything).Return(nil) conSrv.Connect(ctx, cf) //nolint:errcheck err := conSrv.Consume(stream) @@ -977,7 +982,7 @@ func TestConsumerServerConsume_BadOption(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["option1"] = "ok" sourceOptions["badoption"] = "bad" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} @@ -996,14 +1001,14 @@ func TestConsumerServerConsume_AckErr(t *testing.T) { mockp.On("Connect", mock.Anything, mock.AnythingOfType("*api.ConnectionConfiguration"), mock.AnythingOfType("bool")).Return(&pb.Error{}) mockp.On("WaitForConnect", mock.Anything).Return(true) - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} messages := make([]*pb.Message, 0) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("one")}) - messages = append(messages, &pb.Message{Address: &pb.Address{Name: "addressname"}, Body: []byte("two")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("one")}) + messages = append(messages, &pb.Message{Address: &pb.Address{Name: testAddressName}, Body: []byte("two")}) mockp.SetMockMessages(messages) @@ -1014,7 +1019,7 @@ func TestConsumerServerConsume_AckErr(t *testing.T) { stream.On("Recv").Return(nil, errors.New("stop")).Once().After(500 * time.Millisecond) // We are returning an error after 500 ms as a simple way of exiting the subscribe - mockp.On("Subscribe", mock.Anything, mock.Anything, mock.Anything).Return(&pb.Error{Message: "breaking"}).After(250 * time.Millisecond) + mockp.On("Subscribe", mock.Anything, mock.Anything, mock.Anything).Return(&pb.Error{Message: testBreakingError}).After(250 * time.Millisecond) mockp.On("Ack", mock.Anything, mock.Anything).Return(&pb.Error{Message: "ackerr"}) conSrv.Connect(ctx, cf) //nolint:errcheck err := conSrv.Consume(stream) @@ -1031,7 +1036,7 @@ func TestConsumerServerConsume_SourceTwice(t *testing.T) { sourceOptions := make(map[string]string) sourceOptions["option1"] = "ok" - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions, PrefetchCount: 1} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions, PrefetchCount: 1} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} @@ -1039,7 +1044,7 @@ func TestConsumerServerConsume_SourceTwice(t *testing.T) { stream.On("Send", cnsmResp).Return(nil, nil).Once() stream.On("Recv").Return(cnsm, nil).Twice() stream.On("Recv").Return(nil, io.EOF).After(100 * time.Millisecond) - mockp.On("Subscribe", mock.Anything, mock.Anything, mock.Anything).Return(&pb.Error{Message: "breaking"}).After(250 * time.Millisecond) + mockp.On("Subscribe", mock.Anything, mock.Anything, mock.Anything).Return(&pb.Error{Message: testBreakingError}).After(250 * time.Millisecond) conSrv.Connect(ctx, cf) //nolint:errcheck err := conSrv.Consume(stream) fmt.Println("err:", err) @@ -1145,7 +1150,7 @@ func TestConsumerServerConsume_SubscribeDeclareOnly(t *testing.T) { mockp.On("WaitForConnect", mock.Anything).Return(true) sourceOptions := make(map[string]string) - source := &pb.Source{Name: "asdf", Address: &pb.Address{Name: "addressname"}, Options: sourceOptions, DeclareOnly: true} + source := &pb.Source{Name: testSourceName, Address: &pb.Address{Name: testAddressName}, Options: sourceOptions, DeclareOnly: true} stream := &MockConsumerConsumeServerStream{} cnsm := &pb.Consume{Msg: &pb.Consume_Src{Src: source}} diff --git a/internal/util/cgroups_test.go b/internal/util/cgroups_test.go index 8f60b03..2cfe6ee 100644 --- a/internal/util/cgroups_test.go +++ b/internal/util/cgroups_test.go @@ -11,6 +11,8 @@ import ( "testing" ) +const testCgroupMax = "max" + // Helper function to reset the sync.Once for testing func resetCgroupSync() { cgroupSyncOnce = sync.Once{} @@ -135,7 +137,7 @@ func TestReadCgroupLimits_CgroupV2(t *testing.T) { }, { name: "max memory unlimited v2", - memLimit: "max", + memLimit: testCgroupMax, cpuQuota: "100000 100000", expectedMemory: 0, // remains unchanged expectedCPUMillicores: 1000, @@ -149,8 +151,8 @@ func TestReadCgroupLimits_CgroupV2(t *testing.T) { }, { name: "both max v2", - memLimit: "max", - cpuQuota: "max", + memLimit: testCgroupMax, + cpuQuota: testCgroupMax, expectedMemory: 0, expectedCPUMillicores: runtime.NumCPU() * 1000, }, diff --git a/internal/util/concurrentMap_test.go b/internal/util/concurrentMap_test.go index 2864d1e..5914f6e 100644 --- a/internal/util/concurrentMap_test.go +++ b/internal/util/concurrentMap_test.go @@ -15,6 +15,8 @@ type TestItem struct { name string } +const testItemName = "test item" + func TestNewConcurrentMap(t *testing.T) { cMap := NewConcurrentMap() assert.NotNil(t, cMap) @@ -22,7 +24,7 @@ func TestNewConcurrentMap(t *testing.T) { func TestConcurrentMapAdd(t *testing.T) { cMap := NewConcurrentMap() - testItem := TestItem{"test item"} + testItem := TestItem{testItemName} testItem2 := TestItem{"test item 2"} assert.NotNil(t, cMap) cMap.Add("testItem", &testItem) @@ -41,7 +43,7 @@ func TestConcurrentMapAdd(t *testing.T) { func TestConcurrentMapDelete(t *testing.T) { cMap := NewConcurrentMap() assert.NotNil(t, cMap) - testItem := TestItem{"test item"} + testItem := TestItem{testItemName} cMap.Add("testItem", testItem) cItem, ok := cMap.Get("testItem") assert.True(t, ok) @@ -54,7 +56,7 @@ func TestConcurrentMapDelete(t *testing.T) { func TestConcurrentMapDeleteIfEqual(t *testing.T) { cMap := NewConcurrentMap() - testItem := TestItem{"test item"} + testItem := TestItem{testItemName} staleItem := TestItem{"stale item"} replacementItem := TestItem{"replacement item"} @@ -88,7 +90,7 @@ func TestConcurrentMapDeleteIfEqual(t *testing.T) { func TestConcurrentMapGetList(t *testing.T) { cMap := NewConcurrentMap() assert.NotNil(t, cMap) - testItem := TestItem{"test item"} + testItem := TestItem{testItemName} cMap.Add("testItem", testItem) cMap.Add("testItem2", testItem) assert.Len(t, cMap.GetList(), 2) diff --git a/internal/util/file_logger.go b/internal/util/file_logger.go index 22c7942..43208e9 100644 --- a/internal/util/file_logger.go +++ b/internal/util/file_logger.go @@ -17,8 +17,8 @@ type TestLogger struct { // Note that this will close the writer and reader, so it should only be called once per // TestLogger instance func (l *TestLogger) GetOutput() []byte { - l.writer.Close() - defer l.reader.Close() + _ = l.writer.Close() + defer func() { _ = l.reader.Close() }() buf := make([]byte, 2048) n, _ := l.reader.Read(buf) return buf[:n] @@ -31,8 +31,8 @@ func GetTestLoggerWithCleanup() (*TestLogger, func()) { r, w, _ := os.Pipe() logger := NewArkeFileLogger(w) cleanup := func() { - w.Close() - r.Close() + _ = w.Close() + _ = r.Close() // Reset the logger so that other tests can expect an initialized logger NewArkeFileLogger(LogOutputFile) } diff --git a/internal/util/k8s_test.go b/internal/util/k8s_test.go index 7746625..9278adc 100644 --- a/internal/util/k8s_test.go +++ b/internal/util/k8s_test.go @@ -11,6 +11,7 @@ import ( pb "github.com/sassoftware/arke/api" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "k8s.io/client-go/rest" ) @@ -19,14 +20,14 @@ func Test_MonitorHPA(t *testing.T) { tests := []struct { name string expectedMsg string - setupFunc func() func() // returns cleanup function - expectNoReturn bool // whether function should return early + setupFunc func(t *testing.T) func() // returns cleanup function + expectNoReturn bool // whether function should return early }{ { name: "no namespace from file or env", expectedMsg: "No kubernetes namespace detected, not monitoring HPA for changes", expectNoReturn: true, - setupFunc: func() func() { + setupFunc: func(t *testing.T) func() { return func() { } }, @@ -35,12 +36,14 @@ func Test_MonitorHPA(t *testing.T) { name: "namespace from file ErrNotInCluster", expectedMsg: "Could not configure HPA cluster monitoring:", expectNoReturn: true, - setupFunc: func() func() { + setupFunc: func(t *testing.T) func() { origNamespaceFile := namespaceFile tDir := os.TempDir() ns, _ := os.CreateTemp(tDir, "namespace") - ns.WriteString("test-namespace") //nolint:errcheck - ns.Close() + _, err := ns.WriteString("test-namespace") + require.Nil(t, err) + err = ns.Close() + require.Nil(t, err) namespaceFile = ns.Name() // Restore original value on cleanup return func() { @@ -52,7 +55,7 @@ func Test_MonitorHPA(t *testing.T) { name: "namespace from file InClusterConfig error", expectedMsg: "Could not configure HPA cluster monitoring: file does not exist", expectNoReturn: true, - setupFunc: func() func() { + setupFunc: func(t *testing.T) func() { origNamespaceFile := namespaceFile origInClusterConfig := inClusterConfig inClusterConfig = func() (*rest.Config, error) { @@ -60,8 +63,10 @@ func Test_MonitorHPA(t *testing.T) { } tDir := os.TempDir() ns, _ := os.CreateTemp(tDir, "namespace") - ns.WriteString("test-namespace") //nolint:errcheck - ns.Close() + _, err := ns.WriteString("test-namespace") + require.Nil(t, err) + err = ns.Close() + require.Nil(t, err) namespaceFile = ns.Name() // Restore original value on cleanup return func() { @@ -74,18 +79,16 @@ func Test_MonitorHPA(t *testing.T) { name: "namespace from env var invalid namespace", expectedMsg: "Could not configure HPA cluster monitoring:", expectNoReturn: true, - setupFunc: func() func() { - os.Setenv(EnvK8SNamespace, "test-namespace") - return func() { - os.Unsetenv(EnvK8SNamespace) - } + setupFunc: func(t *testing.T) func() { + t.Setenv(EnvK8SNamespace, "test-namespace") + return func() {} }, }, { name: "Could not get HPA watcher", expectedMsg: "Could not get HPA watcher:", expectNoReturn: true, - setupFunc: func() func() { + setupFunc: func(t *testing.T) func() { origNamespaceFile := namespaceFile origInClusterConfig := inClusterConfig inClusterConfig = func() (*rest.Config, error) { @@ -93,8 +96,10 @@ func Test_MonitorHPA(t *testing.T) { } tDir := os.TempDir() ns, _ := os.CreateTemp(tDir, "namespace") - ns.WriteString("test-namespace") //nolint:errcheck - ns.Close() + _, err := ns.WriteString("test-namespace") + require.Nil(t, err) + err = ns.Close() + require.Nil(t, err) namespaceFile = ns.Name() return func() { namespaceFile = origNamespaceFile @@ -116,7 +121,7 @@ func Test_MonitorHPA(t *testing.T) { defer cleanupLogger() // Setup test conditions - cleanup := tt.setupFunc() + cleanup := tt.setupFunc(t) defer cleanup() // Run the function diff --git a/internal/util/tracing/tracing_test.go b/internal/util/tracing/tracing_test.go index 0834b5c..c81f8c2 100644 --- a/internal/util/tracing/tracing_test.go +++ b/internal/util/tracing/tracing_test.go @@ -5,7 +5,6 @@ package tracing import ( "context" - "os" "testing" "github.com/stretchr/testify/assert" @@ -13,56 +12,51 @@ import ( ) func TestGetTelemetryEnabled(t *testing.T) { - defer os.Setenv(EnvOtelSdkDisabled, "true") // A true value for OTEL_SDK_DISABLED is the only way to disable the SDK according to the docs, // even though this env var is not used in the Go SDK // https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#general-sdk-configuration // OTEL_SDK_DISABLED | true | false | '' | other // getTelementryEnabled | false | true | true | true - os.Setenv(EnvOtelSdkDisabled, "true") + t.Setenv(EnvOtelSdkDisabled, "true") enabled := getTelemetryEnabled() assert.False(t, enabled) - os.Setenv(EnvOtelSdkDisabled, "false") + t.Setenv(EnvOtelSdkDisabled, "false") enabled = getTelemetryEnabled() assert.True(t, enabled) - os.Setenv(EnvOtelSdkDisabled, "") + t.Setenv(EnvOtelSdkDisabled, "") enabled = getTelemetryEnabled() assert.True(t, enabled) - os.Setenv(EnvOtelSdkDisabled, "other") + t.Setenv(EnvOtelSdkDisabled, "other") enabled = getTelemetryEnabled() assert.True(t, enabled) } func Test_InitTracerProvider_disabled(t *testing.T) { - os.Setenv(EnvOtelSdkDisabled, "true") + t.Setenv(EnvOtelSdkDisabled, "true") tp, err := InitTracerProvider() assert.Nil(t, tp) assert.Nil(t, err) } func Test_InitTracerProvider_enabled(t *testing.T) { - os.Setenv(EnvOtelSdkDisabled, "false") - defer os.Setenv(EnvOtelSdkDisabled, "true") + t.Setenv(EnvOtelSdkDisabled, "false") tp, err := InitTracerProvider() assert.NotNil(t, tp) assert.Nil(t, err) } func Test_getTelemetryCollectorAddress_unset(t *testing.T) { - os.Unsetenv(EnvOtelExporterOtlpEndpoint) + t.Setenv(EnvOtelExporterOtlpEndpoint, "") addr := getTelemetryCollectorAddress() assert.Equal(t, "localhost:4317", addr) - - os.Getenv(EnvOtelExporterOtlpEndpoint) } func Test_getTelemetryCollectorAddress_set(t *testing.T) { - defer os.Unsetenv(EnvOtelExporterOtlpEndpoint) - os.Setenv(EnvOtelExporterOtlpEndpoint, "localhost:12345") + t.Setenv(EnvOtelExporterOtlpEndpoint, "localhost:12345") addr := getTelemetryCollectorAddress() assert.Equal(t, "localhost:12345", addr) } @@ -80,8 +74,7 @@ func Test_initResource_returnsSamePointer(t *testing.T) { } func Test_SpanFromHeaders(t *testing.T) { - os.Setenv(EnvOtelSdkDisabled, "false") - defer os.Setenv(EnvOtelSdkDisabled, "true") + t.Setenv(EnvOtelSdkDisabled, "false") tp, err := InitTracerProvider() assert.NotNil(t, tp) assert.Nil(t, err) diff --git a/internal/util/util_test.go b/internal/util/util_test.go index 751a7cc..a4f857d 100644 --- a/internal/util/util_test.go +++ b/internal/util/util_test.go @@ -7,7 +7,6 @@ import ( "context" "fmt" "net" - "os" "runtime" "testing" "time" @@ -23,6 +22,9 @@ type fakeAddr struct { network string } +const testLoopbackAddr = "127.0.0.1" +const testNetworkTCP = "tcp" + func (f fakeAddr) String() string { return f.s } @@ -41,17 +43,17 @@ func TestGenUUID(t *testing.T) { func Test_GetClientAddr(t *testing.T) { p := &peer.Peer{} - p.Addr = fakeAddr{s: "127.0.0.1", network: "tcp"} + p.Addr = fakeAddr{s: testLoopbackAddr, network: testNetworkTCP} ctx := peer.NewContext(context.Background(), p) addr, err := GetClientAddr(ctx) assert.Nil(t, err) - assert.Equal(t, addr, "127.0.0.1") + assert.Equal(t, addr, testLoopbackAddr) } func Test_SetClientIdentifier(t *testing.T) { p := &peer.Peer{} - p.Addr = fakeAddr{s: "127.0.0.1", network: "tcp"} + p.Addr = fakeAddr{s: testLoopbackAddr, network: testNetworkTCP} ctx := peer.NewContext(context.Background(), p) id, err := SetClientIdentifier(ctx, "unitTest") assert.Contains(t, id, "unitTest-") @@ -71,7 +73,7 @@ func Test_SetClientIdentifier(t *testing.T) { func Test_RemoveClientIdentifier(t *testing.T) { p := &peer.Peer{} - p.Addr = fakeAddr{s: "127.0.0.1", network: "tcp"} + p.Addr = fakeAddr{s: testLoopbackAddr, network: testNetworkTCP} ctx := peer.NewContext(context.Background(), p) id, err := SetClientIdentifier(ctx, "unitTest") assert.Contains(t, id, "unitTest-") @@ -96,7 +98,6 @@ func Test_TestProcessStats(t *testing.T) { func Test_GetConfig(t *testing.T) { testVar := "TEST_ENV_VAR" - defer os.Unsetenv(testVar) // Default value int intVal := GetConfig(testVar, 11) @@ -111,17 +112,17 @@ func Test_GetConfig(t *testing.T) { assert.Equal(t, "test", strVal) // Env var int - os.Setenv(testVar, "10") + t.Setenv(testVar, "10") intVal = GetConfig(testVar, 11) assert.Equal(t, 10, intVal) // Env var bool - os.Setenv(testVar, "true") + t.Setenv(testVar, "true") boolVal = GetConfig(testVar, false) assert.Equal(t, true, boolVal) // Env var string - os.Setenv(testVar, "foo") + t.Setenv(testVar, "foo") strVal = GetConfig(testVar, "bar") assert.Equal(t, "foo", strVal) } diff --git a/internal/util/zerolog_test.go b/internal/util/zerolog_test.go index d48461e..b1a4b58 100644 --- a/internal/util/zerolog_test.go +++ b/internal/util/zerolog_test.go @@ -10,8 +10,12 @@ import ( "github.com/rs/zerolog" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) +const testLogLevelInfo = "info" +const testLogLevelInvalid = "invalid" + func TestLoadAndParseLevel(t *testing.T) { tests := []struct { name string @@ -35,7 +39,7 @@ func TestLoadAndParseLevel(t *testing.T) { }, { name: "valid info level", - envValue: "info", + envValue: testLogLevelInfo, expected: zerolog.InfoLevel, }, { @@ -50,7 +54,7 @@ func TestLoadAndParseLevel(t *testing.T) { }, { name: "invalid level defaults to info", - envValue: "invalid", + envValue: testLogLevelInvalid, expected: zerolog.InfoLevel, }, { @@ -64,7 +68,7 @@ func TestLoadAndParseLevel(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Setenv(EnvLogLevel, tt.envValue) gotLevel, err := loadAndParseLevel() - if tt.envValue == "invalid" { + if tt.envValue == testLogLevelInvalid { assert.NotNil(t, err, "expected error for invalid log level") } else { assert.Nil(t, err, "did not expect error for valid log level") @@ -83,10 +87,10 @@ func TestCreateFileLogger(t *testing.T) { }{ { name: "creates logger with default settings", - logLevel: "info", + logLevel: testLogLevelInfo, logFormat: "", setupEnv: func() { - t.Setenv(EnvLogLevel, "info") + t.Setenv(EnvLogLevel, testLogLevelInfo) t.Setenv(EnvLogFormat, "") }, }, @@ -100,19 +104,19 @@ func TestCreateFileLogger(t *testing.T) { }, { name: "creates logger with term format", - logLevel: "info", + logLevel: testLogLevelInfo, logFormat: "term", setupEnv: func() { - t.Setenv(EnvLogLevel, "info") + t.Setenv(EnvLogLevel, testLogLevelInfo) t.Setenv(EnvLogFormat, "term") }, }, { name: "creates logger with invalid level defaults to info", - logLevel: "invalid", + logLevel: testLogLevelInvalid, logFormat: "", setupEnv: func() { - t.Setenv(EnvLogLevel, "invalid") + t.Setenv(EnvLogLevel, testLogLevelInvalid) }, }, } @@ -139,7 +143,8 @@ func TestCreateFileLogger(t *testing.T) { assert.Equal(t, expLevel, l.GetLevel(), "logger level should match expected level %s", tt.logLevel) l.Info().Msg("Test log message") - w.Close() + err = w.Close() + require.Nil(t, err) loggerOutput := make([]byte, 1024) bytesRead, _ := r.Read(loggerOutput) assert.Contains(t, string(loggerOutput[:bytesRead]), "Test log message") diff --git a/pkg/arke/arke_test.go b/pkg/arke/arke_test.go index 00901a1..1d17ff1 100644 --- a/pkg/arke/arke_test.go +++ b/pkg/arke/arke_test.go @@ -7,7 +7,6 @@ import ( "context" "fmt" "log" - "os" "testing" "time" @@ -17,6 +16,9 @@ import ( "google.golang.org/grpc/credentials/insecure" ) +const testEnforcedTrue = "true" +const testInvalidValue = "invalid" + func Test_DefaultArkeServer(t *testing.T) { a := DefaultArkeServer().WithCertFilePath("/cert").WithCertKeyPath("/key").WithTLSSkipVerify(true) @@ -40,13 +42,13 @@ func Test_defaultKAEP(t *testing.T) { } func Test_DefaultArkeServerWithPortEnv(t *testing.T) { - os.Setenv("ARKE_PORT", "1234") + t.Setenv("ARKE_PORT", "1234") a := DefaultArkeServer() assert.Equal(t, 1234, a.port) } func Test_DefaultArkeServerWithPortEnvWithPortCall(t *testing.T) { - os.Setenv("ARKE_PORT", "1234") + t.Setenv("ARKE_PORT", "1234") a := DefaultArkeServer().WithPort(1235) assert.Equal(t, 1235, a.port) } @@ -129,7 +131,12 @@ func testHealth(port int) error { if err != nil { log.Fatalf("did not connect: %v", err) } - defer conn.Close() + defer func() { + err = conn.Close() + if err != nil { + log.Printf("error closing connection: %v", err) + } + }() c := pb.NewHealthzClient(conn) ctx := context.Background() _, err = c.Check(ctx) @@ -179,7 +186,7 @@ func Test_GetRateLimitParameters(t *testing.T) { bucketSize: "10", refillIntervalSec: "30", maxAgeStaleClientSec: "60", - enforced: "true", + enforced: testEnforcedTrue, expectedParams: &RateLimitParameters{ BucketSize: 10, RefillInterval: 30 * time.Second, @@ -190,19 +197,19 @@ func Test_GetRateLimitParameters(t *testing.T) { }, { name: "Invalid bucket size", - bucketSize: "invalid", + bucketSize: testInvalidValue, refillIntervalSec: "30", maxAgeStaleClientSec: "60", - enforced: "true", + enforced: testEnforcedTrue, expectedParams: nil, expectError: true, }, { name: "Invalid refill interval", bucketSize: "10", - refillIntervalSec: "invalid", + refillIntervalSec: testInvalidValue, maxAgeStaleClientSec: "60", - enforced: "true", + enforced: testEnforcedTrue, expectedParams: nil, expectError: true, }, @@ -210,8 +217,8 @@ func Test_GetRateLimitParameters(t *testing.T) { name: "Invalid max age stale client", bucketSize: "10", refillIntervalSec: "30", - maxAgeStaleClientSec: "invalid", - enforced: "true", + maxAgeStaleClientSec: testInvalidValue, + enforced: testEnforcedTrue, expectedParams: nil, expectError: true, }, diff --git a/test/config/env_test.go b/test/config/env_test.go index a81330a..781b0d7 100644 --- a/test/config/env_test.go +++ b/test/config/env_test.go @@ -11,6 +11,24 @@ import ( "github.com/stretchr/testify/assert" ) +const testDefaultValue = "default_value" +const testLocalhostName = "localhost" + +// unsetEnvForTest ensures an environment variable is truly absent (not just +// set to an empty string) for the duration of the test, restoring the +// previous state (present or absent) on cleanup. This is needed because +// t.Setenv has no equivalent of os.Unsetenv and getenv() distinguishes +// between an unset variable and one explicitly set to "". +func unsetEnvForTest(t *testing.T, key string) { + t.Helper() + if old, ok := os.LookupEnv(key); ok { + _ = os.Unsetenv(key) + t.Cleanup(func() { + _ = os.Setenv(key, old) + }) + } +} + func TestGetenv(t *testing.T) { tests := []struct { name string @@ -23,14 +41,14 @@ func TestGetenv(t *testing.T) { { name: "returns default when env var not set", key: "TEST_UNSET_VAR", - defaultValue: "default_value", + defaultValue: testDefaultValue, setEnv: false, - expected: "default_value", + expected: testDefaultValue, }, { name: "returns env value when set", key: "TEST_SET_VAR", - defaultValue: "default_value", + defaultValue: testDefaultValue, envValue: "custom_value", setEnv: true, expected: "custom_value", @@ -38,7 +56,7 @@ func TestGetenv(t *testing.T) { { name: "returns empty string when env var is empty", key: "TEST_EMPTY_VAR", - defaultValue: "default_value", + defaultValue: testDefaultValue, envValue: "", setEnv: true, expected: "", @@ -46,9 +64,9 @@ func TestGetenv(t *testing.T) { { name: "returns default when key is empty", key: "", - defaultValue: "default_value", + defaultValue: testDefaultValue, setEnv: false, - expected: "default_value", + expected: testDefaultValue, }, { name: "handles special characters in env value", @@ -62,12 +80,8 @@ func TestGetenv(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - // Ensure clean state - os.Unsetenv(tt.key) - defer os.Unsetenv(tt.key) - if tt.setEnv { - os.Setenv(tt.key, tt.envValue) + t.Setenv(tt.key, tt.envValue) } result := getenv(tt.key, tt.defaultValue) @@ -89,13 +103,8 @@ func TestConnectionConfigurationFromEnv_Defaults(t *testing.T) { } for _, v := range envVars { - os.Unsetenv(v) + unsetEnvForTest(t, v) } - defer func() { - for _, v := range envVars { - os.Unsetenv(v) - } - }() config := ConnectionConfigurationFromEnv() @@ -110,23 +119,13 @@ func TestConnectionConfigurationFromEnv_Defaults(t *testing.T) { func TestConnectionConfigurationFromEnv_CustomValues(t *testing.T) { // Set custom environment variables - os.Setenv("ARKE_BROKER_HOSTNAME", "custom-broker.example.com") - os.Setenv("ARKE_BROKER_PORT", "5671") - os.Setenv("ARKE_BROKER_ADMIN_PORT", "15672") - os.Setenv("ARKE_BROKER_USERNAME", "admin") - os.Setenv("ARKE_BROKER_PASSWORD", "secret123") - os.Setenv("ARKE_BROKER_TYPE", "amqp10") - os.Setenv("ARKE_BROKER_TENANT", "/custom-tenant") - - defer func() { - os.Unsetenv("ARKE_BROKER_HOSTNAME") - os.Unsetenv("ARKE_BROKER_PORT") - os.Unsetenv("ARKE_BROKER_ADMIN_PORT") - os.Unsetenv("ARKE_BROKER_USERNAME") - os.Unsetenv("ARKE_BROKER_PASSWORD") - os.Unsetenv("ARKE_BROKER_TYPE") - os.Unsetenv("ARKE_BROKER_TENANT") - }() + t.Setenv("ARKE_BROKER_HOSTNAME", "custom-broker.example.com") + t.Setenv("ARKE_BROKER_PORT", "5671") + t.Setenv("ARKE_BROKER_ADMIN_PORT", "15672") + t.Setenv("ARKE_BROKER_USERNAME", "admin") + t.Setenv("ARKE_BROKER_PASSWORD", "secret123") + t.Setenv("ARKE_BROKER_TYPE", "amqp10") + t.Setenv("ARKE_BROKER_TENANT", "/custom-tenant") config := ConnectionConfigurationFromEnv() @@ -141,15 +140,9 @@ func TestConnectionConfigurationFromEnv_CustomValues(t *testing.T) { func TestConnectionConfigurationFromEnv_PartialCustomValues(t *testing.T) { // Set only some custom values, others should use defaults - os.Setenv("ARKE_BROKER_HOSTNAME", "my-broker.local") - os.Setenv("ARKE_BROKER_USERNAME", "myuser") - os.Setenv("ARKE_BROKER_TYPE", "rabbitmq") - - defer func() { - os.Unsetenv("ARKE_BROKER_HOSTNAME") - os.Unsetenv("ARKE_BROKER_USERNAME") - os.Unsetenv("ARKE_BROKER_TYPE") - }() + t.Setenv("ARKE_BROKER_HOSTNAME", "my-broker.local") + t.Setenv("ARKE_BROKER_USERNAME", "myuser") + t.Setenv("ARKE_BROKER_TYPE", "rabbitmq") config := ConnectionConfigurationFromEnv() @@ -202,13 +195,8 @@ func TestConnectionConfigurationFromEnv_PortParsing(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - os.Setenv("ARKE_BROKER_PORT", tt.port) - os.Setenv("ARKE_BROKER_ADMIN_PORT", tt.adminPort) - - defer func() { - os.Unsetenv("ARKE_BROKER_PORT") - os.Unsetenv("ARKE_BROKER_ADMIN_PORT") - }() + t.Setenv("ARKE_BROKER_PORT", tt.port) + t.Setenv("ARKE_BROKER_ADMIN_PORT", tt.adminPort) config := ConnectionConfigurationFromEnv() @@ -219,8 +207,8 @@ func TestConnectionConfigurationFromEnv_PortParsing(t *testing.T) { } func TestConnectionConfigurationFromEnv_CredentialsNotNil(t *testing.T) { - os.Unsetenv("ARKE_BROKER_USERNAME") - os.Unsetenv("ARKE_BROKER_PASSWORD") + unsetEnvForTest(t, "ARKE_BROKER_USERNAME") + unsetEnvForTest(t, "ARKE_BROKER_PASSWORD") config := ConnectionConfigurationFromEnv() @@ -259,8 +247,7 @@ func TestConnectionConfigurationFromEnv_BrokerTypes(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - os.Setenv("ARKE_BROKER_TYPE", tt.brokerType) - defer os.Unsetenv("ARKE_BROKER_TYPE") + t.Setenv("ARKE_BROKER_TYPE", tt.brokerType) config := ConnectionConfigurationFromEnv() @@ -299,11 +286,10 @@ func TestConnectionConfigurationFromEnv_TenantValues(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if tt.tenant != "" { - os.Setenv("ARKE_BROKER_TENANT", tt.tenant) - defer os.Unsetenv("ARKE_BROKER_TENANT") + if tt.tenant == "" { + unsetEnvForTest(t, "ARKE_BROKER_TENANT") } else { - os.Unsetenv("ARKE_BROKER_TENANT") + t.Setenv("ARKE_BROKER_TENANT", tt.tenant) } config := ConnectionConfigurationFromEnv() @@ -315,19 +301,11 @@ func TestConnectionConfigurationFromEnv_TenantValues(t *testing.T) { func TestConnectionConfigurationFromEnv_EmptyStringValues(t *testing.T) { // Set all vars to empty strings - os.Setenv("ARKE_BROKER_HOSTNAME", "") - os.Setenv("ARKE_BROKER_USERNAME", "") - os.Setenv("ARKE_BROKER_PASSWORD", "") - os.Setenv("ARKE_BROKER_TYPE", "") - os.Setenv("ARKE_BROKER_TENANT", "") - - defer func() { - os.Unsetenv("ARKE_BROKER_HOSTNAME") - os.Unsetenv("ARKE_BROKER_USERNAME") - os.Unsetenv("ARKE_BROKER_PASSWORD") - os.Unsetenv("ARKE_BROKER_TYPE") - os.Unsetenv("ARKE_BROKER_TENANT") - }() + t.Setenv("ARKE_BROKER_HOSTNAME", "") + t.Setenv("ARKE_BROKER_USERNAME", "") + t.Setenv("ARKE_BROKER_PASSWORD", "") + t.Setenv("ARKE_BROKER_TYPE", "") + t.Setenv("ARKE_BROKER_TENANT", "") config := ConnectionConfigurationFromEnv() @@ -346,9 +324,9 @@ func TestConnectionConfigurationFromEnv_HostnameVariations(t *testing.T) { expectedHost string }{ { - name: "localhost", - hostname: "localhost", - expectedHost: "localhost", + name: testLocalhostName, + hostname: testLocalhostName, + expectedHost: testLocalhostName, }, { name: "IP address", @@ -374,8 +352,7 @@ func TestConnectionConfigurationFromEnv_HostnameVariations(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - os.Setenv("ARKE_BROKER_HOSTNAME", tt.hostname) - defer os.Unsetenv("ARKE_BROKER_HOSTNAME") + t.Setenv("ARKE_BROKER_HOSTNAME", tt.hostname) config := ConnectionConfigurationFromEnv() diff --git a/test/healthz/healthz.go b/test/healthz/healthz.go index de2f979..60ca6c3 100644 --- a/test/healthz/healthz.go +++ b/test/healthz/healthz.go @@ -37,18 +37,23 @@ func main() { if err != nil { log.Fatalf("did not connect: %v", err) } - defer conn.Close() + defer func() { + err = conn.Close() + if err != nil { + log.Printf("error closing connection: %v", err) + } + }() c := pb.NewHealthzClient(conn) ctx := context.Background() stream, err := c.Check(ctx) if err != nil { - fmt.Println("err on check", err) + log.Printf("error on check: %v", err) } go func() { sig := <-sigs fmt.Println() - fmt.Println(sig) + log.Printf("received signal: %v", sig) os.Exit(1) }() @@ -57,11 +62,11 @@ func main() { for { hlth, err := stream.Recv() if err != nil { - fmt.Printf("error on recv %s\n", err) + log.Printf("error on recv: %v", err) return } if status := hlth.GetStatus(); status != nil { - fmt.Printf("health for uuid %s is %s, cpu availability is %v, memory availability is %v\n", + log.Printf("health for uuid %s is %s, cpu availability is %v, memory availability is %v", status.GetUuid(), status.GetCode(), status.GetCpuAvailability(), @@ -79,7 +84,7 @@ func main() { hlth.Resp = hc err := stream.Send(hlth) if err != nil { - fmt.Println("error on send", err) + log.Printf("error on send: %v", err) return } <-ticker.C diff --git a/test/util/util_test.go b/test/util/util_test.go index 43dd131..b810f6f 100644 --- a/test/util/util_test.go +++ b/test/util/util_test.go @@ -16,7 +16,8 @@ func TestConnectToArke(t *testing.T) { assert.Nil(t, err, "error when connecting to arke: %v", err) assert.NotNil(t, conn, "got a nil connection") if conn != nil { - conn.Close() + err = conn.Close() + assert.NoError(t, err) } } From cdef9e3681d87ebf8c74f487886c1f27d3c89921 Mon Sep 17 00:00:00 2001 From: Dane Hunt Date: Tue, 27 Oct 2026 17:00:10 -0400 Subject: [PATCH 3/4] ci: run betterleaks only on current dir Signed-off-by: Dane Hunt --- .github/workflows/ci.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index daacec3..6b12584 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -86,7 +86,7 @@ jobs: url="https://github.com/betterleaks/betterleaks/releases/download/v${{ env.BETTERLEAKS_VERSION }}/betterleaks_${{ env.BETTERLEAKS_VERSION }}_${os}_${arch}.tar.gz" echo "Downloading betterleaks from $url" curl -sSL "$url" | tar zxf - > "$outfile" - "$outfile" git --config ./.github/linters/betterleaks-config.toml -v + "$outfile" dir . --config ./.github/linters/betterleaks-config.toml -v go-lint: runs-on: ubuntu-latest From 94848a716ca74cb6b318af773c87aadb01dc7dec Mon Sep 17 00:00:00 2001 From: Richard Sugg Date: Wed, 28 Oct 2026 13:56:54 -0400 Subject: [PATCH 4/4] fix: have betterleaks ignore test certs and examples Signed-off-by: Richard Sugg --- .github/linters/betterleaks-config.toml | 6 ++++++ .github/linters/betterleaks-ignore.txt | 9 +++++++++ .github/workflows/ci.yaml | 4 +++- 3 files changed, 18 insertions(+), 1 deletion(-) create mode 100644 .github/linters/betterleaks-ignore.txt diff --git a/.github/linters/betterleaks-config.toml b/.github/linters/betterleaks-config.toml index 4d17df2..460beed 100644 --- a/.github/linters/betterleaks-config.toml +++ b/.github/linters/betterleaks-config.toml @@ -3,6 +3,12 @@ title = "betterleaks config extensions" [extend] useDefault = true +[allowlist] +description = "Ignore test certificates and keys" +paths = [ + '''^tests/integration/certs/''' +] + [[rules]] id = "generic-api-key" diff --git a/.github/linters/betterleaks-ignore.txt b/.github/linters/betterleaks-ignore.txt new file mode 100644 index 0000000..6dfe363 --- /dev/null +++ b/.github/linters/betterleaks-ignore.txt @@ -0,0 +1,9 @@ +tests/integration/certs/server/rabbitmq_combined.pem:private-key:21 +tests/integration/certs/server/rabbitmq.key:private-key:1 +tests/integration/certs/server/arke.key:private-key:1 +tests/integration/certs/testca/private/ca_private_key.pem:private-key:1 +doc/design/client-publish-consume.md:generic-password:124 +Makefile:generic-password:187 +tools/anarkey/example-config-queue.json:generic-password:9 +tools/anarkey/example-config-stream.json:generic-password:9 +examples/all_in_one/main.go:generic-password:46 diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 6b12584..0af16cf 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -86,7 +86,9 @@ jobs: url="https://github.com/betterleaks/betterleaks/releases/download/v${{ env.BETTERLEAKS_VERSION }}/betterleaks_${{ env.BETTERLEAKS_VERSION }}_${os}_${arch}.tar.gz" echo "Downloading betterleaks from $url" curl -sSL "$url" | tar zxf - > "$outfile" - "$outfile" dir . --config ./.github/linters/betterleaks-config.toml -v + "$outfile" dir . -v \ + --config ./.github/linters/betterleaks-config.toml \ + --gitleaks-ignore-path ./.github/linters/betterleaks-ignore.txt go-lint: runs-on: ubuntu-latest