1# Copyright 2013 The Chromium Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import os
6import unittest
7
8PERF_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
9from telemetry.testing import system_stub
10from telemetry.internal.testing import system_stub_test_module
11
12class CloudStorageTest(unittest.TestCase):
13  SUCCESS_FILE_HASH = 'success'.zfill(40)
14  PUBLIC_FILE_HASH = 'public'.zfill(40)
15  PARTNER_FILE_HASH = 'partner'.zfill(40)
16  INTERNAL_FILE_HASH = 'internal'.zfill(40)
17  UPDATED_HASH = 'updated'.zfill(40)
18
19  def setUp(self):
20    self.cloud_storage = system_stub.CloudStorageModuleStub()
21
22    # Files in Cloud Storage.
23    self.remote_files = ['preset_public_file.wpr',
24                         'preset_partner_file.wpr',
25                         'preset_internal_file.wpr']
26    self.remote_paths = {
27      self.cloud_storage.PUBLIC_BUCKET:
28        {'preset_public_file.wpr':CloudStorageTest.PUBLIC_FILE_HASH},
29      self.cloud_storage.PARTNER_BUCKET:
30        {'preset_partner_file.wpr':CloudStorageTest.PARTNER_FILE_HASH},
31      self.cloud_storage.INTERNAL_BUCKET:
32        {'preset_internal_file.wpr':CloudStorageTest.INTERNAL_FILE_HASH}}
33
34    # Local data files and hashes.
35    self.data_files = [
36        os.path.join(os.path.sep, 'path', 'to', 'success.wpr'),
37        os.path.join(os.path.sep, 'path', 'to', 'wrong_hash.wpr'),
38        os.path.join(os.path.sep, 'path', 'to', 'preset_public_file.wpr'),
39        os.path.join(os.path.sep, 'path', 'to', 'preset_partner_file.wpr'),
40        os.path.join(os.path.sep, 'path', 'to', 'preset_internal_file.wpr')]
41    self.local_file_hashes = {
42        os.path.join(os.path.sep, 'path', 'to', 'success.wpr'):
43            CloudStorageTest.SUCCESS_FILE_HASH,
44        os.path.join(os.path.sep, 'path', 'to', 'wrong_hash.wpr'):
45            CloudStorageTest.SUCCESS_FILE_HASH,
46        os.path.join(os.path.sep, 'path', 'to', 'preset_public_file.wpr'):
47            CloudStorageTest.PUBLIC_FILE_HASH,
48        os.path.join(os.path.sep, 'path', 'to', 'preset_partner_file.wpr'):
49            CloudStorageTest.PARTNER_FILE_HASH,
50        os.path.join(os.path.sep, 'path', 'to', 'preset_internal_file.wpr'):
51            CloudStorageTest.INTERNAL_FILE_HASH,
52    }
53    self.cloud_storage.SetCalculatedHashesForTesting(self.local_file_hashes)
54    # Local hash files and their contents.
55    local_hash_files = {
56        os.path.join(os.path.sep, 'path', 'to', 'success.wpr.sha1'):
57            CloudStorageTest.SUCCESS_FILE_HASH,
58        os.path.join(os.path.sep, 'path', 'to', 'wrong_hash.wpr.sha1'):
59            'wronghash'.zfill(40),
60        os.path.join(os.path.sep, 'path', 'to', 'preset_public_file.wpr.sha1'):
61            CloudStorageTest.PUBLIC_FILE_HASH,
62        os.path.join(os.path.sep, 'path', 'to', 'preset_partner_file.wpr.sha1'):
63            CloudStorageTest.PARTNER_FILE_HASH,
64        os.path.join(os.path.sep, 'path', 'to',
65                     'preset_internal_file.wpr.sha1'):
66            CloudStorageTest.INTERNAL_FILE_HASH,
67    }
68    self.cloud_storage.SetHashFileContentsForTesting(local_hash_files)
69
70  def testSetup(self):
71    self.assertEqual(self.local_file_hashes,
72                     self.cloud_storage.local_file_hashes)
73    self.assertEqual(set(self.data_files),
74                     set(self.cloud_storage.GetLocalDataFiles()))
75    self.assertEqual(self.cloud_storage.default_remote_paths,
76                     self.cloud_storage.GetRemotePathsForTesting())
77    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
78    self.assertEqual(self.remote_paths,
79                     self.cloud_storage.GetRemotePathsForTesting())
80
81  def testExistsEmptyCloudStorage(self):
82    # Test empty remote files dictionary.
83    self.assertFalse(self.cloud_storage.Exists(self.cloud_storage.PUBLIC_BUCKET,
84                                               'preset_public_file.wpr'))
85    self.assertFalse(self.cloud_storage.Exists(
86        self.cloud_storage.PARTNER_BUCKET, 'preset_partner_file.wpr'))
87    self.assertFalse(self.cloud_storage.Exists(
88        self.cloud_storage.INTERNAL_BUCKET, 'preset_internal_file.wpr'))
89
90  def testExistsNonEmptyCloudStorage(self):
91    # Test non-empty remote files dictionary.
92    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
93    self.assertTrue(self.cloud_storage.Exists(
94        self.cloud_storage.PUBLIC_BUCKET, 'preset_public_file.wpr'))
95    self.assertTrue(self.cloud_storage.Exists(
96        self.cloud_storage.PARTNER_BUCKET, 'preset_partner_file.wpr'))
97    self.assertTrue(self.cloud_storage.Exists(
98        self.cloud_storage.INTERNAL_BUCKET, 'preset_internal_file.wpr'))
99    self.assertFalse(self.cloud_storage.Exists(
100        self.cloud_storage.PUBLIC_BUCKET, 'fake_file'))
101    self.assertFalse(self.cloud_storage.Exists(
102        self.cloud_storage.PARTNER_BUCKET, 'fake_file'))
103    self.assertFalse(self.cloud_storage.Exists(
104        self.cloud_storage.INTERNAL_BUCKET, 'fake_file'))
105    # Reset state.
106    self.cloud_storage.SetRemotePathsForTesting()
107
108  def testNonEmptyInsertAndExistsPublic(self):
109    # Test non-empty remote files dictionary.
110    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
111    self.assertFalse(self.cloud_storage.Exists(self.cloud_storage.PUBLIC_BUCKET,
112                                               'success.wpr'))
113    self.cloud_storage.Insert(
114        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
115        os.path.join(os.path.sep, 'path', 'to', 'success.wpr'))
116    self.assertTrue(self.cloud_storage.Exists(
117        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr'))
118    # Reset state.
119    self.cloud_storage.SetRemotePathsForTesting()
120
121  def testEmptyInsertAndExistsPublic(self):
122    # Test empty remote files dictionary.
123    self.assertFalse(self.cloud_storage.Exists(
124        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr'))
125    self.cloud_storage.Insert(
126        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
127        os.path.join(os.path.sep, 'path', 'to', 'success.wpr'))
128    self.assertTrue(self.cloud_storage.Exists(
129        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr'))
130
131  def testEmptyInsertAndGet(self):
132    self.assertRaises(self.cloud_storage.NotFoundError, self.cloud_storage.Get,
133                      self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
134                      os.path.join(os.path.sep, 'path', 'to', 'success.wpr'))
135    self.cloud_storage.Insert(self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
136                              os.path.join(os.path.sep, 'path', 'to',
137                                           'success.wpr'))
138    self.assertTrue(self.cloud_storage.Exists(
139        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr'))
140    self.assertEqual(CloudStorageTest.SUCCESS_FILE_HASH, self.cloud_storage.Get(
141        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
142        os.path.join(os.path.sep, 'path', 'to', 'success.wpr')))
143
144  def testNonEmptyInsertAndGet(self):
145    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
146    self.assertRaises(self.cloud_storage.NotFoundError, self.cloud_storage.Get,
147                      self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
148                      os.path.join(os.path.sep, 'path', 'to', 'success.wpr'))
149    self.cloud_storage.Insert(self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
150                              os.path.join(os.path.sep, 'path', 'to',
151                                           'success.wpr'))
152    self.assertTrue(self.cloud_storage.Exists(self.cloud_storage.PUBLIC_BUCKET,
153                                              'success.wpr'))
154    self.assertEqual(
155        CloudStorageTest.SUCCESS_FILE_HASH, self.cloud_storage.Get(
156            self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
157            os.path.join(os.path.sep, 'path', 'to', 'success.wpr')))
158    # Reset state.
159    self.cloud_storage.SetRemotePathsForTesting()
160
161  def testGetIfChanged(self):
162    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
163    self.assertRaises(
164        self.cloud_storage.NotFoundError, self.cloud_storage.Get,
165        self.cloud_storage.PUBLIC_BUCKET, 'success.wpr',
166        os.path.join(os.path.sep, 'path', 'to', 'success.wpr'))
167    self.assertFalse(self.cloud_storage.GetIfChanged(
168        os.path.join(os.path.sep, 'path', 'to', 'preset_public_file.wpr'),
169        self.cloud_storage.PUBLIC_BUCKET))
170    self.cloud_storage.ChangeRemoteHashForTesting(
171        self.cloud_storage.PUBLIC_BUCKET, 'preset_public_file.wpr',
172        CloudStorageTest.UPDATED_HASH)
173    self.assertTrue(self.cloud_storage.GetIfChanged(
174        os.path.join(os.path.sep, 'path', 'to', 'preset_public_file.wpr'),
175        self.cloud_storage.PUBLIC_BUCKET))
176    self.assertFalse(self.cloud_storage.GetIfChanged(
177        os.path.join(os.path.sep, 'path', 'to', 'preset_public_file.wpr'),
178        self.cloud_storage.PUBLIC_BUCKET))
179    # Reset state.
180    self.cloud_storage.SetRemotePathsForTesting()
181
182  def testList(self):
183    self.assertEqual([],
184                     self.cloud_storage.List(self.cloud_storage.PUBLIC_BUCKET))
185    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
186    self.assertEqual(['preset_public_file.wpr'],
187                     self.cloud_storage.List(self.cloud_storage.PUBLIC_BUCKET))
188    # Reset state.
189    self.cloud_storage.SetRemotePathsForTesting()
190
191  def testPermissionError(self):
192    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
193    self.cloud_storage.SetPermissionLevelForTesting(
194        self.cloud_storage.PUBLIC_PERMISSION)
195    self.assertRaises(
196        self.cloud_storage.PermissionError, self.cloud_storage.Get,
197        self.cloud_storage.INTERNAL_BUCKET, 'preset_internal_file.wpr',
198        os.path.join(os.path.sep, 'path', 'to', 'preset_internal_file.wpr'))
199    self.assertRaises(
200        self.cloud_storage.PermissionError, self.cloud_storage.GetIfChanged,
201        os.path.join(os.path.sep, 'path', 'to', 'preset_internal_file.wpr'),
202        self.cloud_storage.INTERNAL_BUCKET)
203    self.assertRaises(
204        self.cloud_storage.PermissionError, self.cloud_storage.List,
205        self.cloud_storage.INTERNAL_BUCKET)
206    self.assertRaises(
207        self.cloud_storage.PermissionError, self.cloud_storage.Exists,
208        self.cloud_storage.INTERNAL_BUCKET, 'preset_internal_file.wpr')
209    self.assertRaises(
210        self.cloud_storage.PermissionError, self.cloud_storage.Insert,
211        self.cloud_storage.INTERNAL_BUCKET, 'success.wpr',
212        os.path.join(os.path.sep, 'path', 'to', 'success.wpr'))
213    # Reset state.
214    self.cloud_storage.SetRemotePathsForTesting()
215
216  def testCredentialsError(self):
217    self.cloud_storage.SetRemotePathsForTesting(self.remote_paths)
218    self.cloud_storage.SetPermissionLevelForTesting(
219        self.cloud_storage.CREDENTIALS_ERROR_PERMISSION)
220    self.assertRaises(
221        self.cloud_storage.CredentialsError, self.cloud_storage.Get,
222        self.cloud_storage.INTERNAL_BUCKET, 'preset_internal_file.wpr',
223        os.path.join(os.path.sep, 'path', 'to', 'preset_internal_file.wpr'))
224    self.assertRaises(
225        self.cloud_storage.CredentialsError, self.cloud_storage.GetIfChanged,
226        self.cloud_storage.INTERNAL_BUCKET,
227        os.path.join(os.path.sep, 'path', 'to', 'preset_internal_file.wpr'))
228    self.assertRaises(
229        self.cloud_storage.CredentialsError, self.cloud_storage.List,
230        self.cloud_storage.INTERNAL_BUCKET)
231    self.assertRaises(
232        self.cloud_storage.CredentialsError, self.cloud_storage.Exists,
233        self.cloud_storage.INTERNAL_BUCKET, 'preset_internal_file.wpr')
234    self.assertRaises(
235        self.cloud_storage.CredentialsError, self.cloud_storage.Insert,
236        self.cloud_storage.INTERNAL_BUCKET, 'success.wpr',
237        os.path.join(os.path.sep, 'path', 'to', 'success.wpr'))
238    # Reset state.
239    self.cloud_storage.SetRemotePathsForTesting()
240
241  def testOpenRestoresCorrectly(self):
242    file_path = os.path.realpath(__file__)
243    stubs = system_stub.Override(system_stub_test_module, ['open'])
244    stubs.open.files = {file_path:'contents'}
245    f = system_stub_test_module.SystemStubTest.TestOpen(file_path)
246    self.assertEqual(type(f), system_stub.OpenFunctionStub.FileStub)
247    stubs.open.files = {}
248    stubs.Restore()
249    # This will throw an error if the open stub wasn't restored correctly.
250    f = system_stub_test_module.SystemStubTest.TestOpen(file_path)
251    self.assertEqual(type(f), file)
252