Merge ~sylvain-pineau/checkbox-ng:mini-me into checkbox-ng:remote-api-bump

[Sylvain Pineau]

the checkbox-cli mini-me command (see the commits for details)

How to test?

The standalone command (quick an easy):

$ checkbox-cli mini-me --host 192.168.1.33 com.canonical.certification::usb/detect

Nested inside an existing checkbox remote/service session:

1. Prepare the following launcher (adjust the target host ip of course):

    [launcher]
    app_id = com.canonical.certification:checkbox-test
    launcher_version = 1
    stock_reports = text, submission_files

    [test plan]
    unit = com.canonical.certification::demo
    forced = yes

    [test selection]
    forced = yes

    [ui]
    output = hide-resource-and-attachment

    [environment]
    STRESS_BOOT_ITERATIONS=2
    STRESS_BOOT_WAIT_DELAY=10
    STRESS_BOOT_WAKEUP_DELAY=15
    TARGET_HOST=192.168.1.33

2. A small test plan like this:

    id: demo
    unit: test plan
    _name: demo tests
    _description:
     demo tests
    include:
     demo/usb/detect

3. A demo job calling the new command:

    plugin: shell
    category_id: com.canonical.plainbox::usb
    id: demo/usb/detect
    command:
     checkbox-cli mini-me com.canonical.certification::usb/detect
    _summary: Display USB devices attached to SUT
    _description: Detects and shows USB devices attached to this system.

4. Install checkbox-ng and the checkbox provider from the dev ppa on a spare system

5. From your laptop, start a checkbox service (after adding the test plan and the new job to one of your local providers, side-loading one is a good idea):

    sudo checkbox-cli service

6. Finally execute your launcher file with:

    checkbox-cli master 127.0.0.1 ./mylauncher

If everything goes well, the usb/detect output should list devices from your spare system

[Taihsiang Ho (tai271828)]

Some diff-like unwanted lines are still in the merge proposal. See the inline comments below.

review: Needs Fixing

Unmerged commits

fc1d1f2... by Sylvain Pineau on 2022-02-11

mini-me: new checkbox-cli command

The "remote" version of the "local" `checkbox-cli run` command.

Supported pattern:
- Only a single job id is supported
- Only automated tests, no manual or semi-auto plugins
- Test plan ids are rejected

Selection of the target host:
- Target machine IP can be provided with either --host or the env var TARGET_HOST

Standard streams
- Remote stdout/stderr are printed locally on the the same local streams
- No colorizer

Session management:
- Ephemeral sessions on service side, removed after job execution
- launcher settings are transferred to the TARGET_HOST

73f7921... by Sylvain Pineau on 2022-02-11

session:remote_assistant: Expose the hand_pick_jobs() method

7a8cf0d... by Sylvain Pineau on 2022-02-11

session:assistant: add a method to save the launcher file in the session dir

save_launcher_file is also exposed over rpyc

f9918ab... by Sylvain Pineau on 2022-02-11

execution.py: Set a new env var PLAINBOX_SESSION_DIR

Full path to the current session directory

707dbcf... by Sylvain Pineau on 2022-02-11

execution.py: Set PYTHONUNBUFFERED for all child processes

055cc34... by Sylvain Pineau on 2021-12-18

fix: restart - Disable the checkbox-ng.service if __respawn_checkbox is used

Disable the service provided by checkbox-ng if another mechanism handles
the session restart.
e.g pm_test currently manages several reboots in a single job.

diff --git a/checkbox_ng/launcher/checkbox_cli.py b/checkbox_ng/launcher/checkbox_cli.py
index 9740eb8..0ae7c14 100644
--- a/checkbox_ng/launcher/checkbox_cli.py
+++ b/checkbox_ng/launcher/checkbox_cli.py
@@ -38,6 +38,7 @@ from checkbox_ng.launcher.check_config import CheckConfig
 from checkbox_ng.launcher.merge_reports import MergeReports
 from checkbox_ng.launcher.merge_submissions import MergeSubmissions
 from checkbox_ng.launcher.master import RemoteMaster
+from checkbox_ng.launcher.mini_me import RemoteMe
 from checkbox_ng.launcher.slave import RemoteSlave
 
 
@@ -67,6 +68,7 @@ def main():
         'tp-export': TestPlanExport,
         'service': RemoteSlave,
         'remote': RemoteMaster,
