test_pepu.py 3.73 KB
Newer Older
Vincent Michel's avatar
Vincent Michel committed
1 2 3 4 5 6 7 8 9 10
"""PEPU hardware tests.

Run with:

    $ pytest -c /dev/null tests/controllers_hw/pepu.py -v \
    --cov bliss.controllers.pepu --cov-report html --cov-report term

"""

import pytest
11 12

from bliss.common import scans
13
from bliss.controllers.pepu import PEPU, Signal, Trigger, ChannelMode
Vincent Michel's avatar
Vincent Michel committed
14

15 16
pytestmark = pytest.mark.pepu

Vincent Michel's avatar
Vincent Michel committed
17 18

@pytest.fixture
19
def pepu(request):
20 21
    hostname = request.config.getoption("--pepu")
    pepu = PEPU("test", {"tcp": {"url": hostname}})
Vincent Michel's avatar
Vincent Michel committed
22
    try:
23 24
        pepu.calc_channels[1].formula = "1.5"
        pepu.calc_channels[2].formula = "-1.5"
Vincent Michel's avatar
Vincent Michel committed
25 26 27 28 29 30 31 32
        pepu.out_channels[7].source = pepu.calc_channels[1].name
        pepu.out_channels[8].source = pepu.calc_channels[2].name
        yield pepu
    finally:
        pepu.conn.close()


def test_simple_connection(pepu):
33 34
    assert pepu.app_name == "PEPU"
    assert pepu.version == "00.01"
Vincent Michel's avatar
Vincent Michel committed
35
    assert pepu.up_time > 0
36
    assert pepu.sys_info.startswith("DANCE")
Vincent Michel's avatar
Vincent Michel committed
37
    uptime, uname = pepu.dance_info.splitlines()
38 39 40
    assert uptime.startswith("UPTIME")
    assert uname.startswith("UNAME")
    assert pepu.config.startswith("# %APPNAME% PEPU")
Vincent Michel's avatar
Vincent Michel committed
41 42 43 44 45 46 47 48


@pytest.mark.parametrize("channel_id", range(1, 7))
def test_read_in_channels(pepu, channel_id):
    channel = pepu.in_channels[channel_id]
    assert channel.value in [-1., 0.]


49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68
@pytest.mark.parametrize("channel_id", range(1, 7))
def test_in_channel_config(pepu, channel_id):
    channel = pepu.in_channels[channel_id]

    # test state
    enabled = channel.enabled
    assert channel.enabled is True or channel.enabled is False

    # disabled the state
    channel.enabled = False
    assert channel.enabled is False

    # test mode
    mode = channel.mode
    assert channel.mode in tuple(ChannelMode)

    channel.mode = ChannelMode.BISS
    assert channel.mode == ChannelMode.BISS


Vincent Michel's avatar
Vincent Michel committed
69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90
@pytest.mark.parametrize("channel_id", [7, 8])
def test_read_out_channels(pepu, channel_id):
    channel = pepu.out_channels[channel_id]
    value = channel.value
    pytest.xfail()
    assert value in (1.5, -1.5)


@pytest.mark.parametrize("channel_id", [1, 2])
def test_read_calc_channels(pepu, channel_id):
    channel = pepu.calc_channels[channel_id]
    value = channel.value
    pytest.xfail()
    assert value in (1.5, -1.5)


@pytest.mark.parametrize("acquisitions", [1, 2, 10])
@pytest.mark.parametrize("blocks", [1, 2, 10])
@pytest.mark.parametrize("block_size", [1, 2, 10])
def test_streams_acquisition(pepu, acquisitions, blocks, block_size):
    # Create stream
    stream = pepu.create_stream(
91
        name="TEST",
Vincent Michel's avatar
Vincent Michel committed
92
        trigger=Trigger(Signal.SOFT, Signal.SOFT),
93 94 95 96 97
        frequency=10,
        nb_points=blocks * block_size,
        sources=("CALC2", "CALC1"),
        overwrite=True,
    )
Vincent Michel's avatar
Vincent Michel committed
98
    # Argument testing
99
    assert stream.name == "TEST"
Vincent Michel's avatar
Vincent Michel committed
100 101 102
    assert stream.trigger == Trigger(Signal.SOFT, Signal.SOFT)
    assert stream.frequency == 10
    assert stream.nb_points == blocks * block_size
103
    assert stream.sources == ["CALC1", "CALC2"]
Vincent Michel's avatar
Vincent Michel committed
104 105 106 107 108 109 110 111 112 113 114 115
    assert not stream.active
    # Loop over acquisitions
    for _ in range(acquisitions):
        stream.start()
        # Loop over blocks
        for _ in range(blocks):
            # Loop over points
            for _ in range(block_size):
                pepu.software_trigger()
            # Read block
            assert stream.nb_points_ready == block_size
            data = stream.read(n=block_size)
116 117
            assert data["CALC1"].tolist() == [1.5] * block_size
            assert data["CALC2"].tolist() == [-1.5] * block_size
118 119 120 121


def test_timescan(pepu):
    scan = scans.timescan(
122 123 124 125 126 127 128
        0.1,
        pepu.counters.CALC1,
        pepu.counters.CALC2,
        npoints=3,
        return_scan=True,
        save=False,
    )
129
    data = scan.get_data()
130 131
    assert data["CALC1"].tolist() == [1.5] * 3
    assert data["CALC2"].tolist() == [-1.5] * 3