• 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, 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