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