feat(Anaconda): Local Repo

This commit is contained in:
2026-06-13 02:06:34 +02:00
parent 05ae9c36c3
commit c412cd5d33
203 changed files with 23997 additions and 7 deletions
@@ -0,0 +1,118 @@
# Advanced widgets test cases.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from unittest.mock import patch
from io import StringIO
from simpleline.render.adv_widgets import GetInputScreen, GetPasswordInputScreen
from .. import UtilityMixin
@patch('simpleline.input.input_handler.InputHandlerRequest._get_input')
@patch('sys.stdout', new_callable=StringIO)
class AdvWidgets_TestCase(unittest.TestCase, UtilityMixin):
def setUp(self):
self.correct_input = False
self.args_used = False
def test_gettext(self, stdout_mock, stdin_mock):
prompt = "Type input"
input_text = "user input"
screen = GetInputScreen(message=prompt)
stdin_mock.return_value = input_text
self.schedule_screen_and_run(screen)
expected_output = self.create_output_with_separators(["%s: " % prompt]).rstrip('\n')
self.assertEqual(expected_output, stdout_mock.getvalue())
self.assertEqual(screen.value, input_text)
def test_gettext_with_condition(self, stdout_mock, stdin_mock):
prompt = "Type input"
wrong_input = "wrong"
condition = lambda x, _: x != wrong_input
stdin_mock.side_effect = self.input_generator()
screen = GetInputScreen(message=prompt)
screen.add_acceptance_condition(condition)
self.schedule_screen_and_run(screen)
temp_prompt = ["%s: %s: " % (prompt, prompt)]
expected_output = self.create_output_with_separators(temp_prompt).rstrip("\n")
self.assertEqual(expected_output, stdout_mock.getvalue())
self.assertTrue(self.correct_input)
def test_gettext_with_condition_and_use_arg(self, stdout_mock, stdin_mock):
prompt = "Type input"
user_input = "y"
stdin_mock.return_value = user_input
screen = GetInputScreen(message=prompt)
screen.add_acceptance_condition(self.acceptance_condition_test, "y")
self.schedule_screen_and_run(screen)
expected_msg = "%s: " % prompt
expected_output = self.create_output_with_separators([expected_msg]).rstrip("\n")
self.assertEqual(expected_output, stdout_mock.getvalue())
self.assertTrue(self.args_used)
@patch("simpleline.global_configuration.GlobalConfiguration.password_function")
def test_getpass(self, hiden_stdin_mock, stdout_mock, stdin_mock):
prompt = "Type input"
input_text = "user input"
screen = GetPasswordInputScreen(message=prompt)
hiden_stdin_mock.return_value = input_text
self.schedule_screen_and_run(screen)
self.assertEqual(screen.value, input_text)
@patch("simpleline.global_configuration.GlobalConfiguration.password_function")
def test_getpass_with_condition(self, hiden_stdin_mock, stdout_mock, stdin_mock):
prompt = "Type input"
wrong_input = "wrong"
condition = lambda x, _: x != wrong_input
hiden_stdin_mock.side_effect = self.input_generator()
screen = GetPasswordInputScreen(message=prompt)
screen.add_acceptance_condition(condition)
self.schedule_screen_and_run(screen)
self.assertTrue(self.correct_input)
def input_generator(self):
for i in ("wrong", "correct"):
if i == "correct":
self.correct_input = True
yield i
def acceptance_condition_test(self, user_input, args):
if user_input == args:
self.args_used = True
return True
@@ -0,0 +1,120 @@
# App class test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from unittest import mock
from simpleline import App
from simpleline.global_configuration import GlobalConfiguration
from simpleline.input.input_threading import InputThreadManager
from simpleline.render.screen_scheduler import ScreenScheduler
from simpleline.event_loop.main_loop import MainLoop
from simpleline.errors import NothingScheduledError
class App_TestCase(unittest.TestCase):
def test_create_instance(self):
App.initialize()
self.assertTrue(isinstance(App.get_scheduler(), ScreenScheduler))
self.assertTrue(isinstance(App.get_event_loop(), MainLoop))
self.assertTrue(isinstance(App.get_configuration(), GlobalConfiguration))
def test_create_instance_with_custom_scheduler(self):
App.initialize(scheduler=CustomScreenScheduler(CustomEventLoop()))
self.assertTrue(isinstance(App.get_scheduler(), CustomScreenScheduler))
def test_create_instance_with_event_loop(self):
App.initialize(event_loop=CustomEventLoop())
self.assertTrue(isinstance(App.get_event_loop(), CustomEventLoop))
def test_create_instance_with_configuration(self):
App.initialize(global_configuration=CustomGlobalConfiguration())
self.assertTrue(isinstance(App.get_configuration(), CustomGlobalConfiguration))
def test_create_instance_with_custom_everything(self):
event_loop = CustomEventLoop()
App.initialize(event_loop=event_loop,
scheduler=CustomScreenScheduler(event_loop),
global_configuration=CustomGlobalConfiguration())
self.assertTrue(isinstance(App.get_event_loop(), CustomEventLoop))
self.assertTrue(isinstance(App.get_scheduler(), CustomScreenScheduler))
self.assertTrue(isinstance(App.get_configuration(), CustomGlobalConfiguration))
def test_reinitialize(self):
event_loop1 = CustomEventLoop()
event_loop2 = CustomEventLoop()
scheduler1 = CustomScreenScheduler(event_loop1)
scheduler2 = CustomScreenScheduler(event_loop2)
configuration1 = CustomGlobalConfiguration()
configuration2 = CustomGlobalConfiguration()
App.initialize(event_loop=event_loop1, scheduler=scheduler1,
global_configuration=configuration1)
self._check_app_settings(event_loop1, scheduler1, configuration1)
App.initialize(event_loop=event_loop2, scheduler=scheduler2,
global_configuration=configuration2)
self._check_app_settings(event_loop2, scheduler2, configuration2)
App.initialize()
self.assertNotEqual(App.get_event_loop(), event_loop2)
self.assertNotEqual(App.get_scheduler(), scheduler2)
self.assertNotEqual(App.get_configuration(), configuration2)
def test_input_thread_manager_after_initialize(self):
App.initialize()
thread_mgr = InputThreadManager.get_instance()
App.initialize()
self.assertNotEqual(thread_mgr, InputThreadManager.get_instance())
@mock.patch('simpleline.event_loop.main_loop.MainLoop.run')
def test_run_shortcut(self, run_mock):
App.initialize()
App.get_configuration().should_run_with_empty_stack = True
App.run()
self.assertTrue(run_mock.called)
def test_run_with_empty_screen_stack(self):
App.initialize()
with self.assertRaises(NothingScheduledError):
App.run()
def _check_app_settings(self, event_loop, scheduler, configuration):
self.assertEqual(App.get_event_loop(), event_loop)
self.assertEqual(App.get_scheduler(), scheduler)
self.assertEqual(App.get_configuration(), configuration)
class CustomScreenScheduler(ScreenScheduler):
pass
class CustomEventLoop(MainLoop):
pass
class CustomGlobalConfiguration(GlobalConfiguration):
pass
@@ -0,0 +1,394 @@
# Containers test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from io import StringIO
from unittest.mock import patch
from simpleline import App
from simpleline.render.containers import WindowContainer, ListRowContainer, ListColumnContainer, \
KeyPattern
from simpleline.render.screen import UIScreen, InputState
from simpleline.render.widgets import TextWidget
from .widgets_test import BaseWidgets_TestCase
class Containers_TestCase(BaseWidgets_TestCase):
def _test_callback(self, data):
pass
def test_listrow_container(self):
c = ListRowContainer(columns=2,
items=[self.w2, self.w3, self.w5],
columns_width=10,
spacing=2,
numbering=False)
c.render(25)
expected_result = [u"Test Test 2",
u"Test 3"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_empty(self):
c = ListRowContainer(columns=1)
c.render(10)
result = c.get_lines()
self.assertEqual(len(result), 0)
def test_more_columns_than_widgets(self):
c = ListRowContainer(columns=3, items=[self.w1], columns_width=40, numbering=False)
c.render(80)
expected_result = [u"Můj krásný dlouhý text"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_listrow_wrapping(self):
# spacing is 3 by default
c = ListRowContainer(2,
[self.w1, self.w2, self.w3, self.w4],
columns_width=15,
numbering=False)
c.render(25)
expected_result = [u"Můj krásný Test",
u"dlouhý text",
u"Test 2 Krásný dlouhý",
u" text podruhé"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_newline_wrapping(self):
widgets = [TextWidget("Hello"), TextWidget("Wrap\nthis\ntext"), TextWidget("Hi"),
TextWidget("Hello2")]
c = ListRowContainer(3, widgets, columns_width=6, spacing=1, numbering=False)
c.render(80)
expected_result = [u"Hello Wrap Hi",
u" this",
u" text",
u"Hello2"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_listcolumn_container(self):
c = ListColumnContainer(columns=2,
items=[self.w2, self.w3, self.w5],
columns_width=10,
spacing=2,
numbering=False)
c.render(25)
expected_result = [u"Test Test 3",
u"Test 2"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_listcolumn_wrapping(self):
# spacing is 3 by default
c = ListColumnContainer(2,
[self.w1, self.w2, self.w3, self.w4],
columns_width=15,
numbering=False)
c.render(25)
expected_result = [u"Můj krásný Test 2",
u"dlouhý text",
u"Test Krásný dlouhý",
u" text podruhé"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_add_new_container(self):
c = ListRowContainer(columns=2,
items=[TextWidget("Ahoj")],
columns_width=15,
spacing=0,
numbering=False)
expected_result = [u"Ahoj"]
c.render(80)
self.evaluate_result(c.get_lines(), expected_result)
c.add(TextWidget("Nový widget"))
c.add(TextWidget("Hello"))
expected_result = [u"Ahoj Nový widget",
u"Hello"]
c.render(80)
self.evaluate_result(c.get_lines(), expected_result)
def test_column_numbering(self):
# spacing is 3 by default
c = ListColumnContainer(2, [self.w1, self.w2, self.w3, self.w4], columns_width=16)
c.render(25)
expected_result = [u"1) Můj krásný 3) Test 2",
u" dlouhý text",
u"2) Test 4) Krásný dlouhý",
u" text podruhé"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_row_numbering(self):
# spacing is 3 by default
c = ListRowContainer(2, [self.w1, self.w2, self.w3, self.w4], columns_width=16)
c.render(25)
expected_result = [u"1) Můj krásný 2) Test",
u" dlouhý text",
u"3) Test 2 4) Krásný dlouhý",
u" text podruhé"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_custom_numbering(self):
# spacing is 3 by default
c = ListRowContainer(2, [self.w1, self.w2, self.w3, self.w4], columns_width=20)
c.key_pattern = KeyPattern("a {:d} a ")
c.render(25)
expected_result = [u"a 1 a Můj krásný a 2 a Test",
u" dlouhý text",
u"a 3 a Test 2 a 4 a Krásný dlouhý",
u" text podruhé"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_window_container(self):
c = WindowContainer(title="Test")
c.add(TextWidget("Body"))
c.render(10)
expected_result = [u"Test",
u"",
u"Body"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_window_container_with_multiple_items(self):
c = WindowContainer(title="Test")
c.add(TextWidget("Body"))
c.add(TextWidget("Body second line"))
c.render(30)
expected_result = [u"Test",
u"",
u"Body",
u"Body second line"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_window_container_wrapping(self):
c = WindowContainer(title="Test")
c.add(TextWidget("Body long line"))
c.add(TextWidget("Body"))
c.render(5)
expected_result = [u"Test",
u"",
u"Body",
u"long",
u"line",
u"Body"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_list_container_without_width(self):
column_count = 3
spacing_width = 3
c = ListRowContainer(column_count, spacing=spacing_width, numbering=False)
c.add(TextWidget("AAAA"))
c.add(TextWidget("BBBB"))
c.add(TextWidget("CCCCC")) # this line is too long
c.add(TextWidget("DDDD"))
expected_col_width = 4
expected_spacing_sum = 2 * spacing_width # three columns so 2 spacing between them
render_width = (column_count * expected_col_width) + expected_spacing_sum
c.render(render_width)
expected_result = [u"AAAA BBBB CCCC",
u" C",
u"DDDD"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_list_container_too_small(self):
# to be able to render this container we need at least 11 width
# 8 will take only spacing and then 1 for every column
c = ListRowContainer(3, spacing=4, numbering=False)
c.add(TextWidget("This can't be rendered."))
c.add(TextWidget("Because spacing takes more space than maximal width."))
c.add(TextWidget("Exception will raise."))
with self.assertRaisesRegex(ValueError, "Columns width is too small."):
c.render(10)
def test_list_container_too_small_turn_off_numbering(self):
# to be able to render this container we need
# 11 width + three times numbers (3 characters) = 20
#
# 8 will take only spacing and then 1 for every column
c = ListRowContainer(3, spacing=4, numbering=True)
c.add(TextWidget("This can't be rendered."))
c.add(TextWidget("Because spacing takes more space than maximal width."))
c.add(TextWidget("Exception will raise with info to turn off numbering."))
with self.assertRaisesRegex(ValueError, "Increase column width or disable numbering."):
c.render(19)
@patch('simpleline.input.input_handler.InputHandlerRequest._get_input')
@patch('sys.stdout', new_callable=StringIO)
class ContainerInput_TestCase(unittest.TestCase):
def setUp(self):
self._callback_id = None
self._callback_called = None
def _prepare_callbacks(self, container, count):
for i in range(count):
container.add(TextWidget("Test"), self._callback, i + 1)
def test_list_widget_input_processing(self, out_mock, in_mock):
# call first container callback
in_mock.return_value = "2"
screen = ScreenWithListWidget(3)
App.initialize()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(1, screen.container_callback_input)
# TEST 0 or less as user input
def test_list_input_processing_input_0(self, out_mock, in_mock):
c = ListRowContainer(1)
self._prepare_callbacks(c, 3)
self.assertFalse(c.process_user_input("0"))
def test_list_input_processing_negative_number(self, out_mock, in_mock):
c = ListRowContainer(1)
self._prepare_callbacks(c, 3)
self.assertFalse(c.process_user_input("-2"))
def test_list_input_processing_exceeded(self, out_mock, in_mock):
c = ListRowContainer(1)
self._prepare_callbacks(c, 2)
self.assertFalse(c.process_user_input("3"))
def test_list_without_callback(self, out_mock, in_mock):
c = ListRowContainer(1)
c.add(TextWidget("Test"))
self.assertTrue(c.process_user_input("1"))
def test_list_callback_without_data(self, out_mock, in_mock):
c = ListRowContainer(1)
c.add(TextWidget("Test"), self._callback)
self.assertTrue(c.process_user_input("1"))
self.assertIsNone(self._callback_called)
def test_list_correct_input_processing(self, out_mock, in_mock):
c = ListRowContainer(1)
self._prepare_callbacks(c, 3)
self.assertTrue(c.process_user_input("2"))
self.assertEqual(self._callback_called, 2)
def test_list_wrong_input_processing(self, out_mock, in_mock):
c = ListRowContainer(1)
self._prepare_callbacks(c, 3)
self.assertFalse(c.process_user_input("c"))
def test_list_input_processing_none(self, out_mock, in_mock):
c = ListRowContainer(1)
self._prepare_callbacks(c, 2)
self.assertFalse(c.process_user_input(None))
def _callback(self, data):
self._callback_called = data
class ScreenWithListWidget(UIScreen):
def __init__(self, widgets_count):
super().__init__()
self._widgets_count = widgets_count
self._list_widget = None
self.container_callback_input = -1
def refresh(self, args=None):
super().refresh(args)
self._list_widget = ListRowContainer(2)
for i in range(self._widgets_count):
self._list_widget.add(TextWidget("Test %s" % i), self._callback, i)
self.window.add(self._list_widget)
def input(self, args, key):
self.close()
if self._list_widget.process_user_input(key):
return InputState.PROCESSED
return InputState.DISCARDED
def _callback(self, data):
self.container_callback_input = data
@@ -0,0 +1,310 @@
# Event loop test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from simpleline.event_loop import AbstractSignal
from simpleline.event_loop import EventHandler
from simpleline.event_loop import ExitMainLoop
from simpleline.event_loop.main_loop import MainLoop
class EventLoopHandler_TestCase(unittest.TestCase):
def callback_func(self):
pass
def test_signal_handler_named_params(self):
data = [1, 2, "args"]
ev = EventHandler(callback=self.callback_func, data=data)
self.assertEqual(ev.callback, self.callback_func)
self.assertEqual(ev.data, data)
def test_signal_handler_positional_params(self):
data = [1, 3, "args"]
ev = EventHandler(self.callback_func, data)
self.assertEqual(ev.callback, self.callback_func)
self.assertEqual(ev.data, data)
class ProcessEvents_TestCase(unittest.TestCase):
def setUp(self):
self.signal_counter = 0
self.signal_counter2 = 0
self.signal_counter_copied = 0
self.callback_called = False
self.callback_args = None
self.create_loop()
def create_loop(self):
self.loop = MainLoop()
def test_simple_register_handler(self):
self.callback_called = False
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_callback)
loop.enqueue_signal(TestSignal())
loop.process_signals()
self.assertTrue(self.callback_called)
def test_process_more_signals(self):
self.signal_counter = 0
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_signal_counter)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal())
loop.process_signals()
self.assertEqual(self.signal_counter, 3)
def test_process_signals_multiple_times(self):
self.signal_counter = 0
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_signal_counter)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal())
loop.process_signals()
self.assertEqual(self.signal_counter, 2)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal())
loop.process_signals()
self.assertEqual(self.signal_counter, 4)
def test_wait_on_signal(self):
self.signal_counter = 0
loop = self.loop
loop.register_signal_handler(TestSignal,
self._handler_signal_counter)
loop.register_signal_handler(TestSignal2,
self._handler_process_events_then_register_testsignal,
loop)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal2())
loop.process_signals(return_after=TestSignal2)
self.assertEqual(self.signal_counter, 1)
loop.process_signals()
self.assertEqual(self.signal_counter, 2)
def test_wait_on_signal_skipped_by_inner_process_events(self):
self.signal_counter = 0
loop = self.loop
loop.register_signal_handler(TestSignal,
self._handler_signal_counter)
# run process signals recursively in this handler which will skip processing
loop.register_signal_handler(TestSignal2,
self._handler_process_events_then_register_testsignal,
loop)
loop.enqueue_signal(TestSignal2())
loop.enqueue_signal(TestSignal())
# new signal will be registered in handler method but that shouldn't be processed
# because it should end on the first signal even when it was skipped
loop.process_signals(return_after=TestSignal)
self.assertEqual(self.signal_counter, 1)
def test_multiple_handlers_to_signal(self):
self.signal_counter = 0
self.signal_counter2 = 0
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_signal_counter)
loop.register_signal_handler(TestSignal, self._handler_signal_counter2)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal())
loop.process_signals()
self.assertEqual(self.signal_counter, 2)
self.assertEqual(self.signal_counter2, 2)
def test_priority_signal_processing(self):
self.signal_counter = 0
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_signal_counter)
loop.register_signal_handler(TestPrioritySignal, self._handler_signal_counter)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal())
# should be processed as first signal because of priority
loop.enqueue_signal(TestPrioritySignal())
loop.enqueue_signal(TestSignal())
loop.process_signals()
self.assertEqual(self.signal_counter, 1)
# process rest of the signals
loop.process_signals()
self.assertEqual(self.signal_counter, 4)
def test_low_priority_signal_processing(self):
self.signal_counter = 0
self.signal_counter_copied = 0
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_signal_counter)
loop.register_signal_handler(TestLowPrioritySignal, self._handler_signal_copy_counter)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestLowPrioritySignal())
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal())
loop.process_signals()
self.assertEqual(self.signal_counter, 3)
# process the low priority signal
loop.process_signals()
self.assertEqual(self.signal_counter_copied, 3)
def test_quit_callback(self):
self.callback_called = False
self.callback_args = None
msg = "Test data"
loop = self.loop
loop.set_quit_callback(self._handler_quit_callback, args=msg)
loop.register_signal_handler(TestSignal, self._handler_raise_ExitMainLoop_exception)
loop.enqueue_signal(TestSignal())
loop.run()
self.assertTrue(self.callback_called)
self.assertEqual(msg, self.callback_args)
def test_force_quit(self):
self.callback_called = False
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_callback)
loop.register_signal_handler(TestSignal2, self._handler_force_quit_exception)
loop.enqueue_signal(TestSignal2())
loop.enqueue_signal(TestSignal())
loop.run()
self.assertFalse(self.callback_called)
def test_force_quit_recursive_loop(self):
self.callback_called = False
loop = self.loop
loop.register_signal_handler(TestSignal,
self._handler_start_inner_loop_and_enqueue_event,
TestSignal3())
loop.register_signal_handler(TestSignal2,
self._handler_callback)
loop.register_signal_handler(TestSignal3,
self._handler_force_quit_exception)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal2())
loop.run()
self.assertFalse(self.callback_called)
def test_force_quit_when_waiting_on_signal(self):
self.callback_called = False
loop = self.loop
loop.register_signal_handler(TestSignal, self._handler_force_quit_exception)
loop.register_signal_handler(TestSignal2, self._handler_callback)
loop.enqueue_signal(TestSignal())
loop.enqueue_signal(TestSignal2())
# FIXME: Find a better way how to detect infinite loop
# if force quit won't work properly this will hang up
loop.process_signals(return_after=TestSignal3)
self.assertFalse(self.callback_called)
# HANDLERS FOR TESTING
def _handler_callback(self, signal, data):
self.callback_called = True
def _handler_quit_callback(self, args):
self.callback_called = True
self.callback_args = args
def _handler_signal_counter(self, signal, data):
self.signal_counter += 1
def _handler_signal_counter2(self, signal, data):
self.signal_counter2 += 1
def _handler_signal_copy_counter(self, signal, data):
self.signal_counter_copied = self.signal_counter
@staticmethod
def _handler_process_events_then_register_testsignal(signal, data):
event_loop = data
event_loop.process_signals()
# This shouldn't be processed
event_loop.enqueue_signal(TestSignal())
def _handler_start_inner_loop_and_enqueue_event(self, signal, data):
self.loop.execute_new_loop(data)
@staticmethod
def _handler_raise_ExitMainLoop_exception(signal, data):
raise ExitMainLoop()
def _handler_force_quit_exception(self, signal, data):
self.loop.force_quit()
# TESTING EVENTS
class TestSignal(AbstractSignal):
def __init__(self):
# ignore source
super().__init__(None)
class TestSignal2(AbstractSignal):
def __init__(self):
# ignore source
super().__init__(None)
class TestSignal3(AbstractSignal):
def __init__(self):
# ignore source
super().__init__(None)
class TestPrioritySignal(AbstractSignal):
def __init__(self):
# ignore source
super().__init__(None, -10)
class TestLowPrioritySignal(AbstractSignal):
def __init__(self):
# ignore source
super().__init__(None, 20)
@@ -0,0 +1,105 @@
# Event queue test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from unittest.mock import MagicMock
from simpleline.event_loop.event_queue import EventQueue, EventQueueError
from simpleline.event_loop.signals import AbstractSignal
class EventQueue_TestCase(unittest.TestCase):
def setUp(self):
self.e = EventQueue()
def test_queue_is_empty(self):
self.assertTrue(self.e.empty())
def test_enqueue(self):
fake_signal = MagicMock()
self.e.enqueue(fake_signal)
self.assertFalse(self.e.empty())
self.assertEqual(fake_signal, self.e.get())
self.assertTrue(self.e.empty())
def test_enqueue_priority(self):
signal_low_priority = TestSignal(priority=10)
signal_high_priority = TestSignal(priority=0)
self.e.enqueue(signal_low_priority)
self.e.enqueue(signal_high_priority)
self.assertEqual(signal_high_priority, self.e.get())
self.assertEqual(signal_low_priority, self.e.get())
# Test adding signals in different order (result shouldn't change)
self.e.enqueue(signal_high_priority)
self.e.enqueue(signal_low_priority)
self.assertEqual(signal_high_priority, self.e.get())
self.assertEqual(signal_low_priority, self.e.get())
def test_adding_event_source(self):
fake_source = MagicMock()
self.e.add_source(fake_source)
self.assertTrue(self.e.contains_source(fake_source))
def test_removing_event_source(self):
fake_source = MagicMock()
self.e.add_source(fake_source)
self.e.remove_source(fake_source)
self.assertFalse(self.e.contains_source(fake_source))
def test_remove_empty_source(self):
with self.assertRaises(EventQueueError):
self.e.remove_source(MagicMock())
def test_enqueue_if_source_belongs(self):
source = MagicMock()
signal = TestSignal(source=source)
self.e.add_source(source)
self.assertTrue(self.e.enqueue_if_source_belongs(signal, source))
self.assertEqual(signal, self.e.get())
def test_enqueue_if_source_does_not_belong(self):
signal = TestSignal()
signal_low_priority = TestSignal(priority=25)
# the get method will wait if nothing present so adding low priority signal below
# give us check if the queue is really empty
self.e.enqueue(signal_low_priority)
self.assertFalse(self.e.enqueue_if_source_belongs(signal, MagicMock()))
self.assertEqual(signal_low_priority, self.e.get())
class TestSignal(AbstractSignal):
def __init__(self, source=None, priority=20): # pylint: disable=useless-super-delegation
super().__init__(source, priority)
@@ -0,0 +1,84 @@
# GlobalConfiguration class test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from unittest.mock import MagicMock
from simpleline import App
from simpleline.global_configuration import DEFAULT_WIDTH, DEFAULT_PASSWORD_FUNC
class GlobalConfiguration_TestCase(unittest.TestCase):
def tearDown(self):
App.initialize()
def _check_default_width(self, width=DEFAULT_WIDTH):
self.assertEqual(App.get_configuration().width, width)
def _check_default_password_function(self, password_func=None):
if password_func:
self.assertEqual(App.get_configuration().password_function, password_func)
else:
self.assertEqual(App.get_configuration().password_function, DEFAULT_PASSWORD_FUNC)
def test_clear_width(self):
self._check_default_width()
test_width = 150
App.get_configuration().width = test_width
self._check_default_width(test_width)
App.get_configuration().clear_width()
self._check_default_width()
def test_width(self):
self._check_default_width()
App.initialize()
App.get_configuration().width = 100
self._check_default_width(100)
App.initialize()
self._check_default_width()
def test_password_function(self):
self._check_default_password_function()
test_mock = MagicMock()
App.initialize()
App.get_configuration().password_function = test_mock
self._check_default_password_function(test_mock)
App.initialize()
self._check_default_password_function()
def test_clear_password_function(self):
self._check_default_password_function()
test_func = MagicMock()
App.get_configuration().password_function = test_func
self._check_default_password_function(test_func)
App.get_configuration().clear_password_function()
self._check_default_password_function()
@@ -0,0 +1,247 @@
# Rendering screen test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from unittest.mock import Mock, patch
from threading import Barrier, current_thread, Event
from io import StringIO
from simpleline import App
from simpleline.event_loop.main_loop import MainLoop
from simpleline.input.input_handler import InputHandler, PasswordInputHandler
from simpleline.render.prompt import Prompt
@patch('sys.stdout', new_callable=StringIO)
@patch('simpleline.input.input_handler.InputHandlerRequest._get_input')
class InputHandler_TestCase(unittest.TestCase):
def create_loop(self):
self.loop = MainLoop()
def setUp(self):
super().setUp()
self.create_loop()
App.initialize(event_loop=self.loop)
self._callback_called = False
self._callback_input = ""
self._callback_called2 = False
self._callback_input2 = ""
self._thread_barrier = Barrier(2, timeout=3)
self._thread_event_wait_for_inner = Event()
self._thread_event_wait_for_outer = Event()
self._threads = []
def tearDown(self):
super().tearDown()
self._thread_event_wait_for_outer.set()
for t in self._threads:
t.join()
# process InputReceivedSignal
App.get_event_loop().process_signals()
# process InputReadySignal
App.get_event_loop().process_signals()
def test_async_input(self, input_mock, output_mock):
input_mock.return_value = 'a'
h = InputHandler()
h.get_input(Prompt(message="ABC"))
h.wait_on_input()
self.assertEqual(h.value, 'a')
def test_input_received(self, input_mock, output_mock):
input_mock.return_value = 'a'
h = InputHandler()
self.assertFalse(h.input_received())
h.get_input(Prompt(message="ABC"))
h.wait_on_input()
self.assertTrue(h.input_received())
def test_input_callback(self, input_mock, output_mock):
input_value = 'abc'
input_mock.return_value = input_value
h = InputHandler()
h.set_callback(self._test_callback)
h.get_input(Prompt(message="ABC"))
h.wait_on_input()
self.assertTrue(self._callback_called)
self.assertEqual(self._callback_input, input_value)
self.assertEqual(h.value, input_value)
def test_concurrent_input(self, input_mock, output_mock):
input_mock.side_effect = self._wait_for_concurrent_call
h = InputHandler()
h.set_callback(self._test_callback)
h2 = InputHandler()
h2.set_callback(self._test_callback2)
with self.assertRaisesRegex(KeyError, r'.*Input handler:.*InputHandler object'
r'.*Input requester: Unknown'
r'.*Input handler:.*InputHandler object'
r'.*Input requester: Unknown.*'):
h.get_input(Prompt(message="ABC"))
self._thread_event_wait_for_inner.wait()
h2.get_input(Prompt(message="ABC"))
self.assertFalse(self._callback_called)
self.assertEqual(self._callback_input, "")
self.assertEqual(h.value, None)
self.assertFalse(self._callback_called2)
self.assertEqual(self._callback_input2, "")
self.assertEqual(h2.value, None)
def test_concurrent_input_with_source(self, input_mock, output_mock):
input_mock.side_effect = self._wait_for_concurrent_call
source1 = Mock()
source2 = Mock()
h = InputHandler(source=source1)
h2 = InputHandler(source=source2)
h.set_callback(self._test_callback)
h2.set_callback(self._test_callback2)
with self.assertRaisesRegex(KeyError, r'.*Input handler:.*InputHandler object'
r'.*Input requester:.*Mock'
r'.*Input handler:.*InputHandler object'
r'.*Input requester:.*Mock.*'):
h.get_input(Prompt(message="ABC"))
self._thread_event_wait_for_inner.wait()
h2.get_input(Prompt(message="ABC"))
self.assertFalse(self._callback_called)
self.assertEqual(self._callback_input, "")
self.assertEqual(h.value, None)
self.assertFalse(self._callback_called2)
self.assertEqual(self._callback_input2, "")
self.assertEqual(h2.value, None)
def test_concurrent_input_without_check(self, input_mock, output_mock):
input_mock.side_effect = self._wait_for_concurrent_call
h = InputHandler()
h2 = InputHandler()
h.set_callback(self._test_callback)
h2.set_callback(self._test_callback2)
h2.skip_concurrency_check = True
h.skip_concurrency_check = True
h.get_input(Prompt(message="ABC"))
self._thread_event_wait_for_inner.wait()
h2.get_input(Prompt(message="ABC"))
self._thread_event_wait_for_outer.set()
h.wait_on_input()
h2.wait_on_input()
self.assertFalse(self._callback_called)
self.assertFalse(h.input_successful())
self.assertTrue(self._callback_called2)
self.assertTrue(h2.input_successful())
self.assertEqual(self._callback_input2, "thread 0")
self.assertEqual(h2.value, "thread 0")
def _wait_for_concurrent_call(self):
ret = "thread {}".format(len(self._threads))
self._threads.append(current_thread())
self._thread_event_wait_for_inner.set()
self._thread_event_wait_for_outer.wait()
return ret
def _test_callback(self, user_input):
self._callback_called = True
self._callback_input = user_input
def _test_callback2(self, user_input):
self._callback_called2 = True
self._callback_input2 = user_input
@patch('sys.stdout', new_callable=StringIO)
@patch('simpleline.global_configuration.GlobalConfiguration.password_function')
class PasswordInputHandler_TestCase(unittest.TestCase):
def create_loop(self):
self.loop = MainLoop()
def setUp(self):
super().setUp()
self.create_loop()
App.initialize(event_loop=self.loop)
self._callback_called = False
self._callback_input = ""
def test_async_input(self, input_mock, output_mock):
input_mock.return_value = 'a'
h = PasswordInputHandler()
h.get_input(Prompt(message="ABC"))
h.wait_on_input()
self.assertEqual(h.value, 'a')
def test_input_received(self, input_mock, output_mock):
input_mock.return_value = 'a'
h = PasswordInputHandler()
self.assertFalse(h.input_received())
h.get_input(Prompt(message="ABC"))
h.wait_on_input()
self.assertTrue(h.input_received())
def test_input_callback(self, input_mock, output_mock):
input_value = 'abc'
input_mock.return_value = input_value
h = PasswordInputHandler()
h.set_callback(self._test_callback)
h.get_input(Prompt(message="ABC"))
h.wait_on_input()
self.assertTrue(self._callback_called)
self.assertEqual(self._callback_input, input_value)
self.assertEqual(h.value, input_value)
def _test_callback(self, user_input):
self._callback_called = True
self._callback_input = user_input
@@ -0,0 +1,135 @@
# Prompt test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from simpleline.render.prompt import Prompt
class Prompt_TestCase(unittest.TestCase):
def test_prompt_message(self):
p = Prompt()
self.assertEqual(p.message, Prompt.DEFAULT_MESSAGE)
p.set_message("Brand new message")
self.assertEqual(p.message, "Brand new message")
p.set_message(u"Žluťoučký kůň")
self.assertEqual(p.message, u"Žluťoučký kůň")
p2 = Prompt("Default prompt text")
self.assertEqual(p2.message, "Default prompt text")
def test_add_options(self):
p = Prompt()
# add new option
p.add_option("R", "refresh")
self.assertTrue("R" in p.options)
self.assertEqual(p.options["R"], "refresh")
# add option over the existing option should trigger warning message
with self.assertLogs("simpleline", level="WARNING"):
p.add_option("R", "another refresh")
# update existing option
p.update_option("R", "new refresh option")
self.assertEqual(p.options["R"], "new refresh option")
# update non existing option should trigger warning
with self.assertLogs("simpleline", level="WARNING"):
p.update_option("N", "non existing option")
p = Prompt()
p.add_option("N", "new option")
# remove option
ret = p.remove_option("N")
self.assertFalse("N" in p.options)
self.assertEqual(ret, "new option")
# remove non-existing option
ret = p.remove_option("non-existing")
self.assertIsNone(ret)
def _check_default_option(self, prompt, key, value):
self.assertEqual(len(prompt.options), 1)
self.assertEqual(prompt.options[key], value)
def test_refresh_option(self):
# refresh option
p = Prompt()
p.add_refresh_option()
self._check_default_option(p, Prompt.REFRESH, Prompt.REFRESH_DESCRIPTION)
# test add with description
p = Prompt()
p.add_refresh_option("Other refresh")
self._check_default_option(p, Prompt.REFRESH, "Other refresh")
# change existing description
p.add_refresh_option("New refresh")
self._check_default_option(p, Prompt.REFRESH, "New refresh")
def test_continue_option(self):
# continue option
p = Prompt()
p.add_continue_option()
self._check_default_option(p, Prompt.CONTINUE, Prompt.CONTINUE_DESCRIPTION)
# test add with description
p = Prompt()
p.add_continue_option("Other continue")
self._check_default_option(p, Prompt.CONTINUE, "Other continue")
# change existing description
p.add_continue_option("New continue")
self._check_default_option(p, Prompt.CONTINUE, "New continue")
def test_quit_option(self):
# quit option
p = Prompt()
p.add_quit_option()
self._check_default_option(p, Prompt.QUIT, Prompt.QUIT_DESCRIPTION)
# test add with description
p = Prompt()
p.add_quit_option("Other quit")
self._check_default_option(p, Prompt.QUIT, "Other quit")
# change existing description
p.add_quit_option("New quit")
self._check_default_option(p, Prompt.QUIT, "New quit")
def test_help_option(self):
# help option
p = Prompt()
p.add_help_option()
self._check_default_option(p, Prompt.HELP, Prompt.HELP_DESCRIPTION)
# test add with description
p = Prompt()
p.add_help_option("Other help")
self._check_default_option(p, Prompt.HELP, "Other help")
# change existing description
p.add_help_option("New help")
self._check_default_option(p, Prompt.HELP, "New help")
@@ -0,0 +1,564 @@
# Rendering screen test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from io import StringIO
from unittest import mock
from simpleline import App
from simpleline.render import RenderUnexpectedError
from simpleline.render.screen import UIScreen, InputState
from .. import UtilityMixin
def _fake_input(queue_instance, prompt):
queue_instance.put("a")
@mock.patch('sys.stdout', new_callable=StringIO)
class SeparatorPrinting_TestCase(unittest.TestCase, UtilityMixin):
def setUp(self):
App.initialize()
def test_separator(self, stdout_mock):
ui_screen = EmptyScreen()
self.schedule_screen_and_run(ui_screen)
self.assertEqual(self.calculate_separator(), stdout_mock.getvalue())
def test_other_width_separator(self, stdout_mock):
ui_screen = EmptyScreen()
width = 60
App.get_configuration().width = width
App.get_scheduler().schedule_screen(ui_screen)
App.run()
self.assertEqual(self.calculate_separator(width), stdout_mock.getvalue())
def test_zero_width_no_separator(self, stdout_mock):
ui_screen = EmptyScreen()
width = 0
App.get_configuration().width = width
App.get_scheduler().schedule_screen(ui_screen)
App.run()
self.assertEqual("\n\n", stdout_mock.getvalue())
def test_no_separator_when_screen_setup_fails(self, stdout_mock):
ui_screen = TestScreenSetupFail()
App.initialize()
App.get_scheduler().schedule_screen(ui_screen)
App.run()
self.assertEqual("", stdout_mock.getvalue())
def test_no_separator(self, stdout_mock):
print_text = "testing"
screen = NoSeparatorScreen(print_text)
self.schedule_screen_and_run(screen)
self.schedule_screen_and_run(screen)
expected_output = print_text + "\n" + print_text + "\n"
self.assertEqual(expected_output, stdout_mock.getvalue())
class SimpleUIScreenFeatures_TestCase(unittest.TestCase):
def test_close_screen(self):
screen = UIScreen()
App.initialize()
App.get_scheduler().schedule_screen(screen)
# Program will quit in close_screen when stack is empty
App.get_scheduler().schedule_screen(UIScreen())
screen.close()
def test_close_screen_closed_from_other_source_error(self):
App.initialize()
App.get_scheduler().schedule_screen(UIScreen())
with self.assertRaises(RenderUnexpectedError):
App.get_scheduler().close_screen(closed_from=mock.MagicMock())
def test_failed_screen_setup(self):
screen = FailedSetupScreen()
App.initialize()
App.get_scheduler().schedule_screen(screen)
App.run()
@mock.patch('sys.stdout', new_callable=StringIO)
class SimpleUIScreenProcessing_TestCase(unittest.TestCase, UtilityMixin):
def setUp(self):
self._default_separator = self.calculate_separator(80)
def test_screen_event_loop_processing(self, _):
ui_screen = EmptyScreen()
self.schedule_screen_and_run(ui_screen)
self.assertTrue(ui_screen.is_closed)
def test_running_empty_loop(self, _):
App.initialize()
loop = App.get_event_loop()
loop.process_signals()
def test_screen_event_loop_processing_with_two_screens(self, _):
first_screen = EmptyScreen()
screen = EmptyScreen()
App.initialize()
App.get_scheduler().schedule_screen(first_screen)
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertTrue(first_screen)
self.assertTrue(screen)
def test_screen_title_rendering(self, stdout_mock):
screen = NoInputScreen()
screen.title = u"TestTitle"
self.schedule_screen_and_run(screen)
out = self._default_separator
out += "TestTitle\n\n"
self.assertEqual(stdout_mock.getvalue(), out)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('simpleline.event_loop.AbstractEventLoop.kill_app_with_traceback')
class ScreenException_TestCase(unittest.TestCase, UtilityMixin):
def setUp(self):
self._force_quit_called = False
# The original method calls sys.exit(1) so we don't need to test this functionality
def force_quit_mock(self, signal, data=None):
self._force_quit_called = True
loop = App.get_event_loop()
loop.force_quit()
def test_raise_exception_in_refresh(self, mock_kill_app, _):
screen = ExceptionTestScreen(ExceptionTestScreen.REFRESH)
mock_kill_app.side_effect = self.force_quit_mock
self.schedule_screen_and_run(screen)
self.assertTrue(self._force_quit_called)
def test_raise_exception_in_rendering(self, mock_kill_app, _):
screen = ExceptionTestScreen(ExceptionTestScreen.REDRAW)
mock_kill_app.side_effect = self.force_quit_mock
self.schedule_screen_and_run(screen)
self.assertTrue(self._force_quit_called)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('simpleline.input.input_handler.InputHandlerRequest._get_input')
class InputProcessing_TestCase(unittest.TestCase):
def setUp(self):
App.initialize()
def test_basic_input(self, input_mock, mock_stdout):
input_mock.return_value = "a"
screen = InputScreen()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertTrue(screen.input_processed)
def test_process_input_and_redraw(self, input_mock, mock_stdout):
input_mock.return_value = "a"
screen = InputStateRedrawScreen()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertTrue(screen.refreshed)
def test_process_input_and_close(self, input_mock, mock_stdout):
input_mock.return_value = "a"
screen = InputStateCloseScreen()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertTrue(screen.input_processed)
def test_quit_input(self, mock_stdin, mock_stdout):
mock_stdin.return_value = "q"
screen = UIScreen()
App.get_scheduler().schedule_screen(screen)
App.run()
def test_continue_input(self, mock_stdin, mock_stdout):
mock_stdin.return_value = "c"
screen = UIScreen()
screen2 = EmptyScreen()
App.get_scheduler().schedule_screen(screen)
App.get_scheduler().schedule_screen(screen2)
App.run()
self.assertTrue(screen.screen_ready)
self.assertTrue(screen.screen_ready)
def test_refresh_input(self, mock_stdin, mock_stdout):
mock_stdin.return_value = "r"
screen = RefreshTestScreen()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertTrue(screen.input_processed)
def test_refresh_on_input_error(self, mock_stdin, mock_stdout):
mock_stdin.return_value = "q"
threshold = 5
screen = InputErrorTestScreen(threshold)
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(screen.render_counter, 2)
self.assertEqual(screen.error_counter, threshold)
def test_multiple_refresh_on_input_error(self, mock_stdin, mock_stdout):
mock_stdin.return_value = "q"
threshold = 12
screen = InputErrorTestScreen(threshold)
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(screen.render_counter, 3)
self.assertEqual(screen.error_counter, threshold)
def test_no_refresh_when_prompt_is_none(self, mock_stdin, mock_stdout):
mock_stdin.return_value = "q"
threshold = 5
screen = InputErrorDynamicPromptTestScreen(threshold, not_return_prompt_on=3)
App.get_scheduler().schedule_screen(screen)
App.run()
# first draw and manual redraw when prompt is None
self.assertEqual(screen.render_counter, 2)
self.assertTrue(screen.input_skipped)
self.assertEqual(screen.error_counter, threshold)
def test_input_no_prompt(self, mock_stdin, mock_stdout):
screen = InputWithNoPrompt()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertTrue(screen.prompt_entered)
@mock.patch('simpleline.event_loop.main_loop.MainLoop.process_signals')
def test_custom_getpass(self, mock_stdin, mock_stdout, process_signals):
prompt = mock.MagicMock()
ret = "test"
screen = TestScreenWithPassFunc(prompt, ret)
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertTrue(screen.pass_called)
self.assertEqual(screen.pass_prompt.rstrip(), str(prompt))
def test_blocking_input(self, mock_stdin, mock_stdout):
prompt_message = "test prompt"
ret = "blocking test"
mock_stdin.return_value = ret
screen = BlockingInputTestScreen(prompt_message, False)
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(screen.input_returned, ret)
out = mock_stdout.getvalue()
out = out.split("\n")[-1].strip()
self.assertEqual(out, prompt_message)
@mock.patch('simpleline.global_configuration.GlobalConfiguration.password_function')
def test_blocking_password_input(self, mock_getpass, mock_stdin, mock_stdout):
prompt_message = "test prompt"
ret = "blocking test"
mock_getpass.return_value = ret
screen = BlockingInputTestScreen(prompt_message, True)
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(screen.input_returned, ret)
out = mock_stdout.getvalue()
# can't check for prompt because that is printed by getpass func which is mocked
self.assertGreater(len(out), 1)
# HELPER CLASSES
class EmptyScreen(UIScreen):
def __init__(self):
super().__init__()
self.is_closed = False
EmptyScreen.title = ""
self.input_required = False
def show_all(self):
self.close()
def closed(self):
self.is_closed = True
class TestScreenSetupFail(UIScreen):
def __init__(self):
super().__init__()
self.input_required = False
def setup(self, args):
super().setup(args)
return False
class TestScreenWithPassFunc(UIScreen):
def __init__(self, prompt, return_value):
super().__init__()
self.pass_prompt = ""
self.pass_called = False
self.password_func = self._test_getpass
self.hide_user_input = True
self._prompt = prompt
self._return_value = return_value
def prompt(self, args=None):
return self._prompt
def _test_getpass(self, prompt):
self.pass_prompt = prompt
self.pass_called = True
return self._return_value
def input(self, args, key):
return InputState.PROCESSED_AND_CLOSE
class InputErrorTestScreen(UIScreen):
def __init__(self, error_threshold=5):
super().__init__()
self.error_counter = 0
self.render_counter = 0
self._error_threshold = error_threshold
def input(self, args, key):
if self.error_counter == self._error_threshold:
# let "q" propagate to quit
return key
self.error_counter += 1
return InputState.DISCARDED
def show_all(self):
self.render_counter += 1
class InputErrorDynamicPromptTestScreen(InputErrorTestScreen):
def __init__(self, error_threshold=5, not_return_prompt_on=2):
super().__init__(error_threshold=error_threshold)
self._not_return_prompt_on = not_return_prompt_on
self.input_skipped = False
def prompt(self, args=None):
if self.error_counter == self._not_return_prompt_on and not self.input_skipped:
self.input_skipped = True
self.redraw()
return None
return super().prompt(args)
class InputWithNoPrompt(UIScreen):
def __init__(self):
super().__init__()
self.prompt_entered = False
self.input_required = True
def prompt(self, args=None):
self.prompt_entered = True
self.close()
# do not process input - it was processed here by user
return None
class RefreshTestScreen(UIScreen):
def __init__(self):
super().__init__()
self.input_processed = False
def input(self, args, key):
self.input_processed = True
return key
def refresh(self, args=None):
super().refresh(args)
if self.input_processed:
self.close()
self.input_required = False
return
class FailedSetupScreen(UIScreen):
def setup(self, args):
super().setup(args)
return False
class NoSeparatorScreen(UIScreen):
def __init__(self, print_string):
super().__init__()
self.input_required = False
self.no_separator = True
self._print_string = print_string
def show_all(self):
print(self._print_string)
self.close()
class NoInputScreen(UIScreen):
def __init__(self):
super().__init__()
self.input_required = False
def show_all(self):
super().show_all()
self.close()
class InputScreen(UIScreen):
def __init__(self):
super().__init__()
self.input_processed = False
def input(self, args, key):
if key == "a":
self.input_processed = True
self.close()
return InputState.PROCESSED
class InputStateCloseScreen(UIScreen):
def __init__(self):
super().__init__()
self.input_processed = False
def input(self, args, key):
self.input_processed = not self.input_processed
return InputState.PROCESSED_AND_CLOSE
class InputStateRedrawScreen(UIScreen):
def __init__(self):
super().__init__()
self._refreshing = False
self.refreshed = False
def input(self, args, key):
if not self._refreshing:
self._refreshing = True
return InputState.PROCESSED_AND_REDRAW
self.refreshed = True
return InputState.PROCESSED_AND_CLOSE
class ExceptionTestScreen(UIScreen):
"""Raising an exception in some place of processing."""
REFRESH = 0
REDRAW = 1
def __init__(self, where):
super().__init__()
self._where = where
self.input_required = False
def refresh(self, args=None):
super().refresh()
if self._where == self.REFRESH:
raise TestRefreshException("Refresh test exception happened!")
def show_all(self):
super().show_all()
if self._where == self.REDRAW:
raise TestRedrawException("Redraw test exception happened!")
class BlockingInputTestScreen(EmptyScreen):
def __init__(self, prompt_message, hidden):
super().__init__()
self._prompt_message = prompt_message
self._hidden = hidden
self.input_returned = None
def show_all(self):
self.input_returned = self.get_user_input(self._prompt_message, self._hidden)
super().show_all()
class TestRefreshException(Exception):
pass
class TestRedrawException(Exception):
pass
@@ -0,0 +1,181 @@
# Screen scheduler test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
# Author(s): Jiri Konecny <jkonecny@redhat.com>
#
import unittest
from unittest import mock
from simpleline.event_loop.main_loop import MainLoop
from simpleline.render.screen import UIScreen
from simpleline.render.screen_scheduler import ScreenScheduler
from simpleline.render.screen_stack import ScreenStack, ScreenStackEmptyException
class Scheduler_TestCase(unittest.TestCase):
def setUp(self):
self.stack = None
self.scheduler = None
def create_scheduler_with_stack(self):
self.stack = ScreenStack()
self.scheduler = ScreenScheduler(event_loop=mock.MagicMock(), scheduler_stack=self.stack)
def pop_last_item(self, remove=True):
return self.stack.pop(remove)
def test_create_scheduler(self):
scheduler = ScreenScheduler(MainLoop())
self.assertTrue(isinstance(scheduler._screen_stack, ScreenStack)) # pylint: disable=protected-access
def test_scheduler_quit_screen(self):
def test_callback():
pass
scheduler = ScreenScheduler(MainLoop())
self.assertEqual(scheduler.quit_screen, None)
scheduler.quit_screen = test_callback
self.assertEqual(scheduler.quit_screen, test_callback)
def test_nothing_to_render(self):
self.create_scheduler_with_stack()
self.assertTrue(self.scheduler.nothing_to_render)
self.assertTrue(self.stack.empty())
self.scheduler.schedule_screen(UIScreen())
self.assertFalse(self.scheduler.nothing_to_render)
self.assertFalse(self.stack.empty())
def test_schedule_screen(self):
self.create_scheduler_with_stack()
screen = UIScreen()
self.scheduler.schedule_screen(screen)
test_screen = self.pop_last_item(False)
self.assertEqual(test_screen.ui_screen, screen)
self.assertEqual(test_screen.args, None) # empty field - no arguments
self.assertFalse(test_screen.execute_new_loop)
# Schedule another screen, new one will be added to the bottom of the stack
new_screen = UIScreen()
self.scheduler.schedule_screen(new_screen)
# Here should still be the old screen
self.assertEqual(self.pop_last_item().ui_screen, screen)
# After removing the first we would find the second screen
self.assertEqual(self.pop_last_item().ui_screen, new_screen)
def test_replace_screen_with_empty_stack(self):
self.create_scheduler_with_stack()
with self.assertRaises(ScreenStackEmptyException):
self.scheduler.replace_screen(UIScreen())
def test_replace_screen(self):
self.create_scheduler_with_stack()
old_screen = UIScreen()
screen = UIScreen()
self.scheduler.schedule_screen(old_screen)
self.scheduler.replace_screen(screen)
self.assertEqual(self.pop_last_item(False).ui_screen, screen)
new_screen = UIScreen()
self.scheduler.replace_screen(new_screen)
self.assertEqual(self.pop_last_item().ui_screen, new_screen)
# The old_screen was replaced so the stack is empty now
self.assertTrue(self.stack.empty())
def test_replace_screen_with_args(self):
self.create_scheduler_with_stack()
old_screen = UIScreen()
screen = UIScreen()
self.scheduler.schedule_screen(old_screen)
self.scheduler.replace_screen(screen, "test")
test_screen = self.pop_last_item()
self.assertEqual(test_screen.ui_screen, screen)
self.assertEqual(test_screen.args, "test")
# The old_screen was replaced so the stack is empty now
self.assertTrue(self.stack.empty())
def test_switch_screen_with_empty_stack(self):
self.create_scheduler_with_stack()
screen = UIScreen()
self.scheduler.push_screen(screen)
self.assertEqual(self.pop_last_item().ui_screen, screen)
def test_switch_screen(self):
self.create_scheduler_with_stack()
screen = UIScreen()
new_screen = UIScreen()
self.scheduler.schedule_screen(screen)
self.scheduler.push_screen(new_screen)
test_screen = self.pop_last_item()
self.assertEqual(test_screen.ui_screen, new_screen)
self.assertEqual(test_screen.args, None)
self.assertEqual(test_screen.execute_new_loop, False)
# We popped the new_screen so the old screen should stay here
self.assertEqual(self.pop_last_item().ui_screen, screen)
self.assertTrue(self.stack.empty())
def test_switch_screen_with_args(self):
self.create_scheduler_with_stack()
screen = UIScreen()
self.scheduler.push_screen(screen, args="test")
self.assertEqual(self.pop_last_item(False).ui_screen, screen)
self.assertEqual(self.pop_last_item().args, "test")
@mock.patch('simpleline.render.screen_scheduler.ScreenScheduler._draw_screen')
def test_switch_screen_modal_empty_stack(self, _):
self.create_scheduler_with_stack()
screen = UIScreen()
self.scheduler.push_screen_modal(screen)
self.assertEqual(self.pop_last_item().ui_screen, screen)
@mock.patch('simpleline.render.screen_scheduler.ScreenScheduler._draw_screen')
def test_switch_screen_modal(self, _):
self.create_scheduler_with_stack()
screen = UIScreen()
new_screen = UIScreen()
self.scheduler.schedule_screen(screen)
self.scheduler.push_screen_modal(new_screen)
test_screen = self.pop_last_item()
self.assertEqual(test_screen.ui_screen, new_screen)
self.assertEqual(test_screen.args, None)
self.assertEqual(test_screen.execute_new_loop, True)
@mock.patch('simpleline.render.screen_scheduler.ScreenScheduler._draw_screen')
def test_switch_screen_modal_with_args(self, _):
self.create_scheduler_with_stack()
screen = UIScreen()
self.scheduler.push_screen_modal(screen, args="test")
self.assertEqual(self.pop_last_item(False).ui_screen, screen)
@@ -0,0 +1,207 @@
# Screen scheduling test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from io import StringIO
from unittest import mock
from simpleline.render.screen import UIScreen
from simpleline.render.screen_handler import ScreenHandler
from .. import UtilityMixin
@mock.patch('sys.stdout', new_callable=StringIO)
class ScreenScheduler_TestCase(unittest.TestCase, UtilityMixin):
def test_replace_screen(self, _):
replace_screen = ShowedCounterScreen()
screen = ShowedCounterScreen(replace_screen=replace_screen)
self.schedule_screen_and_run(screen)
self.assertEqual(screen.counter, 1)
self.assertEqual(replace_screen.counter, 1)
def test_switch_screen(self, _):
switched_screen = ShowedCounterScreen()
screen = ShowedCounterScreen(switched_screen)
self.schedule_screen_and_run(screen)
self.assertEqual(screen.counter, 2)
self.assertEqual(switched_screen.counter, 1)
def test_switch_screen_modal_in_render(self, _):
modal_screen = ModalTestScreen()
screen = ModalTestScreen(modal_screen_render=modal_screen)
self.schedule_screen_and_run(screen)
self.assertEqual(screen.copied_modal_counter, ModalTestScreen.AFTER_MODAL_RENDER)
self.assertEqual(modal_screen.copied_modal_counter, ModalTestScreen.BEFORE_MODAL_RENDER)
def test_switch_screen_modal_in_refresh(self, _):
modal_screen = ModalTestScreen()
screen = ModalTestScreen(modal_screen_refresh=modal_screen)
self.schedule_screen_and_run(screen)
self.assertEqual(screen.copied_modal_counter, ModalTestScreen.AFTER_MODAL_REFRESH)
self.assertEqual(modal_screen.copied_modal_counter, ModalTestScreen.BEFORE_MODAL_REFRESH)
def test_switch_screen_modal_refresh_and_render(self, _):
modal_refresh = ModalTestScreen()
modal_render = ModalTestScreen()
screen = ModalTestScreen(modal_screen_refresh=modal_refresh,
modal_screen_render=modal_render)
self.schedule_screen_and_run(screen)
self.assertEqual(screen.copied_modal_counter, ModalTestScreen.AFTER_MODAL_RENDER)
self.assertEqual(modal_refresh.copied_modal_counter, ModalTestScreen.BEFORE_MODAL_REFRESH)
self.assertEqual(modal_render.copied_modal_counter, ModalTestScreen.BEFORE_MODAL_RENDER)
def test_switch_screen_modal_render_recursive(self, _):
modal_render_inner = ModalTestScreen()
modal_render_outer = ModalTestScreen(modal_screen_render=modal_render_inner)
screen = ModalTestScreen(modal_screen_render=modal_render_outer)
self.schedule_screen_and_run(screen)
self.assertEqual(screen.copied_modal_counter, ModalTestScreen.AFTER_MODAL_RENDER)
# outer modal screen has AFTER_MODAL_RENDER because it was set before by inner loop
self.assertEqual(modal_render_outer.copied_modal_counter,
ModalTestScreen.AFTER_MODAL_RENDER)
self.assertEqual(modal_render_inner.copied_modal_counter,
ModalTestScreen.BEFORE_MODAL_RENDER)
@mock.patch('simpleline.input.input_handler.InputHandlerRequest._get_input')
def test_switch_screen_modal_input_order(self, mock_input, mock_stdout):
modal_screen = InputAndDrawScreen("Modal")
parent_screen = EmitDrawThenCreateModal(modal_screen, msg="Parent")
mock_input.return_value = "c"
expected = ["Modal", # modal needs to be printed first
"Parent", # draw enqueued draw signal -- manually registered in refresh()
"Parent"] # draw because modal screen was closed
self.schedule_screen_and_run(parent_screen)
self.maxDiff = None
self.assertEqual(self.create_output_with_separators(expected), mock_stdout.getvalue())
class ShowedCounterScreen(UIScreen):
def __init__(self, switch_to_screen=None, replace_screen=None):
super().__init__()
self._switch_to_screen = switch_to_screen
self._replace_screen = replace_screen
self.counter = 0
self.input_required = False
def show_all(self):
super().show_all()
self.counter += 1
if self._switch_to_screen is not None:
ScreenHandler.push_screen(self._switch_to_screen)
self._switch_to_screen = None
elif self._replace_screen is not None:
ScreenHandler.replace_screen(self._replace_screen)
self._replace_screen = None
else:
self.close()
class ModalTestScreen(UIScreen):
"""Test if the modal screen is really modal and stops the execution in a place where
we start the modal screen.
This class have checkpoints which increment class variable counter. This counter is
copied in the modal instance to the local variable self.copied_modal_counter.
In the end we should check if the instance modal counter have the correct value, which is
before the modal screen was started (1).
"""
INIT = 0
BEFORE_MODAL_REFRESH = 1
AFTER_MODAL_REFRESH = 2
BEFORE_MODAL_RENDER = 3
AFTER_MODAL_RENDER = 4
modal_counter = 0
def __init__(self, modal_screen_render=None, modal_screen_refresh=None):
super().__init__()
self._modal_screen_render = modal_screen_render
self._modal_screen_refresh = modal_screen_refresh
self.copied_modal_counter = 0
self.input_required = False
ModalTestScreen.modal_counter = self.INIT
def refresh(self, args=None):
super().refresh(args)
if self._modal_screen_refresh is not None:
# Start a new modal screen
ModalTestScreen.modal_counter = self.BEFORE_MODAL_REFRESH
ScreenHandler.push_screen_modal(self._modal_screen_refresh)
ModalTestScreen.modal_counter = self.AFTER_MODAL_REFRESH
def show_all(self):
super().show_all()
if self._modal_screen_render is not None:
# Start new modal screen
ModalTestScreen.modal_counter = self.BEFORE_MODAL_RENDER
ScreenHandler.push_screen_modal(self._modal_screen_render)
ModalTestScreen.modal_counter = self.AFTER_MODAL_RENDER
self.copied_modal_counter = ModalTestScreen.modal_counter
self.close()
class EmitDrawThenCreateModal(UIScreen):
def __init__(self, refresh_screen, msg):
super().__init__()
self._refresh_screen = refresh_screen
self.title = msg
self.input_required = False
def refresh(self, args=None):
super().refresh(args)
if self._refresh_screen:
self.redraw()
ScreenHandler.push_screen_modal(self._refresh_screen)
self._refresh_screen = None
else:
self.close()
class InputAndDrawScreen(UIScreen):
def __init__(self, msg):
super().__init__()
self.title = msg
self.input_required = False
def refresh(self, args=None):
super().refresh(args)
self.close()
@@ -0,0 +1,162 @@
# Screen stack test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
# Author(s): Jiri Konecny <jkonecny@redhat.com>
#
import unittest
from simpleline.render.screen import UIScreen
from simpleline.render.screen_stack import ScreenStack, ScreenData, ScreenStackEmptyException
class ScreenStack_TestCase(unittest.TestCase):
def test_append_screen(self):
stack = ScreenStack()
stack.append(ScreenData(None))
def test_is_empty(self):
stack = ScreenStack()
self.assertTrue(stack.empty())
stack.append(ScreenData(None))
self.assertFalse(stack.empty())
def test_pop(self):
stack = ScreenStack()
with self.assertRaises(ScreenStackEmptyException):
stack.pop()
with self.assertRaises(ScreenStackEmptyException):
stack.pop(False)
# stack.pop(True) will remove the item
stack.append(ScreenData(None))
stack.pop(True)
with self.assertRaises(ScreenStackEmptyException):
stack.pop()
# stack.pop() should behave the same as stack.pop(True)
stack.append(ScreenData(None))
stack.pop()
with self.assertRaises(ScreenStackEmptyException):
stack.pop()
stack.append(ScreenData(None))
stack.pop(False)
stack.pop(True)
def test_add_first(self):
stack = ScreenStack()
screen_data = ScreenData(None)
stack.add_first(screen_data)
self.assertEqual(stack.pop(False), screen_data)
# Add new Screen data to the end
new_screen_data = ScreenData(None)
stack.add_first(new_screen_data)
# First the old screen data should be there
self.assertEqual(stack.pop(), screen_data)
# Second should be the new screen data
self.assertEqual(stack.pop(), new_screen_data)
def test_size(self):
stack = ScreenStack()
self.assertEqual(stack.size(), 0)
stack.append(ScreenData(None))
self.assertEqual(stack.size(), 1)
stack.append(ScreenData(None))
self.assertEqual(stack.size(), 2)
# Remove from stack
stack.pop()
self.assertEqual(stack.size(), 1)
stack.pop()
self.assertEqual(stack.size(), 0)
# Add first when stack has items
stack.append(ScreenData(None))
stack.append(ScreenData(None))
self.assertEqual(stack.size(), 2)
stack.add_first(ScreenData(None))
self.assertEqual(stack.size(), 3)
def test_stack_dump(self):
stack = ScreenStack()
stack.append(ScreenData(TestScreen1()))
stack.append(ScreenData(TestScreen2()))
dump = stack.dump_stack()
dump = dump.replace('\n', '')
self.assertRegex(dump, r"TestScreen2.*TestScreen1")
class ScreenData_TestCase(unittest.TestCase):
def setUp(self):
self.ui_screen = None
def _prepare(self):
self.ui_screen = UIScreen()
def _screen_check(self, test_screen, ui_screen, args, execute_new_loop):
self._prepare()
self.assertEqual(test_screen.ui_screen, ui_screen)
self.assertEqual(test_screen.args, args)
self.assertEqual(test_screen.execute_new_loop, execute_new_loop)
def test_screen_data(self):
self._prepare()
screen = ScreenData(self.ui_screen)
self._screen_check(screen, self.ui_screen, None, False)
def test_screen_data_with_args(self):
self._prepare()
screen = ScreenData(ui_screen=self.ui_screen, args=1)
self._screen_check(screen, self.ui_screen, 1, False)
array = [2, "a"]
screen2 = ScreenData(ui_screen=self.ui_screen, args=array)
self._screen_check(screen2, self.ui_screen, array, False)
def test_screen_data_with_execute_loop(self):
self._prepare()
screen = ScreenData(self.ui_screen, execute_new_loop=True)
self._screen_check(screen, self.ui_screen, None, True)
screen2 = ScreenData(self.ui_screen, execute_new_loop=False)
self._screen_check(screen2, self.ui_screen, None, False)
def test_screen_data_with_args_and_execute_loop(self):
self._prepare()
screen = ScreenData(self.ui_screen, "test", True)
self._screen_check(screen, self.ui_screen, "test", True)
class TestScreen1(UIScreen):
pass
class TestScreen2(UIScreen):
pass
@@ -0,0 +1,113 @@
# Signal handler test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from unittest.mock import patch, MagicMock
from simpleline import App
from simpleline.event_loop.signals import RenderScreenSignal, AbstractSignal
from simpleline.render.screen import UIScreen
class SignalHandler_TestCase(unittest.TestCase):
def setUp(self):
self.callback_called = False
self.priority = 0
def _callback(self, signal, data):
self.callback_called = True
signal.test_attribute = True
def test_basic_connect(self):
connect_screen = UIScreen()
App.initialize(scheduler=MagicMock())
connect_screen.connect(TestSignal, self._callback)
App.get_event_loop().enqueue_signal(TestSignal(self))
App.get_event_loop().process_signals()
self.assertTrue(self.callback_called)
def test_create_signal(self):
connect_screen = UIScreen()
App.initialize(scheduler=MagicMock())
signal = connect_screen.create_signal(TestSignal, priority=20)
self.assertEqual(signal.priority, 20)
self.assertTrue(isinstance(signal, TestSignal))
# source is set by create_signal
self.assertEqual(signal.source, connect_screen)
def test_emit(self):
connect_screen = UIScreen()
App.initialize(scheduler=MagicMock())
connect_screen.connect(TestSignal, self._callback)
connect_screen.emit(TestSignal(self))
App.get_event_loop().process_signals()
self.assertTrue(self.callback_called)
@patch('sys.stdout')
def test_connect_react_on_rendering(self, _):
connect_test_screen = TestRenderConnectHandler()
screen2 = EmptyScreen()
App.initialize()
App.get_scheduler().schedule_screen(connect_test_screen)
App.get_scheduler().schedule_screen(screen2)
App.run()
self.assertTrue(connect_test_screen.callback_called)
class TestRenderConnectHandler(UIScreen):
def __init__(self):
super().__init__()
self.callback_called = False
self.input_required = False
def refresh(self, args=None):
super().refresh(args)
self.connect(RenderScreenSignal, self._callback)
self.close()
def _callback(self, signal, args):
self.callback_called = True
class EmptyScreen(UIScreen):
def __init__(self):
super().__init__()
self.input_required = False
def show_all(self):
super().show_all()
self.close()
class TestSignal(AbstractSignal):
pass
@@ -0,0 +1,106 @@
# Widgets test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
from unittest import TestCase
from simpleline.event_loop.ticket_machine import TicketMachine
class TicketMachine_TestCase(TestCase):
def setUp(self):
self._tickets = TicketMachine()
def test_take_ticket(self):
line_id = 0
t = self._tickets.take_ticket(line_id)
self.assertEqual(t, 0)
t2 = self._tickets.take_ticket(line_id)
self.assertNotEqual(t, t2)
def test_check_ticket(self):
line_id = 0
t = self._tickets.take_ticket(line_id)
self.assertFalse(self._tickets.check_ticket(line_id, t))
self._tickets.mark_line_to_go(line_id)
self.assertTrue(self._tickets.check_ticket(line_id, t))
def test_mark_multiple_tickets(self):
line_id = 0
t1 = self._tickets.take_ticket(line_id)
t2 = self._tickets.take_ticket(line_id)
t3 = self._tickets.take_ticket(line_id)
t4 = self._tickets.take_ticket(line_id)
self._tickets.mark_line_to_go(line_id)
self.assertTrue(self._tickets.check_ticket(line_id, t1))
self.assertTrue(self._tickets.check_ticket(line_id, t2))
self.assertTrue(self._tickets.check_ticket(line_id, t3))
self.assertTrue(self._tickets.check_ticket(line_id, t4))
def test_mark_one_of_lines(self):
line_id1 = "a"
line_id2 = "b"
t1 = self._tickets.take_ticket(line_id1)
t2 = self._tickets.take_ticket(line_id1)
t3 = self._tickets.take_ticket(line_id2)
t4 = self._tickets.take_ticket(line_id2)
self._tickets.mark_line_to_go(line_id1)
self.assertTrue(self._tickets.check_ticket(line_id1, t1))
self.assertTrue(self._tickets.check_ticket(line_id1, t2))
self.assertFalse(self._tickets.check_ticket(line_id2, t3))
self.assertFalse(self._tickets.check_ticket(line_id2, t4))
def text_check_re_using(self):
line_id = "a"
t1 = self._tickets.take_ticket(line_id)
t2 = self._tickets.take_ticket(line_id)
t3 = self._tickets.take_ticket(line_id)
self._tickets.mark_line_to_go(line_id)
self.assertTrue(self._tickets.check_ticket(line_id, t1))
self.assertTrue(self._tickets.check_ticket(line_id, t2))
# it needs to be False when you check it again
self.assertFalse(self._tickets.check_ticket(line_id, t1))
self.assertFalse(self._tickets.check_ticket(line_id, t2))
# take new ticket and mark the line again
t4 = self._tickets.take_ticket(line_id)
self._tickets.mark_line_to_go(line_id)
# old checked tickets should be invalid now
self.assertFalse(self._tickets.check_ticket(line_id, t1))
self.assertFalse(self._tickets.check_ticket(line_id, t2))
# old not checked ticket should work
self.assertTrue(self._tickets.check_ticket(line_id, t3))
# new tickets should work
self.assertTrue(self._tickets.check_ticket(line_id, t4))
@@ -0,0 +1,335 @@
# Widgets test classes.
#
# This file is part of Simpleline Text UI library.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Simpleline is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Simpleline is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Simpleline. If not, see <https://www.gnu.org/licenses/>.
#
import unittest
from io import StringIO
from unittest.mock import patch
from simpleline import App
from simpleline.render.prompt import Prompt
from simpleline.render.screen import UIScreen
from simpleline.render.widgets import TextWidget, SeparatorWidget, CheckboxWidget, CenterWidget, \
ColumnWidget, EntryWidget
class BaseWidgets_TestCase(unittest.TestCase):
"""Base class containing helper functions."""
def setUp(self):
self.w1 = TextWidget(u"Můj krásný dlouhý text")
self.w2 = TextWidget(u"Test")
self.w3 = TextWidget(u"Test 2")
self.w4 = TextWidget(u"Krásný dlouhý text podruhé")
self.w5 = TextWidget(u"Test 3")
self.w6 = TextWidget("The rescue environment will now attempt "
"to find your Linux installation and mount it under "
"the directory : bla. You can then make any changes "
"required to your system. Choose '1' to proceed with "
"this step.\nYou can choose to mount your file "
"systems read-only instead of read-write by choosing "
"'2'.\nIf for some reason this process does not work "
"choose '3' to skip directly to a shell.\n\n")
self.w7 = TextWidget("Wrapping toooooooooooooooooooooooooooooooooooooooooooo"
"oooooooooooooooooooooooooooooooooooooooooooooooooooooo long word.")
self.w8 = TextWidget("Text that would be wrapped exactly at the screen width should"
" have special test. This one.")
def evaluate_result(self, test_result, expected_result):
self.assertEqual(len(test_result), len(expected_result))
for i in range(0, len(test_result)): # pylint: disable=consider-using-enumerate
self.assertEqual(test_result[i], expected_result[i])
class Widgets_TestCase(BaseWidgets_TestCase):
def test_separator_widget(self):
w = SeparatorWidget()
w.render(80)
res_lines = w.get_lines()
expected_result = [u""]
self.evaluate_result(res_lines, expected_result)
def test_separator_widget_multiline(self):
w = SeparatorWidget(3)
w.render(80)
res_lines = w.get_lines()
expected_result = [u"",
u"",
u""]
self.evaluate_result(res_lines, expected_result)
def test_column_widget(self):
# Test column text
c = ColumnWidget([(20, [self.w1, self.w2, self.w3]),
(25, [self.w4, self.w5]),
(15, [self.w1, self.w2, self.w3])], spacing=3)
c.render(80)
res_lines = c.get_lines()
expected_result = [u"Můj krásný dlouhý Krásný dlouhý text Můj krásný",
u"text podruhé dlouhý text",
u"Test Test 3 Test",
u"Test 2 Test 2"]
self.evaluate_result(res_lines, expected_result)
def test_column_wrapping(self):
# Test column wrapping text
c = ColumnWidget([(15, [self.w1, self.w2, self.w3]), (10, [self.w4, self.w5])], spacing=1)
c.render(80)
expected_result = [u"Můj krásný Krásný",
u"dlouhý text dlouhý",
u"Test text",
u"Test 2 podruhé",
u" Test 3"]
res_lines = c.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_multiline_text(self):
self.w6.render(80)
expected_result = [
"The rescue environment will now attempt to find your Linux installation and",
"mount it under the directory : bla. You can then make any changes required to",
"your system. Choose '1' to proceed with this step.",
"You can choose to mount your file systems read-only instead of read-write by",
"choosing '2'.",
"If for some reason this process does not work choose '3' to skip directly to a",
"shell.",
"",
""]
res_lines = self.w6.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_wrapping(self):
# wrap long text
self.w7.render(80)
expected_result = [
"Wrapping toooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo",
"oooooooooooooooooooooooooooo long word."]
res_lines = self.w7.get_lines()
self.assertEqual(len(res_lines), len(expected_result))
for i in range(0, len(res_lines)): # pylint: disable=consider-using-enumerate
self.assertEqual(res_lines[i], expected_result[i])
self.w8.render(80)
# pylint: disable=line-too-long
expected_result = ["Text that would be wrapped exactly at the screen width should have special test.",
"This one."]
res_lines = self.w8.get_lines()
self.evaluate_result(res_lines, expected_result)
def test_checkbox(self):
checkbox = CheckboxWidget(title="Test Title", text="Description")
checkbox.render(80)
expected_result = [u"[ ] Test Title",
u" (Description)"]
self.evaluate_result(checkbox.get_lines(), expected_result)
def test_completed_checkbox(self):
checkbox = CheckboxWidget(title="Title", text="Description", completed=True)
checkbox.render(80)
expected_result = [u"[x] Title",
u" (Description)"]
self.evaluate_result(checkbox.get_lines(), expected_result)
def test_key_checkbox(self):
checkbox = CheckboxWidget(key="o", title="Title", text="Description", completed=True)
checkbox.render(80)
expected_result = [u"[o] Title",
u" (Description)"]
self.evaluate_result(checkbox.get_lines(), expected_result)
def test_empty_checkbox(self):
checkbox = CheckboxWidget()
checkbox.render(80)
expected_result = [u"[ ]"]
self.evaluate_result(checkbox.get_lines(), expected_result)
def test_checkbox_wrapping(self):
checkbox = CheckboxWidget(title="Title", text="Testing\nwrapping")
checkbox.render(80)
expected_result = [u"[ ] Title",
u" (Testing",
u" wrapping)"]
self.evaluate_result(checkbox.get_lines(), expected_result)
def test_center_widget(self):
w = CenterWidget(self.w2)
w.render(10)
expected_result = [u" Test"]
self.evaluate_result(w.get_lines(), expected_result)
def test_entry_widget(self):
title = "Title"
value = "Value"
w = EntryWidget(title=title, value=value)
w.render(30)
expected_result = [title,
value]
self.evaluate_result(w.get_lines(), expected_result)
def test_entry_too_long(self):
title = "Title too long"
value = "Value also too long"
w = EntryWidget(title=title, value=value)
w.render(10)
expected_result = [u"Title too",
u"long",
u"Value also",
u"too long"]
self.evaluate_result(w.get_lines(), expected_result)
def test_entry_value_empty(self):
title = "Title"
w = EntryWidget(title=title)
w.render(20)
expected_result = [title]
self.evaluate_result(w.get_lines(), expected_result=expected_result)
@patch('simpleline.input.input_handler.InputHandlerRequest._get_input')
@patch('sys.stdout', new_callable=StringIO)
class WidgetProcessing_TestCase(unittest.TestCase):
@staticmethod
def _calculate_spacer():
# this calculation is taken from scheduler for default width '80'
return '\n'.join(2 * [80 * '='])
def _expected_output(self, text, widget_height=20):
# two lines are always added to the printed size
prompt_height = 2
real_widget_height = widget_height - prompt_height
lines = text.split('\n')
# add Press ENTER... to the text
if len(lines) - 1 >= real_widget_height:
lines.insert(real_widget_height, "\nPress %s to continue: \n" % Prompt.ENTER)
msg = self._calculate_spacer() + '\n'
msg += "\n".join(lines)
msg += "\n"
return msg
def test_draw_simple_widget(self, out_mock, in_mock):
widget_text = "Test"
screen = ScreenWithWidget(widget_text)
App.initialize()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(self._expected_output(widget_text), out_mock.getvalue())
def test_widget_multiline(self, out_mock, in_mock):
widget_text = "Testing output\n\n\nAgain..."
screen = ScreenWithWidget(widget_text)
App.initialize()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(self._expected_output(widget_text), out_mock.getvalue())
def test_widget_too_high(self, out_mock, in_mock):
in_mock.return_value = "\n"
in_mock.side_effect = lambda: print('\n')
widget_text = ("Line\n"
"Line2\n"
"Line3\n"
"Line4\n"
"Line5")
# Screen height take into account also 2 lines for prompt
screen = ScreenWithWidget(widget_text, height=6)
App.initialize()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(self._expected_output(widget_text, widget_height=6), out_mock.getvalue())
def test_widget_is_exactly_height_to_print(self, out_mock, in_mock):
widget_text = ("Line\n"
"Line2\n"
"Line3\n"
"Line4")
# Screen height take into account also 2 lines for prompt
screen = ScreenWithWidget(widget_text, height=6)
App.initialize()
App.get_scheduler().schedule_screen(screen)
App.run()
self.assertEqual(self._expected_output(widget_text, widget_height=6), out_mock.getvalue())
class ScreenWithWidget(UIScreen):
def __init__(self, msg, height=25):
super().__init__(screen_height=height)
self._msg = msg
self.input_required = False
def refresh(self, args=None):
super().refresh(args)
self.window.add(TextWidget(self._msg))
def show_all(self):
super().show_all()
self.close()