sysinfo_test.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471
  1. # Copyright (C) 2012 Internet Systems Consortium.
  2. #
  3. # Permission to use, copy, modify, and distribute this software for any
  4. # purpose with or without fee is hereby granted, provided that the above
  5. # copyright notice and this permission notice appear in all copies.
  6. #
  7. # THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM
  8. # DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
  9. # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
  10. # INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT,
  11. # INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
  12. # FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
  13. # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
  14. # WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  15. from isc.sysinfo import *
  16. import os
  17. import unittest
  18. import platform
  19. import subprocess
  20. import time
  21. # different fake 'number of processors' values used for the different
  22. # operating systems
  23. NPROCESSORS_LINUX = 42
  24. NPROCESSORS_OPENBSD = 43
  25. NPROCESSORS_FREEBSD = 44
  26. NPROCESSORS_OSX = 45
  27. def _my_testcase_platform_system():
  28. return 'BIND10Testcase'
  29. def _my_linux_platform_system():
  30. return 'Linux'
  31. def _my_linux_os_sysconf(key):
  32. if key == 'SC_NPROCESSORS_CONF':
  33. return NPROCESSORS_LINUX
  34. assert False, 'Unhandled key'
  35. class MyLinuxFile:
  36. def __init__(self, filename):
  37. self._filename = filename
  38. def read(self):
  39. if self._filename == '/proc/sys/kernel/hostname':
  40. return 'myhostname'
  41. elif self._filename == '/proc/version':
  42. return 'An SMP version string'
  43. elif self._filename == '/proc/uptime':
  44. return '172800.75 139993.71'
  45. elif self._filename == '/proc/loadavg':
  46. return '0.1 0.2 0.3 0.4'
  47. else:
  48. assert False, 'Unhandled filename'
  49. def readlines(self):
  50. if self._filename == '/proc/meminfo':
  51. return ['MemTotal: 3083872 kB',
  52. 'MemFree: 870492 kB',
  53. 'Buffers: 27412 kB',
  54. 'Cached: 1303860 kB',
  55. 'SwapTotal: 4194300 kB',
  56. 'SwapFree: 3999464 kB']
  57. else:
  58. assert False, 'Unhandled filename'
  59. def close(self):
  60. return
  61. def __enter__(self):
  62. return self
  63. def __exit__(self, type, value, traceback):
  64. return
  65. def _my_linux_open(filename):
  66. return MyLinuxFile(filename)
  67. def _my_linux_subprocess_check_output(command):
  68. assert type(command) == list, 'command argument is not a list'
  69. if command == ['lsb_release', '-a']:
  70. return b'Description: My Distribution\n'
  71. elif command == ['ip', 'addr']:
  72. return b'qB2osV6vUOjqm3P/+tQ4d92xoYz8/U8P9v3KWRpNwlI=\n'
  73. elif command == ['ip', 'route']:
  74. return b'VGWAS92AlS14Pl2xqENJs5P2Ihe6Nv9g181Mu6Zz+aQ=\n'
  75. elif command == ['ip', '-f', 'inet6', 'route']:
  76. return b'XfizswwNA9NkXz6K36ZExpjV08Y5IXkHI8jjDSV+5Nc=\n'
  77. elif command == ['netstat', '-s']:
  78. return b'osuxbrcc1g9VgaF4yf3FrtfodrfATrbSnjhqhuQSAs8=\n'
  79. elif command == ['netstat', '-apn']:
  80. return b'Z+w0lwa02/T+5+EIio84rrst/Dtizoz/aL9Im7J7ESA=\n'
  81. else:
  82. assert False, 'Unhandled command'
  83. def _my_openbsd_platform_system():
  84. return 'OpenBSD'
  85. def _my_openbsd_os_sysconf(key):
  86. if key == 'SC_NPROCESSORS_CONF':
  87. return NPROCESSORS_OPENBSD
  88. assert False, 'Unhandled key'
  89. def _my_openbsd_platform_uname():
  90. return ('OpenBSD', 'test.example.com', '5.0', '', 'amd64')
  91. # For the BSD types, there is a hierarchy that mostly resembles the
  92. # class hierarchy in the sysinfo library;
  93. # These are output strings of commands that sysinfo calls
  94. #
  95. # The test hierarchy is used as follows:
  96. # Each operating system has its own _my_<OS>_subprocess_check_output
  97. # call. If the call is not found, it calls it's 'parent' (e.g.
  98. # for openbsd that is my_bsd_subprocesses_check_output).
  99. #
  100. # If that returns None, the call had no test value and the test fails
  101. # (and needs to be updated).
  102. # The child classes are checked first so that they can override
  103. # output from the parents, if necessary.
  104. #
  105. # Some parents have their own parent
  106. # (e.g. _my_freebsd_osx_subprocess_check_output), in that case,
  107. # if they do not recognize the command, they simply return whatever
  108. # their parent returns
  109. def _my_bsd_subprocess_check_output(command):
  110. '''subprocess output for all bsd types'''
  111. assert type(command) == list, 'command argument is not a list'
  112. if command == ['hostname']:
  113. return b'test.example.com\n'
  114. elif command == ['sysctl', '-n', 'hw.physmem']:
  115. return b'543214321\n'
  116. elif command == ['ifconfig']:
  117. return b'qB2osV6vUOjqm3P/+tQ4d92xoYz8/U8P9v3KWRpNwlI=\n'
  118. elif command == ['netstat', '-s']:
  119. return b'osuxbrcc1g9VgaF4yf3FrtfodrfATrbSnjhqhuQSAs8=\n'
  120. elif command == ['netstat', '-an']:
  121. return b'Z+w0lwa02/T+5+EIio84rrst/Dtizoz/aL9Im7J7ESA=\n'
  122. elif command == ['netstat', '-nr']:
  123. return b'XfizswwNA9NkXz6K36ZExpjV08Y5IXkHI8jjDSV+5Nc=\n'
  124. else:
  125. return None
  126. def _my_openbsd_subprocess_check_output(command):
  127. assert type(command) == list, 'command argument is not a list'
  128. if command == ['sysctl', '-n', 'kern.boottime']:
  129. return bytes(str(int(time.time() - 76632)), 'utf-8')
  130. elif command == ['sysctl', '-n', 'vm.loadavg']:
  131. return b'0.7 0.9 0.8\n'
  132. elif command == ['vmstat']:
  133. return b' procs memory page disks traps cpu\n r b w avm fre flt re pi po fr sr wd0 cd0 int sys cs us sy id\n 0 0 0 121212 123456 47 0 0 0 0 0 2 0 2 80 14 0 1 99\n'
  134. elif command == ['swapctl', '-s', '-k']:
  135. return b'total: 553507 1K-blocks allocated, 2 used, 553505 available'
  136. else:
  137. bsd_output = _my_bsd_subprocess_check_output(command)
  138. if bsd_output is not None:
  139. return bsd_output
  140. else:
  141. assert False, 'Unhandled command'
  142. def _my_freebsd_platform_system():
  143. return 'FreeBSD'
  144. def _my_freebsd_os_sysconf(key):
  145. if key == 'SC_NPROCESSORS_CONF':
  146. return NPROCESSORS_FREEBSD
  147. assert False, 'Unhandled key'
  148. def _my_freebsd_platform_uname():
  149. return ('FreeBSD', 'freebsd', '8.2-RELEASE', '', 'i386')
  150. def _my_freebsd_osx_subprocess_check_output(command, faked_output={}):
  151. '''subprocess output shared for freebsd and osx'''
  152. assert type(command) == list, 'command argument is not a list'
  153. if command == ['sysctl', '-n', 'kern.boottime']:
  154. if 'boottime-sysctl' in faked_output:
  155. return faked_output['boottime-sysctl']
  156. return bytes('{ sec = ' + str(int(time.time() - 76632)) +
  157. ', usec = 0 }\n', 'utf-8')
  158. elif command == ['sysctl', '-n', 'vm.loadavg']:
  159. return b'{ 0.2 0.4 0.6 }\n'
  160. else:
  161. return _my_bsd_subprocess_check_output(command)
  162. def _my_freebsd_subprocess_check_output(command, faked_output):
  163. assert type(command) == list, 'command argument is not a list'
  164. if command == ['sysctl', '-n', 'kern.smp.forward_signal_enabled']:
  165. output = faked_output['smp-sysctl']
  166. if isinstance(output, Exception):
  167. raise output
  168. return faked_output['smp-sysctl']
  169. elif command == ['vmstat', '-H']:
  170. return b' procs memory page disks traps cpu\n r b w avm fre flt re pi po fr sr wd0 cd0 int sys cs us sy id\n 0 0 0 343434 123456 47 0 0 0 0 0 2 0 2 80 14 0 1 99\n'
  171. elif command == ['swapctl', '-s', '-k']:
  172. return b'Total: 1013216 0\n'
  173. else:
  174. freebsd_osx_output = \
  175. _my_freebsd_osx_subprocess_check_output(command, faked_output)
  176. if freebsd_osx_output is not None:
  177. return freebsd_osx_output
  178. else:
  179. assert False, 'Unhandled command'
  180. def _my_osx_platform_system():
  181. return 'Darwin'
  182. def _my_osx_platform_uname():
  183. return ('Darwin', 'test.example.com', '10.6.0', '', '')
  184. def _my_osx_os_sysconf(key):
  185. if key == 'SC_NPROCESSORS_CONF':
  186. return NPROCESSORS_OSX
  187. assert False, 'Unhandled key'
  188. def _my_osx_subprocess_check_output(command):
  189. assert type(command) == list, 'command argument is not a list'
  190. if command == ['sysctl', '-n', 'hw.memsize']:
  191. # Something different than physmem from bsd
  192. return b'123456789\n'
  193. elif command == ['vm_stat']:
  194. return b'Mach Virtual Memory Statistics: (page size of 4096 bytes)\nPages free: 12345.\nPages speculative: 11111.\n'
  195. elif command == ['sysctl', '-n', 'vm.swapusage']:
  196. return b'total = 18432.00M used = 17381.23M free = 1050.77M\n'
  197. else:
  198. freebsd_osx_output = _my_freebsd_osx_subprocess_check_output(command)
  199. if freebsd_osx_output is not None:
  200. return freebsd_osx_output
  201. else:
  202. assert False, 'Unhandled command'
  203. class SysInfoTest(unittest.TestCase):
  204. def setUp(self):
  205. # Save existing implementations of library functions
  206. # (they are replaced in the tests)
  207. self.old_platform_system = platform.system
  208. self.old_os_sysconf = os.sysconf
  209. self.old_open = __builtins__.open
  210. self.old_subprocess_check_output = subprocess.check_output
  211. def tearDown(self):
  212. # Restore the library functions
  213. platform.system = self.old_platform_system
  214. os.sysconf = self.old_os_sysconf
  215. __builtins__.open = self.old_open
  216. subprocess.check_output = self.old_subprocess_check_output
  217. def test_sysinfo(self):
  218. """Test that the various methods on SysInfo exist and return data."""
  219. s = SysInfo()
  220. self.assertEqual(None, s.get_num_processors())
  221. self.assertEqual('Unknown', s.get_endianness())
  222. self.assertEqual('', s.get_platform_hostname())
  223. self.assertEqual('Unknown', s.get_platform_name())
  224. self.assertEqual('Unknown', s.get_platform_version())
  225. self.assertEqual('Unknown', s.get_platform_machine())
  226. self.assertFalse(s.get_platform_is_smp())
  227. self.assertEqual(None, s.get_uptime())
  228. self.assertEqual(None, s.get_uptime_desc())
  229. self.assertEqual(None, s.get_loadavg())
  230. self.assertEqual(None, s.get_mem_total())
  231. self.assertEqual(None, s.get_mem_free())
  232. self.assertEqual(None, s.get_mem_cached())
  233. self.assertEqual(None, s.get_mem_buffers())
  234. self.assertEqual(None, s.get_mem_swap_total())
  235. self.assertEqual(None, s.get_mem_swap_free())
  236. self.assertEqual(None, s.get_platform_distro())
  237. self.assertEqual('Unknown\n', s.get_net_interfaces())
  238. self.assertEqual('Unknown\n', s.get_net_routing_table())
  239. self.assertEqual('Unknown\n', s.get_net_stats())
  240. self.assertEqual('Unknown\n', s.get_net_connections())
  241. def test_sysinfo_factory(self):
  242. """Test that SysInfoFromFactory returns a valid system-specific
  243. SysInfo implementation.
  244. See sysinfo.SysInfoTestcase() for some ofthe parameters.
  245. """
  246. old_platform_system = platform.system
  247. platform.system = _my_testcase_platform_system
  248. s = SysInfoFromFactory()
  249. self.assertEqual(None, s.get_num_processors())
  250. self.assertEqual('bigrastafarian', s.get_endianness())
  251. self.assertEqual('', s.get_platform_hostname())
  252. self.assertEqual('b10test', s.get_platform_name())
  253. self.assertEqual('Unknown', s.get_platform_version())
  254. self.assertEqual('Unknown', s.get_platform_machine())
  255. self.assertFalse(s.get_platform_is_smp())
  256. self.assertEqual(131072, s.get_uptime())
  257. # We check we do NOT add 's' to 'day' (because it's singular):
  258. self.assertEqual('1 day, 12:24', s.get_uptime_desc())
  259. self.assertEqual(None, s.get_loadavg())
  260. self.assertEqual(None, s.get_mem_total())
  261. self.assertEqual(None, s.get_mem_free())
  262. self.assertEqual(None, s.get_mem_cached())
  263. self.assertEqual(None, s.get_mem_buffers())
  264. self.assertEqual(None, s.get_mem_swap_total())
  265. self.assertEqual(None, s.get_mem_swap_free())
  266. self.assertEqual(None, s.get_platform_distro())
  267. self.assertEqual('Unknown\n', s.get_net_interfaces())
  268. self.assertEqual('Unknown\n', s.get_net_routing_table())
  269. self.assertEqual('Unknown\n', s.get_net_stats())
  270. self.assertEqual('Unknown\n', s.get_net_connections())
  271. platform.system = old_platform_system
  272. def test_sysinfo_linux(self):
  273. """Tests the Linux implementation of SysInfo. Note that this
  274. tests deep into the implementation, and not just the
  275. interfaces."""
  276. # Replace existing implementations of library functions
  277. # with mock ones for testing.
  278. platform.system = _my_linux_platform_system
  279. os.sysconf = _my_linux_os_sysconf
  280. __builtins__.open = _my_linux_open
  281. subprocess.check_output = _my_linux_subprocess_check_output
  282. s = SysInfoFromFactory()
  283. self.assertEqual(NPROCESSORS_LINUX, s.get_num_processors())
  284. self.assertEqual('myhostname', s.get_platform_hostname())
  285. self.assertTrue(s.get_platform_is_smp())
  286. self.assertEqual(172801, s.get_uptime())
  287. # We check we add 's' to 'day', and the mm part has an additional
  288. # 0, i.e., not '0:0' but '0:00':
  289. self.assertEqual('2 days, 0:00', s.get_uptime_desc())
  290. self.assertEqual((0.1, 0.2, 0.3), s.get_loadavg())
  291. self.assertEqual(3157884928, s.get_mem_total())
  292. self.assertEqual(891383808, s.get_mem_free())
  293. self.assertEqual(1335152640, s.get_mem_cached())
  294. self.assertEqual(28069888, s.get_mem_buffers())
  295. self.assertEqual(4294963200, s.get_mem_swap_total())
  296. self.assertEqual(4095451136, s.get_mem_swap_free())
  297. self.assertEqual('My Distribution', s.get_platform_distro())
  298. # These test that the corresponding tools are being called (and
  299. # no further processing is done on this data). Please see the
  300. # implementation functions at the top of this file.
  301. self.assertEqual('qB2osV6vUOjqm3P/+tQ4d92xoYz8/U8P9v3KWRpNwlI=\n', s.get_net_interfaces())
  302. self.assertEqual('VGWAS92AlS14Pl2xqENJs5P2Ihe6Nv9g181Mu6Zz+aQ=\n\nXfizswwNA9NkXz6K36ZExpjV08Y5IXkHI8jjDSV+5Nc=\n', s.get_net_routing_table())
  303. self.assertEqual('osuxbrcc1g9VgaF4yf3FrtfodrfATrbSnjhqhuQSAs8=\n', s.get_net_stats())
  304. self.assertEqual('Z+w0lwa02/T+5+EIio84rrst/Dtizoz/aL9Im7J7ESA=\n', s.get_net_connections())
  305. def check_bsd_values(self, s):
  306. # check values shared by all bsd implementations
  307. self.assertEqual('test.example.com', s.get_platform_hostname())
  308. self.assertLess(abs(76632 - s.get_uptime()), 4)
  309. self.assertEqual(None, s.get_mem_cached())
  310. self.assertEqual(None, s.get_mem_buffers())
  311. self.assertEqual(None, s.get_platform_distro())
  312. # These test that the corresponding tools are being called (and
  313. # no further processing is done on this data). Please see the
  314. # implementation functions at the top of this file.
  315. self.assertEqual('qB2osV6vUOjqm3P/+tQ4d92xoYz8/U8P9v3KWRpNwlI=\n', s.get_net_interfaces())
  316. self.assertEqual('XfizswwNA9NkXz6K36ZExpjV08Y5IXkHI8jjDSV+5Nc=\n', s.get_net_routing_table())
  317. self.assertEqual('osuxbrcc1g9VgaF4yf3FrtfodrfATrbSnjhqhuQSAs8=\n', s.get_net_stats())
  318. self.assertEqual('Z+w0lwa02/T+5+EIio84rrst/Dtizoz/aL9Im7J7ESA=\n', s.get_net_connections())
  319. def test_sysinfo_openbsd(self):
  320. """Tests the OpenBSD implementation of SysInfo. Note that this
  321. tests deep into the implementation, and not just the
  322. interfaces."""
  323. # Replace existing implementations of library functions
  324. # with mock ones for testing.
  325. platform.system = _my_openbsd_platform_system
  326. os.sysconf = _my_openbsd_os_sysconf
  327. subprocess.check_output = _my_openbsd_subprocess_check_output
  328. os.uname = _my_openbsd_platform_uname
  329. s = SysInfoFromFactory()
  330. self.assertEqual(NPROCESSORS_OPENBSD, s.get_num_processors())
  331. self.check_bsd_values(s)
  332. self.assertEqual((0.7, 0.9, 0.8), s.get_loadavg())
  333. self.assertFalse(s.get_platform_is_smp())
  334. self.assertEqual(543214321, s.get_mem_total())
  335. self.assertEqual(123456 * 1024, s.get_mem_free())
  336. self.assertEqual(566791168, s.get_mem_swap_total())
  337. self.assertEqual(566789120, s.get_mem_swap_free())
  338. def test_sysinfo_freebsd(self):
  339. """Tests the FreeBSD implementation of SysInfo. Note that this
  340. tests deep into the implementation, and not just the
  341. interfaces."""
  342. # Replace existing implementations of library functions
  343. # with mock ones for testing.
  344. platform.system = _my_freebsd_platform_system
  345. os.sysconf = _my_freebsd_os_sysconf
  346. # We use a lambda object so we can tweak the subprocess output during
  347. # the tests later.
  348. faked_process_output = { 'smp-sysctl': b'1\n' }
  349. subprocess.check_output = lambda command : \
  350. _my_freebsd_subprocess_check_output(command, faked_process_output)
  351. os.uname = _my_freebsd_platform_uname
  352. s = SysInfoFromFactory()
  353. self.assertEqual(NPROCESSORS_FREEBSD, s.get_num_processors())
  354. self.assertTrue(s.get_platform_is_smp())
  355. # We check the kernel SMP support by the availability of an sysctl
  356. # variable. The value (especiall a 0 value) shouldn't matter.
  357. faked_process_output['smp-sysctl'] = b'0\n'
  358. s = SysInfoFromFactory()
  359. self.assertTrue(s.get_platform_is_smp())
  360. # if the sysctl raises CalledProcessError, we treat it as non-SMP
  361. # kernel.
  362. faked_process_output['smp-sysctl'] = \
  363. subprocess.CalledProcessError(1, 'sysctl')
  364. s = SysInfoFromFactory()
  365. self.assertFalse(s.get_platform_is_smp())
  366. # if it results in OSError, no SMP information will be provided.
  367. faked_process_output['smp-sysctl'] = OSError()
  368. s = SysInfoFromFactory()
  369. self.assertIsNone(s.get_platform_is_smp())
  370. self.check_bsd_values(s)
  371. self.assertEqual((0.2, 0.4, 0.6), s.get_loadavg())
  372. self.assertEqual(543214321, s.get_mem_total())
  373. self.assertEqual(123456 * 1024, s.get_mem_free())
  374. self.assertEqual(1037533184, s.get_mem_swap_total())
  375. self.assertEqual(1037533184, s.get_mem_swap_free())
  376. # One more untested case so far: get_uptime_desc() should omit the
  377. # "days" field.
  378. faked_process_output['boottime-sysctl'] = bytes(
  379. '{ sec = ' + str(int(time.time() - 4200)) +
  380. ', usec = 0 }\n', 'utf-8')
  381. s = SysInfoFromFactory()
  382. self.assertEqual('1:10', s.get_uptime_desc())
  383. def test_sysinfo_osx(self):
  384. """Tests the OS X implementation of SysInfo. Note that this
  385. tests deep into the implementation, and not just the
  386. interfaces."""
  387. # Replace existing implementations of library functions
  388. # with mock ones for testing.
  389. platform.system = _my_osx_platform_system
  390. os.sysconf = _my_osx_os_sysconf
  391. subprocess.check_output = _my_osx_subprocess_check_output
  392. os.uname = _my_osx_platform_uname
  393. s = SysInfoFromFactory()
  394. self.assertEqual(NPROCESSORS_OSX, s.get_num_processors())
  395. self.assertFalse(s.get_platform_is_smp())
  396. self.check_bsd_values(s)
  397. self.assertEqual((0.2, 0.4, 0.6), s.get_loadavg())
  398. self.assertEqual(123456789, s.get_mem_total())
  399. self.assertEqual((23456 * 4096), s.get_mem_free())
  400. self.assertEqual(18874368.0, s.get_mem_swap_total())
  401. self.assertEqual(1075988.48, s.get_mem_swap_free())
  402. if __name__ == "__main__":
  403. unittest.main()