test_alias_basics.py 7.28 KB
Newer Older
Linus Pithan's avatar
Linus Pithan committed
1
import pytest
2
import gevent
3
4

import bliss
Piergiorgio Pancino's avatar
Piergiorgio Pancino committed
5
from bliss.shell.standard import wm
6
from bliss import global_map
Linus Pithan's avatar
Linus Pithan committed
7
from bliss.common.scans import loopscan
8
from bliss.common import event
9

10
11
alias_dump = """Alias    Original fullname
-------  ----------------------------------
12
13
14
15
16
mot0     m0
robyy    roby
robzz    robz
dtime    simu1:deadtime_det0
rtime    simu1:realtime_det0
Cyril Guilloud's avatar
Cyril Guilloud committed
17
ltime    simu1:energy_livetime_det0
18
19
dtime1   simu1:deadtime_det1
dtime2   simu1:deadtime_det2
20
robenc   encoder:m1enc
21
22
myroi    lima_simulator:roi_counters:r1_sum
myroi3   lima_simulator:roi_counters:r3_sum"""
23

24
25
alias_dump2 = """Alias    Original fullname
-------  ----------------------------------
26
27
28
29
30
mot0     m0
robyy    roby
robzz    robz
dtime    simu1:deadtime_det0
rtime    simu1:realtime_det0
Cyril Guilloud's avatar
Cyril Guilloud committed
31
ltime    simu1:energy_livetime_det0
32
33
dtime1   simu1:deadtime_det1
dtime2   simu1:deadtime_det2
34
robenc   encoder:m1enc
35
36
myroi    lima_simulator:roi_counters:r1_sum
myroi3   lima_simulator:roi_counters:r3_sum
37
38
m22      m1"""

39
40
alias_dump3 = """Alias    Original fullname
-------  ----------------------------------
41
42
43
44
45
mot0     m0
robyy    roby
robzz    robz
dtime    simu1:deadtime_det0
rtime    simu1:realtime_det0
Cyril Guilloud's avatar
Cyril Guilloud committed
46
ltime    simu1:energy_livetime_det0
47
48
dtime1   simu1:deadtime_det1
dtime2   simu1:deadtime_det2
49
robenc   encoder:m1enc
50
51
myroi    lima_simulator:roi_counters:r1_sum
myroi3   lima_simulator:roi_counters:r3_sum
52
m22      m1
53
myr2sum  lima_simulator:roi_counters:r2_sum"""
Linus Pithan's avatar
Linus Pithan committed
54
55
56
57
58
59
60
61
62
63


def compare_line_by_line(string1, string2):
    list1 = string1.split("\n")
    list2 = string2.split("\n")
    assert len(list1) == len(list2)
    for l1, l2 in zip(list1, list2):
        assert l1 == l2


64
def test_alias_env_dict(alias_session):
65
    env_dict = alias_session.env_dict
Linus Pithan's avatar
Linus Pithan committed
66
    assert env_dict["ALIASES"] is not None
67
    compare_line_by_line(env_dict["ALIASES"].list_aliases(), alias_dump)
Linus Pithan's avatar
Linus Pithan committed
68
69


70
def test_alias_add_dynamically(alias_session, lima_simulator):
71
    env_dict = alias_session.env_dict
72
    lima_simulator = env_dict["lima_simulator"]
Linus Pithan's avatar
Linus Pithan committed
73

74
    m0 = alias_session.config.get("m0")
75
76
    with pytest.raises(RuntimeError):
        global_map.aliases.add("m22", m0)
Linus Pithan's avatar
Linus Pithan committed
77

78
    m1 = alias_session.config.get("m1")
79
    global_map.aliases.add("m22", m1)
Linus Pithan's avatar
Linus Pithan committed
80
    assert "m22" in env_dict
81
    assert m1 == env_dict["m22"]
Linus Pithan's avatar
Linus Pithan committed
82

83
    # create roi counter
Linus Pithan's avatar
Linus Pithan committed
84
    ALIASES = env_dict["ALIASES"]
85
    compare_line_by_line(ALIASES.list_aliases(), alias_dump2)
Linus Pithan's avatar
Linus Pithan committed
86

87
88
    ALIASES.add("myr2sum", lima_simulator.counters.r2_sum)
    compare_line_by_line(ALIASES.list_aliases(), alias_dump3)
Linus Pithan's avatar
Linus Pithan committed
89
90
91


def test_alias_duplication(alias_session):
92
    env_dict = alias_session.env_dict
Linus Pithan's avatar
Linus Pithan committed
93
94

    with pytest.raises(RuntimeError):
95
96
        global_map.aliases.add("blabla", env_dict["robyy"])

97
    simu1 = alias_session.config.get("simu1")
98
99
    with pytest.raises(RuntimeError):
        global_map.aliases.add("dtime", simu1.counters.deadtime_det0)
Linus Pithan's avatar
Linus Pithan committed
100

101
    m0 = alias_session.config.get("m0")
Linus Pithan's avatar
Linus Pithan committed
102
    with pytest.raises(RuntimeError):
103
        global_map.aliases.add("m1", m0)
Linus Pithan's avatar
Linus Pithan committed
104
105

    with pytest.raises(RuntimeError):
106
        global_map.aliases.add("robyy", m0)
Linus Pithan's avatar
Linus Pithan committed
107
108


