Skip to content
Snippets Groups Projects
Commit 1bf57261 authored by Benjamin Bertrand's avatar Benjamin Bertrand
Browse files

Add Cname and NetworkScope table

- refactor Network class
parent 56f8f5ad
No related branches found
No related tags found
No related merge requests found
...@@ -83,6 +83,11 @@ class ItemAdmin(AdminModelView): ...@@ -83,6 +83,11 @@ class ItemAdmin(AdminModelView):
class NetworkAdmin(AdminModelView): class NetworkAdmin(AdminModelView):
# Replace TextAreaField (default for Text) with StringField
form_overrides = {
'vlan_name': fields.StringField,
}
form_args = { form_args = {
'gateway': { 'gateway': {
'filters': [lambda x: x or None], 'filters': [lambda x: x or None],
......
...@@ -244,7 +244,7 @@ def create_status(): ...@@ -244,7 +244,7 @@ def create_status():
def get_networks(): def get_networks():
# TODO: add pagination # TODO: add pagination
query = utils.get_query(Network.query, request.args) query = utils.get_query(Network.query, request.args)
networks = query.order_by(Network.prefix) networks = query.order_by(Network.address)
data = [network.to_dict() for network in networks] data = [network.to_dict() for network in networks]
return jsonify(data) return jsonify(data)
...@@ -254,7 +254,8 @@ def get_networks(): ...@@ -254,7 +254,8 @@ def get_networks():
@jwt_groups_accepted('admin') @jwt_groups_accepted('admin')
def create_network(): def create_network():
"""Create a new network""" """Create a new network"""
return create_generic_model(Network, mandatory_fields=('name', 'prefix', 'first', 'last')) return create_generic_model(Network, mandatory_fields=(
'vlan_name', 'vlan_id', 'address', 'first_ip', 'last_ip', 'scope'))
@bp.route('/hosts') @bp.route('/hosts')
......
...@@ -101,9 +101,11 @@ def create_app(config=None): ...@@ -101,9 +101,11 @@ def create_app(config=None):
admin.add_view(AdminModelView(models.Location, db.session)) admin.add_view(AdminModelView(models.Location, db.session))
admin.add_view(AdminModelView(models.Status, db.session)) admin.add_view(AdminModelView(models.Status, db.session))
admin.add_view(ItemAdmin(models.Item, db.session)) admin.add_view(ItemAdmin(models.Item, db.session))
admin.add_view(AdminModelView(models.NetworkScope, db.session))
admin.add_view(NetworkAdmin(models.Network, db.session)) admin.add_view(NetworkAdmin(models.Network, db.session))
admin.add_view(HostAdmin(models.Host, db.session)) admin.add_view(HostAdmin(models.Host, db.session))
admin.add_view(AdminModelView(models.Mac, db.session)) admin.add_view(AdminModelView(models.Mac, db.session))
admin.add_view(AdminModelView(models.Cname, db.session))
app.register_blueprint(main) app.register_blueprint(main)
app.register_blueprint(users) app.register_blueprint(users)
......
...@@ -46,4 +46,4 @@ class HostForm(FlaskForm): ...@@ -46,4 +46,4 @@ class HostForm(FlaskForm):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
self.network_id.choices = [(str(network.id), network.prefix) for network in models.Network.query.all()] self.network_id.choices = [(str(network.id), network.address) for network in models.Network.query.all()]
...@@ -151,7 +151,7 @@ def hosts_index(): ...@@ -151,7 +151,7 @@ def hosts_index():
@bp.route('/_retrieve_hosts') @bp.route('/_retrieve_hosts')
@login_required @login_required
def retrieve_hosts(): def retrieve_hosts():
data = [(host.name, host.ip, host.network.prefix) for host in models.Host.query.all()] data = [(host.name, host.ip, host.network.vlan_name) for host in models.Host.query.all()]
return jsonify(data=data) return jsonify(data=data)
......
...@@ -177,7 +177,6 @@ class Model(QRCodeMixin, db.Model): ...@@ -177,7 +177,6 @@ class Model(QRCodeMixin, db.Model):
class Location(QRCodeMixin, db.Model): class Location(QRCodeMixin, db.Model):
items = db.relationship('Item', back_populates='location') items = db.relationship('Item', back_populates='location')
networks = db.relationship('Network', backref='location')
class Status(QRCodeMixin, db.Model): class Status(QRCodeMixin, db.Model):
...@@ -271,36 +270,45 @@ class Item(db.Model): ...@@ -271,36 +270,45 @@ class Item(db.Model):
class Network(db.Model): class Network(db.Model):
id = db.Column(db.Integer, primary_key=True) id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.Text, nullable=False, unique=True) vlan_name = db.Column(CIText, nullable=False, unique=True)
prefix = db.Column(postgresql.CIDR, nullable=False, unique=True) vlan_id = db.Column(db.Integer, nullable=False, unique=True)
first = db.Column(postgresql.INET, nullable=False, unique=True) address = db.Column(postgresql.CIDR, nullable=False, unique=True)
last = db.Column(postgresql.INET, nullable=False, unique=True) first_ip = db.Column(postgresql.INET, nullable=False, unique=True)
last_ip = db.Column(postgresql.INET, nullable=False, unique=True)
gateway = db.Column(postgresql.INET) gateway = db.Column(postgresql.INET)
vlanid = db.Column(db.Integer, unique=True) description = db.Column(db.Text)
location_id = db.Column(db.Integer, db.ForeignKey('location.id')) admin_only = db.Column(db.Boolean, nullable=False, default=False)
scope_id = db.Column(db.Integer, db.ForeignKey('network_scope.id'), nullable=False)
hosts = db.relationship('Host', backref='network') hosts = db.relationship('Host', backref='network')
__table_args__ = ( __table_args__ = (
sa.CheckConstraint('first < last', name='first_less_than_last'), sa.CheckConstraint('first_ip < last_ip', name='first_ip_less_than_last_ip'),
sa.CheckConstraint('first << prefix', name='first_in_prefix'), sa.CheckConstraint('first_ip << address', name='first_ip_in_network'),
sa.CheckConstraint('last << prefix', name='last_in_prefix'), sa.CheckConstraint('last_ip << address', name='last_ip_in_network'),
) )
def __init__(self, **kwargs):
# Automatically convert scope to an instance of NetworkScope if it was passed
# as a string
if 'scope' in kwargs:
kwargs['scope'] = utils.convert_to_model(kwargs['scope'], NetworkScope, 'name')
super().__init__(**kwargs)
def __str__(self): def __str__(self):
return str(self.prefix) return str(self.vlan_name)
@property @property
def network_ip(self): def network_ip(self):
return ipaddress.ip_network(self.prefix) return ipaddress.ip_network(self.address)
@property @property
def first_ip(self): def first(self):
return ipaddress.ip_address(self.first) return ipaddress.ip_address(self.first_ip)
@property @property
def last_ip(self): def last(self):
return ipaddress.ip_address(self.last) return ipaddress.ip_address(self.last_ip)
def ip_range(self): def ip_range(self):
"""Return the list of IP addresses that can be assigned for this network """Return the list of IP addresses that can be assigned for this network
...@@ -308,7 +316,7 @@ class Network(db.Model): ...@@ -308,7 +316,7 @@ class Network(db.Model):
The range is defined by the first and last IP The range is defined by the first and last IP
""" """
return [addr for addr in self.network_ip.hosts() return [addr for addr in self.network_ip.hosts()
if self.first_ip <= addr <= self.last_ip] if self.first <= addr <= self.last]
def used_ips(self): def used_ips(self):
"""Return the list of IP addresses in use """Return the list of IP addresses in use
...@@ -323,7 +331,7 @@ class Network(db.Model): ...@@ -323,7 +331,7 @@ class Network(db.Model):
if addr not in self.used_ips()] if addr not in self.used_ips()]
@staticmethod @staticmethod
def ip_in_network(ip, prefix): def ip_in_network(ip, address):
"""Ensure the IP is in the network """Ensure the IP is in the network
:param str user_id: unicode ID of a user :param str user_id: unicode ID of a user
...@@ -331,43 +339,45 @@ class Network(db.Model): ...@@ -331,43 +339,45 @@ class Network(db.Model):
:raises: ValidationError if the IP is not in the network :raises: ValidationError if the IP is not in the network
""" """
addr = ipaddress.ip_address(ip) addr = ipaddress.ip_address(ip)
net = ipaddress.ip_network(prefix) net = ipaddress.ip_network(address)
if addr not in net: if addr not in net:
raise ValidationError(f'IP address {ip} is not in network {prefix}') raise ValidationError(f'IP address {ip} is not in network {address}')
return (addr, net) return (addr, net)
@validates('first') @validates('first_ip')
def validate_first(self, key, ip): def validate_first_ip(self, key, ip):
"""Ensure the first IP is in the network""" """Ensure the first IP is in the network"""
self.ip_in_network(ip, self.prefix) self.ip_in_network(ip, self.address)
return ip return ip
@validates('last') @validates('last_ip')
def validate_last(self, key, ip): def validate_last_ip(self, key, ip):
"""Ensure the last IP is in the network""" """Ensure the last IP is in the network and greater than first_ip"""
addr, net = self.ip_in_network(ip, self.prefix) addr, net = self.ip_in_network(ip, self.address)
if addr < ipaddress.ip_address(self.first): if addr < self.first:
raise ValidationError(f'Last IP address {ip} is less than the first address {self.first}') raise ValidationError(f'Last IP address {ip} is less than the first address {self.first}')
return ip return ip
@validates('hosts') @validates('hosts')
def validate_hosts(self, key, host): def validate_hosts(self, key, host):
"""Ensure the host IP is in the network range""" """Ensure the host IP is in the network range"""
addr, net = self.ip_in_network(host.ip, self.prefix) addr, net = self.ip_in_network(host.ip, self.address)
if addr < ipaddress.ip_address(self.first) or addr > ipaddress.ip_address(self.last): if addr < self.first or addr > self.last:
raise ValidationError(f'IP address {host.ip} is not in range {self.first} - {self.last}') raise ValidationError(f'IP address {host.ip} is not in range {self.first} - {self.last}')
return host return host
def to_dict(self): def to_dict(self):
return { return {
'id': self.id, 'id': self.id,
'name': self.name, 'vlan_name': self.vlan_name,
'prefix': self.prefix, 'address': self.address,
'first': self.first, 'first_ip': self.first_ip,
'last': self.last, 'last_ip': self.last_ip,
'gateway': self.gateway, 'gateway': self.gateway,
'vlanid': self.vlanid, 'vlan_id': self.vlan_id,
'location': utils.format_field(self.location), 'description': self.description,
'admin_only': self.admin_only,
'scope': utils.format_field(self.scope),
} }
...@@ -379,12 +389,13 @@ class Host(db.Model): ...@@ -379,12 +389,13 @@ class Host(db.Model):
# This is a One To One relationship (set uselist to False) # This is a One To One relationship (set uselist to False)
mac = db.relationship('Mac', backref='host', uselist=False) mac = db.relationship('Mac', backref='host', uselist=False)
cnames = db.relationship('Cname', backref='host')
def __init__(self, **kwargs): def __init__(self, **kwargs):
# Automatically convert network to an instance of Network if it was passed # Automatically convert network to an instance of Network if it was passed
# as a prefix string # as an address string
if 'network' in kwargs: if 'network' in kwargs:
kwargs['network'] = utils.convert_to_model(kwargs['network'], Network, 'prefix') kwargs['network'] = utils.convert_to_model(kwargs['network'], Network, 'address')
# WARNING! Setting self.network will call validates_hosts in the Network class # WARNING! Setting self.network will call validates_hosts in the Network class
# For the validation to work, self.ip must be set before! # For the validation to work, self.ip must be set before!
# Ensure that ip is passed before network # Ensure that ip is passed before network
...@@ -442,6 +453,45 @@ class Mac(db.Model): ...@@ -442,6 +453,45 @@ class Mac(db.Model):
return d return d
class Cname(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.Text, nullable=False, unique=True)
host_id = db.Column(db.Integer, db.ForeignKey('host.id'), nullable=False, unique=True)
def __str__(self):
return str(self.name)
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'host_id': self.host_id,
}
class NetworkScope(db.Model):
__tablename__ = 'network_scope'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(CIText, nullable=False, unique=True)
first_vlan = db.Column(db.Integer, nullable=False, unique=True)
last_vlan = db.Column(db.Integer, nullable=False, unique=True)
subnet = db.Column(postgresql.CIDR, nullable=False, unique=True)
networks = db.relationship('Network', backref='scope')
def __str__(self):
return str(self.name)
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'first_vlan': self.first_vlan,
'last_vlan': self.last_vlan,
'subnet': self.subnet,
}
# call configure_mappers after defining all the models # call configure_mappers after defining all the models
# required by sqlalchemy_continuum # required by sqlalchemy_continuum
sa.orm.configure_mappers() sa.orm.configure_mappers()
......
...@@ -23,6 +23,7 @@ register(factories.ModelFactory) ...@@ -23,6 +23,7 @@ register(factories.ModelFactory)
register(factories.LocationFactory) register(factories.LocationFactory)
register(factories.StatusFactory) register(factories.StatusFactory)
register(factories.ItemFactory) register(factories.ItemFactory)
register(factories.NetworkScopeFactory)
register(factories.NetworkFactory) register(factories.NetworkFactory)
register(factories.HostFactory) register(factories.HostFactory)
......
...@@ -78,31 +78,44 @@ class ItemFactory(factory.alchemy.SQLAlchemyModelFactory): ...@@ -78,31 +78,44 @@ class ItemFactory(factory.alchemy.SQLAlchemyModelFactory):
status = factory.SubFactory(StatusFactory) status = factory.SubFactory(StatusFactory)
class NetworkScopeFactory(factory.alchemy.SQLAlchemyModelFactory):
class Meta:
model = models.NetworkScope
sqlalchemy_session = common.Session
sqlalchemy_session_persistence = 'commit'
name = factory.Sequence(lambda n: f'scope{n}')
first_vlan = factory.Sequence(lambda n: 1600 + 10 * n)
last_vlan = factory.Sequence(lambda n: 1609 + 10 * n)
subnet = factory.Faker('ipv4', network=True)
class NetworkFactory(factory.alchemy.SQLAlchemyModelFactory): class NetworkFactory(factory.alchemy.SQLAlchemyModelFactory):
class Meta: class Meta:
model = models.Network model = models.Network
sqlalchemy_session = common.Session sqlalchemy_session = common.Session
sqlalchemy_session_persistence = 'commit' sqlalchemy_session_persistence = 'commit'
name = factory.Sequence(lambda n: f'network{n}') vlan_name = factory.Sequence(lambda n: f'vlan{n}')
prefix = factory.Faker('ipv4', network=True) address = factory.Faker('ipv4', network=True)
vlanid = factory.Sequence(lambda n: 1600 + n) vlan_id = factory.Sequence(lambda n: 1600 + n)
scope = factory.SubFactory(NetworkScopeFactory)
@factory.lazy_attribute @factory.lazy_attribute
def first(self): def first_ip(self):
net = ipaddress.ip_network(self.prefix) net = ipaddress.ip_network(self.address)
hosts = list(net.hosts()) hosts = list(net.hosts())
return str(hosts[4]) return str(hosts[4])
@factory.lazy_attribute @factory.lazy_attribute
def last(self): def last_ip(self):
net = ipaddress.ip_network(self.prefix) net = ipaddress.ip_network(self.address)
hosts = list(net.hosts()) hosts = list(net.hosts())
return str(hosts[-5]) return str(hosts[-5])
@factory.lazy_attribute @factory.lazy_attribute
def gateway(self): def gateway(self):
net = ipaddress.ip_network(self.prefix) net = ipaddress.ip_network(self.address)
hosts = list(net.hosts()) hosts = list(net.hosts())
return str(hosts[-1]) return str(hosts[-1])
......
...@@ -431,26 +431,19 @@ def test_get_items(client, session, readonly_token): ...@@ -431,26 +431,19 @@ def test_get_items(client, session, readonly_token):
check_response_message(response, 'Invalid query arguments', 422) check_response_message(response, 'Invalid query arguments', 422)
def test_get_networks(client, location_factory, network_factory, readonly_token): def test_get_networks(client, network_factory, readonly_token):
# Create some networks # Create some networks
location = location_factory(name='G02') network1 = network_factory(address='172.16.1.0/24', first_ip='172.16.1.1', last_ip='172.16.1.254')
network1 = network_factory(prefix='172.16.1.0/24', first='172.16.1.1', last='172.16.1.254') network2 = network_factory(address='172.16.20.0/22', first_ip='172.16.20.11', last_ip='172.16.20.250')
network2 = network_factory(prefix='172.16.20.0/22', first='172.16.20.11', last='172.16.20.250') network3 = network_factory(address='172.16.5.0/24', first_ip='172.16.5.10', last_ip='172.16.5.254')
network3 = network_factory(prefix='172.16.5.0/24', first='172.16.5.10', last='172.16.5.254', location=location)
response = get(client, '/api/networks', token=readonly_token) response = get(client, '/api/networks', token=readonly_token)
assert response.status_code == 200 assert response.status_code == 200
assert len(response.json) == 3 assert len(response.json) == 3
check_input_is_subset_of_response(response, (network1.to_dict(), network2.to_dict(), network3.to_dict())) check_input_is_subset_of_response(response, (network1.to_dict(), network2.to_dict(), network3.to_dict()))
# test filtering by location_id # test filtering by address
response = get(client, f'/api/networks?location_id={location.id}', token=readonly_token) response = get(client, '/api/networks?address=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, (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 response.status_code == 200
assert len(response.json) == 1 assert len(response.json) == 1
check_input_is_subset_of_response(response, (network2.to_dict(),)) check_input_is_subset_of_response(response, (network2.to_dict(),))
...@@ -462,125 +455,146 @@ def test_create_network_auth_fail(client, session, user_token): ...@@ -462,125 +455,146 @@ def test_create_network_auth_fail(client, session, user_token):
check_response_message(response, "User doesn't have the required group", 403) check_response_message(response, "User doesn't have the required group", 403)
def test_create_network(client, location_factory, admin_token): def test_create_network(client, admin_token, network_scope_factory):
location = location_factory(name='G02') scope = network_scope_factory(subnet='172.16.0.0/16')
# check that name, prefix, first and last are mandatory # check that vlan_name, vlan_id, address, first_ip, last_ip and scope are mandatory
response = post(client, '/api/networks', data={}, token=admin_token) response = post(client, '/api/networks', data={}, token=admin_token)
check_response_message(response, "Missing mandatory field 'name'", 422) check_response_message(response, "Missing mandatory field 'vlan_name'", 422)
response = post(client, '/api/networks', data={'first': '172.16.1.10', 'last': '172.16.1.250'}, token=admin_token) response = post(client, '/api/networks', data={'first_ip': '172.16.1.10', 'last_ip': '172.16.1.250'}, token=admin_token)
check_response_message(response, "Missing mandatory field 'name'", 422) check_response_message(response, "Missing mandatory field 'vlan_name'", 422)
response = post(client, '/api/networks', data={'prefix': '172.16.1.0/24'}, token=admin_token) response = post(client, '/api/networks', data={'address': '172.16.1.0/24'}, token=admin_token)
check_response_message(response, "Missing mandatory field 'name'", 422) check_response_message(response, "Missing mandatory field 'vlan_name'", 422)
response = post(client, '/api/networks', data={'name': 'network1'}, token=admin_token) response = post(client, '/api/networks', data={'vlan_name': 'network1'}, token=admin_token)
check_response_message(response, "Missing mandatory field 'prefix'", 422) check_response_message(response, "Missing mandatory field 'vlan_id'", 422)
response = post(client, '/api/networks', data={'name': 'network1', 'prefix': '172.16.1.0/24', 'first': '172.16.1.10'}, token=admin_token) response = post(client, '/api/networks', data={'vlan_name': 'network1', 'vlan_id': 1600}, token=admin_token)
check_response_message(response, "Missing mandatory field 'last'", 422) check_response_message(response, "Missing mandatory field 'address'", 422)
response = post(client, '/api/networks', data={'vlan_name': 'network1', 'vlan_id': 1600, 'address': '172.16.1.0/24', 'first_ip': '172.16.1.10'}, token=admin_token)
data = {'name': 'network1', check_response_message(response, "Missing mandatory field 'last_ip'", 422)
'prefix': '172.16.1.0/24',
'first': '172.16.1.10', data = {'vlan_name': 'network1',
'last': '172.16.1.250'} 'vlan_id': 1600,
'address': '172.16.1.0/24',
'first_ip': '172.16.1.10',
'last_ip': '172.16.1.250',
'scope': scope.name}
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
assert response.status_code == 201 assert response.status_code == 201
assert {'id', 'name', 'prefix', 'first', 'last', 'vlanid', 'gateway', 'location'} == set(response.json.keys()) assert {'id', 'vlan_name', 'vlan_id', 'address', 'first_ip',
assert response.json['name'] == 'network1' 'last_ip', 'gateway', 'description', 'is_admin', 'scope'} == set(response.json.keys())
assert response.json['prefix'] == '172.16.1.0/24' assert response.json['vlan_name'] == 'network1'
assert response.json['first'] == '172.16.1.10' assert response.json['vlan_id'] == 1600
assert response.json['last'] == '172.16.1.250' assert response.json['address'] == '172.16.1.0/24'
assert response.json['first_ip'] == '172.16.1.10'
# Check that prefix and name shall be unique assert response.json['last_ip'] == '172.16.1.250'
# Check that address and name shall be unique
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422) check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422)
data_same_prefix = data.copy() data_same_address = data.copy()
data_same_prefix['name'] = 'networkX' data_same_address['vlan_name'] = 'networkX'
response = post(client, '/api/networks', data=data_same_prefix, token=admin_token) response = post(client, '/api/networks', data=data_same_address, token=admin_token)
check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422) check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422)
data_same_name = {'name': 'network1', data_same_name = {'vlan_name': 'network1',
'prefix': '172.16.2.0/24', 'vlan_id': '1600',
'first': '172.16.2.10', 'address': '172.16.2.0/24',
'last': '172.16.2.250'} 'first_ip': '172.16.2.10',
'last_ip': '172.16.2.250',
'scope': scope.name}
response = post(client, '/api/networks', data=data_same_name, token=admin_token) response = post(client, '/api/networks', data=data_same_name, token=admin_token)
check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422) check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422)
# Check that all parameters can be passed # Check that all parameters can be passed
data2 = {'name': 'network2', data2 = {'vlan_name': 'network2',
'prefix': '172.16.5.0/24', 'vlan_id': '1601',
'first': '172.16.5.11', 'address': '172.16.5.0/24',
'last': '172.16.5.250', 'first_ip': '172.16.5.11',
'last_ip': '172.16.5.250',
'gateway': '172.16.5.10', 'gateway': '172.16.5.10',
'vlanid': 1601, 'description': 'long description',
'location_id': location.id} 'scope': scope.name}
response = post(client, '/api/networks', data=data2, token=admin_token) response = post(client, '/api/networks', data=data2, token=admin_token)
assert response.status_code == 201 assert response.status_code == 201
assert response.json['location'] == location.name assert response.json['description'] == 'long description'
# check all items that were created # check all items that were created
assert models.Network.query.count() == 2 assert models.Network.query.count() == 2
def test_create_network_invalid_prefix(client, session, admin_token): def test_create_network_invalid_address(client, session, admin_token, network_scope):
# invalid network address # invalid network address
data = {'name': 'network1', data = {'vlan_name': 'network1',
'prefix': 'foo', 'vlan_id': '1600',
'first': '172.16.1.10', 'address': 'foo',
'last': '172.16.1.250'} 'first_ip': '172.16.1.10',
'last_ip': '172.16.1.250',
'scope': network_scope.name}
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, "'foo' does not appear to be an IPv4 or IPv6 network", 422) check_response_message(response, "'foo' does not appear to be an IPv4 or IPv6 network", 422)
data['prefix'] = '172.16.1' data['address'] = '172.16.1'
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, "'172.16.1' does not appear to be an IPv4 or IPv6 network", 422) check_response_message(response, "'172.16.1' does not appear to be an IPv4 or IPv6 network", 422)
# prefix address has host bits set # address has host bits set
data['prefix'] = '172.16.1.1/24' data['address'] = '172.16.1.1/24'
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, '172.16.1.1/24 has host bits set', 422) check_response_message(response, '172.16.1.1/24 has host bits set', 422)
@pytest.mark.parametrize('address', ('', 'foo', '192.168')) @pytest.mark.parametrize('address', ('', 'foo', '192.168'))
def test_create_network_invalid_ip(address, client, session, admin_token): def test_create_network_invalid_ip(address, client, session, admin_token, network_scope):
# invalid first IP address # invalid first IP address
data = {'name': 'network1', data = {'vlan_name': 'network1',
'prefix': '192.168.0.0/24', 'vlan_id': '1600',
'first': address, 'address': '192.168.0.0/24',
'last': '192.168.0.250'} 'first_ip': address,
'last_ip': '192.168.0.250',
'scope': network_scope.name}
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, f"'{address}' does not appear to be an IPv4 or IPv6 address", 422) check_response_message(response, f"'{address}' does not appear to be an IPv4 or IPv6 address", 422)
# invalid last IP address # invalid last IP address
data = {'name': 'network1', data = {'vlan_name': 'network1',
'prefix': '192.168.0.0/24', 'vlan_id': '1600',
'first': '192.168.0.250', 'address': '192.168.0.0/24',
'last': address} 'first_ip': '192.168.0.250',
'last_ip': address,
'scope': network_scope.name}
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, f"'{address}' does not appear to be an IPv4 or IPv6 address", 422) check_response_message(response, f"'{address}' does not appear to be an IPv4 or IPv6 address", 422)
def test_create_network_invalid_range(client, session, admin_token): def test_create_network_invalid_range(client, session, admin_token, network_scope):
# first not in prefix # first_ip not in network address
data = {'name': 'network1', data = {'vlan_name': 'network1',
'prefix': '172.16.1.0/24', 'vlan_id': '1600',
'first': '172.16.2.10', 'address': '172.16.1.0/24',
'last': '172.16.1.250'} 'first_ip': '172.16.2.10',
'last_ip': '172.16.1.250',
'scope': network_scope.name}
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, 'IP address 172.16.2.10 is not in network 172.16.1.0/24', 422) check_response_message(response, 'IP address 172.16.2.10 is not in network 172.16.1.0/24', 422)
# last not in prefix # last_ip not in network address
data = {'name': 'network1', data = {'vlan_name': 'network1',
'prefix': '172.16.1.0/24', 'vlan_id': '1600',
'first': '172.16.1.10', 'address': '172.16.1.0/24',
'last': '172.16.5.250'} 'first_ip': '172.16.1.10',
'last_ip': '172.16.5.250',
'scope': network_scope.name}
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, 'IP address 172.16.5.250 is not in network 172.16.1.0/24', 422) check_response_message(response, 'IP address 172.16.5.250 is not in network 172.16.1.0/24', 422)
# first > last # first_ip > last_ip
data = {'name': 'network1', data = {'vlan_name': 'network1',
'prefix': '172.16.1.0/24', 'vlan_id': '1600',
'first': '172.16.1.10', 'address': '172.16.1.0/24',
'last': '172.16.1.9'} 'first_ip': '172.16.1.10',
'last_ip': '172.16.1.9',
'scope': network_scope.name}
response = post(client, '/api/networks', data=data, token=admin_token) response = post(client, '/api/networks', data=data, token=admin_token)
check_response_message(response, 'Last IP address 172.16.1.9 is less than the first address 172.16.1.10', 422) check_response_message(response, 'Last IP address 172.16.1.9 is less than the first address 172.16.1.10', 422)
def test_get_hosts(client, network_factory, host_factory, readonly_token): def test_get_hosts(client, network_factory, host_factory, readonly_token):
# Create some hosts # Create some hosts
network1 = network_factory(prefix='192.168.1.0/24', first='192.168.1.10', last='192.168.1.250') network1 = network_factory(address='192.168.1.0/24', first_ip='192.168.1.10', last_ip='192.168.1.250')
network2 = network_factory(prefix='192.168.2.0/24', first='192.168.2.10', last='192.168.2.250') network2 = network_factory(address='192.168.2.0/24', first_ip='192.168.2.10', last_ip='192.168.2.250')
host1 = host_factory(network=network1, ip='192.168.1.10') host1 = host_factory(network=network1, ip='192.168.1.10')
host2 = host_factory(network=network1, ip='192.168.1.11', name='hostname2') host2 = host_factory(network=network1, ip='192.168.1.11', name='hostname2')
host3 = host_factory(network=network2, ip='192.168.2.10') host3 = host_factory(network=network2, ip='192.168.2.10')
...@@ -598,16 +612,16 @@ def test_get_hosts(client, network_factory, host_factory, readonly_token): ...@@ -598,16 +612,16 @@ def test_get_hosts(client, network_factory, host_factory, readonly_token):
def test_create_host(client, network_factory, user_token): def test_create_host(client, network_factory, user_token):
network = network_factory(prefix='192.168.1.0/24', first='192.168.1.10', last='192.168.1.250') network = network_factory(address='192.168.1.0/24', first_ip='192.168.1.10', last_ip='192.168.1.250')
# check that network_id and ip are mandatory # check that network_id and ip are mandatory
response = post(client, '/api/hosts', data={}, token=user_token) response = post(client, '/api/hosts', data={}, token=user_token)
check_response_message(response, "Missing mandatory field 'network'", 422) check_response_message(response, "Missing mandatory field 'network'", 422)
response = post(client, '/api/hosts', data={'ip': '192.168.1.20'}, token=user_token) response = post(client, '/api/hosts', data={'ip': '192.168.1.20'}, token=user_token)
check_response_message(response, "Missing mandatory field 'network'", 422) check_response_message(response, "Missing mandatory field 'network'", 422)
response = post(client, '/api/hosts', data={'network': network.prefix}, token=user_token) response = post(client, '/api/hosts', data={'network': network.address}, token=user_token)
check_response_message(response, "Missing mandatory field 'ip'", 422) check_response_message(response, "Missing mandatory field 'ip'", 422)
data = {'network': network.prefix, data = {'network': network.address,
'ip': '192.168.1.20'} 'ip': '192.168.1.20'}
response = post(client, '/api/hosts', data=data, token=user_token) response = post(client, '/api/hosts', data=data, token=user_token)
assert response.status_code == 201 assert response.status_code == 201
...@@ -620,7 +634,7 @@ def test_create_host(client, network_factory, user_token): ...@@ -620,7 +634,7 @@ def test_create_host(client, network_factory, user_token):
check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422) check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422)
# Check that all parameters can be passed # Check that all parameters can be passed
data2 = {'network': network.prefix, data2 = {'network': network.address,
'ip': '192.168.1.21', 'ip': '192.168.1.21',
'name': 'myhostname'} 'name': 'myhostname'}
response = post(client, '/api/hosts', data=data2, token=user_token) response = post(client, '/api/hosts', data=data2, token=user_token)
...@@ -633,18 +647,18 @@ def test_create_host(client, network_factory, user_token): ...@@ -633,18 +647,18 @@ def test_create_host(client, network_factory, user_token):
@pytest.mark.parametrize('ip', ('', 'foo', '192.168')) @pytest.mark.parametrize('ip', ('', 'foo', '192.168'))
def test_create_host_invalid_ip(ip, client, network_factory, user_token): def test_create_host_invalid_ip(ip, client, network_factory, user_token):
network = network_factory(prefix='192.168.1.0/24', first='192.168.1.10', last='192.168.1.250') network = network_factory(address='192.168.1.0/24', first_ip='192.168.1.10', last_ip='192.168.1.250')
# invalid IP address # invalid IP address
data = {'network': network.prefix, data = {'network': network.address,
'ip': ip} 'ip': ip}
response = post(client, '/api/hosts', data=data, token=user_token) response = post(client, '/api/hosts', data=data, token=user_token)
check_response_message(response, f"'{ip}' does not appear to be an IPv4 or IPv6 address", 422) check_response_message(response, f"'{ip}' does not appear to be an IPv4 or IPv6 address", 422)
def test_create_host_ip_not_in_network(client, network_factory, user_token): def test_create_host_ip_not_in_network(client, network_factory, user_token):
network = network_factory(prefix='192.168.1.0/24', first='192.168.1.10', last='192.168.1.250') network = network_factory(address='192.168.1.0/24', first_ip='192.168.1.10', last_ip='192.168.1.250')
# IP address not in range # IP address not in range
data = {'network': network.prefix, data = {'network': network.address,
'ip': '192.168.2.4'} 'ip': '192.168.2.4'}
response = post(client, '/api/hosts', data=data, token=user_token) response = post(client, '/api/hosts', data=data, token=user_token)
check_response_message(response, 'IP address 192.168.2.4 is not in network 192.168.1.0/24', 422) check_response_message(response, 'IP address 192.168.2.4 is not in network 192.168.1.0/24', 422)
...@@ -14,20 +14,20 @@ import ipaddress ...@@ -14,20 +14,20 @@ import ipaddress
def test_network_ip_properties(network_factory): def test_network_ip_properties(network_factory):
# Create some networks # Create some networks
network1 = network_factory(prefix='172.16.1.0/24', first='172.16.1.10', last='172.16.1.250') network1 = network_factory(address='172.16.1.0/24', first_ip='172.16.1.10', last_ip='172.16.1.250')
network2 = network_factory(prefix='172.16.20.0/26', first='172.16.20.11', last='172.16.20.14') network2 = network_factory(address='172.16.20.0/26', first_ip='172.16.20.11', last_ip='172.16.20.14')
assert network1.network_ip == ipaddress.ip_network('172.16.1.0/24') assert network1.network_ip == ipaddress.ip_network('172.16.1.0/24')
assert network1.first_ip == ipaddress.ip_address('172.16.1.10') assert network1.first == ipaddress.ip_address('172.16.1.10')
assert network1.last_ip == ipaddress.ip_address('172.16.1.250') assert network1.last == ipaddress.ip_address('172.16.1.250')
assert len(network1.ip_range()) == 241 assert len(network1.ip_range()) == 241
assert network1.ip_range() == [ipaddress.ip_address(f'172.16.1.{i}') for i in range(10, 251)] assert network1.ip_range() == [ipaddress.ip_address(f'172.16.1.{i}') for i in range(10, 251)]
assert network1.ip_range() == network1.available_ips() assert network1.ip_range() == network1.available_ips()
assert network1.used_ips() == [] assert network1.used_ips() == []
assert network2.network_ip == ipaddress.ip_network('172.16.20.0/26') assert network2.network_ip == ipaddress.ip_network('172.16.20.0/26')
assert network2.first_ip == ipaddress.ip_address('172.16.20.11') assert network2.first == ipaddress.ip_address('172.16.20.11')
assert network2.last_ip == ipaddress.ip_address('172.16.20.14') assert network2.last == ipaddress.ip_address('172.16.20.14')
assert len(network2.ip_range()) == 4 assert len(network2.ip_range()) == 4
assert network2.ip_range() == [ipaddress.ip_address(f'172.16.20.{i}') for i in range(11, 15)] assert network2.ip_range() == [ipaddress.ip_address(f'172.16.20.{i}') for i in range(11, 15)]
assert network2.ip_range() == network2.available_ips() assert network2.ip_range() == network2.available_ips()
...@@ -36,8 +36,8 @@ def test_network_ip_properties(network_factory): ...@@ -36,8 +36,8 @@ def test_network_ip_properties(network_factory):
def test_network_available_and_used_ips(network_factory, host_factory): def test_network_available_and_used_ips(network_factory, host_factory):
# Create some networks and hosts # Create some networks and hosts
network1 = network_factory(prefix='172.16.1.0/24', first='172.16.1.10', last='172.16.1.250') network1 = network_factory(address='172.16.1.0/24', first_ip='172.16.1.10', last_ip='172.16.1.250')
network2 = network_factory(prefix='172.16.20.0/26', first='172.16.20.11', last='172.16.20.14') network2 = network_factory(address='172.16.20.0/26', first_ip='172.16.20.11', last_ip='172.16.20.14')
for i in range(10, 20): for i in range(10, 20):
host_factory(network=network1, ip=f'172.16.1.{i}') host_factory(network=network1, ip=f'172.16.1.{i}')
host_factory(network=network2, ip='172.16.20.13') host_factory(network=network2, ip='172.16.20.13')
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment