Produits Entr'ouvert » Passerelle

Functional tests for the passerelle Planitech connector

Description

===========

This test suite will use the web API of a passerelle Planitech connector

to find an available reservation slot on a random place and perform a reservation. 

Usage

=====

You will need a running passerelle instance, whith a Planitech connector instance configured.

Suppose that the Planitech connector instance is listening here :

    http://127.0.0.1:8000/planitech/planitech

Then you would start the test suite with the following command:

    $ py.test --url=http://127.0.0.1:8000/planitech/planitech test_planitech.py

import pytest

def pytest_addoption(parser):

    parser.addoption(

        "--url", help="Url of a passerelle Planitech connector instance")

@pytest.fixture(scope='session')

def conn(request):

    return request.config.getoption("--url")

import datetime

import random

import urllib

import requests

def test_main(conn):

    # get a free gap

    today = datetime.datetime.now().date().isoformat()

    query_string = urllib.urlencode({

        'start_date': today, 'start_time': '10:00', 'end_time': '11:00'

    })

    url = conn + '/getdays?%s' % query_string

    resp = requests.get(url)

    resp.raise_for_status()

    res = resp.json()

    assert res['err'] == 0

    data = res['data']

    assert data

    date = data[0]['id']

    # get places

    query_string = urllib.urlencode({

        'start_date': date, 'start_time': '10:00', 'end_time': '11:00'

    })

    url = conn + '/getplaces?%s' % query_string

    resp = requests.get(url)

    resp.raise_for_status()

    res = resp.json()

    assert res['err'] == 0

    data = res['data']

    assert data

    place = data[random.randint(0, len(data) - 1)]['id']

    # create reservation

    params = {

        'date': date, 'start_time': '10:00', 'end_time': '11:00',

        'place_id': place, 'price': 200

    }

    url = conn + '/createreservation'

    resp = requests.post(url, json=params)

    resp.raise_for_status()

    res = resp.json()

    assert res['err'] == 0

    data = res['data']

    assert 'reservation_id' in data

    reservation_id = data['reservation_id']

    # confirm reservation

    params = {

        'reservation_id': reservation_id, 'status': 'standard'

    }

    url = conn + '/updatereservation'

    resp = requests.post(url, json=params)

    resp.raise_for_status()

    res = resp.json()

    assert res['err'] == 0

    data = res['data']

    assert data

    # FORBIDDEN BY PLANITECH - NO RESERVATION CANCELATION ?

    # cancel reservation

    # params = {

    #     'reservation_id': reservation_id, 'status': 'invalid'

    # }

    # url = conn + '/updatereservation'

    # resp = requests.post(url, json=params)

    # resp.raise_for_status()

    # res = resp.json()

    # assert res['err'] == 0

    # data = res['data']

    # assert data

# -*- coding: utf-8 -*-

