diff --git a/tests/conftest.py b/tests/conftest.py index 0512b94..f4e30ea 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,6 +2,7 @@ import os import socket +import string import subprocess import sys import time as ttime @@ -106,4 +107,8 @@ def caproto_ioc_channel_types(wait=5): @pytest.fixture() def ophyd_channel_types(): - return OphydChannelTypes(OPHYD_PV_PREFIX, name="ophyd_channel_type") + dev = OphydChannelTypes(OPHYD_PV_PREFIX, name="ophyd_channel_type") + letters = iter(list(string.ascii_letters)) + for cpt in sorted(dev.component_names): + getattr(dev, cpt).put(next(letters)) + return dev diff --git a/tests/test_string_ioc.py b/tests/test_string_ioc.py index 31af7ac..af2e46c 100644 --- a/tests/test_string_ioc.py +++ b/tests/test_string_ioc.py @@ -1,5 +1,6 @@ from __future__ import annotations +import re import string import subprocess @@ -13,7 +14,7 @@ @pytest.mark.cloud_friendly() @pytest.mark.parametrize("value", [STRING_39, STRING_LONGER]) def test_strings( - # caproto_ioc_channel_types, + caproto_ioc_channel_types, ophyd_channel_types, value, ): @@ -31,6 +32,12 @@ def test_strings( with pytest.raises(ValueError, match="byte string too long"): ophyd_channel_types.string_type.put(value) + if len(value) <= LIMIT: + ophyd_channel_types.string_type_enum.put(value) + else: + with pytest.raises(ValueError, match="byte string too long"): + ophyd_channel_types.string_type_enum.put(value) + if len(value) <= LIMIT: ophyd_channel_types.char_type_as_string.put(value) else: @@ -40,13 +47,64 @@ def test_strings( ophyd_channel_types.char_type.put(value) +@pytest.mark.cloud_friendly() +@pytest.mark.needs_epics_core() +def test_cainfo(caproto_ioc_channel_types, ophyd_channel_types): + for cpt in sorted(ophyd_channel_types.component_names): + command = ["cainfo", getattr(ophyd_channel_types, cpt).pvname] + command_str = " ".join(command) + ret = subprocess.run( + command, + check=False, + capture_output=True, + ) + stdout = ret.stdout.decode() + print( + f"command: {command_str}\n {ret.returncode=}\n STDOUT:\n{ret.stdout.decode()}\n STDERR:\n{ret.stderr.decode()}\n" + ) + assert ret.returncode == 0 + if cpt in [ + "char_type_as_string", + "implicit_string_type", + "string_type", + "string_type_enum", + ]: + assert "Native data type: DBF_STRING" in stdout + else: + assert "Native data type: DBF_CHAR" in stdout + + @pytest.mark.cloud_friendly() @pytest.mark.needs_epics_core() @pytest.mark.parametrize("value", [STRING_39, STRING_LONGER]) -def test_with_epics_core(ophyd_channel_types, value): - for cpt in ophyd_channel_types.component_names: +def test_caput(caproto_ioc_channel_types, ophyd_channel_types, value): + option = "" + for cpt in sorted(ophyd_channel_types.component_names): + if cpt in [ + "char_type_as_string", + "implicit_string_type", + "string_type", + "string_type_enum", + ]: + option = "-s" + would_trim = True + else: + option = "-S" + would_trim = False + command = ["caput", option, getattr(ophyd_channel_types, cpt).pvname, value] + command_str = " ".join(command) ret = subprocess.run( - ["caput", "-S", getattr(ophyd_channel_types, cpt).pvname, value], + command, check=False, + capture_output=True, + ) + stdout = ret.stdout.decode() + print( + f"command: {command_str}\n {ret.returncode=}\n STDOUT:\n{stdout}\n STDERR:\n{ret.stderr.decode()}\n" ) - print(f"{cpt=}: {ret.returncode=}\n") + assert ret.returncode == 0 + actual = re.search("New : (.*)", stdout).group(1).split()[-1].rstrip() + if not would_trim or len(value) == LIMIT: + assert actual == value + else: + assert len(actual) < len(value)