Merge ~sylvain-pineau/checkbox-ng:remote-interact-fixes into checkbox-ng:master

[Sylvain Pineau]

Multiple fixes to support all interactions type for manual jobs (skip, rerun).

Fix user interact types jobs where command output was not streamed resulting in a re-connection after 30s.

Unmerged commits

32b2d59... by Sylvain Pineau on 2019-10-29

remote:sa: pep8

05688f1... by Sylvain Pineau on 2019-10-29

remote:master: extend the socket timeout to 60s

37b1d3b... by Sylvain Pineau on 2019-10-29

remote: Fix user-interact jobs command not streamed and rerun interactions

- Properly catch ReRunJob exception on master side
- Always yield something from the run_job generator
- on master keep looping until next_job is False
- skip results are now handled on master side (interaction.kind == 'skip')

b2083cf... by Sylvain Pineau on 2019-10-01

session:remote_assistant: Only sleep 0.01s to stream faster

8fa0e77... by Sylvain Pineau on 2019-10-01

session:assistant: Ensure all jobs can write to CHECKBOX_DATA

21efbb1... by Sylvain Pineau on 2019-10-01

execution: Ensure all jobs can write to CHECKBOX_DATA

2f78a4f... by Sylvain Pineau on 2019-10-01

master: Add a --user to specify the target (i.e slave) normal user

c6bdeb3... by Sylvain Pineau on 2019-10-01

master: Hide connection info when running on localhost

e41ebe0... by Sylvain Pineau on 2019-10-01

master: Use getpass.getuser() to find the normal user when running on localhost

f897c6e... by Sylvain Pineau on 2019-09-25

execution: pep8

diff --git a/checkbox_ng/launcher/master.py b/checkbox_ng/launcher/master.py
index 7fb87f7..de46dc9 100644
--- a/checkbox_ng/launcher/master.py
+++ b/checkbox_ng/launcher/master.py
@@ -19,7 +19,9 @@
 This module contains implementation of the master end of the remote execution
 functionality.
 """
+import getpass
 import gettext
+import ipaddress
 import logging
 import os
 import select
@@ -43,7 +45,7 @@ from checkbox_ng.urwid_ui import CategoryBrowser
 from checkbox_ng.urwid_ui import ReRunBrowser
 from checkbox_ng.urwid_ui import interrupt_dialog
 from checkbox_ng.urwid_ui import resume_dialog
-from checkbox_ng.launcher.run import NormalUI
+from checkbox_ng.launcher.run import NormalUI, ReRunJob
 from checkbox_ng.launcher.stages import MainLoopStage
 from checkbox_ng.launcher.stages import ReportsStage
 _ = gettext.gettext
@@ -122,6 +124,7 @@ class RemoteMaster(Command, ReportsStage, MainLoopStage):
         self._is_bootstrapping = False
         self._target_host = ctx.args.host
         self._sudo_provider = None
+        self._normal_user = ''
         self.launcher = DefaultLauncherDefinition()
         if ctx.args.launcher:
             expanded_path = os.path.expanduser(ctx.args.launcher)
@@ -131,11 +134,18 @@ class RemoteMaster(Command, ReportsStage, MainLoopStage):
             with open(expanded_path, 'rt') as f:
                 self._launcher_text = f.read()
             self.launcher.read_string(self._launcher_text)
+        if 'local' in ctx.args.host:
+            ctx.args.host = '127.0.0.1'
+            if ipaddress.ip_address(ctx.args.host).is_loopback:
+                self._normal_user = getpass.getuser()
+        if ctx.args.user:
+            self._normal_user = ctx.args.user
         timeout = 600
         deadline = time.time() + timeout
         port = ctx.args.port
-        print(_("Connecting to {}:{}. Timeout: {}s").format(
-            ctx.args.host, port, timeout))
+        if not ipaddress.ip_address(ctx.args.host).is_loopback:
+            print(_("Connecting to {}:{}. Timeout: {}s").format(
+                ctx.args.host, port, timeout))
         while time.time() < deadline:
             try:
                 self.connect_and_run(ctx.args.host, port)
@@ -149,6 +159,7 @@ class RemoteMaster(Command, ReportsStage, MainLoopStage):
     def connect_and_run(self, host, port=18871):
         config = rpyc.core.protocol.DEFAULT_CONFIG.copy()
         config['allow_all_attrs'] = True
+        config['sync_request_timeout'] = 60
         keep_running = False
         self._prepare_transports()
         interrupted = False
@@ -217,6 +228,7 @@ class RemoteMaster(Command, ReportsStage, MainLoopStage):
         _logger.info("master: Starting new session.")
         configuration = dict()
         configuration['launcher'] = self._launcher_text
+        configuration['normal_user'] = self._normal_user
 
         tps = self.sa.start_session(configuration)
         _logger.debug("master: Session started. Available TPs:\n%s",
@@ -296,6 +308,8 @@ class RemoteMaster(Command, ReportsStage, MainLoopStage):
             "launcher definition file to use"))
         parser.add_argument('--port', type=int, default=18871, help=_(
             "port to connect to"))
+        parser.add_argument('-u', '--user', help=_(
+            "normal user to run non-root jobs"))
 
     def _handle_interrupt(self):
         """