# Generated by Django 1.11.15 on 2018-10-29 15:09

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0006_resourcestatus'),

    ]

    operations = [

        migrations.CreateModel(

            name='PlanitechConnector',

            fields=[

                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('description', models.TextField(verbose_name='Description')),

                ('slug', models.SlugField(unique=True)),

                ('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')),

                ('url', models.URLField(help_text='URL of the Planitech API endpoint', max_length=400, verbose_name='Planitech API endpoint')),

                ('username', models.CharField(max_length=128, verbose_name='Service username')),

                ('password', models.CharField(blank=True, max_length=128, null=True, verbose_name='Service password')),

                ('users', models.ManyToManyField(blank=True, to='base.ApiUser')),

                ('verify_cert', models.BooleanField(default=True, verbose_name='Check HTTPS Certificate validity')),

            ],

            options={

                'verbose_name': 'Planitech connector',

            },

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2018 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from datetime import datetime, time, timedelta

import hashlib

import json

import re

import urlparse

from django.conf import settings

from django.core.cache import cache

from django.db import models

from django.utils import dateformat

from django.utils import dateparse

from django.utils.translation import ugettext_lazy as _

from pytz import timezone, utc

from requests.exceptions import RequestException

from passerelle.base.models import BaseResource

from passerelle.contrib.planitech import mste

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

TZ = timezone(settings.TIME_ZONE)

CREATE_RESERVATION_SCHEMA = {

    "$schema": "http://json-schema.org/draft-03/schema#",

    "title": "Planitech createreservation",

    "description": "",

    "type": "object",

    "properties": {

        "date": {

            "description": "Date",

            "type": "string",

            "required": True

        },

        "start_time": {

            "description": "Start time",

            "type": "string",

            "required": True

        },

        "end_time": {

            "description": "End time",

            "type": "string",

            "required": True

        },

        "place_id": {

            "description": "Place idenfier",

            "type": "number",

            "required": True

        },

        "price": {

            "description": "Price",

            "type": "number",

            "required": True

        }

    }

}

RESERVATION_STATUS = {

    "confirmed": 3, "invalid": 0, " pre-reservation": 1,  "standard": 2

}

UPDATE_RESERVATION_SCHEMA = {

    "$schema": "http://json-schema.org/draft-03/schema#",

    "title": "Planitech updatereservation",

    "description": "",

    "type": "object",

    "properties": {

        "reservation_id": {

            "description": "Reservation Identifier",

            "type": "number",

            "required": True

        },

        "status": {

            "description": "Status of the reservation",

            "type": "string",

            "required": True,

            "enum": list(RESERVATION_STATUS.keys())

        }

    }

}

def parse_date(date_str):

    date_obj = dateparse.parse_date(date_str)

    if date_obj is None:

        raise APIError("Invalid date format: %s" % date_str)

    return date_obj

def parse_time(time_str):

    timeobj = dateparse.parse_time(time_str)

    if timeobj is None:

        raise APIError("Invalid time format: %s" % time_str)

    return timeobj

def compute_hash(content, hardness, salt):

    sha = hashlib.new('sha512', salt + content)

    for idx in range(hardness):

        sha = hashlib.new('sha512', sha.digest())

    return sha.hexdigest().upper()

def date_to_datetime(date_str):

    date_obj = parse_date(date_str)

    if date_obj is None:

        raise APIError("Invalid date string: %s" % date_str)

    return datetime.combine(date_obj, time(hour=12))

def get_duration(start_time_str, end_time_str):

    start_time = parse_time(start_time_str)

    end_time = parse_time(end_time_str)

    return time_to_minutes(end_time) - time_to_minutes(start_time)

def get_salt(salt):

    return re.match(r'<(.*?)>', salt).groups()[0]

def get_utc_datetime(date_str, time_str):

    date_obj = parse_date(date_str)

    time_obj = parse_time(time_str)

    datetime_obj = datetime.combine(date_obj, time_obj)

    return TZ.localize(datetime_obj).astimezone(utc)

def time_to_minutes(timeobj):

    return float(timeobj.hour * 60 + timeobj.minute)

class PlanitechConnector(BaseResource):

    url = models.URLField(

        max_length=400, verbose_name=_('Planitech API endpoint'),

        help_text=_('URL of the Planitech API endpoint'))

    username = models.CharField(max_length=128, verbose_name=_('Service username'))

    password = models.CharField(

        max_length=128, verbose_name=_('Service password'), null=True, blank=True)

    verify_cert = models.BooleanField(

        default=True, verbose_name=_('Check HTTPS Certificate validity'))

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('Planitech')

    def _call_planitech(self, session_meth, endpoint, params=None):

        if not getattr(self, '_planitech_session', False):

            self._login()

            self._planitech_session = True

        kwargs = {}

        if params is not None:

            kwargs['data'] = json.dumps(mste.encode(params))

        response = session_meth(urlparse.urljoin(self.url, endpoint), **kwargs)

        if response.status_code != 200:

            error_msg = "Planitech error %s" % response.status_code

            try:

                data = mste.decode(response.json())

                if hasattr(data, 'get'):

                    error = data.get('errors')

                    if error:

                        error_msg += " - %s" % error

            except TypeError:

                pass

            raise APIError(error_msg)

        return mste.decode(response.json())

    def _get_places_by_capacity(self, min_capacity, max_capacity):

        places = self._get_places_referential()

        min_capacity = int(min_capacity)

        max_capacity = int(max_capacity)

        return [place['identifier'] for place in places.values()

                if (min_capacity <= place['capacity'] <= max_capacity)

                and place['capacity']]

    def _get_places_referential(self, refresh_cache=False):

        cache_key = 'planitech-%s-places' % self.id

        ref = cache.get(cache_key)

        if ref is None or refresh_cache:

            data = self._call_planitech(self.requests.get, 'getPlacesList')

            ref = {}

            for place in data['placesList']:

                ref[place['identifier']] = {

                    'identifier': place['identifier'], 'label': place['label']

                }

            data = self._call_planitech(

                self.requests.post, 'getPlacesInfo',

                {

                    "placeIdentifiers": list(ref.keys()),

                    "extensionAttributes": {

                        "capacity": {

                            "name": "TOTCAP",

                            "type": "int"

                        }

                    }

                }

            )

            for place in data['requestedPlaces']:

                ref[place['identifier']]['capacity'] = place.get('capacity')

            cache.set(cache_key, ref)

        return ref

    def _login(self):

        try:

            auth_url = urlparse.urljoin(self.url, 'auth')

            response = self.requests.get(auth_url, headers={'MH-LOGIN': self.username})

            response.raise_for_status()

            part1, salt1, part2, salt2, _ = re.split(r'(\<.*?\>)', response.content)

            hardness1 = int(part1.split(':')[1])

            hardness2 = int(part2.split(':')[1])

            salt1 = get_salt(salt1)

            salt2 = get_salt(salt2)

            tmp_hash = compute_hash(self.password, hardness1, salt1)

            hash_pass = compute_hash(tmp_hash, hardness2, salt2)

            response = self.requests.get(auth_url, headers={'MH-PASSWORD': hash_pass})

            response.raise_for_status()

            # the last response should have set a cookie which will be used for authentication

        except RequestException as e:

            raise APIError("Authentication to Planitech failed: %s" % str(e))

    def check_status(self):

        self._call_planitech(self.requests.get, 'getCapabilities')

    @endpoint(

        perm='can_access',

        post={

            'description': _('Create reservation'),

            'request_body': {

                'schema': {

                    'application/json': CREATE_RESERVATION_SCHEMA

                }

            }

        }

    )

    def createreservation(self, request, post_data):

        start_datetime = get_utc_datetime(post_data['date'], post_data['start_time'])

        end_datetime = get_utc_datetime(post_data['date'], post_data['end_time'])

        request_date = datetime.now(tz=utc)

        params = {

            "activityID": mste.Uint32(6),  # relaxation FIXME

            "contractorExternalIdentifier": "test-entrouvert",  # FIXME

            "end": end_datetime,

            "isWeekly": False,

            "object": "reservation  test entouvert",  # FIXME

            "places": [float(post_data['place_id'])],

            "price": mste.Uint32(post_data['price']),

            "requestDate": request_date,

            "start": start_datetime,

            "typeID": mste.Uint32(2),  # payant FIXME

            "vatRate": mste.Uint32(4000)  # FIXME

        }

        data = self._call_planitech(self.requests.post, 'createReservation', params)

        if data.get('creationStatus') != 'OK':

            raise APIError("Reservation creation failed: %s" % data.get('creationStatus'))

        reservation_id = data.get('reservationIdentifier')

        if not reservation_id:

            raise APIError("Reservation creation failed: no reservation ID")

        return {

            'data': {

                'reservation_id': reservation_id,

                'raw_data': data

            }

        }

    def hourly(self):

        self._get_places_referential(refresh_cache=True)

    @endpoint(

        description_get=_('Get days available for reservation'),

        methods=['get'], perm='can_access',

        parameters={

            'min_capacity': {

                'description': _('Minimum capacity'),

                'example_value': '1',

            },

            'max_capacity': {

                'description': _('Maximum capacity'),

                'example_value': '10',

            },

            'start_date': {

                'description': _('Start date'),

                'example_value': '2018-10-10',

            },

            'end_date': {

                'description': _('End date'),

                'example_value': '2018-12-10',

            },

            'start_time': {

                'description': _('Start time'),

                'example_value': '10:00',

            },

            'end_time': {

                'description': _('End time'),

                'example_value': '18:00',

            },

            'weekdays': {

                'description': _('Week days'),

                'example_value': 'true',

                'type': 'bool',

            },

        })

    def getdays(

            self, request, start_date, start_time, end_time, min_capacity=0,

            end_date=None, max_capacity=100000, weekdays=False):

        places_id = self._get_places_by_capacity(int(min_capacity), int(max_capacity))

        utc_start_datetime = get_utc_datetime(start_date, start_time)

        if end_date is None:

            utc_end_datetime = utc_start_datetime + timedelta(days=365)

        else:

            utc_end_datetime = get_utc_datetime(end_date, '00:00')

        duration = get_duration(start_time, end_time)

        params = {

            "placeIdentifiers": places_id,

            "startingDate": utc_start_datetime,

            "endingDate": utc_end_datetime,

            "requestedStartingTime": float(0),

            "requestedEndingTime": duration

        }

        if not weekdays:

            params['reservationDays'] = [mste.Uint32(0), mste.Uint32(6)]

        raw_data = self._call_planitech(self.requests.post, 'getFreeGaps', params)

        available_dates = set()

        for place in raw_data.get('availablePlaces', []):

            for freegap in place.get('freeGaps', []):

                available_dates.add(freegap[0].date())

        res = []

        available_dates = list(available_dates)

        available_dates.sort()

        for date_obj in available_dates:

            date_text = dateformat.format(date_obj, 'l d F Y')

            res.append({"id": date_obj.isoformat(), "text": date_text})

        return {'data': res}

    @endpoint(

        description_get=_('Get places available for reservation'),

        methods=['get'], perm='can_access',

        parameters={

            'min_capacity': {

                'description': _('Minimum capacity'),

                'example_value': '1',

            },

            'max_capacity': {

                'description': _('Maximum capacity'),

                'example_value': '10',

            },

            'start_date': {

                'description': _('Start date'),

                'example_value': '2018-10-10',

            },

            'start_time': {

                'description': _('Start time'),

                'example_value': '10:00',

            },

            'end_time': {

                'description': _('End time'),

                'example_value': '18:00',

            },

        })

    def getplaces(

            self, request, start_date, start_time, end_time, min_capacity=0, max_capacity=100000):

        places_id = self._get_places_by_capacity(min_capacity, max_capacity)

        utc_start_datetime = get_utc_datetime(start_date, start_time)

        utc_end_datetime = utc_start_datetime + timedelta(days=1)

        duration = get_duration(start_time, end_time)

        params = {

            "placeIdentifiers": places_id,

            "startingDate": utc_start_datetime,

            "endingDate": utc_end_datetime,

            "requestedStartingTime": float(0),

            "requestedEndingTime": duration

        }

        raw_data = self._call_planitech(self.requests.post, 'getFreeGaps', params)

        available_places = []

        for place in raw_data.get('availablePlaces', []):

            available_places.append(place['placeIdentifier'])

        places_ref = self._get_places_referential()

        res = []

        for place in available_places:

            res.append({"id": place, "text": places_ref[place]['label']})

        return {'data': res}

    @endpoint(description_get=_('Get places referential'), methods=['get'], perm='can_access')

    def getplacesreferential(self, request):

        return {'data': self._get_places_referential()}

    @endpoint(

        methods=['post'], perm='can_access',

        post={

            'description': _('Update reservation'),

            'request_body': {

                'schema': {

                    'application/json': UPDATE_RESERVATION_SCHEMA

                }

            }

        }

    )

    def updatereservation(self, request, post_data):

        params = {

            "reservationIdentifier": mste.Uint32(post_data['reservation_id']),

            "situation": mste.Uint32(RESERVATION_STATUS[post_data['status']])

        }

        data = self._call_planitech(self.requests.post, 'updateReservation', params)

        if data.get('modificationStatus') != 'OK':

            raise APIError("Update reservation failed: %s" % data.get('modificationStatus'))

        return {

            'data': {

                'raw_data': data

            }

        }

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2018 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from datetime import datetime

import calendar

ENCODE_TOKENS = {

    'integer': 16,

    'real': 19,

    'nil':  0,

    'true': 1,

    'false': 2,

    'emptyString': 3,

    'emptyData': 4,

    'ref': 9,

    'i1': 10,

    'u1': 11,

    'i2': 12,

    'u2': 13,

    'i4': 14,

    'uint32': 15,

    'i8': 16,

    'u8': 17,

    'float': 18,

    'double': 19,

    'decimal': 20,

    'string': 21,

    'localdate': 22,

    'gmtDate': 23,

    'color': 24,

    'data': 25,

    'naturals': 26,

    'dictionary': 30,

    'array': 31,

    'couple': 32,

    'object': 50,

}

DECODE_TOKENS = {v: k for k, v in ENCODE_TOKENS.items()}

class Couple(list):

    pass

class Uint32(int):

    pass

class MSTEDecoder(object):

    def __init__(self, data):

        self._idx = 4

        self._keys = []

        self._refs = []

        self._data = data

    def _next_token(self):

        self._idx += 1

        return self._data[self._idx]

    def _parse_array(self):

        count = self._next_token()

        res = []

        self._refs.append(res)

        while count > 0:

            res.append(self._parse_item())

            count -= 1

        return res

    def _parse_couple(self):

        res = Couple()

        self._refs.append(res)

        for _ in range(2):

            res.append(self._parse_item())

        return res

    def _parse_decimal(self):

        res = float(self._next_token())

        self._refs.append(res)

        return res

    def _parse_dictionary(self):

        count = self._next_token()

        res = {}

        self._refs.append(res)

        while count > 0:

            key = self._keys[self._next_token()]

            res[key] = self._parse_item()

            count -= 1

        return res

    def _parse_emptyString(self):

        return ''

    def _parse_gmtDate(self):

        return self._parse_localdate()

    def _parse_item(self):

        token = self._next_token()

        _type = DECODE_TOKENS[token]

        return getattr(self, "_parse_%s" % _type)()

    def _parse_localdate(self):

        timestamp = self._next_token()

        res = datetime.fromtimestamp(timestamp)

        self._refs.append(res)

        return res

    def _parse_nil(self):

        return None

    def _parse_ref(self):

        pos = self._next_token()

        return self._refs[pos]

    def _parse_string(self):

        res = self._next_token()

        self._refs.append(res)

        return res

    def _parse_true(self):

        return True

    def _parse_false(self):

        return False

    def _parse_uint32(self):

        return int(self._next_token())

    def decode(self):

        num_keys = self._data[self._idx]

        while num_keys > 0:

            self._keys.append(self._next_token())

            num_keys -= 1

        return self._parse_item()

class ObjectStore(list):

    def add(self, obj):

        """ Add object in the store

        and return its reference

        """

        ref = self.getref(obj)

        if ref is None:

            self.append(obj)

            ref = len(self) - 1

        return ref

    def getref(self, obj):

        """ Return the reference of obj,

        None if the object is not in the store

        """

        try:

            return self.index(obj)

        except ValueError:

            return None

class MSTEEncoder(object):

    def __init__(self, data):

        self._data = data

        self._stream = []

        self._refs_store = ObjectStore()

        self._keys_store = ObjectStore()

    def _push_token_type(self, token_type):

        self._stream.append(ENCODE_TOKENS[token_type])

    def _push(self, item):

        self._stream.append(item)

    def _encode_array(self, obj):

        self._refs_store.add(obj)

        self._push_token_type('array')

        self._push(len(obj))

        for elem in obj:

            self._encode_obj(elem)

    def _encode_boolean(self, obj):

        if obj:

            self._push_token_type('true')

        else:

            self._push_token_type('false')

    def _encode_couple(self, obj):

        self._refs_store.add(obj)

        self._push_token_type('couple')

        for elem in obj:

            self._encode_obj(elem)

    def _encode_decimal(self, obj):

        self._refs_store.add(obj)

        self._push_token_type('decimal')

        self._push(int(obj))

    def _encode_dictionary(self, obj):

        self._refs_store.add(obj)

        self._push_token_type('dictionary')

        self._push(len(obj))

        for key, value in obj.items():

            key_ref = self._keys_store.add(key)

            self._push(key_ref)

            self._encode_obj(value)

    def _encode_localdate(self, obj):

        # obj must be utc

        self._refs_store.add(obj)

        self._push_token_type('localdate')

        self._push(int(calendar.timegm(obj.timetuple())))

    def _encode_obj(self, obj):

        ref = self._refs_store.getref(obj)

        if ref is not None:

            self._push_token_type('ref')

            self._push(ref)

        elif isinstance(obj, unicode) or isinstance(obj, str):

            self._encode_string(obj)

        elif obj is None:

            self._encode_nil()

        elif isinstance(obj, Couple):

            self._encode_couple(obj)

        elif isinstance(obj, bool):

            self._encode_boolean(obj)

        elif isinstance(obj, list):

            self._encode_array(obj)

        elif isinstance(obj, dict):

            self._encode_dictionary(obj)

        elif isinstance(obj, float):

            self._encode_decimal(obj)

        elif isinstance(obj, datetime):

            self._encode_localdate(obj)

        elif isinstance(obj, Uint32):

            self._encode_uint32(obj)

        else:

            raise TypeError("%s encoding not supported" % type(obj))

    def _encode_nil(self):

        self._push_token_type('nil')

    def _encode_string(self, _str):

        if _str:

            self._push_token_type('string')

            self._push(_str)

            self._refs_store.add(_str)

        else:

            self._push_token_type('emptyString')

    def _encode_uint32(self, obj):

        self._push_token_type('uint32')

        self._push(obj)

    def encode(self):

        res = ["MSTE0102"]

        self._encode_obj(self._data)

        nb_token = 5 + len(self._keys_store) + len(self._stream)

        res = ["MSTE0102", nb_token, 'CRC00000000', 0, len(self._keys_store)] + self._keys_store

        res.extend(self._stream)

        return res

def decode(data):

    return MSTEDecoder(data).decode()

def encode(data):

    return MSTEEncoder(data).encode()

        'passerelle.contrib.planitech',

from datetime import datetime

from django.contrib.contenttypes.models import ContentType

from django.core.cache import cache

from httmock import urlmatch, HTTMock

import mock

import pytest

from pytz import utc

import requests

from passerelle.base.models import ApiUser, AccessRight

from passerelle.contrib.planitech import mste

from passerelle.contrib.planitech.models import PlanitechConnector

from passerelle.utils.jsonresponse import APIError

def assert_mste(data, ref_data):

    """ skip CRC verification

    """

    assert len(data) == len(ref_data)

    for i in range(len(data)):

        if i != 2:

            assert data[i] == ref_data[i]

@pytest.mark.parametrize("data,mste_data", [

    (None, ["MSTE0102", 6, "CRC82413E70", 0, 0, 0]),

    ("toto", ["MSTE0102", 7, "CRCD45ACB10", 0, 0, 21, "toto"]),  # string

    (mste.Couple(("toto", "tata")), ["MSTE0102", 10, "CRCD45ACB10", 0, 0, 32, 21, "toto", 21,

                                     "tata"]),

    # couple

    ([mste.Couple(("toto", "tata")), mste.Couple(("toto", "tata"))],

     ["MSTE0102", 14, "CRCD45ACB10", 0, 0, 31, 2, 32, 21, "toto", 21, "tata", 9, 1]),

    # couple are stored in refs

    (["toto"], ["MSTE0102", 9, "CRCD4E14B75", 0, 0, 31, 1, 21, "toto"]),  # array

    (["toto", "tata", "toto"], ["MSTE0102", 13, "CRC7311752F", 0, 0, 31, 3, 21, "toto", 21,

                                "tata", 9, 1]),  # array with reference

    ({"mykey": "toto"}, ["MSTE0102",  11, "CRC1C9E9FE1", 0, 1, "mykey", 30, 1, 0, 21, "toto"]),

    # dictionnary

    ([{"mykey": "toto"}, {"mykey": "toto"}],

     ["MSTE0102",  15, "CRC1C9E9FE1", 0, 1, "mykey", 31, 2, 30, 1, 0, 21,

      "toto", 9, 1]),

    # dictionnary are stored in refs

    (float(2), ["MSTE0102",  7, "CRC1C9E9FE1", 0, 0, 20, 2]),  # decimal

    ([float(2), float(2)], ["MSTE0102",  11, "CRC1C9E9FE1", 0, 0, 31, 2, 20, 2, 9, 1]),

    # decimal are stored in refs

    (mste.Uint32(1), ["MSTE0102",  7, "CRC1C9E9FE1", 0, 0, 15, 1]),  # uint32

    (True, ["MSTE0102",  6, "CRC1C9E9FE1", 0, 0, 1]),  # True

    (False, ["MSTE0102",  6, "CRC1C9E9FE1", 0, 0, 2]),  # False

    ('', ["MSTE0102",  6, "CRC1C9E9FE1", 0, 0, 3]),  # empty string

    (datetime.fromtimestamp(1537364340), ["MSTE0102",  7, "CRC1C9E9FE1", 0, 0, 22, 1537364340]),

    # local date

    ([datetime.fromtimestamp(1537364340), datetime.fromtimestamp(1537364340)],

     ["MSTE0102", 11, "CRC1C9E9FE1", 0, 0, 31, 2, 22, 1537364340, 9, 1]),

    # local date in refs

])

def test_mste(mste_data, data):

    assert data == mste.decode(mste_data)

    assert_mste(mste.encode(data), mste_data)

def test_encode_unsupported_type():

    with pytest.raises(TypeError):

        mste.encode(set())

def test_real():

    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]

    mste.decode(mste_data)

