Add replication_util, a CLI for replication_lib.
- The CLI just reads a ReplicationConfig and passes it
to replication_lib. In the future, may support more
functionality, such as dry running.
TEST=run_tests scripts/replication_util_unittest
BUG=chromium:1021252
Change-Id: I2561ee506447424153f6ea280e219552966c6a0d
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/chromite/+/1898378
Tested-by: Andrew Lamb <andrewlamb@chromium.org>
Commit-Queue: Andrew Lamb <andrewlamb@chromium.org>
Reviewed-by: David Burger <dburger@chromium.org>
diff --git a/scripts/replication_util_unittest.py b/scripts/replication_util_unittest.py
new file mode 100644
index 0000000..62fe330
--- /dev/null
+++ b/scripts/replication_util_unittest.py
@@ -0,0 +1,92 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 The Chromium OS Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Unit tests for replication_util."""
+
+from __future__ import print_function
+
+import json
+import os
+
+from google.protobuf import json_format
+
+from chromite.api.gen.config.replication_config_pb2 import (
+ ReplicationConfig, FileReplicationRule, FILE_TYPE_OTHER,
+ REPLICATION_TYPE_COPY)
+from chromite.lib import cros_test_lib
+from chromite.lib import osutils
+from chromite.scripts import replication_util
+
+D = cros_test_lib.Directory
+
+
+class RunTest(cros_test_lib.TempDirTestCase):
+ """Tests of the run command.
+
+ Note that detailed tests of replication behavior should be done in
+ replication_lib_unittest.
+ """
+
+ def setUp(self):
+ file_layout = (D('src', ['audio_file']),)
+ cros_test_lib.CreateOnDiskHierarchy(self.tempdir, file_layout)
+
+ self.audio_path = os.path.join(self.tempdir, 'src', 'audio_file')
+ osutils.WriteFile(self.audio_path, '[Speaker A Settings]')
+
+ def testRun(self):
+ """Basic test of the 'run' command.
+
+ """
+ audio_dst_path = os.path.join(self.tempdir, 'dst', 'audio_file')
+
+ replication_config = ReplicationConfig(file_replication_rules=[
+ FileReplicationRule(
+ source_path=self.audio_path,
+ destination_path=audio_dst_path,
+ file_type=FILE_TYPE_OTHER,
+ replication_type=REPLICATION_TYPE_COPY,
+ ),
+ ])
+
+ replication_config_path = os.path.join(self.tempdir,
+ 'replication_config.jsonpb')
+ osutils.WriteFile(replication_config_path,
+ json_format.MessageToJson(replication_config))
+
+ replication_util.main(['run', replication_config_path])
+
+ expected_file_layout = (
+ D('src', ['audio_file']),
+ D('dst', ['audio_file']),
+ 'replication_config.jsonpb',
+ )
+
+ cros_test_lib.VerifyOnDiskHierarchy(self.tempdir, expected_file_layout)
+
+ self.assertFileContents(audio_dst_path, '[Speaker A Settings]')
+
+ def testUnknownFieldInConfig(self):
+ """Test that unknown fields in the ReplicationConfig cause an error."""
+ audio_dst_path = os.path.join(self.tempdir, 'dst', 'audio_file')
+
+ replication_config = ReplicationConfig(file_replication_rules=[
+ FileReplicationRule(
+ source_path=self.audio_path,
+ destination_path=audio_dst_path,
+ file_type=FILE_TYPE_OTHER,
+ replication_type=REPLICATION_TYPE_COPY,
+ ),
+ ])
+
+ replication_config_path = os.path.join(self.tempdir,
+ 'replication_config.jsonpb')
+ replication_config_dict = json_format.MessageToDict(replication_config)
+ replication_config_dict['new_field'] = 1
+ osutils.WriteFile(replication_config_path,
+ json.dumps(replication_config_dict))
+
+ with self.assertRaises(json_format.ParseError):
+ replication_util.main(['run', replication_config_path])