OLD | NEW |
(Empty) | |
| 1 # Copyright 2014 Dirk Pranke. All rights reserved. |
| 2 # |
| 3 # Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 # you may not use this file except in compliance with the License. |
| 5 # You may obtain a copy of the License at |
| 6 # |
| 7 # http://www.apache.org/licenses/LICENSE-2.0 |
| 8 # |
| 9 # Unless required by applicable law or agreed to in writing, software |
| 10 # distributed under the License is distributed on an "AS IS" BASIS, |
| 11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 # See the License for the specific language governing permissions and |
| 13 # limitations under the License. |
| 14 |
| 15 import logging |
| 16 import pickle |
| 17 import sys |
| 18 import unittest |
| 19 |
| 20 from typ.host import Host |
| 21 |
| 22 |
| 23 class TestHost(unittest.TestCase): |
| 24 |
| 25 def host(self): |
| 26 return Host() |
| 27 |
| 28 def test_capture_output(self): |
| 29 try: |
| 30 logging.basicConfig() |
| 31 h = self.host() |
| 32 h.capture_output() |
| 33 h.print_('on stdout') |
| 34 h.print_('on stderr', stream=h.stderr) |
| 35 logging.critical('critical log failure') |
| 36 out, err = h.restore_output() |
| 37 self.assertEqual(out, 'on stdout\n') |
| 38 self.assertEqual(err, 'on stderr\ncritical log failure\n') |
| 39 finally: |
| 40 h.logger.handlers = [] |
| 41 |
| 42 # TODO: Add tests for divert=False or eliminate the flag? |
| 43 |
| 44 def test_abspath_and_realpath(self): |
| 45 h = self.host() |
| 46 self.assertNotEqual(h.abspath(h.getcwd()), None) |
| 47 self.assertNotEqual(h.realpath(h.getcwd()), None) |
| 48 |
| 49 def test_chdir(self): |
| 50 h = self.host() |
| 51 orig_cwd = h.getcwd() |
| 52 h.chdir('.') |
| 53 self.assertEqual(orig_cwd, h.getcwd()) |
| 54 h.chdir('..') |
| 55 self.assertNotEqual(orig_cwd, h.getcwd()) |
| 56 |
| 57 def test_files(self): |
| 58 h = self.host() |
| 59 orig_cwd = h.getcwd() |
| 60 try: |
| 61 now = h.time() |
| 62 |
| 63 # TODO: MacOS does goofy things with temp dirs by default, so |
| 64 # we can't compare for equality. Figure out how to get the normpath |
| 65 # from mkdtemp |
| 66 dirpath = h.mkdtemp(suffix='host_test') |
| 67 self.assertTrue(h.isdir(dirpath)) |
| 68 h.chdir(dirpath) |
| 69 self.assertIn(dirpath, h.getcwd()) |
| 70 |
| 71 h.maybe_mkdir('bar') |
| 72 self.assertTrue(h.exists(dirpath, 'bar')) |
| 73 self.assertTrue(h.isdir(dirpath, 'bar')) |
| 74 self.assertFalse(h.isfile(dirpath, 'bar')) |
| 75 |
| 76 bar_path = h.join(dirpath, 'bar') |
| 77 self.assertEqual(dirpath, h.dirname(bar_path)) |
| 78 |
| 79 h.write_text_file('bar/foo.txt', 'foo') |
| 80 self.assertTrue(h.exists('bar', 'foo.txt')) |
| 81 self.assertEqual(h.read_text_file('bar/foo.txt'), 'foo') |
| 82 self.assertTrue(h.exists(dirpath, 'bar', 'foo.txt')) |
| 83 self.assertTrue(h.isfile(dirpath, 'bar', 'foo.txt')) |
| 84 self.assertFalse(h.isdir(dirpath, 'bar', 'foo.txt')) |
| 85 |
| 86 h.write_binary_file('binfile', b'bin contents') |
| 87 self.assertEqual(h.read_binary_file('binfile'), |
| 88 b'bin contents') |
| 89 |
| 90 self.assertEqual(sorted(h.files_under(dirpath)), |
| 91 ['bar' + h.sep + 'foo.txt', 'binfile']) |
| 92 |
| 93 mtime = h.mtime(dirpath, 'bar', 'foo.txt') |
| 94 self.assertGreaterEqual(now, mtime - 0.1) |
| 95 h.remove(dirpath, 'bar', 'foo.txt') |
| 96 self.assertFalse(h.exists(dirpath, 'bar', 'foo.txt')) |
| 97 self.assertFalse(h.isfile(dirpath, 'bar', 'foo.txt')) |
| 98 |
| 99 h.chdir(orig_cwd) |
| 100 h.rmtree(dirpath) |
| 101 self.assertFalse(h.exists(dirpath)) |
| 102 self.assertFalse(h.isdir(dirpath)) |
| 103 finally: |
| 104 h.chdir(orig_cwd) |
| 105 |
| 106 def test_terminal_width(self): |
| 107 h = self.host() |
| 108 self.assertGreaterEqual(h.terminal_width(), 0) |
| 109 |
| 110 def test_for_mp_and_pickling(self): |
| 111 h = self.host() |
| 112 mp_host = h.for_mp() |
| 113 s = pickle.dumps(mp_host) |
| 114 pickle.loads(s) |
| 115 |
| 116 def test_cpu_count(self): |
| 117 h = self.host() |
| 118 self.assertGreaterEqual(h.cpu_count(), 1) |
| 119 |
| 120 def test_getenv(self): |
| 121 h = self.host() |
| 122 self.assertNotEqual(h.getenv('PATH', ''), None) |
| 123 |
| 124 def test_getpid(self): |
| 125 h = self.host() |
| 126 self.assertNotEqual(h.getpid(), 0) |
| 127 |
| 128 def test_basename(self): |
| 129 h = self.host() |
| 130 self.assertEqual(h.basename('foo.txt'), 'foo.txt') |
| 131 self.assertEqual(h.basename('foo/bar.txt'), 'bar.txt') |
| 132 |
| 133 def test_mktempfile(self, delete=False): |
| 134 h = self.host() |
| 135 f= h.mktempfile() |
| 136 f.close() |
| 137 self.assertNotEqual(f.name, None) |
| 138 |
| 139 def test_splitext(self): |
| 140 h = self.host() |
| 141 self.assertEqual(h.splitext('foo'), ('foo', '')) |
| 142 self.assertEqual(h.splitext('foo.txt'), ('foo', '.txt')) |
| 143 self.assertEqual(h.splitext('foo/bar'), ('foo/bar', '')) |
| 144 self.assertEqual(h.splitext('foo/bar.txt'), ('foo/bar', '.txt')) |
| 145 |
| 146 def test_print(self): |
| 147 h = self.host() |
| 148 |
| 149 class FakeStream(object): |
| 150 |
| 151 def __init__(self): |
| 152 self.contents = None |
| 153 self.flush_called = False |
| 154 |
| 155 def write(self, m): |
| 156 self.contents = m |
| 157 |
| 158 def flush(self): |
| 159 self.flush_called = True |
| 160 |
| 161 s = FakeStream() |
| 162 h.print_('hello', stream=s) |
| 163 self.assertEqual(s.contents, 'hello\n') |
| 164 self.assertTrue(s.flush_called) |
| 165 |
| 166 s = FakeStream() |
| 167 h.stdout = s |
| 168 h.print_('hello') |
| 169 self.assertEqual(s.contents, 'hello\n') |
| 170 |
| 171 s = FakeStream() |
| 172 h.stdout = s |
| 173 h.print_('hello', '') |
| 174 self.assertEqual(s.contents, 'hello') |
| 175 |
| 176 def test_call(self): |
| 177 h = self.host() |
| 178 ret, out, err = h.call( |
| 179 [h.python_interpreter, |
| 180 '-c', 'import sys; sys.stdout.write(sys.stdin.read())'], |
| 181 stdin='foo', env={}) |
| 182 self.assertEqual(ret, 0) |
| 183 self.assertEqual(out, 'foo') |
| 184 self.assertEqual(err, '') |
| 185 |
| 186 ret, out, err = h.call( |
| 187 [h.python_interpreter, |
| 188 '-c', 'import sys; sys.stderr.write("err\\n")']) |
| 189 self.assertEqual(ret, 0) |
| 190 self.assertEqual(out, '') |
| 191 self.assertIn(err, ('err\n', 'err\r\n')) |
| 192 |
| 193 def test_call_inline(self): |
| 194 h = self.host() |
| 195 h.stdout = None |
| 196 h.stderr = None |
| 197 ret = h.call_inline([h.python_interpreter, |
| 198 '-c', 'import sys; sys.exit(0)']) |
| 199 self.assertEqual(ret, 0) |
| 200 |
| 201 def test_add_to_path(self): |
| 202 orig_sys_path = sys.path[:] |
| 203 try: |
| 204 h = self.host() |
| 205 h.add_to_path(sys.path[-1]) |
| 206 self.assertEqual(sys.path, orig_sys_path) |
| 207 |
| 208 dirpath = h.mkdtemp() |
| 209 h.add_to_path(dirpath) |
| 210 self.assertNotEqual(sys.path, orig_sys_path) |
| 211 finally: |
| 212 sys.path = orig_sys_path |
| 213 |
| 214 def test_platform(self): |
| 215 h = self.host() |
| 216 self.assertNotEqual(h.platform, None) |
OLD | NEW |