+        'mini-me': RemoteMe,
     }
     deprecated_commands = {
         'slave': 'service',
diff --git a/checkbox_ng/launcher/mini_me.py b/checkbox_ng/launcher/mini_me.py
new file mode 100644
index 0000000..266ed80
--- /dev/null
+++ b/checkbox_ng/launcher/mini_me.py
@@ -0,0 +1,327 @@
+# This file is part of Checkbox.
+#
+# Copyright 2022 Canonical Ltd.
+# Written by:
+#   Sylvain Pineau <[e-mail]>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+"""
+This module contains implementation of the mini me end of the remote execution
+functionality.
+"""
+import contextlib
+import ipaddress
+import json
+import logging
+import os
+import select
+import socket
+import time
+import sys
+
+from functools import partial
+
+from plainbox.impl.session.remote_assistant import RemoteSessionAssistant
+from plainbox.vendor import rpyc
+from checkbox_ng.launcher.stages import MainLoopStage
+from checkbox_ng.launcher.stages import ReportsStage
+from tqdm import tqdm
+_logger = logging.getLogger("mini-me")
+
+
+def environ_or_required(key):
+    """Mapping for argparse to supply required or default from $ENV."""
+    if os.environ.get(key):
+        return {'default': os.environ.get(key)}
+    else:
+        return {'required': True}
+
+
+class RemoteMe(ReportsStage, MainLoopStage):
+    """
+    Control remote slave instance
+
+    This class implements the part that presents UI to the operator and
+    steers the session.
+    """
+
+    name = 'remote-me'
+
+    @property
+    def is_interactive(self):
+        return False
+
+    @property
+    def C(self):
+        return None
+
+    @property
+    def sa(self):
+        return self._sa
+
+    def invoked(self, ctx):
+        self._launcher_text = ''
+        self._is_bootstrapping = False
+        self._target_host = ctx.args.host
+        self._normal_user = ''
+        timeout = 600
+        deadline = time.time() + timeout
+        port = ctx.args.port
+        self.selection = [ctx.args.PATTERN]
+        if not ipaddress.ip_address(ctx.args.host).is_loopback:
+            _logger.info("Connecting to {}:{}. Timeout: {}s".format(
+                ctx.args.host, port, timeout))
+        while time.time() < deadline:
+            try:
+                self.connect_and_run(ctx.args.host, port)
+                break
+            except (ConnectionRefusedError, socket.timeout, OSError):
+                print('.', end='', flush=True)
+                time.sleep(1)
+        else:
+            print("\nConnection timed out.")
+
+    def connect_and_run(self, host, port=18871):
+        config = rpyc.core.protocol.DEFAULT_CONFIG.copy()
+        config['allow_all_attrs'] = True
+        config['sync_request_timeout'] = 120
+        keep_running = False
+        server_msg = None
+        self._prepare_transports()
+        interrupted = False
+        while True:
+            try:
+                if interrupted:
+                    _logger.info("remote: Session interrupted")
+                    interrupted = False  # we are handling the interruption ATM
+                    # next line can raise exception due to connection being
+                    # lost so let's set the default behavior to quitting
+                    keep_running = False
+                    keep_running = self._handle_interrupt()
+                    if not keep_running:
+                        break
+                conn = rpyc.connect(host, port, config=config)
+                keep_running = True
+
+                def quitter(msg):
+                    # this will be called when the slave decides to disconnect
+                    # this master
+                    nonlocal server_msg
+                    nonlocal keep_running
+                    keep_running = False
+                    server_msg = msg
+                with contextlib.suppress(AttributeError):
+                    # TODO: REMOTE_API
+                    # when bumping the remote api make this bit obligatory
+                    # i.e. remove the suppressing
+                    conn.root.register_master_blaster(quitter)
+                self._sa = conn.root.get_sa()
+                self.sa.conn = conn
+                # TODO: REMOTE API RAPI: Remove this API on the next RAPI bump
+                # the check and bailout is not needed if the slave as up to
+                # date as this master, so after bumping RAPI we can assume
+                # that slave is always passwordless
+                if not self.sa.passwordless_sudo:
+                    raise SystemExit(
+                        "This version of Checkbox requires the service"
+                        " to be run as root")
+                try:
+                    slave_api_version = self.sa.get_remote_api_version()
+                except AttributeError:
+                    raise SystemExit("Service doesn't declare Remote API"
+                                     " version. Update Checkbox on the"
+                                     " SUT!")
+                master_api_version = RemoteSessionAssistant.REMOTE_API_VERSION
+                if slave_api_version != master_api_version:
+                    raise SystemExit(
+                        "Remote API version mismatch. Service "
+                        "uses: {}. Remote uses: {}".format(
+                            slave_api_version, master_api_version))
+                state, payload = self.sa.whats_up()
+                _logger.info("remote: Main dispatch with state: %s", state)
+                keep_running = {
+                    'idle': self.new_session,
+                    'running': self.wait_and_continue,
+                    'finalizing': self.finish_session,
+                    'testsselected': partial(
+                        self.run_jobs, resumed_session_info=payload),
+                    'bootstrapping': self.restart,
+                    'started': self.restart,
+                    'interacting': partial(
+                        self.resume_interacting, interaction=payload),
+                }[state]()
+            except EOFError as exc:
+                if keep_running:
+                    print("Connection lost!")
+                    # this is yucky but it works, in case of explicit
+                    # connection closing by the slave we get this msg
+                    _logger.info("remote: Connection lost due to: %s", exc)
+                    if str(exc) == 'stream has been closed':
+                        print('Service explicitly disconnected you. Possible '
+                              'reason: new remote connected to the service')
+                        break
+                    print(exc)
+                    time.sleep(1)
+                else:
+                    # if keep_running got set to False it means that the
+                    # network interruption was planned, AKA slave disconnected
+                    # this master
+                    print(server_msg)
+                    break
+            except (ConnectionRefusedError, socket.timeout, OSError) as exc:
+                _logger.info("remote: Connection lost due to: %s", exc)
+                if not keep_running:
+                    raise
+                # it's reconnecting, so we can ignore refuses
+                print('.', flush=True)
+                time.sleep(0.5)
+            except KeyboardInterrupt:
+                interrupted = True
+
+            if not keep_running:
+                break
+
+    def new_session(self):
+        _logger.info("remote: Starting new session.")
+        configuration = dict()
+        session_dir = os.environ.get('PLAINBOX_SESSION_DIR', '/tmp')
+        launcher_file = os.path.join(session_dir, 'launcher')
+        if os.path.isfile(launcher_file):
+            with open(launcher_file, 'rt', encoding='UTF-8') as f:
+                self._launcher_text = f.read()
+        configuration['launcher'] = self._launcher_text
+        configuration['normal_user'] = self._normal_user
+        self.sa.start_session(configuration)
+        self.sa.hand_pick_jobs(self.selection)
+        self.run_jobs()
+
+    def register_arguments(self, parser):
+        parser.add_argument('--host', help="target host",
+                        **environ_or_required('TARGET_HOST'))
+        parser.add_argument('--port', type=int, default=18871, help=(
+            "port to connect to"))
+        parser.add_argument(
+            'PATTERN',
+            help="run a job matching the given regular expression")
+
+    def _handle_interrupt(self):
+        """
+        Returns True if the remote should keep running.
+        And False if it should quit.
+        """
+        self._sa.terminate()
+        return False
+
+    def cleanup_ephemeral_session(self):
+        storage = self.sa.manager.storage
+        self.sa.finalize_session()
+        storage.remove()
+
+    def finish_session(self):
+        self.cleanup_ephemeral_session()
+        return_code = self.job_result.return_code
+        if return_code:
+            raise SystemExit(return_code)
+
+    def wait_and_continue(self):
+        progress = self.sa.whats_up()[1]
+        print("Rejoined session.")
+        print("In progress: {} ({}/{})".format(
+            progress[2], progress[0], progress[1]))
+        self.wait_for_job()
+        self.run_jobs()
+
+    def _handle_last_job_after_resume(self, resumed_session_info):
+        jobs_repr = json.loads(
+            self.sa.get_jobs_repr([resumed_session_info['last_job']]))
+        job = jobs_repr[-1]
+        self.job_result = self.sa.get_job_result(job['id'])
+
+    def run_jobs(self, resumed_session_info=None):
+        if resumed_session_info and resumed_session_info['last_job']:
+            self._handle_last_job_after_resume(resumed_session_info)
+        _logger.info("remote: Running jobs.")
+        jobs = self.sa.get_session_progress()
+        _logger.debug("remote: Jobs to be run:\n%s",
+                      '\n'.join(['  ' + job for job in jobs]))
+        total_num = len(jobs['done']) + len(jobs['todo'])
+        if total_num > 1:
+            self.cleanup_ephemeral_session()
+            raise SystemExit("More than one job to run!")
+        if total_num == 0:
+            self.cleanup_ephemeral_session()
+            raise SystemExit(
+                "No job found with this id: {}".format(self.selection[0]))
+        jobs_repr = json.loads(
+            self.sa.get_jobs_repr(jobs['todo'], len(jobs['done'])))
+
+        self._run_jobs(jobs_repr, total_num)
+        self.finish_session()
+
+    def resume_interacting(self, interaction):
+        self.sa.remember_users_response('rollback')
+        self.run_jobs()
+
+    def wait_for_job(self, dont_finish=False):
+        _logger.info("remote: Waiting for job to finish.")
+        while True:
+            state, payload = self.sa.monitor_job()
+            if payload and not self._is_bootstrapping:
+                for line in payload.splitlines():
+                    if line.startswith('stderr'):
+                        print(line[6:], file=sys.stderr)
+                    else:
+                        print(line[6:])
+            if state == 'running':
+                time.sleep(0.5)
+                while True:
+                    res = select.select([sys.stdin], [], [], 0)
+                    if not res[0]:
+                        break
+                    # XXX: this assumes that sys.stdin is chunked in lines
+                    buff = res[0][0].readline()
+                    self.sa.transmit_input(buff)
+                    if not buff:
+                        break
+            else:
+                if dont_finish:
+                    return
+                self.finish_job()
+                break
+
+    def finish_job(self, result=None):
+        _logger.info("remote: Finishing job with a result: %s", result)
+        self.job_result = self.sa.finish_job(result)
+
+    def abandon(self):
+        _logger.info("remote: Abandoning session.")
+        self.sa.finalize_session()
+
+    def restart(self):
+        _logger.info("remote: Restarting session.")
+        self.abandon()
+        self.new_session()
+
+    def _run_jobs(self, jobs_repr, total_num=0):
+        for job in jobs_repr:
+            # print("ID: {0}".format(job['id']))
+            if 'user' in job['plugin'] or 'manual' in job['plugin']:
+                self.cleanup_ephemeral_session()
+                raise SystemExit("Not an automated job: {0}".format(job['id']))
+            next_job = False
+            while next_job is False:
+                list(self.sa.run_job(job['id']))
+                self.wait_for_job()
+                break
+            if next_job:
+                continue
diff --git a/plainbox/impl/execution.py b/plainbox/impl/execution.py
index e1aa65b..06c19a1 100644
--- a/plainbox/impl/execution.py
+++ b/plainbox/impl/execution.py
@@ -468,6 +468,11 @@ def get_execution_environment(job, environ, session_id, nest_dir):
     # Add per-session shared state directory
     env['PLAINBOX_SESSION_SHARE'] = WellKnownDirsHelper.session_share(
         session_id)
+    # Add the session directory
+    env['PLAINBOX_SESSION_DIR'] = WellKnownDirsHelper.session_dir(
+        session_id)
+    # Force the stdout and stderr streams to be unbuffered
+    env['PYTHONUNBUFFERED'] = "1"
 
     def set_if_not_none(envvar, source):
         """Update env if the source variable is not None"""
@@ -542,6 +547,8 @@ def get_differential_execution_environment(job, environ, session_id, nest_dir,
         for key, value in base_env.items():
             if key in copy_vars or key.startswith('SNAP'):
                 delta_env[key] = value
+    # Force the stdout and stderr streams to be unbuffered
+    delta_env['PYTHONUNBUFFERED'] = "1"
     return delta_env
 
 
diff --git a/plainbox/impl/session/assistant.py b/plainbox/impl/session/assistant.py
index c560828..253148e 100644
--- a/plainbox/impl/session/assistant.py
+++ b/plainbox/impl/session/assistant.py
@@ -613,6 +613,12 @@ class SessionAssistant:
         self._context.state.metadata.app_blob = updated_blob
         self._manager.checkpoint()
 
+    def save_launcher_file(self, launcher_text):
+        session_dir = WellKnownDirsHelper.session_dir(self.get_session_id())
+        launcher_file = os.path.join(session_dir, 'launcher')
+        with open(launcher_file, 'wt', encoding='UTF-8') as f:
+                f.write(launcher_text)
+
     @morris.signal
     def session_available(self, session_id):
         """
diff --git a/plainbox/impl/session/remote_assistant.py b/plainbox/impl/session/remote_assistant.py
index 96f5de4..6681c9b 100644
--- a/plainbox/impl/session/remote_assistant.py
+++ b/plainbox/impl/session/remote_assistant.py
@@ -31,6 +31,7 @@ from subprocess import CalledProcessError, check_output
 
 from plainbox.impl.config import Configuration
 from plainbox.impl.execution import UnifiedRunner
+from plainbox.impl.session import SessionMetaData
 from plainbox.impl.session.assistant import SessionAssistant
 from plainbox.impl.session.assistant import SA_RESTARTABLE
 from plainbox.impl.session.jobs import InhibitionCause
@@ -185,6 +186,9 @@ class RemoteSessionAssistant():
     def update_app_blob(self, app_blob):
         self._sa.update_app_blob(app_blob)
 
+    def save_launcher_file(self, launcher_text):
+        self._sa.save_launcher_file(launcher_text)
+
     def allowed_when(*states):
         def wrap(f):
             def fun(self, *args):
@@ -301,6 +305,7 @@ class RemoteSessionAssistant():
             'effective_normal_user': self._normal_user,
         }).encode("UTF-8")
         self._sa.update_app_blob(new_blob)
