Andrew Lamb | 16f02bf | 2019-11-04 14:44:34 -0700 | [diff] [blame^] | 1 | # -*- coding: utf-8 -*- |
| 2 | # Copyright 2019 The Chromium OS Authors. All rights reserved. |
| 3 | # Use of this source code is governed by a BSD-style license that can be |
| 4 | # found in the LICENSE file. |
| 5 | |
| 6 | """Unit tests for replication_util.""" |
| 7 | |
| 8 | from __future__ import print_function |
| 9 | |
| 10 | import json |
| 11 | import os |
| 12 | |
| 13 | from google.protobuf import json_format |
| 14 | |
| 15 | from chromite.api.gen.config.replication_config_pb2 import ( |
| 16 | ReplicationConfig, FileReplicationRule, FILE_TYPE_OTHER, |
| 17 | REPLICATION_TYPE_COPY) |
| 18 | from chromite.lib import cros_test_lib |
| 19 | from chromite.lib import osutils |
| 20 | from chromite.scripts import replication_util |
| 21 | |
| 22 | D = cros_test_lib.Directory |
| 23 | |
| 24 | |
| 25 | class RunTest(cros_test_lib.TempDirTestCase): |
| 26 | """Tests of the run command. |
| 27 | |
| 28 | Note that detailed tests of replication behavior should be done in |
| 29 | replication_lib_unittest. |
| 30 | """ |
| 31 | |
| 32 | def setUp(self): |
| 33 | file_layout = (D('src', ['audio_file']),) |
| 34 | cros_test_lib.CreateOnDiskHierarchy(self.tempdir, file_layout) |
| 35 | |
| 36 | self.audio_path = os.path.join(self.tempdir, 'src', 'audio_file') |
| 37 | osutils.WriteFile(self.audio_path, '[Speaker A Settings]') |
| 38 | |
| 39 | def testRun(self): |
| 40 | """Basic test of the 'run' command. |
| 41 | |
| 42 | """ |
| 43 | audio_dst_path = os.path.join(self.tempdir, 'dst', 'audio_file') |
| 44 | |
| 45 | replication_config = ReplicationConfig(file_replication_rules=[ |
| 46 | FileReplicationRule( |
| 47 | source_path=self.audio_path, |
| 48 | destination_path=audio_dst_path, |
| 49 | file_type=FILE_TYPE_OTHER, |
| 50 | replication_type=REPLICATION_TYPE_COPY, |
| 51 | ), |
| 52 | ]) |
| 53 | |
| 54 | replication_config_path = os.path.join(self.tempdir, |
| 55 | 'replication_config.jsonpb') |
| 56 | osutils.WriteFile(replication_config_path, |
| 57 | json_format.MessageToJson(replication_config)) |
| 58 | |
| 59 | replication_util.main(['run', replication_config_path]) |
| 60 | |
| 61 | expected_file_layout = ( |
| 62 | D('src', ['audio_file']), |
| 63 | D('dst', ['audio_file']), |
| 64 | 'replication_config.jsonpb', |
| 65 | ) |
| 66 | |
| 67 | cros_test_lib.VerifyOnDiskHierarchy(self.tempdir, expected_file_layout) |
| 68 | |
| 69 | self.assertFileContents(audio_dst_path, '[Speaker A Settings]') |
| 70 | |
| 71 | def testUnknownFieldInConfig(self): |
| 72 | """Test that unknown fields in the ReplicationConfig cause an error.""" |
| 73 | audio_dst_path = os.path.join(self.tempdir, 'dst', 'audio_file') |
| 74 | |
| 75 | replication_config = ReplicationConfig(file_replication_rules=[ |
| 76 | FileReplicationRule( |
| 77 | source_path=self.audio_path, |
| 78 | destination_path=audio_dst_path, |
| 79 | file_type=FILE_TYPE_OTHER, |
| 80 | replication_type=REPLICATION_TYPE_COPY, |
| 81 | ), |
| 82 | ]) |
| 83 | |
| 84 | replication_config_path = os.path.join(self.tempdir, |
| 85 | 'replication_config.jsonpb') |
| 86 | replication_config_dict = json_format.MessageToDict(replication_config) |
| 87 | replication_config_dict['new_field'] = 1 |
| 88 | osutils.WriteFile(replication_config_path, |
| 89 | json.dumps(replication_config_dict)) |
| 90 | |
| 91 | with self.assertRaises(json_format.ParseError): |
| 92 | replication_util.main(['run', replication_config_path]) |