diff --git a/tests/API-dhcp/app.py b/tests/API-dhcp/app.py new file mode 100644 index 0000000..89d3753 --- /dev/null +++ b/tests/API-dhcp/app.py @@ -0,0 +1,371 @@ +#!/usr/bin/env python + +from flask import Flask, jsonify, request +import ipaddress +import json + + +app = Flask(__name__) + + +def convert_to_cidr(address, mask): + """ + Convert an IP address and subnet mask to CIDR notation. + + Args: + address (str): The IP address in dotted decimal format (e.g., '192.168.1.1'). + mask (str): The subnet mask in dotted decimal format (e.g., '255.255.255.0'). + + Returns: + str: The CIDR notation of the network (e.g., '192.168.1.1/24'). + None: If there is an error in conversion, returns None and prints an error message. + + Raises: + ValueError: If the provided address or mask is invalid. + """ + # Convertir dirección y máscara a formato CIDR + try: + # Convertir la máscara de red de formato largo a formato corto + # Cada octeto se convierte a su representación binaria y se cuenta el número de bits '1' + cidr_mask = ipaddress.IPv4Network(f"0.0.0.0/{mask}").prefixlen + red_objeto = f"{address}/{cidr_mask}" + return red_objeto + except ValueError as e: + print(f"Error al convertir a CIDR: {e}") + return None + + + +subnets_collection = [{"id": 1,"subnet": "192.168.1.0/24", "next-server": "192.168.1.1", "boot-file-name": "pxelinux.0", "reservations": [] }] + + + +# Endpoint GET /ogdhcp/v1/status +""" +Endpoint to get the status of the DHCP service. + +This endpoint returns a JSON response with the current status of the DHCP service, +including disk usage, subnets configuration, and the status of various services. + +Returns: + Response: A JSON response with the following structure: + "total": str, # Total disk space + "used": str, # Used disk space + "available": str, # Available disk space + "percentage": str # Percentage of disk space used + "id": int, # Subnet ID + "subnet": str, # Subnet address + "pools": [ + "pool": str # IP address pool range + "reservations": [ + "ip-address": str, # Reserved IP address + "hw-address": str # Hardware address associated with the reservation + ] + ... + "kea-ctrl-agent": str, # Status of kea-ctrl-agent service + "kea-dhcp4": str, # Status of kea-dhcp4 service + "nginx": str # Status of nginx service + HTTP Status Code: + 200: If the request was successful. +""" +@app.route('/ogdhcp/v1/status', methods=['GET']) +def get_status(): + # Simular respuesta de éxito + return jsonify({ + "success": "Mensaje éxito", + "message": { + "disk_usage": { + "total": "20G", + "used": "15G", + "available": "5G", + "percentage": "75%" + }, + "subnets": [ + { + "id": 1, + "subnet": "192.168.1.0/24", + "pools": [{"pool": "192.168.1.10-192.168.1.100"}], + "reservations": [{"ip-address": "192.168.1.20", "hw-address": "00:0c:29:6b:5e:71"}] + }, + { + "id": 2, + "subnet": "10.0.0.0/24", + "pools": [{"pool": "10.0.0.10-10.0.0.100"}], + "reservations": [] + } + ], + "services_status": { + "kea-ctrl-agent": "active", + "kea-dhcp4": "active", + "nginx": "active" + } + } + }), 200 + +# Endpoint GET /ogdhcp/v1/subnets + +@app.route('/ogdhcp/v1/subnets', methods=['GET']) +def get_subnets(): + # Simular respuesta de éxito + return jsonify({ + "success": "Subredes obtenidas correctamente", + "message": subnets_collection + }), 200 + +# Endpoint POST /ogdhcp/v1/subnets +""" +Create a new subnet. +Endpoint: /ogdhcp/v1/subnets +Method: POST +Request Body (JSON): + { + "id": , # Unique identifier for the subnet + "address": , # IP address of the subnet + "mask": , # Subnet mask + "nextServer": , # (Optional) Next server IP address + "bootFileName": # (Optional) Boot file name + } +Responses: + 200 OK: + { + "message": + } + 400 Bad Request: + - "Invalid JSON" + - "Missing 'id', 'address' or 'mask' key" + - { + "error": "Error: La subred con el id '' ya existe." + } + - { + "error": "Error: La subred con la dirección '' ya existe." + } + - "Invalid data format" +Description: + This endpoint allows the creation of a new subnet. It expects a JSON payload with the subnet details. + The function checks if the provided JSON is valid and contains the required keys ('id', 'address', 'mask'). + It also ensures that the subnet ID and subnet address do not already exist in the collection. + If the validation passes, the new subnet is added to the collection and a success response is returned. +""" +@app.route('/ogdhcp/v1/subnets', methods=['POST']) +def create_subnet(): + try: + new_subnet = json.loads(request.data) + new_subnet["next-server"] = new_subnet.get("nextServer") + new_subnet["boot-file-name"] = new_subnet.get("bootFileName") + except json.JSONDecodeError: + return "Invalid JSON", 400 + + + + if isinstance(new_subnet, dict): + if "id" in new_subnet and "address" in new_subnet and "mask" in new_subnet: + new_subnet_cidr = convert_to_cidr(new_subnet["address"], new_subnet["mask"]) + # Comprobar si el id ya existe en la colección + for subnet in subnets_collection: + if subnet.get("id") == new_subnet["id"]: + return jsonify({ + "error": f"Error: La subred con el id '{new_subnet['id']}' ya existe." + }), 400 + if subnet.get("subnet") == new_subnet_cidr: + return jsonify({ + "error": f"Error: La subred con la dirección '{new_subnet_cidr}' ya existe." + }), 400 + + # Si el id no existe, continuar con la creación de la subred + new_subnet["subnet"] = convert_to_cidr(new_subnet["address"], new_subnet["mask"]) + subnets_collection.append(new_subnet) + return jsonify({ + "success": "Subred agregada correctamente", + "message": new_subnet + }), 200 + + else: + return "Missing 'id', 'address' or 'mask' key", 400 + else: + return "Invalid data format", 400 + + +# Endpoint DELETE /ogdhcp/v1/subnets/ +""" +Deletes a subnet from the subnets_collection based on the provided subnetId. + +Args: + subnetId (int): The ID of the subnet to be deleted. + +Returns: + Response: A JSON response indicating the success or failure of the deletion. + - If the subnet is successfully deleted, returns a JSON response with a success message and HTTP status 200. + - If the subnet with the given ID does not exist, returns a JSON response with an error message and HTTP status 404. +""" +@app.route('/ogdhcp/v1/subnets/', methods=['DELETE']) +def delete_subnet(subnetId): + subnet_to_delete = None + for subnet in subnets_collection: + if subnet["id"] == subnetId: + subnet_to_delete = subnet + break + if subnet_to_delete: + subnets_collection.remove(subnet_to_delete) + return jsonify({ + "success": "Subred eliminada correctamente" + }), 200 + else: + return jsonify({ + "error": f"Error: La subred con el id '{subnetId}' no existe" + }), 404 + +# Endpoint PUT /ogdhcp/v1/subnets/ +""" +Updates a subnet with the given subnetId based on the provided JSON data in the request. +Args: + subnetId (str): The ID of the subnet to be updated. +Returns: + Response: A JSON response indicating the success or failure of the update operation. + - On success: Returns a JSON response with a success message and the updated subnet data, with a status code of 200. + - On failure: Returns a JSON response with an error message and a status code of 400. +Raises: + json.JSONDecodeError: If the request data is not valid JSON. +The function performs the following steps: + 1. Parses the JSON data from the request. + 2. Validates the presence of required fields ('address' and 'mask') in the JSON data. + 3. Converts the 'address' and 'mask' fields to a 'subnet' field in CIDR notation. + 4. Searches for the subnet with the given subnetId in the subnets_collection. + 5. Updates the subnet fields ('subnet', 'nextServer', 'bootFileName') if they are present in the JSON data. + 6. Returns a success response if the subnet is found and updated. + 7. Returns an error response if the subnet is not found. +""" +@app.route('/ogdhcp/v1/subnets/', methods=['PUT']) +def update_subnet(subnetId): + + + if subnetId == 0: + return jsonify({ + "error": "Error al guardar la configuración en Kea DHCP: Unable to save configuration" + }), 400 + try: + modify_data = json.loads(request.data) + except json.JSONDecodeError: + return "Invalid JSON", 400 + + print ("Modify data", modify_data) + print ("Modify data type", type(modify_data)) + if modify_data.get("address") != None and modify_data.get("mask") == None: + print ("Address", modify_data.get("address")) + return jsonify({ + "error": f"Error: La máscara de red es requerida con el parametro 'address'" + }), 400 + if modify_data.get("mask") != None and modify_data.get("address") == None: + return jsonify({ + "error": f"Error: La dirección de red es requerida con el parametro 'mask'" + }), 400 + + subnet_to_update = None + + for subnet in subnets_collection: + # Casting subnet["id"] to int to avoid type mismatch + if str(subnet["id"]) == str(subnetId): + # subnet_to_update = subnet + if modify_data.get("subnet"): + subnet["subnet"] = modify_data["subnet"] + + if modify_data.get("nextServer"): + subnet["next-server"] = modify_data["nextServer"] + + if modify_data.get("bootFileName"): + subnet["boot-file-name"] = modify_data["bootFileName"] + subnet_to_update = subnet + break + + if subnet_to_update: + # subnet_to_update.update(modify_data) + print ("Subnet to update", subnet_to_update) + return jsonify({ + "success": "Subred modificada correctamente", + "message": subnet_to_update + }), 200 + + else: + # Si no se encuentra la subred, devolver un error + print (f"Subnet Collection {subnets_collection}") + response = jsonify({ + "error": f"Error: La subred con el id '{subnetId}' no existe" + }) + response.status_code = 404 + response.headers["Content-Type"] = "application/json" + return response + + +# Endpoint GET /ogdhcp/v1/subnets//hosts +@app.route('/ogdhcp/v1/subnets//hosts', methods=['GET']) +def get_hosts(subnetId): + # Simular respuesta de éxito + return jsonify({ + "success": "Hosts retrieved successfully", + "message": [ + {"ip-address": "192.168.1.10", "hw-address": "00:0c:29:6b:5e:71", "hostname": "host1"}, + {"ip-address": "192.168.1.20", "hw-address": "00:0c:29:6b:5e:72", "hostname": "host2"} + ] + }), 200 + +# Endpoint POST /ogdhcp/v1/subnets//hosts +@app.route('/ogdhcp/v1/subnets//hosts', methods=['POST']) +def create_host(subnetId): + # Simular respuesta de éxito + return jsonify({ + "success": "Host agregado correctamente", + "message": { + "id": 1, + "subnet": "192.168.1.0/24", + "next-server": "192.168.1.1", + "boot-file-name": "pxelinux.0", + "reservations": [ + {"hostname": "pc11", "hw-address": "56:6f:c7:4f:00:4f", "ip-address": "172.30.4.11"} + ] + } + }), 200 + +# Endpoint DELETE /ogdhcp/v1/subnets//hosts +@app.route('/ogdhcp/v1/subnets//hosts', methods=['DELETE']) +def delete_host(subnetId): + # Simular respuesta de éxito + return jsonify({ + "success": "Host eliminado correctamente", + "message": { + "id": 1, + "subnet": "192.168.1.0/24", + "next-server": "192.168.1.1", + "boot-file-name": "pxelinux.0", + "reservations": [ + {"hostname": "host2", "hw-address": "00:0c:29:6b:5e:72", "ip-address": "172.30.4.12"} + ] + } + }), 200 + +# Endpoint PUT /ogdhcp/v1/subnets//hosts +@app.route('/ogdhcp/v1/subnets//hosts', methods=['PUT']) +def update_host(subnetId): + # Simular respuesta de éxito + return jsonify({ + "success": "Host actualizado correctamente", + "message": { + "id": 1, + "subnet": "192.168.1.0/24", + "next-server": "192.168.1.1", + "boot-file-name": "pxelinux.0", + "reservations": [ + {"hostname": "pc11", "hw-address": "56:6f:c7:4f:01:01", "ip-address": "192.168.1.11"}, + {"hostname": "host2", "hw-address": "00:0c:29:6b:5e:72", "ip-address": "192.168.1.12"} + ] + } + }), 200 + +# Endpoint POST /ogdhcp/v1/backup +@app.route('/ogdhcp/v1/backup', methods=['POST']) +def backup_config(): + # Simular respuesta de éxito + return jsonify({ + "success": "Configuración cargada correctamente" + }), 200 + +if __name__ == '__main__': + app.run(debug=True, port=8006)