Allow NELM and length as aliases for waveform length

softioc/builder.py

@@ -134,6 +134,19 @@ def Action(name, **fields):
 }
 
 
+def _get_length(fields, default = None):
+    '''Helper function for getting 'length' or 'NELM' from arguments'''
+
+    if 'length' in fields:
+        assert 'NELM' not in fields, 'Cannot specify NELM and length together'
+        return fields.pop('length')
+    elif 'NELM' in fields:
+        return fields.pop('NELM')
+    else:
+        assert default is not None, 'Must specify waveform length'
+        return default
+
+
 def _waveform(value, fields):
     '''Helper routine for waveform construction.  If a value is given it is
     interpreted as an initial value and used to configure length and datatype
@@ -165,7 +178,7 @@ def _waveform(value, fields):
         # If a value is specified it should be the *only* non keyword argument.
         value, = value
         initial_value = device._require_waveform(value, datatype)
-        length = fields.pop('length', len(initial_value))
+        length = _get_length(fields, len(initial_value))
 
         # Special case for [u]int64: if the initial value comes in as 64 bit
         # integers we cannot represent that, so recast it as [u]int32
@@ -176,7 +189,7 @@ def _waveform(value, fields):
                 initial_value = numpy.require(initial_value, numpy.uint32)
     else:
         initial_value = numpy.array([], dtype = datatype)
-        length = fields.pop('length')
+        length = _get_length(fields)
     datatype = initial_value.dtype
 
     assert length > 0, 'Array cannot be of zero length'

tests/conftest.py

@@ -15,11 +15,20 @@
     sys.platform.startswith("win"), reason="Cothread doesn't work on windows"
 )
 
+# Default length used to initialise Waveform and longString records.
+# Length picked to match string record length, so we can re-use test strings.
+WAVEFORM_LENGTH = 40
+
+# Default timeout for many operations across testing
+TIMEOUT = 10  # Seconds
+
+def create_random_prefix():
+    """Create 12-character random string, for generating unique Device Names"""
+    return "".join(random.choice(string.ascii_uppercase) for _ in range(12))
+
 class SubprocessIOC:
     def __init__(self, ioc_py):
-        self.pv_prefix = "".join(
-            random.choice(string.ascii_uppercase) for _ in range(12)
-        )
+        self.pv_prefix = create_random_prefix()
         sim_ioc = os.path.join(os.path.dirname(__file__), ioc_py)
         cmd = [sys.executable, sim_ioc, self.pv_prefix]
         self.proc = subprocess.Popen(
@@ -76,9 +85,9 @@ def _clear_records():
     # https://github.com/dls-controls/pythonSoftIOC/issues/56
     RecordLookup._RecordDirectory.clear()
 
-@pytest.fixture
+@pytest.fixture(autouse=True)
 def clear_records():
-    """Fixture to delete all records before and after a test."""
+    """Deletes all records before and after every test"""
     _clear_records()
     yield
     _clear_records()
@@ -93,3 +102,21 @@ def enable_code_coverage():
         pass
     else:
         cleanup_on_sigterm()
+
+def select_and_recv(conn, expected_char = None):
+    """Wait for the given Connection to have data to receive, and return it.
+    If a character is provided check its correct before returning it.
+    This function imports Cothread, and so must NOT be called before any
+    multiprocessing sub-processes are spawned."""
+    from cothread import select
+    rrdy, _, _ = select([conn], [], [], TIMEOUT)
+    if rrdy:
+        val = conn.recv()
+    else:
+        pytest.fail("Did not receive expected char before TIMEOUT expired")
+
+    if expected_char:
+        assert val == expected_char, \
+            "Expected character did not match"
+
+    return val