0001-create-planitech-connector-27653.patch
functests/planitech/README | ||
---|---|---|
1 |
Functional tests for the passerelle Planitech connector |
|
2 | ||
3 |
Description |
|
4 |
=========== |
|
5 | ||
6 |
This test suite will use the web API of a passerelle Planitech connector |
|
7 |
to find an available reservation slot on a random place and perform a reservation. |
|
8 | ||
9 | ||
10 |
Usage |
|
11 |
===== |
|
12 | ||
13 |
You will need a running passerelle instance, whith a Planitech connector instance configured. |
|
14 |
Suppose that the Planitech connector instance is listening here : |
|
15 | ||
16 |
http://127.0.0.1:8000/planitech/planitech |
|
17 | ||
18 |
Then you would start the test suite with the following command: |
|
19 | ||
20 |
$ py.test --url=http://127.0.0.1:8000/planitech/planitech test_planitech.py |
functests/planitech/conftest.py | ||
---|---|---|
1 |
import pytest |
|
2 | ||
3 | ||
4 |
def pytest_addoption(parser): |
|
5 |
parser.addoption( |
|
6 |
"--url", help="Url of a passerelle Planitech connector instance") |
|
7 | ||
8 | ||
9 |
@pytest.fixture(scope='session') |
|
10 |
def conn(request): |
|
11 |
return request.config.getoption("--url") |
functests/planitech/test_planitech.py | ||
---|---|---|
1 |
import datetime |
|
2 |
import random |
|
3 |
import urllib |
|
4 | ||
5 | ||
6 |
import requests |
|
7 | ||
8 | ||
9 |
def test_main(conn): |
|
10 |
# get a free gap |
|
11 |
today = datetime.datetime.now().date().isoformat() |
|
12 |
query_string = urllib.urlencode({ |
|
13 |
'start_date': today, 'start_time': '10:00', 'end_time': '11:00' |
|
14 |
}) |
|
15 |
url = conn + '/getdays?%s' % query_string |
|
16 |
resp = requests.get(url) |
|
17 |
resp.raise_for_status() |
|
18 |
res = resp.json() |
|
19 |
assert res['err'] == 0 |
|
20 |
data = res['data'] |
|
21 |
assert data |
|
22 |
date = data[0]['id'] |
|
23 | ||
24 |
# get places |
|
25 |
query_string = urllib.urlencode({ |
|
26 |
'start_date': date, 'start_time': '10:00', 'end_time': '11:00' |
|
27 |
}) |
|
28 |
url = conn + '/getplaces?%s' % query_string |
|
29 |
resp = requests.get(url) |
|
30 |
resp.raise_for_status() |
|
31 |
res = resp.json() |
|
32 |
assert res['err'] == 0 |
|
33 |
data = res['data'] |
|
34 |
assert data |
|
35 |
place = data[random.randint(0, len(data) - 1)]['id'] |
|
36 | ||
37 |
# create reservation |
|
38 |
params = { |
|
39 |
'date': date, 'start_time': '10:00', 'end_time': '11:00', |
|
40 |
'place_id': place, 'price': 200 |
|
41 |
} |
|
42 |
url = conn + '/createreservation' |
|
43 |
resp = requests.post(url, json=params) |
|
44 |
resp.raise_for_status() |
|
45 |
res = resp.json() |
|
46 |
assert res['err'] == 0 |
|
47 |
data = res['data'] |
|
48 |
assert 'reservation_id' in data |
|
49 |
reservation_id = data['reservation_id'] |
|
50 | ||
51 |
# confirm reservation |
|
52 |
params = { |
|
53 |
'reservation_id': reservation_id, 'status': 'standard' |
|
54 |
} |
|
55 |
url = conn + '/updatereservation' |
|
56 |
resp = requests.post(url, json=params) |
|
57 |
resp.raise_for_status() |
|
58 |
res = resp.json() |
|
59 |
assert res['err'] == 0 |
|
60 |
data = res['data'] |
|
61 |
assert data |
|
62 | ||
63 |
# FORBIDDEN BY PLANITECH - NO RESERVATION CANCELATION ? |
|
64 |
# cancel reservation |
|
65 |
# params = { |
|
66 |
# 'reservation_id': reservation_id, 'status': 'invalid' |
|
67 |
# } |
|
68 |
# url = conn + '/updatereservation' |
|
69 |
# resp = requests.post(url, json=params) |
|
70 |
# resp.raise_for_status() |
|
71 |
# res = resp.json() |
|
72 |
# assert res['err'] == 0 |
|
73 |
# data = res['data'] |
|
74 |
# assert data |
passerelle/contrib/planitech/migrations/0001_initial.py | ||
---|---|---|
1 |
# -*- coding: utf-8 -*- |
|
2 |
# Generated by Django 1.11.15 on 2018-10-29 15:09 |
|
3 |
from __future__ import unicode_literals |
|
4 | ||
5 |
from django.db import migrations, models |
|
6 | ||
7 | ||
8 |
class Migration(migrations.Migration): |
|
9 | ||
10 |
initial = True |
|
11 | ||
12 |
dependencies = [ |
|
13 |
('base', '0006_resourcestatus'), |
|
14 |
] |
|
15 | ||
16 |
operations = [ |
|
17 |
migrations.CreateModel( |
|
18 |
name='PlanitechConnector', |
|
19 |
fields=[ |
|
20 |
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), |
|
21 |
('title', models.CharField(max_length=50, verbose_name='Title')), |
|
22 |
('description', models.TextField(verbose_name='Description')), |
|
23 |
('slug', models.SlugField(unique=True)), |
|
24 |
('log_level', models.CharField(choices=[(b'NOTSET', b'NOTSET'), (b'DEBUG', b'DEBUG'), (b'INFO', b'INFO'), (b'WARNING', b'WARNING'), (b'ERROR', b'ERROR'), (b'CRITICAL', b'CRITICAL')], default=b'INFO', max_length=10, verbose_name='Log Level')), |
|
25 |
('url', models.URLField(help_text='URL of the Planitech API endpoint', max_length=400, verbose_name='Planitech API endpoint')), |
|
26 |
('username', models.CharField(max_length=128, verbose_name='Service username')), |
|
27 |
('password', models.CharField(blank=True, max_length=128, null=True, verbose_name='Service password')), |
|
28 |
('users', models.ManyToManyField(blank=True, to='base.ApiUser')), |
|
29 |
('verify_cert', models.BooleanField(default=True, verbose_name='Check HTTPS Certificate validity')), |
|
30 |
], |
|
31 |
options={ |
|
32 |
'verbose_name': 'Planitech connector', |
|
33 |
}, |
|
34 |
), |
|
35 |
] |
passerelle/contrib/planitech/models.py | ||
---|---|---|
1 |
# passerelle - uniform access to multiple data sources and services |
|
2 |
# Copyright (C) 2018 Entr'ouvert |
|
3 |
# |
|
4 |
# This program is free software: you can redistribute it and/or modify it |
|
5 |
# under the terms of the GNU Affero General Public License as published |
|
6 |
# by the Free Software Foundation, either version 3 of the License, or |
|
7 |
# (at your option) any later version. |
|
8 |
# |
|
9 |
# This program is distributed in the hope that it will be useful, |
|
10 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 |
# GNU Affero General Public License for more details. |
|
13 |
# |
|
14 |
# You should have received a copy of the GNU Affero General Public License |
|
15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | ||
17 |
from datetime import datetime, time, timedelta |
|
18 |
import hashlib |
|
19 |
import json |
|
20 |
import re |
|
21 |
import urlparse |
|
22 | ||
23 |
from django.conf import settings |
|
24 |
from django.core.cache import cache |
|
25 |
from django.db import models |
|
26 |
from django.utils import dateformat |
|
27 |
from django.utils.dateparse import parse_date, parse_time |
|
28 |
from django.utils.translation import ugettext_lazy as _ |
|
29 |
from pytz import timezone, utc |
|
30 |
from requests.exceptions import RequestException |
|
31 | ||
32 |
from passerelle.base.models import BaseResource |
|
33 |
from passerelle.contrib.planitech import mste |
|
34 |
from passerelle.utils.api import endpoint |
|
35 |
from passerelle.utils.jsonresponse import APIError |
|
36 | ||
37 | ||
38 |
TZ = timezone(settings.TIME_ZONE) |
|
39 | ||
40 | ||
41 |
CREATE_RESERVATION_SCHEMA = { |
|
42 |
"$schema": "http://json-schema.org/draft-03/schema#", |
|
43 |
"title": "Planitech createreservation", |
|
44 |
"description": "", |
|
45 |
"type": "object", |
|
46 |
"properties": { |
|
47 |
"date": { |
|
48 |
"description": "Date", |
|
49 |
"type": "string", |
|
50 |
"required": True |
|
51 |
}, |
|
52 |
"start_time": { |
|
53 |
"description": "Start time", |
|
54 |
"type": "string", |
|
55 |
"required": True |
|
56 |
}, |
|
57 |
"end_time": { |
|
58 |
"description": "End time", |
|
59 |
"type": "string", |
|
60 |
"required": True |
|
61 |
}, |
|
62 |
"place_id": { |
|
63 |
"description": "Place idenfier", |
|
64 |
"type": "number", |
|
65 |
"required": True |
|
66 |
}, |
|
67 |
"price": { |
|
68 |
"description": "Price", |
|
69 |
"type": "number", |
|
70 |
"required": True |
|
71 |
} |
|
72 |
} |
|
73 |
} |
|
74 | ||
75 | ||
76 |
RESERVATION_STATUS = { |
|
77 |
"confirmed": 3, "invalid": 0, " pre-reservation": 1, "standard": 2 |
|
78 |
} |
|
79 | ||
80 |
UPDATE_RESERVATION_SCHEMA = { |
|
81 |
"$schema": "http://json-schema.org/draft-03/schema#", |
|
82 |
"title": "Planitech updatereservation", |
|
83 |
"description": "", |
|
84 |
"type": "object", |
|
85 |
"properties": { |
|
86 |
"reservation_id": { |
|
87 |
"description": "Reservation Identifier", |
|
88 |
"type": "number", |
|
89 |
"required": True |
|
90 |
}, |
|
91 |
"status": { |
|
92 |
"description": "Status of the reservation", |
|
93 |
"type": "string", |
|
94 |
"required": True, |
|
95 |
"enum": list(RESERVATION_STATUS.keys()) |
|
96 |
} |
|
97 |
} |
|
98 |
} |
|
99 | ||
100 | ||
101 |
def _parse_date(date_str): |
|
102 |
date_obj = parse_date(date_str) |
|
103 |
if date_obj is None: |
|
104 |
raise APIError("Invalid date format: %s" % date_str) |
|
105 |
return date_obj |
|
106 | ||
107 | ||
108 |
def _parse_time(time_str): |
|
109 |
timeobj = parse_time(time_str) |
|
110 |
if timeobj is None: |
|
111 |
raise APIError("Invalid time format : %s" % time_str) |
|
112 |
return timeobj |
|
113 | ||
114 | ||
115 |
def compute_hash(content, hardness, salt): |
|
116 |
sha = hashlib.new('sha512', salt + content) |
|
117 |
for idx in range(hardness): |
|
118 |
sha = hashlib.new('sha512', sha.digest()) |
|
119 |
return sha.hexdigest().upper() |
|
120 | ||
121 | ||
122 |
def date_to_datetime(date_str): |
|
123 |
date_obj = parse_date(date_str) |
|
124 |
if date_obj is None: |
|
125 |
raise APIError("Invalid date string: %s" % date_str) |
|
126 |
return datetime.combine(date_obj, time(hour=12)) |
|
127 | ||
128 | ||
129 |
def get_duration(start_time_str, end_time_str): |
|
130 |
start_time = parse_time(start_time_str) |
|
131 |
end_time = parse_time(end_time_str) |
|
132 |
return time_to_minutes(end_time) - time_to_minutes(start_time) |
|
133 | ||
134 | ||
135 |
def get_salt(salt): |
|
136 |
return re.match(r'<(.*?)>', salt).groups()[0] |
|
137 | ||
138 | ||
139 |
def get_utc_datetime(date_str, time_str): |
|
140 |
date_obj = parse_date(date_str) |
|
141 |
time_obj = parse_time(time_str) |
|
142 |
datetime_obj = datetime.combine(date_obj, time_obj) |
|
143 |
return TZ.localize(datetime_obj).astimezone(utc) |
|
144 | ||
145 | ||
146 |
def time_to_minutes(timeobj): |
|
147 |
return float(timeobj.hour * 60 + timeobj.minute) |
|
148 | ||
149 | ||
150 |
class PlanitechConnector(BaseResource): |
|
151 |
url = models.URLField( |
|
152 |
max_length=400, verbose_name=_('Planitech API endpoint'), |
|
153 |
help_text=_('URL of the Planitech API endpoint')) |
|
154 |
username = models.CharField(max_length=128, verbose_name=_('Service username')) |
|
155 |
password = models.CharField( |
|
156 |
max_length=128, verbose_name=_('Service password'), null=True, blank=True) |
|
157 |
verify_cert = models.BooleanField( |
|
158 |
default=True, verbose_name=_('Check HTTPS Certificate validity')) |
|
159 | ||
160 |
category = _('Business Process Connectors') |
|
161 | ||
162 |
class Meta: |
|
163 |
verbose_name = _('Planitech') |
|
164 | ||
165 |
def _call_planitech(self, session_meth, endpoint, params=None): |
|
166 |
if not getattr(self, '_planitech_session', False): |
|
167 |
self._login() |
|
168 |
self._planitech_session = True |
|
169 | ||
170 |
kwargs = {} |
|
171 |
if params is not None: |
|
172 |
kwargs['data'] = json.dumps(mste.encode(params)) |
|
173 |
response = session_meth(urlparse.urljoin(self.url, endpoint), **kwargs) |
|
174 |
if response.status_code != 200: |
|
175 |
error_msg = "Planitech error %s" % response.status_code |
|
176 |
try: |
|
177 |
data = mste.decode(response.json()) |
|
178 |
if hasattr(data, 'get'): |
|
179 |
error = data.get('errors') |
|
180 |
if error: |
|
181 |
error_msg += " - %s" % error |
|
182 |
except TypeError: |
|
183 |
pass |
|
184 |
raise APIError(error_msg) |
|
185 |
return mste.decode(response.json()) |
|
186 | ||
187 |
def _get_places_by_capacity(self, min_capacity, max_capacity): |
|
188 |
places = self._get_places_referential() |
|
189 |
min_capacity = int(min_capacity) |
|
190 |
max_capacity = int(max_capacity) |
|
191 |
return [place['identifier'] for place in places.values() |
|
192 |
if (min_capacity <= place['capacity'] <= max_capacity) |
|
193 |
and place['capacity']] |
|
194 | ||
195 |
def _get_places_referential(self, refresh_cache=False): |
|
196 |
cache_key = 'planitech-%s-places' % self.id |
|
197 |
ref = cache.get(cache_key) |
|
198 |
if ref is None or refresh_cache: |
|
199 |
data = self._call_planitech(self.requests.get, 'getPlacesList') |
|
200 |
ref = {} |
|
201 |
for place in data['placesList']: |
|
202 |
ref[place['identifier']] = { |
|
203 |
'identifier': place['identifier'], 'label': place['label'] |
|
204 |
} |
|
205 | ||
206 |
data = self._call_planitech( |
|
207 |
self.requests.post, 'getPlacesInfo', |
|
208 |
{ |
|
209 |
"placeIdentifiers": list(ref.keys()), |
|
210 |
"extensionAttributes": { |
|
211 |
"capacity": { |
|
212 |
"name": "TOTCAP", |
|
213 |
"type": "int" |
|
214 |
} |
|
215 |
} |
|
216 |
} |
|
217 |
) |
|
218 |
for place in data['requestedPlaces']: |
|
219 |
ref[place['identifier']]['capacity'] = place.get('capacity') |
|
220 |
cache.set(cache_key, ref) |
|
221 |
return ref |
|
222 | ||
223 |
def _login(self): |
|
224 |
try: |
|
225 |
auth_url = urlparse.urljoin(self.url, 'auth') |
|
226 |
response = self.requests.get(auth_url, headers={'MH-LOGIN': self.username}) |
|
227 |
response.raise_for_status() |
|
228 |
part1, salt1, part2, salt2, _ = re.split(r'(\<.*?\>)', response.content) |
|
229 |
hardness1 = int(part1.split(':')[1]) |
|
230 |
hardness2 = int(part2.split(':')[1]) |
|
231 |
salt1 = get_salt(salt1) |
|
232 |
salt2 = get_salt(salt2) |
|
233 |
tmp_hash = compute_hash(self.password, hardness1, salt1) |
|
234 |
hash_pass = compute_hash(tmp_hash, hardness2, salt2) |
|
235 |
response = self.requests.get(auth_url, headers={'MH-PASSWORD': hash_pass}) |
|
236 |
response.raise_for_status() |
|
237 |
# the last response should have set a cookie which will be used for authentication |
|
238 |
except RequestException as e: |
|
239 |
raise APIError("Authentication to Planitech failed: %s" % str(e)) |
|
240 | ||
241 |
def check_status(self): |
|
242 |
self._call_planitech(self.requests.get, 'getCapabilities') |
|
243 | ||
244 |
@endpoint( |
|
245 |
perm='can_access', |
|
246 |
post={ |
|
247 |
'description': _('Create reservation'), |
|
248 |
'request_body': { |
|
249 |
'schema': { |
|
250 |
'application/json': CREATE_RESERVATION_SCHEMA |
|
251 |
} |
|
252 |
} |
|
253 |
} |
|
254 |
) |
|
255 |
def createreservation(self, request, post_data): |
|
256 |
start_datetime = get_utc_datetime(post_data['date'], post_data['start_time']) |
|
257 |
end_datetime = get_utc_datetime(post_data['date'], post_data['end_time']) |
|
258 |
request_date = datetime.now(tz=utc) |
|
259 | ||
260 |
params = { |
|
261 |
"activityID": mste.Uint32(6), # relaxation FIXME |
|
262 |
"contractorExternalIdentifier": "test-entrouvert", # FIXME |
|
263 |
"end": end_datetime, |
|
264 |
"isWeekly": False, |
|
265 |
"object": "reservation test entouvert", # FIXME |
|
266 |
"places": [float(post_data['place_id'])], |
|
267 |
"price": mste.Uint32(post_data['price']), |
|
268 |
"requestDate": request_date, |
|
269 |
"start": start_datetime, |
|
270 |
"typeID": mste.Uint32(2), # payant FIXME |
|
271 |
"vatRate": mste.Uint32(4000) # FIXME |
|
272 |
} |
|
273 |
data = self._call_planitech(self.requests.post, 'createReservation', params) |
|
274 |
if data.get('creationStatus') != 'OK': |
|
275 |
raise APIError("Reservation creation failed: %s" % data.get('creationStatus')) |
|
276 |
reservation_id = data.get('reservationIdentifier') |
|
277 |
if not reservation_id: |
|
278 |
raise APIError("Reservation creation failed: no reservation ID") |
|
279 |
return { |
|
280 |
'data': { |
|
281 |
'reservation_id': reservation_id, |
|
282 |
'raw_data': data |
|
283 |
} |
|
284 |
} |
|
285 | ||
286 |
def hourly(self): |
|
287 |
self._get_places_referential(refresh_cache=True) |
|
288 | ||
289 |
@endpoint( |
|
290 |
description_get='Get days available for reservation', |
|
291 |
methods=['get'], perm='can_access', |
|
292 |
parameters={ |
|
293 |
'min_capacity': { |
|
294 |
'description': _('Minimum capacity'), |
|
295 |
'example_value': '1', |
|
296 |
}, |
|
297 |
'max_capacity': { |
|
298 |
'description': _('Maximum capacity'), |
|
299 |
'example_value': '10', |
|
300 |
}, |
|
301 |
'start_date': { |
|
302 |
'description': _('Start date'), |
|
303 |
'example_value': '2018-10-10', |
|
304 |
}, |
|
305 |
'end_date': { |
|
306 |
'description': _('End date'), |
|
307 |
'example_value': '2018-12-10', |
|
308 |
}, |
|
309 |
'start_time': { |
|
310 |
'description': _('Start time'), |
|
311 |
'example_value': '10:00', |
|
312 |
}, |
|
313 |
'end_time': { |
|
314 |
'description': _('End time'), |
|
315 |
'example_value': '18:00', |
|
316 |
}, |
|
317 |
'weekdays': { |
|
318 |
'description': _('Week days'), |
|
319 |
'example_value': 'true', |
|
320 |
'type': 'bool', |
|
321 |
}, |
|
322 |
}) |
|
323 |
def getdays( |
|
324 |
self, request, start_date, start_time, end_time, min_capacity=0, |
|
325 |
end_date=None, max_capacity=100000, weekdays=False): |
|
326 |
places_id = self._get_places_by_capacity(int(min_capacity), int(max_capacity)) |
|
327 | ||
328 |
utc_start_datetime = get_utc_datetime(start_date, start_time) |
|
329 |
if end_date is None: |
|
330 |
utc_end_datetime = utc_start_datetime + timedelta(days=365) |
|
331 |
else: |
|
332 |
utc_end_datetime = get_utc_datetime(end_date, '00:00') |
|
333 | ||
334 |
duration = get_duration(start_time, end_time) |
|
335 | ||
336 |
params = { |
|
337 |
"placeIdentifiers": places_id, |
|
338 |
"startingDate": utc_start_datetime, |
|
339 |
"endingDate": utc_end_datetime, |
|
340 |
"requestedStartingTime": float(0), |
|
341 |
"requestedEndingTime": duration |
|
342 |
} |
|
343 |
if not weekdays: |
|
344 |
params['reservationDays'] = [mste.Uint32(0), mste.Uint32(6)] |
|
345 | ||
346 |
raw_data = self._call_planitech(self.requests.post, 'getFreeGaps', params) |
|
347 |
available_dates = set() |
|
348 |
for place in raw_data.get('availablePlaces', []): |
|
349 |
for freegap in place.get('freeGaps', []): |
|
350 |
available_dates.add(freegap[0].date()) |
|
351 | ||
352 |
res = [] |
|
353 |
available_dates = list(available_dates) |
|
354 |
available_dates.sort() |
|
355 |
for date_obj in available_dates: |
|
356 |
date_text = dateformat.format(date_obj, 'l d F Y') |
|
357 |
res.append({"id": date_obj.isoformat(), "text": date_text}) |
|
358 | ||
359 |
return {'data': res} |
|
360 | ||
361 |
@endpoint( |
|
362 |
description_get='Get places available for reservation', |
|
363 |
methods=['get'], perm='can_access', |
|
364 |
parameters={ |
|
365 |
'min_capacity': { |
|
366 |
'description': _('Minimum capacity'), |
|
367 |
'example_value': '1', |
|
368 |
}, |
|
369 |
'max_capacity': { |
|
370 |
'description': _('Maximum capacity'), |
|
371 |
'example_value': '10', |
|
372 |
}, |
|
373 |
'start_date': { |
|
374 |
'description': _('Start date'), |
|
375 |
'example_value': '2018-10-10', |
|
376 |
}, |
|
377 |
'start_time': { |
|
378 |
'description': _('Start time'), |
|
379 |
'example_value': '10:00', |
|
380 |
}, |
|
381 |
'end_time': { |
|
382 |
'description': _('End time'), |
|
383 |
'example_value': '18:00', |
|
384 |
}, |
|
385 |
}) |
|
386 |
def getplaces( |
|
387 |
self, request, start_date, start_time, end_time, min_capacity=0, max_capacity=100000): |
|
388 | ||
389 |
places_id = self._get_places_by_capacity(min_capacity, max_capacity) |
|
390 | ||
391 |
utc_start_datetime = get_utc_datetime(start_date, start_time) |
|
392 |
utc_end_datetime = utc_start_datetime + timedelta(days=1) |
|
393 |
duration = get_duration(start_time, end_time) |
|
394 | ||
395 |
params = { |
|
396 |
"placeIdentifiers": places_id, |
|
397 |
"startingDate": utc_start_datetime, |
|
398 |
"endingDate": utc_end_datetime, |
|
399 |
"requestedStartingTime": float(0), |
|
400 |
"requestedEndingTime": duration |
|
401 |
} |
|
402 | ||
403 |
raw_data = self._call_planitech(self.requests.post, 'getFreeGaps', params) |
|
404 |
available_places = [] |
|
405 |
for place in raw_data.get('availablePlaces', []): |
|
406 |
available_places.append(place['placeIdentifier']) |
|
407 | ||
408 |
places_ref = self._get_places_referential() |
|
409 | ||
410 |
res = [] |
|
411 |
for place in available_places: |
|
412 |
res.append({"id": place, "text": places_ref[place]['label']}) |
|
413 | ||
414 |
return {'data': res} |
|
415 | ||
416 |
@endpoint(description_get='Get every places', methods=['get'], perm='can_access') |
|
417 |
def getplacesreferential(self, request): |
|
418 |
return {'data': self._get_places_referential()} |
|
419 | ||
420 |
@endpoint( |
|
421 |
methods=['post'], perm='can_access', |
|
422 |
post={ |
|
423 |
'description': _('Update reservation'), |
|
424 |
'request_body': { |
|
425 |
'schema': { |
|
426 |
'application/json': UPDATE_RESERVATION_SCHEMA |
|
427 |
} |
|
428 |
} |
|
429 |
} |
|
430 |
) |
|
431 |
def updatereservation(self, request, post_data): |
|
432 |
params = { |
|
433 |
"reservationIdentifier": mste.Uint32(post_data['reservation_id']), |
|
434 |
"situation": mste.Uint32(RESERVATION_STATUS[post_data['status']]) |
|
435 |
} |
|
436 |
data = self._call_planitech(self.requests.post, 'updateReservation', params) |
|
437 |
if data.get('modificationStatus') != 'OK': |
|
438 |
raise APIError("Update reservation failed: %s" % data.get('modificationStatus')) |
|
439 |
return { |
|
440 |
'data': { |
|
441 |
'raw_data': data |
|
442 |
} |
|
443 |
} |
passerelle/contrib/planitech/mste.py | ||
---|---|---|
1 |
# passerelle - uniform access to multiple data sources and services |
|
2 |
# Copyright (C) 2018 Entr'ouvert |
|
3 |
# |
|
4 |
# This program is free software: you can redistribute it and/or modify it |
|
5 |
# under the terms of the GNU Affero General Public License as published |
|
6 |
# by the Free Software Foundation, either version 3 of the License, or |
|
7 |
# (at your option) any later version. |
|
8 |
# |
|
9 |
# This program is distributed in the hope that it will be useful, |
|
10 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 |
# GNU Affero General Public License for more details. |
|
13 |
# |
|
14 |
# You should have received a copy of the GNU Affero General Public License |
|
15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | ||
17 |
from datetime import datetime |
|
18 |
import calendar |
|
19 | ||
20 | ||
21 |
ENCODE_TOKENS = { |
|
22 |
'integer': 16, |
|
23 |
'real': 19, |
|
24 |
'nil': 0, |
|
25 |
'true': 1, |
|
26 |
'false': 2, |
|
27 |
'emptyString': 3, |
|
28 |
'emptyData': 4, |
|
29 |
'ref': 9, |
|
30 |
'i1': 10, |
|
31 |
'u1': 11, |
|
32 |
'i2': 12, |
|
33 |
'u2': 13, |
|
34 |
'i4': 14, |
|
35 |
'uint32': 15, |
|
36 |
'i8': 16, |
|
37 |
'u8': 17, |
|
38 |
'float': 18, |
|
39 |
'double': 19, |
|
40 |
'decimal': 20, |
|
41 |
'string': 21, |
|
42 |
'localdate': 22, |
|
43 |
'gmtDate': 23, |
|
44 |
'color': 24, |
|
45 |
'data': 25, |
|
46 |
'naturals': 26, |
|
47 |
'dictionary': 30, |
|
48 |
'array': 31, |
|
49 |
'couple': 32, |
|
50 |
'object': 50, |
|
51 |
} |
|
52 | ||
53 |
DECODE_TOKENS = {v: k for k, v in ENCODE_TOKENS.items()} |
|
54 | ||
55 | ||
56 |
class Couple(list): |
|
57 |
pass |
|
58 | ||
59 | ||
60 |
class Uint32(int): |
|
61 |
pass |
|
62 | ||
63 | ||
64 |
class MSTEDecoder(object): |
|
65 | ||
66 |
def __init__(self, data): |
|
67 |
self._idx = 4 |
|
68 |
self._keys = [] |
|
69 |
self._refs = [] |
|
70 |
self._data = data |
|
71 | ||
72 |
def _next_token(self): |
|
73 |
self._idx += 1 |
|
74 |
return self._data[self._idx] |
|
75 | ||
76 |
def _parse_array(self): |
|
77 |
count = self._next_token() |
|
78 |
res = [] |
|
79 |
self._refs.append(res) |
|
80 |
while count > 0: |
|
81 |
res.append(self._parse_item()) |
|
82 |
count -= 1 |
|
83 |
return res |
|
84 | ||
85 |
def _parse_couple(self): |
|
86 |
res = Couple() |
|
87 |
self._refs.append(res) |
|
88 |
for _ in range(2): |
|
89 |
res.append(self._parse_item()) |
|
90 |
return res |
|
91 | ||
92 |
def _parse_decimal(self): |
|
93 |
res = float(self._next_token()) |
|
94 |
self._refs.append(res) |
|
95 |
return res |
|
96 | ||
97 |
def _parse_dictionary(self): |
|
98 |
count = self._next_token() |
|
99 |
res = {} |
|
100 |
self._refs.append(res) |
|
101 |
while count > 0: |
|
102 |
key = self._keys[self._next_token()] |
|
103 |
res[key] = self._parse_item() |
|
104 |
count -= 1 |
|
105 |
return res |
|
106 | ||
107 |
def _parse_emptyString(self): |
|
108 |
return '' |
|
109 | ||
110 |
def _parse_gmtDate(self): |
|
111 |
return self._parse_localdate() |
|
112 | ||
113 |
def _parse_item(self): |
|
114 |
token = self._next_token() |
|
115 |
_type = DECODE_TOKENS[token] |
|
116 |
return getattr(self, "_parse_%s" % _type)() |
|
117 | ||
118 |
def _parse_localdate(self): |
|
119 |
timestamp = self._next_token() |
|
120 |
res = datetime.fromtimestamp(timestamp) |
|
121 |
self._refs.append(res) |
|
122 |
return res |
|
123 | ||
124 |
def _parse_nil(self): |
|
125 |
return None |
|
126 | ||
127 |
def _parse_ref(self): |
|
128 |
pos = self._next_token() |
|
129 |
return self._refs[pos] |
|
130 | ||
131 |
def _parse_string(self): |
|
132 |
res = self._next_token() |
|
133 |
self._refs.append(res) |
|
134 |
return res |
|
135 | ||
136 |
def _parse_true(self): |
|
137 |
return True |
|
138 | ||
139 |
def _parse_false(self): |
|
140 |
return False |
|
141 | ||
142 |
def _parse_uint32(self): |
|
143 |
return int(self._next_token()) |
|
144 | ||
145 |
def decode(self): |
|
146 |
num_keys = self._data[self._idx] |
|
147 |
while num_keys > 0: |
|
148 |
self._keys.append(self._next_token()) |
|
149 |
num_keys -= 1 |
|
150 |
return self._parse_item() |
|
151 | ||
152 | ||
153 |
class ObjectStore(list): |
|
154 | ||
155 |
def add(self, obj): |
|
156 |
""" Add object in the store |
|
157 |
and return its reference |
|
158 |
""" |
|
159 |
ref = self.getref(obj) |
|
160 |
if ref is None: |
|
161 |
self.append(obj) |
|
162 |
ref = len(self) - 1 |
|
163 |
return ref |
|
164 | ||
165 |
def getref(self, obj): |
|
166 |
""" Return the reference of obj, |
|
167 |
None if the object is not in the store |
|
168 |
""" |
|
169 |
try: |
|
170 |
return self.index(obj) |
|
171 |
except ValueError: |
|
172 |
return None |
|
173 | ||
174 | ||
175 |
class MSTEEncoder(object): |
|
176 | ||
177 |
def __init__(self, data): |
|
178 |
self._data = data |
|
179 |
self._stream = [] |
|
180 |
self._refs_store = ObjectStore() |
|
181 |
self._keys_store = ObjectStore() |
|
182 | ||
183 |
def _push_token_type(self, token_type): |
|
184 |
self._stream.append(ENCODE_TOKENS[token_type]) |
|
185 | ||
186 |
def _push(self, item): |
|
187 |
self._stream.append(item) |
|
188 | ||
189 |
def _encode_array(self, obj): |
|
190 |
self._refs_store.add(obj) |
|
191 |
self._push_token_type('array') |
|
192 |
self._push(len(obj)) |
|
193 |
for elem in obj: |
|
194 |
self._encode_obj(elem) |
|
195 | ||
196 |
def _encode_boolean(self, obj): |
|
197 |
if obj: |
|
198 |
self._push_token_type('true') |
|
199 |
else: |
|
200 |
self._push_token_type('false') |
|
201 | ||
202 |
def _encode_couple(self, obj): |
|
203 |
self._refs_store.add(obj) |
|
204 |
self._push_token_type('couple') |
|
205 |
for elem in obj: |
|
206 |
self._encode_obj(elem) |
|
207 | ||
208 |
def _encode_decimal(self, obj): |
|
209 |
self._refs_store.add(obj) |
|
210 |
self._push_token_type('decimal') |
|
211 |
self._push(int(obj)) |
|
212 | ||
213 |
def _encode_dictionary(self, obj): |
|
214 |
self._refs_store.add(obj) |
|
215 |
self._push_token_type('dictionary') |
|
216 |
self._push(len(obj)) |
|
217 |
for key, value in obj.items(): |
|
218 |
key_ref = self._keys_store.add(key) |
|
219 |
self._push(key_ref) |
|
220 |
self._encode_obj(value) |
|
221 | ||
222 |
def _encode_localdate(self, obj): |
|
223 |
# obj must be utc |
|
224 |
self._refs_store.add(obj) |
|
225 |
self._push_token_type('localdate') |
|
226 |
self._push(int(calendar.timegm(obj.timetuple()))) |
|
227 | ||
228 |
def _encode_obj(self, obj): |
|
229 |
ref = self._refs_store.getref(obj) |
|
230 |
if ref is not None: |
|
231 |
self._push_token_type('ref') |
|
232 |
self._push(ref) |
|
233 |
elif isinstance(obj, unicode) or isinstance(obj, str): |
|
234 |
self._encode_string(obj) |
|
235 |
elif obj is None: |
|
236 |
self._encode_nil() |
|
237 |
elif isinstance(obj, Couple): |
|
238 |
self._encode_couple(obj) |
|
239 |
elif isinstance(obj, bool): |
|
240 |
self._encode_boolean(obj) |
|
241 |
elif isinstance(obj, list): |
|
242 |
self._encode_array(obj) |
|
243 |
elif isinstance(obj, dict): |
|
244 |
self._encode_dictionary(obj) |
|
245 |
elif isinstance(obj, float): |
|
246 |
self._encode_decimal(obj) |
|
247 |
elif isinstance(obj, datetime): |
|
248 |
self._encode_localdate(obj) |
|
249 |
elif isinstance(obj, Uint32): |
|
250 |
self._encode_uint32(obj) |
|
251 |
else: |
|
252 |
raise TypeError("%s encoding not supported" % type(obj)) |
|
253 | ||
254 |
def _encode_nil(self): |
|
255 |
self._push_token_type('nil') |
|
256 | ||
257 |
def _encode_string(self, _str): |
|
258 |
if _str: |
|
259 |
self._push_token_type('string') |
|
260 |
self._push(_str) |
|
261 |
self._refs_store.add(_str) |
|
262 |
else: |
|
263 |
self._push_token_type('emptyString') |
|
264 | ||
265 |
def _encode_uint32(self, obj): |
|
266 |
self._push_token_type('uint32') |
|
267 |
self._push(obj) |
|
268 | ||
269 |
def encode(self): |
|
270 |
res = ["MSTE0102"] |
|
271 |
self._encode_obj(self._data) |
|
272 |
nb_token = 5 + len(self._keys_store) + len(self._stream) |
|
273 |
res = ["MSTE0102", nb_token, 'CRC00000000', 0, len(self._keys_store)] + self._keys_store |
|
274 |
res.extend(self._stream) |
|
275 |
return res |
|
276 | ||
277 | ||
278 |
def decode(data): |
|
279 |
return MSTEDecoder(data).decode() |
|
280 | ||
281 | ||
282 |
def encode(data): |
|
283 |
return MSTEEncoder(data).encode() |
tests/settings.py | ||
---|---|---|
27 | 27 |
'passerelle.contrib.mdel', |
28 | 28 |
'passerelle.contrib.meyzieu_newsletters', |
29 | 29 |
'passerelle.contrib.nancypoll', |
30 |
'passerelle.contrib.planitech', |
|
30 | 31 |
'passerelle.contrib.seisin_by_email', |
31 | 32 |
'passerelle.contrib.solis_apa', |
32 | 33 |
'passerelle.contrib.strasbourg_eu', |
tests/test_planitech.py | ||
---|---|---|
1 |
from datetime import datetime |
|
2 | ||
3 |
from django.contrib.contenttypes.models import ContentType |
|
4 |
from django.core.cache import cache |
|
5 |
from httmock import urlmatch, HTTMock |
|
6 |
import mock |
|
7 |
import pytest |
|
8 |
from pytz import utc |
|
9 |
import requests |
|
10 | ||
11 |
from passerelle.base.models import ApiUser, AccessRight |
|
12 |
from passerelle.contrib.planitech import mste |
|
13 |
from passerelle.contrib.planitech.models import PlanitechConnector |
|
14 |
from passerelle.utils.jsonresponse import APIError |
|
15 | ||
16 | ||
17 |
def assert_mste(data, ref_data): |
|
18 |
""" skip CRC verification |
|
19 |
""" |
|
20 |
assert len(data) == len(ref_data) |
|
21 |
for i in range(len(data)): |
|
22 |
if i != 2: |
|
23 |
assert data[i] == ref_data[i] |
|
24 | ||
25 | ||
26 |
@pytest.mark.parametrize("data,mste_data", [ |
|
27 |
(None, ["MSTE0102", 6, "CRC82413E70", 0, 0, 0]), |
|
28 |
("toto", ["MSTE0102", 7, "CRCD45ACB10", 0, 0, 21, "toto"]), # string |
|
29 |
(mste.Couple(("toto", "tata")), ["MSTE0102", 10, "CRCD45ACB10", 0, 0, 32, 21, "toto", 21, |
|
30 |
"tata"]), |
|
31 |
# couple |
|
32 |
([mste.Couple(("toto", "tata")), mste.Couple(("toto", "tata"))], |
|
33 |
["MSTE0102", 14, "CRCD45ACB10", 0, 0, 31, 2, 32, 21, "toto", 21, "tata", 9, 1]), |
|
34 |
# couple are stored in refs |
|
35 |
(["toto"], ["MSTE0102", 9, "CRCD4E14B75", 0, 0, 31, 1, 21, "toto"]), # array |
|
36 |
(["toto", "tata", "toto"], ["MSTE0102", 13, "CRC7311752F", 0, 0, 31, 3, 21, "toto", 21, |
|
37 |
"tata", 9, 1]), # array with reference |
|
38 |
({"mykey": "toto"}, ["MSTE0102", 11, "CRC1C9E9FE1", 0, 1, "mykey", 30, 1, 0, 21, "toto"]), |
|
39 |
# dictionnary |
|
40 |
([{"mykey": "toto"}, {"mykey": "toto"}], |
|
41 |
["MSTE0102", 15, "CRC1C9E9FE1", 0, 1, "mykey", 31, 2, 30, 1, 0, 21, |
|
42 |
"toto", 9, 1]), |
|
43 |
# dictionnary are stored in refs |
|
44 |
(float(2), ["MSTE0102", 7, "CRC1C9E9FE1", 0, 0, 20, 2]), # decimal |
|
45 |
([float(2), float(2)], ["MSTE0102", 11, "CRC1C9E9FE1", 0, 0, 31, 2, 20, 2, 9, 1]), |
|
46 |
# decimal are stored in refs |
|
47 |
(mste.Uint32(1), ["MSTE0102", 7, "CRC1C9E9FE1", 0, 0, 15, 1]), # uint32 |
|
48 |
(True, ["MSTE0102", 6, "CRC1C9E9FE1", 0, 0, 1]), # True |
|
49 |
(False, ["MSTE0102", 6, "CRC1C9E9FE1", 0, 0, 2]), # False |
|
50 |
('', ["MSTE0102", 6, "CRC1C9E9FE1", 0, 0, 3]), # empty string |
|
51 |
(datetime.fromtimestamp(1537364340), ["MSTE0102", 7, "CRC1C9E9FE1", 0, 0, 22, 1537364340]), |
|
52 |
# local date |
|
53 |
([datetime.fromtimestamp(1537364340), datetime.fromtimestamp(1537364340)], |
|
54 |
["MSTE0102", 11, "CRC1C9E9FE1", 0, 0, 31, 2, 22, 1537364340, 9, 1]), |
|
55 |
# local date in refs |
|
56 |
]) |
|
57 |
def test_mste(mste_data, data): |
|
58 |
assert data == mste.decode(mste_data) |
|
59 |
assert_mste(mste.encode(data), mste_data) |
|
60 | ||
61 | ||
62 |
def test_encode_unsupported_type(): |
|
63 |
with pytest.raises(TypeError): |
|
64 |
mste.encode(set()) |
|
65 | ||
66 | ||
67 |
def test_real(): |
|
68 |
mste_data = ["MSTE0102", 128, "CRC99D9BCEB", 0, 11, "requestDate", "responseDate", "requestName", "requestedEndingTime", "availablePlaces", "label", "freeGaps", "placeIdentifier", "resourceIdentifier", "daysMask", "requestedStartingTime", 30, 7, 0, 22, 1538404500, 1, 22, 1538404500, 2, 21, "getFreeGaps", 3, 20, 600, 4, 31, 1, 30, 4, 5, 21, "M.F.F. 2", 6, 31, 15, 32, 22, 1538384400, 22, 1538388000, 32, 22, 1538470800, 22, 1538474400, 32, 22, 1538557200, 22, 1538560800, 32, 22, 1538643600, 22, 1538647200, 32, 22, 1538989200, 22, 1538992800, 32, 22, 1539075600, 22, 1539079200, 32, 22, 1539162000, 22, 1539165600, 32, 22, 1539248400, 22, 1539252000, 32, 22, 1539334800, 22, 1539338400, 32, 22, 1539507600, 22, 1539511200, 32, 22, 1539594000, 22, 1539597600, 32, 22, 1539680400, 22, 1539684000, 32, 22, 1539766800, 22, 1539770400, 32, 22, 1539853200, 22, 1539856800, 32, 22, 1539939600, 22, 1539943200, 7, 20, 2, 8, 9, 54, 9, 20, 127, 10, 20, 540] |
|
69 |
mste.decode(mste_data) |
|
70 | ||
71 | ||
72 |
@pytest.fixture() |
|
73 |
def connector(db): |
|
74 |
api = ApiUser.objects.create(username='all', keytype='', key='') |
|
75 |
connector = PlanitechConnector.objects.create( |
|
76 |
url='http://example.planitech.com/', username='admin', password='admin', |
|
77 |
verify_cert=False, slug='slug-planitech') |
|
78 |
obj_type = ContentType.objects.get_for_model(connector) |
|
79 |
AccessRight.objects.create( |
|
80 |
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=connector.pk) |
|
81 |
return connector |
|
82 | ||
83 | ||
84 |
def mock_planitech(monkeypatch, return_value=None, side_effect=None, referential=None): |
|
85 |
from passerelle.contrib.planitech import models |
|
86 |
monkeypatch.setattr(models.PlanitechConnector, '_login', mock.Mock()) |
|
87 |
kwargs = {} |
|
88 |
if return_value is not None: |
|
89 |
kwargs['return_value'] = return_value |
|
90 |
if side_effect is not None: |
|
91 |
kwargs['side_effect'] = side_effect |
|
92 |
mock_call_planitech = mock.Mock(**kwargs) |
|
93 |
monkeypatch.setattr(models.PlanitechConnector, '_call_planitech', mock_call_planitech) |
|
94 | ||
95 |
if referential is not None: |
|
96 |
mock_get_referential = mock.Mock(return_value=referential) |
|
97 |
monkeypatch.setattr( |
|
98 |
models.PlanitechConnector, '_get_places_referential', mock_get_referential) |
|
99 | ||
100 |
return mock_call_planitech |
|
101 | ||
102 | ||
103 |
def test_call_planitech(connector, monkeypatch): |
|
104 | ||
105 |
class MockResponse(object): |
|
106 | ||
107 |
status_code = 200 |
|
108 |
content = None |
|
109 | ||
110 |
def __init__(self, content=None, status_code=None): |
|
111 |
if content is not None: |
|
112 |
self.content = content |
|
113 |
if status_code is not None: |
|
114 |
self.status_code = status_code |
|
115 | ||
116 |
def session_meth(self, *args, **kwargs): |
|
117 |
return self |
|
118 | ||
119 |
def json(self): |
|
120 |
return mste.encode(self.content) |
|
121 | ||
122 |
connector._planitech_session = True |
|
123 | ||
124 |
response = MockResponse(content='somestring') |
|
125 |
assert connector._call_planitech(response.session_meth, 'endpoint') == "somestring" |
|
126 | ||
127 |
response = MockResponse(content=set(), status_code=400) |
|
128 |
with pytest.raises(APIError) as excinfo: |
|
129 |
connector._call_planitech(response.session_meth, 'endpoint') |
|
130 |
assert str(excinfo.value) == 'Planitech error 400' |
|
131 | ||
132 |
response = MockResponse(content='unexpected error format', status_code=400) |
|
133 |
with pytest.raises(APIError) as excinfo: |
|
134 |
connector._call_planitech(response.session_meth, 'endpoint') |
|
135 |
assert str(excinfo.value) == 'Planitech error 400' |
|
136 | ||
137 |
response = MockResponse(content={'errors': 'planitech error message'}, status_code=400) |
|
138 |
with pytest.raises(APIError) as excinfo: |
|
139 |
connector._call_planitech(response.session_meth, 'endpoint') |
|
140 |
assert str(excinfo.value) == 'Planitech error 400 - planitech error message' |
|
141 | ||
142 | ||
143 |
def test_create_reservation(app, connector, monkeypatch): |
|
144 |
mock_call_planitech = mock_planitech( |
|
145 |
monkeypatch, return_value={ |
|
146 |
'creationStatus': 'OK', |
|
147 |
'reservationIdentifier': 1 |
|
148 |
}) |
|
149 |
response = app.post_json( |
|
150 |
'/planitech/slug-planitech/createreservation', |
|
151 |
params={ |
|
152 |
'date': '2018-11-11', |
|
153 |
'start_time': '10:00', |
|
154 |
'end_time': '11:00', |
|
155 |
'place_id': 1, |
|
156 |
'price': 10 |
|
157 |
} |
|
158 |
) |
|
159 |
json_resp = response.json |
|
160 |
assert json_resp['err'] == 0 |
|
161 |
assert json_resp['data']['reservation_id'] == 1 |
|
162 |
call_params = mock_call_planitech.call_args[0][2] |
|
163 |
assert call_params['start'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc) |
|
164 |
assert call_params['end'] == datetime(2018, 11, 11, 11, 0, tzinfo=utc) |
|
165 |
assert call_params['places'] == [1] |
|
166 |
assert call_params['price'] == 10 |
|
167 | ||
168 |
# Create reservation failed |
|
169 |
mock_call_planitech = mock_planitech( |
|
170 |
monkeypatch, return_value={ |
|
171 |
'creationStatus': 'NOTOK', |
|
172 |
'reservationIdentifier': 1 |
|
173 |
}) |
|
174 |
response = app.post_json( |
|
175 |
'/planitech/slug-planitech/createreservation', |
|
176 |
params={ |
|
177 |
'date': '2018-11-11', |
|
178 |
'start_time': '10:00', |
|
179 |
'end_time': '11:00', |
|
180 |
'place_id': 1, |
|
181 |
'price': 10 |
|
182 |
} |
|
183 |
) |
|
184 |
json_resp = response.json |
|
185 |
assert json_resp['err'] == 1 |
|
186 |
assert json_resp['err_desc'] == 'Reservation creation failed: NOTOK' |
|
187 | ||
188 |
# Create reservation failed - nor reservation ID |
|
189 |
mock_call_planitech = mock_planitech( |
|
190 |
monkeypatch, return_value={ |
|
191 |
'creationStatus': 'OK' |
|
192 |
}) |
|
193 |
response = app.post_json( |
|
194 |
'/planitech/slug-planitech/createreservation', |
|
195 |
params={ |
|
196 |
'date': '2018-11-11', |
|
197 |
'start_time': '10:00', |
|
198 |
'end_time': '11:00', |
|
199 |
'place_id': 1, |
|
200 |
'price': 10 |
|
201 |
} |
|
202 |
) |
|
203 |
json_resp = response.json |
|
204 |
assert json_resp['err'] == 1 |
|
205 |
assert json_resp['err_desc'] == 'Reservation creation failed: no reservation ID' |
|
206 | ||
207 | ||
208 |
def test_getplaces_referential(app, connector, monkeypatch): |
|
209 |
side_effect = [ |
|
210 |
{ |
|
211 |
'placesList': [ |
|
212 |
{'identifier': 1.0, 'label': 'salle 1'}, |
|
213 |
{'identifier': 2.0, 'label': 'salle 2'} |
|
214 |
] |
|
215 |
}, |
|
216 |
{ |
|
217 |
'requestedPlaces': [ |
|
218 |
{'identifier': 1.0, 'capacity': 10.0}, |
|
219 |
{'identifier': 2.0, 'capacity': 20.0} |
|
220 |
] |
|
221 |
} |
|
222 |
] |
|
223 |
mock_planitech(monkeypatch, side_effect=side_effect) |
|
224 |
response = app.get('/planitech/slug-planitech/getplacesreferential') |
|
225 |
expected_res = { |
|
226 |
u'2.0': { |
|
227 |
u'capacity': 20.0, u'label': u'salle 2', u'identifier': 2.0 |
|
228 |
}, |
|
229 |
u'1.0': { |
|
230 |
u'capacity': 10.0, u'label': u'salle 1', u'identifier': 1.0 |
|
231 |
} |
|
232 |
} |
|
233 |
assert response.json['data'] == expected_res |
|
234 | ||
235 | ||
236 |
def test_getplaces_referential_use_cache(app, connector): |
|
237 |
cache_key = 'planitech-%s-places' % connector.id |
|
238 |
cache.set(cache_key, {'some': 'data'}) |
|
239 |
response = app.get('/planitech/slug-planitech/getplacesreferential') |
|
240 |
assert response.json_body['data'] == {'some': 'data'} |
|
241 |
cache.delete(cache_key) |
|
242 | ||
243 | ||
244 |
def test_get_days(app, connector, monkeypatch, settings): |
|
245 |
settings.LANGUAGE_CODE = 'fr-fr' |
|
246 |
referential = { |
|
247 |
2.0: { |
|
248 |
u'capacity': 20.0, u'label': u'salle 2', u'identifier': 2.0 |
|
249 |
}, |
|
250 |
1.0: { |
|
251 |
u'capacity': 10.0, u'label': u'salle 1', u'identifier': 1.0 |
|
252 |
} |
|
253 |
} |
|
254 | ||
255 |
def get_free_gaps(): |
|
256 |
return [ |
|
257 |
[datetime(year=2018, month=11, day=11, hour=10, minute=0), |
|
258 |
datetime(year=2018, month=11, day=11, hour=11, minute=0)], |
|
259 |
[datetime(year=2018, month=11, day=12, hour=10, minute=0), |
|
260 |
datetime(year=2018, month=11, day=12, hour=11, minute=0)] |
|
261 |
] |
|
262 | ||
263 |
getfreegaps = { |
|
264 |
'availablePlaces': [ |
|
265 |
{ |
|
266 |
'freeGaps': get_free_gaps() |
|
267 |
}, |
|
268 |
{ |
|
269 |
'freeGaps': get_free_gaps() |
|
270 |
} |
|
271 |
] |
|
272 |
} |
|
273 |
mock_call_planitech = mock_planitech( |
|
274 |
monkeypatch, return_value=getfreegaps, referential=referential) |
|
275 |
response = app.get( |
|
276 |
'/planitech/slug-planitech/getdays?start_time=10:00&&end_time=11:00' |
|
277 |
'&start_date=2018-11-11' |
|
278 |
) |
|
279 |
assert response.json['data'] == [ |
|
280 |
{u'id': u'2018-11-11', u'text': u'dimanche 11 novembre 2018'}, |
|
281 |
{u'id': u'2018-11-12', u'text': u'lundi 12 novembre 2018'} |
|
282 |
] |
|
283 |
call_params = mock_call_planitech.call_args[0][2] |
|
284 |
assert call_params['startingDate'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc) |
|
285 |
assert call_params['endingDate'] == datetime(2019, 11, 11, 10, 0, tzinfo=utc) |
|
286 |
assert call_params['placeIdentifiers'] == [1.0, 2.0] |
|
287 |
assert call_params['requestedStartingTime'] == 0.0 |
|
288 |
assert call_params['requestedEndingTime'] == 60.0 |
|
289 |
assert call_params['reservationDays'] == [0, 6] |
|
290 | ||
291 |
# capcacity |
|
292 |
mock_call_planitech.reset_mock() |
|
293 |
app.get( |
|
294 |
'/planitech/slug-planitech/getdays?start_time=10:00&&end_time=11:00' |
|
295 |
'&start_date=2018-11-11&min_capacity=15' |
|
296 |
) |
|
297 |
call_params = mock_call_planitech.call_args[0][2] |
|
298 |
assert call_params['placeIdentifiers'] == [2.0] |
|
299 | ||
300 |
freegaps = get_free_gaps()[:1] |
|
301 |
getfreegaps = { |
|
302 |
'availablePlaces': [ |
|
303 |
{ |
|
304 |
'freeGaps': freegaps |
|
305 |
}, |
|
306 |
{ |
|
307 |
'freeGaps': freegaps |
|
308 |
} |
|
309 |
] |
|
310 |
} |
|
311 |
mock_call_planitech = mock_planitech( |
|
312 |
monkeypatch, return_value=getfreegaps, referential=referential) |
|
313 |
response = app.get( |
|
314 |
'/planitech/slug-planitech/getdays?max_capacity=3&start_time=10:00&&end_time=11:00' |
|
315 |
'&start_date=2018-11-11' |
|
316 |
) |
|
317 |
assert response.json['data'] == [ |
|
318 |
{u'id': u'2018-11-11', u'text': u'dimanche 11 novembre 2018'}, |
|
319 |
] |
|
320 | ||
321 |
getfreegaps = { |
|
322 |
'availablePlaces': [ |
|
323 |
{ |
|
324 |
'freeGaps': [] |
|
325 |
}, |
|
326 |
{ |
|
327 |
'freeGaps': [] |
|
328 |
} |
|
329 |
] |
|
330 |
} |
|
331 |
mock_call_planitech = mock_planitech( |
|
332 |
monkeypatch, return_value=getfreegaps, referential=referential) |
|
333 |
response = app.get( |
|
334 |
'/planitech/slug-planitech/getdays?max_capacity=3&start_time=10:00&&end_time=11:00' |
|
335 |
'&start_date=2018-11-11' |
|
336 |
) |
|
337 |
assert response.json['data'] == [] |
|
338 | ||
339 |
# end date |
|
340 |
mock_call_planitech.reset_mock() |
|
341 |
response = app.get( |
|
342 |
'/planitech/slug-planitech/getdays?start_time=10:00&&end_time=11:00' |
|
343 |
'&start_date=2018-11-11&end_date=2018-11-12' |
|
344 |
) |
|
345 |
call_params = mock_call_planitech.call_args[0][2] |
|
346 |
assert call_params['startingDate'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc) |
|
347 |
assert call_params['endingDate'] == datetime(2018, 11, 12, 00, 0, tzinfo=utc) |
|
348 | ||
349 |
# start time |
|
350 |
mock_call_planitech.reset_mock() |
|
351 |
response = app.get( |
|
352 |
'/planitech/slug-planitech/getdays?start_time=11:00&&end_time=12:00' |
|
353 |
'&start_date=2018-11-11' |
|
354 |
) |
|
355 |
call_params = mock_call_planitech.call_args[0][2] |
|
356 |
assert call_params['startingDate'] == datetime(2018, 11, 11, 11, 0, tzinfo=utc) |
|
357 | ||
358 |
# duration |
|
359 |
mock_call_planitech.reset_mock() |
|
360 |
response = app.get( |
|
361 |
'/planitech/slug-planitech/getdays?start_time=11:00&&end_time=14:00' |
|
362 |
'&start_date=2018-11-11' |
|
363 |
) |
|
364 |
call_params = mock_call_planitech.call_args[0][2] |
|
365 |
assert call_params['requestedEndingTime'] == 180.0 |
|
366 | ||
367 | ||
368 |
def test_get_places(app, connector, monkeypatch, settings): |
|
369 |
referential = { |
|
370 |
2.0: { |
|
371 |
u'capacity': 20.0, u'label': u'salle 2', u'identifier': 2.0 |
|
372 |
}, |
|
373 |
1.0: { |
|
374 |
u'capacity': 10.0, u'label': u'salle 1', u'identifier': 1.0 |
|
375 |
} |
|
376 |
} |
|
377 | ||
378 |
def get_free_gaps(): |
|
379 |
return [ |
|
380 |
[datetime(year=2018, month=11, day=11, hour=10, minute=0), |
|
381 |
datetime(year=2018, month=11, day=11, hour=11, minute=0)], |
|
382 |
[datetime(year=2018, month=11, day=12, hour=10, minute=0), |
|
383 |
datetime(year=2018, month=11, day=12, hour=11, minute=0)] |
|
384 |
] |
|
385 | ||
386 |
getfreegaps = { |
|
387 |
'availablePlaces': [ |
|
388 |
{ |
|
389 |
'placeIdentifier': 1.0, |
|
390 |
'freeGaps': get_free_gaps() |
|
391 |
}, |
|
392 |
{ |
|
393 |
'placeIdentifier': 2.0, |
|
394 |
'freeGaps': get_free_gaps() |
|
395 |
} |
|
396 |
] |
|
397 |
} |
|
398 |
mock_call_planitech = mock_planitech( |
|
399 |
monkeypatch, return_value=getfreegaps, referential=referential) |
|
400 |
response = app.get( |
|
401 |
'/planitech/slug-planitech/getplaces?start_time=10:00&&end_time=11:00' |
|
402 |
'&start_date=2018-11-11' |
|
403 |
) |
|
404 |
assert response.json['data'] == [ |
|
405 |
{u'id': 1.0, u'text': u'salle 1'}, |
|
406 |
{u'id': 2.0, u'text': u'salle 2'} |
|
407 |
] |
|
408 |
call_params = mock_call_planitech.call_args[0][2] |
|
409 |
assert call_params['startingDate'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc) |
|
410 |
assert call_params['endingDate'] == datetime(2018, 11, 12, 10, 0, tzinfo=utc) |
|
411 |
assert call_params['placeIdentifiers'] == [1.0, 2.0] |
|
412 |
assert call_params['requestedStartingTime'] == 0.0 |
|
413 |
assert call_params['requestedEndingTime'] == 60.0 |
|
414 | ||
415 | ||
416 |
def test_login(connector): |
|
417 | ||
418 |
@urlmatch(netloc=r'(.*\.)?planitech\.com$') |
|
419 |
def planitech_mock(url, request): |
|
420 |
raise requests.exceptions.RequestException("Bad news") |
|
421 | ||
422 |
with HTTMock(planitech_mock): |
|
423 |
with pytest.raises(APIError) as excinfo: |
|
424 |
connector._login() |
|
425 |
assert str(excinfo.value) == 'Authentication to Planitech failed: Bad news' |
|
426 | ||
427 | ||
428 |
def test_update_reservation(app, connector, monkeypatch): |
|
429 |
mock_call_planitech = mock_planitech( |
|
430 |
monkeypatch, return_value={'modificationStatus': 'OK'} |
|
431 |
) |
|
432 |
response = app.post_json( |
|
433 |
'/planitech/slug-planitech/updatereservation', |
|
434 |
params={'status': 'confirmed', 'reservation_id': 1} |
|
435 |
) |
|
436 |
json_resp = response.json |
|
437 |
assert json_resp['err'] == 0 |
|
438 |
assert json_resp['data']['raw_data'] == {'modificationStatus': 'OK'} |
|
439 |
call_params = mock_call_planitech.call_args[0][2] |
|
440 |
assert call_params['reservationIdentifier'] == 1 |
|
441 |
assert call_params['situation'] == 3 |
|
442 | ||
443 |
# Update failed |
|
444 |
mock_planitech( |
|
445 |
monkeypatch, return_value={'modificationStatus': 'NOTOK'} |
|
446 |
) |
|
447 |
response = app.post_json( |
|
448 |
'/planitech/slug-planitech/updatereservation', |
|
449 |
params={'status': 'confirmed', 'reservation_id': 1} |
|
450 |
) |
|
451 |
json_resp = response.json |
|
452 |
assert json_resp['err'] == 1 |
|
453 |
assert json_resp['err_desc'] == 'Update reservation failed: NOTOK' |
|
454 | ||
455 |
# Connector bad param |
|
456 |
response = app.post_json( |
|
457 |
'/planitech/slug-planitech/updatereservation', params={'status': 'confirmed'}, status=400) |
|
458 |
json_resp = response.json |
|
459 |
assert json_resp['err'] == 1 |
|
460 |
assert json_resp['err_desc'] == "'reservation_id' is a required property" |
|
0 |
- |