mirror of
https://github.com/Telecominfraproject/OpenNetworkLinux.git
synced 2025-12-25 17:27:01 +00:00
Merge pull request #255 from carlroth/roth_onlp
Use Python weakref to streamline C memory management
This commit is contained in:
Jeffrey Townsend
@@ -13,15 +13,17 @@ libc = ctypes.cdll.LoadLibrary(ctypes.util.find_library("c"))
|
||||
|
||||
import onlp.onlplib
|
||||
import onlp.sff
|
||||
from onlp.onlp import aim_weakref
|
||||
|
||||
from onlp.onlp.enums import *
|
||||
|
||||
# AIM/aim_memory.h
|
||||
|
||||
class aim_void_p(ctypes.c_void_p):
|
||||
"""Generic data allocated by AIM."""
|
||||
def __del__(self):
|
||||
libonlp.aim_free(self)
|
||||
class aim_void_p(aim_weakref.AimPointer):
|
||||
|
||||
@classmethod
|
||||
def deletePointer(cls, aimPtr):
|
||||
libonlp.aim_free(aimPtr)
|
||||
|
||||
class aim_char_p(aim_void_p):
|
||||
"""AIM data that is a printable string."""
|
||||
@@ -155,13 +157,18 @@ class aim_bitmap_hdr(ctypes.Structure):
|
||||
class aim_bitmap(ctypes.Structure):
|
||||
_fields_ = [("hdr", aim_bitmap_hdr,),]
|
||||
|
||||
@classmethod
|
||||
def fromAim(cls, bitcount):
|
||||
"""Return a pointer to a bitmap from aim_alloc().
|
||||
class aim_bitmap_ref(aim_weakref.AimReference):
|
||||
"""Dynamically allocated aim_bitmap."""
|
||||
|
||||
Pre-initialized; needs to be freed with aim_free().
|
||||
"""
|
||||
return libonlp.aim_bitmap_alloc(None, bitcount)
|
||||
_fields_ = aim_bitmap._fields_
|
||||
|
||||
def __init__(self, bitcount):
|
||||
ptr = libonlp.aim_bitmap_alloc(None, bitcount)
|
||||
super(aim_bitmap_ref, self).__init__(ptr)
|
||||
|
||||
@classmethod
|
||||
def deleteReference(self, aimPtr):
|
||||
libonlp.aim_free(aimPtr)
|
||||
|
||||
class aim_bitmap256(aim_bitmap):
|
||||
"""Statically-allocated AIM bitmap."""
|
||||
@@ -299,16 +306,19 @@ def onlp_oid_init_prototypes():
|
||||
|
||||
# onlp/sys.h
|
||||
|
||||
class onlp_sys_info(ctypes.Structure):
|
||||
|
||||
initialized = False
|
||||
class onlp_sys_info(aim_weakref.AimStruct):
|
||||
|
||||
_fields_ = [("hdr", onlp_oid_hdr,),
|
||||
("onie_info", onlp.onlplib.onlp_onie_info,),
|
||||
("platform_info", onlp.onlplib.onlp_platform_info,),]
|
||||
|
||||
def __del__(self):
|
||||
if self.initialized:
|
||||
def __init__(self):
|
||||
self.initialized = False
|
||||
super(onlp_sys_info, self).__init__()
|
||||
|
||||
def deleteStruct(self):
|
||||
initialized, self.initialized = self.initialized, False
|
||||
if initialized:
|
||||
libonlp.onlp_sys_info_free(ctypes.byref(self))
|
||||
|
||||
def onlp_sys_init_prototypes():
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
"""aim_weakref.py
|
||||
|
||||
Use weakref to implement smart AIM pointers.
|
||||
|
||||
See e.g.
|
||||
http://code.activestate.com/recipes/577242-calling-c-level-finalizers-without-__del__/
|
||||
"""
|
||||
|
||||
import ctypes
|
||||
import logging
|
||||
import weakref
|
||||
|
||||
logger = logging.getLogger("weakref")
|
||||
|
||||
def getLogger():
|
||||
global logger
|
||||
return logger
|
||||
|
||||
class AimOwnerRef(weakref.ref):
|
||||
pass
|
||||
|
||||
def _run_finalizer(ref):
|
||||
"""Internal weakref callback to run finalizers"""
|
||||
del _finalize_refs[id(ref)]
|
||||
finalizer = ref.finalizer
|
||||
item = ref.item
|
||||
try:
|
||||
finalizer(item)
|
||||
except Exception:
|
||||
getLogger().exception("finalizer failed")
|
||||
|
||||
_finalize_refs = {}
|
||||
|
||||
def track_for_finalization(owner, item, finalizer):
|
||||
"""Register an object for finalization.
|
||||
|
||||
``owner`` is the the object which is responsible for ``item``.
|
||||
``finalizer`` will be called with ``item`` as its only argument when
|
||||
``owner`` is destroyed by the garbage collector.
|
||||
"""
|
||||
ref = AimOwnerRef(owner, _run_finalizer)
|
||||
ref.item = item
|
||||
ref.finalizer = finalizer
|
||||
_finalize_refs[id(ref)] = ref
|
||||
|
||||
class AimReference(object):
|
||||
"""Manage an AIM pointer using reference semantics."""
|
||||
|
||||
@classmethod
|
||||
def deleteReference(cls, aimPtr):
|
||||
"""Override this with the proper delete semantics."""
|
||||
raise NotImplementedError
|
||||
|
||||
def __init__(self, aimPtr):
|
||||
self.ptr = aimPtr
|
||||
track_for_finalization(self, self.ptr, self.deleteReference)
|
||||
|
||||
def __getattr__(self, attr, dfl='__none__'):
|
||||
if dfl == '__none__':
|
||||
return getattr(self.ptr.contents, attr)
|
||||
else:
|
||||
return getattr(self.ptr.contents, attr, dfl)
|
||||
|
||||
def __setattr___(self, attr, val):
|
||||
setattr(self.ptr.contents, attr, val)
|
||||
|
||||
class AimPointer(ctypes.c_void_p):
|
||||
"""Manage an AIM pointer using pointer semantics."""
|
||||
|
||||
@classmethod
|
||||
def deletePointer(cls, aimPtr):
|
||||
"""Override this with the proper delete semantics."""
|
||||
raise NotImplementedError
|
||||
|
||||
def __init__(self, aimPtr):
|
||||
|
||||
super(ctypes.c_void_p, self).__init__(aimPtr)
|
||||
# XXX roth -- casting may be necessary
|
||||
|
||||
track_for_finalization(self, aimPtr, self.deletePointer)
|
||||
|
||||
class AimStruct(ctypes.Structure):
|
||||
"""Manage an AIM struct with internally-allocated data."""
|
||||
|
||||
@classmethod
|
||||
def deleteStruct(cls, aimPtr):
|
||||
"""Override this with the proper delete semantics."""
|
||||
raise NotImplementedError
|
||||
|
||||
def __init__(self):
|
||||
|
||||
super(ctypes.Structure, self).__init__()
|
||||
|
||||
track_for_finalization(self, self, self.deleteStruct)
|
||||
@@ -18,6 +18,8 @@ import onlp.sff
|
||||
|
||||
libonlp = onlp.onlp.libonlp
|
||||
|
||||
import onlp.onlp.aim_weakref
|
||||
|
||||
def isVirtual():
|
||||
with open("/etc/onl/platform") as fd:
|
||||
buf = fd.read()
|
||||
@@ -26,6 +28,20 @@ def isVirtual():
|
||||
def skipIfVirtual(reason="this test only works with a physical device"):
|
||||
return unittest.skipIf(isVirtual(), reason)
|
||||
|
||||
class aim_pvs_buffer(onlp.onlp.aim_weakref.AimReference):
|
||||
|
||||
def __init__(self):
|
||||
ptr = libonlp.aim_pvs_buffer_create()
|
||||
super(aim_pvs_buffer, self).__init__(ptr)
|
||||
|
||||
@classmethod
|
||||
def deleteReference(self, ptr):
|
||||
libonlp.aim_pvs_destroy(ptr)
|
||||
|
||||
def string_at(self):
|
||||
buf = libonlp.aim_pvs_buffer_get(self.ptr)
|
||||
return buf.string_at()
|
||||
|
||||
class OnlpTestMixin(object):
|
||||
|
||||
def setUp(self):
|
||||
@@ -33,12 +49,12 @@ class OnlpTestMixin(object):
|
||||
self.log = logging.getLogger("onlp")
|
||||
self.log.setLevel(logging.DEBUG)
|
||||
|
||||
self.aim_pvs_buffer_p = libonlp.aim_pvs_buffer_create()
|
||||
self.aim_pvs_buffer = self.aim_pvs_buffer_p.contents
|
||||
self.aim_pvs_buffer = aim_pvs_buffer()
|
||||
|
||||
onlp.onlp.aim_weakref.logger = self.log.getChild("weakref")
|
||||
|
||||
def tearDown(self):
|
||||
|
||||
libonlp.aim_pvs_destroy(self.aim_pvs_buffer_p)
|
||||
pass
|
||||
|
||||
def assertStatusOK(self, sts):
|
||||
if sts == onlp.onlp.ONLP_STATUS.OK:
|
||||
@@ -61,25 +77,19 @@ class InitTest(OnlpTestMixin,
|
||||
def testBuffer(self):
|
||||
"""Verify that the AIM buffer type is usable."""
|
||||
|
||||
nullString = onlp.onlp.aim_char_p(None)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
self.assertEqual(nullString, buf)
|
||||
self.assertIsNone(self.aim_pvs_buffer.string_at())
|
||||
|
||||
libonlp.aim_printf(self.aim_pvs_buffer_p, "hello\n")
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
self.assertEqual("hello\n", buf.string_at())
|
||||
libonlp.aim_printf(self.aim_pvs_buffer.ptr, "hello\n")
|
||||
self.assertEqual("hello\n", self.aim_pvs_buffer.string_at())
|
||||
|
||||
libonlp.aim_printf(self.aim_pvs_buffer_p, "world\n")
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
self.assertEqual("hello\nworld\n", buf.string_at())
|
||||
libonlp.aim_printf(self.aim_pvs_buffer.ptr, "world\n")
|
||||
self.assertEqual("hello\nworld\n", self.aim_pvs_buffer.string_at())
|
||||
|
||||
libonlp.aim_printf(self.aim_pvs_buffer_p, "%d\n", 42)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
self.assertEqual("hello\nworld\n42\n", buf.string_at())
|
||||
libonlp.aim_printf(self.aim_pvs_buffer.ptr, "%d\n", 42)
|
||||
self.assertEqual("hello\nworld\n42\n", self.aim_pvs_buffer.string_at())
|
||||
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer_p)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
self.assertEqual(nullString, buf)
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer.ptr)
|
||||
self.assertIsNone(self.aim_pvs_buffer.string_at())
|
||||
|
||||
class OnlpTest(OnlpTestMixin,
|
||||
unittest.TestCase):
|
||||
@@ -95,8 +105,8 @@ class OnlpTest(OnlpTestMixin,
|
||||
"""Verify basic platform dump output."""
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_platform_dump(self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_platform_dump(self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("System Information:", bufStr)
|
||||
self.assertIn("thermal @ 1", bufStr)
|
||||
@@ -105,17 +115,17 @@ class OnlpTest(OnlpTestMixin,
|
||||
"""Verify platform dump flags are honored."""
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_platform_dump(self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_platform_dump(self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("psu @ 1", bufStr)
|
||||
self.assertNotIn("PSU-1 Fan", bufStr)
|
||||
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer_p)
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer.ptr)
|
||||
|
||||
flags = onlp.onlp.ONLP_OID_DUMP.RECURSE
|
||||
libonlp.onlp_platform_dump(self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_platform_dump(self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("psu @ 1", bufStr)
|
||||
self.assertIn("PSU-1 Fan", bufStr)
|
||||
@@ -127,8 +137,8 @@ class OnlpTest(OnlpTestMixin,
|
||||
"""Verify basic platform show output."""
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_platform_show(self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_platform_show(self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("System Information:", bufStr)
|
||||
|
||||
@@ -136,17 +146,17 @@ class OnlpTest(OnlpTestMixin,
|
||||
"""Verify that onlp_platform_show honors flags."""
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_platform_show(self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_platform_show(self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertNotIn("PSU 1", bufStr)
|
||||
self.assertNotIn("PSU-1 Fan", bufStr)
|
||||
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer_p)
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer.ptr)
|
||||
|
||||
flags = onlp.onlp.ONLP_OID_SHOW.RECURSE
|
||||
libonlp.onlp_platform_show(self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_platform_show(self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("PSU 1", bufStr)
|
||||
self.assertIn("PSU-1 Fan", bufStr)
|
||||
@@ -288,18 +298,18 @@ class SysTest(OnlpTestMixin,
|
||||
|
||||
oid = onlp.onlp.ONLP_OID_SYS
|
||||
flags = 0
|
||||
libonlp.onlp_sys_dump(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_sys_dump(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("System Information", bufStr)
|
||||
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer_p)
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer.ptr)
|
||||
|
||||
# this is not the system OID
|
||||
|
||||
oid = self.sys_info.hdr.coids[0]
|
||||
libonlp.onlp_sys_dump(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_sys_dump(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIsNone(bufStr)
|
||||
|
||||
@@ -308,18 +318,18 @@ class SysTest(OnlpTestMixin,
|
||||
|
||||
oid = onlp.onlp.ONLP_OID_SYS
|
||||
flags = 0
|
||||
libonlp.onlp_sys_show(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_sys_show(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("System Information", bufStr)
|
||||
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer_p)
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer.ptr)
|
||||
|
||||
# this is not the system OID
|
||||
|
||||
oid = self.sys_info.hdr.coids[0]
|
||||
libonlp.onlp_sys_show(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_sys_show(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIsNone(bufStr)
|
||||
|
||||
@@ -431,15 +441,15 @@ class OidTest(OnlpTestMixin,
|
||||
def testOidDump(self):
|
||||
oid = self.hdr.coids[0]
|
||||
flags = 0
|
||||
libonlp.onlp_oid_dump(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_oid_dump(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
self.assertIn("Description:", buf.string_at())
|
||||
|
||||
def testOidTableDump(self):
|
||||
tbl = self.hdr.coids
|
||||
flags = 0
|
||||
libonlp.onlp_oid_table_dump(tbl, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_oid_table_dump(tbl, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
lines = buf.string_at().splitlines(False)
|
||||
lines = [x for x in lines if 'Description' in x]
|
||||
self.assertGreater(len(lines), 1)
|
||||
@@ -447,15 +457,15 @@ class OidTest(OnlpTestMixin,
|
||||
def testOidShow(self):
|
||||
oid = self.hdr.coids[0]
|
||||
flags = 0
|
||||
libonlp.onlp_oid_show(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_oid_show(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
self.assertIn("Description:", buf.string_at())
|
||||
|
||||
def testOidTableShow(self):
|
||||
tbl = self.hdr.coids
|
||||
flags = 0
|
||||
libonlp.onlp_oid_table_show(tbl, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_oid_table_show(tbl, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
lines = buf.string_at().splitlines(False)
|
||||
lines = [x for x in lines if 'Description' in x]
|
||||
self.assertGreater(len(lines), 1)
|
||||
@@ -544,13 +554,13 @@ class FanTest(OnlpTestMixin,
|
||||
self.auditFanDir(oid)
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_fan_dump(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_fan_dump(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("Fan", bufStr)
|
||||
|
||||
libonlp.onlp_fan_show(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_fan_show(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("Fan", bufStr)
|
||||
|
||||
@@ -814,13 +824,13 @@ class LedTest(OnlpTestMixin,
|
||||
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_led_dump(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_led_dump(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("led @", bufStr)
|
||||
|
||||
libonlp.onlp_led_show(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_led_show(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("led @", bufStr)
|
||||
|
||||
@@ -1038,8 +1048,8 @@ class ConfigTest(OnlpTestMixin,
|
||||
|
||||
def testConfigShow(self):
|
||||
|
||||
libonlp.onlp_config_show(self.aim_pvs_buffer_p)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_config_show(self.aim_pvs_buffer.ptr)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
self.assertIn("ONLP_CONFIG_INFO_STR_MAX = 64\n", buf.string_at())
|
||||
|
||||
class ThermalTest(OnlpTestMixin,
|
||||
@@ -1112,13 +1122,13 @@ class ThermalTest(OnlpTestMixin,
|
||||
self.assertEqual(onlp.onlp.ONLP_STATUS.E_UNSUPPORTED, code)
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_thermal_dump(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_thermal_dump(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("thermal @", bufStr)
|
||||
|
||||
libonlp.onlp_thermal_show(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_thermal_show(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("thermal @", bufStr)
|
||||
|
||||
@@ -1205,13 +1215,13 @@ class PsuTest(OnlpTestMixin,
|
||||
self.assertEqual(onlp.onlp.ONLP_STATUS.E_UNSUPPORTED, code)
|
||||
|
||||
flags = 0
|
||||
libonlp.onlp_psu_dump(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_psu_dump(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("psu @", bufStr)
|
||||
|
||||
libonlp.onlp_psu_show(oid, self.aim_pvs_buffer_p, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_psu_show(oid, self.aim_pvs_buffer.ptr, flags)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
bufStr = buf.string_at()
|
||||
self.assertIn("psu @", bufStr)
|
||||
|
||||
@@ -1412,8 +1422,8 @@ class SfpTest(OnlpTestMixin,
|
||||
cl.append(onlp.sff.SFF_MODULE_CAPS.name(fl))
|
||||
self.log.info("module caps %s", "+".join(cl))
|
||||
|
||||
libonlp.sff_info_show(ctypes.byref(sffEeprom.info), self.aim_pvs_buffer_p)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.sff_info_show(ctypes.byref(sffEeprom.info), self.aim_pvs_buffer.ptr)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
self.assertIn("Vendor:", buf.string_at())
|
||||
|
||||
# cons up a new info structure
|
||||
@@ -1428,9 +1438,9 @@ class SfpTest(OnlpTestMixin,
|
||||
sffEeprom.info.length)
|
||||
self.assertStatusOK(sts)
|
||||
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer_p)
|
||||
libonlp.sff_info_show(ctypes.byref(info), self.aim_pvs_buffer_p)
|
||||
buf2 = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer.ptr)
|
||||
libonlp.sff_info_show(ctypes.byref(info), self.aim_pvs_buffer.ptr)
|
||||
buf2 = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
self.assertEqual(buf2.string_at(), buf.string_at())
|
||||
|
||||
# test parsing from a file
|
||||
@@ -1465,9 +1475,9 @@ class SfpTest(OnlpTestMixin,
|
||||
def testDump(self):
|
||||
unittest.skip("this is a really slow command")
|
||||
return
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer_p)
|
||||
libonlp.onlp_sfp_dump(self.aim_pvs_buffer_p)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer_p)
|
||||
libonlp.aim_pvs_buffer_reset(self.aim_pvs_buffer.ptr)
|
||||
libonlp.onlp_sfp_dump(self.aim_pvs_buffer.ptr)
|
||||
buf = libonlp.aim_pvs_buffer_get(self.aim_pvs_buffer.ptr)
|
||||
self.assertIn("Presence Bitmap", buf.string_at())
|
||||
|
||||
def auditIoctl(self, port):
|
||||
|
||||
Reference in New Issue
Block a user