#!/usr/bin/env python3 # Copyright 2019 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Unittests for the utils module.""" import datetime import os from pathlib import Path import sys import unittest _path = os.path.realpath(__file__ + '/../..') if sys.path[0] != _path: sys.path.insert(0, _path) del _path # We have to import our local modules after the sys.path tweak. We can't use # relative imports because this is an executable program, not a module. # pylint: disable=wrong-import-position import rh import rh.utils class TimeDeltaStrTests(unittest.TestCase): """Verify behavior of timedelta_str object.""" def test_same(self): """Check timedelta of 0 seconds.""" delta = datetime.timedelta(0) self.assertEqual('0.000s', rh.utils.timedelta_str(delta)) def test_millisecondss(self): """Check timedelta of milliseconds.""" delta = datetime.timedelta(seconds=0.123456) self.assertEqual('0.123s', rh.utils.timedelta_str(delta)) def test_seconds(self): """Check timedelta of seconds.""" delta = datetime.timedelta(seconds=12.3) self.assertEqual('12.300s', rh.utils.timedelta_str(delta)) def test_minutes(self): """Check timedelta of minutes.""" delta = datetime.timedelta(seconds=72.3) self.assertEqual('1m12.300s', rh.utils.timedelta_str(delta)) def test_hours(self): """Check timedelta of hours.""" delta = datetime.timedelta(seconds=4000.3) self.assertEqual('1h6m40.300s', rh.utils.timedelta_str(delta)) class CompletedProcessTests(unittest.TestCase): """Verify behavior of CompletedProcess object.""" def test_empty_cmdstr(self): """Check cmdstr with an empty command.""" result = rh.utils.CompletedProcess(args=[]) self.assertEqual('', result.cmdstr) def test_basic_cmdstr(self): """Check cmdstr with a basic command command.""" result = rh.utils.CompletedProcess(args=['ls', 'a b']) self.assertEqual("ls 'a b'", result.cmdstr) def test_str(self): """Check str() handling.""" # We don't enforce much, just that it doesn't crash. result = rh.utils.CompletedProcess() self.assertNotEqual('', str(result)) result = rh.utils.CompletedProcess(args=[]) self.assertNotEqual('', str(result)) def test_repr(self): """Check repr() handling.""" # We don't enforce much, just that it doesn't crash. result = rh.utils.CompletedProcess() self.assertNotEqual('', repr(result)) result = rh.utils.CompletedProcess(args=[]) self.assertNotEqual('', repr(result)) class CalledProcessErrorTests(unittest.TestCase): """Verify behavior of CalledProcessError object.""" def test_basic(self): """Basic test we can create a normal instance.""" rh.utils.CalledProcessError(0, ['mycmd']) def test_stringify(self): """Check stringify() handling.""" # We don't assert much so we leave flexibility in changing format. err = rh.utils.CalledProcessError(0, ['mycmd']) self.assertIn('mycmd', err.stringify()) def test_str(self): """Check str() handling.""" # We don't assert much so we leave flexibility in changing format. err = rh.utils.CalledProcessError(0, ['mycmd']) self.assertIn('mycmd', str(err)) def test_repr(self): """Check repr() handling.""" # We don't assert much so we leave flexibility in changing format. err = rh.utils.CalledProcessError(0, ['mycmd']) self.assertNotEqual('', repr(err)) def test_output(self): """Make sure .output is removed and .stdout works.""" e = rh.utils.CalledProcessError( 0, ['true'], stdout='STDOUT', stderr='STDERR') with self.assertRaises(AttributeError): assert e.output is None assert e.stdout == 'STDOUT' assert e.stderr == 'STDERR' e.stdout = 'STDout' e.stderr = 'STDerr' with self.assertRaises(AttributeError): assert e.output is None assert e.stdout == 'STDout' assert e.stderr == 'STDerr' class RunCommandTests(unittest.TestCase): """Verify behavior of run helper.""" def test_basic(self): """Simple basic test.""" ret = rh.utils.run(['true']) self.assertEqual('true', ret.cmdstr) self.assertIsNone(ret.stdout) self.assertIsNone(ret.stderr) def test_stdout_capture(self): """Verify output capturing works.""" ret = rh.utils.run(['echo', 'hi'], redirect_stdout=True) self.assertEqual('hi\n', ret.stdout) self.assertIsNone(ret.stderr) def test_stderr_capture(self): """Verify stderr capturing works.""" ret = rh.utils.run(['sh', '-c', 'echo hi >&2'], redirect_stderr=True) self.assertIsNone(ret.stdout) self.assertEqual('hi\n', ret.stderr) def test_stdout_utf8(self): """Verify reading UTF-8 data works.""" ret = rh.utils.run(['printf', r'\xc3\x9f'], redirect_stdout=True) self.assertEqual('ß', ret.stdout) self.assertIsNone(ret.stderr) def test_stdin_utf8(self): """Verify writing UTF-8 data works.""" ret = rh.utils.run(['cat'], redirect_stdout=True, input='ß') self.assertEqual('ß', ret.stdout) self.assertIsNone(ret.stderr) def test_check_false(self): """Verify handling of check=False.""" ret = rh.utils.run(['false'], check=False) self.assertNotEqual(0, ret.returncode) self.assertIn('false', str(ret)) ret = rh.utils.run(['true'], check=False) self.assertEqual(0, ret.returncode) self.assertIn('true', str(ret)) def test_check_true(self): """Verify handling of check=True.""" with self.assertRaises(rh.utils.CalledProcessError) as e: rh.utils.run(['false'], check=True) err = e.exception self.assertNotEqual(0, err.returncode) self.assertIn('false', str(err)) ret = rh.utils.run(['true'], check=True) self.assertEqual(0, ret.returncode) self.assertIn('true', str(ret)) def test_check_false_output(self): """Verify handling of output capturing w/check=False.""" with self.assertRaises(rh.utils.CalledProcessError) as e: rh.utils.run(['sh', '-c', 'echo out; echo err >&2; false'], check=True, capture_output=True) err = e.exception self.assertNotEqual(0, err.returncode) self.assertIn('false', str(err)) def test_check_true_missing_prog_output(self): """Verify handling of output capturing w/missing progs.""" with self.assertRaises(rh.utils.CalledProcessError) as e: rh.utils.run(['./!~a/b/c/d/'], check=True, capture_output=True) err = e.exception self.assertNotEqual(0, err.returncode) self.assertIn('a/b/c/d', str(err)) def test_check_false_missing_prog_output(self): """Verify handling of output capturing w/missing progs.""" ret = rh.utils.run(['./!~a/b/c/d/'], check=False, capture_output=True) self.assertNotEqual(0, ret.returncode) self.assertIn('a/b/c/d', str(ret)) def test_check_false_missing_prog_combined_output(self): """Verify handling of combined output capturing w/missing progs.""" with self.assertRaises(rh.utils.CalledProcessError) as e: rh.utils.run(['./!~a/b/c/d/'], check=True, combine_stdout_stderr=True) err = e.exception self.assertNotEqual(0, err.returncode) self.assertIn('a/b/c/d', str(err)) def test_pathlib(self): """Verify pathlib arguments work.""" result = rh.utils.run(['true', Path('/')]) # Verify stringify behavior. str(result) self.assertEqual(result.cmdstr, 'true /') if __name__ == '__main__': unittest.main()