+        self._sa.save_launcher_file(configuration['launcher'])
         self._sa.configure_application_restart(self._cmd_callback)
 
         self._session_id = self._sa.get_session_id()
@@ -333,6 +338,11 @@ class RemoteSessionAssistant():
     def get_bootstrapping_todo_list(self):
         return self._sa.get_bootstrap_todo_list()
 
+    @allowed_when(Started)
+    def hand_pick_jobs(self, id_patterns):
+        self._sa.hand_pick_jobs(id_patterns)
+        self._state = TestsSelected
+
     def finish_bootstrap(self):
         self._sa.finish_bootstrap()
         self._state = Bootstrapped
@@ -646,6 +656,7 @@ class RemoteSessionAssistant():
             'extra_env': self.prepare_extra_env,
         }
         meta = self._sa.resume_session(session_id, runner_kwargs=runner_kwargs)
+<<<<<<< plainbox/impl/session/remote_assistant.py
         app_blob = json.loads(meta.app_blob.decode("UTF-8"))
         launcher = app_blob['launcher']
         self._launcher = Configuration.from_text(launcher, 'Remote launcher')
@@ -659,6 +670,20 @@ class RemoteSessionAssistant():
         test_plan_id = app_blob['testplan_id']
         self._sa.select_test_plan(test_plan_id)
         self._sa.bootstrap()
