1#!/usr/bin/env python3
2
3# Copyright 2016 - The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9#     http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16
17# Import the python3 compatible bytes()
18from builtins import bytes
19
20import mock
21import os
22import sys
23import unittest
24
25from acts.libs.proc import job
26
27if os.name == 'posix' and sys.version_info[0] < 3:
28    import subprocess32 as subprocess
29else:
30    import subprocess
31
32
33class FakePopen(object):
34    """A fake version of the object returned from subprocess.Popen()."""
35
36    def __init__(self,
37                 stdout=None,
38                 stderr=None,
39                 returncode=0,
40                 will_timeout=False):
41        self.returncode = returncode
42        self._stdout = bytes(stdout,
43                             'utf-8') if stdout is not None else bytes()
44        self._stderr = bytes(stderr,
45                             'utf-8') if stderr is not None else bytes()
46        self._will_timeout = will_timeout
47
48    def communicate(self, timeout=None):
49        if self._will_timeout:
50            raise subprocess.TimeoutExpired(
51                -1, 'Timed out according to test logic')
52        return self._stdout, self._stderr
53
54    def kill(self):
55        pass
56
57    def wait(self):
58        pass
59
60
61class JobTestCases(unittest.TestCase):
62    @mock.patch(
63        'acts.libs.proc.job.subprocess.Popen',
64        return_value=FakePopen(stdout='TEST\n'))
65    def test_run_success(self, popen):
66        """Test running a simple shell command."""
67        result = job.run('echo TEST')
68        self.assertTrue(result.stdout.startswith('TEST'))
69
70    @mock.patch(
71        'acts.libs.proc.job.subprocess.Popen',
72        return_value=FakePopen(stderr='TEST\n'))
73    def test_run_stderr(self, popen):
74        """Test that we can read process stderr."""
75        result = job.run('echo TEST 1>&2')
76        self.assertEqual(len(result.stdout), 0)
77        self.assertTrue(result.stderr.startswith('TEST'))
78        self.assertFalse(result.stdout)
79
80    @mock.patch(
81        'acts.libs.proc.job.subprocess.Popen',
82        return_value=FakePopen(returncode=1))
83    def test_run_error(self, popen):
84        """Test that we raise on non-zero exit statuses."""
85        self.assertRaises(job.Error, job.run, 'exit 1')
86
87    @mock.patch(
88        'acts.libs.proc.job.subprocess.Popen',
89        return_value=FakePopen(returncode=1))
90    def test_run_with_ignored_error(self, popen):
91        """Test that we can ignore exit status on request."""
92        result = job.run('exit 1', ignore_status=True)
93        self.assertEqual(result.exit_status, 1)
94
95    @mock.patch(
96        'acts.libs.proc.job.subprocess.Popen',
97        return_value=FakePopen(will_timeout=True))
98    def test_run_timeout(self, popen):
99        """Test that we correctly implement command timeouts."""
100        self.assertRaises(job.Error, job.run, 'sleep 5', timeout=0.1)
101
102    @mock.patch(
103        'acts.libs.proc.job.subprocess.Popen',
104        return_value=FakePopen(stdout='TEST\n'))
105    def test_run_no_shell(self, popen):
106        """Test that we handle running without a wrapping shell."""
107        result = job.run(['echo', 'TEST'])
108        self.assertTrue(result.stdout.startswith('TEST'))
109
110    @mock.patch(
111        'acts.libs.proc.job.subprocess.Popen',
112        return_value=FakePopen(stdout='TEST\n'))
113    def test_job_env(self, popen):
114        """Test that we can set environment variables correctly."""
115        test_env = {'MYTESTVAR': '20'}
116        result = job.run('printenv', env=test_env.copy())
117        popen.assert_called_once()
118        _, kwargs = popen.call_args
119        self.assertTrue('env' in kwargs)
120        self.assertEqual(kwargs['env'], test_env)
121
122
123if __name__ == '__main__':
124    unittest.main()
125