test: Add report creation to synthetic correctness test

.github/workflows/run-k8s-tests.yml

@@ -52,13 +52,15 @@ jobs:
         MC_NAME: ${{ vars.MC_NAME }}
         MC_API_KEY: ${{ secrets.MC_API_KEY }}
         GCLOUD_PROJECT: ${{ vars.GCLOUD_PROJECT }}
+        REPORTING_PUBLIC_API_TARGET: v2alpha.reporting.dev.halo-cmm.org:8443
       run: |
         cat << EOF > ~/.bazelrc
         common --config=ci
         build --define kingdom_public_api_target=$KINGDOM_PUBLIC_API_TARGET
         build --define mc_name=$MC_NAME
         build --define mc_api_key=$MC_API_KEY
         build --define google_cloud_project=$GCLOUD_PROJECT
+        build --define reporting_public_api_target=$REPORTING_PUBLIC_API_TARGET
         test --test_output=streamed
         test --test_timeout=3600
         EOF

docs/gke/correctness-test.md

@@ -80,7 +80,8 @@ bazel test //src/test/kotlin/org/wfanet/measurement/integration/k8s:SyntheticGen
 --test_output=streamed \
 --define=kingdom_public_api_target=v2alpha.kingdom.dev.halo-cmm.org:8443 \
 --define=mc_name=measurementConsumers/Rcn7fKd25C8 \
