|
| 1 | +from typing import Any |
| 2 | + |
1 | 3 | import pytest |
2 | 4 | from pytest_mock import MockerFixture |
3 | 5 |
|
4 | | -from fastcs.attributes import AttrR, AttrRW |
| 6 | +from fastcs.attributes import AttrR, AttrRW, AttrW |
5 | 7 | from fastcs.backends.epics.ioc import ( |
6 | 8 | EPICS_MAX_NAME_LENGTH, |
7 | 9 | EpicsIOC, |
8 | 10 | _add_attr_pvi_info, |
9 | 11 | _add_pvi_info, |
10 | 12 | _add_sub_controller_pvi_info, |
| 13 | + _get_input_record, |
| 14 | + _get_output_record, |
11 | 15 | ) |
12 | 16 | from fastcs.controller import Controller |
13 | 17 | from fastcs.cs_methods import Command |
14 | | -from fastcs.datatypes import Int |
| 18 | +from fastcs.datatypes import Int, String |
| 19 | +from fastcs.exceptions import FastCSException |
15 | 20 | from fastcs.mapping import Mapping |
16 | 21 |
|
17 | 22 | DEVICE = "DEVICE" |
18 | 23 |
|
| 24 | +SEVENTEEN_VALUES = [str(i) for i in range(1, 18)] |
| 25 | +ONOFF_STATES = {"ZRST": "disabled", "ONST": "enabled"} |
| 26 | + |
| 27 | + |
| 28 | +@pytest.mark.parametrize( |
| 29 | + "attribute,record_type,kwargs", |
| 30 | + ( |
| 31 | + (AttrR(String()), "longStringIn", {}), |
| 32 | + ( |
| 33 | + AttrR(String(), allowed_values=list(ONOFF_STATES.values())), |
| 34 | + "mbbIn", |
| 35 | + ONOFF_STATES, |
| 36 | + ), |
| 37 | + (AttrR(String(), allowed_values=SEVENTEEN_VALUES), "longStringIn", {}), |
| 38 | + ), |
| 39 | +) |
| 40 | +def test_get_input_record( |
| 41 | + attribute: AttrR, |
| 42 | + record_type: str, |
| 43 | + kwargs: dict[str, Any], |
| 44 | + mocker: MockerFixture, |
| 45 | +): |
| 46 | + builder = mocker.patch("fastcs.backends.epics.ioc.builder") |
| 47 | + |
| 48 | + pv = "PV" |
| 49 | + _get_input_record(pv, attribute) |
| 50 | + |
| 51 | + getattr(builder, record_type).assert_called_once_with(pv, **kwargs) |
| 52 | + |
| 53 | + |
| 54 | +def test_get_input_record_raises(mocker: MockerFixture): |
| 55 | + # Pass a mock as attribute to provoke the fallback case matching on datatype |
| 56 | + with pytest.raises(FastCSException): |
| 57 | + _get_input_record("PV", mocker.MagicMock()) |
| 58 | + |
| 59 | + |
| 60 | +@pytest.mark.parametrize( |
| 61 | + "attribute,record_type,kwargs", |
| 62 | + ( |
| 63 | + (AttrR(String()), "longStringOut", {}), |
| 64 | + ( |
| 65 | + AttrR(String(), allowed_values=list(ONOFF_STATES.values())), |
| 66 | + "mbbOut", |
| 67 | + ONOFF_STATES, |
| 68 | + ), |
| 69 | + (AttrR(String(), allowed_values=SEVENTEEN_VALUES), "longStringOut", {}), |
| 70 | + ), |
| 71 | +) |
| 72 | +def test_get_output_record( |
| 73 | + attribute: AttrW, |
| 74 | + record_type: str, |
| 75 | + kwargs: dict[str, Any], |
| 76 | + mocker: MockerFixture, |
| 77 | +): |
| 78 | + builder = mocker.patch("fastcs.backends.epics.ioc.builder") |
| 79 | + update = mocker.MagicMock() |
| 80 | + |
| 81 | + pv = "PV" |
| 82 | + _get_output_record(pv, attribute, on_update=update) |
| 83 | + |
| 84 | + getattr(builder, record_type).assert_called_once_with( |
| 85 | + pv, always_update=True, on_update=update, **kwargs |
| 86 | + ) |
| 87 | + |
| 88 | + |
| 89 | +def test_get_output_record_raises(mocker: MockerFixture): |
| 90 | + # Pass a mock as attribute to provoke the fallback case matching on datatype |
| 91 | + with pytest.raises(FastCSException): |
| 92 | + _get_output_record("PV", mocker.MagicMock(), on_update=mocker.MagicMock()) |
| 93 | + |
19 | 94 |
|
20 | 95 | def test_ioc(mocker: MockerFixture, mapping: Mapping): |
21 | 96 | builder = mocker.patch("fastcs.backends.epics.ioc.builder") |
|
0 commit comments