code: def test_variance_calls_aggregate_with_correct_params(
      self, contribution_bounds_already_enforced: bool, mock_aggregate
  ):
      # Arrange
      dist_data = PrivateRDDTest.sc.parallelize(
          [(1, 0.0, "pk1"), (2, 10.0, "pk1")]
      )
      MetricsTuple = collections.namedtuple(
          "MetricsTuple", ["variance"]
      )
      mock_aggregate.return_value = PrivateRDDTest.sc.parallelize(
          [("pk1", MetricsTuple(variance=25.0))]
      )
      budget_accountant = budget_accounting.NaiveBudgetAccountant(
          1, 1e-10
      )
  
      def privacy_id_extractor(x):
          return x[1]
  
      prdd = private_spark.make_private(
          dist_data, budget_accountant, privacy_id_extractor
      )
      variance_params = agg.VarianceParams(
          noise_kind=pipeline_dp.NoiseKind.GAUSSIAN,
          max_partitions_contributed=2,
          max_contributions_per_partition=3,
          min_value=1.5,
          max_value=5.78,
          budget_weight=1.1,
          partition_extractor=lambda x: x[0],
          value_extractor=lambda x: x,
          contribution_bounds_already_enforced=contribution_bounds_already_enforced,
      )
  
      # Act
      actual_result = prdd.variance(variance_params)
  
      # Assert
      mock_aggregate.assert_called_once()
      args = mock_aggregate.call_args[0]
  
      rdd = dist_data.map(lambda x: (privacy_id_extractor(x), x))
      self.assertListEqual(args[0].collect(), rdd.collect())
  
      params = pipeline_dp.AggregateParams(
          noise_kind=pipeline_dp.NoiseKind.GAUSSIAN,
          metrics=[pipeline_dp.Metrics.VARIANCE],
          max_partitions_contributed=variance_params.max_partitions_contributed,
          max_contributions_per_partition=variance_params.max_contributions_per_partition,
          min_value=variance_params.min_value,
          max_value=variance_params.max_value,
          budget_weight=variance_params.budget_weight,
          contribution_bounds_already_enforced=contribution_bounds_already_enforced,
      )
      self.assertEqual(args[1], params)
  
      self.assertEqual(actual_result.collect(), [("pk1", 25.0)])
  
imports: import pyspark
  from pyspark import SparkContext
  from absl.testing import parameterized
  from unittest.mock import patch
  import unittest
  import sys
  import collections
  import pipeline_dp
  from pipeline_dp import aggregate_params
  from pipeline_dp import budget_accounting, private_spark
label_desc: Use image node_id 6 for calling the PrivateRDDTest obj's underlying member method code with example usage: obj.test_variance_calls_aggregate_with_correct_params(contribution_bounds_already_enforced, mock_aggregate) and returns: x