+=======
+        if SessionMetaData.FLAG_TESTPLANLESS not in meta.flags:
+            app_blob = json.loads(meta.app_blob.decode("UTF-8"))
+            launcher = app_blob['launcher']
+            self._launcher.read_string(launcher, False)
+            self._sa.use_alternate_configuration(self._launcher)
+            self._normal_user = app_blob.get(
+                'effective_normal_user', self._launcher.normal_user)
+            _logger.info(
+                "normal_user after loading metadata: %r", self._normal_user)
+            test_plan_id = app_blob['testplan_id']
+            self._sa.select_test_plan(test_plan_id)
+            self._sa.bootstrap()
+>>>>>>> plainbox/impl/session/remote_assistant.py
         self._last_job = meta.running_job_name
 
         result_dict = {
diff --git a/plainbox/impl/session/restart.py b/plainbox/impl/session/restart.py
index b3c72cd..a79741b 100644
--- a/plainbox/impl/session/restart.py
+++ b/plainbox/impl/session/restart.py
@@ -255,7 +255,7 @@ class RemoteDebRestartStrategy(RemoteSnappyRestartStrategy):
         with open(self.session_resume_filename, 'wt') as f:
             f.write(session_id)
             os.fsync(f.fileno())
-        if cmd == self.service_name:
+        if cmd != self.service_name:
             subprocess.call(['systemctl', 'disable', self.service_name])