test_pepu.py 3.68 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 20 21
def pepu(request):
    hostname = request.config.getoption('--pepu')
    pepu = PEPU('test', {'tcp': {'url': hostname}})
Vincent Michel's avatar
Vincent Michel committed
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
    try:
        pepu.calc_channels[1].formula = '1.5'
        pepu.calc_channels[2].formula = '-1.5'
        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):
    assert pepu.app_name == 'PEPU'
    assert pepu.version == '00.01'
    assert pepu.up_time > 0
    assert pepu.sys_info.startswith('DANCE')
    uptime, uname = pepu.dance_info.splitlines()
    assert uptime.startswith('UPTIME')
    assert uname.startswith('UNAME')
    assert pepu.config.startswith('# %APPNAME% PEPU')


@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 69
@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
70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94
@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(
        name='TEST',
        trigger=Trigger(Signal.SOFT, Signal.SOFT),
        frequency=10, nb_points=blocks * block_size,
95
        sources=('CALC2', 'CALC1'),
Vincent Michel's avatar
Vincent Michel committed
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114
        overwrite=True)
    # Argument testing
    assert stream.name == 'TEST'
    assert stream.trigger == Trigger(Signal.SOFT, Signal.SOFT)
    assert stream.frequency == 10
    assert stream.nb_points == blocks * block_size
    assert stream.sources == ['CALC1', 'CALC2']
    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)
115 116 117 118 119 120 121 122 123 124 125
            assert data['CALC1'].tolist() == [1.5] * block_size
            assert data['CALC2'].tolist() == [-1.5] * block_size


def test_timescan(pepu):
    scan = scans.timescan(
        0.1, pepu.counters.CALC1, pepu.counters.CALC2,
        npoints=3, return_scan=True, save=False)
    data = scans.get_data(scan)
    assert data['CALC1'].tolist() == [1.5] * 3
    assert data['CALC2'].tolist() == [-1.5] * 3