Add mock api
parent
43de742eae
commit
12e215b44c
|
|
@ -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": <str>, # Unique identifier for the subnet
|
||||
"address": <str>, # IP address of the subnet
|
||||
"mask": <str>, # Subnet mask
|
||||
"nextServer": <str>, # (Optional) Next server IP address
|
||||
"bootFileName": <str> # (Optional) Boot file name
|
||||
}
|
||||
Responses:
|
||||
200 OK:
|
||||
{
|
||||
"message": <new_subnet>
|
||||
}
|
||||
400 Bad Request:
|
||||
- "Invalid JSON"
|
||||
- "Missing 'id', 'address' or 'mask' key"
|
||||
- {
|
||||
"error": "Error: La subred con el id '<id>' ya existe."
|
||||
}
|
||||
- {
|
||||
"error": "Error: La subred con la dirección '<subnet>' 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/<subnetId>
|
||||
"""
|
||||
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/<int:subnetId>', 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/<subnetId>
|
||||
"""
|
||||
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/<int:subnetId>', 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/<subnetId>/hosts
|
||||
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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/<subnetId>/hosts
|
||||
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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/<subnetId>/hosts
|
||||
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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/<subnetId>/hosts
|
||||
@app.route('/ogdhcp/v1/subnets/<int:subnetId>/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)
|
||||
Loading…
Reference in New Issue