diff --git a/bitelab/__init__.py b/bitelab/__init__.py index 5440d67..048fc72 100644 --- a/bitelab/__init__.py +++ b/bitelab/__init__.py @@ -83,8 +83,8 @@ async def snmpget(host, oid, type): raise RuntimeError('unknown type: %s' % repr(type)) -#async def snmpset(host, oid, value): -# return await _snmpwrapper('set', host, oid, value=value) +async def snmpset(host, oid, value): + return await _snmpwrapper('set', host, oid, value=value) class Attribute: '''Base class for board attributes. This is for both read-only @@ -98,7 +98,7 @@ class Attribute: async def getvalue(self): # pragma: no cover raise NotImplementedError - async def setvalue(self): # pragma: no cover + async def setvalue(self, v): # pragma: no cover raise NotImplementedError class Power(Attribute): @@ -114,8 +114,9 @@ class SNMPPower(Power): return await snmpget(self.host, 'pethPsePortAdminEnable.1.%d' % self.port, 'bool') - #async def setvalue(self, v): - # pass + async def setvalue(self, v): + return await snmpset(self.host, + 'pethPsePortAdminEnable.1.%d' % self.port, v) class BoardImpl: def __init__(self, name, brdclass, options): @@ -484,6 +485,17 @@ async def _setup_data(data): await data.APIKey.objects.create(user='foo', key='thisisanapikey') await data.APIKey.objects.create(user='bar', key='anotherlongapikey') +def _wrap_subprocess_exec(mockobj, stdout=b'', stderr=b'', retcode=0): + assert isinstance(stdout, bytes) + assert isinstance(stderr, bytes) + proc = Mock() + proc.communicate = AsyncMock() + proc.communicate.return_value = (stdout, stderr) + proc.wait = AsyncMock() + proc.wait.return_value = retcode + proc.returncode = retcode + mockobj.return_value = proc + # Per RFC 5737 (https://tools.ietf.org/html/rfc5737): # The blocks 192.0.2.0/24 (TEST-NET-1), 198.51.100.0/24 (TEST-NET-2), # and 203.0.113.0/24 (TEST-NET-3) are provided for use in @@ -557,45 +569,6 @@ class TestBiteLab(unittest.IsolatedAsyncioTestCase): self.assertEqual(res.json(), { 'cora-z7s': BoardClassInfo(**{ 'arch': 'arm-armv7', 'clsname': 'cora-z7s', }) }) - @staticmethod - def _wrap_subprocess_exec(mockobj, stdout=b'', stderr=b'', retcode=0): - assert isinstance(stdout, bytes) - assert isinstance(stderr, bytes) - proc = Mock() - proc.communicate = AsyncMock() - proc.communicate.return_value = (stdout, stderr) - proc.wait = AsyncMock() - proc.wait.return_value = retcode - proc.returncode = retcode - mockobj.return_value = proc - - @patch('asyncio.create_subprocess_exec') - async def test_snmpwrapper(self, cse): - self._wrap_subprocess_exec(cse, b'false\n') - - r = await snmpget('somehost', 'snmpoid', 'bool') - - self.assertEqual(r, False) - - cse.assert_called_with('snmpget', '-Oqv', 'somehost', - 'snmpoid', stdout=subprocess.PIPE) - - self._wrap_subprocess_exec(cse, b'true\n') - r = await snmpget('somehost', 'snmpoid', 'bool') - - self.assertEqual(r, True) - - # that a bogus return value - self._wrap_subprocess_exec(cse, b'bogus\n') - - # raises an error - with self.assertRaises(RuntimeError): - await snmpget('somehost', 'snmpoid', 'bool') - - # that an unknown type, raises an error - with self.assertRaises(RuntimeError): - await snmpget('somehost', 'snmpoid', 'randomtype') - @patch('asyncio.create_subprocess_exec') @patch('bitelab.snmpget') async def test_board_reserve_release(self, sg, cse): @@ -610,7 +583,7 @@ class TestBiteLab(unittest.IsolatedAsyncioTestCase): sg.return_value = False # that when the setup script will fail - self._wrap_subprocess_exec(cse, stderr=b'error', retcode=1) + _wrap_subprocess_exec(cse, stderr=b'error', retcode=1) # that reserving the board res = await self.client.post('/board/cora-1/reserve', @@ -634,7 +607,7 @@ class TestBiteLab(unittest.IsolatedAsyncioTestCase): stderr=subprocess.PIPE) # that when the setup script returns - self._wrap_subprocess_exec(cse, + _wrap_subprocess_exec(cse, json.dumps(dict(ip='192.0.2.10')).encode('utf-8')) # that reserving the board @@ -792,6 +765,57 @@ class TestDatabase(unittest.IsolatedAsyncioTestCase): self.assertEqual((await data.APIKey.objects.get( key='anotherlongapikey')).user, 'bar') +class TestSNMPWrapper(unittest.IsolatedAsyncioTestCase): + @patch('asyncio.create_subprocess_exec') + async def test_snmpwrapper(self, cse): + _wrap_subprocess_exec(cse, b'false\n') + + r = await snmpget('somehost', 'snmpoid', 'bool') + + self.assertEqual(r, False) + + cse.assert_called_with('snmpget', '-Oqv', 'somehost', + 'snmpoid', stdout=subprocess.PIPE) + + _wrap_subprocess_exec(cse, b'true\n') + r = await snmpget('somehost', 'snmpoid', 'bool') + + self.assertEqual(r, True) + + # that a bogus return value + _wrap_subprocess_exec(cse, b'bogus\n') + + # raises an error + with self.assertRaises(RuntimeError): + await snmpget('somehost', 'snmpoid', 'bool') + + # that an unknown type, raises an error + with self.assertRaises(RuntimeError): + await snmpget('somehost', 'snmpoid', 'randomtype') + + +class TestSNMPPower(unittest.IsolatedAsyncioTestCase): + @patch('bitelab.snmpset') + @patch('bitelab.snmpget') + async def test_snmppower(self, sg, ss): + sp = SNMPPower('host', 5) + + # that when snmpget returns False + sg.return_value = False + + self.assertFalse(await sp.getvalue()) + + # calls snmpget w/ the correct args + sg.assert_called_with('host', 'pethPsePortAdminEnable.1.5', + 'bool') + + # that when setvalue is called + await sp.setvalue(True) + + # calls snmpset w/ the correct args + ss.assert_called_with('host', 'pethPsePortAdminEnable.1.5', + True) + @patch.dict(os.environ, dict(BITELAB_URL='http://someserver/')) @patch.dict(os.environ, dict(BITELAB_AUTH='thisisanapikey')) class TestClient(unittest.TestCase):