@pytest.fixture()

def connector(db):

    api = ApiUser.objects.create(username='all', keytype='', key='')

    connector = PlanitechConnector.objects.create(

        url='http://example.planitech.com/', username='admin', password='admin',

        verify_cert=False, slug='slug-planitech')

    obj_type = ContentType.objects.get_for_model(connector)

    AccessRight.objects.create(

        codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=connector.pk)

    return connector

def mock_planitech(monkeypatch, return_value=None, side_effect=None, referential=None):

    from passerelle.contrib.planitech import models

    monkeypatch.setattr(models.PlanitechConnector, '_login', mock.Mock())

    kwargs = {}

    if return_value is not None:

        kwargs['return_value'] = return_value

    if side_effect is not None:

        kwargs['side_effect'] = side_effect

    mock_call_planitech = mock.Mock(**kwargs)

    monkeypatch.setattr(models.PlanitechConnector, '_call_planitech', mock_call_planitech)

    if referential is not None:

        mock_get_referential = mock.Mock(return_value=referential)

        monkeypatch.setattr(

            models.PlanitechConnector, '_get_places_referential', mock_get_referential)

    return mock_call_planitech

def test_call_planitech(connector, monkeypatch):

    class MockResponse(object):

        status_code = 200

        content = None

        def __init__(self, content=None, status_code=None):

            if content is not None:

                self.content = content

            if status_code is not None:

                self.status_code = status_code

        def session_meth(self, *args, **kwargs):

            return self

        def json(self):

            return mste.encode(self.content)

    connector._planitech_session = True

    response = MockResponse(content='somestring')

    assert connector._call_planitech(response.session_meth, 'endpoint') == "somestring"

    response = MockResponse(content=set(), status_code=400)

    with pytest.raises(APIError) as excinfo:

        connector._call_planitech(response.session_meth, 'endpoint')

    assert str(excinfo.value) == 'Planitech error 400'

    response = MockResponse(content='unexpected error format', status_code=400)

    with pytest.raises(APIError) as excinfo:

        connector._call_planitech(response.session_meth, 'endpoint')

    assert str(excinfo.value) == 'Planitech error 400'

    response = MockResponse(content={'errors': 'planitech error message'}, status_code=400)

    with pytest.raises(APIError) as excinfo:

        connector._call_planitech(response.session_meth, 'endpoint')

    assert str(excinfo.value) == 'Planitech error 400 - planitech error message'

