From 07dce613eb5ac760bf5316b606eef5664e595eb3 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:54:08 +1000 Subject: [PATCH] fix: replace fixed test waits with exact barriers and overridable deadlines --- .../DescribeInstanceAttribute_test.go | 14 ++ ...scribeInstanceCreditSpecifications_test.go | 5 + .../instance/DescribeInstanceStatus_test.go | 23 +++ .../instance/DescribeInstanceTypes_test.go | 9 + .../gateway/ec2/instance/DescribeInstances.go | 35 +++- .../ec2/instance/DescribeInstances_test.go | 29 +++- .../ec2/instance/GetPasswordData_test.go | 5 + .../IamInstanceProfileAssociation_test.go | 36 ++++ .../instance/ModifyInstanceAttribute_test.go | 18 ++ .../ModifyInstanceMetadataOptions_test.go | 1 + .../ec2/instance/MonitorInstances_test.go | 5 + .../ec2/instance/RebootInstances_test.go | 6 + .../RunInstancesIntoReservation_test.go | 14 +- .../gateway/ec2/instance/RunInstances_test.go | 13 ++ .../ec2/instance/StartInstances_test.go | 6 + .../ec2/instance/StopInstances_test.go | 6 + .../ec2/instance/TerminateInstances_test.go | 5 + .../gateway/ec2/instance/clienttoken_test.go | 10 ++ .../gateway/ec2/instance/placement_test.go | 81 +++++++-- spinifex/gateway/ec2/instance/state_test.go | 1 + spinifex/handlers/dns/zone_lookup_test.go | 7 +- .../handlers/ec2/routetable/helpers_test.go | 20 +++ .../routetable/lifecycle_invariants_test.go | 1 + .../ec2/routetable/record_tags_test.go | 2 + .../ec2/routetable/service_impl_test.go | 163 +++++++++++------- 25 files changed, 421 insertions(+), 94 deletions(-) diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceAttribute_test.go b/spinifex/gateway/ec2/instance/DescribeInstanceAttribute_test.go index 6692bf8c8..1f405393a 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstanceAttribute_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstanceAttribute_test.go @@ -18,12 +18,14 @@ import ( // --- Validation tests --- func TestValidateDescribeInstanceAttributeInput_NilInput(t *testing.T) { + t.Parallel() err := ValidateDescribeInstanceAttributeInput(nil) require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) } func TestValidateDescribeInstanceAttributeInput_MissingInstanceId(t *testing.T) { + t.Parallel() err := ValidateDescribeInstanceAttributeInput(&ec2.DescribeInstanceAttributeInput{ Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), }) @@ -32,6 +34,7 @@ func TestValidateDescribeInstanceAttributeInput_MissingInstanceId(t *testing.T) } func TestValidateDescribeInstanceAttributeInput_EmptyInstanceId(t *testing.T) { + t.Parallel() err := ValidateDescribeInstanceAttributeInput(&ec2.DescribeInstanceAttributeInput{ InstanceId: aws.String(""), Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), @@ -41,6 +44,7 @@ func TestValidateDescribeInstanceAttributeInput_EmptyInstanceId(t *testing.T) { } func TestValidateDescribeInstanceAttributeInput_BadPrefix(t *testing.T) { + t.Parallel() err := ValidateDescribeInstanceAttributeInput(&ec2.DescribeInstanceAttributeInput{ InstanceId: aws.String("x-12345"), Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), @@ -50,6 +54,7 @@ func TestValidateDescribeInstanceAttributeInput_BadPrefix(t *testing.T) { } func TestValidateDescribeInstanceAttributeInput_MissingAttribute(t *testing.T) { + t.Parallel() err := ValidateDescribeInstanceAttributeInput(&ec2.DescribeInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), }) @@ -58,6 +63,7 @@ func TestValidateDescribeInstanceAttributeInput_MissingAttribute(t *testing.T) { } func TestValidateDescribeInstanceAttributeInput_EmptyAttribute(t *testing.T) { + t.Parallel() err := ValidateDescribeInstanceAttributeInput(&ec2.DescribeInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), Attribute: aws.String(""), @@ -67,6 +73,7 @@ func TestValidateDescribeInstanceAttributeInput_EmptyAttribute(t *testing.T) { } func TestValidateDescribeInstanceAttributeInput_Valid(t *testing.T) { + t.Parallel() err := ValidateDescribeInstanceAttributeInput(&ec2.DescribeInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), @@ -100,6 +107,7 @@ func respondWithError(code string) nats.MsgHandler { } func TestDescribeInstanceAttribute_SingleNode(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceAttribute", respondWithInstance(t, "i-test123", "t3.micro")) @@ -125,6 +133,7 @@ func TestDescribeInstanceAttribute_SingleNode(t *testing.T) { // ~50% of the time. Post-fix, the aggregator drops the NotFound and surfaces // the success. func TestDescribeInstanceAttribute_MultipleNodes_OwnerWins(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceAttribute", respondWithError(awserrors.ErrorInvalidInstanceIDNotFound)) @@ -154,6 +163,7 @@ func TestDescribeInstanceAttribute_MultipleNodes_OwnerWins(t *testing.T) { // TestDescribeInstanceAttribute_AllNodesNotFound covers the genuine missing // instance case: every daemon confirms the instance is absent. func TestDescribeInstanceAttribute_AllNodesNotFound(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceAttribute", respondWithError(awserrors.ErrorInvalidInstanceIDNotFound)) @@ -182,6 +192,7 @@ func TestDescribeInstanceAttribute_AllNodesNotFound(t *testing.T) { // validation-class error (deterministic across daemons) is surfaced to the // caller rather than getting masked by NotFound. func TestDescribeInstanceAttribute_ClientErrorPropagates(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceAttribute", respondWithError(awserrors.ErrorInvalidParameterValue)) @@ -203,6 +214,7 @@ func TestDescribeInstanceAttribute_ClientErrorPropagates(t *testing.T) { // masked by sibling NotFound replies — otherwise terraform treats a live // instance as deleted. func TestDescribeInstanceAttribute_ServerErrorNotMaskedByNotFound(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceAttribute", respondWithError(awserrors.ErrorServerInternal)) @@ -230,6 +242,7 @@ func TestDescribeInstanceAttribute_ServerErrorNotMaskedByNotFound(t *testing.T) // TestDescribeInstanceAttribute_NoResponders returns NotFound (not a NATS // timeout) so terraform retries cleanly rather than hanging. func TestDescribeInstanceAttribute_NoResponders(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.DescribeInstanceAttributeInput{ @@ -248,6 +261,7 @@ func TestDescribeInstanceAttribute_NoResponders(t *testing.T) { } func TestDescribeInstanceAttribute_ValidationFailure(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := DescribeInstanceAttribute(context.Background(), nil, nc, 1, "123456789012") diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceCreditSpecifications_test.go b/spinifex/gateway/ec2/instance/DescribeInstanceCreditSpecifications_test.go index 9581cc0bb..2103f1978 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstanceCreditSpecifications_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstanceCreditSpecifications_test.go @@ -11,6 +11,7 @@ import ( ) func TestDescribeInstanceCreditSpecifications_NilInput(t *testing.T) { + t.Parallel() out, err := DescribeInstanceCreditSpecifications(nil) require.Error(t, err) assert.Nil(t, out) @@ -18,12 +19,14 @@ func TestDescribeInstanceCreditSpecifications_NilInput(t *testing.T) { } func TestDescribeInstanceCreditSpecifications_EmptyInstanceIds(t *testing.T) { + t.Parallel() out, err := DescribeInstanceCreditSpecifications(&ec2.DescribeInstanceCreditSpecificationsInput{}) require.NoError(t, err) assert.Empty(t, out.InstanceCreditSpecifications) } func TestDescribeInstanceCreditSpecifications_SingleInstance(t *testing.T) { + t.Parallel() out, err := DescribeInstanceCreditSpecifications(&ec2.DescribeInstanceCreditSpecificationsInput{ InstanceIds: []*string{aws.String("i-abc123")}, }) @@ -34,6 +37,7 @@ func TestDescribeInstanceCreditSpecifications_SingleInstance(t *testing.T) { } func TestDescribeInstanceCreditSpecifications_MultipleInstances(t *testing.T) { + t.Parallel() out, err := DescribeInstanceCreditSpecifications(&ec2.DescribeInstanceCreditSpecificationsInput{ InstanceIds: []*string{aws.String("i-aaa"), aws.String("i-bbb"), aws.String("i-ccc")}, }) @@ -45,6 +49,7 @@ func TestDescribeInstanceCreditSpecifications_MultipleInstances(t *testing.T) { } func TestDescribeInstanceCreditSpecifications_SkipsNilAndEmpty(t *testing.T) { + t.Parallel() out, err := DescribeInstanceCreditSpecifications(&ec2.DescribeInstanceCreditSpecificationsInput{ InstanceIds: []*string{nil, aws.String(""), aws.String("i-valid"), nil, aws.String("i-also")}, }) diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go b/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go index 7d35fa3e8..6145f2afd 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go @@ -42,6 +42,7 @@ func respondJSON(t *testing.T, msg *nats.Msg, payload any) { } func TestDescribeInstanceStatus_SingleNode(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { @@ -60,6 +61,7 @@ func TestDescribeInstanceStatus_SingleNode(t *testing.T) { } func TestDescribeInstanceStatus_TwoNodesDedup(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // First node returns i-001 (running) @@ -92,6 +94,7 @@ func TestDescribeInstanceStatus_TwoNodesDedup(t *testing.T) { } func TestDescribeInstanceStatus_OneNodeErrorOthersData(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Node 1: data @@ -121,6 +124,7 @@ func TestDescribeInstanceStatus_OneNodeErrorOthersData(t *testing.T) { } func TestDescribeInstanceStatus_AllNodesError(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { @@ -134,6 +138,7 @@ func TestDescribeInstanceStatus_AllNodesError(t *testing.T) { } func TestDescribeInstanceStatus_AllNodesTimeoutReturnsEmpty(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) out, err := DescribeInstanceStatus(context.Background(), &ec2.DescribeInstanceStatusInput{}, nc, 0, "123456789012", "az-a") @@ -143,6 +148,7 @@ func TestDescribeInstanceStatus_AllNodesTimeoutReturnsEmpty(t *testing.T) { } func TestDescribeInstanceStatus_IncludeAllAddsStoppedAsNotApplicable(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Fan-out responder: a running instance @@ -192,6 +198,7 @@ func TestDescribeInstanceStatus_IncludeAllAddsStoppedAsNotApplicable(t *testing. } func TestDescribeInstanceStatus_RunningWinsOverStoppedDuringRace(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Fan-out: i-001 is running @@ -225,6 +232,7 @@ func TestDescribeInstanceStatus_RunningWinsOverStoppedDuringRace(t *testing.T) { } func TestDescribeInstanceStatus_NilInput(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { @@ -239,6 +247,7 @@ func TestDescribeInstanceStatus_NilInput(t *testing.T) { } func TestDescribeInstanceStatus_ClosedConnection(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) closed, err := nats.Connect(nc.ConnectedUrl()) @@ -250,6 +259,7 @@ func TestDescribeInstanceStatus_ClosedConnection(t *testing.T) { } func TestBuildInstanceStatusFromInstance_PopulatesState(t *testing.T) { + t.Parallel() in := &ec2.Instance{ InstanceId: aws.String("i-x"), State: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, @@ -265,6 +275,7 @@ func TestBuildInstanceStatusFromInstance_PopulatesState(t *testing.T) { } func TestBuildInstanceStatusFromInstance_FallbackStateName(t *testing.T) { + t.Parallel() in := &ec2.Instance{InstanceId: aws.String("i-x")} got := buildInstanceStatusFromInstance(in, "az-a") assert.Equal(t, "stopped", *got.InstanceState.Name) @@ -275,6 +286,7 @@ func TestBuildInstanceStatusFromInstance_FallbackStateName(t *testing.T) { // because of the IncludeAllInstances field. The gateway must project to // DescribeInstancesInput before publishing. func TestDescribeInstanceStatus_IncludeAllProjectsInputForStoppedHandler(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { @@ -313,6 +325,7 @@ func TestDescribeInstanceStatus_IncludeAllProjectsInputForStoppedHandler(t *test } func TestStoppedCompatibleFilters_StripsStatusOnlyNames(t *testing.T) { + t.Parallel() filters := []*ec2.Filter{ {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-a")}}, {Name: aws.String("instance-state-code"), Values: []*string{aws.String("80")}}, @@ -326,11 +339,13 @@ func TestStoppedCompatibleFilters_StripsStatusOnlyNames(t *testing.T) { } func TestStoppedCompatibleFilters_Empty(t *testing.T) { + t.Parallel() assert.Nil(t, stoppedCompatibleFilters(nil)) assert.Nil(t, stoppedCompatibleFilters([]*ec2.Filter{})) } func TestFilterStoppedStatuses_AvailabilityZoneMatch(t *testing.T) { + t.Parallel() in := []*ec2.InstanceStatus{ {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, @@ -343,6 +358,7 @@ func TestFilterStoppedStatuses_AvailabilityZoneMatch(t *testing.T) { } func TestFilterStoppedStatuses_AvailabilityZoneMiss(t *testing.T) { + t.Parallel() in := []*ec2.InstanceStatus{ {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, @@ -355,6 +371,7 @@ func TestFilterStoppedStatuses_AvailabilityZoneMiss(t *testing.T) { } func TestFilterStoppedStatuses_StateCodeMatch(t *testing.T) { + t.Parallel() in := []*ec2.InstanceStatus{ {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, @@ -367,6 +384,7 @@ func TestFilterStoppedStatuses_StateCodeMatch(t *testing.T) { } func TestFilterStoppedStatuses_StateCodeMiss(t *testing.T) { + t.Parallel() in := []*ec2.InstanceStatus{ {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, @@ -379,12 +397,14 @@ func TestFilterStoppedStatuses_StateCodeMiss(t *testing.T) { } func TestFilterStoppedStatuses_NoFiltersPassthrough(t *testing.T) { + t.Parallel() in := []*ec2.InstanceStatus{{InstanceId: aws.String("i-1")}} got := filterStoppedStatuses(in, nil, "az-a") assert.Equal(t, in, got) } func TestFilterStoppedStatuses_UnknownFilterIgnored(t *testing.T) { + t.Parallel() in := []*ec2.InstanceStatus{ {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, @@ -399,6 +419,7 @@ func TestFilterStoppedStatuses_UnknownFilterIgnored(t *testing.T) { } func TestDescribeInstanceStatus_IncludeAllWithAZFilterMatches(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { @@ -429,6 +450,7 @@ func TestDescribeInstanceStatus_IncludeAllWithAZFilterMatches(t *testing.T) { } func TestDescribeInstanceStatus_IncludeAllWithAZFilterMisses(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { @@ -458,6 +480,7 @@ func TestDescribeInstanceStatus_IncludeAllWithAZFilterMisses(t *testing.T) { } func TestDedupStatuses_FirstWins(t *testing.T) { + t.Parallel() first := runningStatus("i-001", "az-a") second := &ec2.InstanceStatus{ InstanceId: aws.String("i-001"), diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceTypes_test.go b/spinifex/gateway/ec2/instance/DescribeInstanceTypes_test.go index 1d6f4aef8..d920f5b15 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstanceTypes_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstanceTypes_test.go @@ -14,6 +14,7 @@ import ( ) func TestDescribeInstanceTypes_SingleNode(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstanceTypes", func(msg *nats.Msg) { @@ -37,6 +38,7 @@ func TestDescribeInstanceTypes_SingleNode(t *testing.T) { } func TestDescribeInstanceTypes_DeduplicatesAcrossNodes(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Node 1 @@ -86,6 +88,7 @@ func TestDescribeInstanceTypes_DeduplicatesAcrossNodes(t *testing.T) { } func TestDescribeInstanceTypes_CapacityFilterShowsDuplicates(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Node 1 @@ -133,6 +136,7 @@ func TestDescribeInstanceTypes_CapacityFilterShowsDuplicates(t *testing.T) { } func TestDescribeInstanceTypes_CapacityFilterFalseDeduplicates(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstanceTypes", func(msg *nats.Msg) { @@ -162,6 +166,7 @@ func TestDescribeInstanceTypes_CapacityFilterFalseDeduplicates(t *testing.T) { } func TestDescribeInstanceTypes_NoSubscribers(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.DescribeInstanceTypesInput{} @@ -173,6 +178,7 @@ func TestDescribeInstanceTypes_NoSubscribers(t *testing.T) { } func TestDescribeInstanceTypes_NodeReturnsError(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstanceTypes", func(msg *nats.Msg) { @@ -189,6 +195,7 @@ func TestDescribeInstanceTypes_NodeReturnsError(t *testing.T) { } func TestDescribeInstanceTypes_MalformedJSON(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstanceTypes", func(msg *nats.Msg) { @@ -204,6 +211,7 @@ func TestDescribeInstanceTypes_MalformedJSON(t *testing.T) { } func TestDescribeInstanceTypes_NilInstanceTypeSkipped(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstanceTypes", func(msg *nats.Msg) { @@ -227,6 +235,7 @@ func TestDescribeInstanceTypes_NilInstanceTypeSkipped(t *testing.T) { } func TestDescribeInstanceTypes_ClosedConnection(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) closedNC, err := nats.Connect(nc.ConnectedUrl()) diff --git a/spinifex/gateway/ec2/instance/DescribeInstances.go b/spinifex/gateway/ec2/instance/DescribeInstances.go index 774a501bc..eb33b953b 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstances.go +++ b/spinifex/gateway/ec2/instance/DescribeInstances.go @@ -17,6 +17,24 @@ import ( "github.com/nats-io/nats.go" ) +// defaultDescribeFanoutTimeout is the hard deadline the describe fan-out waits +// for node replies before treating the sweep as incomplete. +const defaultDescribeFanoutTimeout = 3 * time.Second + +// describeConfig holds the tunable knobs of the describe fan-out. +type describeConfig struct { + fanoutTimeout time.Duration +} + +// DescribeOption customises a describe fan-out. Production callers pass none +// and get defaultDescribeFanoutTimeout. +type DescribeOption func(*describeConfig) + +// WithFanoutTimeout overrides the fan-out deadline. +func WithFanoutTimeout(d time.Duration) DescribeOption { + return func(c *describeConfig) { c.fanoutTimeout = d } +} + // DescribeInstances fans out to all nodes via NATS and aggregates the results. // It is lenient: partial results from a slow or unreachable node are returned // without error, and an explicitly named instance ID that is simply absent @@ -28,8 +46,8 @@ import ( // which adds the InvalidInstanceID.NotFound assertion AWS callers expect. // Callers that must not act on a partial view at all (the quota reconcile) // use DescribeInstancesForReconcile. -func DescribeInstances(ctx context.Context, input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int, accountID string) (*ec2.DescribeInstancesOutput, error) { - reservations, _, firstClient4xx, err := gatherInstances(ctx, input, natsConn, expectedNodes, accountID) +func DescribeInstances(ctx context.Context, input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int, accountID string, opts ...DescribeOption) (*ec2.DescribeInstancesOutput, error) { + reservations, _, firstClient4xx, err := gatherInstances(ctx, input, natsConn, expectedNodes, accountID, opts...) if err != nil { return nil, err } @@ -49,8 +67,8 @@ func DescribeInstances(ctx context.Context, input *ec2.DescribeInstancesInput, n // fan-out must never turn into a false NotFound for an instance that actually // exists on that node. A --filters-only query (no explicit IDs) is never // affected and keeps returning an empty list with no error. -func DescribeInstancesChecked(ctx context.Context, input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int, accountID string) (*ec2.DescribeInstancesOutput, error) { - reservations, complete, firstClient4xx, err := gatherInstances(ctx, input, natsConn, expectedNodes, accountID) +func DescribeInstancesChecked(ctx context.Context, input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int, accountID string, opts ...DescribeOption) (*ec2.DescribeInstancesOutput, error) { + reservations, complete, firstClient4xx, err := gatherInstances(ctx, input, natsConn, expectedNodes, accountID, opts...) if err != nil { return nil, err } @@ -95,7 +113,12 @@ func DescribeInstancesForReconcile(ctx context.Context, input *ec2.DescribeInsta // both bucket queries succeeded — the precondition reconcile needs before it may // lower a counter. firstClient4xx carries the first deterministic 4xx for the // lenient caller to surface when nothing was collected. -func gatherInstances(ctx context.Context, input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int, accountID string) (reservations []*ec2.Reservation, complete bool, firstClient4xx string, err error) { +func gatherInstances(ctx context.Context, input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int, accountID string, opts ...DescribeOption) (reservations []*ec2.Reservation, complete bool, firstClient4xx string, err error) { + cfg := describeConfig{fanoutTimeout: defaultDescribeFanoutTimeout} + for _, opt := range opts { + opt(&cfg) + } + jsonData, err := json.Marshal(input) if err != nil { slog.ErrorContext(ctx, "DescribeInstances: Failed to marshal input", "err", err) @@ -103,7 +126,7 @@ func gatherInstances(ctx context.Context, input *ec2.DescribeInstancesInput, nat } frames, sum, err := utils.Gather(ctx, natsConn, "ec2.DescribeInstances", jsonData, - utils.GatherOpts{Timeout: 3 * time.Second, ExpectedNodes: expectedNodes, AccountID: accountID}) + utils.GatherOpts{Timeout: cfg.fanoutTimeout, ExpectedNodes: expectedNodes, AccountID: accountID}) if err != nil { return nil, false, "", err } diff --git a/spinifex/gateway/ec2/instance/DescribeInstances_test.go b/spinifex/gateway/ec2/instance/DescribeInstances_test.go index 3f90a0496..94dfc9b18 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstances_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstances_test.go @@ -18,6 +18,7 @@ import ( ) func TestDescribeInstances_SingleNode(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) reservation := &ec2.Reservation{ @@ -50,6 +51,7 @@ func TestDescribeInstances_SingleNode(t *testing.T) { } func TestDescribeInstances_MultipleNodes(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Two nodes each return different instances @@ -99,6 +101,7 @@ func TestDescribeInstances_MultipleNodes(t *testing.T) { } func TestDescribeInstances_NoSubscribers(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.DescribeInstancesInput{} @@ -111,6 +114,7 @@ func TestDescribeInstances_NoSubscribers(t *testing.T) { } func TestDescribeInstances_NodeReturnsError(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { @@ -128,6 +132,7 @@ func TestDescribeInstances_NodeReturnsError(t *testing.T) { } func TestDescribeInstances_MixedResponses(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Node 1: returns valid data @@ -167,6 +172,7 @@ func TestDescribeInstances_MixedResponses(t *testing.T) { } func TestDescribeInstances_MalformedJSON(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { @@ -183,6 +189,7 @@ func TestDescribeInstances_MalformedJSON(t *testing.T) { } func TestDescribeInstances_EmptyReservations(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { @@ -201,11 +208,12 @@ func TestDescribeInstances_EmptyReservations(t *testing.T) { } func TestDescribeInstances_TimeoutCollection(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Node responds after a delay (but within timeout) nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { - time.Sleep(500 * time.Millisecond) + time.Sleep(50 * time.Millisecond) data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ Reservations: []*ec2.Reservation{ {ReservationId: aws.String("r-delayed")}, @@ -215,7 +223,9 @@ func TestDescribeInstances_TimeoutCollection(t *testing.T) { }) input := &ec2.DescribeInstancesInput{} - output, err := DescribeInstances(context.Background(), input, nc, 0, "123456789012") // 0 = timeout-based collection + // 0 expected nodes = timeout-based collection: the whole window is waited out. + output, err := DescribeInstances(context.Background(), input, nc, 0, "123456789012", + WithFanoutTimeout(300*time.Millisecond)) require.NoError(t, err) require.NotNil(t, output) @@ -223,6 +233,7 @@ func TestDescribeInstances_TimeoutCollection(t *testing.T) { } func TestDescribeInstances_EarlyExitWithExpectedNodes(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { @@ -267,6 +278,7 @@ func subscribeEmptyInstanceBuckets(t *testing.T, nc *nats.Conn) { // fully-answered fan-out returns InvalidInstanceID.NotFound instead of a // silent empty list. func TestDescribeInstancesChecked_ExplicitIDNotFound_CompleteSweep(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -297,6 +309,7 @@ func TestDescribeInstancesChecked_ExplicitIDNotFound_CompleteSweep(t *testing.T) // meaning an empty list, not an error. Only DescribeInstancesChecked — used // solely by the gateway's customer-facing action — may assert NotFound. func TestDescribeInstances_ExplicitIDNotFound_StaysSilent(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -321,6 +334,7 @@ func TestDescribeInstances_ExplicitIDNotFound_StaysSilent(t *testing.T) { // counterpart: the same complete sweep, but the requested ID is present, so // no error is raised. func TestDescribeInstancesChecked_ExplicitIDFound_CompleteSweep(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -350,6 +364,7 @@ func TestDescribeInstancesChecked_ExplicitIDFound_CompleteSweep(t *testing.T) { // must still return rc=0 with an empty list, never NotFound (no instance IDs // were named). func TestDescribeInstancesChecked_FilterOnlyNoMatch_ReturnsEmptyNotNotFound(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -376,11 +391,12 @@ func TestDescribeInstancesChecked_FilterOnlyNoMatch_ReturnsEmptyNotNotFound(t *t // sweep is provably incomplete and the naive found-set check must be // suppressed. func TestDescribeInstancesChecked_ExplicitIDNotFound_NodeTimeout(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) // Only one of the two expected nodes has a subscriber; the second never - // answers, so the fan-out hits its 3s deadline with sum.TimedOut=true. + // answers, so the fan-out hits its deadline with sum.TimedOut=true. _, err := nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { data, _ := json.Marshal(&ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{}}) msg.Respond(data) @@ -389,7 +405,8 @@ func TestDescribeInstancesChecked_ExplicitIDNotFound_NodeTimeout(t *testing.T) { nc.Flush() input := &ec2.DescribeInstancesInput{InstanceIds: []*string{aws.String("i-on-the-slow-node")}} - output, err := DescribeInstancesChecked(context.Background(), input, nc, 2, "123456789012") + output, err := DescribeInstancesChecked(context.Background(), input, nc, 2, "123456789012", + WithFanoutTimeout(300*time.Millisecond)) require.NoError(t, err, "an incomplete sweep must never assert a false NotFound") require.NotNil(t, output) @@ -397,6 +414,7 @@ func TestDescribeInstancesChecked_ExplicitIDNotFound_NodeTimeout(t *testing.T) { } func TestDescribeInstances_ClosedConnection(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) closedNC, err := nats.Connect(nc.ConnectedUrl()) @@ -427,6 +445,7 @@ func describeOutputWithProfiles(arns ...string) *ec2.DescribeInstancesOutput { } func TestEnrichInstanceProfileIDs_CachesPerARN(t *testing.T) { + t.Parallel() arn := "arn:aws:iam::123456789012:instance-profile/shared" var calls int svc := &fakeIAMService{ @@ -446,6 +465,7 @@ func TestEnrichInstanceProfileIDs_CachesPerARN(t *testing.T) { } func TestEnrichInstanceProfileIDs_ResolveErrorLeavesIdEmpty(t *testing.T) { + t.Parallel() failingARN := "arn:aws:iam::123456789012:instance-profile/missing" okARN := "arn:aws:iam::123456789012:instance-profile/present" svc := &fakeIAMService{ @@ -467,6 +487,7 @@ func TestEnrichInstanceProfileIDs_ResolveErrorLeavesIdEmpty(t *testing.T) { } func TestEnrichInstanceProfileIDs_NoOpInputs(t *testing.T) { + t.Parallel() // Should not panic on nil out or nil iamSvc, and instances without a // profile attached are untouched. EnrichInstanceProfileIDs(nil, &fakeIAMService{}, "123456789012") diff --git a/spinifex/gateway/ec2/instance/GetPasswordData_test.go b/spinifex/gateway/ec2/instance/GetPasswordData_test.go index 31e135787..649b594e8 100644 --- a/spinifex/gateway/ec2/instance/GetPasswordData_test.go +++ b/spinifex/gateway/ec2/instance/GetPasswordData_test.go @@ -26,6 +26,7 @@ func shrinkPasswordDataTimeout(t *testing.T) { } func TestValidateGetPasswordDataInput(t *testing.T) { + t.Parallel() tests := []struct { name string input *ec2.GetPasswordDataInput @@ -66,6 +67,7 @@ func TestValidateGetPasswordDataInput(t *testing.T) { } func TestGetPasswordData_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-0123456789abcdef0" @@ -108,6 +110,7 @@ func TestGetPasswordData_Success(t *testing.T) { } func TestGetPasswordData_NoResponders(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // No subscriber on the topic: an unknown/stopped/terminated instance. @@ -140,6 +143,7 @@ func TestGetPasswordData_Timeout(t *testing.T) { } func TestGetPasswordData_DaemonError(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-error" @@ -158,6 +162,7 @@ func TestGetPasswordData_DaemonError(t *testing.T) { } func TestGetPasswordData_MalformedResponse(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-malformed" diff --git a/spinifex/gateway/ec2/instance/IamInstanceProfileAssociation_test.go b/spinifex/gateway/ec2/instance/IamInstanceProfileAssociation_test.go index 20afe8fdc..ab0cae354 100644 --- a/spinifex/gateway/ec2/instance/IamInstanceProfileAssociation_test.go +++ b/spinifex/gateway/ec2/instance/IamInstanceProfileAssociation_test.go @@ -76,6 +76,7 @@ func profileNoRole() *handlers_iam.InstanceProfile { // --- helper-level tests -------------------------------------------------- func TestResolveAndAuthorizeProfile_NameForm(t *testing.T) { + t.Parallel() gotNameOrARN := "" svc := &fakeIAMService{resolveFn: func(_, nameOrARN string) (*handlers_iam.InstanceProfile, error) { gotNameOrARN = nameOrARN @@ -99,6 +100,7 @@ func TestResolveAndAuthorizeProfile_NameForm(t *testing.T) { } func TestResolveAndAuthorizeProfile_ArnForm(t *testing.T) { + t.Parallel() gotNameOrARN := "" svc := &fakeIAMService{resolveFn: func(_, nameOrARN string) (*handlers_iam.InstanceProfile, error) { gotNameOrARN = nameOrARN @@ -112,6 +114,7 @@ func TestResolveAndAuthorizeProfile_ArnForm(t *testing.T) { } func TestResolveAndAuthorizeProfile_MissingNameAndArn(t *testing.T) { + t.Parallel() svc := &fakeIAMService{} _, err := resolveAndAuthorizeProfile( &ec2.IamInstanceProfileSpecification{}, @@ -121,6 +124,7 @@ func TestResolveAndAuthorizeProfile_MissingNameAndArn(t *testing.T) { } func TestResolveAndAuthorizeProfile_NilIAMService(t *testing.T) { + t.Parallel() _, err := resolveAndAuthorizeProfile( &ec2.IamInstanceProfileSpecification{Name: aws.String(testProfileNameApp)}, nil, testGwAccountID, nil) @@ -129,6 +133,7 @@ func TestResolveAndAuthorizeProfile_NilIAMService(t *testing.T) { } func TestResolveAndAuthorizeProfile_NotFoundMapsToEC2Error(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return nil, errors.New(awserrors.ErrorIAMNoSuchEntity) }} @@ -141,6 +146,7 @@ func TestResolveAndAuthorizeProfile_NotFoundMapsToEC2Error(t *testing.T) { } func TestResolveAndAuthorizeProfile_PassthroughOtherIAMError(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return nil, errors.New(awserrors.ErrorAccessDenied) }} @@ -153,6 +159,7 @@ func TestResolveAndAuthorizeProfile_PassthroughOtherIAMError(t *testing.T) { } func TestResolveAndAuthorizeProfile_PassRoleDenied(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileWithRole(), nil }} @@ -165,6 +172,7 @@ func TestResolveAndAuthorizeProfile_PassRoleDenied(t *testing.T) { } func TestResolveAndAuthorizeProfile_NoRoleSkipsPassRole(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileNoRole(), nil }} @@ -183,6 +191,7 @@ func TestResolveAndAuthorizeProfile_NoRoleSkipsPassRole(t *testing.T) { } func TestEnrichProfileID_FillsIdFromResolvedProfile(t *testing.T) { + t.Parallel() assoc := &ec2.IamInstanceProfileAssociation{ IamInstanceProfile: &ec2.IamInstanceProfile{Arn: aws.String(testProfileARNApp)}, } @@ -193,10 +202,12 @@ func TestEnrichProfileID_FillsIdFromResolvedProfile(t *testing.T) { } func TestEnrichProfileID_NilAssociationIsNoOp(t *testing.T) { + t.Parallel() enrichProfileID(nil, profileWithRole()) // must not panic } func TestEnrichProfileID_NilInnerProfileIsAllocated(t *testing.T) { + t.Parallel() assoc := &ec2.IamInstanceProfileAssociation{} enrichProfileID(assoc, profileWithRole()) require.NotNil(t, assoc.IamInstanceProfile) @@ -206,12 +217,14 @@ func TestEnrichProfileID_NilInnerProfileIsAllocated(t *testing.T) { // --- AssociateIamInstanceProfile --------------------------------------------- func TestAssociateIamInstanceProfile_NilInput(t *testing.T) { + t.Parallel() _, err := AssociateIamInstanceProfile(context.Background(), nil, nil, &fakeIAMService{}, testGwAccountID, nil) require.Error(t, err) assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) } func TestAssociateIamInstanceProfile_MissingInstanceID(t *testing.T) { + t.Parallel() in := &ec2.AssociateIamInstanceProfileInput{ IamInstanceProfile: &ec2.IamInstanceProfileSpecification{Name: aws.String(testProfileNameApp)}, } @@ -221,6 +234,7 @@ func TestAssociateIamInstanceProfile_MissingInstanceID(t *testing.T) { } func TestAssociateIamInstanceProfile_MissingProfileSpec(t *testing.T) { + t.Parallel() in := &ec2.AssociateIamInstanceProfileInput{InstanceId: aws.String("i-001")} _, err := AssociateIamInstanceProfile(context.Background(), in, nil, &fakeIAMService{}, testGwAccountID, nil) require.Error(t, err) @@ -228,6 +242,7 @@ func TestAssociateIamInstanceProfile_MissingProfileSpec(t *testing.T) { } func TestAssociateIamInstanceProfile_ProfileNotFound(t *testing.T) { + t.Parallel() svc := &fakeIAMService{} // default resolve → NoSuchEntity in := &ec2.AssociateIamInstanceProfileInput{ InstanceId: aws.String("i-001"), @@ -239,6 +254,7 @@ func TestAssociateIamInstanceProfile_ProfileNotFound(t *testing.T) { } func TestAssociateIamInstanceProfile_PassRoleDenied(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileWithRole(), nil }} @@ -253,6 +269,7 @@ func TestAssociateIamInstanceProfile_PassRoleDenied(t *testing.T) { } func TestAssociateIamInstanceProfile_NoResponders(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileNoRole(), nil @@ -270,6 +287,7 @@ func TestAssociateIamInstanceProfile_NoResponders(t *testing.T) { } func TestAssociateIamInstanceProfile_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileWithRole(), nil @@ -315,6 +333,7 @@ func TestAssociateIamInstanceProfile_Success(t *testing.T) { } func TestAssociateIamInstanceProfile_DaemonAlreadyAssociated(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileNoRole(), nil @@ -336,18 +355,21 @@ func TestAssociateIamInstanceProfile_DaemonAlreadyAssociated(t *testing.T) { // --- DisassociateIamInstanceProfile ----------------------------------------- func TestDisassociateIamInstanceProfile_MissingAssociationID(t *testing.T) { + t.Parallel() _, err := DisassociateIamInstanceProfile(context.Background(), &ec2.DisassociateIamInstanceProfileInput{}, nil, 0, testGwAccountID) require.Error(t, err) assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) } func TestDisassociateIamInstanceProfile_NilInput(t *testing.T) { + t.Parallel() _, err := DisassociateIamInstanceProfile(context.Background(), nil, nil, 0, testGwAccountID) require.Error(t, err) assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) } func TestDisassociateIamInstanceProfile_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) const assocID = "iip-assoc-disassoc00000001" const instanceID = "i-dis-success" @@ -396,6 +418,7 @@ func TestDisassociateIamInstanceProfile_Success(t *testing.T) { } func TestDisassociateIamInstanceProfile_NoSuchAssociation(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Every daemon NoOps with JSON null — gateway must surface // NoSuchAssociation, not timeout. @@ -414,12 +437,14 @@ func TestDisassociateIamInstanceProfile_NoSuchAssociation(t *testing.T) { // --- ReplaceIamInstanceProfileAssociation ----------------------------------- func TestReplaceIamInstanceProfileAssociation_NilInput(t *testing.T) { + t.Parallel() _, err := ReplaceIamInstanceProfileAssociation(context.Background(), nil, nil, &fakeIAMService{}, 0, testGwAccountID, nil) require.Error(t, err) assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) } func TestReplaceIamInstanceProfileAssociation_MissingAssociationID(t *testing.T) { + t.Parallel() in := &ec2.ReplaceIamInstanceProfileAssociationInput{ IamInstanceProfile: &ec2.IamInstanceProfileSpecification{Name: aws.String(testProfileNameApp)}, } @@ -429,6 +454,7 @@ func TestReplaceIamInstanceProfileAssociation_MissingAssociationID(t *testing.T) } func TestReplaceIamInstanceProfileAssociation_MissingProfileSpec(t *testing.T) { + t.Parallel() in := &ec2.ReplaceIamInstanceProfileAssociationInput{AssociationId: aws.String("iip-assoc-x")} _, err := ReplaceIamInstanceProfileAssociation(context.Background(), in, nil, &fakeIAMService{}, 0, testGwAccountID, nil) require.Error(t, err) @@ -436,6 +462,7 @@ func TestReplaceIamInstanceProfileAssociation_MissingProfileSpec(t *testing.T) { } func TestReplaceIamInstanceProfileAssociation_PassRoleDeniedBeforeBroadcast(t *testing.T) { + t.Parallel() // nc=nil proves the call is rejected before any NATS dispatch. svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileWithRole(), nil @@ -450,6 +477,7 @@ func TestReplaceIamInstanceProfileAssociation_PassRoleDeniedBeforeBroadcast(t *t } func TestReplaceIamInstanceProfileAssociation_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileWithRole(), nil @@ -490,6 +518,7 @@ func TestReplaceIamInstanceProfileAssociation_Success(t *testing.T) { } func TestReplaceIamInstanceProfileAssociation_NoSuchAssociation(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileNoRole(), nil @@ -510,6 +539,7 @@ func TestReplaceIamInstanceProfileAssociation_NoSuchAssociation(t *testing.T) { // --- DescribeIamInstanceProfileAssociations --------------------------------- func TestDescribeIamInstanceProfileAssociations_FanOutAggregates(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Two daemons each return one record; the gateway must concatenate them. _, err := nc.Subscribe("ec2.IamProfileAssociation.describe", func(msg *nats.Msg) { @@ -550,6 +580,7 @@ func TestDescribeIamInstanceProfileAssociations_FanOutAggregates(t *testing.T) { } func TestDescribeIamInstanceProfileAssociations_ForwardsFilters(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) var gotInput ec2.DescribeIamInstanceProfileAssociationsInput _, err := nc.Subscribe("ec2.IamProfileAssociation.describe", func(msg *nats.Msg) { @@ -575,6 +606,7 @@ func TestDescribeIamInstanceProfileAssociations_ForwardsFilters(t *testing.T) { } func TestDescribeIamInstanceProfileAssociations_InvalidFilterName(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := DescribeIamInstanceProfileAssociations(context.Background(), &ec2.DescribeIamInstanceProfileAssociationsInput{ Filters: []*ec2.Filter{{Name: aws.String("not-a-valid-filter")}}, @@ -584,6 +616,7 @@ func TestDescribeIamInstanceProfileAssociations_InvalidFilterName(t *testing.T) } func TestDescribeIamInstanceProfileAssociations_EmptyResultIsValid(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.IamProfileAssociation.describe", func(msg *nats.Msg) { resp, _ := json.Marshal(&ec2.DescribeIamInstanceProfileAssociationsOutput{}) @@ -599,6 +632,7 @@ func TestDescribeIamInstanceProfileAssociations_EmptyResultIsValid(t *testing.T) // --- CountInstanceProfileAssociations --------------------------------------- func TestCountInstanceProfileAssociations_MatchesByARN(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.IamProfileAssociation.describe", func(msg *nats.Msg) { resp, _ := json.Marshal(&ec2.DescribeIamInstanceProfileAssociationsOutput{ @@ -618,6 +652,7 @@ func TestCountInstanceProfileAssociations_MatchesByARN(t *testing.T) { } func TestCountInstanceProfileAssociations_NoMatches(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := nc.Subscribe("ec2.IamProfileAssociation.describe", func(msg *nats.Msg) { resp, _ := json.Marshal(&ec2.DescribeIamInstanceProfileAssociationsOutput{ @@ -640,6 +675,7 @@ func TestCountInstanceProfileAssociations_NoMatches(t *testing.T) { // (which produces it) and clients (who parse it via SDKs); regressions in the // utils generator would break replay parsing of Disassociate/Replace. func TestAssociationIDFormat(t *testing.T) { + t.Parallel() id := utils.GenerateResourceID("iip-assoc") assert.True(t, strings.HasPrefix(id, "iip-assoc-"), "GenerateResourceID must preserve the AWS-style hyphen between prefix and suffix") diff --git a/spinifex/gateway/ec2/instance/ModifyInstanceAttribute_test.go b/spinifex/gateway/ec2/instance/ModifyInstanceAttribute_test.go index 5f980eb7e..843049b65 100644 --- a/spinifex/gateway/ec2/instance/ModifyInstanceAttribute_test.go +++ b/spinifex/gateway/ec2/instance/ModifyInstanceAttribute_test.go @@ -16,18 +16,21 @@ import ( // --- Validation tests --- func TestValidateModifyInstanceAttributeInput_NilInput(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(nil) require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) } func TestValidateModifyInstanceAttributeInput_MissingInstanceId(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{}) require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidInstanceIDMalformed, err.Error()) } func TestValidateModifyInstanceAttributeInput_EmptyInstanceId(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String(""), }) @@ -36,6 +39,7 @@ func TestValidateModifyInstanceAttributeInput_EmptyInstanceId(t *testing.T) { } func TestValidateModifyInstanceAttributeInput_BadPrefix(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("x-12345"), InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")}, @@ -45,6 +49,7 @@ func TestValidateModifyInstanceAttributeInput_BadPrefix(t *testing.T) { } func TestValidateModifyInstanceAttributeInput_NoAttributeSet(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), }) @@ -53,6 +58,7 @@ func TestValidateModifyInstanceAttributeInput_NoAttributeSet(t *testing.T) { } func TestValidateModifyInstanceAttributeInput_MultipleAttributes(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")}, @@ -63,6 +69,7 @@ func TestValidateModifyInstanceAttributeInput_MultipleAttributes(t *testing.T) { } func TestValidateModifyInstanceAttributeInput_EmptyInstanceType(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), InstanceType: &ec2.AttributeValue{Value: aws.String("")}, @@ -72,6 +79,7 @@ func TestValidateModifyInstanceAttributeInput_EmptyInstanceType(t *testing.T) { } func TestValidateModifyInstanceAttributeInput_NilInstanceTypeValue(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), InstanceType: &ec2.AttributeValue{}, @@ -81,6 +89,7 @@ func TestValidateModifyInstanceAttributeInput_NilInstanceTypeValue(t *testing.T) } func TestValidateModifyInstanceAttributeInput_ValidInstanceType(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")}, @@ -89,6 +98,7 @@ func TestValidateModifyInstanceAttributeInput_ValidInstanceType(t *testing.T) { } func TestValidateModifyInstanceAttributeInput_ValidUserData(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), UserData: &ec2.BlobAttributeValue{Value: []byte("IyEvYmluL2Jhc2g=")}, @@ -97,6 +107,7 @@ func TestValidateModifyInstanceAttributeInput_ValidUserData(t *testing.T) { } func TestValidateModifyInstanceAttributeInput_ValidDisableApiTermination(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(true)}, @@ -105,6 +116,7 @@ func TestValidateModifyInstanceAttributeInput_ValidDisableApiTermination(t *test } func TestValidateModifyInstanceAttributeInput_DisableApiTerminationWithOther(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(true)}, @@ -115,6 +127,7 @@ func TestValidateModifyInstanceAttributeInput_DisableApiTerminationWithOther(t * } func TestValidateModifyInstanceAttributeInput_ValidSourceDestCheck(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), SourceDestCheck: &ec2.AttributeBooleanValue{Value: aws.Bool(true)}, @@ -123,6 +136,7 @@ func TestValidateModifyInstanceAttributeInput_ValidSourceDestCheck(t *testing.T) } func TestValidateModifyInstanceAttributeInput_SourceDestCheckWithOtherAttribute(t *testing.T) { + t.Parallel() err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{ InstanceId: aws.String("i-abc123"), SourceDestCheck: &ec2.AttributeBooleanValue{Value: aws.Bool(false)}, @@ -135,6 +149,7 @@ func TestValidateModifyInstanceAttributeInput_SourceDestCheckWithOtherAttribute( // --- Gateway function tests --- func TestModifyInstanceAttribute_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", func(msg *nats.Msg) { @@ -155,6 +170,7 @@ func TestModifyInstanceAttribute_Success(t *testing.T) { } func TestModifyInstanceAttribute_SourceDestCheck(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", func(msg *nats.Msg) { @@ -171,6 +187,7 @@ func TestModifyInstanceAttribute_SourceDestCheck(t *testing.T) { } func TestModifyInstanceAttribute_DaemonError(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) nc.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", func(msg *nats.Msg) { @@ -188,6 +205,7 @@ func TestModifyInstanceAttribute_DaemonError(t *testing.T) { } func TestModifyInstanceAttribute_ValidationFailure(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := ModifyInstanceAttribute(context.Background(), nil, nc, "123456789012") diff --git a/spinifex/gateway/ec2/instance/ModifyInstanceMetadataOptions_test.go b/spinifex/gateway/ec2/instance/ModifyInstanceMetadataOptions_test.go index 1bbe4ce3a..44759fa48 100644 --- a/spinifex/gateway/ec2/instance/ModifyInstanceMetadataOptions_test.go +++ b/spinifex/gateway/ec2/instance/ModifyInstanceMetadataOptions_test.go @@ -14,6 +14,7 @@ import ( // A nil, empty, or non-"i-" instance ID is rejected at the gateway before any // NATS round-trip, so a nil connection is never dereferenced. func TestModifyInstanceMetadataOptions_RejectsMalformedID(t *testing.T) { + t.Parallel() cases := map[string]*ec2.ModifyInstanceMetadataOptionsInput{ "nil input": nil, "nil id": {}, diff --git a/spinifex/gateway/ec2/instance/MonitorInstances_test.go b/spinifex/gateway/ec2/instance/MonitorInstances_test.go index e707fcf4d..ee072e67d 100644 --- a/spinifex/gateway/ec2/instance/MonitorInstances_test.go +++ b/spinifex/gateway/ec2/instance/MonitorInstances_test.go @@ -39,6 +39,7 @@ func serveMonitoring(t *testing.T, nc *nats.Conn, subject, state string, ids ... } func TestMonitorInstances_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) seen := serveMonitoring(t, nc, "ec2.MonitorInstances", ec2.MonitoringStateEnabled, "i-0123456789abcdef0") @@ -53,6 +54,7 @@ func TestMonitorInstances_Success(t *testing.T) { } func TestUnmonitorInstances_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) seen := serveMonitoring(t, nc, "ec2.UnmonitorInstances", ec2.MonitoringStateDisabled, "i-0123456789abcdef0") @@ -69,6 +71,7 @@ func TestUnmonitorInstances_Success(t *testing.T) { // The two actions must not share a subject, or disabling monitoring would // enable it. func TestMonitorInstances_DistinctSubjects(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) serveMonitoring(t, nc, "ec2.MonitorInstances", ec2.MonitoringStateEnabled, "i-0123456789abcdef0") @@ -79,6 +82,7 @@ func TestMonitorInstances_DistinctSubjects(t *testing.T) { } func TestMonitorInstances_InvalidInput(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) tests := []struct { @@ -107,6 +111,7 @@ func TestMonitorInstances_InvalidInput(t *testing.T) { } func TestMonitorInstances_NilInput(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := MonitorInstances(context.Background(), nil, nc, "123456789012") diff --git a/spinifex/gateway/ec2/instance/RebootInstances_test.go b/spinifex/gateway/ec2/instance/RebootInstances_test.go index e0b057966..927cfa13b 100644 --- a/spinifex/gateway/ec2/instance/RebootInstances_test.go +++ b/spinifex/gateway/ec2/instance/RebootInstances_test.go @@ -16,6 +16,7 @@ import ( ) func TestRebootInstances_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-0123456789abcdef0" @@ -42,6 +43,7 @@ func TestRebootInstances_Success(t *testing.T) { } func TestRebootInstances_MultipleInstances(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) ids := []string{"i-001", "i-002"} @@ -63,6 +65,7 @@ func TestRebootInstances_MultipleInstances(t *testing.T) { } func TestRebootInstances_EmptyInstanceIds(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.RebootInstancesInput{ @@ -75,6 +78,7 @@ func TestRebootInstances_EmptyInstanceIds(t *testing.T) { } func TestRebootInstances_NATSRequestFails(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.RebootInstancesInput{ @@ -87,6 +91,7 @@ func TestRebootInstances_NATSRequestFails(t *testing.T) { } func TestRebootInstances_StoppedInstance(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-stopped" @@ -113,6 +118,7 @@ func TestRebootInstances_StoppedInstance(t *testing.T) { } func TestRebootInstances_DaemonError(t *testing.T) { + t.Parallel() for _, code := range []string{ awserrors.ErrorIncorrectInstanceState, awserrors.ErrorServerInternal, diff --git a/spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go b/spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go index ab6e8ce52..308a5d6f3 100644 --- a/spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go +++ b/spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "testing" - "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" @@ -29,6 +28,7 @@ func crTargetedInput(crID string) *ec2.RunInstancesInput { } func TestCapacityReservationTargetID(t *testing.T) { + t.Parallel() assert.Empty(t, capacityReservationTargetID(&ec2.RunInstancesInput{}), "no spec → general path") assert.Empty(t, capacityReservationTargetID(&ec2.RunInstancesInput{ CapacityReservationSpecification: &ec2.CapacityReservationSpecification{ @@ -40,6 +40,7 @@ func TestCapacityReservationTargetID(t *testing.T) { // A malformed reservation id is rejected at the gateway before any NATS call. func TestRunInstancesInner_TargetedMalformedID(t *testing.T) { + t.Parallel() _, err := runInstancesInner(context.Background(), crTargetedInput("bogus-id"), nil, nil, "123456789012", nil, nil, 1) require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdMalformed, err.Error()) @@ -47,6 +48,7 @@ func TestRunInstancesInner_TargetedMalformedID(t *testing.T) { // The placement-group + capacity-reservation combination is rejected. func TestRunInstancesInner_TargetedWithPlacementGroup(t *testing.T) { + t.Parallel() input := crTargetedInput(testCRID) input.Placement = &ec2.Placement{GroupName: aws.String("pg-cluster")} _, err := runInstancesInner(context.Background(), input, nil, nil, "123456789012", nil, nil, 1) @@ -57,6 +59,7 @@ func TestRunInstancesInner_TargetedWithPlacementGroup(t *testing.T) { // No responder on the cr-subject (cancelled / restarted reservation) maps the // transport-level ErrNoResponders to InvalidCapacityReservationId.NotFound. func TestRunIntoReservation_NoResponder(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) _, err := runIntoReservation(context.Background(), crTargetedInput(testCRID), nc, "123456789012", testCRID) require.Error(t, err) @@ -66,6 +69,7 @@ func TestRunIntoReservation_NoResponder(t *testing.T) { // A targeted launch routes to ec2.RunInstances.cr. and returns the owning // daemon's reservation verbatim. func TestRunIntoReservation_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) { @@ -77,7 +81,7 @@ func TestRunIntoReservation_Success(t *testing.T) { }) require.NoError(t, err) defer func() { _ = sub.Unsubscribe() }() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) reservation, err := runIntoReservation(context.Background(), crTargetedInput(testCRID), nc, "123456789012", testCRID) require.NoError(t, err) @@ -88,6 +92,7 @@ func TestRunIntoReservation_Success(t *testing.T) { // A semantic error from the owning daemon (e.g. a full reservation) rides back // as its awserror code, not the generic NotFound. func TestRunIntoReservation_DaemonErrorPropagates(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) { @@ -95,7 +100,7 @@ func TestRunIntoReservation_DaemonErrorPropagates(t *testing.T) { }) require.NoError(t, err) defer func() { _ = sub.Unsubscribe() }() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) _, err = runIntoReservation(context.Background(), crTargetedInput(testCRID), nc, "123456789012", testCRID) require.Error(t, err) @@ -105,6 +110,7 @@ func TestRunIntoReservation_DaemonErrorPropagates(t *testing.T) { // The full runInstancesInner branch routes a targeted launch to the cr-subject // (not the general distribute path) and enriches the result. func TestRunInstancesInner_TargetedRoutesToCRSubject(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) { @@ -116,7 +122,7 @@ func TestRunInstancesInner_TargetedRoutesToCRSubject(t *testing.T) { }) require.NoError(t, err) defer func() { _ = sub.Unsubscribe() }() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) reservation, err := runInstancesInner(context.Background(), crTargetedInput(testCRID), nc, nil, "123456789012", nil, nil, 1) require.NoError(t, err) diff --git a/spinifex/gateway/ec2/instance/RunInstances_test.go b/spinifex/gateway/ec2/instance/RunInstances_test.go index ca358d16b..664c5cd2f 100644 --- a/spinifex/gateway/ec2/instance/RunInstances_test.go +++ b/spinifex/gateway/ec2/instance/RunInstances_test.go @@ -26,6 +26,7 @@ var defaults = ec2.RunInstancesInput{ } func TestParseRunInstances(t *testing.T) { + t.Parallel() // Group multiple tests tests := []struct { name string @@ -231,6 +232,7 @@ func runValidInputWithProfile(spec *ec2.IamInstanceProfileSpecification) *ec2.Ru } func TestResolveAndAuthorizeInstanceProfile_Absent(t *testing.T) { + t.Parallel() // No IamInstanceProfile on the input → no resolution, no PassRole check. in := &ec2.RunInstancesInput{} // empty profile, err := resolveAndAuthorizeInstanceProfile(in, nil, testGwAccountID, nil) @@ -239,6 +241,7 @@ func TestResolveAndAuthorizeInstanceProfile_Absent(t *testing.T) { } func TestResolveAndAuthorizeInstanceProfile_NameForm_NormalisesToARN(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(_, nameOrARN string) (*handlers_iam.InstanceProfile, error) { assert.Equal(t, testProfileNameApp, nameOrARN) return profileWithRole(), nil @@ -257,6 +260,7 @@ func TestResolveAndAuthorizeInstanceProfile_NameForm_NormalisesToARN(t *testing. } func TestResolveAndAuthorizeInstanceProfile_ArnForm(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(_, nameOrARN string) (*handlers_iam.InstanceProfile, error) { assert.Equal(t, testProfileARNApp, nameOrARN) return profileWithRole(), nil @@ -269,6 +273,7 @@ func TestResolveAndAuthorizeInstanceProfile_ArnForm(t *testing.T) { } func TestResolveAndAuthorizeInstanceProfile_MissingNameAndArn(t *testing.T) { + t.Parallel() in := runValidInputWithProfile(&ec2.IamInstanceProfileSpecification{}) _, err := resolveAndAuthorizeInstanceProfile(in, &fakeIAMService{}, testGwAccountID, nil) require.Error(t, err) @@ -276,6 +281,7 @@ func TestResolveAndAuthorizeInstanceProfile_MissingNameAndArn(t *testing.T) { } func TestResolveAndAuthorizeInstanceProfile_NilIAMService(t *testing.T) { + t.Parallel() in := runValidInputWithProfile(&ec2.IamInstanceProfileSpecification{Name: aws.String(testProfileNameApp)}) _, err := resolveAndAuthorizeInstanceProfile(in, nil, testGwAccountID, nil) require.Error(t, err) @@ -283,6 +289,7 @@ func TestResolveAndAuthorizeInstanceProfile_NilIAMService(t *testing.T) { } func TestResolveAndAuthorizeInstanceProfile_NotFoundMapsToEC2Error(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return nil, errors.New(awserrors.ErrorIAMNoSuchEntity) }} @@ -293,6 +300,7 @@ func TestResolveAndAuthorizeInstanceProfile_NotFoundMapsToEC2Error(t *testing.T) } func TestResolveAndAuthorizeInstanceProfile_CrossAccountARNPassthrough(t *testing.T) { + t.Parallel() // IAM service rejects cross-account; gateway surfaces the AccessDenied verbatim. svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return nil, errors.New(awserrors.ErrorAccessDenied) @@ -304,6 +312,7 @@ func TestResolveAndAuthorizeInstanceProfile_CrossAccountARNPassthrough(t *testin } func TestResolveAndAuthorizeInstanceProfile_PassRoleDenied(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileWithRole(), nil }} @@ -318,6 +327,7 @@ func TestResolveAndAuthorizeInstanceProfile_PassRoleDenied(t *testing.T) { } func TestResolveAndAuthorizeInstanceProfile_NoRoleSkipsPassRole(t *testing.T) { + t.Parallel() svc := &fakeIAMService{resolveFn: func(string, string) (*handlers_iam.InstanceProfile, error) { return profileNoRole(), nil }} @@ -334,6 +344,7 @@ func TestResolveAndAuthorizeInstanceProfile_NoRoleSkipsPassRole(t *testing.T) { } func TestEnrichReservationWithProfileID_PopulatesID(t *testing.T) { + t.Parallel() r := &ec2.Reservation{Instances: []*ec2.Instance{ {InstanceId: aws.String("i-001"), IamInstanceProfile: &ec2.IamInstanceProfile{Arn: aws.String(testProfileARNApp)}}, {InstanceId: aws.String("i-002")}, // no IamInstanceProfile yet — gateway fills both Arn + Id @@ -348,11 +359,13 @@ func TestEnrichReservationWithProfileID_PopulatesID(t *testing.T) { } func TestEnrichReservationWithProfileID_NoProfileIsNoOp(t *testing.T) { + t.Parallel() r := &ec2.Reservation{Instances: []*ec2.Instance{{InstanceId: aws.String("i-001")}}} enrichReservationWithProfileID(r, nil) assert.Nil(t, r.Instances[0].IamInstanceProfile, "nil profile must not synthesise an empty IamInstanceProfile") } func TestEnrichReservationWithProfileID_NilReservationIsSafe(t *testing.T) { + t.Parallel() enrichReservationWithProfileID(nil, profileWithRole()) // must not panic } diff --git a/spinifex/gateway/ec2/instance/StartInstances_test.go b/spinifex/gateway/ec2/instance/StartInstances_test.go index e78e2b9f0..41f261fed 100644 --- a/spinifex/gateway/ec2/instance/StartInstances_test.go +++ b/spinifex/gateway/ec2/instance/StartInstances_test.go @@ -14,6 +14,7 @@ import ( ) func TestStartInstances_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-0123456789abcdef0" @@ -48,6 +49,7 @@ func TestStartInstances_Success(t *testing.T) { } func TestStartInstances_MultipleInstances(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) ids := []string{"i-001", "i-002", "i-003"} @@ -76,6 +78,7 @@ func TestStartInstances_MultipleInstances(t *testing.T) { } func TestStartInstances_EmptyInstanceIds(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.StartInstancesInput{ @@ -88,6 +91,7 @@ func TestStartInstances_EmptyInstanceIds(t *testing.T) { } func TestStartInstances_NilInstanceIdSkipped(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-valid" @@ -109,6 +113,7 @@ func TestStartInstances_NilInstanceIdSkipped(t *testing.T) { } func TestStartInstances_NATSRequestFails(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // No subscriber for ec2.start, so NATS request will fail @@ -134,6 +139,7 @@ func TestStartInstances_NATSRequestFails(t *testing.T) { } func TestStartInstances_MixedSuccessAndFailure(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) goodID := "i-good" diff --git a/spinifex/gateway/ec2/instance/StopInstances_test.go b/spinifex/gateway/ec2/instance/StopInstances_test.go index 24959bee8..80ec4f9b9 100644 --- a/spinifex/gateway/ec2/instance/StopInstances_test.go +++ b/spinifex/gateway/ec2/instance/StopInstances_test.go @@ -15,6 +15,7 @@ import ( ) func TestStopInstances_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-0123456789abcdef0" @@ -50,6 +51,7 @@ func TestStopInstances_Success(t *testing.T) { } func TestStopInstances_MultipleInstances(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) ids := []string{"i-001", "i-002"} @@ -77,6 +79,7 @@ func TestStopInstances_MultipleInstances(t *testing.T) { } func TestStopInstances_EmptyInstanceIds(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.StopInstancesInput{ @@ -89,6 +92,7 @@ func TestStopInstances_EmptyInstanceIds(t *testing.T) { } func TestStopInstances_NilInstanceIdSkipped(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-valid" @@ -108,6 +112,7 @@ func TestStopInstances_NilInstanceIdSkipped(t *testing.T) { } func TestStopInstances_NATSRequestFails(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-nosubscriber" @@ -131,6 +136,7 @@ func TestStopInstances_NATSRequestFails(t *testing.T) { } func TestStopInstances_MixedSuccessAndFailure(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) goodID := "i-good" diff --git a/spinifex/gateway/ec2/instance/TerminateInstances_test.go b/spinifex/gateway/ec2/instance/TerminateInstances_test.go index 675ae13db..2c562afa0 100644 --- a/spinifex/gateway/ec2/instance/TerminateInstances_test.go +++ b/spinifex/gateway/ec2/instance/TerminateInstances_test.go @@ -15,6 +15,7 @@ import ( ) func TestTerminateInstances_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-0123456789abcdef0" @@ -50,6 +51,7 @@ func TestTerminateInstances_Success(t *testing.T) { } func TestTerminateInstances_MultipleInstances(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) ids := []string{"i-001", "i-002", "i-003"} @@ -77,6 +79,7 @@ func TestTerminateInstances_MultipleInstances(t *testing.T) { } func TestTerminateInstances_EmptyInstanceIds(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) input := &ec2.TerminateInstancesInput{ @@ -89,6 +92,7 @@ func TestTerminateInstances_EmptyInstanceIds(t *testing.T) { } func TestTerminateInstances_NilInstanceIdSkipped(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-valid" @@ -148,6 +152,7 @@ func TestTerminateInstances_MixedSuccessAndFailure(t *testing.T) { } func TestTerminateInstances_VerifiesQMPAttributes(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) instanceID := "i-verify" diff --git a/spinifex/gateway/ec2/instance/clienttoken_test.go b/spinifex/gateway/ec2/instance/clienttoken_test.go index 30602effc..08dc2b875 100644 --- a/spinifex/gateway/ec2/instance/clienttoken_test.go +++ b/spinifex/gateway/ec2/instance/clienttoken_test.go @@ -28,6 +28,7 @@ func newTestClientTokenStore(t *testing.T) *ClientTokenStore { // A completed token replays the stored reservation for a duplicate caller with // matching params. func TestClientToken_ReplaysCompletedReservation(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok, hash = "tok-1", "hash-a" @@ -47,6 +48,7 @@ func TestClientToken_ReplaysCompletedReservation(t *testing.T) { // Reusing a token with different params is IdempotentParameterMismatch. func TestClientToken_ParamMismatchRejected(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok = "tok-2" @@ -62,6 +64,7 @@ func TestClientToken_ParamMismatchRejected(t *testing.T) { // A failed launch aborts the token so a later retry with the same token // re-launches instead of replaying a non-existent reservation. func TestClientToken_AbortAllowsRelaunch(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok, hash = "tok-3", "hash-a" @@ -79,6 +82,7 @@ func TestClientToken_AbortAllowsRelaunch(t *testing.T) { // Concurrent callers with the same token must yield exactly one owner; the // others replay the owner's reservation. Proves single-launch under -race. func TestClientToken_ConcurrentSingleOwner(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok, hash = "tok-4", "hash-a" res := &ec2.Reservation{ReservationId: aws.String("r-only")} @@ -122,6 +126,7 @@ func TestClientToken_ConcurrentSingleOwner(t *testing.T) { // clientTokenParamHash ignores ClientToken (same params, different token → // same hash) but reflects a real parameter change. func TestClientTokenParamHash_IgnoresTokenReflectsParams(t *testing.T) { + t.Parallel() base := &ec2.RunInstancesInput{ ImageId: aws.String("ami-123"), InstanceType: aws.String("t3.micro"), @@ -144,6 +149,7 @@ func TestClientTokenParamHash_IgnoresTokenReflectsParams(t *testing.T) { // A completed token replays its reservation and must NOT invoke the launcher. func TestRunInstancesWithClientToken_ReplaySkipsLaunch(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok, hash = "rt-1", "h" _, owned, err := store.Claim(t.Context(), ctTestAccount, tok, hash) @@ -163,6 +169,7 @@ func TestRunInstancesWithClientToken_ReplaySkipsLaunch(t *testing.T) { // The owner launches once and finalizes; a duplicate replays without launching. func TestRunInstancesWithClientToken_OwnerLaunchesOnceThenReplay(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok, hash = "rt-2", "h" launches := 0 @@ -183,6 +190,7 @@ func TestRunInstancesWithClientToken_OwnerLaunchesOnceThenReplay(t *testing.T) { // A launch failure aborts the token so a retry re-launches. func TestRunInstancesWithClientToken_LaunchFailureAborts(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok, hash = "rt-3", "h" @@ -204,6 +212,7 @@ func TestRunInstancesWithClientToken_LaunchFailureAborts(t *testing.T) { // Token reuse with different params maps to the AWS IdempotentParameterMismatch // error code and never launches. func TestRunInstancesWithClientToken_ParamMismatchMapsAWSError(t *testing.T) { + t.Parallel() store := newTestClientTokenStore(t) const tok = "rt-4" _, owned, err := store.Claim(t.Context(), ctTestAccount, tok, "hA") @@ -223,6 +232,7 @@ func TestRunInstancesWithClientToken_ParamMismatchMapsAWSError(t *testing.T) { // getClientTokenStore binds the process-wide store once and returns the same // instance on subsequent calls. func TestGetClientTokenStore_BindsOnce(t *testing.T) { + t.Parallel() _, nc, _ := testutil.StartTestJetStream(t) s1, err := getClientTokenStore(t.Context(), nc) require.NoError(t, err) diff --git a/spinifex/gateway/ec2/instance/placement_test.go b/spinifex/gateway/ec2/instance/placement_test.go index ca21da2de..95fc4ed72 100644 --- a/spinifex/gateway/ec2/instance/placement_test.go +++ b/spinifex/gateway/ec2/instance/placement_test.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "testing" - "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" @@ -21,6 +20,7 @@ import ( // --- spreadAllocate tests (pure algorithm) --- func TestSpreadAllocate_EqualDistribution(t *testing.T) { + t.Parallel() // 3 instances across 3 nodes → 1 per node nodes := []nodeAllocation{ {NodeID: "A", Available: 4}, @@ -36,6 +36,7 @@ func TestSpreadAllocate_EqualDistribution(t *testing.T) { } func TestSpreadAllocate_SpreadThenPack(t *testing.T) { + t.Parallel() // 5 instances across 3 nodes (capacities: A=4, B=3, C=2) // Round 1: A=1, B=1, C=1 (3 assigned, 2 remaining) // Round 2: A gets 1 (remaining cap 3), B gets 1 (remaining cap 2) @@ -57,6 +58,7 @@ func TestSpreadAllocate_SpreadThenPack(t *testing.T) { } func TestSpreadAllocate_SingleNode(t *testing.T) { + t.Parallel() // All 3 instances on 1 node nodes := []nodeAllocation{ {NodeID: "A", Available: 5}, @@ -69,6 +71,7 @@ func TestSpreadAllocate_SingleNode(t *testing.T) { } func TestSpreadAllocate_MoreNodesThanInstances(t *testing.T) { + t.Parallel() // 2 instances across 5 nodes → only 2 get assigned nodes := []nodeAllocation{ {NodeID: "A", Available: 4}, @@ -89,6 +92,7 @@ func TestSpreadAllocate_MoreNodesThanInstances(t *testing.T) { } func TestSpreadAllocate_HeavyPacking(t *testing.T) { + t.Parallel() // 10 instances across 2 nodes (A=8, B=6) // Round 1: A=1, B=1 // Packing: each round picks node with most remaining @@ -110,6 +114,7 @@ func TestSpreadAllocate_HeavyPacking(t *testing.T) { } func TestSpreadAllocate_ExactCapacity(t *testing.T) { + t.Parallel() // Request exactly matches total capacity nodes := []nodeAllocation{ {NodeID: "A", Available: 2}, @@ -127,6 +132,7 @@ func TestSpreadAllocate_ExactCapacity(t *testing.T) { } func TestSpreadAllocate_ZeroCount(t *testing.T) { + t.Parallel() nodes := []nodeAllocation{ {NodeID: "A", Available: 4}, } @@ -135,6 +141,7 @@ func TestSpreadAllocate_ZeroCount(t *testing.T) { } func TestSpreadAllocate_EmptyNodes(t *testing.T) { + t.Parallel() result := spreadAllocate(nil, 3) assert.Empty(t, result) } @@ -142,6 +149,7 @@ func TestSpreadAllocate_EmptyNodes(t *testing.T) { // --- queryNodeCapacity tests (NATS-based) --- func TestQueryNodeCapacity_FiltersEligibleNodes(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Simulate 3 daemons responding to spinifex.node.status @@ -190,6 +198,7 @@ func TestQueryNodeCapacity_FiltersEligibleNodes(t *testing.T) { } func TestQueryNodeCapacity_NoNodes(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // No subscribers → timeout, empty result @@ -201,6 +210,7 @@ func TestQueryNodeCapacity_NoNodes(t *testing.T) { // --- aggregateResults tests --- func TestAggregateResults_AllSucceed(t *testing.T) { + t.Parallel() results := []nodeLaunchResult{ { NodeID: "node-1", @@ -231,6 +241,7 @@ func TestAggregateResults_AllSucceed(t *testing.T) { } func TestAggregateResults_PartialSuccessMeetsMinCount(t *testing.T) { + t.Parallel() results := []nodeLaunchResult{ { NodeID: "node-1", @@ -284,6 +295,7 @@ func TestAggregateResults_PartialFailureBelowMinCount(t *testing.T) { } func TestAggregateResults_AllFail(t *testing.T) { + t.Parallel() results := []nodeLaunchResult{ {NodeID: "node-1", Err: assert.AnError}, {NodeID: "node-2", Err: assert.AnError}, @@ -298,6 +310,7 @@ func TestAggregateResults_AllFail(t *testing.T) { // A node-side capacity race (queryNodeCapacity saw room, the node then rejected) // still propagates as InsufficientInstanceCapacity via the client-error whitelist. func TestAggregateResults_NodeCapacityRacePropagates(t *testing.T) { + t.Parallel() inner := errors.New(awserrors.ErrorInsufficientInstanceCapacity) wrapped := fmt.Errorf("launch on node-1: %w", inner) results := []nodeLaunchResult{ @@ -311,6 +324,7 @@ func TestAggregateResults_NodeCapacityRacePropagates(t *testing.T) { // A node RPC timeout must surface as the real error, never masquerade as capacity. func TestAggregateResults_TimeoutSurfacedNotMasked(t *testing.T) { + t.Parallel() timeoutErr := fmt.Errorf("launch on node-1: %w", context.DeadlineExceeded) results := []nodeLaunchResult{ {NodeID: "node-1", Err: timeoutErr}, @@ -346,6 +360,7 @@ func TestAggregateResults_ShortWithoutNodeErrors(t *testing.T) { // --- extractClientError tests --- func TestExtractClientError_NoErrors(t *testing.T) { + t.Parallel() results := []nodeLaunchResult{ {NodeID: "node-1", Reservation: &ec2.Reservation{}}, } @@ -353,6 +368,7 @@ func TestExtractClientError_NoErrors(t *testing.T) { } func TestExtractClientError_GenericError(t *testing.T) { + t.Parallel() results := []nodeLaunchResult{ {NodeID: "node-1", Err: assert.AnError}, } @@ -360,6 +376,7 @@ func TestExtractClientError_GenericError(t *testing.T) { } func TestExtractClientError_AMINotFound(t *testing.T) { + t.Parallel() // Simulate the error wrapping that launchOnNodes does inner := errors.New(awserrors.ErrorInvalidAMIIDNotFound) wrapped := fmt.Errorf("launch on node-1: %w", inner) @@ -372,6 +389,7 @@ func TestExtractClientError_AMINotFound(t *testing.T) { } func TestExtractClientError_NestedWrappedCode(t *testing.T) { + t.Parallel() inner := errors.New(awserrors.ErrorInvalidParameterValue) wrapped := fmt.Errorf("launch on node-1: %w", fmt.Errorf("prepare instance: %w", inner)) results := []nodeLaunchResult{ @@ -383,6 +401,7 @@ func TestExtractClientError_NestedWrappedCode(t *testing.T) { } func TestExtractClientError_JoinedCode(t *testing.T) { + t.Parallel() joined := errors.Join(assert.AnError, errors.New(awserrors.ErrorInvalidGroupNotFound)) results := []nodeLaunchResult{ {NodeID: "node-1", Err: fmt.Errorf("launch on node-1: %w", joined)}, @@ -393,6 +412,7 @@ func TestExtractClientError_JoinedCode(t *testing.T) { } func TestExtractClientError_KeyPairNotFound(t *testing.T) { + t.Parallel() inner := errors.New(awserrors.ErrorInvalidKeyPairNotFound) wrapped := fmt.Errorf("launch on node-1: %w", inner) results := []nodeLaunchResult{ @@ -404,6 +424,7 @@ func TestExtractClientError_KeyPairNotFound(t *testing.T) { } func TestAggregateResults_PropagatesClientError(t *testing.T) { + t.Parallel() inner := errors.New(awserrors.ErrorInvalidAMIIDNotFound) wrapped := fmt.Errorf("launch on node-1: %w", inner) results := []nodeLaunchResult{ @@ -419,6 +440,7 @@ func TestAggregateResults_PropagatesClientError(t *testing.T) { // --- distributeInstances integration tests (end-to-end with mock daemons) --- func TestDistributeInstances_SuccessfulSpread(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Mock node.status responder @@ -458,7 +480,7 @@ func TestDistributeInstances_SuccessfulSpread(t *testing.T) { require.NoError(t, err) defer sub2.Unsubscribe() - time.Sleep(50 * time.Millisecond) // let subscriptions propagate + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -481,6 +503,7 @@ func TestDistributeInstances_SuccessfulSpread(t *testing.T) { } func TestDistributeInstances_InsufficientCapacity(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Mock node.status with only 1 available slot total @@ -495,7 +518,7 @@ func TestDistributeInstances_InsufficientCapacity(t *testing.T) { require.NoError(t, err) defer statusSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -510,6 +533,7 @@ func TestDistributeInstances_InsufficientCapacity(t *testing.T) { } func TestDistributeInstances_PropagatesAMINotFound(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Mock node.status with capacity available @@ -531,7 +555,7 @@ func TestDistributeInstances_PropagatesAMINotFound(t *testing.T) { require.NoError(t, err) defer sub1.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-0000000000000dead"), @@ -554,6 +578,7 @@ func TestDistributeInstances_PropagatesAMINotFound(t *testing.T) { // Terraform / SDK consumers branch on the specific code; masking it as // "no capacity" makes a typo'd SG ID look like a transient cluster issue. func TestDistributeInstances_PropagatesSGValidationErrors(t *testing.T) { + t.Parallel() cases := []struct { name string daemonErrCode string @@ -583,7 +608,7 @@ func TestDistributeInstances_PropagatesSGValidationErrors(t *testing.T) { require.NoError(t, err) defer daemonSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -601,6 +626,7 @@ func TestDistributeInstances_PropagatesSGValidationErrors(t *testing.T) { } func TestDistributeInstances_LaunchCountCappedToMaxCount(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // 3 nodes with capacity, but MaxCount=2 @@ -631,7 +657,7 @@ func TestDistributeInstances_LaunchCountCappedToMaxCount(t *testing.T) { defer sub.Unsubscribe() } - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -647,10 +673,11 @@ func TestDistributeInstances_LaunchCountCappedToMaxCount(t *testing.T) { } func TestDistributeInstances_NoNodesAvailable(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // No responders to node.status - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -667,6 +694,7 @@ func TestDistributeInstances_NoNodesAvailable(t *testing.T) { // --- RunInstances routing tests --- func TestRunInstances_SingleInstanceDistributes(t *testing.T) { + t.Parallel() // For MinCount=MaxCount=1, RunInstances should still query node capacity // and route to a specific node via targeted topic (not queue group). _, nc := startTestNATSServer(t) @@ -695,7 +723,7 @@ func TestRunInstances_SingleInstanceDistributes(t *testing.T) { require.NoError(t, err) defer nodeSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -714,6 +742,7 @@ func TestRunInstances_SingleInstanceDistributes(t *testing.T) { // --- placementGroupName tests --- func TestPlacementGroupName_WithGroupName(t *testing.T) { + t.Parallel() input := &ec2.RunInstancesInput{ Placement: &ec2.Placement{ GroupName: aws.String("my-group"), @@ -723,11 +752,13 @@ func TestPlacementGroupName_WithGroupName(t *testing.T) { } func TestPlacementGroupName_NilPlacement(t *testing.T) { + t.Parallel() input := &ec2.RunInstancesInput{} assert.Empty(t, placementGroupName(input)) } func TestPlacementGroupName_NilGroupName(t *testing.T) { + t.Parallel() input := &ec2.RunInstancesInput{ Placement: &ec2.Placement{}, } @@ -735,6 +766,7 @@ func TestPlacementGroupName_NilGroupName(t *testing.T) { } func TestPlacementGroupName_EmptyGroupName(t *testing.T) { + t.Parallel() input := &ec2.RunInstancesInput{ Placement: &ec2.Placement{ GroupName: aws.String(""), @@ -746,6 +778,7 @@ func TestPlacementGroupName_EmptyGroupName(t *testing.T) { // --- lookupPlacementGroupStrategy tests --- func TestLookupPlacementGroupStrategy_Success(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Mock DescribePlacementGroups responder @@ -765,7 +798,7 @@ func TestLookupPlacementGroupStrategy_Success(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) strategy, err := lookupPlacementGroupStrategy(context.Background(), nc, "test-account", "my-group") require.NoError(t, err) @@ -773,6 +806,7 @@ func TestLookupPlacementGroupStrategy_Success(t *testing.T) { } func TestLookupPlacementGroupStrategy_NotFound(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) // Mock responder returns error @@ -783,13 +817,14 @@ func TestLookupPlacementGroupStrategy_NotFound(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) _, err = lookupPlacementGroupStrategy(context.Background(), nc, "test-account", "ghost-group") require.Error(t, err) } func TestLookupPlacementGroupStrategy_NotAvailable(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) sub, err := nc.QueueSubscribe("ec2.DescribePlacementGroups", "spinifex-workers", func(msg *nats.Msg) { @@ -808,7 +843,7 @@ func TestLookupPlacementGroupStrategy_NotAvailable(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) _, err = lookupPlacementGroupStrategy(context.Background(), nc, "test-account", "my-group") require.Error(t, err) @@ -816,6 +851,7 @@ func TestLookupPlacementGroupStrategy_NotAvailable(t *testing.T) { } func TestLookupPlacementGroupStrategy_EmptyResult(t *testing.T) { + t.Parallel() _, nc := startTestNATSServer(t) sub, err := nc.QueueSubscribe("ec2.DescribePlacementGroups", "spinifex-workers", func(msg *nats.Msg) { @@ -828,7 +864,7 @@ func TestLookupPlacementGroupStrategy_EmptyResult(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) _, err = lookupPlacementGroupStrategy(context.Background(), nc, "test-account", "my-group") require.Error(t, err) @@ -838,6 +874,7 @@ func TestLookupPlacementGroupStrategy_EmptyResult(t *testing.T) { // --- distributeInstancesCluster tests --- func TestDistributeInstancesCluster_FirstLaunchPicksBestNode(t *testing.T) { + t.Parallel() // First launch on empty cluster group should pick node with most capacity _, nc := startTestNATSServer(t) @@ -889,7 +926,7 @@ func TestDistributeInstancesCluster_FirstLaunchPicksBestNode(t *testing.T) { require.NoError(t, err) defer finalizeSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -909,6 +946,7 @@ func TestDistributeInstancesCluster_FirstLaunchPicksBestNode(t *testing.T) { } func TestDistributeInstancesCluster_SubsequentLaunchPinsToExistingNode(t *testing.T) { + t.Parallel() // Subsequent launch should go to the same node the group already uses _, nc := startTestNATSServer(t) @@ -966,7 +1004,7 @@ func TestDistributeInstancesCluster_SubsequentLaunchPinsToExistingNode(t *testin require.NoError(t, err) defer finalizeSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -982,6 +1020,7 @@ func TestDistributeInstancesCluster_SubsequentLaunchPinsToExistingNode(t *testin } func TestDistributeInstancesCluster_InsufficientCapacityOnPinnedNode(t *testing.T) { + t.Parallel() // Pinned node doesn't have enough capacity → InsufficientInstanceCapacity _, nc := startTestNATSServer(t) @@ -1009,7 +1048,7 @@ func TestDistributeInstancesCluster_InsufficientCapacityOnPinnedNode(t *testing. require.NoError(t, err) defer reserveSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -1024,6 +1063,7 @@ func TestDistributeInstancesCluster_InsufficientCapacityOnPinnedNode(t *testing. } func TestDistributeInstancesCluster_PinnedNodeNotInCapacityResults(t *testing.T) { + t.Parallel() // Pinned node has no capacity at all (not in fan-out results) → InsufficientInstanceCapacity _, nc := startTestNATSServer(t) @@ -1050,7 +1090,7 @@ func TestDistributeInstancesCluster_PinnedNodeNotInCapacityResults(t *testing.T) require.NoError(t, err) defer reserveSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -1065,6 +1105,7 @@ func TestDistributeInstancesCluster_PinnedNodeNotInCapacityResults(t *testing.T) } func TestDistributeInstancesCluster_LaunchCountCappedByCapacityAndMaxCount(t *testing.T) { + t.Parallel() // Target node has 3 available but MaxCount=2 → should launch 2 _, nc := startTestNATSServer(t) @@ -1115,7 +1156,7 @@ func TestDistributeInstancesCluster_LaunchCountCappedByCapacityAndMaxCount(t *te require.NoError(t, err) defer finalizeSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -1130,6 +1171,7 @@ func TestDistributeInstancesCluster_LaunchCountCappedByCapacityAndMaxCount(t *te } func TestRunInstances_ClusterPlacementGroupRouting(t *testing.T) { + t.Parallel() // RunInstances with a cluster placement group should route through // the cluster path (lookupPlacementGroupStrategy → distributeInstancesCluster). _, nc := startTestNATSServer(t) @@ -1197,7 +1239,7 @@ func TestRunInstances_ClusterPlacementGroupRouting(t *testing.T) { require.NoError(t, err) defer finalizeSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), @@ -1216,6 +1258,7 @@ func TestRunInstances_ClusterPlacementGroupRouting(t *testing.T) { } func TestRunInstances_MultiInstanceUsesDistribution(t *testing.T) { + t.Parallel() // For MaxCount > 1, RunInstances should use the distribution path, // which queries spinifex.node.status. _, nc := startTestNATSServer(t) @@ -1248,7 +1291,7 @@ func TestRunInstances_MultiInstanceUsesDistribution(t *testing.T) { require.NoError(t, err) defer nodeSub.Unsubscribe() - time.Sleep(50 * time.Millisecond) + require.NoError(t, nc.Flush()) input := &ec2.RunInstancesInput{ ImageId: aws.String("ami-test"), diff --git a/spinifex/gateway/ec2/instance/state_test.go b/spinifex/gateway/ec2/instance/state_test.go index 3b9f4ac95..975af786f 100644 --- a/spinifex/gateway/ec2/instance/state_test.go +++ b/spinifex/gateway/ec2/instance/state_test.go @@ -8,6 +8,7 @@ import ( ) func TestNewStateChange(t *testing.T) { + t.Parallel() tests := []struct { name string instanceID string diff --git a/spinifex/handlers/dns/zone_lookup_test.go b/spinifex/handlers/dns/zone_lookup_test.go index c79477855..0b2255e9d 100644 --- a/spinifex/handlers/dns/zone_lookup_test.go +++ b/spinifex/handlers/dns/zone_lookup_test.go @@ -33,8 +33,9 @@ class = 1 } // fakeZoneS3 is a mutable path-style S3 endpoint (GET only) backing HostsZone -// lookups. Keys listed in errorKeys return 500 instead of the usual 200/404, -// simulating a transient S3 failure for that specific zone object. requested +// lookups. Keys listed in errorKeys return 403 instead of the usual 200/404, +// simulating a read failure for that zone object — 403 rather than a retried +// (and so slow) 5xx, since HostsZone branches only on err != nil. requested // records every key fetched, letting a test assert which candidate zones were // actually queried. func fakeZoneS3(t *testing.T, bucket string, objects map[string]string, errorKeys map[string]bool) (endpoint string, requested *[]string) { @@ -48,7 +49,7 @@ func fakeZoneS3(t *testing.T, bucket string, objects map[string]string, errorKey log = append(log, key) mu.Unlock() if errorKeys[key] { - http.Error(w, "internal error", http.StatusInternalServerError) + http.Error(w, "access denied", http.StatusForbidden) return } body, ok := objects[key] diff --git a/spinifex/handlers/ec2/routetable/helpers_test.go b/spinifex/handlers/ec2/routetable/helpers_test.go index 30521e239..e29d1f531 100644 --- a/spinifex/handlers/ec2/routetable/helpers_test.go +++ b/spinifex/handlers/ec2/routetable/helpers_test.go @@ -11,23 +11,27 @@ import ( // --- rtbMatchesFilters --- func TestRtbMatchesFilters_EmptyFilters(t *testing.T) { + t.Parallel() record := &RouteTableRecord{RouteTableId: "rtb-1", VpcId: "vpc-1"} assert.True(t, rtbMatchesFilters(record, map[string][]string{})) } func TestRtbMatchesFilters_VpcId(t *testing.T) { + t.Parallel() record := &RouteTableRecord{RouteTableId: "rtb-1", VpcId: "vpc-aaa"} assert.True(t, rtbMatchesFilters(record, map[string][]string{"vpc-id": {"vpc-aaa"}})) assert.False(t, rtbMatchesFilters(record, map[string][]string{"vpc-id": {"vpc-bbb"}})) } func TestRtbMatchesFilters_RouteTableId(t *testing.T) { + t.Parallel() record := &RouteTableRecord{RouteTableId: "rtb-xyz", VpcId: "vpc-1"} assert.True(t, rtbMatchesFilters(record, map[string][]string{"route-table-id": {"rtb-xyz"}})) assert.False(t, rtbMatchesFilters(record, map[string][]string{"route-table-id": {"rtb-other"}})) } func TestRtbMatchesFilters_AssociationMain_True(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ Associations: []AssociationRecord{ {AssociationId: "rtbassoc-1", Main: true}, @@ -38,6 +42,7 @@ func TestRtbMatchesFilters_AssociationMain_True(t *testing.T) { } func TestRtbMatchesFilters_AssociationMain_False(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ Associations: []AssociationRecord{ {AssociationId: "rtbassoc-1", Main: false}, @@ -47,6 +52,7 @@ func TestRtbMatchesFilters_AssociationMain_False(t *testing.T) { } func TestRtbMatchesFilters_AssociationId(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ Associations: []AssociationRecord{ {AssociationId: "rtbassoc-abc"}, @@ -61,6 +67,7 @@ func TestRtbMatchesFilters_AssociationId(t *testing.T) { } func TestRtbMatchesFilters_AssociationSubnetId(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ Associations: []AssociationRecord{ {AssociationId: "rtbassoc-1", SubnetId: "subnet-aaa"}, @@ -75,6 +82,7 @@ func TestRtbMatchesFilters_AssociationSubnetId(t *testing.T) { } func TestRtbMatchesFilters_RouteDestinationCidr(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ Routes: []RouteRecord{ {DestinationCidrBlock: "10.0.0.0/16", GatewayId: "local", State: "active"}, @@ -90,6 +98,7 @@ func TestRtbMatchesFilters_RouteDestinationCidr(t *testing.T) { } func TestRtbMatchesFilters_RouteGatewayId(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ Routes: []RouteRecord{ {DestinationCidrBlock: "0.0.0.0/0", GatewayId: "igw-abc", State: "active"}, @@ -104,6 +113,7 @@ func TestRtbMatchesFilters_RouteGatewayId(t *testing.T) { } func TestRtbMatchesFilters_MultipleFilters(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ RouteTableId: "rtb-1", VpcId: "vpc-aaa", @@ -124,6 +134,7 @@ func TestRtbMatchesFilters_MultipleFilters(t *testing.T) { } func TestRtbMatchesFilters_MultipleValues(t *testing.T) { + t.Parallel() record := &RouteTableRecord{VpcId: "vpc-bbb"} assert.True(t, rtbMatchesFilters(record, map[string][]string{ "vpc-id": {"vpc-aaa", "vpc-bbb", "vpc-ccc"}, @@ -131,6 +142,7 @@ func TestRtbMatchesFilters_MultipleValues(t *testing.T) { } func TestRtbMatchesFilters_NoAssociations(t *testing.T) { + t.Parallel() record := &RouteTableRecord{RouteTableId: "rtb-1"} assert.False(t, rtbMatchesFilters(record, map[string][]string{ "association.subnet-id": {"subnet-1"}, @@ -138,6 +150,7 @@ func TestRtbMatchesFilters_NoAssociations(t *testing.T) { } func TestRtbMatchesFilters_NoRoutes(t *testing.T) { + t.Parallel() record := &RouteTableRecord{RouteTableId: "rtb-1"} assert.False(t, rtbMatchesFilters(record, map[string][]string{ "route.destination-cidr-block": {"0.0.0.0/0"}, @@ -145,6 +158,7 @@ func TestRtbMatchesFilters_NoRoutes(t *testing.T) { } func TestRtbMatchesFilters_UnknownFilter(t *testing.T) { + t.Parallel() record := &RouteTableRecord{RouteTableId: "rtb-1", VpcId: "vpc-1"} assert.False(t, rtbMatchesFilters(record, map[string][]string{ "bogus-filter": {"value"}, @@ -152,6 +166,7 @@ func TestRtbMatchesFilters_UnknownFilter(t *testing.T) { } func TestRtbMatchesFilters_Wildcard(t *testing.T) { + t.Parallel() record := &RouteTableRecord{RouteTableId: "rtb-1", VpcId: "vpc-aaa"} assert.True(t, rtbMatchesFilters(record, map[string][]string{ "vpc-id": {"vpc-*"}, @@ -159,6 +174,7 @@ func TestRtbMatchesFilters_Wildcard(t *testing.T) { } func TestRtbMatchesFilters_TagFilter(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ RouteTableId: "rtb-1", VpcId: "vpc-1", @@ -178,6 +194,7 @@ func TestRtbMatchesFilters_TagFilter(t *testing.T) { // --- recordToEC2 --- func TestRecordToEC2_Basic(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ RouteTableId: "rtb-abc", VpcId: "vpc-123", @@ -193,6 +210,7 @@ func TestRecordToEC2_Basic(t *testing.T) { } func TestRecordToEC2_Routes(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ RouteTableId: "rtb-1", VpcId: "vpc-1", @@ -217,6 +235,7 @@ func TestRecordToEC2_Routes(t *testing.T) { } func TestRecordToEC2_Associations(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ RouteTableId: "rtb-1", VpcId: "vpc-1", @@ -240,6 +259,7 @@ func TestRecordToEC2_Associations(t *testing.T) { } func TestRecordToEC2_Tags(t *testing.T) { + t.Parallel() record := &RouteTableRecord{ RouteTableId: "rtb-1", VpcId: "vpc-1", diff --git a/spinifex/handlers/ec2/routetable/lifecycle_invariants_test.go b/spinifex/handlers/ec2/routetable/lifecycle_invariants_test.go index ebedcd385..8d31aa2bf 100644 --- a/spinifex/handlers/ec2/routetable/lifecycle_invariants_test.go +++ b/spinifex/handlers/ec2/routetable/lifecycle_invariants_test.go @@ -16,6 +16,7 @@ import ( // table, not success. Idempotent convergence belongs to destroy orchestration, // which tolerates NotFound via awserrors.IsNotFound; the public API stays AWS compatible. func TestRLC1_RouteTableDeleteNotFoundOnAbsent(t *testing.T) { + t.Parallel() svc := setupTestService(t) _, err := svc.DeleteRouteTable(t.Context(), &ec2.DeleteRouteTableInput{ diff --git a/spinifex/handlers/ec2/routetable/record_tags_test.go b/spinifex/handlers/ec2/routetable/record_tags_test.go index 091dad137..3708a7dfb 100644 --- a/spinifex/handlers/ec2/routetable/record_tags_test.go +++ b/spinifex/handlers/ec2/routetable/record_tags_test.go @@ -19,6 +19,7 @@ func rtbTagVal(tags []*ec2.Tag, key string) string { } func TestRouteTableRecordTagsMirror(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -57,6 +58,7 @@ func TestRouteTableRecordTagsMirror(t *testing.T) { } func TestRouteTableRecordTagsMirror_UnownedNoError(t *testing.T) { + t.Parallel() svc := setupTestService(t) // Absent rtb + non-rtb id: both skipped without error. require.NoError(t, svc.ApplyRecordTags(&ec2.CreateTagsInput{ diff --git a/spinifex/handlers/ec2/routetable/service_impl_test.go b/spinifex/handlers/ec2/routetable/service_impl_test.go index 515827a0a..4f3ca08d0 100644 --- a/spinifex/handlers/ec2/routetable/service_impl_test.go +++ b/spinifex/handlers/ec2/routetable/service_impl_test.go @@ -53,6 +53,32 @@ func setupTestService(t *testing.T) *RouteTableServiceImpl { return svc } +// pendingEvents flushes the service connection and reports how many events are +// already queued on sub. Publishes and deliveries share one connection, so a +// completed flush is an exact barrier: no sleep is needed to prove a negative. +func pendingEvents(t *testing.T, svc *RouteTableServiceImpl, sub *nats.Subscription) int { + t.Helper() + require.NoError(t, svc.natsConn.Flush()) + msgs, _, err := sub.Pending() + require.NoError(t, err) + return msgs +} + +// assertNoEvents asserts that nothing was published on sub. +func assertNoEvents(t *testing.T, svc *RouteTableServiceImpl, sub *nats.Subscription, msg string) { + t.Helper() + assert.Zero(t, pendingEvents(t, svc, sub), msg) +} + +// drainEvents discards every event already queued on sub. +func drainEvents(t *testing.T, svc *RouteTableServiceImpl, sub *nats.Subscription) { + t.Helper() + for range pendingEvents(t, svc, sub) { + _, err := sub.NextMsg(time.Second) + require.NoError(t, err) + } +} + func seedKV(t *testing.T, js jetstream.JetStream, bucket string, entries map[string][]byte) jetstream.KeyValue { t.Helper() kv, err := js.CreateOrUpdateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: bucket, History: 1}) @@ -74,6 +100,7 @@ func createTestRtb(t *testing.T, svc *RouteTableServiceImpl) string { } func TestCreateRouteTable(t *testing.T) { + t.Parallel() svc := setupTestService(t) out, err := svc.CreateRouteTable(t.Context(), &ec2.CreateRouteTableInput{ VpcId: aws.String("vpc-test1"), @@ -97,6 +124,7 @@ func TestCreateRouteTable(t *testing.T) { // teardown can find the private route table, clear its NAT-GW route, and let // the NAT gateway (and its billable EIP) be reclaimed. func TestCreateRouteTable_PersistsTagsForTagFilterDiscovery(t *testing.T) { + t.Parallel() svc := setupTestService(t) _, err := svc.CreateRouteTable(t.Context(), &ec2.CreateRouteTableInput{ VpcId: aws.String("vpc-test1"), @@ -128,6 +156,7 @@ func TestCreateRouteTable_PersistsTagsForTagFilterDiscovery(t *testing.T) { } func TestCreateRouteTable_VpcNotFound(t *testing.T) { + t.Parallel() svc := setupTestService(t) _, err := svc.CreateRouteTable(t.Context(), &ec2.CreateRouteTableInput{ VpcId: aws.String("vpc-nonexistent"), @@ -136,6 +165,7 @@ func TestCreateRouteTable_VpcNotFound(t *testing.T) { } func TestDeleteRouteTable(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -150,6 +180,7 @@ func TestDeleteRouteTable(t *testing.T) { } func TestDeleteRouteTable_Main(t *testing.T) { + t.Parallel() svc := setupTestService(t) record, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "") require.NoError(t, err) @@ -161,6 +192,7 @@ func TestDeleteRouteTable_Main(t *testing.T) { } func TestDeleteRouteTable_WithAssociations(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -179,6 +211,7 @@ func TestDeleteRouteTable_WithAssociations(t *testing.T) { } func TestDescribeRouteTables(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -189,6 +222,7 @@ func TestDescribeRouteTables(t *testing.T) { } func TestDescribeRouteTables_FilterByVpcId(t *testing.T) { + t.Parallel() svc := setupTestService(t) createTestRtb(t, svc) @@ -210,6 +244,7 @@ func TestDescribeRouteTables_FilterByVpcId(t *testing.T) { } func TestDescribeRouteTables_FilterByMain(t *testing.T) { + t.Parallel() svc := setupTestService(t) // Create a main route table @@ -230,6 +265,7 @@ func TestDescribeRouteTables_FilterByMain(t *testing.T) { } func TestDescribeRouteTables_FilterByRouteState(t *testing.T) { + t.Parallel() svc := setupTestService(t) createTestRtb(t, svc) // has local route with state=active @@ -249,6 +285,7 @@ func TestDescribeRouteTables_FilterByRouteState(t *testing.T) { } func TestDescribeRouteTables_FilterByRouteOrigin(t *testing.T) { + t.Parallel() svc := setupTestService(t) createTestRtb(t, svc) // local route has origin=CreateRouteTable @@ -266,6 +303,7 @@ func TestDescribeRouteTables_FilterByRouteOrigin(t *testing.T) { } func TestDescribeRouteTables_FilterByRouteNatGatewayId(t *testing.T) { + t.Parallel() svc := setupTestService(t) createTestRtb(t, svc) // no NAT GW route @@ -285,6 +323,7 @@ func TestDescribeRouteTables_FilterByRouteNatGatewayId(t *testing.T) { } func TestDescribeRouteTables_FilterByOwnerId(t *testing.T) { + t.Parallel() svc := setupTestService(t) createTestRtb(t, svc) @@ -311,6 +350,7 @@ func TestDescribeRouteTables_FilterByOwnerId(t *testing.T) { } func TestCreateRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -330,6 +370,7 @@ func TestCreateRoute(t *testing.T) { } func TestCreateRoute_DuplicateDestination(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -350,6 +391,7 @@ func TestCreateRoute_DuplicateDestination(t *testing.T) { } func TestDeleteRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -372,6 +414,7 @@ func TestDeleteRoute(t *testing.T) { } func TestDeleteRoute_LocalRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -383,6 +426,7 @@ func TestDeleteRoute_LocalRoute(t *testing.T) { } func TestDeleteRoute_NotFound(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -394,6 +438,7 @@ func TestDeleteRoute_NotFound(t *testing.T) { } func TestReplaceRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -414,6 +459,7 @@ func TestReplaceRoute(t *testing.T) { } func TestAssociateRouteTable(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -427,6 +473,7 @@ func TestAssociateRouteTable(t *testing.T) { } func TestAssociateRouteTable_DuplicateSubnet(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -450,6 +497,7 @@ func TestAssociateRouteTable_DuplicateSubnet(t *testing.T) { // from the same revision and was clobbered, so the provider timed out waiting // for 'associated'); the CAS path must persist both. func TestAssociateRouteTable_ConcurrentDistinctSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -483,6 +531,7 @@ func TestAssociateRouteTable_ConcurrentDistinctSubnets(t *testing.T) { } func TestDisassociateRouteTable(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtbID := createTestRtb(t, svc) @@ -504,6 +553,7 @@ func TestDisassociateRouteTable(t *testing.T) { } func TestDisassociateRouteTable_Main(t *testing.T) { + t.Parallel() svc := setupTestService(t) record, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "") require.NoError(t, err) @@ -515,6 +565,7 @@ func TestDisassociateRouteTable_Main(t *testing.T) { } func TestReplaceRouteTableAssociation(t *testing.T) { + t.Parallel() svc := setupTestService(t) rtb1ID := createTestRtb(t, svc) rtb2ID := createTestRtb(t, svc) @@ -546,6 +597,7 @@ func TestReplaceRouteTableAssociation(t *testing.T) { } func TestCreateRouteTableForVPC_Main(t *testing.T) { + t.Parallel() svc := setupTestService(t) record, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "") require.NoError(t, err) @@ -581,6 +633,7 @@ func receiveNatGWEvent(t *testing.T, sub *nats.Subscription, wantCidr string) (n // TestAssociateRouteTable_PublishesNatGatewayEvent covers the terraform flow where // AssociateRouteTable must emit a NAT GW event so vpcd installs the SNAT rule. func TestAssociateRouteTable_PublishesNatGatewayEvent(t *testing.T) { + t.Parallel() svc := setupTestService(t) sub, err := svc.natsConn.SubscribeSync("vpc.add-nat-gateway") require.NoError(t, err) @@ -597,8 +650,7 @@ func TestAssociateRouteTable_PublishesNatGatewayEvent(t *testing.T) { require.NoError(t, err) // No subnet yet → no event from CreateRoute. - _, err = sub.NextMsg(100 * time.Millisecond) - assert.Error(t, err, "CreateRoute must not publish when table has no associations") + assertNoEvents(t, svc, sub, "CreateRoute must not publish when table has no associations") _, err = svc.AssociateRouteTable(t.Context(), &ec2.AssociateRouteTableInput{ RouteTableId: aws.String(rtbID), @@ -614,6 +666,7 @@ func TestAssociateRouteTable_PublishesNatGatewayEvent(t *testing.T) { // TestAssociateRouteTable_NoNatGatewayRoute ensures Associate stays silent when // the route table has no NAT GW routes. func TestAssociateRouteTable_NoNatGatewayRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) sub, err := svc.natsConn.SubscribeSync("vpc.add-nat-gateway") require.NoError(t, err) @@ -627,13 +680,13 @@ func TestAssociateRouteTable_NoNatGatewayRoute(t *testing.T) { }, testAccountID) require.NoError(t, err) - _, err = sub.NextMsg(100 * time.Millisecond) - assert.Error(t, err, "Associate must not publish when table has no NAT GW routes") + assertNoEvents(t, svc, sub, "Associate must not publish when table has no NAT GW routes") } // TestDisassociateRouteTable_PublishesNatGatewayDeleteEvent verifies SNAT // teardown when a subnet leaves a table with a NAT GW route. func TestDisassociateRouteTable_PublishesNatGatewayDeleteEvent(t *testing.T) { + t.Parallel() svc := setupTestService(t) delSub, err := svc.natsConn.SubscribeSync("vpc.delete-nat-gateway") require.NoError(t, err) @@ -665,6 +718,7 @@ func TestDisassociateRouteTable_PublishesNatGatewayDeleteEvent(t *testing.T) { // TestDeleteRoute_NATGW_PublishesDeleteForAssociatedSubnets verifies that every // associated subnet receives a vpc.delete-nat-gateway event when a NATGW route is removed. func TestDeleteRoute_NATGW_PublishesDeleteForAssociatedSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) addSub, err := svc.natsConn.SubscribeSync("vpc.add-nat-gateway") require.NoError(t, err) @@ -703,6 +757,7 @@ func TestDeleteRoute_NATGW_PublishesDeleteForAssociatedSubnets(t *testing.T) { // path: subnet leaves a NAT-routed table for a NAT-free table. We expect a // delete event against the old table's NAT GW route and no add event. func TestReplaceRouteTableAssociation_PublishesNatGatewayEvents(t *testing.T) { + t.Parallel() svc := setupTestService(t) addSub, err := svc.natsConn.SubscribeSync("vpc.add-nat-gateway") require.NoError(t, err) @@ -740,8 +795,7 @@ func TestReplaceRouteTableAssociation_PublishesNatGatewayEvents(t *testing.T) { assert.Equal(t, "nat-test1", natgwID) // New table has no NAT GW routes → no add event. - _, err = addSub.NextMsg(100 * time.Millisecond) - assert.Error(t, err, "Replace must not publish add when new table has no NAT GW routes") + assertNoEvents(t, svc, addSub, "Replace must not publish add when new table has no NAT GW routes") } func receiveIGWRouteEvent(t *testing.T, sub *nats.Subscription, wantCidr string) (subnetID, igwID, destCidr string) { @@ -763,6 +817,7 @@ func receiveIGWRouteEvent(t *testing.T, sub *nats.Subscription, wantCidr string) // CreateRoute(IGW) flow: emits one vpc.add-igw-route event per associated // subnet so the network subscriber installs per-subnet egress policies. func TestCreateRoute_IGW_PublishesAddIGWRouteForAssociatedSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) sub, err := svc.natsConn.SubscribeSync("vpc.add-igw-route") require.NoError(t, err) @@ -793,6 +848,7 @@ func TestCreateRoute_IGW_PublishesAddIGWRouteForAssociatedSubnets(t *testing.T) // TestAssociateRouteTable_PublishesAddIGWRouteForExistingRoute covers the terraform // ordering: AssociateRouteTable must emit an IGW route event for the joining subnet. func TestAssociateRouteTable_PublishesAddIGWRouteForExistingRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) sub, err := svc.natsConn.SubscribeSync("vpc.add-igw-route") require.NoError(t, err) @@ -808,8 +864,7 @@ func TestAssociateRouteTable_PublishesAddIGWRouteForExistingRoute(t *testing.T) require.NoError(t, err) // No subnet yet → no event from CreateRoute. - _, err = sub.NextMsg(100 * time.Millisecond) - assert.Error(t, err, "CreateRoute(IGW) must not publish when table has no associations") + assertNoEvents(t, svc, sub, "CreateRoute(IGW) must not publish when table has no associations") _, err = svc.AssociateRouteTable(t.Context(), &ec2.AssociateRouteTableInput{ RouteTableId: aws.String(rtbID), @@ -825,6 +880,7 @@ func TestAssociateRouteTable_PublishesAddIGWRouteForExistingRoute(t *testing.T) // TestDeleteRoute_IGW_PublishesDeleteIGWRoute verifies policy teardown when // the IGW route is removed from a table with associations. func TestDeleteRoute_IGW_PublishesDeleteIGWRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) delSub, err := svc.natsConn.SubscribeSync("vpc.delete-igw-route") require.NoError(t, err) @@ -858,6 +914,7 @@ func TestDeleteRoute_IGW_PublishesDeleteIGWRoute(t *testing.T) { // TestDisassociateRouteTable_PublishesDeleteIGWRoute verifies the egress policy // is torn down when a subnet leaves a table that carries an IGW route. func TestDisassociateRouteTable_PublishesDeleteIGWRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) delSub, err := svc.natsConn.SubscribeSync("vpc.delete-igw-route") require.NoError(t, err) @@ -889,11 +946,11 @@ func TestDisassociateRouteTable_PublishesDeleteIGWRoute(t *testing.T) { // drainIGWSubnets reads up to `count` events from sub and returns the unique // SubnetIds. Used to assert fan-out shape regardless of NATS message order. -func drainIGWSubnets(t *testing.T, sub *nats.Subscription, count int) map[string]bool { +func drainIGWSubnets(t *testing.T, svc *RouteTableServiceImpl, sub *nats.Subscription, count int) map[string]bool { t.Helper() got := map[string]bool{} - for range count { - msg, err := sub.NextMsg(500 * time.Millisecond) + for range min(count, pendingEvents(t, svc, sub)) { + msg, err := sub.NextMsg(time.Second) if err != nil { break } @@ -909,6 +966,7 @@ func drainIGWSubnets(t *testing.T, sub *nats.Subscription, count int) map[string // TestCreateRoute_IGW_OnMainRT_FansOutToImplicitSubnets verifies CreateRoute(IGW) on // the main RT emits one vpc.add-igw-route per implicit-main subnet. func TestCreateRoute_IGW_OnMainRT_FansOutToImplicitSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) sub, err := svc.natsConn.SubscribeSync("vpc.add-igw-route") require.NoError(t, err) @@ -924,7 +982,7 @@ func TestCreateRoute_IGW_OnMainRT_FansOutToImplicitSubnets(t *testing.T) { }, testAccountID) require.NoError(t, err) - got := drainIGWSubnets(t, sub, 4) + got := drainIGWSubnets(t, svc, sub, 4) assert.True(t, got["subnet-test1"], "implicit main-RT subnet must receive add event") assert.True(t, got["subnet-priv1"], "implicit main-RT subnet must receive add event") assert.Len(t, got, 2) @@ -934,6 +992,7 @@ func TestCreateRoute_IGW_OnMainRT_FansOutToImplicitSubnets(t *testing.T) { // subnet with an explicit non-main RT association does NOT receive the main // RT's IGW route event — its effective route table is the explicit one. func TestCreateRoute_IGW_OnMainRT_SkipsExplicitlyAssociatedSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) sub, err := svc.natsConn.SubscribeSync("vpc.add-igw-route") require.NoError(t, err) @@ -956,7 +1015,7 @@ func TestCreateRoute_IGW_OnMainRT_SkipsExplicitlyAssociatedSubnets(t *testing.T) }, testAccountID) require.NoError(t, err) - got := drainIGWSubnets(t, sub, 4) + got := drainIGWSubnets(t, svc, sub, 4) assert.True(t, got["subnet-test1"], "implicit subnet must receive add event") assert.False(t, got["subnet-priv1"], "explicitly-associated subnet must NOT receive main-RT event") } @@ -964,6 +1023,7 @@ func TestCreateRoute_IGW_OnMainRT_SkipsExplicitlyAssociatedSubnets(t *testing.T) // TestAssociateRouteTable_RemovesMainRTRoutesForJoiningSubnet verifies that the main // RT's per-subnet rules are torn down when a subnet joins an explicit non-main RT. func TestAssociateRouteTable_RemovesMainRTRoutesForJoiningSubnet(t *testing.T) { + t.Parallel() svc := setupTestService(t) delSub, err := svc.natsConn.SubscribeSync("vpc.delete-igw-route") require.NoError(t, err) @@ -985,7 +1045,7 @@ func TestAssociateRouteTable_RemovesMainRTRoutesForJoiningSubnet(t *testing.T) { }, testAccountID) require.NoError(t, err) - got := drainIGWSubnets(t, delSub, 2) + got := drainIGWSubnets(t, svc, delSub, 2) assert.True(t, got["subnet-priv1"], "joining subnet must receive delete event for main-RT routes") assert.Len(t, got, 1) } @@ -1005,11 +1065,11 @@ func receiveGateEvent(t *testing.T, sub *nats.Subscription) (subnetID, destCidr return evt.SubnetId, evt.DestinationCidr } -func drainGateSubnets(t *testing.T, sub *nats.Subscription, limit int) map[string]string { +func drainGateSubnets(t *testing.T, svc *RouteTableServiceImpl, sub *nats.Subscription, limit int) map[string]string { t.Helper() got := map[string]string{} - for range limit { - msg, err := sub.NextMsg(200 * time.Millisecond) + for range min(limit, pendingEvents(t, svc, sub)) { + msg, err := sub.NextMsg(time.Second) if err != nil { break } @@ -1026,6 +1086,7 @@ func drainGateSubnets(t *testing.T, sub *nats.Subscription, limit int) map[strin // TestCreateRoute_IGW_PublishesUngateForAssociatedSubnets: adding an IGW default route // to a table must flip associated subnets to "egress allowed" (ungate). func TestCreateRoute_IGW_PublishesUngateForAssociatedSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) ungate, err := svc.natsConn.SubscribeSync("vpc.ungate-subnet-egress") require.NoError(t, err) @@ -1038,11 +1099,7 @@ func TestCreateRoute_IGW_PublishesUngateForAssociatedSubnets(t *testing.T) { }, testAccountID) require.NoError(t, err) // Drain Associate-time gate event (gate, no egress yet). - for { - if _, err := ungate.NextMsg(100 * time.Millisecond); err != nil { - break - } - } + drainEvents(t, svc, ungate) _, err = svc.CreateRoute(t.Context(), &ec2.CreateRouteInput{ RouteTableId: aws.String(rtbID), @@ -1060,6 +1117,7 @@ func TestCreateRoute_IGW_PublishesUngateForAssociatedSubnets(t *testing.T) { // default route from a table flips associated subnets back to "no egress" // so the subscriber must reinstall the drop policy. func TestDeleteRoute_IGW_PublishesGateForAssociatedSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) gate, err := svc.natsConn.SubscribeSync("vpc.gate-subnet-egress") require.NoError(t, err) @@ -1077,11 +1135,7 @@ func TestDeleteRoute_IGW_PublishesGateForAssociatedSubnets(t *testing.T) { GatewayId: aws.String("igw-test1"), }, testAccountID) require.NoError(t, err) - for { - if _, err := gate.NextMsg(100 * time.Millisecond); err != nil { - break - } - } + drainEvents(t, svc, gate) _, err = svc.DeleteRoute(t.Context(), &ec2.DeleteRouteInput{ RouteTableId: aws.String(rtbID), @@ -1097,6 +1151,7 @@ func TestDeleteRoute_IGW_PublishesGateForAssociatedSubnets(t *testing.T) { // TestAssociateRouteTable_NoDefaultRoute_PublishesGate: joining a table that // has no 0.0.0.0/0 route must publish a gate event so the subnet is dropped. func TestAssociateRouteTable_NoDefaultRoute_PublishesGate(t *testing.T) { + t.Parallel() svc := setupTestService(t) gate, err := svc.natsConn.SubscribeSync("vpc.gate-subnet-egress") require.NoError(t, err) @@ -1118,6 +1173,7 @@ func TestAssociateRouteTable_NoDefaultRoute_PublishesGate(t *testing.T) { // already carries 0.0.0.0/0 -> IGW must publish ungate so any pre-existing // drop policy is removed. func TestAssociateRouteTable_HasIGWRoute_PublishesUngate(t *testing.T) { + t.Parallel() svc := setupTestService(t) ungate, err := svc.natsConn.SubscribeSync("vpc.ungate-subnet-egress") require.NoError(t, err) @@ -1145,6 +1201,7 @@ func TestAssociateRouteTable_HasIGWRoute_PublishesUngate(t *testing.T) { // TestDisassociateRouteTable_MainHasNoEgress_PublishesGate: leaving an // egress-carrying table for a main RT without 0.0.0.0/0 must gate. func TestDisassociateRouteTable_MainHasNoEgress_PublishesGate(t *testing.T) { + t.Parallel() svc := setupTestService(t) gate, err := svc.natsConn.SubscribeSync("vpc.gate-subnet-egress") require.NoError(t, err) @@ -1165,11 +1222,7 @@ func TestDisassociateRouteTable_MainHasNoEgress_PublishesGate(t *testing.T) { SubnetId: aws.String("subnet-priv1"), }, testAccountID) require.NoError(t, err) - for { - if _, err := gate.NextMsg(100 * time.Millisecond); err != nil { - break - } - } + drainEvents(t, svc, gate) _, err = svc.DisassociateRouteTable(t.Context(), &ec2.DisassociateRouteTableInput{ AssociationId: assocOut.AssociationId, @@ -1184,6 +1237,7 @@ func TestDisassociateRouteTable_MainHasNoEgress_PublishesGate(t *testing.T) { // TestCreateRoute_NonDefaultPrefix_NoGateEvent: only 0.0.0.0/0 routes drive // gate decisions today (subnet-level scoping is what AWS RTs gate on). func TestCreateRoute_NonDefaultPrefix_NoGateEvent(t *testing.T) { + t.Parallel() svc := setupTestService(t) gate, err := svc.natsConn.SubscribeSync("vpc.gate-subnet-egress") require.NoError(t, err) @@ -1198,12 +1252,7 @@ func TestCreateRoute_NonDefaultPrefix_NoGateEvent(t *testing.T) { SubnetId: aws.String("subnet-priv1"), }, testAccountID) require.NoError(t, err) - // Drain the Associate-time gate event. - for { - if _, err := gate.NextMsg(100 * time.Millisecond); err != nil { - break - } - } + drainEvents(t, svc, gate) _, err = svc.CreateRoute(t.Context(), &ec2.CreateRouteInput{ RouteTableId: aws.String(rtbID), @@ -1212,15 +1261,14 @@ func TestCreateRoute_NonDefaultPrefix_NoGateEvent(t *testing.T) { }, testAccountID) require.NoError(t, err) - _, err = gate.NextMsg(200 * time.Millisecond) - assert.Error(t, err, "non-default prefix must not emit a gate event") - _, err = ungate.NextMsg(200 * time.Millisecond) - assert.Error(t, err, "non-default prefix must not emit an ungate event") + assertNoEvents(t, svc, gate, "non-default prefix must not emit a gate event") + assertNoEvents(t, svc, ungate, "non-default prefix must not emit an ungate event") } // TestCreateRoute_IGW_OnMainRT_FansOutGateToImplicitSubnets: when the main // RT acquires an IGW default route, implicit-main subnets receive ungate. func TestCreateRoute_IGW_OnMainRT_FansOutUngateToImplicitSubnets(t *testing.T) { + t.Parallel() svc := setupTestService(t) ungate, err := svc.natsConn.SubscribeSync("vpc.ungate-subnet-egress") require.NoError(t, err) @@ -1235,12 +1283,13 @@ func TestCreateRoute_IGW_OnMainRT_FansOutUngateToImplicitSubnets(t *testing.T) { }, testAccountID) require.NoError(t, err) - got := drainGateSubnets(t, ungate, 3) + got := drainGateSubnets(t, svc, ungate, 3) assert.Equal(t, "0.0.0.0/0", got["subnet-test1"]) assert.Equal(t, "0.0.0.0/0", got["subnet-priv1"]) } func TestDisassociateRouteTable_RestoresMainRTRoutesForDepartingSubnet(t *testing.T) { + t.Parallel() svc := setupTestService(t) addSub, err := svc.natsConn.SubscribeSync("vpc.add-igw-route") require.NoError(t, err) @@ -1262,20 +1311,14 @@ func TestDisassociateRouteTable_RestoresMainRTRoutesForDepartingSubnet(t *testin }, testAccountID) require.NoError(t, err) - // Drain everything published during setup so the Disassociate assertions - // only see post-disassociate events. - for { - if _, err := addSub.NextMsg(100 * time.Millisecond); err != nil { - break - } - } + drainEvents(t, svc, addSub) _, err = svc.DisassociateRouteTable(t.Context(), &ec2.DisassociateRouteTableInput{ AssociationId: assocOut.AssociationId, }, testAccountID) require.NoError(t, err) - got := drainIGWSubnets(t, addSub, 2) + got := drainIGWSubnets(t, svc, addSub, 2) assert.True(t, got["subnet-priv1"], "departing subnet must receive add event for main-RT routes") assert.Len(t, got, 1) } @@ -1284,6 +1327,7 @@ func TestDisassociateRouteTable_RestoresMainRTRoutesForDepartingSubnet(t *testin // carrying only the local route, every subnet in the VPC must receive a gate // event. Exercises the IGW attach/detach fan-out hook. func TestPublishGateDecisionsForVPC_GatesAllSubnetsWhenNoEgress(t *testing.T) { + t.Parallel() svc := setupTestService(t) _, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "") require.NoError(t, err) @@ -1294,7 +1338,7 @@ func TestPublishGateDecisionsForVPC_GatesAllSubnetsWhenNoEgress(t *testing.T) { svc.PublishGateDecisionsForVPC(testAccountID, "vpc-test1", "0.0.0.0/0") - got := drainGateSubnets(t, gate, 4) + got := drainGateSubnets(t, svc, gate, 4) assert.Equal(t, "0.0.0.0/0", got["subnet-test1"]) assert.Equal(t, "0.0.0.0/0", got["subnet-priv1"]) assert.Len(t, got, 2) @@ -1304,6 +1348,7 @@ func TestPublishGateDecisionsForVPC_GatesAllSubnetsWhenNoEgress(t *testing.T) { // effective RT carries 0.0.0.0/0 -> IGW must receive ungate; the other // (implicit-main, no egress) subnet still receives gate. func TestPublishGateDecisionsForVPC_UngatesSubnetWithIGWRoute(t *testing.T) { + t.Parallel() svc := setupTestService(t) _, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "") require.NoError(t, err) @@ -1331,8 +1376,8 @@ func TestPublishGateDecisionsForVPC_UngatesSubnetWithIGWRoute(t *testing.T) { svc.PublishGateDecisionsForVPC(testAccountID, "vpc-test1", "0.0.0.0/0") - gateGot := drainGateSubnets(t, gate, 4) - ungateGot := drainGateSubnets(t, ungate, 4) + gateGot := drainGateSubnets(t, svc, gate, 4) + ungateGot := drainGateSubnets(t, svc, ungate, 4) assert.Equal(t, "0.0.0.0/0", ungateGot["subnet-test1"], "subnet-test1 has IGW route → ungate") assert.Equal(t, "0.0.0.0/0", gateGot["subnet-priv1"], "subnet-priv1 implicit-main no egress → gate") assert.Len(t, ungateGot, 1) @@ -1343,6 +1388,7 @@ func TestPublishGateDecisionsForVPC_UngatesSubnetWithIGWRoute(t *testing.T) { // for the same VPC and asserts effectiveRouteTable deterministically picks the one // with more routes (the "real" RT) over the orphan with only the implicit local route. func TestEffectiveRouteTable_DeterministicWithDuplicateMain(t *testing.T) { + t.Parallel() svc := setupTestService(t) now := time.Now() @@ -1383,6 +1429,7 @@ func TestEffectiveRouteTable_DeterministicWithDuplicateMain(t *testing.T) { // primary tiebreaker — a later-created RT with more routes beats an older // orphan with only the implicit local route. func TestPreferMain_RouteCountWinsOverCreatedAt(t *testing.T) { + t.Parallel() older := &RouteTableRecord{RouteTableId: "rtb-a", CreatedAt: time.Unix(100, 0), Routes: []RouteRecord{{}}} newer := &RouteTableRecord{RouteTableId: "rtb-b", CreatedAt: time.Unix(200, 0), Routes: []RouteRecord{{}, {}}} assert.True(t, preferMain(older, newer), "more routes wins regardless of CreatedAt") @@ -1392,6 +1439,7 @@ func TestPreferMain_RouteCountWinsOverCreatedAt(t *testing.T) { // TestPreferMain_CreatedAtTiebreak: equal route counts → oldest CreatedAt // wins (original record beats the late dup). func TestPreferMain_CreatedAtTiebreak(t *testing.T) { + t.Parallel() older := &RouteTableRecord{RouteTableId: "rtb-b", CreatedAt: time.Unix(100, 0), Routes: []RouteRecord{{}}} newer := &RouteTableRecord{RouteTableId: "rtb-a", CreatedAt: time.Unix(200, 0), Routes: []RouteRecord{{}}} assert.True(t, preferMain(newer, older), "older CreatedAt wins on equal route count") @@ -1401,6 +1449,7 @@ func TestPreferMain_CreatedAtTiebreak(t *testing.T) { // TestPublishGateDecisionsForVPC_EmptyVPCNoOp: VPC with zero subnets must // silently emit no events. func TestPublishGateDecisionsForVPC_EmptyVPCNoOp(t *testing.T) { + t.Parallel() svc := setupTestService(t) gate, err := svc.natsConn.SubscribeSync("vpc.gate-subnet-egress") require.NoError(t, err) @@ -1411,8 +1460,6 @@ func TestPublishGateDecisionsForVPC_EmptyVPCNoOp(t *testing.T) { svc.PublishGateDecisionsForVPC(testAccountID, "vpc-empty", "0.0.0.0/0") - _, err = gate.NextMsg(150 * time.Millisecond) - assert.Error(t, err, "expected no gate event") - _, err = ungate.NextMsg(150 * time.Millisecond) - assert.Error(t, err, "expected no ungate event") + assertNoEvents(t, svc, gate, "expected no gate event") + assertNoEvents(t, svc, ungate, "expected no ungate event") }