Merge pull request #118 from simonsobs/koopman/test-util-refactor

Refactor mock clients in tests

tests/test__internal.py

@@ -2,24 +2,17 @@
 import ocs
 import pytest
 
-from unittest.mock import MagicMock
-from ocs.ocs_agent import OpSession
 from ocs.ocs_client import OCSReply
 
 from sorunlib._internal import check_response, check_running
 
-os.environ["OCS_CONFIG_DIR"] = "./test_util/"
+from util import create_session as create_unencoded_session
 
+os.environ["OCS_CONFIG_DIR"] = "./test_util/"
 
-def create_session(op_name, status=None, success=None):
-    """Create an OpSession with a mocked app for testing."""
-    mock_app = MagicMock()
-    session = OpSession(1, op_name, app=mock_app)
-    session.op_name = 'test_op'
-    if status is not None:
-        session.set_status(status)
-    session.success = success
 
+def create_session(*args, **kwargs):
+    session = create_unencoded_session(*args, **kwargs)
     return session.encoded()
 
 

tests/test_acu.py

@@ -3,74 +3,40 @@
 
 import pytest
 
-from unittest.mock import MagicMock, patch
-
-import ocs
 from ocs.ocs_client import OCSReply
 from sorunlib import acu
 
-from util import create_session
-
-
-def create_acu_client(platform_type):
-    """Create an ACU client with mock monitor Process session.data.
-
-    Args:
-        platform_type (str): Either 'satp' or 'ccat'.
-
-    """
-    acu_client = MagicMock()
-    session = create_session('monitor')
-    session.data = {'PlatformType': platform_type}
-    reply = OCSReply(ocs.OK, 'msg', session.encoded())
-    acu_client.monitor.status = MagicMock(return_value=reply)
-
-    return acu_client
+from util import create_patch_clients
 
 
-def mocked_clients(test_mode):
-    clients = {'acu': MagicMock(),
-               'smurf': [MagicMock(), MagicMock(), MagicMock()]}
+patch_clients_satp = create_patch_clients('satp')
+patch_clients_lat = create_patch_clients('ccat')
 
-    return clients
 
-
-@patch('sorunlib.create_clients', mocked_clients)
-def test_move_to():
-    acu.run.initialize(test_mode=True)
+def test_move_to(patch_clients_satp):
     acu.move_to(180, 60)
     acu.run.CLIENTS['acu'].go_to.assert_called_with(az=180, el=60)
 
 
-@patch('sorunlib.create_clients', mocked_clients)
-def test_set_boresight():
-    acu.run.initialize(test_mode=True)
-    acu.run.CLIENTS['acu'] = create_acu_client('satp')
+def test_move_to_failed(patch_clients_satp):
+    mocked_response = OCSReply(
+        0, 'msg', {'success': False, 'op_name': 'go_to'})
+    acu.run.CLIENTS['acu'].go_to.side_effect = [mocked_response]
+    with pytest.raises(RuntimeError):
+        acu.move_to(180, 90)
+
+
+def test_set_boresight(patch_clients_satp):
     acu.set_boresight(20)
     acu.run.CLIENTS['acu'].set_boresight.assert_called_with(target=20)
 
 
-@patch('sorunlib.create_clients', mocked_clients)
-def test_set_boresight_lat():
-    acu.run.initialize(test_mode=True)
-    acu.run.CLIENTS['acu'] = create_acu_client('ccat')
+def test_set_boresight_lat(patch_clients_lat):
     with pytest.raises(RuntimeError):
         acu.set_boresight(20)
 
 