def test_create_reservation(app, connector, monkeypatch):

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value={

            'creationStatus': 'OK',

            'reservationIdentifier': 1

        })

    response = app.post_json(

        '/planitech/slug-planitech/createreservation',

        params={

            'date': '2018-11-11',

            'start_time': '10:00',

            'end_time': '11:00',

            'place_id': 1,

            'price': 10

        }

    )

    json_resp = response.json

    assert json_resp['err'] == 0

    assert json_resp['data']['reservation_id'] == 1

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['start'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc)

    assert call_params['end'] == datetime(2018, 11, 11, 11, 0, tzinfo=utc)

    assert call_params['places'] == [1]

    assert call_params['price'] == 10

    # Create reservation failed

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value={

            'creationStatus': 'NOTOK',

            'reservationIdentifier': 1

        })

    response = app.post_json(

        '/planitech/slug-planitech/createreservation',

        params={

            'date': '2018-11-11',

            'start_time': '10:00',

            'end_time': '11:00',

            'place_id': 1,

            'price': 10

        }

    )

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == 'Reservation creation failed: NOTOK'

    # Create reservation failed - nor reservation ID

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value={

            'creationStatus': 'OK'

        })

    response = app.post_json(

        '/planitech/slug-planitech/createreservation',

        params={

            'date': '2018-11-11',

            'start_time': '10:00',

            'end_time': '11:00',

            'place_id': 1,

            'price': 10

        }

    )

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == 'Reservation creation failed: no reservation ID'

