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