job tests faulty: need to mock `os.environ`.
I had to pass a recent PR despite a failing test.  The test was the ssh launch method which failed because the OS environment for `LD_LIBRARY_PATH` differed from what we encoded in the expected result.  If I change the expected result, then the tests fail when running for me locally.  Ultimately the LM is making a runtime decision based on the actual os env, and thus the result cannot be hard coded this way.  I assume we need to mock the os env for the tests, and this is what this ticket is about.
diff --git a/src/radical/pilot/agent/lm/mpirun.py b/src/radical/pilot/agent/lm/mpirun.py
index 76e6a5066..827c1c29c 100644
--- a/src/radical/pilot/agent/lm/mpirun.py
+++ b/src/radical/pilot/agent/lm/mpirun.py
@@ -19,6 +19,9 @@ class MPIRun(LaunchMethod):
 
         LaunchMethod.__init__(self, name, cfg, session)
 
+        self._mpt = False
+        self._rsh = False
+
 
     # --------------------------------------------------------------------------
     #
@@ -27,12 +30,12 @@ class MPIRun(LaunchMethod):
         self._mpt = False
         self._rsh = False
 
-        if   '_rsh' in self.name.lower():
+        if '_rsh' in self.name.lower():
             self._rsh = True
             self.launch_command = ru.which(['mpirun_rsh',         # Gordon (SDSC)
                                             'mpirun'              # general case
                                            ])
