Move return codes into main module

Signed-off-by: Andrea Cervesato <[email protected]>

Author: acerv

ltp/main.py

@@ -7,7 +7,6 @@
 """
 import os
 import re
-import sys
 import inspect
 import argparse
 import importlib
@@ -19,11 +18,17 @@
 from ltp.session import Session
 from ltp.ui import SimpleUserInterface
 from ltp.ui import VerboseUserInterface
+from ltp.dispatcher import SuiteTimeoutError
 
 
 # runtime loaded SUT(s)
 LOADED_SUT = []
 
+RC_OK = 0
+RC_ERROR = 1
+RC_TIMEOUT = 124
+RC_INTERRUPT = 130
+
 
 def _from_params_to_config(params: list) -> dict:
     """
@@ -211,17 +216,30 @@ def _start_session(parser: ArgumentParser, args: Namespace) -> None:
     else:
         SimpleUserInterface(args.no_colors)
 
-    exit_code = session.run_single(
-        command=args.run_cmd,
-        suites=args.run_suite,
-        report_path=args.json_report)
-
-    ltp.events.stop_event_loop()
-
-    sys.exit(exit_code)
-
-
-def run() -> None:
+    exit_code = RC_OK
+
+    try:
+        session.run_single(
+            command=args.run_cmd,
+            suites=args.run_suite,
+            report_path=args.json_report)
+    except KeyboardInterrupt:
+        session.stop(timeout=60)
+        exit_code = RC_INTERRUPT
+        ltp.events.fire("session_stopped")
+    except SuiteTimeoutError:
+        exit_code = RC_TIMEOUT
+    except ltp.LTPException as err:
+        session.stop(timeout=60)
+        exit_code = RC_ERROR
+        ltp.events.fire("session_error", str(err))
+    finally:
+        ltp.events.stop_event_loop()
+
+    parser.exit(exit_code)
+
+
+def run(cmd_args: list = None) -> None:
     """
     Entry point of the application.
     """
@@ -298,11 +316,11 @@ def run() -> None:
         type=str,
         help="JSON output report")
 
-    args = parser.parse_args()
+    args = parser.parse_args(cmd_args)
 
     if args.sut and "help" in args.sut:
         print(args.sut["help"])
-        return
+        parser.exit(RC_OK)
 
     if args.json_report and os.path.exists(args.json_report):
         parser.error(f"JSON report file already exists: {args.json_report}")

ltp/session.py

@@ -15,7 +15,6 @@
 from ltp.sut import IOBuffer
 from ltp.tempfile import TempDir
 from ltp.dispatcher import SerialDispatcher
-from ltp.dispatcher import SuiteTimeoutError
 from ltp.export import JSONExporter
 from ltp.utils import Timeout
 
@@ -50,11 +49,6 @@ class Session:
     The main session handler.
     """
 
-    RC_OK = 0
-    RC_ERROR = 1
-    RC_TIMEOUT = 124
-    RC_INTERRUPT = 130
-
     def __init__(self, **kwargs) -> None:
         """
         :param tmpdir: temporary directory path
@@ -212,7 +206,7 @@ def run_single(
             self,
             command: str = None,
             suites: list = None,
-            report_path: str = None) -> int:
+            report_path: str = None) -> None:
         """
         Run some testing suites with a specific SUT configurations.
         :param command: command to execute
@@ -221,10 +215,7 @@ def run_single(
         :type suites: list
         :param report_path: path of the report file. If None, it won't be saved
         :type report_path: None | str
-        :returns: exit code for the session
         """
-        exit_code = self.RC_OK
-
         with self._lock_run:
             ltp.events.fire("session_started", self._tmpdir.abspath)
 
@@ -252,21 +243,12 @@ def run_single(
                         suites, skip_tests=self._skip_tests)
 
                     self._dispatcher.stop()
-            except SuiteTimeoutError:
-                exit_code = self.RC_TIMEOUT
             except LTPException as err:
-                self._stop_all(timeout=60)
-
                 self._logger.exception(err)