def test_getplaces_referential(app, connector, monkeypatch):

    side_effect = [

        {

            'placesList': [

                {'identifier': 1.0, 'label': 'salle 1'},

                {'identifier': 2.0, 'label': 'salle 2'}

            ]

        },

        {

            'requestedPlaces': [

                {'identifier': 1.0, 'capacity': 10.0},

                {'identifier': 2.0, 'capacity': 20.0}

            ]

        }

    ]

    mock_planitech(monkeypatch, side_effect=side_effect)

    response = app.get('/planitech/slug-planitech/getplacesreferential')

    expected_res = {

        u'2.0': {

            u'capacity': 20.0, u'label': u'salle 2', u'identifier': 2.0

        },

        u'1.0': {

            u'capacity': 10.0, u'label': u'salle 1', u'identifier': 1.0

        }

    }

    assert response.json['data'] == expected_res

def test_getplaces_referential_use_cache(app, connector):

    cache_key = 'planitech-%s-places' % connector.id

    cache.set(cache_key, {'some': 'data'})

    response = app.get('/planitech/slug-planitech/getplacesreferential')

    assert response.json_body['data'] == {'some': 'data'}

    cache.delete(cache_key)

def test_get_days(app, connector, monkeypatch, settings):

    settings.LANGUAGE_CODE = 'fr-fr'

    referential = {

        2.0: {

            u'capacity': 20.0, u'label': u'salle 2', u'identifier': 2.0

        },

        1.0: {

            u'capacity': 10.0, u'label': u'salle 1', u'identifier': 1.0

        }

    }

    def get_free_gaps():

        return [

            [datetime(year=2018, month=11, day=11, hour=10, minute=0),

             datetime(year=2018, month=11, day=11, hour=11, minute=0)],

            [datetime(year=2018, month=11, day=12, hour=10, minute=0),

             datetime(year=2018, month=11, day=12, hour=11, minute=0)]

        ]

    getfreegaps = {

        'availablePlaces': [

            {

                'freeGaps': get_free_gaps()

            },

            {

                'freeGaps': get_free_gaps()

            }

        ]

    }

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value=getfreegaps, referential=referential)

    response = app.get(

        '/planitech/slug-planitech/getdays?start_time=10:00&&end_time=11:00'

        '&start_date=2018-11-11'

    )

    assert response.json['data'] == [

        {u'id': u'2018-11-11', u'text': u'dimanche 11 novembre 2018'},

        {u'id': u'2018-11-12', u'text': u'lundi 12 novembre 2018'}

    ]

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['startingDate'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc)

    assert call_params['endingDate'] == datetime(2019, 11, 11, 10, 0, tzinfo=utc)

    assert call_params['placeIdentifiers'] == [1.0, 2.0]

    assert call_params['requestedStartingTime'] == 0.0

    assert call_params['requestedEndingTime'] == 60.0

    assert call_params['reservationDays'] == [0, 6]

    # capcacity

    mock_call_planitech.reset_mock()

    app.get(

        '/planitech/slug-planitech/getdays?start_time=10:00&&end_time=11:00'

        '&start_date=2018-11-11&min_capacity=15'

    )

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['placeIdentifiers'] == [2.0]

    freegaps = get_free_gaps()[:1]

    getfreegaps = {

        'availablePlaces': [

            {

                'freeGaps': freegaps

            },

            {

                'freeGaps': freegaps

            }

        ]

    }

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value=getfreegaps, referential=referential)

    response = app.get(

        '/planitech/slug-planitech/getdays?max_capacity=3&start_time=10:00&&end_time=11:00'

        '&start_date=2018-11-11'

    )

    assert response.json['data'] == [

        {u'id': u'2018-11-11', u'text': u'dimanche 11 novembre 2018'},

    ]

    getfreegaps = {

        'availablePlaces': [

            {

                'freeGaps': []

            },

            {

                'freeGaps': []

            }

        ]

    }

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value=getfreegaps, referential=referential)

    response = app.get(

        '/planitech/slug-planitech/getdays?max_capacity=3&start_time=10:00&&end_time=11:00'

        '&start_date=2018-11-11'

    )

    assert response.json['data'] == []

    # end date

    mock_call_planitech.reset_mock()

    response = app.get(

        '/planitech/slug-planitech/getdays?start_time=10:00&&end_time=11:00'

        '&start_date=2018-11-11&end_date=2018-11-12'

    )

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['startingDate'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc)

    assert call_params['endingDate'] == datetime(2018, 11, 12, 00, 0, tzinfo=utc)

    # start time

    mock_call_planitech.reset_mock()

    response = app.get(

        '/planitech/slug-planitech/getdays?start_time=11:00&&end_time=12:00'

        '&start_date=2018-11-11'

    )

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['startingDate'] == datetime(2018, 11, 11, 11, 0, tzinfo=utc)

    # duration

    mock_call_planitech.reset_mock()

    response = app.get(

        '/planitech/slug-planitech/getdays?start_time=11:00&&end_time=14:00'

        '&start_date=2018-11-11'

    )

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['requestedEndingTime'] == 180.0

    # date bad format

    response = app.get(

        '/planitech/slug-planitech/getdays?start_time=11:00&&end_time=14:00'

        '&start_date=notadate'

    )

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == "Invalid date format: notadate"

    # date time format

    response = app.get(

        '/planitech/slug-planitech/getdays?start_time=notatime&&end_time=14:00'

        '&start_date=2018-11-11'

    )

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == "Invalid time format: notatime"

