l.behm/bumblebee-status
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

[core] Non-blocking input thread for i3bar events

Browse source 
Make input thread non-blocking by using select(). This increases the CPU
utilization a bit (depending on the timeout), but makes the thread exit
cleanly, even if an exception is thrown in the main thread.

see #23
This commit is contained in:
Tobi-wan Kenobi 2016-12-10 13:45:54 +01:00
parent 0489ce1b51
commit 029492e16d
4 changed files with 45 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

40
tests/test_i3barinput.py
View file

@ -21,23 +21,29 @@ class TestI3BarInput(unittest.TestCase):
def callback(self, event):
self._called += 1
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_basic_read_event(self, mock_input):
def test_basic_read_event(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = ""
self.input.start()
self.input.stop()
mock_input.readline.assert_any_call()
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_ignore_invalid_data(self, mock_input):
def test_ignore_invalid_data(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = "garbage"
self.input.start()
self.assertEquals(self.input.alive(), True)
self.assertEquals(self.input.stop(), True)
mock_input.readline.assert_any_call()
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_ignore_invalid_event(self, mock_input):
def test_ignore_invalid_event(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": None,
"instance": None,
@ -48,8 +54,10 @@ class TestI3BarInput(unittest.TestCase):
self.assertEquals(self.input.stop(), True)
mock_input.readline.assert_any_call()
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_global_callback(self, mock_input):
def test_global_callback(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": "somename",
"instance": "someinstance",
@ -61,8 +69,10 @@ class TestI3BarInput(unittest.TestCase):
mock_input.readline.assert_any_call()
self.assertTrue(self._called > 0)
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_remove_global_callback(self, mock_input):
def test_remove_global_callback(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": "somename",
"instance": "someinstance",
@ -75,8 +85,10 @@ class TestI3BarInput(unittest.TestCase):
mock_input.readline.assert_any_call()
self.assertTrue(self._called == 0)
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_global_callback_button_missmatch(self, mock_input):
def test_global_callback_button_missmatch(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": "somename",
"instance": "someinstance",
@ -88,8 +100,10 @@ class TestI3BarInput(unittest.TestCase):
mock_input.readline.assert_any_call()
self.assertTrue(self._called == 0)
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_module_callback(self, mock_input):
def test_module_callback(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": self.anyModule.id,
"instance": None,
@ -101,8 +115,10 @@ class TestI3BarInput(unittest.TestCase):
mock_input.readline.assert_any_call()
self.assertTrue(self._called > 0)
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_remove_module_callback(self, mock_input):
def test_remove_module_callback(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": self.anyModule.id,
"instance": None,
@ -115,8 +131,10 @@ class TestI3BarInput(unittest.TestCase):
mock_input.readline.assert_any_call()
self.assertTrue(self._called == 0)
@mock.patch("select.select")
@mock.patch("sys.stdin")
def test_widget_callback(self, mock_input):
def test_widget_callback(self, mock_input, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": "test",
"instance": self.anyWidget.id,
@ -128,9 +146,11 @@ class TestI3BarInput(unittest.TestCase):
mock_input.readline.assert_any_call()
self.assertTrue(self._called > 0)
@mock.patch("select.select")
@mock.patch("subprocess.Popen")
@mock.patch("sys.stdin")
def test_widget_cmd_callback(self, mock_input, mock_output):
def test_widget_cmd_callback(self, mock_input, mock_output, mock_select):
mock_select.return_value = (1,2,3)
mock_input.readline.return_value = json.dumps({
"name": "test",
"instance": self.anyWidget.id,