@@ -478,37 +492,68 @@ class RemoteMaster(Command, ReportsStage, MainLoopStage):
             print(_("Category: {0}").format(job['category_name']))
             SimpleUI.horiz_line()
             next_job = False
-            for interaction in self.sa.run_job(job['id']):
-                if interaction.kind == 'sudo_input':
-                    self.sa.save_password(
-                        self._sudo_provider.encrypted_password)
-                if interaction.kind == 'purpose':
-                    SimpleUI.description(_('Purpose:'), interaction.message)
-                elif interaction.kind in ['description', 'steps']:
-                    SimpleUI.description(_('Steps:'), interaction.message)
-                    if job['command'] is None:
-                        cmd = 'run'
-                    else:
-                        cmd = SimpleUI(None).wait_for_interaction_prompt(None)
-                    if cmd == 'skip':
+            while next_job is False:
+                for interaction in self.sa.run_job(job['id']):
+                    if interaction.kind == 'sudo_input':
+                        self.sa.save_password(
+                            self._sudo_provider.encrypted_password)
+                    if interaction.kind == 'purpose':
+                        SimpleUI.description(_('Purpose:'),
+                                             interaction.message)
+                    elif interaction.kind == 'description':
+                        SimpleUI.description(_('Description:'),
+                                             interaction.message)
+                        if job['command'] is None:
+                            cmd = 'run'
+                        else:
+                            cmd = SimpleUI(
+                                None).wait_for_interaction_prompt(None)
+                        if cmd == 'skip':
+                            next_job = True
+                        self.sa.remember_users_response(cmd)
+                        self.wait_for_job(dont_finish=True)
+                    elif interaction.kind in 'steps':
+                        SimpleUI.description(_('Steps:'), interaction.message)
+                        if job['command'] is None:
+                            cmd = 'run'
+                        else:
+                            cmd = SimpleUI(
+                                None).wait_for_interaction_prompt(None)
+                        if cmd == 'skip':
+                            next_job = True
+                        self.sa.remember_users_response(cmd)
+                    elif interaction.kind == 'verification':
+                        self.wait_for_job(dont_finish=True)
+                        if interaction.message:
+                            SimpleUI.description(
+                                _('Verification:'), interaction.message)
+                        JobAdapter = namedtuple('job_adapter', ['command'])
+                        job_lite = JobAdapter(job['command'])
+                        try:
+                            cmd = SimpleUI(None)._interaction_callback(
+                                job_lite, interaction.extra._builder)
+                            self.sa.remember_users_response(cmd)
+                            self.finish_job(
+                                interaction.extra._builder.get_result())
+                            next_job = True
+                            break
+                        except ReRunJob:
+                            next_job = False
+                            self.sa.rerun_job(
+                                job['id'],
+                                interaction.extra._builder.get_result())
+                            break
+                    elif interaction.kind == 'comment':
+                        new_comment = input(SimpleUI.C.BLUE(
+                            _('Please enter your comments:') + '\n'))
+                        self.sa.remember_users_response(new_comment + '\n')
+                    elif interaction.kind == 'skip':
+                        self.finish_job(
+                            interaction.extra._builder.get_result())
                         next_job = True
-                    self.sa.remember_users_response(cmd)
-                elif interaction.kind == 'verification':
-                    self.wait_for_job(dont_finish=True)
-                    if interaction.message:
-                        SimpleUI.description(
-                            _('Verification:'), interaction.message)
-                    JobAdapter = namedtuple('job_adapter', ['command'])
-                    job = JobAdapter(job['command'])
-                    cmd = SimpleUI(None)._interaction_callback(
-                        job, interaction.extra)
-                    self.sa.remember_users_response(cmd)
-                    self.finish_job(interaction.extra.get_result())
-                    next_job = True
-                elif interaction.kind == 'comment':
-                    new_comment = input(SimpleUI.C.BLUE(
-                        _('Please enter your comments:') + '\n'))
-                    self.sa.remember_users_response(new_comment + '\n')
+                        break
+                else:
+                    self.wait_for_job()
+                    break
             if next_job:
                 continue
-            self.wait_for_job()
diff --git a/plainbox/impl/execution.py b/plainbox/impl/execution.py
index 38c1221..8c93f16 100644
--- a/plainbox/impl/execution.py
+++ b/plainbox/impl/execution.py
@@ -265,7 +265,9 @@ class UnifiedRunner(IJobRunner):
             extcmd_popen._delegate.on_end(proc.returncode)
             return proc.returncode
         if not os.path.isdir(os.path.join(self._session_dir, "CHECKBOX_DATA")):
+            oldmask = os.umask(000)
             os.makedirs(os.path.join(self._session_dir, "CHECKBOX_DATA"))
+            os.umask(oldmask)
         # Setup the executable nest directory
         with self.configured_filesystem(job) as nest_dir:
             # Get the command and the environment.
