Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import logging, pytest, requests, time
from google.protobuf.json_format import MessageToDict
from common.orm.Database import Database
from common.orm.Factory import get_database_backend, BackendEnum
from context.proto.context_pb2 import Context, Topology
from context.Config import RESTAPI_SERVICE_PORT, RESTAPI_BASE_URL
from context.service.rest_server.Server import Server
from context.service.rest_server.resources.Context import Context
#from .populate_database import populate_example
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
RESTAPI_PORT = 10000 + RESTAPI_SERVICE_PORT # avoid privileged ports
SCENARIOS = [
(BackendEnum.INMEMORY, {}),
#(BackendEnum.REDIS, {
# 'REDIS_SERVICE_HOST': '10.1.7.194',
# 'REDIS_SERVICE_PORT': 30283,
# 'REDIS_DATABASE_ID': 0,
#}),
]
@pytest.fixture(scope='session', ids=[str(scenario[0].value) for scenario in SCENARIOS], params=SCENARIOS)
def context_database(request):
backend,settings = request.param
LOGGER.info('Running fixture with backend={}, settings={}...'.format(str(backend), str(settings)))
database_backend = get_database_backend(backend=backend, **settings)
_database = Database(database_backend)
return _database
@pytest.fixture(scope='session')
def context_service_rest(context_database : Database): # pylint: disable=redefined-outer-name
_rest_server = Server(port=RESTAPI_PORT, base_url=RESTAPI_BASE_URL)
_rest_server.add_resource(Context, '/context', endpoint='api.context', resource_class_args=(context_database,))
_rest_server.start()
time.sleep(1) # bring time for the server to start
yield _rest_server
_rest_server.shutdown()
_rest_server.join()
def test_get_topology_completed_rest_api(context_service_rest : Server): # pylint: disable=redefined-outer-name
# should work
request_url = 'http://127.0.0.1:{}{}/context'.format(RESTAPI_PORT, RESTAPI_BASE_URL)
LOGGER.warning('Request: GET {}'.format(str(request_url)))
reply = requests.get(request_url)
LOGGER.warning('Reply: {}'.format(str(reply.text)))
assert reply.status_code == 200, 'Reply failed with code {}'.format(reply.status_code)
json_reply = reply.json()
topology = MessageToDict(
Topology(**json_reply['topologies'][0]),
including_default_value_fields=True, preserving_proto_field_name=True,
use_integers_for_enums=False)
validate_topology(topology)
validate_topology_has_devices(topology)
validate_topology_has_links(topology)