src/tools/telemetry/telemetry/page/page_set.py

   1 # Copyright 2012 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
   5 import csv
   6 import inspect
   7 import os
   8
   9 from telemetry.page import page as page_module
  10 from telemetry.page import page_set_archive_info
  11 from telemetry.user_story import user_story_set
  12 from telemetry.util import cloud_storage
  13
  14 PUBLIC_BUCKET = cloud_storage.PUBLIC_BUCKET
  15 PARTNER_BUCKET = cloud_storage.PARTNER_BUCKET
  16 INTERNAL_BUCKET = cloud_storage.INTERNAL_BUCKET
  17
  18
  19 class PageSetError(Exception):
  20   pass
  21
  22
  23 class PageSet(user_story_set.UserStorySet):
  24   def __init__(self, file_path=None, archive_data_file='', user_agent_type=None,
  25                make_javascript_deterministic=True, startup_url='',
  26                serving_dirs=None, bucket=None):
  27     super(PageSet, self).__init__()
  28     # The default value of file_path is location of the file that define this
  29     # page set instance's class.
  30     if file_path is None:
  31       file_path = inspect.getfile(self.__class__)
  32       # Turn pyc file into py files if we can
  33       if file_path.endswith('.pyc') and os.path.exists(file_path[:-1]):
  34         file_path = file_path[:-1]
  35
  36     self.file_path = file_path
  37     # These attributes can be set dynamically by the page set.
  38     self.archive_data_file = archive_data_file
  39     self.user_agent_type = user_agent_type
  40     self.make_javascript_deterministic = make_javascript_deterministic
  41     self._wpr_archive_info = None
  42     self.startup_url = startup_url
  43     self.user_stories = []
  44     # Convert any relative serving_dirs to absolute paths.
  45     self._serving_dirs = set(os.path.realpath(os.path.join(self.base_dir, d))
  46                              for d in serving_dirs or [])
  47     if self._IsValidPrivacyBucket(bucket):
  48       self._bucket = bucket
  49     else:
  50       raise ValueError("Pageset privacy bucket %s is invalid" % bucket)
  51
  52   @property
  53   def pages(self):
  54     return self.user_stories
  55
  56   def AddUserStory(self, user_story):
  57     assert isinstance(user_story, page_module.Page)
  58     assert user_story.page_set is self
  59     super(PageSet, self).AddUserStory(user_story)
  60
  61   def AddPage(self, page):
  62     self.AddUserStory(page)
  63
  64   def AddPageWithDefaultRunNavigate(self, page_url):
  65     """ Add a simple page with url equals to page_url that contains only default
  66     RunNavigateSteps.
  67     """
  68     self.AddUserStory(page_module.Page(
  69       page_url, self, self.base_dir))
  70
  71   @staticmethod
  72   def _IsValidPrivacyBucket(bucket_name):
  73     return bucket_name in (None, PUBLIC_BUCKET, PARTNER_BUCKET, INTERNAL_BUCKET)
  74
  75   @property
  76   def base_dir(self):
  77     if os.path.isfile(self.file_path):
  78       return os.path.dirname(self.file_path)
  79     else:
  80       return self.file_path
  81
  82   @property
  83   def serving_dirs(self):
  84     return self._serving_dirs
  85
  86   @property
  87   def wpr_archive_info(self):  # pylint: disable=E0202
  88     """Lazily constructs wpr_archive_info if it's not set and returns it."""
  89     if self.archive_data_file and not self._wpr_archive_info:
  90       self._wpr_archive_info = (
  91           page_set_archive_info.PageSetArchiveInfo.FromFile(
  92             os.path.join(self.base_dir, self.archive_data_file)))
  93     return self._wpr_archive_info
  94
  95   @property
  96   def bucket(self):
  97     return self._bucket
  98
  99   @wpr_archive_info.setter
 100   def wpr_archive_info(self, value):  # pylint: disable=E0202
 101     self._wpr_archive_info = value
 102
 103   def ContainsOnlyFileURLs(self):
 104     for page in self.user_stories:
 105       if not page.is_file:
 106         return False
 107     return True
 108
 109   def ReorderPageSet(self, results_file):
 110     """Reorders this page set based on the results of a past run."""
 111     page_set_dict = {}
 112     for page in self.user_stories:
 113       page_set_dict[page.url] = page
 114
 115     user_stories = []
 116     with open(results_file, 'rb') as csv_file:
 117       csv_reader = csv.reader(csv_file)
 118       csv_header = csv_reader.next()
 119
 120       if 'url' not in csv_header:
 121         raise Exception('Unusable results_file.')
 122
 123       url_index = csv_header.index('url')
 124
 125       for csv_row in csv_reader:
 126         if csv_row[url_index] in page_set_dict:
 127           self.AddPage(page_set_dict[csv_row[url_index]])
 128         else:
 129           raise Exception('Unusable results_file.')
 130
 131     return user_stories
 132
 133   def WprFilePathForPage(self, page):
 134     if not self.wpr_archive_info:
 135       return None
 136     return self.wpr_archive_info.WprFilePathForPage(page)