/test/services/version1/test_BeaconsHttpServiceV1.py

import json
import time
from json import JSONDecodeError
from typing import Union

import requests
from pip_services3_commons.config import ConfigParams
from pip_services3_commons.refer import References, Descriptor
from pip_services3_commons.reflect import PropertyReflector
from pip_services3_commons.run import Parameters

from src.data.version1 import BeaconV1, BeaconTypeV1
from src.logic.BeaconsController import BeaconsController
from src.persistence.BeaconsMemoryPersistence import BeaconsMemoryPersistence
from src.services.version1.BeaconsHttpServiceV1 import BeaconsHttpServiceV1

BEACON1 = BeaconV1("1", "1", BeaconTypeV1.AltBeacon, "00001", "TestBeacon1", {"type": 'Point', "coordinates": [0, 0]},
                   50.0)
BEACON2 = BeaconV1("2", "1", BeaconTypeV1.iBeacon, "00002", "TestBeacon2", {"type": 'Point', "coordinates": [2, 2]},
                   70.0)
BEACON3 = BeaconV1("3", "2", BeaconTypeV1.AltBeacon, "00003", "TestBeacon3", {"type": 'Point', "coordinates": [10, 10]},
                   50.0)


class TestBeaconsHttpServiceV1:
    _persistence: BeaconsMemoryPersistence
    _controller: BeaconsController
    _service: BeaconsHttpServiceV1

    @classmethod
    def setup_class(cls):
        cls._persistence = BeaconsMemoryPersistence()
        cls._controller = BeaconsController()
        cls._service = BeaconsHttpServiceV1()

        cls._service.configure(ConfigParams.from_tuples(
            'connection.protocol', 'http',
            'connection.port', 3002,
            'connection.host', 'localhost'))

        references = References.from_tuples(Descriptor('beacons', 'persistence', 'memory', 'default', '1.0'),
                                            cls._persistence,
                                            Descriptor('beacons', 'controller', 'default', 'default', '1.0'),
                                            cls._controller,
                                            Descriptor('beacons', 'service', 'http', 'default', '1.0'),
                                            cls._service)
        cls._controller.set_references(references)
        cls._service.set_references(references)

        cls._persistence.open(None)
        cls._service.open(None)

    @classmethod
    def teardown_class(cls):
        cls._persistence.close(None)
        cls._service.close(None)

    def test_crud_operations(self):
        time.sleep(2)
        # Create the first beacon
        beacon1 = self.invoke("/v1/beacons/create_beacon",
                              Parameters.from_tuples("beacon", PropertyReflector.get_properties(BEACON1)))

        assert beacon1 is not None
        assert beacon1['id'] == BEACON1.id
        assert beacon1['site_id'] == BEACON1.site_id
        assert beacon1['udi'] == BEACON1.udi
        assert beacon1['type'] == BEACON1.type
        assert beacon1['label'] == BEACON1.label
        assert beacon1['center'] is not None

        # Create the second beacon
        beacon2 = self.invoke("/v1/beacons/create_beacon",
                              Parameters.from_tuples("beacon", PropertyReflector.get_properties(BEACON2)))

        assert beacon2 is not None
        assert beacon2['id'] == BEACON2.id
        assert beacon2['site_id'] == BEACON2.site_id
        assert beacon2['udi'] == BEACON2.udi
        assert beacon2['type'] == BEACON2.type
        assert beacon2['label'] == BEACON2.label
        assert beacon2['center'] is not None

        # Get all beacons
        page = self.invoke("/v1/beacons/get_beacons", Parameters.from_tuples("beacons"))
        assert page is not None
        assert len(page['data']) == 2

        beacon1 = page['data'][0]

        # Update the beacon
        beacon1['label'] = "ABC"
        beacon = self.invoke("/v1/beacons/update_beacon", Parameters.from_tuples("beacon", beacon1))
        assert beacon is not None
        assert beacon1['id'] == beacon['id']
        assert "ABC" == beacon['label']

        # Get beacon by udi
        beacon = self.invoke("/v1/beacons/get_beacon_by_udi", Parameters.from_tuples("udi", beacon1['udi']))
        assert beacon is not None
        assert beacon['id'] == beacon1['id']

        # Calculate position for one beacon
        position = self.invoke("/v1/beacons/calculate_position",
                               Parameters.from_tuples("site_id", '1', "udis", ['00001']))
        assert position is not None
        assert "Point" == position["type"]
        assert 2 == len(position["coordinates"])
        assert 0 == position["coordinates"][0]
        assert 0 == position["coordinates"][1]

        # Delete beacon
        self.invoke("/v1/beacons/delete_beacon_by_id", Parameters.from_tuples("id", beacon1['id']))

        # Try to get deleted beacon
        beacon = self.invoke("/v1/beacons/get_beacon_by_id", Parameters.from_tuples("id", beacon1['id']))
        assert beacon is False

    def invoke(self, route, entity) -> Union[bool, dict]:
        params = {}
        route = "http://localhost:3002" + route
        response = None
        timeout = 10000
        # Call the service
        data = json.dumps(entity)
        try:
            response = requests.request('POST', route, params=params, json=data, timeout=timeout)
            return response.json()
        except JSONDecodeError:
            if response.status_code == 404:
                return False

Run the python test.py command and make sure that all of the tests pass successfully.