---define=mc_api_key=W9q4zad246g
+--define=mc_api_key=W9q4zad246g \
+--define=reporting_public_api_target=v2alpha.reporting.dev.halo-cmm.org:8443
 ```
 
 The time the test takes depends on the size of the data set. With the default

src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/BUILD.bazel

@@ -8,6 +8,7 @@ package(default_visibility = [
     "//src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/testing:__pkg__",
     "//src/main/kotlin/org/wfanet/measurement/loadtest/panelmatch:__pkg__",
     "//src/main/kotlin/org/wfanet/measurement/loadtest/panelmatchresourcesetup:__pkg__",
+    "//src/main/kotlin/org/wfanet/measurement/loadtest/reporting:__pkg__",
     "//src/main/kotlin/org/wfanet/measurement/loadtest/resourcesetup:__pkg__",
     "//src/test/kotlin/org/wfanet/measurement/integration/common:__pkg__",
 ])

src/main/kotlin/org/wfanet/measurement/loadtest/reporting/BUILD.bazel

@@ -0,0 +1,26 @@
+load("@wfa_rules_kotlin_jvm//kotlin:defs.bzl", "kt_jvm_library")
+
+package(
+    default_testonly = True,
+    default_visibility = [
+        "//src/main/kotlin/org/wfanet/measurement/integration:__subpackages__",
+        "//src/main/kotlin/org/wfanet/measurement/loadtest:__subpackages__",
+        "//src/test/kotlin/org/wfanet/measurement/integration:__subpackages__",
+        "//src/test/kotlin/org/wfanet/measurement/loadtest:__subpackages__",
+    ],
+)
+
+kt_jvm_library(
+    name = "simulator",
+    srcs = ["ReportingUserSimulator.kt"],
+    deps = [
+        "//src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha:data_providers_service",
+        "//src/main/kotlin/org/wfanet/measurement/loadtest/config:test_identifiers",
+        "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v2alpha:event_groups_service",
+        "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v2alpha:metric_calculation_specs_service",
+        "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v2alpha:reporting_sets_service",
+        "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v2alpha:reports_service",
+        "//src/main/proto/wfa/measurement/api/v2alpha/event_templates/testing:test_event_kt_jvm_proto",
+        "@wfa_common_jvm//imports/java/com/google/common/truth/extensions/proto",
+    ],
+)

src/main/kotlin/org/wfanet/measurement/loadtest/reporting/ReportingUserSimulator.kt

@@ -0,0 +1,241 @@
+/*
+ * Copyright 2024 The Cross-Media Measurement Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.wfanet.measurement.loadtest.reporting
+
+import com.google.common.truth.Truth.assertThat
+import com.google.type.DayOfWeek
+import com.google.type.date
+import com.google.type.dateTime
+import com.google.type.timeZone
+import io.grpc.StatusException
+import java.util.logging.Logger
+import kotlinx.coroutines.delay
+import org.wfanet.measurement.api.v2alpha.DataProvider
+import org.wfanet.measurement.api.v2alpha.DataProvidersGrpcKt
+import org.wfanet.measurement.api.v2alpha.getDataProviderRequest
+import org.wfanet.measurement.api.withAuthenticationKey
+import org.wfanet.measurement.loadtest.config.TestIdentifiers
+import org.wfanet.measurement.reporting.v2alpha.EventGroup
+import org.wfanet.measurement.reporting.v2alpha.EventGroupsGrpcKt
+import org.wfanet.measurement.reporting.v2alpha.MetricCalculationSpec
+import org.wfanet.measurement.reporting.v2alpha.MetricCalculationSpecKt
+import org.wfanet.measurement.reporting.v2alpha.MetricCalculationSpecsGrpcKt
+import org.wfanet.measurement.reporting.v2alpha.MetricSpecKt
+import org.wfanet.measurement.reporting.v2alpha.Report
+import org.wfanet.measurement.reporting.v2alpha.ReportKt
+import org.wfanet.measurement.reporting.v2alpha.ReportingSet
+import org.wfanet.measurement.reporting.v2alpha.ReportingSetKt
+import org.wfanet.measurement.reporting.v2alpha.ReportingSetsGrpcKt
+import org.wfanet.measurement.reporting.v2alpha.ReportsGrpcKt
+import org.wfanet.measurement.reporting.v2alpha.createMetricCalculationSpecRequest
+import org.wfanet.measurement.reporting.v2alpha.createReportRequest
+import org.wfanet.measurement.reporting.v2alpha.createReportingSetRequest
+import org.wfanet.measurement.reporting.v2alpha.getReportRequest
+import org.wfanet.measurement.reporting.v2alpha.listEventGroupsRequest
+import org.wfanet.measurement.reporting.v2alpha.metricCalculationSpec
+import org.wfanet.measurement.reporting.v2alpha.metricSpec
+import org.wfanet.measurement.reporting.v2alpha.report
+import org.wfanet.measurement.reporting.v2alpha.reportingSet
+
+/** Simulator for Reporting operations on the Reporting public API. */
+class ReportingUserSimulator(
+  private val measurementConsumerName: String,
+  private val apiAuthenticationKey: String,
+  private val dataProvidersClient: DataProvidersGrpcKt.DataProvidersCoroutineStub,
+  private val eventGroupsClient: EventGroupsGrpcKt.EventGroupsCoroutineStub,
+  private val reportingSetsClient: ReportingSetsGrpcKt.ReportingSetsCoroutineStub,
+  private val metricCalculationSpecsClient:
+    MetricCalculationSpecsGrpcKt.MetricCalculationSpecsCoroutineStub,
+  private val reportsClient: ReportsGrpcKt.ReportsCoroutineStub,
+) {
+  suspend fun testCreateReport(runId: String) {
+    logger.info("Creating report...")
+
+    val eventGroup =
+      listEventGroups()
+        .filter {
+          it.eventGroupReferenceId.startsWith(
+            TestIdentifiers.SIMULATOR_EVENT_GROUP_REFERENCE_ID_PREFIX
+          )
+        }
+        .firstOrNull {
+          getDataProvider(it.cmmsDataProvider).capabilities.honestMajorityShareShuffleSupported
+        } ?: listEventGroups().first()
+    val createdPrimitiveReportingSet = createPrimitiveReportingSet(eventGroup)
+    val createdMetricCalculationSpec = createMetricCalculationSpec()
+
+    val report = report {
+      reportingMetricEntries +=
+        ReportKt.reportingMetricEntry {
+          key = createdPrimitiveReportingSet.name
+          value =
+            ReportKt.reportingMetricCalculationSpec {
+              metricCalculationSpecs += createdMetricCalculationSpec.name
+            }
+        }
+      reportingInterval =
+        ReportKt.reportingInterval {
+          reportStart = dateTime {
+            year = 2024
+            month = 1
+            day = 3
+            timeZone = timeZone { id = "America/Los_Angeles" }
+          }
+          reportEnd = date {
+            year = 2024
+            month = 1
+            day = 4
+          }
+        }
+    }
+
+    val createdReport =
+      try {
+        reportsClient
+          .withAuthenticationKey(apiAuthenticationKey)
+          .createReport(
+            createReportRequest {
+              parent = measurementConsumerName
+              this.report = report
+              reportId = "a-$runId"
+            }
+          )
+      } catch (e: StatusException) {
+        throw Exception("Error creating Report", e)
+      }
+
+    val completedReport = pollForCompletedReport(createdReport.name)
+
+    assertThat(completedReport.state).isEqualTo(Report.State.SUCCEEDED)
+    logger.info("Report creation succeeded")
+  }
+
+  private suspend fun listEventGroups(): List<EventGroup> {
+    try {
+      return eventGroupsClient
+        .withAuthenticationKey(apiAuthenticationKey)
+        .listEventGroups(
+          listEventGroupsRequest {
+            parent = measurementConsumerName
+            pageSize = 1000
+          }
+        )
+        .eventGroupsList
+    } catch (e: StatusException) {
+      throw Exception("Error listing EventGroups", e)
+    }
+  }
+
+  private suspend fun getDataProvider(dataProviderName: String): DataProvider {
+    try {
+      return dataProvidersClient
+        .withAuthenticationKey(apiAuthenticationKey)
+        .getDataProvider(getDataProviderRequest { name = dataProviderName })
+    } catch (e: StatusException) {
+      throw Exception("Error getting DataProvider $dataProviderName", e)
+    }
+  }
+
+  private suspend fun createPrimitiveReportingSet(eventGroup: EventGroup): ReportingSet {
+    val primitiveReportingSet = reportingSet {
+      primitive = ReportingSetKt.primitive { cmmsEventGroups += eventGroup.cmmsEventGroup }
+    }
+
+    try {
+      return reportingSetsClient
+        .withAuthenticationKey(apiAuthenticationKey)
+        .createReportingSet(
+          createReportingSetRequest {
+            parent = measurementConsumerName
+            reportingSet = primitiveReportingSet
+            reportingSetId = "a-123"
+          }
+        )
+    } catch (e: StatusException) {
+      throw Exception("Error creating ReportingSet", e)
+    }
+  }
+
+  private suspend fun createMetricCalculationSpec(): MetricCalculationSpec {
+    try {
+      return metricCalculationSpecsClient
+        .withAuthenticationKey(apiAuthenticationKey)
+        .createMetricCalculationSpec(
+          createMetricCalculationSpecRequest {
+            parent = measurementConsumerName
+            metricCalculationSpecId = "a-123"
+            metricCalculationSpec = metricCalculationSpec {
+              displayName = "union reach"
+              metricSpecs += metricSpec {
+                reach =
+                  MetricSpecKt.reachParams {
+                    singleDataProviderParams =
+                      MetricSpecKt.samplingAndPrivacyParams {
+                        privacyParams = MetricSpecKt.differentialPrivacyParams {}
+                      }
+                    multipleDataProviderParams =
+                      MetricSpecKt.samplingAndPrivacyParams {
+                        privacyParams = MetricSpecKt.differentialPrivacyParams {}
+                      }
+                  }
+              }
+              metricFrequencySpec =
+                MetricCalculationSpecKt.metricFrequencySpec {
+                  weekly =
+                    MetricCalculationSpecKt.MetricFrequencySpecKt.weekly {
+                      dayOfWeek = DayOfWeek.WEDNESDAY
+                    }
+                }
+              trailingWindow =
+                MetricCalculationSpecKt.trailingWindow {
+                  count = 1
+                  increment = MetricCalculationSpec.TrailingWindow.Increment.WEEK
+                }
+            }
+          }
+        )
+    } catch (e: StatusException) {
+      throw Exception("Error creating MetricCalculationSpec", e)
+    }
+  }
+
+  private suspend fun pollForCompletedReport(reportName: String): Report {
+    while (true) {
+      val retrievedReport =
+        try {
+          reportsClient
+            .withAuthenticationKey(apiAuthenticationKey)
+            .getReport(getReportRequest { name = reportName })
+        } catch (e: StatusException) {
+          throw Exception("Error getting Report", e)
+        }
+
+      @Suppress("WHEN_ENUM_CAN_BE_NULL_IN_JAVA") // Proto enum fields are never null.
+      when (retrievedReport.state) {
+        Report.State.SUCCEEDED,
+        Report.State.FAILED -> return retrievedReport
+        Report.State.RUNNING,
+        Report.State.UNRECOGNIZED,
+        Report.State.STATE_UNSPECIFIED -> delay(5000)
+      }
+    }
+  }
+
+  companion object {
+    private val logger: Logger = Logger.getLogger(this::class.java.name)
+  }
+}

src/main/proto/wfa/measurement/integration/k8s/testing/correctness_test_config.proto

@@ -38,4 +38,13 @@ message CorrectnessTestConfig {
 
   // Authentication key for the CMMS public API.
   string api_authentication_key = 4;
+
+  // gRPC target of Reporting public API server.
+  string reporting_public_api_target = 5;
+
+  // Expected hostname (DNS-ID) in the reporting public API server's TLS
+  // certificate.
+  //
+  // If not specified, standard TLS DNS-ID derivation will be used.
+  string reporting_public_api_cert_host = 6;
 }

src/test/kotlin/org/wfanet/measurement/integration/k8s/AbstractCorrectnessTest.kt

@@ -28,6 +28,7 @@ import org.wfanet.measurement.common.crypto.SigningKeyHandle
 import org.wfanet.measurement.integration.common.loadEncryptionPrivateKey
 import org.wfanet.measurement.integration.common.loadSigningKey
 import org.wfanet.measurement.loadtest.measurementconsumer.MeasurementConsumerSimulator
+import org.wfanet.measurement.loadtest.reporting.ReportingUserSimulator
 
 /** Test for correctness of the CMMS on Kubernetes. */
 abstract class AbstractCorrectnessTest(private val measurementSystem: MeasurementSystem) {
@@ -37,6 +38,9 @@ abstract class AbstractCorrectnessTest(private val measurementSystem: Measuremen
   private val testHarness: MeasurementConsumerSimulator
     get() = measurementSystem.testHarness
 
+  private val reportingTestHarness: ReportingUserSimulator?
+    get() = measurementSystem.reportingTestHarness
+
   @Test(timeout = 1 * 60 * 1000)
   fun `impression measurement completes with expected result`() = runBlocking {
     testHarness.testImpression("$runId-impression")
@@ -63,9 +67,17 @@ abstract class AbstractCorrectnessTest(private val measurementSystem: Measuremen
     )
   }
 
+  @Test(timeout = 1 * 60 * 1000)
+  fun `report can be created`() = runBlocking {
+    if (reportingTestHarness != null) {
+      reportingTestHarness!!.testCreateReport("$runId-test-report")
+    }
+  }
+
   interface MeasurementSystem {
     val runId: String
     val testHarness: MeasurementConsumerSimulator
+    val reportingTestHarness: ReportingUserSimulator?
   }
 
   companion object {
@@ -97,6 +109,14 @@ abstract class AbstractCorrectnessTest(private val measurementSystem: Measuremen
       SigningCerts.fromPemFiles(cert, key, trustedCerts)
     }
 
+    val REPORTING_SIGNING_CERTS: SigningCerts by lazy {
+      val secretFiles = getRuntimePath(SECRET_FILES_PATH)
+      val trustedCerts = secretFiles.resolve("reporting_root.pem").toFile()
+      val cert = secretFiles.resolve("mc_tls.pem").toFile()
+      val key = secretFiles.resolve("mc_tls.key").toFile()
+      SigningCerts.fromPemFiles(cert, key, trustedCerts)
+    }
+
     val MC_ENCRYPTION_PRIVATE_KEY: PrivateKeyHandle by lazy {
       loadEncryptionPrivateKey(MC_ENCRYPTION_PRIVATE_KEY_NAME)
     }

src/test/kotlin/org/wfanet/measurement/integration/k8s/BUILD.bazel

@@ -13,11 +13,13 @@ kt_jvm_library(
     srcs = ["AbstractCorrectnessTest.kt"],
     data = [
         "//src/main/k8s/testing/secretfiles:mc_trusted_certs.pem",
+        "//src/main/k8s/testing/secretfiles:reporting_root.pem",
         "//src/main/k8s/testing/secretfiles:secret_files",
     ],
     deps = [
         "//src/main/kotlin/org/wfanet/measurement/integration/common:configs",
         "//src/main/kotlin/org/wfanet/measurement/loadtest/measurementconsumer:simulator",
+        "//src/main/kotlin/org/wfanet/measurement/loadtest/reporting:simulator",
         "@wfa_common_jvm//imports/java/com/google/common/truth",
         "@wfa_common_jvm//imports/java/org/junit",
         "@wfa_common_jvm//imports/kotlin/kotlinx/coroutines:core",
@@ -58,6 +60,7 @@ kt_jvm_library(
         "//src/main/kotlin/org/wfanet/measurement/integration/common:synthetic_generation_specs",
         "//src/main/kotlin/org/wfanet/measurement/loadtest/config:vid_sampling",
         "//src/main/kotlin/org/wfanet/measurement/loadtest/measurementconsumer:synthetic_generator_event_query",
+        "//src/main/kotlin/org/wfanet/measurement/loadtest/reporting:simulator",
         "//src/main/proto/wfa/measurement/integration/k8s/testing:correctness_test_config_kt_jvm_proto",
         "@wfa_common_jvm//imports/java/org/junit",
         "@wfa_common_jvm//imports/kotlin/kotlinx/coroutines:core",
@@ -74,6 +77,8 @@ expand_template(
         "{kingdom_public_api_cert_host}": "localhost",
         "{mc_name}": TEST_K8S_SETTINGS.mc_name,
         "{mc_api_key}": TEST_K8S_SETTINGS.mc_api_key,
+        "{reporting_public_api_target}": "$(reporting_public_api_target)",
+        "{reporting_public_api_cert_host}": "localhost",
     },
     tags = ["manual"],
     template = "correctness_test_config.tmpl.textproto",