except (subprocess.CalledProcessError, OSError):
pass
+ try:
+ s = subprocess.check_output(['sysctl', '-n', 'hw.physmem'])
+ self._mem_total = int(s.decode('utf-8').strip())
+ except (subprocess.CalledProcessError, OSError):
+ pass
+
+ self._platform_distro = self._platform_name + ' ' + self._platform_version
+
+ try:
+ s = subprocess.check_output(['ifconfig'])
+ self._net_interfaces = s.decode('utf-8')
+ except (subprocess.CalledProcessError, OSError):
+ self._net_interfaces = 'Warning: "ifconfig" command failed.\n'
+
+ try:
+ s = subprocess.check_output(['netstat', '-s'])
+ self._net_stats = s.decode('utf-8')
+ except (subprocess.CalledProcessError, OSError):
+ self._net_stats = 'Warning: "netstat -s" command failed.\n'
+
+ try:
+ s = subprocess.check_output(['netstat', '-an'])
+ self._net_connections = s.decode('utf-8')
+ except (subprocess.CalledProcessError, OSError):
+ self._net_connections = 'Warning: "netstat -an" command failed.\n'
+
+class SysInfoOpenBSD(SysInfoBSD):
+ """OpenBSD implementation of the SysInfo class.
+ See the SysInfo class documentation for more information.
+ """
+ def __init__(self):
+ super().__init__()
+
+ # Don't know how to gather these
+ self._platform_is_smp = False
+ self._mem_cached = -1
+ self._mem_buffers = -1
+
try:
s = subprocess.check_output(['sysctl', '-n', 'kern.boottime'])
t = s.decode('utf-8').strip()
except (subprocess.CalledProcessError, OSError):
pass
- try:
- s = subprocess.check_output(['sysctl', '-n', 'hw.physmem'])
- self._mem_total = int(s.decode('utf-8').strip())
- except (subprocess.CalledProcessError, OSError):
- pass
-
try:
s = subprocess.check_output(['vmstat'])
lines = s.decode('utf-8').split('\n')
except (subprocess.CalledProcessError, OSError):
pass
- self._platform_distro = self._platform_name + ' ' + self._platform_version
+ try:
+ s = subprocess.check_output(['swapctl', '-s', '-k'])
+ l = s.decode('utf-8').strip()
+ r = re.match('^total: (\d+) 1K-blocks allocated, (\d+) used, (\d+) available', l)
+ if r:
+ self._mem_swap_total = int(r.group(1).strip()) * 1024
+ self._mem_swap_free = int(r.group(3).strip()) * 1024
+ except (subprocess.CalledProcessError, OSError):
+ pass
-class SysInfoOpenBSD(SysInfoBSD):
- """OpenBSD implementation of the SysInfo class.
+ try:
+ s = subprocess.check_output(['route', '-n', 'show'])
+ self._net_routing_table = s.decode('utf-8')
+ except (subprocess.CalledProcessError, OSError):
+ self._net_routing_table = 'Warning: "route -n show" command failed.\n'
+
+class SysInfoFreeBSD(SysInfoBSD):
+ """FreeBSD implementation of the SysInfo class.
See the SysInfo class documentation for more information.
"""
def __init__(self):
super().__init__()
# Don't know how to gather these
- self._platform_is_smp = False
self._mem_cached = -1
self._mem_buffers = -1
try:
- s = subprocess.check_output(['swapctl', '-s', '-k'])
- l = s.decode('utf-8').strip()
- r = re.match('^total: (\d+) 1K-blocks allocated, (\d+) used, (\d+) available', l)
+ s = subprocess.check_output(['sysctl', '-n', 'kern.smp.active'])
+ self._platform_is_smp = int(s.decode('utf-8').strip()) > 0
+ except (subprocess.CalledProcessError, OSError):
+ pass
+
+ try:
+ s = subprocess.check_output(['sysctl', '-n', 'kern.boottime'])
+ t = s.decode('utf-8').strip()
+ r = re.match('^\{\s+sec\s+\=\s+(\d+),.*', t)
if r:
- self._mem_swap_total = int(r.group(1).strip()) * 1024
- self._mem_swap_free = int(r.group(3).strip()) * 1024
+ sec = time.time() - int(r.group(1))
+ self._uptime = int(round(sec))
except (subprocess.CalledProcessError, OSError):
pass
try:
- s = subprocess.check_output(['ifconfig'])
- self._net_interfaces = s.decode('utf-8')
+ s = subprocess.check_output(['sysctl', '-n', 'vm.loadavg'])
+ l = s.decode('utf-8').strip()
+ r = re.match('^\{(.*)\}$', l)
+ if r:
+ la = r.group(1).strip().split(' ')
+ else:
+ la = l.split(' ')
+ if len(la) >= 3:
+ self._loadavg = [float(la[0]), float(la[1]), float(la[2])]
except (subprocess.CalledProcessError, OSError):
- self._net_interfaces = 'Warning: "ifconfig" command failed.\n'
+ pass
try:
- s = subprocess.check_output(['route', '-n', 'show'])
- self._net_routing_table = s.decode('utf-8')
+ s = subprocess.check_output(['vmstat', '-H'])
+ lines = s.decode('utf-8').split('\n')
+ v = re.split('\s+', lines[2])
+ used = int(v[4]) * 1024
+ self._mem_free = self._mem_total - used
except (subprocess.CalledProcessError, OSError):
- self._net_routing_table = 'Warning: "route -n show" command failed.\n'
+ pass
try:
- s = subprocess.check_output(['netstat', '-s'])
- self._net_stats = s.decode('utf-8')
+ s = subprocess.check_output(['swapctl', '-s', '-k'])
+ l = s.decode('utf-8').strip()
+ r = re.match('^Total:\s+(\d+)\s+(\d+)', l)
+ if r:
+ self._mem_swap_total = int(r.group(1).strip()) * 1024
+ self._mem_swap_free = self._mem_swap_total - (int(r.group(2).strip()) * 1024)
except (subprocess.CalledProcessError, OSError):
- self._net_stats = 'Warning: "netstat -s" command failed.\n'
+ pass
try:
- s = subprocess.check_output(['netstat', '-an'])
- self._net_connections = s.decode('utf-8')
+ s = subprocess.check_output(['netstat', '-nr'])
+ self._net_routing_table = s.decode('utf-8')
except (subprocess.CalledProcessError, OSError):
- self._net_connections = 'Warning: "netstat -an" command failed.\n'
+ self._net_connections = 'Warning: "netstat -nr" command failed.\n'
class SysInfoTestcase(SysInfo):
def __init__(self):
return SysInfoLinux()
elif osname == 'OpenBSD':
return SysInfoOpenBSD()
+ elif osname == 'FreeBSD':
+ return SysInfoFreeBSD()
elif osname == 'BIND10Testcase':
return SysInfoTestcase()
else:
elif command == ['sysctl', '-n', 'kern.boottime']:
return bytes(str(int(time.time() - 76632)), 'utf-8')
elif command == ['sysctl', '-n', 'vm.loadavg']:
- return b'0.7 0.9 0.8'
+ return b'0.7 0.9 0.8\n'
elif command == ['sysctl', '-n', 'hw.physmem']:
- return b'543214321'
+ return b'543214321\n'
elif command == ['vmstat']:
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'
elif command == ['swapctl', '-s', '-k']:
else:
assert False, 'Unhandled command'
+def _my_freebsd_platform_system():
+ return 'FreeBSD'
+
+def _my_freebsd_os_sysconf(key):
+ if key == 'SC_NPROCESSORS_CONF':
+ return 91
+ assert False, 'Unhandled key'
+
+def _my_freebsd_subprocess_check_output(command):
+ assert type(command) == list, 'command argument is not a list'
+ if command == ['hostname']:
+ return b'daemon.example.com\n'
+ elif command == ['sysctl', '-n', 'kern.smp.active']:
+ return b'1\n'
+ elif command == ['sysctl', '-n', 'kern.boottime']:
+ return bytes('{ sec = ' + str(int(time.time() - 76632)) + ', usec = 0 }\n', 'utf-8')
+ elif command == ['sysctl', '-n', 'vm.loadavg']:
+ return b'{ 0.2 0.4 0.6 }\n'
+ elif command == ['sysctl', '-n', 'hw.physmem']:
+ return b'987654321\n'
+ elif command == ['vmstat', '-H']:
+ 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'
+ elif command == ['swapctl', '-s', '-k']:
+ return b'Total: 1013216 0\n'
+ elif command == ['ifconfig']:
+ return b'qB2osV6vUOjqm3P/+tQ4d92xoYz8/U8P9v3KWRpNwlI=\n'
+ elif command == ['netstat', '-nr']:
+ return b'XfizswwNA9NkXz6K36ZExpjV08Y5IXkHI8jjDSV+5Nc=\n'
+ elif command == ['netstat', '-s']:
+ return b'osuxbrcc1g9VgaF4yf3FrtfodrfATrbSnjhqhuQSAs8=\n'
+ elif command == ['netstat', '-an']:
+ return b'Z+w0lwa02/T+5+EIio84rrst/Dtizoz/aL9Im7J7ESA=\n'
+ else:
+ assert False, 'Unhandled command'
+
class SysInfoTest(unittest.TestCase):
def test_sysinfo(self):
"""Test that the various methods on SysInfo exist and return data."""
tests deep into the implementation, and not just the
interfaces."""
- # Don't run this test on platform other than Openbsd as some
+ # Don't run this test on platform other than OpenBSD as some
# system calls may not even be available.
osname = platform.system()
if osname != 'OpenBSD':
os.sysconf = old_os_sysconf
subprocess.check_output = old_subprocess_check_output
+ def test_sysinfo_freebsd(self):
+ """Tests the FreeBSD implementation of SysInfo. Note that this
+ tests deep into the implementation, and not just the
+ interfaces."""
+
+ # Don't run this test on platform other than FreeBSD as some
+ # system calls may not even be available.
+ osname = platform.system()
+ if osname != 'FreeBSD':
+ return
+
+ # Save and replace existing implementations of library functions
+ # with mock ones for testing.
+ old_platform_system = platform.system
+ platform.system = _my_freebsd_platform_system
+ old_os_sysconf = os.sysconf
+ os.sysconf = _my_freebsd_os_sysconf
+ old_subprocess_check_output = subprocess.check_output
+ subprocess.check_output = _my_freebsd_subprocess_check_output
+
+ s = SysInfoFromFactory()
+ self.assertEqual(91, s.get_num_processors())
+ self.assertEqual('daemon.example.com', s.get_platform_hostname())
+ self.assertTrue(s.get_platform_is_smp())
+
+ self.assertLess(abs(76632 - s.get_uptime()), 4)
+ self.assertEqual([0.2, 0.4, 0.6], s.get_loadavg())
+ self.assertEqual(987654321, s.get_mem_total())
+ self.assertEqual(987654321 - (343434 * 1024), s.get_mem_free())
+ self.assertEqual(-1, s.get_mem_cached())
+ self.assertEqual(-1, s.get_mem_buffers())
+ self.assertEqual(1037533184, s.get_mem_swap_total())
+ self.assertEqual(1037533184, s.get_mem_swap_free())
+ self.assertRegexpMatches(s.get_platform_distro(), '^FreeBSD\s+.*')
+
+ # These test that the corresponding tools are being called (and
+ # no further processing is done on this data). Please see the
+ # implementation functions at the top of this file.
+ self.assertEqual('qB2osV6vUOjqm3P/+tQ4d92xoYz8/U8P9v3KWRpNwlI=\n', s.get_net_interfaces())
+ self.assertEqual('XfizswwNA9NkXz6K36ZExpjV08Y5IXkHI8jjDSV+5Nc=\n', s.get_net_routing_table())
+ self.assertEqual('osuxbrcc1g9VgaF4yf3FrtfodrfATrbSnjhqhuQSAs8=\n', s.get_net_stats())
+ self.assertEqual('Z+w0lwa02/T+5+EIio84rrst/Dtizoz/aL9Im7J7ESA=\n', s.get_net_connections())
+
+ # Restore original implementations.
+ platform.system = old_platform_system
+ os.sysconf = old_os_sysconf
+ subprocess.check_output = old_subprocess_check_output
+
if __name__ == "__main__":
unittest.main()
projects / thirdparty / kea.git / commitdiff