-@patch('sorunlib.create_clients', mocked_clients)
-def test_move_to_failed():
-    acu.run.initialize()
-    mocked_response = OCSReply(
-        0, 'msg', {'success': False, 'op_name': 'go_to'})
-    acu.run.CLIENTS['acu'].go_to.side_effect = [mocked_response]
-    with pytest.raises(RuntimeError):
-        acu.move_to(180, 90)
-
-
-@patch('sorunlib.create_clients', mocked_clients)
-def test_set_scan_params():
-    acu.run.initialize(test_mode=True)
+def test_set_scan_params(patch_clients_satp):
     acu.set_scan_params(az_speed=2, az_accel=2, reset=True)
     acu.run.CLIENTS['acu'].set_scan_params.assert_called_with(
         az_speed=2,

tests/test_seq.py

@@ -1,5 +1,6 @@
 import os
 os.environ["OCS_CONFIG_DIR"] = "./test_util/"
+os.environ["SORUNLIB_CONFIG"] = "./data/example_config.yaml"
 import datetime as dt
 
 import pytest
@@ -8,27 +9,20 @@
 import sorunlib
 from sorunlib import seq
 
+from util import create_patch_clients
 
-def mocked_clients(test_mode):
-    clients = {'acu': MagicMock(),
-               'smurf': [MagicMock(), MagicMock(), MagicMock()]}
 
-    return clients
+patch_clients = create_patch_clients('satp')
 
 
-@patch('sorunlib.create_clients', mocked_clients)
 @patch('sorunlib.commands.time.sleep', MagicMock())
-def test_scan():
-    seq.run.initialize(test_mode=True)
+def test_scan(patch_clients):
     target = dt.datetime.now() + dt.timedelta(seconds=1)
     seq.scan(description='test', stop_time=target.isoformat(), width=20.)
 
 
-@patch('sorunlib.create_clients', mocked_clients)
 @patch('sorunlib.commands.time.sleep', MagicMock())
-def test_scan_failed_to_start():
-    seq.run.initialize(test_mode=True)
-
+def test_scan_failed_to_start(patch_clients):
     # Setup mock OCSReply without session object
     mock_reply = MagicMock()
     mock_reply.session = None

tests/test_smurf.py

@@ -8,16 +8,13 @@
 from ocs.ocs_client import OCSReply
 
 from sorunlib import smurf
-from util import mocked_clients
+from util import create_patch_clients
 
 
 # Use pytest-mock plugin to patch CLIENTS on all tests
-@pytest.fixture(autouse=True)
-def patch_clients(mocker):
-    mocker.patch('sorunlib.smurf.run.CLIENTS', mocked_clients())
+patch_clients = create_patch_clients('satp', autouse=True)
 
 
-@patch('sorunlib.create_clients', mocked_clients)
 def test_set_targets():
     smurf.set_targets(['smurf1'])
     assert len(smurf.run.CLIENTS['smurf']) == 1

tests/test_sorunlib.py

@@ -0,0 +1,13 @@
+from unittest.mock import MagicMock, patch
+
+from util import mocked_clients
+
+import sorunlib
+from sorunlib import initialize
+
+
+@patch('sorunlib.commands.time.sleep', MagicMock())
+@patch('sorunlib.create_clients', mocked_clients)
+def test_initialize():
+    initialize()
+    assert sorunlib.CLIENTS is not None

tests/test_wiregrid.py

@@ -10,7 +10,9 @@
 from ocs.ocs_client import OCSReply
 from sorunlib import wiregrid
 
-from util import create_session
+from util import create_session, create_patch_clients, mocked_clients
+
+patch_clients = create_patch_clients('satp')
 
 
 def create_acu_client(az, el, boresight):
@@ -111,17 +113,6 @@ def create_encoder_client():
     return client
 
 
-def mocked_clients():
-    clients = {'acu': MagicMock(),
-               'smurf': [MagicMock(), MagicMock(), MagicMock()],
-               'wiregrid': {'actuator': MagicMock(),
-                            'encoder': MagicMock(),
-                            'kikusui': MagicMock(),
-                            'labjack': MagicMock()}}
-
-    return clients
-
-
 @patch('sorunlib.wiregrid.run.CLIENTS', mocked_clients())
 def test_insert():
     wiregrid.insert()
@@ -193,9 +184,7 @@ def test__check_temperature_sensors():
     wiregrid.run.CLIENTS['wiregrid']['labjack'].acq.status.assert_called_once()
 
 
-@patch('sorunlib.wiregrid.run.CLIENTS', mocked_clients())
-def test__check_telescope_position():
-    wiregrid.run.CLIENTS['acu'] = create_acu_client(180, 50, 0)
+def test__check_telescope_position(patch_clients):
     wiregrid._check_telescope_position()
     wiregrid.run.CLIENTS['acu'].monitor.status.assert_called_once()
 
@@ -244,9 +233,8 @@ def test__check_agents_online():
 
 
 @pytest.mark.parametrize('continuous', [(True), (False)])
-@patch('sorunlib.wiregrid.run.CLIENTS', mocked_clients())
 @patch('sorunlib.wiregrid.time.sleep', MagicMock())
-def test_calibrate_stepwise(continuous):
+def test_calibrate_stepwise(patch_clients, continuous):
     # Setup all mock clients
     wiregrid.run.CLIENTS['acu'] = create_acu_client(180, 50, 0)
     wiregrid.run.CLIENTS['wiregrid']['actuator'] = create_actuator_client(motor=1)

tests/util.py

@@ -1,12 +1,20 @@
+import pytest
+
 from unittest.mock import MagicMock
 
+import ocs
 from ocs.ocs_agent import OpSession
+from ocs.ocs_client import OCSReply
 
 
-def create_session(op_name):
+def create_session(op_name, status=None, success=None):
     """Create an OpSession with a mocked app for testing."""
     mock_app = MagicMock()
     session = OpSession(1, op_name, app=mock_app)
+    session.op_name = 'test_op'
+    if status is not None:
+        session.set_status(status)
+    session.success = success
 
     return session
 
@@ -35,11 +43,57 @@ def _mock_smurf_client(instance_id):
     return smurf
 
 
+def _mock_acu_client(platform_type, az=180, el=50, boresight=0):
+    """Create an ACU client with mock monitor Process session.data."""
+    acu = MagicMock()
+    session = create_session('monitor')
+    session.data = {'PlatformType': platform_type,
+                    'StatusDetailed': {'Azimuth current position': az,
+                                       'Elevation current position': el,
+                                       'Boresight current position': boresight}}
+    reply = OCSReply(ocs.OK, 'msg', session.encoded())
+    acu.monitor.status = MagicMock(return_value=reply)
+
+    session = create_session('generate_scan', status='running')
+    reply = OCSReply(ocs.OK, 'msg', session.encoded())
+    acu.generate_scan.start = MagicMock(return_value=reply)
+    acu.generate_scan.status = MagicMock(return_value=reply)
+
+    return acu
+
+
 def mocked_clients(**kwargs):
+    platform_type = kwargs.get('platform_type', 'satp')
+
     smurf_ids = ['smurf1', 'smurf2', 'smurf3']
     smurfs = [_mock_smurf_client(id_) for id_ in smurf_ids]
 
-    clients = {'acu': MagicMock(),
-               'smurf': smurfs}
+    clients = {'acu': _mock_acu_client(platform_type),
+               'smurf': smurfs,
+               'wiregrid': {'actuator': MagicMock(),
+                            'encoder': MagicMock(),
+                            'kikusui': MagicMock(),
+                            'labjack': MagicMock()}}
 
     return clients
+
+
+def create_patch_clients(platform_type, autouse=False):
+    """Create patch_clients fixture that patches out the global CLIENTS list
+    with a set of mocked clients using the ``pytest-mock`` plugin.
+
+    Args:
+        platform_type (str): Either 'satp' or 'ccat'.
+        autouse (bool): Whether to enable 'autouse' on the fixture. This will
+            enable the fixture for all tests within a test module.
+
+    Returns:
+        function: A pytest fixture that patches out ``sorunlib.CLIENTS`` with a
+            set of mocked clients.
+
+    """
+    @pytest.fixture(autouse=autouse)
+    def patch_clients(mocker):
+        mocker.patch('sorunlib.CLIENTS', mocked_clients(platform_type=platform_type))
+
+    return patch_clients