add unit tests for synApps support

apstools/devices/tests/test_kohzu_monochromator.py

@@ -34,7 +34,7 @@ class MyKohzu(KohzuSeqCtl_Monochromator):
     m_y = Component(EpicsMotor, "m46")
     m_z = Component(EpicsMotor, "m47")
 
-    def into_control_range(self, p_theta=2, p_y=-15, p_z=90):
+    def into_control_range(self, p_theta=11, p_y=-18, p_z=90):
         """
         Move the Kohzu motors into range so the wavelength controls will work.
 
@@ -44,10 +44,13 @@ def into_control_range(self, p_theta=2, p_y=-15, p_z=90):
         args = []
         if self.m_theta.position < p_theta:
             args += [self.m_theta, p_theta]
+            yield from bps.mv(self.m_theta.velocity, 5)
         if self.m_y.position > p_y:
             args += [self.m_y, p_y]
+            yield from bps.mv(self.m_y.velocity, 5)
         if self.m_z.position < p_z:
             args += [self.m_z, p_z]
+            yield from bps.mv(self.m_z.velocity, 5)
         if (len(args) == 0):
             # all motors in range, no work to do, MUST yield something
             yield from bps.null()

apstools/synApps/calcout.py

@@ -92,6 +92,7 @@ class CalcoutRecord(EpicsRecordFloatFields, EpicsRecordDeviceCommonAll):
 
     :see: https://wiki-ext.aps.anl.gov/epics/index.php/RRM_3-14_Calcout
     """
+    enable = Cpt(EpicsSignal, "Enable", kind="omitted")
 
     units = Cpt(EpicsSignal, ".EGU", kind="config")
     precision = Cpt(EpicsSignal, ".PREC", kind="config")
@@ -131,7 +132,7 @@ def reset(self):
         self.units.put("")
         self.precision.put("5")
 
-        self.calculation.put("")
+        self.calculation.put("0")
         self.calculated_value.put(0)
         self.output_calculation.put("")
         self.output_value.put(0)

apstools/synApps/swait.py