- 
+
         elif '_mpt' in self.name.lower():
             self._mpt = True
             self.launch_command = ru.which(['mpirun_mpt',         # Cheyenne (NCAR)
@@ -47,17 +50,17 @@ class MPIRun(LaunchMethod):
         # don't use the full pathname as the user might load a different
         # compiler / MPI library suite from his CU pre_exec that requires
         # the launcher from that version -- see #572.
-        self.launch_command = os.path.basename(self.launch_command)
+        # FIXME: then why are we doing this LM setup in the first place??
+        if self.launch_command:
+            self.launch_command = os.path.basename(self.launch_command)
 
 
         # do we need ccmrun or dplace?
-        self.ccmrun_command = ''
         if '_ccmrun' in self.name:
             self.ccmrun_command = ru.which('ccmrun')
             if not self.ccmrun_command:
                 raise RuntimeError("ccmrun not found!")
 
-        self.dplace_command = ''
         if '_dplace' in self.name:
             self.dplace_command = ru.which('dplace')
             if not self.dplace_command:
diff --git a/src/radical/pilot/pilot_manager.py b/src/radical/pilot/pilot_manager.py
index 60dbb466c..87355b2da 100644
--- a/src/radical/pilot/pilot_manager.py
+++ b/src/radical/pilot/pilot_manager.py
@@ -70,6 +70,8 @@ class PilotManager(rpu.Component):
         **Arguments:**
             * session [:class:`rp.Session`]:
               The session instance to use.
+            * cfg (`dict` or `string`):
+              The configuration or name of configuration to use.
 
         **Returns:**
             * A new `PilotManager` object [:class:`rp.PilotManager`].
@@ -77,11 +79,10 @@ class PilotManager(rpu.Component):
 
         assert(session.primary), 'pmgr needs primary session'
 
-        self._session     = session
         self._pilots      = dict()
-        self._pilots_lock = ru.RLock()
+        self._pilots_lock = ru.RLock('pmgr.pilots_lock')
         self._callbacks   = dict()
-        self._pcb_lock    = ru.RLock()
+        self._pcb_lock    = ru.RLock('pmgr.pcb_lock')
         self._terminate   = mt.Event()
         self._closed      = False
         self._rec_id      = 0       # used for session recording
@@ -103,22 +104,23 @@ class PilotManager(rpu.Component):
         cfg           = ru.Config('radical.pilot.pmgr', name=name, cfg=cfg)
         cfg.uid       = self._uid
         cfg.owner     = self._uid
-        cfg.sid       = self._session.uid
-        cfg.base      = self._session.base
-        cfg.path      = self._session.path
-        cfg.dburl     = self._session.dburl
-        cfg.heartbeat = self._session.cfg.heartbeat
+        cfg.sid       = session.uid
+        cfg.base      = session.base
+        cfg.path      = session.path
+        cfg.dburl     = session.dburl
+        cfg.heartbeat = session.cfg.heartbeat
 
-        rpu.Component.__init__(self, cfg, session=self._session)
+        rpu.Component.__init__(self, cfg, session=session)
+        self.start()
+
+        self._log.info('started pmgr %s', self._uid)
+        self._rep.info('<<create pilot manager')
 
         # create pmgr bridges and components, use session cmgr for that
         self._cmgr = rpu.ComponentManager(self._cfg)
         self._cmgr.start_bridges()
         self._cmgr.start_components()
 
-        # only now we have a logger... :/
-        self._rep.info('<<create pilot manager')
-
         # The output queue is used to forward submitted pilots to the
         # launching component.
         self.register_output(rps.PMGR_LAUNCHING_PENDING,
@@ -128,9 +130,9 @@ class PilotManager(rpu.Component):
         # directives
         self.register_subscriber(rpc.CONTROL_PUBSUB, self._staging_ack_cb)
         self._active_sds = dict()
-        self._sds_lock   = ru.Lock()
+        self._sds_lock   = ru.Lock('pmgr_sds_lock')
 
-        # register the state notification pull cb
+        # register the state notification pull cb and hb pull cb
         # FIXME: we may want to have the frequency configurable
         # FIXME: this should be a tailing cursor in the update worker
         self.register_timed_cb(self._state_pull_cb,
@@ -150,7 +152,7 @@ class PilotManager(rpu.Component):
 
     # --------------------------------------------------------------------------
     #
-    def initialize_common(self):
+    def initialize(self):
 
         # the manager must not carry bridge and component handles across forks
         ru.atfork(self._atfork_prepare, self._atfork_parent, self._atfork_child)
@@ -158,6 +160,8 @@ class PilotManager(rpu.Component):
 
     # --------------------------------------------------------------------------
     #
+    # EnTK forks, make sure we don't carry traces of children across the fork
+    #
     def _atfork_prepare(self): pass
     def _atfork_parent(self) : pass
     def _atfork_child(self)  :
@@ -167,26 +171,17 @@ class PilotManager(rpu.Component):
 
     # --------------------------------------------------------------------------
     #
-    def finalize_parent(self):
+    def finalize(self):
 
+        self._cmgr.close()
         self._fail_missing_pilots()
 
-        # terminate pmgr components
-        for c in self._components:
-            c.stop()
-            c.join()
-
-        # terminate pmgr bridges
-        for b in self._bridges:
-            b.stop()
-            b.join()
-
 
     # --------------------------------------------------------------------------
     #
     def close(self, terminate=True):
         """
-        Shuts down the PilotManager.
+        Shut down the PilotManager and all its components.
 
         **Arguments:**
             * **terminate** [`bool`]: cancel non-final pilots if True (default)
@@ -198,7 +193,7 @@ class PilotManager(rpu.Component):
         self._terminate.set()
         self._rep.info('<<close pilot manager')
 
-        # we don't want any callback invokations during shutdown
+        # disable callbacks during shutdown
         # FIXME: really?
         with self._pcb_lock:
             for m in rpc.PMGR_METRICS:
diff --git a/src/radical/pilot/unit_manager.py b/src/radical/pilot/unit_manager.py
index 7c3558609..34bd54c86 100644
--- a/src/radical/pilot/unit_manager.py
+++ b/src/radical/pilot/unit_manager.py
@@ -5,7 +5,7 @@ __license__   = "MIT"
 
 import os
 import time
-import threading
+import threading as mt
 
 import radical.utils as ru
 
@@ -50,8 +50,7 @@ class UnitManager(rpu.Component):
 
         # Combine the two pilots, the workload and a scheduler via
         # a UnitManager.
-        um = rp.UnitManager(session=session,
-                            scheduler=rp.SCHEDULER_ROUND_ROBIN)
+        um = rp.UnitManager(session=session, scheduler=rp.SCHEDULER_ROUND_ROBIN)
         um.add_pilot(p1)
         um.submit_units(compute_units)
 
@@ -65,13 +64,15 @@ class UnitManager(rpu.Component):
 
     # --------------------------------------------------------------------------
     #
-    def __init__(self, session, scheduler=None):
+    def __init__(self, session, cfg='default', scheduler=None):
         """
         Creates a new UnitManager and attaches it to the session.
 
         **Arguments:**
             * session [:class:`radical.pilot.Session`]:
               The session instance to use.
+            * cfg (`dict` or `string`):
+              The configuration or name of configuration to use.
             * scheduler (`string`):
               The name of the scheduler plug-in to use.
 
@@ -84,33 +85,38 @@ class UnitManager(rpu.Component):
         self._units       = dict()
         self._units_lock  = ru.RLock('umgr.units_lock')
         self._callbacks   = dict()
-        self._cb_lock     = ru.RLock()
-        self._terminate   = threading.Event()
+        self._cb_lock     = ru.RLock('umgr.cb_lock')
+        self._terminate   = mt.Event()
         self._closed      = False
         self._rec_id      = 0       # used for session recording
+        self._uid         = ru.generate_id('umgr.%(item_counter)04d',
+                                           ru.ID_CUSTOM, ns=session.uid)
 
         for m in rpc.UMGR_METRICS:
             self._callbacks[m] = dict()
 
-        # FIXME: expose in API
-        name = 'default'
-        cfg  = None
-        cfg = ru.Config('radical.pilot.umgr', name=name, cfg=cfg)
+        # NOTE: `name` and `cfg` are overloaded, the user cannot point to
+        #       a predefined config and amed it at the same time.  This might
+        #       be ok for the session, but introduces a minor API inconsistency.
+        #
+        name = None
+        if isinstance(cfg, str):
+            name = cfg
+            cfg  = None
+
+        cfg           = ru.Config('radical.pilot.umgr', name=name, cfg=cfg)
+        cfg.uid       = self._uid
+        cfg.owner     = self._uid
+        cfg.sid       = session.uid
+        cfg.base      = session.base
+        cfg.path      = session.path
+        cfg.dburl     = session.dburl
+        cfg.heartbeat = session.cfg.heartbeat
 
         if scheduler:
             # overwrite the scheduler from the config file
             cfg.scheduler = scheduler
 
-        # initialize the base class (with no intent to fork)
-        self._uid  = ru.generate_id('umgr.%(item_counter)04d', ru.ID_CUSTOM,
-                                    ns=session.uid)
-        cfg.uid       = self.uid
-        cfg.owner     = self.uid
-        cfg.sid       = self._session.uid
-        cfg.base      = self._session.base
-        cfg.path      = self._session.path
-        cfg.dburl     = self._session.dburl
-        cfg.heartbeat = self._session.cfg.heartbeat
 
         rpu.Component.__init__(self, cfg, session=session)
         self.start()
@@ -118,6 +124,7 @@ class UnitManager(rpu.Component):
         self._log.info('started umgr %s', self._uid)
         self._rep.info('<<create unit manager')
 
+        # create pmgr bridges and components, use session cmgr for that
         self._cmgr = rpu.ComponentManager(self._cfg)
         self._cmgr.start_bridges()
         self._cmgr.start_components()
@@ -152,11 +159,38 @@ class UnitManager(rpu.Component):
         self._rep.ok('>>ok\n')
 
 
+    # --------------------------------------------------------------------------
+    #
+    def initialize(self):
+
+        # the manager must not carry bridge and component handles across forks
+        ru.atfork(self._atfork_prepare, self._atfork_parent, self._atfork_child)
+
+
+    # --------------------------------------------------------------------------
+    #
+    # EnTK forks, make sure we don't carry traces of children across the fork
+    #
+    def _atfork_prepare(self): pass
+    def _atfork_parent(self) : pass
+    def _atfork_child(self)  :
+        self._bridges    = dict()
+        self._components = dict()
+
+
+    # --------------------------------------------------------------------------
+    #
+    def finalize(self):
+
+        self._cmgr.close()
+        self._fail_missing_pilots()
+
+
     # --------------------------------------------------------------------------
     #
     def close(self):
         """
-        Shut down the UnitManager, and all umgr components.
+        Shut down the UnitManager and all its components.
         """
 
         # we do not cancel units at this point, in case any component or pilot
@@ -167,7 +201,6 @@ class UnitManager(rpu.Component):
             return
 
         self._terminate.set()
-
         self._rep.info('<<close unit manager')
 
         # disable callbacks during shutdown
diff --git a/tests/test_cases/unit.000000.json b/tests/test_cases/unit.000000.json
index de35648a8..107d060cf 100644
--- a/tests/test_cases/unit.000000.json
+++ b/tests/test_cases/unit.000000.json
@@ -43,7 +43,7 @@
             "fork"    : ["/bin/sleep", null],
             # "orte"    : ["orterun  --hnp \"dvm_uri\" -np 1 --bind-to none -host node1 -x \"LD_LIBRARY_PATH\" -x \"PATH\" -x \"PYTHONPATH\"  /bin/sleep", null],
             "aprun"   : ["aprun -n 1 -N 1 -L node1 -d 1 -cc 0 /bin/sleep ", null],
-            "mpirun"  : ["mpirun -np 1  -host node1  /bin/sleep", null],
+            "mpirun"  : ["mpirun  -np 1  -host node1  /bin/sleep", null],
             "ibrun"   : ["ibrun -n 1 -o 1 /bin/sleep", null],
             "ssh"     : "ValueError",
             "spark"   : ["//bin/sleep     ",null],
diff --git a/tests/test_cases/unit.000003.json b/tests/test_cases/unit.000003.json
index 689441177..8c6d5feb9 100644
--- a/tests/test_cases/unit.000003.json
+++ b/tests/test_cases/unit.000003.json
@@ -38,7 +38,7 @@
             "fork"    : ["/bin/sleep \"10\" ", null],
             # "orte"    : ["orterun  --hnp \"dvm_uri\" -np 1 --bind-to none -host node1 -x \"LD_LIBRARY_PATH\" -x \"PATH\" -x \"PYTHONPATH\"  /bin/sleep \"10\" ", null],
             "aprun"   : ["aprun -n 1 -N 1 -L node1 -d 4 -cc 0,1,2,3 /bin/sleep \"10\"  ", null],
-            "mpirun"  : ["mpirun -np 1  -host node1  /bin/sleep \"10\"", null],
+            "mpirun"  : ["mpirun  -np 1  -host node1  /bin/sleep \"10\"", null],
             "ibrun"   : ["ibrun -n 1 -o 4 /bin/sleep \"10\" ", null],
             "ssh"     : ["/bin/sleep \"10\" ","ssh node1 LD_LIBRARY_PATH=/usr/local/lib/ PATH=test_path 1"],
             "rsh"     : ["/bin/sleep \"10\" ","rsh node1 LD_LIBRARY_PATH=/usr/local/lib/ PATH=test_path 1"],
diff --git a/tests/test_cases/unit.000004.json b/tests/test_cases/unit.000004.json
index dba5bdf7b..c74128871 100644
--- a/tests/test_cases/unit.000004.json
+++ b/tests/test_cases/unit.000004.json
@@ -35,7 +35,7 @@
             "fork"  : ["/bin/sleep \"10\" ", null],
           # "orte"  : "RuntimeError",
             "aprun" : ["aprun -n 1 -N 1 -L node1 -d 1 -cc 0 /bin/sleep \"10\"  ", null],
-            "mpirun": ["mpirun -np 0  -host   /bin/sleep \"10\"", null],
+            "mpirun": ["mpirun  -np 0  -host   /bin/sleep \"10\"", null],
             "ibrun" : ["ibrun -n 0 -o 0 /bin/sleep \"10\" ", null],
             "ssh"   : ["/bin/sleep \"10\" ","ssh node1 LD_LIBRARY_PATH=/usr/local/lib/ PATH=test_path 1"],
             "spark" : "RuntimeError",
diff --git a/tests/test_cases/unit.000005.json b/tests/test_cases/unit.000005.json
index f1fca51e5..2df4d7eb0 100644
--- a/tests/test_cases/unit.000005.json
+++ b/tests/test_cases/unit.000005.json
@@ -41,7 +41,7 @@
             "fork"    : ["/bin/sleep \"10\" ", null],
             # "orte"    : ["orterun  --hnp \"dvm_uri\" -np 2 --bind-to none -host node1,node1 -x \"LD_LIBRARY_PATH\" -x \"PATH\" -x \"PYTHONPATH\"  /bin/sleep \"10\" ", null],
             "aprun"   : ["aprun -n 2 -N 2 -L node1 -d 1 -cc 0:0 /bin/sleep \"10\"  ", null],
-            "mpirun"  : ["mpirun -np 1  -host node1  /bin/sleep \"10\"", null],
+            "mpirun"  : ["mpirun  -np 1  -host node1  /bin/sleep \"10\"", null],
             "ibrun"   : ["ibrun -n 1 -o 0 /bin/sleep \"10\" ", null],
             "ssh"     : ["/bin/sleep \"10\" ","ssh node1 LD_LIBRARY_PATH=/usr/local/lib/ PATH=test_path 1"],
             "jsrun"   : ["jsrun --erf_input rs_layout_cu_000005 --smpiargs=\"-gpu\"  /bin/sleep \"10\" ", null],
diff --git a/tests/test_cases/unit.000006.json b/tests/test_cases/unit.000006.json
index 8b9c67dc4..079fadc79 100644
--- a/tests/test_cases/unit.000006.json
+++ b/tests/test_cases/unit.000006.json
@@ -39,7 +39,7 @@
             "fork"  : ["/bin/sleep \"10\" ", null],
             # "orte"  : ["orterun  --hnp \"dvm_uri\" -np 2 --bind-to none -host node1,node1 -x \"LD_LIBRARY_PATH\" -x \"PATH\" -x \"PYTHONPATH\"  /bin/sleep \"10\" ", null],
             "aprun" : ["aprun -n 2 -N 2 -L node1 -d 2 -cc 0,1:2,3 /bin/sleep \"10\"  ", null],
-            "mpirun": ["mpirun -np 2  -host node1,node1  /bin/sleep \"10\"", null],
+            "mpirun": ["mpirun  -np 2  -host node1,node1  /bin/sleep \"10\"", null],
             "ibrun" : ["ibrun -n 2 -o 2 /bin/sleep \"10\" ", null],
             "ssh"   : ["/bin/sleep \"10\" ","ssh node1 LD_LIBRARY_PATH=/usr/local/lib/ PATH=test_path 1"],
             "jsrun"  : ["jsrun --erf_input rs_layout_cu_000006   /bin/sleep \"10\" ",null]
diff --git a/tests/test_lm/test_unit/test_mpirun.py b/tests/test_lm/test_unit/test_mpirun.py
index 01748a018..31a464075 100644
--- a/tests/test_lm/test_unit/test_mpirun.py
+++ b/tests/test_lm/test_unit/test_mpirun.py
@@ -111,17 +111,19 @@ def test_configure_dplace(mocked_init, mocked_get_mpi_info, mocked_raise_on,
 # ------------------------------------------------------------------------------
 #
 @mock.patch.object(MPIRun, '__init__',   return_value=None)
-@mock.patch.object(MPIRun, '_configure', return_value=None)
+@mock.patch.object(MPIRun, '_get_mpi_info', return_value=[5,'ORTE'])
 @mock.patch('radical.utils.raise_on')
 def test_construct_command(mocked_init,
-                           mocked_configure,
+                           mocked_get_mpi_info,
                            mocked_raise_on):
 
     test_cases = setUp('lm', 'mpirun')
+
     component  = MPIRun(name=None, cfg=None, session=None)
+    component.name = 'MPIRun'
+    component._configure()
 
     component._log           = ru.Logger('dummy')
-    component.name           = 'MPIRun'
     component.mpi_flavor     = None
     component.launch_command = 'mpirun'
     component.ccmrun_command = ''
@@ -129,6 +131,7 @@ def test_construct_command(mocked_init,
 
     for unit, result in test_cases:
         command, hop = component.construct_command(unit, None)
-        assert([command, hop] == result)
+        assert([command, hop] == result), unit['uid']
+
 
 # ------------------------------------------------------------------------------
['src/radical/pilot/unit_manager.py', 'src/radical/pilot/pilot_manager.py', 'tests/test_cases/unit.000005.json', 'src/radical/pilot/agent/lm/mpirun.py', 'tests/test_cases/unit.000000.json', 'tests/test_lm/test_unit/test_mpirun.py', 'tests/test_cases/unit.000004.json', 'tests/test_cases/unit.000006.json', 'tests/test_cases/unit.000003.json']