Home | History | Annotate | Download | only in client_lib 
      1 #! /usr/bin/python
      2 
      3 # Copyright 2014 The Chromium OS Authors. All rights reserved.
      4 # Use of this source code is governed by a BSD-style license that can be
      5 # found in the LICENSE file.
      6 
      7 """Small integration test for registration client
      8 
      9 This client can work either with the fake_device_server or a live server.
     10 To use locally (with the fake device server), start the server in the
     11 background (e.g. ../server.py) and run the program without arguments.
     12 
     13 Otherwise, if you want to run against a live server, you must provide an
     14 auth code. To get an auth code, run this script with the argument URL
     15 which will print out a link for you to visit and get your auth code.
     16 
     17 Then re-run the test with that auth code like so:
     18 
     19 ./client_lib_test <YOUR_AUTH_CODE>.
     20 ."""
     21 
     22 import argparse
     23 import logging
     24 import sys
     25 import urllib2
     26 
     27 import commands
     28 import devices
     29 import oauth_helpers
     30 import registration
     31 
     32 
     33 API_KEY = 'AIzaSyC55ef0RkaFTQvGvTXL_HIh6KI3pzVq4w0'
     34 CLIENT_ID = ('522003936346-odpbgftanpuruuqhf1puk9e0' +
     35              'p2d5ldho.apps.googleusercontent.com')
     36 CLIENT_SECRET = '9Om2cR2_5cKIKhSY5OFFo8uX'
     37 SERVER_URL = 'https://www.googleapis.com/clouddevices/v1'
     38 
     39 
     40 def parse_args(args):
     41     """Arg parser for this tiny program."""
     42     parser = argparse.ArgumentParser(usage=__doc__)
     43     parser.add_argument('auth_code', nargs='?',
     44                         help=('Either your auth code or "URL" to return the'
     45                               ' url to visit to get the code. If not'
     46                               ' specified, runs test through local fake server.'
     47                               ))
     48     return parser.parse_args(args)
     49 
     50 
     51 def main(args):
     52     """Main method for integration test."""
     53     server_url, api_key = 'http://localhost:8080', None
     54     access_token = None
     55 
     56     parsed_args = parse_args(args)
     57     if parsed_args.auth_code == 'URL':
     58         print oauth_helpers.get_oauth2_auth_url(CLIENT_ID)
     59         return 0
     60     elif parsed_args.auth_code:
     61         server_url, api_key = SERVER_URL, API_KEY
     62         access_token = oauth_helpers.get_oauth2_user_token(
     63               CLIENT_ID, CLIENT_SECRET, parsed_args.auth_code)
     64 
     65     r_client = registration.RegistrationClient(server_url=server_url,
     66                                                api_key=api_key,
     67                                                access_token=access_token)
     68     # Device should support base.reboot command.
     69     base_reboot_command = {'reboot': {}}
     70     finalized_ticket = r_client.register_device(
     71             'test_device', 'vendor', 'xmpp', oauth_client_id=CLIENT_ID,
     72             base=base_reboot_command)
     73     new_device_id = finalized_ticket['deviceDraft']['id']
     74     print 'Registered new device', finalized_ticket
     75 
     76     # TODO(sosa): Do better. Change this to use fake auth server when it exists.
     77     if not parsed_args.auth_code:
     78         robot_token = None
     79     else:
     80         robot_token = oauth_helpers.get_oauth2_robot_token(
     81                 CLIENT_ID, CLIENT_SECRET,
     82                 finalized_ticket['robotAccountAuthorizationCode'])
     83 
     84     d_client = devices.DevicesClient(server_url=server_url,
     85                                      api_key=api_key, access_token=robot_token)
     86     if not d_client.get_device(new_device_id):
     87         print 'Device not found in database'
     88         return 1
     89 
     90     device_list = d_client.list_devices()['devices']
     91     device_ids = [device['id'] for device in device_list]
     92     if not new_device_id in device_ids:
     93         print 'Device found but not listed correctly'
     94         return 1
     95 
     96 
     97     # TODO(sosa): Figure out why I can't send commands.
     98     c_client = commands.CommandsClient(server_url=server_url,
     99                                        api_key=api_key,
    100                                        access_token=robot_token)
    101     command_dict = {'base': {'reboot': {}}}
    102     new_command = c_client.create_command(device['id'], command_dict)
    103     if not c_client.get_command(new_command['id']):
    104         print 'Command not found'
    105         return 1
    106 
    107     command_list = c_client.list_commands(device['id'])['commands']
    108     command_ids = [c['id'] for c in command_list]
    109     if not new_command['id'] in command_ids:
    110         print 'Command found but not listed correctly'
    111         return 1
    112 
    113     new_command = c_client.update_command(new_command['id'],
    114                                           {'state':'finished'})
    115     return 0
    116 
    117 
    118 if __name__ == '__main__':
    119     logging_format = '%(asctime)s - %(filename)s - %(levelname)-8s: %(message)s'
    120     date_format = '%H:%M:%S'
    121     logging.basicConfig(level=logging.DEBUG, format=logging_format,
    122                         datefmt=date_format)
    123     try:
    124         error_code = main(sys.argv[1:])
    125         if error_code != 0:
    126             print 'Test Failed'
    127 
    128         sys.exit(error_code)
    129     except urllib2.HTTPError as e:
    130         print 'Received an HTTPError exception!!!'
    131         print e
    132         print e.read()
    133         sys.exit(1)
    134