blob: 9cf85c679cee6b48ee65d82d433f4ecfdbe65f07 [file] [log] [blame]
# Copyright (c) 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import os
import sys
import urllib2
from multiprocessing import Process
from autotest_lib.client.bin import utils
policy_testserver = None
class FakeDMServer(object):
"""Utility class for policy tests."""
def __init__(self, proto_path):
Import the DM testserver from chrome source.
@param proto_path: location of proto files.
self.server_url = None
telemetry_src = '/usr/local/telemetry/src'
for path in ['chrome/browser/policy/test',
sys.path.append(os.path.join(telemetry_src, path))
global policy_testserver
import policy_testserver
def start(self, tmpdir, debugdir):
Start the local DM testserver.
@param tmpdir: location of the Autotest tmp dir.
@param debugdir: location of the Autotest debug directory.
policy_server_runner = policy_testserver.PolicyServerRunner()
self._policy_location = os.path.join(tmpdir, 'policy.json')
port = utils.get_unused_port()
# The first argument is always ignored since it is expected to be the
# path to the executable. Hence passing an empty string for first
# argument.
sys.argv = ['',
'--config-file=%s' % self._policy_location,
'--log-file=%s/dm_server.log' % debugdir,
'--port=%d' % port
self.process = Process(target=policy_server_runner.main)
self.server_url = '' % port
def stop(self):
"""Terminate the fake DM server instance."""
if urllib2.urlopen('%stest/ping' % self.server_url).getcode() == 200:
urllib2.urlopen('%sconfiguration/test/exit' % self.server_url)
if self.process.is_alive():
def setup_policy(self, policy_blob):
Write policy blob to file used by the DM server to read policy.
@param policy_blob: JSON policy blob to be written to the policy file.
with open(self._policy_location, 'w') as f: