Add TestPipelinesAggregatorForEachReader

Test each pipeline is updated with the aggregators it syncs.

sdk/metric/pipeline_registry_test.go

@@ -227,6 +227,11 @@ func testCreateAggregators[N int64 | float64](t *testing.T) {
 	}
 }
 
+func TestCreateAggregators(t *testing.T) {
+	t.Run("Int64", testCreateAggregators[int64])
+	t.Run("Float64", testCreateAggregators[float64])
+}
+
 func testInvalidInstrumentShouldPanic[N int64 | float64]() {
 	c := newInstrumentCache[N](nil, nil)
 	i := newInserter(newPipeline(nil, NewManualReader(), []View{defaultView}), c)
@@ -242,9 +247,25 @@ func TestInvalidInstrumentShouldPanic(t *testing.T) {
 	assert.Panics(t, testInvalidInstrumentShouldPanic[float64])
 }
 
-func TestCreateAggregators(t *testing.T) {
-	t.Run("Int64", testCreateAggregators[int64])
-	t.Run("Float64", testCreateAggregators[float64])
+func TestPipelinesAggregatorForEachReader(t *testing.T) {
+	r0, r1 := NewManualReader(), NewManualReader()
+	pipes := newPipelines(resource.Empty(), []Reader{r0, r1}, nil)
+	require.Len(t, pipes, 2, "created pipelines")
+
+	inst := Instrument{Name: "foo", Kind: InstrumentKindCounter}
+	c := newInstrumentCache[int64](nil, nil)
+	r := newResolver(pipes, c)
+	aggs, err := r.Aggregators(inst)
+	require.NoError(t, err, "resolved Aggregators error")
+	require.Len(t, aggs, 2, "instrument aggregators")
+
+	for i, p := range pipes {
+		var aggN int
+		for _, is := range p.aggregations {
+			aggN += len(is)
+		}
+		assert.Equalf(t, 1, aggN, "pipeline %d: number of instrumentSync", i)
+	}
 }
 
 func TestPipelineRegistryCreateAggregators(t *testing.T) {