diff --git a/app/api/network.py b/app/api/network.py
index 24be880c364144ccf8bc2717677b64204d8bd13c..e5f689df5ac4538a42f889456670d8d943a6bf8c 100644
--- a/app/api/network.py
+++ b/app/api/network.py
@@ -197,3 +197,38 @@ def create_domain():
"""
return create_generic_model(models.Domain,
mandatory_fields=('name',))
+
+
+@bp.route('/cnames')
+@jwt_required
+def get_cnames():
+ """Return cnames
+
+ .. :quickref: Network; Get cnames
+ """
+ domain = request.args.get('domain', None)
+ if domain is not None:
+ query = models.Cname.query
+ query = query.join(models.Cname.interface).join(
+ models.Interface.network).join(
+ models.Network.domain).filter(
+ models.Domain.name == domain)
+ query = query.order_by(models.Cname.name)
+ return get_generic_model(model=None, query=query)
+ return get_generic_model(models.Cname,
+ order_by=models.Cname.name)
+
+
+@bp.route('/cnames', methods=['POST'])
+@jwt_required
+@jwt_groups_accepted('admin')
+def create_cname():
+ """Create a new cname
+
+ .. :quickref: Network; Create new cname
+
+ :jsonparam name: full cname
+ :jsonparam interface_id: primary key of the associated interface
+ """
+ return create_generic_model(models.Cname,
+ mandatory_fields=('name', 'interface_id'))
diff --git a/tests/functional/conftest.py b/tests/functional/conftest.py
index 75c6b356a6d5239d6784a6c967b15686d1a44025..d9166d2ef32146351f0f153e37550f296ca8f141 100644
--- a/tests/functional/conftest.py
+++ b/tests/functional/conftest.py
@@ -30,6 +30,7 @@ register(factories.InterfaceFactory)
register(factories.HostFactory)
register(factories.MacFactory)
register(factories.DomainFactory)
+register(factories.CnameFactory)
@pytest.fixture(scope='session')
diff --git a/tests/functional/factories.py b/tests/functional/factories.py
index d0554a6ea93d1a735dec5b385deae072ccff06b9..523103ed31db5bf8ae409d27f96f89556d5a1c56 100644
--- a/tests/functional/factories.py
+++ b/tests/functional/factories.py
@@ -130,7 +130,10 @@ class NetworkFactory(factory.alchemy.SQLAlchemyModelFactory):
def first_ip(self):
net = ipaddress.ip_network(self.address)
hosts = list(net.hosts())
- return str(hosts[4])
+ try:
+ return str(hosts[4])
+ except IndexError:
+ return str(hosts[0])
@factory.lazy_attribute
def last_ip(self):
@@ -168,3 +171,14 @@ class MacFactory(factory.alchemy.SQLAlchemyModelFactory):
sqlalchemy_session_persistence = 'commit'
address = factory.Faker('mac_address')
+
+
+class CnameFactory(factory.alchemy.SQLAlchemyModelFactory):
+ class Meta:
+ model = models.Cname
+ sqlalchemy_session = common.Session
+ sqlalchemy_session_persistence = 'commit'
+
+ name = factory.Sequence(lambda n: f'host{n}.example.org')
+ interface = factory.SubFactory(InterfaceFactory)
+ user = factory.SubFactory(UserFactory)
diff --git a/tests/functional/test_api.py b/tests/functional/test_api.py
index c6ce317f621ef81696366a913b103b8b04b2716e..859f115de91139828aa89a4cbbed69fc210625f6 100644
--- a/tests/functional/test_api.py
+++ b/tests/functional/test_api.py
@@ -28,6 +28,7 @@ ENDPOINT_MODEL = {
'network/hosts': models.Host,
'network/macs': models.Mac,
'network/domains': models.Domain,
+ 'network/cnames': models.Cname,
}
GENERIC_GET_ENDPOINTS = [key for key in ENDPOINT_MODEL.keys()
if key.startswith('inventory') and key != 'inventory/items']
@@ -827,3 +828,62 @@ def test_create_domain(client, admin_token):
# Check that name shall be unique
response = post(client, f'{API_URL}/network/domains', data=data, token=admin_token)
check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422)
+
+
+def test_get_cnames(client, cname_factory, readonly_token):
+ # Create some cnames
+ cname1 = cname_factory()
+ cname2 = cname_factory()
+ response = get(client, f'{API_URL}/network/cnames', token=readonly_token)
+ assert response.status_code == 200
+ assert len(response.json) == 2
+ check_input_is_subset_of_response(response, (cname1.to_dict(), cname2.to_dict()))
+
+
+def test_get_cnames_by_domain(client, domain_factory, network_factory, interface_factory, cname_factory, readonly_token):
+ # Create some cnames
+ domain_a = domain_factory(name='a.esss.lu.se')
+ domain_b = domain_factory(name='b.esss.lu.se')
+ network_a = network_factory(domain=domain_a)
+ network_b = network_factory(domain=domain_b)
+ interface_a1 = interface_factory(network=network_a)
+ interface_a2 = interface_factory(network=network_a)
+ interface_b1 = interface_factory(network=network_b)
+ cname_a1 = cname_factory(interface=interface_a1)
+ cname_a2 = cname_factory(interface=interface_a1)
+ cname_a3 = cname_factory(interface=interface_a2)
+ cname_b1 = cname_factory(interface=interface_b1)
+ cname_b2 = cname_factory(interface=interface_b1)
+ response = get(client, f'{API_URL}/network/cnames', token=readonly_token)
+ assert response.status_code == 200
+ assert len(response.json) == 5
+ response = get(client, f'{API_URL}/network/cnames?domain=a.esss.lu.se', token=readonly_token)
+ assert response.status_code == 200
+ assert len(response.json) == 3
+ check_input_is_subset_of_response(response, (cname_a1.to_dict(), cname_a2.to_dict(), cname_a3.to_dict()))
+ response = get(client, f'{API_URL}/network/cnames?domain=b.esss.lu.se', token=readonly_token)
+ assert response.status_code == 200
+ assert len(response.json) == 2
+ check_input_is_subset_of_response(response, (cname_b1.to_dict(), cname_b2.to_dict()))
+
+
+def test_create_cname(client, interface, admin_token):
+ # check that name and interface_id are mandatory
+ response = post(client, f'{API_URL}/network/cnames', data={}, token=admin_token)
+ check_response_message(response, "Missing mandatory field 'name'", 422)
+ response = post(client, f'{API_URL}/network/cnames', data={'name': 'myhost.example.org'}, token=admin_token)
+ check_response_message(response, "Missing mandatory field 'interface_id'", 422)
+ response = post(client, f'{API_URL}/network/cnames', data={'interface_id': interface.id}, token=admin_token)
+ check_response_message(response, "Missing mandatory field 'name'", 422)
+
+ data = {'name': 'myhost.tn.esss.lu.se',
+ 'interface_id': interface.id}
+ response = post(client, f'{API_URL}/network/cnames', data=data, token=admin_token)
+ assert response.status_code == 201
+ assert {'id', 'name', 'interface',
+ 'created_at', 'updated_at', 'user'} == set(response.json.keys())
+ assert response.json['name'] == data['name']
+
+ # Check that name shall be unique
+ response = post(client, f'{API_URL}/network/cnames', data=data, token=admin_token)
+ check_response_message(response, '(psycopg2.IntegrityError) duplicate key value violates unique constraint', 422)