• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright (c) 2012 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 logging
6import os
7import pwd
8import subprocess
9import tempfile
10
11from autotest_lib.client.bin import test
12from autotest_lib.client.common_lib import error
13
14class TLSDate:
15    """
16    A single tlsdate invocation. Takes care of setting up a temporary cachedir
17    for it, along with collecting output from both it and its helper processes.
18    """
19    def __init__(self, test_obj):
20        self._proc = None
21        self._testdir = tempfile.mkdtemp(suffix='tlsdate')
22        self._cachedir = self._testdir + '/cache'
23        self._outfile = self._testdir + '/out'
24        self._subprog = '?'
25        self._test_obj = test_obj
26        self._output = None
27        self._tlsdate_uid = pwd.getpwnam('tlsdate').pw_uid
28        os.mkdir(self._cachedir)
29        # Let the tlsdate user (tlsdate) write.
30        os.chown(self._testdir, self._tlsdate_uid, -1)
31        # Allow support shell library to be sourced.
32        os.chown(self._test_obj.srcdir + '/time.sh', self._tlsdate_uid, -1)
33
34
35    def start(self, subprog):
36        logging.info('running with %s/%s', self._test_obj.srcdir, subprog)
37        self._subprog = subprog
38        # Make sure the tlsdate user can access the files
39        fake_tlsdate = self._test_obj.srcdir + '/' + subprog
40        os.chown(fake_tlsdate, self._tlsdate_uid, -1)
41        args = ['/usr/bin/tlsdated', '-p',
42                '-f', self._test_obj.srcdir + '/test.conf',
43                '-c', self._cachedir,
44                '-v',
45                fake_tlsdate,
46                self._outfile]
47        self._proc = subprocess.Popen(args, stdin=subprocess.PIPE,
48                                      stderr=subprocess.PIPE)
49
50
51    def route_up(self):
52        self._proc.stdin.write('n')
53        self._proc.stdin.flush()
54
55
56    def kill(self):
57        self._proc.terminate()
58
59
60    def output(self):
61        if not self._output:
62            self._output = self._proc.communicate()[1].split('\n')
63        return self._output
64
65
66    def in_output(self, string):
67        for x in self.output():
68            if string in x:
69                return True
70        return False
71
72
73    def subproc_output(self):
74        with open(self._outfile) as f:
75            return [x.rstrip() for x in f.readlines()]
76
77
78    def ok(self):
79        return 'ok' in self.subproc_output()
80
81
82class platform_TLSDate(test.test):
83    version = 1
84
85    def require_ok(self, t):
86        if not t.ok():
87            raise error.TestFail('Expected success, got:' +
88                                 ';'.join(t.subproc_output()))
89
90
91    def require_output(self, t, string):
92        if not t.in_output(string):
93            raise error.TestFail('Needed "%s" but got "%s"' % (string,
94                                 ';'.join(t.output())))
95
96
97    def require_not_output(self, t, string):
98        if t.in_output(string):
99            raise error.TestFail('Needed no "%s" but got "%s"' % (string,
100                                 ';'.join(t.output())))
101
102
103    def test_delay_subproc(self):
104        """
105        Tests that a subprocess that delays for one second is waited on
106        successfully the second time.
107        """
108        t = TLSDate(self)
109        t.start('delay_subproc')
110        self.require_output(t, 'attempt 1 backoff')
111        self.require_output(t, 'time set from the network')
112        self.require_ok(t)
113
114
115    def test_hang_subproc(self):
116        """
117        Tests that a subprocess that delays for too long is considered hung and
118        killed.
119        """
120        t = TLSDate(self)
121        t.start('hang_subproc')
122        self.require_output(t, 'attempt 1 backoff')
123        self.require_output(t, 'tlsdate timed out')
124        self.require_ok(t)
125
126
127    def test_fail_routes(self):
128        """
129        Tests that if the initial tlsdate call fails, we wait for a route to
130        appear, then rerun tlsdate.
131        """
132        t = TLSDate(self)
133        t.start('fail_routes')
134        t.route_up()
135        self.require_output(t, 'status:2')
136        self.require_output(t, 'stdin')
137        self.require_output(t, 'time set from the network')
138        self.require_ok(t)
139
140
141    def run_once(self):
142        self.test_delay_subproc()
143        self.test_hang_subproc()
144        self.test_fail_routes()
145