diff --git a/plainbox/impl/session/assistant.py b/plainbox/impl/session/assistant.py
index abcd58c..f941793 100644
--- a/plainbox/impl/session/assistant.py
+++ b/plainbox/impl/session/assistant.py
@@ -1321,7 +1321,9 @@ class SessionAssistant:
                 checkbox_data_dir = os.path.join(
                     self.get_session_dir(), 'CHECKBOX_DATA')
                 if not os.path.exists(checkbox_data_dir):
+                    oldmask = os.umask(000)
                     os.mkdir(checkbox_data_dir)
+                    os.umask(oldmask)
                 respawn_cmd_file = os.path.join(
                     checkbox_data_dir, '__respawn_checkbox')
                 if self._restart_cmd_callback:
diff --git a/plainbox/impl/session/remote_assistant.py b/plainbox/impl/session/remote_assistant.py
index 35a5b11..48b2ac0 100644
--- a/plainbox/impl/session/remote_assistant.py
+++ b/plainbox/impl/session/remote_assistant.py
@@ -25,12 +25,10 @@ import time
 import sys
 from collections import namedtuple
 from threading import Thread, Lock
-from subprocess import DEVNULL, CalledProcessError, check_call
 
 from plainbox.impl.execution import UnifiedRunner
 from plainbox.impl.session.assistant import SessionAssistant
 from plainbox.impl.session.assistant import SA_RESTARTABLE
-from plainbox.impl.session.jobs import InhibitionCause
 from plainbox.impl.secure.sudo_broker import SudoBroker, EphemeralKey
 from plainbox.impl.secure.sudo_broker import is_passwordless_sudo
 from plainbox.impl.secure.sudo_broker import validate_pass
@@ -103,7 +101,7 @@ class BackgroundExecutor(Thread):
     def wait(self):
         while self.is_alive() or not self._started_real_run:
             # return control to RPC server
-            time.sleep(0.1)
+            time.sleep(0.01)
         self.join()
         return self._builder
 
@@ -213,6 +211,8 @@ class RemoteSessionAssistant():
         self._sa.load_providers()
 
         self._normal_user = self._launcher.normal_user
+        if configuration['normal_user']:
+            self._normal_user = configuration['normal_user']
         pass_provider = (None if self._passwordless_sudo else
                          self.get_decrypted_password)
         runner_kwargs = {
@@ -275,6 +275,13 @@ class RemoteSessionAssistant():
         self._jobs_count = len(self._sa.get_dynamic_todo_list())
         self._state = TestsSelected
 
+    @allowed_when(Interacting)
+    def rerun_job(self, job_id, result):
+        self._sa.use_job_result(job_id, result)
+        self.session_change_lock.acquire(blocking=False)
+        self.session_change_lock.release()
+        self._state = TestsSelected
+
     @allowed_when(TestsSelected)
     def run_job(self, job_id):
         """
@@ -299,7 +306,8 @@ class RemoteSessionAssistant():
                     yield from self.interact(
                         Interaction('purpose', job.tr_purpose()))
                 if job.tr_steps():
-                    yield from self.interact(Interaction('steps', job.tr_steps()))
+                    yield from self.interact(
+                        Interaction('steps', job.tr_steps()))
                 if self._last_response == 'comment':
                     yield from self.interact(Interaction('comment'))
                     if self._last_response:
@@ -307,13 +315,16 @@ class RemoteSessionAssistant():
                     may_comment = True
                     continue
             if self._last_response == 'skip':
-                result_builder = JobResultBuilder(
-                    outcome=IJobResult.OUTCOME_SKIP,
-                    comments=_("Explicitly skipped before execution"))
-                if self._current_comments != "":
-                    result_builder.comments = self._current_comments
-                self.finish_job(result_builder.get_result())
-                return
+                def skipped_builder(*args, **kwargs):
+                    result_builder = JobResultBuilder(
+                        outcome=IJobResult.OUTCOME_SKIP,
+                        comments=_("Explicitly skipped before execution"))
+                    if self._current_comments != "":
+                        result_builder.comments = self._current_comments
+                    return result_builder
+                self._be = BackgroundExecutor(self, job_id, skipped_builder)
+                yield from self.interact(
+                    Interaction('skip', job.verification, self._be))
         if job.command:
             if (job.user and not self._passwordless_sudo
                     and not self._sudo_password):
@@ -338,14 +349,8 @@ class RemoteSessionAssistant():
             self._be = BackgroundExecutor(self, job_id, undecided_builder)
         if self._sa.get_job(self._currently_running_job).plugin in [
                 'manual', 'user-interact-verify']:
-            rb = self._be.wait()
-            # by this point the ui will handle adding comments via
-            # ResultBuilder.add_comment method that adds \n in front
-            # of the addition, let's rstrip it
-            rb.comments = self._current_comments.rstrip()
             yield from self.interact(
-                Interaction('verification', job.verification, rb))
-            self.finish_job(rb.get_result())
+                Interaction('verification', job.verification, self._be))
 
     @allowed_when(Started, Bootstrapping)
     def run_bootstrapping_job(self, job_id):