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