def test_get_places(app, connector, monkeypatch, settings):

    referential = {

        2.0: {

            u'capacity': 20.0, u'label': u'salle 2', u'identifier': 2.0

        },

        1.0: {

            u'capacity': 10.0, u'label': u'salle 1', u'identifier': 1.0

        }

    }

    def get_free_gaps():

        return [

            [datetime(year=2018, month=11, day=11, hour=10, minute=0),

             datetime(year=2018, month=11, day=11, hour=11, minute=0)],

            [datetime(year=2018, month=11, day=12, hour=10, minute=0),

             datetime(year=2018, month=11, day=12, hour=11, minute=0)]

        ]

    getfreegaps = {

        'availablePlaces': [

            {

                'placeIdentifier': 1.0,

                'freeGaps': get_free_gaps()

            },

            {

                'placeIdentifier': 2.0,

                'freeGaps': get_free_gaps()

            }

        ]

    }

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value=getfreegaps, referential=referential)

    response = app.get(

        '/planitech/slug-planitech/getplaces?start_time=10:00&&end_time=11:00'

        '&start_date=2018-11-11'

    )

    assert response.json['data'] == [

        {u'id': 1.0, u'text': u'salle 1'},

        {u'id': 2.0, u'text': u'salle 2'}

    ]

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['startingDate'] == datetime(2018, 11, 11, 10, 0, tzinfo=utc)

    assert call_params['endingDate'] == datetime(2018, 11, 12, 10, 0, tzinfo=utc)

    assert call_params['placeIdentifiers'] == [1.0, 2.0]

    assert call_params['requestedStartingTime'] == 0.0

    assert call_params['requestedEndingTime'] == 60.0

    # date bad format

    response = app.get(

        '/planitech/slug-planitech/getplaces?start_time=10:00&&end_time=11:00'

        '&start_date=notadate'

    )

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == "Invalid date format: notadate"

    # date time format

    response = app.get(

        '/planitech/slug-planitech/getplaces?start_time=notatime&&end_time=11:00'

        '&start_date=2018-11-11'

    )

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == "Invalid time format: notatime"