@@ -97,6 +97,7 @@ class SwaitRecord(EpicsRecordDeviceCommonAll):
         ~reset
 
     """
+    enable = Cpt(EpicsSignal, "Enable", kind="config")
 
     precision = Cpt(EpicsSignal, ".PREC", kind="config")
     high_operating_range = Cpt(EpicsSignal, ".HOPR", kind="config")

apstools/synApps/tests/test_calcout.py

@@ -0,0 +1,54 @@
+import time
+
+from ..calcout import setup_incrementer_calcout
+from ..calcout import CalcoutRecord
+from ..calcout import UserCalcoutDevice
+
+
+IOC = "gp:"
+
+
+def test_read():
+    calcout = CalcoutRecord(f"{IOC}userCalcOut10", name="sseq")
+    assert calcout is not None
+    calcout.wait_for_connection()
+
+    assert len(calcout.read_attrs) == 12
+    assert len(calcout.configuration_attrs) == 52
+    assert len(calcout._summary().splitlines()) == 173
+
+
+def test_calcout_reset():
+    user = UserCalcoutDevice(IOC, name="user")
+    user.wait_for_connection()
+    user.enable.put("Enable")
+    assert len(user.read()) == 500
+
+    calcout = user.calcout10
+    assert isinstance(calcout, CalcoutRecord)
+    calcout.enable.put("E")  # Note: only "E"
+
+    setup_incrementer_calcout(calcout)
+    time.sleep(0.2)
+    assert calcout.description.get() == "incrementer"
+    assert calcout.calculation.get() == "(A+1) % B"
+    v1 = calcout.calculated_value.get()
+    time.sleep(0.2)
+    assert v1 < calcout.calculated_value.get()
+
+    calcout.reset()
+    assert calcout.description.get() == calcout.prefix
+    assert calcout.calculation.get() == "0"
+    v1 = calcout.calculated_value.get()
+    time.sleep(0.2)
+    assert v1 == calcout.calculated_value.get()
+
+# -----------------------------------------------------------------------------
+# :author:    Pete R. Jemian
+# :email:     [email protected]
+# :copyright: (c) 2017-2022, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Creative Commons Attribution 4.0 International Public License.
+#
+# The full license is in the file LICENSE.txt, distributed with this software.
+# -----------------------------------------------------------------------------

apstools/synApps/tests/test_iocstats.py

@@ -0,0 +1,58 @@
+import pytest
+
+from ..iocstats import IocStatsDevice
+
+
+IOC = "gp:"
+
+
+def test_read():
+    gp_info = IocStatsDevice(IOC, name="gp_info")
+    gp_info.wait_for_connection()
+    assert gp_info.connected
+
+    assert len(gp_info.read_attrs) == 19
+    assert len(gp_info.configuration_attrs) == 8
+    assert len(gp_info._summary().splitlines()) == 73
+
+
+@pytest.mark.parametrize(
+    "attr, expected, comparison",
+    [
+        ["access", "Running", None],
+        ["application_directory", "iocxxx", "end"],
+        ["engineer", "engineer", None],
+        ["epics_version", "EPICS", "start"],
+        ["kernel_version", "Linux", "start"],
+        ["kernel_version", "x86_64", "end"],
+        ["location", "location", None],
+        ["records_count", 100, ">="],
+    ]
+)
+def test_running(attr, expected, comparison):
+    gp_info = IocStatsDevice(IOC, name="gp_info")
+    gp_info.wait_for_connection()
+    assert gp_info.connected
+
+    assert hasattr(gp_info, attr)
+    value = getattr(gp_info, attr).get()
+    if comparison == "start":
+        assert value.startswith(expected)
+    elif comparison == "end":
+        assert value.endswith(expected)
+    elif comparison == "find":
+        assert value.find(expected) >= 0
+    elif comparison == ">=":
+        assert value >= 0
+    else:
+        assert value == expected
+
+# -----------------------------------------------------------------------------
+# :author:    Pete R. Jemian
+# :email:     [email protected]
+# :copyright: (c) 2017-2022, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Creative Commons Attribution 4.0 International Public License.
+#
+# The full license is in the file LICENSE.txt, distributed with this software.
+# -----------------------------------------------------------------------------

apstools/synApps/tests/test_sseq.py

@@ -6,22 +6,23 @@
 
 
 IOC = "gp:"
-TEST_SSEQ_PV = f"{IOC}userStringSeq10"
 
 
-def test_sseq_read():
-    sseq = SseqRecord(TEST_SSEQ_PV, name="sseq")
+def test_read():
+    sseq = SseqRecord(f"{IOC}userStringSeq10", name="sseq")
     assert sseq is not None
     sseq.wait_for_connection()
 
-    r = sseq.read()
-    assert len(r) == 20
+    assert len(sseq.read_attrs) == 31
+    assert len(sseq.configuration_attrs) == 109
+    assert len(sseq._summary().splitlines()) == 276
 
 
 def test_sseq_reset():
     user = UserStringSequenceDevice(IOC, name="user")
     user.wait_for_connection()
     user.enable.put("Enable")
+    assert len(user.read()) == 200
 
     sseq = user.sseq10
     assert isinstance(sseq, SseqRecord)
@@ -44,3 +45,13 @@ def test_sseq_reset():
     user.reset()
     assert step.input_pv.get() == ""
     assert step.string_value.get() == f"{0:.5f}"
+
+# -----------------------------------------------------------------------------
+# :author:    Pete R. Jemian
+# :email:     [email protected]
+# :copyright: (c) 2017-2022, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Creative Commons Attribution 4.0 International Public License.
+#
+# The full license is in the file LICENSE.txt, distributed with this software.
+# -----------------------------------------------------------------------------

apstools/synApps/tests/test_swait.py

@@ -0,0 +1,54 @@
+import time
+
+from ..swait import setup_random_number_swait
+from ..swait import SwaitRecord
+from ..swait import UserCalcsDevice
+
+
+IOC = "gp:"
+
+
+def test_read():
+    swait = SwaitRecord(f"{IOC}userCalc10", name="sseq")
+    assert swait is not None
+    swait.wait_for_connection()
+
+    assert len(swait.read_attrs) == 12
+    assert len(swait.configuration_attrs) == 61
+    assert len(swait._summary().splitlines()) == 152
+
+
+def test_swait_reset():
+    user = UserCalcsDevice(IOC, name="user")
+    user.wait_for_connection()
+    user.enable.put("Enable")
+    assert len(user.read()) == 130
+
+    swait = user.calc10
+    assert isinstance(swait, SwaitRecord)
+    swait.enable.put("E")  # Note: only "E"
+
+    setup_random_number_swait(swait)
+    time.sleep(0.2)
+    assert swait.description.get() == "uniform random numbers"
+    assert swait.calculation.get() == "RNDM"
+    v1 = swait.calculated_value.get()
+    time.sleep(0.2)
+    assert v1 != swait.calculated_value.get()
+
+    swait.reset()
+    assert swait.description.get() == swait.prefix
+    assert swait.calculation.get() == "0"
+    v1 = swait.calculated_value.get()
+    time.sleep(0.2)
+    assert v1 == swait.calculated_value.get()
+
+# -----------------------------------------------------------------------------
+# :author:    Pete R. Jemian
+# :email:     [email protected]
+# :copyright: (c) 2017-2022, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Creative Commons Attribution 4.0 International Public License.
+#
+# The full license is in the file LICENSE.txt, distributed with this software.
+# -----------------------------------------------------------------------------

apstools/utils/tests/test_listdevice.py

@@ -78,7 +78,7 @@ def test_listdevice(obj, length):
 @pytest.mark.parametrize(
     "obj, length, ref",
     [
-        (calcs, 126, EpicsSignalBase),
+        (calcs, 128, EpicsSignalBase),
         (calcs.calc5.description, 1, EpicsSignalBase),
         (signal, None, None),
         (motor, 19, EpicsSignalBase),
@@ -122,13 +122,13 @@ def test_spotchecks(function, row, column, value):
     "device, scope, ancient, length",
     [
         (calcs, "epics", False, 0),
-        (calcs, "epics", True, 126),
+        (calcs, "epics", True, 128),
         (calcs, "full", False, 4),
-        (calcs, "full", True, 130),
+        (calcs, "full", True, 132),
         (calcs, "read", False, 2),
         (calcs, "read", True, 28),
         (calcs, None, False, 4),
-        (calcs, None, True, 130),
+        (calcs, None, True, 132),
     ],
 )
 def test_listdevice_filters(device, scope, ancient, length):