-                ltp.events.fire("session_error", str(err))
-
-                exit_code = self.RC_ERROR
-            except KeyboardInterrupt:
+                raise err
+            except KeyboardInterrupt as err:
                 self._logger.info("Keyboard interrupt")
-                self._stop_all(timeout=60)
-                ltp.events.fire("session_stopped")
-
-                exit_code = self.RC_INTERRUPT
+                raise err
             finally:
                 results = self._dispatcher.last_results
                 if results:
@@ -288,5 +270,3 @@ def run_single(
                     self._stop_sut(timeout=60)
                     ltp.events.fire("session_completed", results)
                     self._logger.info("Session completed")
-
-        return exit_code

ltp/tests/test_session.py

@@ -8,7 +8,7 @@
 import pytest
 import ltp
 from ltp.session import Session
-from ltp.tempfile import TempDir
+from ltp.dispatcher import SuiteTimeoutError
 from ltp.host import HostSUT
 
 
@@ -186,9 +186,8 @@ def test_run_cmd(self, sut, tmpdir, sut_config, ltpdir):
                 ltpdir=ltpdir,
                 tmpdir=str(tmpdir))
 
-            retcode = session.run_single(command="ls -l")
+            session.run_single(command="ls -l")
 
-            assert retcode == Session.RC_OK
             assert tracer.next_event() == "session_started"
             assert tracer.next_event() == "sut_start"
             assert tracer.next_event() == "run_cmd_start"
@@ -228,12 +227,11 @@ def test_run_single(
                 ltpdir=ltpdir,
                 tmpdir=str(tmpdir))
 
-            retcode = session.run_single(
+            session.run_single(
                 report_path=report_path,
                 suites=suites,
                 command=command)
 
-            assert retcode == Session.RC_OK
             assert tracer.next_event() == "session_started"
             assert tracer.next_event() == "sut_start"
 
@@ -274,7 +272,7 @@ def test_skip_tests(
                 tmpdir=str(tmpdir),
                 skip_tests="dir0[12]|dir0(1|3)|dir05")
 
-            retcode = session.run_single(
+            session.run_single(
                 report_path=report_path,
                 suites=[
                     "dirsuite0",
@@ -288,7 +286,6 @@ def test_skip_tests(
             with open(report_path, 'r') as report_f:
                 report_d = json.loads(report_f.read())
 
-            assert retcode == Session.RC_OK
             tests = [item['test_fqn'] for item in report_d["results"]]
             assert "dir01" not in tests
             assert "dir02" not in tests
@@ -321,9 +318,8 @@ def stop_exec_suites(test):
             sut_config["name"],
             None)
 
-        retcode = session.run_single(report_path=report_path, suites=suites)
+        session.run_single(report_path=report_path, suites=suites)
 
-        assert retcode == Session.RC_OK
         assert os.path.exists(report_path)
         assert tracer.next_event() == "session_started"
         assert tracer.next_event() == "sut_start"
@@ -349,11 +345,11 @@ def test_suite_timeout_report(self, sut, tmpdir, sut_config, ltpdir):
             sut_config["name"],
             None)
 
-        retcode = session.run_single(
-            suites=["sleep"],
-            report_path=report_path)
+        with pytest.raises(SuiteTimeoutError):
+            session.run_single(
+                suites=["sleep"],
+                report_path=report_path)
 
-        assert retcode == Session.RC_TIMEOUT
         assert os.path.exists(report_path)
         assert tracer.next_event() == "session_started"
         assert tracer.next_event() == "sut_start"
@@ -385,11 +381,10 @@ def test_env(self, sut, tmpdir, sut_config, ltpdir):
                 tmpdir=str(tmpdir),
                 env=dict(VAR0="0", VAR1="1"))
 
-            retcode = session.run_single(
+            session.run_single(
                 report_path=report_path,
                 suites=["suite"])
 
-            assert retcode == Session.RC_OK
             assert os.path.isfile(report_path)
 
             report_d = None