diff --git a/app/api/main.py b/app/api/main.py
index e2cf9ca50382db8cd61a974147881cc5f284cbe5..f778224b76bfdea9b67062177d984a6918d73725 100644
--- a/app/api/main.py
+++ b/app/api/main.py
@@ -14,7 +14,7 @@ from flask import (current_app, Blueprint, jsonify, request)
from flask_jwt_extended import create_access_token, jwt_required
from flask_ldap3_login import AuthenticationResponseStatus
from ..extensions import ldap_manager, db
-from ..models import Item, Manufacturer, Model, Location, Status, Action
+from ..models import Item, Manufacturer, Model, Location, Status, Action, Network
from .. import utils
from ..decorators import jwt_groups_accepted
@@ -48,13 +48,14 @@ def get_generic_model(model, args):
return jsonify(data)
-def create_generic_model(model, mandatory_field='name'):
+def create_generic_model(model, mandatory_fields=('name',)):
data = request.get_json()
if data is None:
raise utils.CSEntryError('Body should be a JSON object')
current_app.logger.debug(f'Received: {data}')
- if mandatory_field not in data:
- raise utils.CSEntryError(f"Missing mandatory field '{mandatory_field}'", status_code=422)
+ for mandatory_field in mandatory_fields:
+ if mandatory_field not in data:
+ raise utils.CSEntryError(f"Missing mandatory field '{mandatory_field}'", status_code=422)
try:
instance = model(**data)
except TypeError as e:
@@ -119,7 +120,7 @@ def create_item():
# an item so ics_id should also be a mandatory field.
# But there are existing items (in confluence and JIRA) that we want to
# import and associate after they have been created.
- return create_generic_model(Item, mandatory_field='serial_number')
+ return create_generic_model(Item, mandatory_fields=('serial_number',))
@bp.route('/items/<id_>', methods=['PATCH'])
@@ -234,3 +235,21 @@ def get_status():
@jwt_groups_accepted('admin', 'create')
def create_status():
return create_generic_model(Status)
+
+
+@bp.route('/networks')
+@jwt_required
+def get_networks():
+ # TODO: add pagination
+ query = utils.get_query(Network.query, request.args)
+ networks = query.order_by(Network.id)
+ data = [network.to_dict() for network in networks]
+ return jsonify(data)
+
+
+@bp.route('/networks', methods=['POST'])
+@jwt_required
+@jwt_groups_accepted('admin')
+def create_network():
+ """Create a new network"""
+ return create_generic_model(Network, mandatory_fields=('prefix', 'first', 'last'))
diff --git a/tests/functional/test_api.py b/tests/functional/test_api.py
index 11dd1984bfab7625f151956d1eecc345b4f8d227..4b0b7375c8419747c19c0f290b9ebebee0d3b264 100644
--- a/tests/functional/test_api.py
+++ b/tests/functional/test_api.py
@@ -21,9 +21,10 @@ ENDPOINT_MODEL = {
'locations': models.Location,
'status': models.Status,
'items': models.Item,
+ 'networks': models.Network,
}
-GENERIC_GET_ENDPOINTS = [key for key in ENDPOINT_MODEL.keys() if key != 'items']
-GENERIC_CREATE_ENDPOINTS = [key for key in ENDPOINT_MODEL.keys() if key not in ('items', 'actions')]
+GENERIC_GET_ENDPOINTS = [key for key in ENDPOINT_MODEL.keys() if key not in ('items', 'networks')]
+GENERIC_CREATE_ENDPOINTS = [key for key in ENDPOINT_MODEL.keys() if key not in ('items', 'actions', 'networks')]
CREATE_AUTH_ENDPOINTS = [key for key in ENDPOINT_MODEL.keys() if key != 'actions']
@@ -106,10 +107,10 @@ def check_names(response, names):
assert set(names) == response_names
-def check_items(response, inputs):
+def check_input_is_subset_of_response(response, inputs):
# Sort the response by id to match the inputs order
- response_items = sorted(response.json, key=lambda d: d['id'])
- for d1, d2 in zip(inputs, response_items):
+ response_elts = sorted(response.json, key=lambda d: d['id'])
+ for d1, d2 in zip(inputs, response_elts):
assert set(d1.items()).issubset(set(d2.items()))
@@ -211,7 +212,7 @@ def test_create_item(client, user_token):
# check all items that were created
assert models.Item.query.count() == 3
response = get(client, '/api/items', user_token)
- check_items(response, (data, data, data2))
+ check_input_is_subset_of_response(response, (data, data, data2))
def test_create_item_invalid_ics_id(client, user_token):
@@ -410,20 +411,121 @@ def test_get_items(client, session, readonly_token):
response = get(client, '/api/items', token=readonly_token)
assert response.status_code == 200
assert len(response.json) == 3
- check_items(response, (item1.to_dict(), item2.to_dict(), item3.to_dict()))
+ check_input_is_subset_of_response(response, (item1.to_dict(), item2.to_dict(), item3.to_dict()))
# test filtering
response = get(client, '/api/items?serial_number=234567', token=readonly_token)
assert response.status_code == 200
assert len(response.json) == 1
- check_items(response, (item2.to_dict(),))
+ check_input_is_subset_of_response(response, (item2.to_dict(),))
# filtering on location_id works but not location (might want to change that)
response = get(client, f'/api/items?location_id={item1.location_id}', token=readonly_token)
assert response.status_code == 200
assert len(response.json) == 1
- check_items(response, (item1.to_dict(),))
+ check_input_is_subset_of_response(response, (item1.to_dict(),))
response = get(client, '/api/items?location=ESS', token=readonly_token)
check_response_message(response, 'Invalid query arguments', 422)
# using an unknown key raises a 422
response = get(client, '/api/items?foo=bar', token=readonly_token)
check_response_message(response, 'Invalid query arguments', 422)
+
+
+def test_get_networks(client, session, readonly_token):
+ # Create some networks
+ location = models.Location(name='G02')
+ session.add(location)
+ session.flush()
+ network1 = models.Network(prefix='172.16.1.0/24', first='172.16.1.1', last='172.16.1.254', label='network1')
+ network2 = models.Network(prefix='172.16.20.0/22', first='172.16.20.11', last='172.16.20.250')
+ network3 = models.Network(prefix='172.16.5.0/24', first='172.16.5.10', last='172.16.5.254', location_id=location.id)
+ for network in (network1, network2, network3):
+ session.add(network)
+ session.commit()
+
+ response = get(client, '/api/networks', token=readonly_token)
+ assert response.status_code == 200
+ assert len(response.json) == 3
+ check_input_is_subset_of_response(response, (network1.to_dict(), network2.to_dict(), network3.to_dict()))
+
+ # test filtering by location_id
+ response = get(client, f'/api/networks?location_id={location.id}', token=readonly_token)
+ assert response.status_code == 200
+ assert len(response.json) == 1
+ check_input_is_subset_of_response(response, (network3.to_dict(),))
+
+ # test filtering by prefix
+ response = get(client, '/api/networks?prefix=172.16.20.0/22', token=readonly_token)
+ assert response.status_code == 200
+ assert len(response.json) == 1
+ check_input_is_subset_of_response(response, (network2.to_dict(),))
+
+
+def test_create_network_auth_fail(client, session, user_token):
+ # admin is required to create networks
+ response = post(client, '/api/networks', data={}, token=user_token)
+ check_response_message(response, "User doesn't have the required group", 403)
+
+
+def test_create_network(client, session, admin_token):
+ location = models.Location(name='G02')
+ session.add(location)
+ session.commit()
+ # check that prefix, first and last are mandatory
+ response = post(client, '/api/networks', data={}, token=admin_token)
+ check_response_message(response, "Missing mandatory field 'prefix'", 422)
+ response = post(client, '/api/networks', data={'first': '172.16.1.10', 'last': '172.16.1.250'}, token=admin_token)
+ check_response_message(response, "Missing mandatory field 'prefix'", 422)
+ response = post(client, '/api/networks', data={'prefix': '172.16.1.0/24'}, token=admin_token)
+ check_response_message(response, "Missing mandatory field 'first'", 422)
+ response = post(client, '/api/networks', data={'prefix': '172.16.1.0/24', 'first': '172.16.1.10'}, token=admin_token)
+ check_response_message(response, "Missing mandatory field 'last'", 422)
+
+ data = {'prefix': '172.16.1.0/24',
+ 'first': '172.16.1.10',
+ 'last': '172.16.1.250'}
+ response = post(client, '/api/networks', data=data, token=admin_token)
+ assert response.status_code == 201
+ assert {'id', 'prefix', 'first', 'last', 'label', 'vlanid', 'gateway', 'location'} == set(response.json.keys())
+ assert response.json['prefix'] == '172.16.1.0/24'
+ assert response.json['first'] == '172.16.1.10'
+ assert response.json['last'] == '172.16.1.250'
+
+ # Check that prefix shall be unique
+ response = post(client, '/api/networks', data=data, token=admin_token)
+ check_response_message(response, 'IntegrityError', 409)
+
+ # Check that all parameters can be passed
+ data2 = {'prefix': '172.16.5.0/24',
+ 'first': '172.16.5.11',
+ 'last': '172.16.5.250',
+ 'label': 'G02',
+ 'gateway': '172.16.5.10',
+ 'vlanid': 1601,
+ 'location_id': location.id}
+ response = post(client, '/api/networks', data=data2, token=admin_token)
+ assert response.status_code == 201
+ assert response.json['location'] == location.name
+
+ # check all items that were created
+ assert models.Network.query.count() == 2
+
+
+def test_create_network_constraint_fail(client, session, admin_token):
+ # first not in prefix
+ data = {'prefix': '172.16.1.0/24',
+ 'first': '172.16.2.10',
+ 'last': '172.16.1.250'}
+ response = post(client, '/api/networks', data=data, token=admin_token)
+ check_response_message(response, 'IntegrityError', 409)
+ # last not in prefix
+ data = {'prefix': '172.16.1.0/24',
+ 'first': '172.16.1.10',
+ 'last': '172.16.5.250'}
+ response = post(client, '/api/networks', data=data, token=admin_token)
+ check_response_message(response, 'IntegrityError', 409)
+ # first > last
+ data = {'prefix': '172.16.1.0/24',
+ 'first': '172.16.1.10',
+ 'last': '172.16.1.9'}
+ response = post(client, '/api/networks', data=data, token=admin_token)
+ check_response_message(response, 'IntegrityError', 409)