browser_credentials.py revision 6e8cce623b6e4fe0c9e4af605d675dd9d0338c38 
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
5import json
6import logging
7import os
8
9from telemetry.core import util
10from telemetry.core.backends import facebook_credentials_backend
11from telemetry.core.backends import google_credentials_backend
12from telemetry.unittest import options_for_unittests
13
14
15class CredentialsError(Exception):
16  """Error that can be thrown when logging in."""
17
18
19class BrowserCredentials(object):
20  def __init__(self, backends = None):
21    self._credentials = {}
22    self._credentials_path = None
23    self._extra_credentials = {}
24
25    if backends is None:
26      backends = [
27        facebook_credentials_backend.FacebookCredentialsBackend(),
28        google_credentials_backend.GoogleCredentialsBackend()]
29
30    self._backends = {}
31    for backend in backends:
32      self._backends[backend.credentials_type] = backend
33
34  def AddBackend(self, backend):
35    assert backend.credentials_type not in self._backends
36    self._backends[backend.credentials_type] = backend
37
38  def IsLoggedIn(self, credentials_type):
39    if credentials_type not in self._backends:
40      raise CredentialsError(
41          'Unrecognized credentials type: %s', credentials_type)
42    if credentials_type not in self._credentials:
43      return False
44    return self._backends[credentials_type].IsLoggedIn()
45
46  def CanLogin(self, credentials_type):
47    if credentials_type not in self._backends:
48      raise CredentialsError(
49          'Unrecognized credentials type: %s', credentials_type)
50    return credentials_type in self._credentials
51
52  def LoginNeeded(self, tab, credentials_type):
53    if credentials_type not in self._backends:
54      raise CredentialsError(
55          'Unrecognized credentials type: %s', credentials_type)
56    if credentials_type not in self._credentials:
57      return False
58    return self._backends[credentials_type].LoginNeeded(
59      tab, self._credentials[credentials_type])
60
61  def LoginNoLongerNeeded(self, tab, credentials_type):
62    assert credentials_type in self._backends
63    self._backends[credentials_type].LoginNoLongerNeeded(tab)
64
65  @property
66  def credentials_path(self):
67    return self._credentials_path
68
69  @credentials_path.setter
70  def credentials_path(self, credentials_path):
71    self._credentials_path = credentials_path
72    self._RebuildCredentials()
73
74  def Add(self, credentials_type, data):
75    if credentials_type not in self._extra_credentials:
76      self._extra_credentials[credentials_type] = {}
77    for k, v in data.items():
78      assert k not in self._extra_credentials[credentials_type]
79      self._extra_credentials[credentials_type][k] = v
80    self._RebuildCredentials()
81
82  def _ResetLoggedInState(self):
83    """Makes the backends think we're not logged in even though we are.
84    Should only be used in unit tests to simulate --dont-override-profile.
85    """
86    for backend in self._backends.keys():
87      self._backends[backend]._ResetLoggedInState() # pylint: disable=W0212
88
89  def _RebuildCredentials(self):
90    credentials = {}
91    if self._credentials_path == None:
92      pass
93    elif os.path.exists(self._credentials_path):
94      with open(self._credentials_path, 'r') as f:
95        credentials = json.loads(f.read())
96
97    # TODO(nduca): use system keychain, if possible.
98    homedir_credentials_path = os.path.expanduser('~/.telemetry-credentials')
99    homedir_credentials = {}
100
101    if (not options_for_unittests.GetCopy() and
102        os.path.exists(homedir_credentials_path)):
103      logging.info("Found ~/.telemetry-credentials. Its contents will be used "
104                   "when no other credentials can be found.")
105      with open(homedir_credentials_path, 'r') as f:
106        homedir_credentials = json.loads(f.read())
107
108    self._credentials = {}
109    all_keys = set(credentials.keys()).union(
110      homedir_credentials.keys()).union(
111      self._extra_credentials.keys())
112
113    for k in all_keys:
114      if k in credentials:
115        self._credentials[k] = credentials[k]
116      if k in homedir_credentials:
117        logging.info("Will use ~/.telemetry-credentials for %s logins." % k)
118        self._credentials[k] = homedir_credentials[k]
119      if k in self._extra_credentials:
120        self._credentials[k] = self._extra_credentials[k]
121
122  def WarnIfMissingCredentials(self, page_set):
123    num_pages_missing_login = 0
124    missing_credentials = set()
125    for page in page_set:
126      if (page.credentials
127          and not self.CanLogin(page.credentials)):
128        num_pages_missing_login += 1
129        missing_credentials.add(page.credentials)
130
131    if num_pages_missing_login > 0:
132      files_to_tweak = []
133      if page_set.credentials_path:
134        files_to_tweak.append(
135          os.path.relpath(os.path.join(os.path.dirname(page_set.file_path),
136                                       page_set.credentials_path)))
137      files_to_tweak.append('~/.telemetry-credentials')
138
139      example_credentials_file = os.path.join(
140          util.GetTelemetryDir(), 'examples', 'credentials_example.json')
141
142      logging.warning("""
143        Credentials for %s were not found. %i pages will not be tested.
144
145        To fix this, either add svn-internal to your .gclient using
146        http://goto/read-src-internal, or add your own credentials to:
147            %s
148        An example credentials file you can copy from is here:
149            %s\n""" % (', '.join(missing_credentials),
150         num_pages_missing_login,
151         ' or '.join(files_to_tweak),
152         example_credentials_file))
153