[NaCl SDK] update_nacl_manifest: Allow manual updating of a version.

native_client_sdk/src/build_tools/tests/test_update_nacl_manifest.py

 #!/usr/bin/env python
 # Copyright (c) 2012 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.
 
 import copy
 import datetime
 import hashlib
 import os
 import posixpath
@@ skipping 118 matching lines @@
 
   def Add(self, host_oses, channel, version):
     for host_os in host_oses:
       timestamp = self.datetime.strftime('%Y-%m-%d %H:%M:%S.%f')
       self.history.append((host_os, channel, version, timestamp))
       self.datetime += datetime.timedelta(0, -3600) # one hour earlier
     self.datetime += datetime.timedelta(-1) # one day earlier
 
 
 class MakeFiles(dict):
+  def AddOnlineManifest(self, manifest_string):
+    self['naclsdk_manifest2.json'] = manifest_string
+
   def Add(self, bundle, add_archive_for_os=OS_MLW, add_json_for_os=OS_MLW):
     for archive in bundle.GetArchives():
       if not archive.host_os in add_archive_for_os:
         continue
 
       # add a dummy file for each archive
       path = GetPathFromHttpsUrl(archive.url)
       self[path] = 'My Dummy Archive'
 
       if archive.host_os in add_json_for_os:
         # add .json manifest snippet, it should look like a normal Bundle, but
         # only has one archive.
         new_bundle = manifest_util.Bundle('')
         new_bundle.CopyFrom(bundle)
         del new_bundle.archives[:]
         new_bundle.AddArchive(archive)
         self[path + '.json'] = new_bundle.GetDataAsString()
 
 
 class TestDelegate(update_nacl_manifest.Delegate):
   def __init__(self, manifest, history, files, version_mapping):
     self.manifest = manifest
     self.history = history
     self.files = files
     self.version_mapping = version_mapping
     self.dryrun = 0
+    self.called_gsutil_cp = False
 
   def GetRepoManifest(self):
     return self.manifest
 
   def GetHistory(self):
     return self.history
 
   def GetTrunkRevision(self, version):
     return self.version_mapping[version]
 
   def GsUtil_ls(self, url):
     path = GetPathFromGsUrl(url)
     result = []
     for filename, _ in self.files.iteritems():
       if filename.startswith(path):
         result.append(MakeGsUrl(filename))
     return result
 
   def GsUtil_cat(self, url):
     path = GetPathFromGsUrl(url)
     if path not in self.files:
       raise subprocess.CalledProcessError(1, 'gsutil cat %s' % (url,))
     return self.files[path]
 
   def GsUtil_cp(self, src, dest, stdin=None):
+    self.called_gsutil_cp = True
     dest_path = GetPathFromGsUrl(dest)
     if src == '-':
       self.files[dest_path] = stdin
     else:
       src_path = GetPathFromGsUrl(src)
       if src_path not in self.files:
         raise subprocess.CalledProcessError(1, 'gsutil cp %s %s' % (src, dest))
       self.files[dest_path] = self.files[src_path]
 
   def Print(self, *args):
@@ skipping 30 matching lines @@
     self.files = MakeFiles()
     self.version_mapping = {}
     self.delegate = None
     self.uploaded_manifest = None
     self.manifest = None
 
   def _MakeDelegate(self):
     self.delegate = TestDelegate(self.manifest, self.history.history,
         self.files, self.version_mapping)
 
-  def _Run(self, host_oses):
-    update_nacl_manifest.Run(self.delegate, host_oses)
+  def _Run(self, host_oses, fixed_bundle_versions=None):
+    update_nacl_manifest.Run(self.delegate, host_oses, fixed_bundle_versions)
 
   def _HasUploadedManifest(self):
     return 'naclsdk_manifest2.json' in self.files
 
   def _ReadUploadedManifest(self):
     self.uploaded_manifest = manifest_util.SDKManifest()
     self.uploaded_manifest.LoadDataFromString(
         self.files['naclsdk_manifest2.json'])
 
   def _AssertUploadedManifestHasBundle(self, bundle, stability):
@@ skipping 16 matching lines @@
     import csv
     import cStringIO
     history_stream = cStringIO.StringIO(history)
     self.history.history = [(platform, channel, version, date)
         for platform, channel, version, date in csv.reader(history_stream)]
 
   def testNoUpdateNeeded(self):
     self.manifest = MakeManifest(B18_0_1025_163_R1_MLW)
     self._MakeDelegate()
     self._Run(OS_MLW)
-    self.assertEqual(self._HasUploadedManifest(), False)
+    self.assertFalse(self._HasUploadedManifest())
 
     # Add another bundle, make sure it still doesn't update
     self.manifest.AddBundle(B19_0_1084_41_R1_MLW)
     self._Run(OS_MLW)
-    self.assertEqual(self._HasUploadedManifest(), False)
+    self.assertFalse(self._HasUploadedManifest())
 
   def testSimpleUpdate(self):
     self.manifest = MakeManifest(B18_R1_NONE)
     self.history.Add(OS_MLW, BETA, V18_0_1025_163)
     self.files.Add(B18_0_1025_163_R1_MLW)
     self._MakeDelegate()
     self._Run(OS_MLW)
     self._ReadUploadedManifest()
     self._AssertUploadedManifestHasBundle(B18_0_1025_163_R1_MLW, BETA)
     self.assertEqual(len(self.uploaded_manifest.GetBundles()), 1)
