jrmccluskey commented on code in PR #38979: URL: https://github.com/apache/beam/pull/38979#discussion_r3723781857
########## sdks/go/pkg/beam/core/runtime/harness/sampler_test.go: ########## @@ -0,0 +1,80 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package harness + +import ( + "context" + "testing" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics" +) + +func TestNewSampler(t *testing.T) { + ctx := metrics.SetBundleID(context.Background(), "test-bundle") + store := metrics.GetStore(ctx) + if store == nil { + t.Fatal("GetStore returned nil") + } + + s := newSampler(store, 0) + if s == nil { + t.Fatal("newSampler returned nil") + } + if s.done == nil { + t.Error("sampler done channel is nil") + } +} + +func TestStateSampler_Stop(t *testing.T) { + ctx := metrics.SetBundleID(context.Background(), "test-bundle") + store := metrics.GetStore(ctx) + s := newSampler(store, 0) + // stop should not panic when called on a properly initialized sampler. + s.stop() +} + +func TestStateSampler_Start_ContextCancel(t *testing.T) { + ctx := metrics.SetBundleID(context.Background(), "test-bundle") + store := metrics.GetStore(ctx) + s := newSampler(store, 0) + + cancelCtx, cancel := context.WithCancel(ctx) + cancel() // Cancel immediately + + // start should return nil immediately when context is already canceled. + err := s.start(cancelCtx, samplePeriod) + if err != nil { + t.Errorf("start returned error on canceled context: %v", err) + } Review Comment: +1 ########## sdks/go/pkg/beam/core/runtime/harness/harness_test.go: ########## @@ -258,3 +277,112 @@ func TestElementProcessingTimeoutParsing(t *testing.T) { } } } + +func TestControl_MetStoreToString(t *testing.T) { + ctx := metrics.SetBundleID(context.Background(), "test-bundle") + store := metrics.GetStore(ctx) + if store == nil { + t.Fatal("GetStore returned nil") + } + ctrl := &control{ + metStore: map[instructionID]*metrics.Store{ + "inst1": store, + }, + } + b := &strings.Builder{} + ctrl.metStoreToString(b) + out := b.String() + if !strings.Contains(out, "Bundle ID: inst1") { + t.Errorf("metStoreToString output missing bundle ID, got: %s", out) + } +} + +func TestControl_GetPlanOrResponse(t *testing.T) { + tests := []struct { + name string + active map[instructionID]*exec.Plan + awaitFinalize map[instructionID]awaitingFinalization + failed map[instructionID]error + inactive circleBuffer + wantErr bool // response has Error field set (non-nil response) + wantNilPlan bool // response is non-nil and plan is nil -> response is returned + wantEmpty bool // both plan and response are nil - empty response needed + }{ + { + name: "active", + active: map[instructionID]*exec.Plan{ + "ref": {}, + }, + }, + { + name: "awaitingFinalization", + awaitFinalize: map[instructionID]awaitingFinalization{ + "ref": {plan: &exec.Plan{}}, + }, + }, + { + name: "failed", + failed: map[instructionID]error{"ref": fmt.Errorf("test failure")}, + wantErr: true, + }, + { + name: "inactive", + inactive: func() circleBuffer { + c := newCircleBuffer() + c.Add("ref") + return c + }(), + wantEmpty: true, + }, + { + name: "notFound", + wantErr: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctrl := &control{ + active: make(map[instructionID]*exec.Plan), + awaitingFinalization: make(map[instructionID]awaitingFinalization), + failed: make(map[instructionID]error), + inactive: newCircleBuffer(), + } + if test.active != nil { + ctrl.active = test.active + } + if test.awaitFinalize != nil { + ctrl.awaitingFinalization = test.awaitFinalize + } + if test.failed != nil { + ctrl.failed = test.failed + } + if len(test.inactive.buf) > 0 { + ctrl.inactive = test.inactive + } Review Comment: +1 ########## sdks/go/pkg/beam/core/runtime/harness/harness_test.go: ########## @@ -166,6 +167,24 @@ func TestControl_getOrCreatePlan(t *testing.T) { } +func TestFail(t *testing.T) { + ctx := context.Background() + resp := fail(ctx, "test-id", "error %s %d", "msg", 42) + + if resp == nil { + t.Fatal("fail returned nil") + } + if got, want := resp.GetInstructionId(), "test-id"; got != want { Review Comment: A check like this is somewhat inherently unhelpful since `fail()` is written to always instantiate an `InstructionResponse` struct, it will _never_ be `nil`. Checking the contents for correctness is valid afterwards though ########## sdks/go/pkg/beam/core/runtime/harness/statemgr_test.go: ########## @@ -508,3 +510,99 @@ func contains(got, want error) bool { } return strings.Contains(got.Error(), want.Error()) } + +func TestNewScopedStateReader(t *testing.T) { + mgr := &StateChannelManager{} + s := NewScopedStateReader(mgr, "inst1") + if s == nil { + t.Fatal("NewScopedStateReader returned nil") + } + if s.mgr != mgr { + t.Error("mgr field not set") + } + if s.instID != "inst1" { + t.Errorf("instID = %v, want inst1", s.instID) + } + if s.cache != nil { + t.Error("cache should be nil for NewScopedStateReader") + } +} + +func TestNewScopedStateReaderWithCache(t *testing.T) { + mgr := &StateChannelManager{} + cache := &statecache.SideInputCache{} + cache.Init(1) + s := NewScopedStateReaderWithCache(mgr, "inst2", cache) + if s == nil { + t.Fatal("NewScopedStateReaderWithCache returned nil") + } + if s.mgr != mgr { + t.Error("mgr field not set") + } + if s.instID != "inst2" { + t.Errorf("instID = %v, want inst2", s.instID) + } + if s.cache != cache { + t.Error("cache field not set correctly") + } +} + +func TestScopedStateReader_GetSideInputCache(t *testing.T) { + cache := &statecache.SideInputCache{} + cache.Init(1) + s := NewScopedStateReaderWithCache(&StateChannelManager{}, "inst", cache) + got := s.GetSideInputCache() + if got != cache { + t.Error("GetSideInputCache returned wrong cache") + } +} + +func TestScopedStateReader_Close(t *testing.T) { + mgr := &StateChannelManager{} + s := NewScopedStateReader(mgr, "inst") + err := s.Close() + if err != nil { + t.Errorf("Close returned error: %v", err) + } + if !s.closed { + t.Error("Close did not set closed flag") + } + // Second close should not error but should no-op. + err = s.Close() + if err != nil { + t.Errorf("Second Close returned error: %v", err) + } +} + +func TestStateChannelManager_PortsInit(t *testing.T) { + mgr := &StateChannelManager{} + if mgr.ports != nil { + t.Error("ports should be nil before first Open call") + } + // Force-initialize ports by attempting Open; even if it fails, the map should be created. + // Restore: just set it directly since we don't want to dial. + mgr.mu.Lock() + mgr.ports = make(map[string]*StateChannel) + mgr.mu.Unlock() + if mgr.ports == nil { + t.Error("ports map should exist after initialization") + } Review Comment: +1 ########## sdks/go/pkg/beam/core/runtime/harness/worker_status_test.go: ########## @@ -94,3 +94,74 @@ func TestSendStatusResponse(t *testing.T) { t.Error(err) } } + +func TestWorkerStatusHandler_IsAliveAndShutdown(t *testing.T) { + w := &workerStatusHandler{shouldShutdown: 0} + if !w.isAlive() { + t.Error("isAlive: expected true after init") + } + w.shutdown() + if w.isAlive() { + t.Error("isAlive: expected false after shutdown") + } +} + +func TestMemoryUsage(t *testing.T) { + b := &strings.Builder{} + memoryUsage(b) + out := b.String() + required := []string{"heap in-use-spans", "stack in-use-spans", "GC-CPU percentage", "Last GC time", "Next GC"} + for _, r := range required { + if !strings.Contains(out, r) { + t.Errorf("memoryUsage output missing %q, got:\n%s", r, out) + } + } +} + +func TestGoroutineDump(t *testing.T) { + b := &strings.Builder{} + goroutineDump(b) + out := b.String() + if !strings.Contains(out, "goroutine") { + t.Errorf("goroutineDump output missing 'goroutine', got:\n%s", out) + } +} + +func TestBuildInfo(t *testing.T) { + b := &strings.Builder{} + buildInfo(b) + out := b.String() + if !strings.Contains(out, "Build Info") { + t.Errorf("buildInfo output missing 'Build Info', got:\n%s", out) + } +} Review Comment: These aren't really testing interesting functionality here, just looking for the hard-coded headers from the functions. It doesn't feel particularly useful to have unit tests that will break if the string we write during the dump changes. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
