Home | History | Annotate | Download | only in enterprise 
      1 # Copyright (c) 2016 The Chromium OS 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 os
      6 import sys
      7 import urllib2
      8 from multiprocessing import Process
      9 
     10 from autotest_lib.client.bin import utils
     11 
     12 policy_testserver = None
     13 
     14 
     15 class FakeDMServer(object):
     16     """Utility class for policy tests."""
     17 
     18     def __init__(self, proto_path):
     19         """
     20         Import the DM testserver from chrome source.
     21 
     22         @param proto_path: location of proto files.
     23 
     24         """
     25         self.server_url = None
     26         telemetry_src = '/usr/local/telemetry/src'
     27         for path in ['chrome/browser/policy/test',
     28                      'net/tools/testserver',
     29                      'third_party/protobuf/python/google',
     30                      'third_party/tlslite']:
     31             sys.path.append(os.path.join(telemetry_src, path))
     32         sys.path.append(proto_path)
     33         global policy_testserver
     34         import policy_testserver
     35 
     36     def start(self, tmpdir, debugdir):
     37         """
     38         Start the local DM testserver.
     39 
     40         @param tmpdir: location of the Autotest tmp dir.
     41         @param debugdir: location of the Autotest debug directory.
     42 
     43         """
     44         policy_server_runner = policy_testserver.PolicyServerRunner()
     45         self._policy_location = os.path.join(tmpdir, 'policy.json')
     46         port = utils.get_unused_port()
     47         # The first argument is always ignored since it is expected to be the
     48         # path to the executable. Hence passing an empty string for first
     49         # argument.
     50         sys.argv = ['',
     51                     '--config-file=%s' % self._policy_location,
     52                     '--host=127.0.0.1',
     53                     '--log-file=%s/dm_server.log' % debugdir,
     54                     '--log-level=DEBUG',
     55                     '--port=%d' % port
     56 
     57                    ]
     58         self.process = Process(target=policy_server_runner.main)
     59         self.process.start()
     60         self.server_url = 'http://127.0.0.1:%d/' % port
     61 
     62     def stop(self):
     63         """Terminate the fake DM server instance."""
     64         if urllib2.urlopen('%stest/ping' % self.server_url).getcode() == 200:
     65             urllib2.urlopen('%sconfiguration/test/exit' % self.server_url)
     66         if self.process.is_alive():
     67             self.process.join()
     68 
     69     def setup_policy(self, policy_blob):
     70         """
     71         Write policy blob to file used by the DM server to read policy.
     72 
     73         @param policy_blob: JSON policy blob to be written to the policy file.
     74 
     75         """
     76         with open(self._policy_location, 'w') as f:
     77             f.write(policy_blob)
     78