109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def test_alias_overwriting_setup_object(alias_session):
    # related to https://gitlab.esrf.fr/bliss/bliss/merge_requests/1455#note_34532
    env_dict = alias_session.env_dict

    m2 = env_dict["m2"]
    roby = alias_session.config.get("roby")

    with pytest.raises(RuntimeError):
        global_map.aliases.add("roby", m2)

    with pytest.raises(RuntimeError):
        # 'bad' is an existing object in config (but not loaded in this session)
        global_map.aliases.add("bad", m2)


Linus Pithan's avatar
Linus Pithan committed
124
def test_alias_get(alias_session):
125
    env_dict = alias_session.env_dict
Linus Pithan's avatar
Linus Pithan committed
126
127
128
129

    r = env_dict["ALIASES"].get("robyy")
    assert r is env_dict["robyy"]

130
    assert env_dict["ALIASES"].get("does_not_exist") is None
Linus Pithan's avatar
Linus Pithan committed
131
132


133
def test_alias_included_session(alias_session):
134
    env_dict = alias_session.env_dict
135
    assert "mot0" in global_map.aliases.names_iter()
136
    m0 = alias_session.config.get("m0")
137
    assert env_dict["mot0"] == m0
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185


def test_alias_add_remove_set_get(alias_session):
    env_dict = alias_session.env_dict
    roby = alias_session.config.get("roby")
    assert env_dict["robyy"] == env_dict["roby"]
    ALIASES = env_dict["ALIASES"]
    assert ALIASES.get_alias("m0") == "mot0"

    with pytest.raises(RuntimeError):
        # mot0 alias already exists
        ALIASES.add("mot0", env_dict["lima_simulator"].counters.x)

    try:
        ALIASES.remove("mot0")
        ALIASES.remove("robyy")

        assert env_dict.get("mot0") is None
        assert env_dict.get("m0") is None  # aliased object has never been there

        ALIASES.add("mot0", roby)

        assert env_dict.get("roby") is None
        assert ALIASES.get("mot0") == roby
    finally:
        ALIASES.set("mot0", "m0")


def test_alias_original_object_removal(alias_session):
    env_dict = alias_session.env_dict
    ALIASES = env_dict["ALIASES"]

    assert "roby" not in env_dict  # only robyy is there

    diode = alias_session.config.get("diode")

    assert "diode" in env_dict

    # add alias with object
    ALIASES.add("dd", diode)

    assert "dd" in env_dict
    assert "diode" not in env_dict

    ALIASES.remove("dd")

    diode = alias_session.config.get("diode")
    # add alias with name
186
    ALIASES.add("dd", diode)
187
188
189
190
191
192
193
    assert "dd" in env_dict
    assert "diode" not in env_dict

    # check with a counter within a controller
    ALIASES.add("x", "lima_simulator:bpm:x")
    lima_simulator = env_dict["lima_simulator"]
    assert env_dict["x"] == lima_simulator.counters.x
194
195
196
197
198
199
200
201
202
203


def test_alias_wm(alias_session, capsys):
    robyy = alias_session.env_dict["robyy"]

    wm(robyy)

    output = capsys.readouterr()[0]
    assert "robyy" in output
    assert "inf" in output
Linus Pithan's avatar
Linus Pithan committed
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226


def test_encoder_alias(default_session):
    m1 = default_session.config.get("m1")
    m1enc = default_session.config.get("m1enc")
    mot_maxee = default_session.config.get("mot_maxee")
    mot_maxee_enc = default_session.config.get("mot_maxee_enc")

    ALIASES = default_session.env_dict["ALIASES"]

    s1 = loopscan(1, .01, mot_maxee_enc)
    assert "encoder:mot_maxee_enc" in list(s1.get_data())

    s2 = loopscan(1, .01, m1enc)
    assert "encoder:m1enc" in list(s2.get_data())

    ALIASES.add("mynewalias", m1enc.counter)

    s3 = loopscan(1, .01, m1enc)
    assert "encoder:mynewalias" in list(s3.get_data())

    s4 = loopscan(1, .01, mot_maxee_enc)
    assert "encoder:mot_maxee_enc" in list(s4.get_data())
227
228
229
230
231
232
233
234
235
236
237
238
239
240


def test_alias_connect(alias_session):
    robzz = alias_session.env_dict["robzz"]
    move_done = gevent.event.Event()

    def callback(done):
        if done:
            move_done.set()

    event.connect(robzz, "move_done", callback)

    robzz.rmove(0.1)
    assert move_done.is_set()
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271


def test_alias_lsobj(alias_session):
    """
    Ensure that lsobj() find all objects in the session.
    """
    obj_in_session = [
        "simu1",
        "robu",
        "lima_simulator",
        "m1",
        "m2",
        "mot0",
        "robyy",
        "robzz",
        "dtime",
        "rtime",
        "ltime",
        "dtime1",
        "dtime2",
        "robenc",
    ]
    obj_in_session.append("myroi")  # myroi and myroi3 are added in test_alias
    obj_in_session.append("myroi3")  #   session only during tests.
    obj_in_session.sort()

    obj_in_lsobj = bliss.common.standard._lsobj()
    obj_in_lsobj = list(set(obj_in_lsobj))  # for uniqueness of names.
    obj_in_lsobj.sort()

    assert obj_in_session == obj_in_lsobj