openstack
/
virtualpdu
Code Issues Proposed changes

Taking executor out of the context

This commit is contained in:
srobert 2016-11-09 14:36:02 -05:00
parent 6daec3f99d
commit 1980992828
3 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
virtualpdu/core.py
View File

@ -26,6 +26,7 @@ class Core(object):
self.store = store
self.default_state = default_state
self.logger = logging.getLogger(__name__)
self.executor = ThreadPoolExecutor(max_workers=1)
def pdu_outlet_state_changed(self, pdu, outlet, state):
self.store[(pdu, outlet)] = state
@ -60,8 +61,7 @@ class Core(object):
self.logger.error("Unknown power state: {}".format(state))
return
with ThreadPoolExecutor(max_workers=1) as executor:
executor.submit(switch_power)
self.executor.submit(switch_power)
def get_pdu_outlet_state(self, pdu, outlet):
try:

6
virtualpdu/tests/integration/test_core_integration.py
View File

@ -13,9 +13,9 @@
# limitations under the License.
import random
import time
from pysnmp.entity.rfc3413.oneliner import cmdgen
from virtualpdu import core
from virtualpdu import drivers
from virtualpdu.drivers import libvirt_driver
@ -67,12 +67,16 @@ class TestCoreIntegration(base.TestCase):
snmp_client_.set(
self.outlet_oid,
apc_rackpdu.APCRackPDUOutletControl.states.IMMEDIATE_OFF)
time.sleep(0.1)
self.assertEqual(drivers.POWER_OFF,
self.driver.get_power_state('test'))
snmp_client_.set(
self.outlet_oid,
apc_rackpdu.APCRackPDUOutletControl.states.IMMEDIATE_ON)
time.sleep(0.1)
self.assertEqual(drivers.POWER_ON,
self.driver.get_power_state('test'))

6
virtualpdu/tests/unit/test_core.py
View File

@ -36,7 +36,7 @@ class TestCore(base.TestCase):
self.core.pdu_outlet_state_changed(pdu='my_pdu',
outlet=1,
state=core.POWER_OFF)
time.sleep(0)
time.sleep(0.1)
self.driver_mock.power_off.assert_called_with('server_one')
def test_pdu_outlet_state_changed_machine_not_in_mapping_noop(self):
@ -51,14 +51,14 @@ class TestCore(base.TestCase):
self.core.pdu_outlet_state_changed(pdu='my_pdu',
outlet=1,
state=core.POWER_ON)
time.sleep(0)
time.sleep(0.1)
self.driver_mock.power_on.assert_called_with('server_one')
def test_pdu_outlet_state_changed_on_reboot(self):
self.core.pdu_outlet_state_changed(pdu='my_pdu',
outlet=1,
state=core.REBOOT)
time.sleep(0)
time.sleep(0.1)
self.driver_mock.assert_has_calls([mock.call.power_off('server_one'),
mock.call.power_on('server_one')])