Add UploadClient

UploadClient handles uploading records.

Bug: chromium:1078512
Change-Id: I76d3d23cabdb14c7a02b666daed094e8b57f1869
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2316688
Commit-Queue: Zach Trudo <zatrudo@google.com>
Reviewed-by: Leonid Baraz <lbaraz@chromium.org>
Cr-Commit-Position: refs/heads/master@{#797051}

chrome/browser/BUILD.gn

@@ -1178,6 +1178,8 @@ static_library("browser") {
     "policy/messaging_layer/upload/app_install_report_handler.h",
     "policy/messaging_layer/upload/dm_server_upload_service.cc",
     "policy/messaging_layer/upload/dm_server_upload_service.h",
+    "policy/messaging_layer/upload/upload_client.cc",
+    "policy/messaging_layer/upload/upload_client.h",
     "policy/messaging_layer/util/backoff_settings.cc",
     "policy/messaging_layer/util/backoff_settings.h",
     "policy/messaging_layer/util/shared_queue.h",

chrome/browser/policy/messaging_layer/public/report_client.cc

@@ -7,6 +7,8 @@
 #include <memory>
 #include <utility>
 
+#include "base/bind.h"
+#include "base/callback.h"
 #include "base/memory/ptr_util.h"
 #include "base/no_destructor.h"
 #include "base/path_service.h"
@@ -17,38 +19,191 @@
 #include "chrome/browser/policy/messaging_layer/util/status.h"
 #include "chrome/browser/policy/messaging_layer/util/status_macros.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "chrome/browser/policy/messaging_layer/util/task_runner_context.h"
 #include "chrome/common/chrome_paths.h"
+#include "components/enterprise/browser/controller/browser_dm_token_storage.h"
+#include "components/policy/core/common/cloud/device_management_service.h"
+
+#include "chrome/browser/browser_process.h"
+#include "chrome/browser/browser_process_platform_part.h"
+#include "chrome/browser/net/system_network_context_manager.h"
+#include "services/network/public/cpp/shared_url_loader_factory.h"
+
+#if defined(OS_CHROMEOS)
+#include "chrome/browser/chromeos/login/users/chrome_user_manager.h"
+#include "chrome/browser/chromeos/policy/browser_policy_connector_chromeos.h"
+#else
+#include "chrome/browser/policy/chrome_browser_policy_connector.h"
+#endif
 
 namespace reporting {
 
 namespace {
 
+// policy::CloudPolicyClient is needed by the UploadClient, but is built in two
+// different ways for ChromeOS and non-ChromeOS browsers.
+#if defined(OS_CHROMEOS)
+std::unique_ptr<policy::CloudPolicyClient> BuildCloudPolicyClient() {
+  policy::DeviceManagementService* const device_management_service =
+      g_browser_process->browser_policy_connector()
+          ->device_management_service();
+
+  scoped_refptr<network::SharedURLLoaderFactory>
+      signin_profile_url_loader_factory =
+          g_browser_process->system_network_context_manager()
+              ->GetSharedURLLoaderFactory();
+
+  auto* user_manager_ptr = g_browser_process->platform_part()->user_manager();
+  auto* primary_user = user_manager_ptr->GetPrimaryUser();
+
+  auto dm_token_getter = chromeos::GetDeviceDMTokenForUserPolicyGetter(
+      primary_user->GetAccountId());
+
+  auto client = std::make_unique<policy::CloudPolicyClient>(
+      device_management_service, signin_profile_url_loader_factory,
+      dm_token_getter);
+
+  policy::CloudPolicyClient::RegistrationParameters registration(
+      enterprise_management::DeviceRegisterRequest::USER,
+      enterprise_management::DeviceRegisterRequest::FLAVOR_USER_REGISTRATION);
+
+  // Register the client with the device management service.
+  client->Register(registration,
+                   /*client_id=*/std::string(),
+                   /*oauth_token=*/"oauth_token_unused");
+  return client;
+}
+#else
+std::unique_ptr<policy::CloudPolicyClient> BuildCloudPolicyClient() {
+  policy::DeviceManagementService* const device_management_service =
+      g_browser_process->browser_policy_connector()
+          ->device_management_service();
+
+  scoped_refptr<network::SharedURLLoaderFactory>
+      signin_profile_url_loader_factory =
+          g_browser_process->system_network_context_manager()
+              ->GetSharedURLLoaderFactory();
+
+  auto client = std::make_unique<policy::CloudPolicyClient>(
+      device_management_service, signin_profile_url_loader_factory,
+      policy::CloudPolicyClient::DeviceDMTokenCallback());
+
+  policy::DMToken browser_dm_token =
+      policy::BrowserDMTokenStorage::Get()->RetrieveDMToken();
+  std::string client_id =
+      policy::BrowserDMTokenStorage::Get()->RetrieveClientId();
+
+  client->SetupRegistration(browser_dm_token.value(), client_id,
+                            std::vector<std::string>());
+  return client;
+}
+#endif
+
 const base::FilePath::CharType kReportingDirectory[] =
     FILE_PATH_LITERAL("reporting");
 
 }  // namespace
 
-class ReportingClient::UploadClient : public Storage::UploaderInterface {
- public:
-  static StatusOr<std::unique_ptr<Storage::UploaderInterface>> Build(
-      Priority priority) {
-    // Cannot use make_unique, since constructor is private.
-    return base::WrapUnique(new UploadClient());
+using Uploader = ReportingClient::Uploader;
+
+Uploader::Uploader(UploadCallback upload_callback)
+    : upload_callback_(std::move(upload_callback)),
+      completed_(false),
+      encrypted_records_(std::make_unique<std::vector<EncryptedRecord>>()) {}
+
+Uploader::~Uploader() = default;
+
+StatusOr<std::unique_ptr<Uploader>> Uploader::Create(
+    UploadCallback upload_callback) {
+  auto uploader = base::WrapUnique(new Uploader(std::move(upload_callback)));
+  return uploader;
+}
+
+void Uploader::ProcessBlob(Priority priority,
+                           StatusOr<base::span<const uint8_t>> data,
+                           base::OnceCallback<void(bool)> processed_cb) {
+  if (completed_ || !data.ok()) {
+    std::move(processed_cb).Run(false);
+    return;
   }
 
-  void ProcessBlob(Priority priority,
-                   StatusOr<base::span<const uint8_t>> blob,
-                   base::OnceCallback<void(bool)> processed_cb) override {
-    std::move(processed_cb).Run(false);  // Do not proceed.
+  class ProcessBlobContext : public TaskRunnerContext<bool> {
+   public:
+    ProcessBlobContext(
+        base::span<const uint8_t> data,
+        std::vector<EncryptedRecord>* records,
+        base::OnceCallback<void(bool)> processed_callback,
+        scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner)
+        : TaskRunnerContext<bool>(std::move(processed_callback),
+                                  sequenced_task_runner),
+          records_(records),
+          data_(data.begin(), data.end()) {}
+
+   private:
+    ~ProcessBlobContext() override = default;
+
+    void OnStart() override {
+      if (data_.empty()) {
+        Complete(true);
+        return;
+      }
+      ProcessBlob();
+    }
+
+    void ProcessBlob() {
+      EncryptedRecord record;
+      if (!record.ParseFromArray(data_.data(), data_.size())) {
+        Complete(false);
+        return;
+      }
+      records_->push_back(record);
+      Complete(true);
+    }
+
+    void Complete(bool success) {
+      if (!success) {
+        LOG(ERROR) << "Unable to process blob";
+      }
+      Response(success);
+    }
+
+    std::vector<EncryptedRecord>* const records_;
+    const std::vector<const uint8_t> data_;
+  };
+
+  Start<ProcessBlobContext>(data.ValueOrDie(), encrypted_records_.get(),
+                            std::move(processed_cb), sequenced_task_runner_);
+}
+
+void Uploader::Completed(Priority priority, Status final_status) {
+  if (!final_status.ok()) {
+    // No work to do - something went wrong with storage and it no longer wants
+    // to upload the records. Let the records die with |this|.
+    return;
   }
 
-  void Completed(Priority priority, Status status) override {
-    LOG(ERROR) << "Not implemented yet, status=" << status;
+  if (completed_) {
+    // RunUpload has already been invoked. Return.
+    return;
   }
 
- private:
-  UploadClient() = default;
-};
+  sequenced_task_runner_->PostTask(
+      FROM_HERE, base::BindOnce(&Uploader::RunUpload, base::Unretained(this)));
+}
+
+void Uploader::RunUpload() {
+  if (completed_) {
+    // RunUpload has already been invoked. Return.
+    return;
+  }
+  completed_ = true;
+
+  Status upload_status =
+      std::move(upload_callback_).Run(std::move(encrypted_records_));
+  if (!upload_status.ok()) {
+    LOG(ERROR) << "Unable to upload records: " << upload_status;
+  }
+}
 
 ReportingClient::ReportingClient(scoped_refptr<StorageModule> storage)
     : storage_(std::move(storage)),
@@ -86,7 +241,7 @@ StatusOr<std::unique_ptr<ReportingClient>> ReportingClient::Create() {
   StatusOr<scoped_refptr<StorageModule>> storage_result;
   StorageModule::Create(
       Storage::Options().set_directory(reporting_path),
-      base::BindRepeating(&ReportingClient::UploadClient::Build),
+      base::BindRepeating(&ReportingClient::BuildUploader),
       base::BindOnce(
           [](StatusOr<scoped_refptr<StorageModule>>* result,
              base::WaitableEvent* done,
@@ -102,4 +257,20 @@ StatusOr<std::unique_ptr<ReportingClient>> ReportingClient::Create() {
   return client;
 }
 
+// static
+StatusOr<std::unique_ptr<Storage::UploaderInterface>>
+ReportingClient::BuildUploader(Priority priority) {
+  ASSIGN_OR_RETURN(ReportingClient * instance, GetInstance());
+  if (instance->upload_client_ == nullptr) {
+    ASSIGN_OR_RETURN(
+        instance->upload_client_,
+        UploadClient::Create(BuildCloudPolicyClient(),
+                             base::BindRepeating(&StorageModule::ReportSuccess,
+                                                 instance->storage_)));
+  }
+  return Uploader::Create(
+      base::BindOnce(&UploadClient::EnqueueUpload,
+                     base::Unretained(instance->upload_client_.get())));
+}
+
 }  // namespace reporting

chrome/browser/policy/messaging_layer/public/report_client.h

@@ -12,7 +12,9 @@
 #include "chrome/browser/policy/messaging_layer/public/report_queue.h"
 #include "chrome/browser/policy/messaging_layer/public/report_queue_configuration.h"
 #include "chrome/browser/policy/messaging_layer/storage/storage_module.h"
+#include "chrome/browser/policy/messaging_layer/upload/upload_client.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "chrome/browser/policy/messaging_layer/util/task_runner_context.h"
 
 namespace reporting {
 
@@ -31,6 +33,40 @@ namespace reporting {
 // }
 class ReportingClient {
  public:
+  // Uploader is passed to Storage in order to upload messages using the
+  // UploadClient.
+  class Uploader : public Storage::UploaderInterface {
+   public:
+    using UploadCallback = base::OnceCallback<Status(
+        std::unique_ptr<std::vector<EncryptedRecord>>)>;
+
+    static StatusOr<std::unique_ptr<Uploader>> Create(
+        UploadCallback upload_callback);
+
+    ~Uploader() override;
+    Uploader(const Uploader& other) = delete;
+    Uploader& operator=(const Uploader& other) = delete;
+
+    // TODO(chromium:1078512) Priority is unused, remove it.
+    void ProcessBlob(Priority priority,
+                     StatusOr<base::span<const uint8_t>> data,
+                     base::OnceCallback<void(bool)> processed_cb) override;
+
+    // TODO(chromium:1078512) Priority is unused, remove it.
+    void Completed(Priority priority, Status final_status) override;
+
+   private:
+    explicit Uploader(UploadCallback upload_callback_);
+
+    void RunUpload();
+
+    UploadCallback upload_callback_;
+
+    bool completed_;
+    std::unique_ptr<std::vector<EncryptedRecord>> encrypted_records_;
+    scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
+  };
+
   ~ReportingClient();
   ReportingClient(const ReportingClient& other) = delete;
   ReportingClient& operator=(const ReportingClient& other) = delete;
@@ -47,9 +83,6 @@ class ReportingClient {
       std::unique_ptr<ReportQueueConfiguration> config);
 
  private:
-  // Client for Upload operation.
-  class UploadClient;
-
   explicit ReportingClient(scoped_refptr<StorageModule> storage);
 
   static StatusOr<ReportingClient*> GetInstance();
@@ -57,8 +90,13 @@ class ReportingClient {
   // ReportingClient is not meant to be used directly.
   static StatusOr<std::unique_ptr<ReportingClient>> Create();
 
+  // TODO(chromium:1078512) Priority is unused, remove it.
+  static StatusOr<std::unique_ptr<Storage::UploaderInterface>> BuildUploader(
+      Priority priority);
+
   scoped_refptr<StorageModule> storage_;
   scoped_refptr<EncryptionModule> encryption_;
+  std::unique_ptr<UploadClient> upload_client_;
 };
 
 }  // namespace reporting

chrome/browser/policy/messaging_layer/storage/storage_module.cc

@@ -22,8 +22,8 @@ StorageModule::StorageModule() = default;
 
 StorageModule::~StorageModule() = default;
 
-void StorageModule::AddRecord(reporting::EncryptedRecord record,
-                              reporting::Priority priority,
+void StorageModule::AddRecord(EncryptedRecord record,
+                              Priority priority,
                               base::OnceCallback<void(Status)> callback) {
   size_t record_size = record.ByteSizeLong();
   auto data = std::make_unique<uint8_t[]>(record_size);
@@ -32,6 +32,11 @@ void StorageModule::AddRecord(reporting::EncryptedRecord record,
                   std::move(callback));
 }
 
+void StorageModule::ReportSuccess(
+    SequencingInformation sequencing_information) {
+  LOG(ERROR) << "ReportSuccess isn't implemented";
+}
+
 // static
 void StorageModule::Create(
     const Storage::Options& options,

chrome/browser/policy/messaging_layer/storage/storage_module.h

@@ -35,6 +35,10 @@ class StorageModule : public base::RefCountedThreadSafe<StorageModule> {
                          reporting::Priority priority,
                          base::OnceCallback<void(Status)> callback);
 
+  // Once a record has been successfully uploaded, the sequencing information
+  // can be passed back to the StorageModule here for record deletion.
+  virtual void ReportSuccess(SequencingInformation sequencing_information);
+
  protected:
   // Constructor can only be called by |Create| factory method.
   StorageModule();

chrome/browser/policy/messaging_layer/upload/app_install_report_handler.cc

@@ -136,7 +136,6 @@ AppInstallReportHandler::AppInstallReportHandler(
 AppInstallReportHandler::~AppInstallReportHandler() = default;
 
 Status AppInstallReportHandler::HandleRecord(Record record) {
-  RETURN_IF_ERROR(ValidateClientState());
   ASSIGN_OR_RETURN(base::Value report, ValidateRecord(record));
 
   ClientCallback client_cb = base::BindOnce([](bool finished_running) {

chrome/browser/policy/messaging_layer/upload/dm_server_upload_service.cc

@@ -16,7 +16,6 @@
 #include "chrome/browser/policy/messaging_layer/util/status_macros.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
 #include "chrome/browser/policy/messaging_layer/util/task_runner_context.h"
-#include "chrome/browser/profiles/profile.h"
 #include "components/policy/core/common/cloud/cloud_policy_client.h"
 #include "components/policy/core/common/cloud/user_cloud_policy_manager.h"
 #include "components/policy/proto/record.pb.h"
@@ -100,6 +99,8 @@ void DmServerUploader::HandleRecords() {
   // |ArcAppInstallEventLogUploader|).
   // TODO(chromium:1078512) Consider creating a whitelist/blacklist for retry
   // and continue.
+  // TODO(chromium:1078512) Cannot verify client state on this thread. Find a
+  // way to do that and restructure this loop to handle it.
   for (auto record_info_it = record_infos_.begin();
        record_info_it != record_infos_.end();) {
     for (auto& handler : *handlers_) {
@@ -197,13 +198,13 @@ void DmServerUploader::ResetDelay() {
 }
 
 StatusOr<std::unique_ptr<DmServerUploadService>> DmServerUploadService::Create(
-    Profile* profile,
+    std::unique_ptr<policy::CloudPolicyClient> client,
     ReportSuccessfulUploadCallback upload_cb) {
-  if (profile == nullptr) {
-    return Status(error::INVALID_ARGUMENT, "Profile may not be nullptr.");
+  if (client == nullptr) {
+    return Status(error::INVALID_ARGUMENT, "client may not be nullptr.");
   }
   auto uploader =
-      base::WrapUnique(new DmServerUploadService(profile, upload_cb));
+      base::WrapUnique(new DmServerUploadService(std::move(client), upload_cb));
 
   RETURN_IF_ERROR(uploader->InitRecordHandlers());
 
@@ -211,9 +212,9 @@ StatusOr<std::unique_ptr<DmServerUploadService>> DmServerUploadService::Create(
 }
 
 DmServerUploadService::DmServerUploadService(
-    Profile* profile,
+    std::unique_ptr<policy::CloudPolicyClient> client,
     ReportSuccessfulUploadCallback upload_cb)
-    : profile_(profile),
+    : client_(std::move(client)),
       upload_cb_(upload_cb),
       sequenced_task_runner_(base::ThreadPool::CreateSequencedTaskRunner({})) {}
 
@@ -257,25 +258,7 @@ void DmServerUploadService::UploadCompletion(
 }
 
 CloudPolicyClient* DmServerUploadService::GetClient() {
-#if defined(OS_CHROMEOS)
-  auto* policy_manager = profile_->GetUserCloudPolicyManagerChromeOS();
-#else
-  auto* policy_manager = profile_->GetUserCloudPolicyManager();
-#endif
-  if (policy_manager == nullptr) {
-    LOG(ERROR) << "Policy manager was null";
-    return nullptr;
-  }
-  auto* core = policy_manager->core();
-  if (core == nullptr) {
-    LOG(ERROR) << "Core was null";
-    return nullptr;
-  }
-  auto* client = core->client();
-  if (client == nullptr) {
-    LOG(ERROR) << "Client was null";
-  }
-  return client;
+  return client_.get();
 }
 
 }  // namespace reporting

chrome/browser/policy/messaging_layer/upload/dm_server_upload_service.h

@@ -14,7 +14,6 @@
 #include "chrome/browser/policy/messaging_layer/util/status_macros.h"
 #include "chrome/browser/policy/messaging_layer/util/statusor.h"
 #include "chrome/browser/policy/messaging_layer/util/task_runner_context.h"
-#include "chrome/browser/profiles/profile.h"
 #include "components/policy/core/common/cloud/cloud_policy_client.h"
 #include "components/policy/proto/record.pb.h"
 #include "components/policy/proto/record_constants.pb.h"
@@ -22,7 +21,7 @@
 
 namespace reporting {
 
-// DMServerUploadService uploads events to the DMServer. It does not manage
+// DmServerUploadService uploads events to the DMServer. It does not manage
 // sequence information, instead reporting the highest sequence number for each
 // generation id and priority.
 //
@@ -115,32 +114,31 @@ class DmServerUploadService {
 
   // Will create a DMServerUploadService with handlers.
   // On successful completion returns a DMServerUploadService.
-  // If |profile| is null, will return error::INVALID_ARGUMENT.
+  // If |client| is null, will return error::INVALID_ARGUMENT.
   // If any handlers fail to create, or the policy::CloudPolicyClient is null,
   // will return error::UNAVAILABLE.
   //
-  // |profile| must not be null.
+  // |client| must not be null.
   // |completion_cb| should report back to the holder of the created object
   // whenever a record set is successfully uploaded.
   static StatusOr<std::unique_ptr<DmServerUploadService>> Create(
-      Profile* profile,
+      std::unique_ptr<policy::CloudPolicyClient> client,
       ReportSuccessfulUploadCallback completion_cb);
   ~DmServerUploadService();
 
   Status EnqueueUpload(std::unique_ptr<std::vector<EncryptedRecord>> record);
 
  private:
-  DmServerUploadService(Profile* profile,
+  DmServerUploadService(std::unique_ptr<policy::CloudPolicyClient> client,
                         ReportSuccessfulUploadCallback completion_cb);
 
   Status InitRecordHandlers();
 
   void UploadCompletion(StatusOr<std::vector<SequencingInformation>>) const;
 
-  // Must check for nullptr.
   policy::CloudPolicyClient* GetClient();
 
-  Profile* profile_;
+  std::unique_ptr<policy::CloudPolicyClient> client_;
   ReportSuccessfulUploadCallback upload_cb_;
   scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
 

chrome/browser/policy/messaging_layer/upload/upload_client.cc

+// Copyright 2020 The Chromium 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 "chrome/browser/policy/messaging_layer/upload/upload_client.h"
+
+#include "base/memory/ptr_util.h"
+#include "chrome/browser/policy/messaging_layer/upload/dm_server_upload_service.h"
+#include "chrome/browser/policy/messaging_layer/util/status.h"
+#include "chrome/browser/policy/messaging_layer/util/status_macros.h"
+#include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "components/policy/core/common/cloud/cloud_policy_client.h"
+#include "components/user_manager/user_manager.h"
+#include "content/public/browser/browser_context.h"
+#include "content/public/browser/storage_partition.h"
+
+namespace reporting {
+
+// static
+StatusOr<std::unique_ptr<UploadClient>> UploadClient::Create(
+    std::unique_ptr<policy::CloudPolicyClient> cloud_policy_client,
+    ReportSuccessfulUploadCallback report_success_cb) {
+  auto upload_client = base::WrapUnique(new UploadClient());
+  ASSIGN_OR_RETURN(upload_client->dm_server_upload_service_,
+                   DmServerUploadService::Create(std::move(cloud_policy_client),
+                                                 report_success_cb));
+
+  return upload_client;
+}
+
+Status UploadClient::EnqueueUpload(
+    std::unique_ptr<std::vector<EncryptedRecord>> records) {
+  if (records->empty()) {
+    return Status::StatusOK();
+  }
+
+  return dm_server_upload_service_->EnqueueUpload(std::move(records));
+}
+
+UploadClient::UploadClient() = default;
+
+UploadClient::~UploadClient() = default;
+
+}  // namespace reporting

chrome/browser/policy/messaging_layer/upload/upload_client.h

+// Copyright 2020 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROME_BROWSER_POLICY_MESSAGING_LAYER_UPLOAD_UPLOAD_CLIENT_H_
+#define CHROME_BROWSER_POLICY_MESSAGING_LAYER_UPLOAD_UPLOAD_CLIENT_H_
+
+#include <memory>
+#include <vector>
+
+#include "base/task/post_task.h"
+#include "base/task_runner.h"
+#include "chrome/browser/policy/messaging_layer/upload/dm_server_upload_service.h"
+#include "chrome/browser/policy/messaging_layer/util/status.h"
+#include "chrome/browser/policy/messaging_layer/util/statusor.h"
+#include "components/policy/core/common/cloud/cloud_policy_client.h"
+#include "components/policy/proto/record.pb.h"
+
+namespace reporting {
+
+// UploadClient handles sending records to the correct upload service.
+class UploadClient {
+ public:
+  // ReportSuccessfulUploadCallback is used to pass server responses back to
+  // the owner of |this|.
+  using ReportSuccessfulUploadCallback =
+      base::RepeatingCallback<void(SequencingInformation)>;
+
+  static StatusOr<std::unique_ptr<UploadClient>> Create(
+      std::unique_ptr<policy::CloudPolicyClient> cloud_policy_client,
+      ReportSuccessfulUploadCallback report_success_cb);
+
+  ~UploadClient();
+  UploadClient(const UploadClient& other) = delete;
+  UploadClient& operator=(const UploadClient& other) = delete;
+
+  Status EnqueueUpload(std::unique_ptr<std::vector<EncryptedRecord>> record);
+
+ private:
+  UploadClient();
+
+  std::unique_ptr<DmServerUploadService> dm_server_upload_service_;
+};
+
+}  // namespace reporting
+
+#endif  // CHROME_BROWSER_POLICY_MESSAGING_LAYER_UPLOAD_UPLOAD_CLIENT_H_

chrome/browser/policy/messaging_layer/upload/upload_client_unittest.cc

+// Copyright 2020 The Chromium 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 "chrome/browser/policy/messaging_layer/upload/upload_client.h"
+#include "base/json/json_writer.h"
+#include "base/test/task_environment.h"
+#include "base/values.h"
+#include "chrome/browser/policy/messaging_layer/upload/app_install_report_handler.h"
+#include "components/policy/core/common/cloud/dm_token.h"
+#include "components/policy/core/common/cloud/mock_cloud_policy_client.h"
+#include "components/policy/proto/record.pb.h"
+#include "components/policy/proto/record_constants.pb.h"
+
+#include "base/bind.h"
+#include "base/files/file_path.h"
+#include "base/test/test_mock_time_task_runner.h"
+#include "components/account_id/account_id.h"
+#include "services/network/test/test_network_connection_tracker.h"
+
+namespace reporting {
+namespace {
+
+using policy::MockCloudPolicyClient;
+using testing::_;
+using testing::Invoke;
+using testing::InvokeArgument;
+using testing::WithArgs;
+
+class TestCallbackWaiter {
+ public:
+  TestCallbackWaiter()
+      : completed_(base::WaitableEvent::ResetPolicy::MANUAL,
+                   base::WaitableEvent::InitialState::NOT_SIGNALED) {}
+
+  virtual void Signal() {
+    DCHECK(!completed_.IsSignaled());
+    completed_.Signal();
+  }
+
+  void Wait() { completed_.Wait(); }
+
+ protected:
+  base::WaitableEvent completed_;
+};
+
+class TestCallbackWaiterWithCounter : public TestCallbackWaiter {
+ public:
+  explicit TestCallbackWaiterWithCounter(int counter_limit)
+      : counter_limit_(counter_limit) {}
+
+  void Signal() override {
+    DCHECK(!completed_.IsSignaled());
+    DCHECK_GT(counter_limit_, 0);
+    if (--counter_limit_ == 0) {
+      completed_.Signal();
+    }
+  }
+
+ private:
+  std::atomic<int> counter_limit_;
+};
+
+TEST(UploadClientTest, CreateUploadClient) {
+  base::test::TaskEnvironment task_envrionment{
+      base::test::TaskEnvironment::TimeSource::MOCK_TIME};
+
+  const int kExpectedCallTimes = 10;
+  const uint64_t kGenerationId = 1234;
+
+  TestCallbackWaiterWithCounter waiter(kExpectedCallTimes);
+
+  auto client = std::make_unique<MockCloudPolicyClient>();
+  client->SetDMToken(
+      policy::DMToken::CreateValidTokenForTesting("FAKE_DM_TOKEN").value());
+
+  EXPECT_CALL(*client, UploadAppInstallReport_(_, _))
+      .WillRepeatedly(WithArgs<1>(
+          Invoke([&waiter](AppInstallReportHandler::ClientCallback& callback) {
+            std::move(callback).Run(true);
+            waiter.Signal();
+          })));
+
+  auto upload_client_result =
+      UploadClient::Create(std::move(client), base::DoNothing());
+
+  ASSERT_TRUE(upload_client_result.ok());
+
+  base::Value data{base::Value::Type::DICTIONARY};
+  data.SetKey("TEST_KEY", base::Value("TEST_VALUE"));
+
+  std::string json_data;
+  ASSERT_TRUE(base::JSONWriter::Write(data, &json_data));
+
+  WrappedRecord wrapped_record;
+  Record* record = wrapped_record.mutable_record();
+  record->set_data(json_data);
+  record->set_destination(Destination::APP_INSTALL_EVENT);
+
+  std::string serialized_record;
+  wrapped_record.SerializeToString(&serialized_record);
+
+  std::unique_ptr<std::vector<EncryptedRecord>> records =
+      std::make_unique<std::vector<EncryptedRecord>>();
+  for (int i = 0; i < kExpectedCallTimes; i++) {
+    EncryptedRecord encrypted_record;
+    encrypted_record.set_encrypted_wrapped_record(serialized_record);
+
+    SequencingInformation* sequencing_information =
+        encrypted_record.mutable_sequencing_information();
+    sequencing_information->set_sequencing_id(i);
+    sequencing_information->set_generation_id(kGenerationId);
+    sequencing_information->set_priority(Priority::IMMEDIATE);
+    records->push_back(encrypted_record);
+  }
+
+  auto upload_client = std::move(upload_client_result.ValueOrDie());
+  auto enqueue_result = upload_client->EnqueueUpload(std::move(records));
+  EXPECT_TRUE(enqueue_result.ok());
+
+  waiter.Wait();
+}
+
+}  // namespace
+}  // namespace reporting

chrome/test/BUILD.gn

@@ -3410,6 +3410,7 @@ test("unit_tests") {
     "../browser/policy/messaging_layer/storage/test_storage_module.h",
     "../browser/policy/messaging_layer/upload/app_install_report_handler_unittest.cc",
     "../browser/policy/messaging_layer/upload/dm_server_upload_service_unittest.cc",
+    "../browser/policy/messaging_layer/upload/upload_client_unittest.cc",
     "../browser/policy/messaging_layer/util/shared_queue_unittest.cc",
     "../browser/policy/messaging_layer/util/status_macros_unittest.cc",
     "../browser/policy/messaging_layer/util/status_unittest.cc",