def test_login(connector):

    @urlmatch(netloc=r'(.*\.)?planitech\.com$')

    def planitech_mock(url, request):

        raise requests.exceptions.RequestException("Bad news")

    with HTTMock(planitech_mock):

        with pytest.raises(APIError) as excinfo:

            connector._login()

        assert str(excinfo.value) == 'Authentication to Planitech failed: Bad news'

def test_update_reservation(app, connector, monkeypatch):

    mock_call_planitech = mock_planitech(

        monkeypatch, return_value={'modificationStatus': 'OK'}

    )

    response = app.post_json(

        '/planitech/slug-planitech/updatereservation',

        params={'status': 'confirmed', 'reservation_id': 1}

    )

    json_resp = response.json

    assert json_resp['err'] == 0

    assert json_resp['data']['raw_data'] == {'modificationStatus': 'OK'}

    call_params = mock_call_planitech.call_args[0][2]

    assert call_params['reservationIdentifier'] == 1

    assert call_params['situation'] == 3

    # Update failed

    mock_planitech(

        monkeypatch, return_value={'modificationStatus': 'NOTOK'}

    )

    response = app.post_json(

        '/planitech/slug-planitech/updatereservation',

        params={'status': 'confirmed', 'reservation_id': 1}

    )

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == 'Update reservation failed: NOTOK'

    # Connector bad param

    response = app.post_json(

        '/planitech/slug-planitech/updatereservation', params={'status': 'confirmed'}, status=400)

    json_resp = response.json

    assert json_resp['err'] == 1

    assert json_resp['err_desc'] == "'reservation_id' is a required property"