@@ skipping 84 matching lines @@
       if channel == STABLE:
         self.assertEqual(uploaded_bundle.recommended, 'yes')
       else:
         self.assertEqual(uploaded_bundle.recommended, 'no')
 
   def testNoUpdateWithNonPepperBundle(self):
     self.manifest = MakeManifest(NON_PEPPER_BUNDLE_NOARCHIVES,
         B18_0_1025_163_R1_MLW)
     self._MakeDelegate()
     self._Run(OS_MLW)
-    self.assertEqual(self._HasUploadedManifest(), False)
+    self.assertFalse(self._HasUploadedManifest())
 
   def testUpdateWithHistoryWithExtraneousPlatforms(self):
     self.manifest = MakeManifest(B18_R1_NONE)
     self.history.Add(OS_ML, BETA, V18_0_1025_184)
     self.history.Add(OS_CR, BETA, V18_0_1025_184)
     self.history.Add(OS_CR, BETA, V18_0_1025_175)
     self.history.Add(OS_MLW, BETA, V18_0_1025_163)
     self.files.Add(B18_0_1025_163_R1_MLW)
     self._MakeDelegate()
     self._Run(OS_MLW)
@@ skipping 124 matching lines @@
     self.files.Add(B18_0_1025_163_R1_MLW)
     self.files.Add(B19_0_1084_41_R1_MLW)
     self._MakeDelegate()
     self._Run(OS_MLW)
     self._ReadUploadedManifest()
     p18_bundle = self.uploaded_manifest.GetBundle(B18_R1_NONE.name)
     self.assertEqual(p18_bundle.stability, POST_STABLE)
     p19_bundle = self.uploaded_manifest.GetBundle(B19_R1_NONE.name)
     self.assertEqual(p19_bundle.stability, STABLE)
 
+  def testDontPushIfNoChange(self):
+    # Make an online manifest that already has this bundle.
+    online_manifest = MakeManifest(B18_0_1025_163_R1_MLW)
+    self.files.AddOnlineManifest(online_manifest.GetDataAsString())
+
+    self.manifest = MakeManifest(B18_R1_NONE)
+    self.history.Add(OS_MLW, DEV, V18_0_1025_163)
+    self.files.Add(B18_0_1025_163_R1_MLW)
+
+    self._MakeDelegate()
+    self._Run(OS_MLW)
+    self.assertFalse(self.delegate.called_gsutil_cp)
+
+  def testDontPushIfRollback(self):
+    # Make an online manifest that has a newer bundle
+    online_manifest = MakeManifest(B18_0_1025_184_R1_MLW)
+    self.files.AddOnlineManifest(online_manifest.GetDataAsString())
+
+    self.manifest = MakeManifest(B18_R1_NONE)
+    self.history.Add(OS_MLW, DEV, V18_0_1025_163)
+    self.files.Add(B18_0_1025_163_R1_MLW)
+
+    self._MakeDelegate()
+    self._Run(OS_MLW)
+    self.assertFalse(self.delegate.called_gsutil_cp)
+
+  def testRunWithFixedBundleVersions(self):
+    self.manifest = MakeManifest(B18_R1_NONE)
+    self.history.Add(OS_MLW, BETA, V18_0_1025_163)
+    self.files.Add(B18_0_1025_163_R1_MLW)
+    self.files.Add(B18_0_1025_184_R1_MLW)
+
+    self._MakeDelegate()
+    self._Run(OS_MLW, [('pepper_18', '18.0.1025.184')])
+    self._ReadUploadedManifest()
+    self._AssertUploadedManifestHasBundle(B18_0_1025_184_R1_MLW, BETA)
+    self.assertEqual(len(self.uploaded_manifest.GetBundles()), 1)
+
+  def testRunWithMissingFixedBundleVersions(self):
+    self.manifest = MakeManifest(B18_R1_NONE)
+    self.history.Add(OS_MLW, BETA, V18_0_1025_163)
+    self.files.Add(B18_0_1025_163_R1_MLW)
+
+    self._MakeDelegate()
+    self._Run(OS_MLW, [('pepper_18', '18.0.1025.184')])
+    # Nothing should be uploaded if the user gives a missing fixed version.
+    self.assertFalse(self.delegate.called_gsutil_cp)
+
 
 class TestUpdateVitals(unittest.TestCase):
   def setUp(self):
     f = tempfile.NamedTemporaryFile('w', prefix="test_update_nacl_manifest")
     self.test_file = f.name
     f.close()
     test_data = "Some test data"
     self.sha1 = hashlib.sha1(test_data).hexdigest()
     self.data_len = len(test_data)
     with open(self.test_file, 'w') as f:
@@ skipping 36 matching lines @@
       '23.0.1271.89': '167132',
       '24.0.1305.4': '164971',
     }
     for version, revision in revision_dict.iteritems():
       self.assertEqual('trunk.%s' % revision,
                        self.delegate.GetTrunkRevision(version))
 
 
 if __name__ == '__main__':
   sys.exit(unittest.main())