src/local_aggregation_1_1/aggregation_procedures/sum_and_count_aggregation_procedure_test.cc

// Copyright 2020 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "src/local_aggregation_1_1/aggregation_procedures/sum_and_count_aggregation_procedure.h"

#include <gtest/gtest.h>

#include "src/local_aggregation_1_1/aggregation_procedures/aggregation_procedure.h"
#include "src/local_aggregation_1_1/aggregation_procedures/testing/test_aggregation_procedure.h"
#include "src/local_aggregation_1_1/testing/test_registry.cb.h"
#include "src/logger/project_context_factory.h"

namespace cobalt::local_aggregation {

class SumAndCountAggregationProcedureTest : public testing::TestAggregationProcedure {
 protected:
  std::unique_ptr<SumAndCountAggregationProcedure> GetProcedure(uint32_t metric_id,
                                                                int report_index) {
    return std::make_unique<SumAndCountAggregationProcedure>(GetMetricDef(metric_id),
                                                             GetReportDef(metric_id, report_index));
  }
};

TEST_F(SumAndCountAggregationProcedureTest, UpdateAggregateWorks) {
  uint32_t metric_id = kIntegerMetricMetricId;
  int report_index = kIntegerMetricFleetwideMeansReportReportIndex;
  std::unique_ptr<AggregationProcedure> procedure = GetProcedureFor(metric_id, report_index);

  ReportAggregate aggregate;
  const uint32_t kNumEventCodes = 100;
  ASSERT_GE(GetReportDef(metric_id, report_index).event_vector_buffer_max(), kNumEventCodes);

  const int64_t kValue = 42;
  const uint32_t kHourId = 1;
  const uint64_t system_profile_hash = uint64_t{634354};
  AddIntegerEvents(kNumEventCodes, kValue, kHourId, system_profile_hash, *procedure, aggregate);

  ASSERT_EQ(aggregate.hourly().by_hour_id().count(kHourId), 1u);
  ASSERT_EQ(aggregate.hourly().by_hour_id().at(kHourId).system_profile_aggregates_size(), 1u);
  const SystemProfileAggregate &system_profile_agg =
      aggregate.hourly().by_hour_id().at(kHourId).system_profile_aggregates(0);
  EXPECT_EQ(system_profile_agg.system_profile_hash(), system_profile_hash);
  EXPECT_EQ(system_profile_agg.by_event_code_size(), kNumEventCodes);

  for (uint32_t i = 0; i < kNumEventCodes; ++i) {
    EXPECT_EQ(system_profile_agg.by_event_code(i).data().sum_and_count().count(), i + 1);
    EXPECT_EQ(system_profile_agg.by_event_code(i).data().sum_and_count().sum(), (i + 1) * kValue);
  }
}

TEST_F(SumAndCountAggregationProcedureTest, MergeAggregateDataBothSet) {
  AggregateData data;
  data.mutable_sum_and_count()->set_count(10);
  data.mutable_sum_and_count()->set_sum(200);
  AggregateData merged_data;
  merged_data.mutable_sum_and_count()->set_count(20);
  merged_data.mutable_sum_and_count()->set_sum(100);

  std::unique_ptr<SumAndCountAggregationProcedure> procedure =
      GetProcedure(kOccurrenceMetricMetricId, kOccurrenceMetricHourlyDeviceHistogramsReportIndex);
  procedure->MergeAggregateData(merged_data, data);

  EXPECT_EQ(merged_data.sum_and_count().count(), 30);
  EXPECT_EQ(merged_data.sum_and_count().sum(), 300);
}

TEST_F(SumAndCountAggregationProcedureTest, MergeAggregateDataNeitherSet) {
  AggregateData data;
  AggregateData merged_data;

  std::unique_ptr<SumAndCountAggregationProcedure> procedure =
      GetProcedure(kOccurrenceMetricMetricId, kOccurrenceMetricHourlyDeviceHistogramsReportIndex);
  procedure->MergeAggregateData(merged_data, data);

  EXPECT_EQ(merged_data.sum_and_count().count(), 0);
  EXPECT_EQ(merged_data.sum_and_count().sum(), 0);
}

TEST_F(SumAndCountAggregationProcedureTest, MergeAggregateDataFromSet) {
  AggregateData data;
  data.mutable_sum_and_count()->set_count(10);
  data.mutable_sum_and_count()->set_sum(200);
  AggregateData merged_data;

  std::unique_ptr<SumAndCountAggregationProcedure> procedure =
      GetProcedure(kOccurrenceMetricMetricId, kOccurrenceMetricHourlyDeviceHistogramsReportIndex);
  procedure->MergeAggregateData(merged_data, data);

  EXPECT_EQ(merged_data.sum_and_count().count(), 10);
  EXPECT_EQ(merged_data.sum_and_count().sum(), 200);
}

TEST_F(SumAndCountAggregationProcedureTest, MergeAggregateDataToSet) {
  AggregateData data;
  AggregateData merged_data;
  merged_data.mutable_sum_and_count()->set_count(20);
  merged_data.mutable_sum_and_count()->set_sum(100);

  std::unique_ptr<SumAndCountAggregationProcedure> procedure =
      GetProcedure(kOccurrenceMetricMetricId, kOccurrenceMetricHourlyDeviceHistogramsReportIndex);
  procedure->MergeAggregateData(merged_data, data);

  EXPECT_EQ(merged_data.sum_and_count().count(), 20);
  EXPECT_EQ(merged_data.sum_and_count().sum(), 100);
}

TEST_F(SumAndCountAggregationProcedureTest, GenerateObservationWorks) {
  uint32_t metric_id = kIntegerMetricMetricId;
  int report_index = kIntegerMetricFleetwideMeansReportReportIndex;
  std::unique_ptr<AggregationProcedure> procedure = GetProcedureFor(metric_id, report_index);

  ReportAggregate aggregate;
  const uint32_t kNumEventCodes = 10;
  ASSERT_GE(GetReportDef(metric_id, report_index).event_vector_buffer_max(), kNumEventCodes);

  const int64_t kValue = 42;
  const uint32_t kEndHourId = 11;
  const uint64_t system_profile_hash = uint64_t{634354};
  for (int hour_id = 1; hour_id <= kEndHourId; hour_id += 2) {
    AddIntegerEvents(kNumEventCodes, kValue, hour_id, system_profile_hash, *procedure, aggregate);
  }

  util::TimeInfo info;
  info.hour_id = kEndHourId;
  lib::statusor::StatusOr<std::vector<ObservationAndSystemProfile>> observations_or =
      procedure->GenerateObservations(info, aggregate);
  ASSERT_EQ(observations_or.status().error_code(), StatusCode::OK);
  std::vector<ObservationAndSystemProfile> observations = std::move(observations_or).value();

  // Should only generate for kEndHourId
  ASSERT_EQ(observations.size(), 1u);
  EXPECT_EQ(observations[0].system_profile_hash, system_profile_hash);

  EXPECT_EQ(observations[0].observation->sum_and_count().sums_and_counts_size(), kNumEventCodes);

  for (const SumAndCountObservation_SumAndCount &sum_and_count :
       observations[0].observation->sum_and_count().sums_and_counts()) {
    EXPECT_EQ(sum_and_count.event_codes(0), sum_and_count.count());
    EXPECT_EQ(sum_and_count.event_codes(0) * kValue, sum_and_count.sum());
  }
  // Check that obsolete aggregates get cleaned up.
  procedure->ObservationsCommitted(aggregate, info, system_profile_hash);
  EXPECT_EQ(aggregate.hourly().by_hour_id_size(), 0);
}

}  // namespace cobalt::local_aggregation