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