1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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
|
import pytest
import grpc
import labgrid.remote.generated.labgrid_coordinator_pb2_grpc as labgrid_coordinator_pb2_grpc
import labgrid.remote.generated.labgrid_coordinator_pb2 as labgrid_coordinator_pb2
@pytest.fixture(scope='function')
def channel_stub():
import queue
queue = queue.Queue()
channel = grpc.insecure_channel("127.0.0.1:20408")
stub = labgrid_coordinator_pb2_grpc.CoordinatorStub(channel)
def generate_startup(queue):
msg = labgrid_coordinator_pb2.ClientInMessage()
msg.startup.version = "2.0.0"
msg.startup.name = "testclient"
messages = [
msg
]
for msg in messages:
yield msg
while True:
msg = queue.get()
yield msg
queue.task_done()
stream = stub.ClientStream(generate_startup(queue))
yield stub
channel.close()
@pytest.fixture(scope='function')
def coordinator_place(channel_stub):
name = "test"
place = labgrid_coordinator_pb2.AddPlaceRequest(name=name)
res = channel_stub.AddPlace(place)
assert res, f"There was an error: {res}"
return channel_stub
def test_startup(coordinator):
pass
def test_coordinator_add_place(coordinator, channel_stub):
name = "test"
place = labgrid_coordinator_pb2.AddPlaceRequest(name=name)
res = channel_stub.AddPlace(place)
assert res, f"There was an error: {res}"
def test_coordinator_del_place(coordinator, channel_stub):
name = "test"
place = labgrid_coordinator_pb2.AddPlaceRequest(name=name)
res = channel_stub.AddPlace(place)
assert res, f"There was an error: {res}"
place = labgrid_coordinator_pb2.DeletePlaceRequest(name=name)
res = channel_stub.DeletePlace(place)
assert res, f"There was an error: {res}"
def test_coordinator_get_places(coordinator, channel_stub):
name = "test"
place = labgrid_coordinator_pb2.AddPlaceRequest(name=name)
res = channel_stub.AddPlace(place)
assert res, f"There was an error: {res}"
name = "test2"
place = labgrid_coordinator_pb2.AddPlaceRequest(name=name)
res = channel_stub.AddPlace(place)
assert res, f"There was an error: {res}"
request = labgrid_coordinator_pb2.GetPlacesRequest()
res = channel_stub.GetPlaces(request)
from labgrid.remote.common import Place
places = set()
names = set()
for pb2 in res.places:
place = Place.from_pb2(pb2)
places.add(place)
names.add(place.name)
assert len(places) == 2, f"Returned places not two: {places}"
assert set(names) == {"test", "test2"}, f"There was an error: {res}"
def test_coordinator_exporter_session(coordinator, channel_stub):
import queue
queue = queue.Queue()
def generate_startup(queue):
msg = labgrid_coordinator_pb2.ExporterInMessage()
msg.startup.version = "2.0.0"
msg.startup.name = "testporter"
messages = [
msg
]
for msg in messages:
yield msg
while True:
msg = queue.get()
yield msg
queue.task_done()
coordinator = channel_stub.ExporterStream(generate_startup(queue), wait_for_ready=True)
def test_coordinator_place_acquire(coordinator, coordinator_place):
stub = coordinator_place
res = stub.AcquirePlace(labgrid_coordinator_pb2.AcquirePlaceRequest(placename="test"))
assert res
def test_coordinator_place_acquire_release(coordinator, coordinator_place):
stub = coordinator_place
res = stub.AcquirePlace(labgrid_coordinator_pb2.AcquirePlaceRequest(placename="test"))
assert res
res = stub.ReleasePlace(labgrid_coordinator_pb2.ReleasePlaceRequest(placename="test"))
assert res
def test_coordinator_place_add_alias(coordinator, coordinator_place):
stub = coordinator_place
res = stub.AddPlaceAlias(labgrid_coordinator_pb2.AddPlaceAliasRequest(placename="test", alias="testalias"))
assert res
def test_coordinator_place_add_remove_alias(coordinator, coordinator_place):
stub = coordinator_place
res = stub.AddPlaceAlias(labgrid_coordinator_pb2.AddPlaceAliasRequest(placename="test", alias="testalias"))
assert res
res = stub.DeletePlaceAlias(labgrid_coordinator_pb2.DeletePlaceAliasRequest(placename="test", alias="testalias"))
assert res
def test_coordinator_place_set_tags(coordinator, coordinator_place):
stub = coordinator_place
res = stub.SetPlaceTags(labgrid_coordinator_pb2.SetPlaceTagsRequest(placename="test", tags={"one": "two"}))
assert res
def test_coordinator_place_set_comment(coordinator, coordinator_place):
stub = coordinator_place
res = stub.SetPlaceComment(labgrid_coordinator_pb2.SetPlaceCommentRequest(placename="test", comment="testcomment"))
assert res
def test_coordinator_place_add_match(coordinator, coordinator_place):
stub = coordinator_place
res = stub.AddPlaceMatch(labgrid_coordinator_pb2.AddPlaceMatchRequest(placename="test", pattern="this/test/pattern"))
assert res
def test_coordinator_place_add_delete_match(coordinator, coordinator_place):
stub = coordinator_place
res = stub.AddPlaceMatch(labgrid_coordinator_pb2.AddPlaceMatchRequest(placename="test", pattern="this/test/pattern"))
assert res
res = stub.DeletePlaceMatch(labgrid_coordinator_pb2.DeletePlaceMatchRequest(placename="test", pattern="this/test/pattern"))
assert res
def test_coordinator_place_allow(coordinator, coordinator_place):
stub = coordinator_place
res = stub.AcquirePlace(labgrid_coordinator_pb2.AcquirePlaceRequest(placename="test"))
assert res
res = stub.AllowPlace(labgrid_coordinator_pb2.AllowPlaceRequest(placename="test", user="othertest"))
assert res
def test_coordinator_create_reservation(coordinator, coordinator_place):
tags = {"board": "test"}
stub = coordinator_place
res = stub.SetPlaceTags(labgrid_coordinator_pb2.SetPlaceTagsRequest(placename="test", tags=tags))
assert res
res = stub.CreateReservation(labgrid_coordinator_pb2.CreateReservationRequest(filters={
"main": labgrid_coordinator_pb2.Reservation.Filter(filter={"board": "test"}),
}, prio=1.0))
assert res
res: labgrid_coordinator_pb2.CreateReservationResponse
assert len(res.reservation.token) > 0
|
