(20250218) Will MikroTik quick rollback URL for easy testing

This commit is contained in:
2025-02-18 15:52:51 +05:30
parent 5074b3f040
commit a56269373a
4 changed files with 169 additions and 3 deletions
+167
View File
@@ -0,0 +1,167 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Tuesday, 18th Feb., 2025.
OBJECTIVE:
For simple MikroTik Rollbacks for testing.
NOT TO BE USED FOR LONG.
REFERENCES:
N/A
DOWNLOADS:
N/A
NOTES:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# For using Quart:
from quart import Blueprint, current_app, request, render_template
# My utils:
from utils_v2.string import json
from utils_v2.mikrotik.controllers.async_mikrotik import AsyncMikroTik
from utils_v2.api.codes import StatusCodes, HttpCodes
from utils_v2.api.response import ResponseModel
from utils_v2.api.async_quart import (
set_api_version,
read_input,
get_session_info,
log_request_to_mongo,
log_chain_to_mongo,
should_not_be_under_maintenance,
only_whitelisted_ips,
limit_rate,
validate_input,
handle_cancelled_request
)
# Common:
from shared import constants
# For asynchronous activities:
import asyncio
# For random choices:
import random
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# Related to Quart:
test_mikrotik_rollback_bp = Blueprint("test_mt_rb", __name__)
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
@test_mikrotik_rollback_bp.record_once
def init(blueprint_setup_state):
# This gets called when the blueprint is registered.
# Consider this to be a one-time setup for the whole blueprint:
pass
# ---------------------------------------------------------------------------------------------------------------------
@test_mikrotik_rollback_bp.route("/mikrotik/rollback", methods = ["GET", "POST", "PUT", "PATCH", "DELETE"])
@set_api_version(api_version = "1.0.0")
@read_input(sanitize_headers = False, sanitize_data = False)
@get_session_info(key = "X-Session-Token", session_coro = "get_session")
@log_request_to_mongo(
attr_name = "logs_mongo",
project = constants.PROJECT_NAME,
log_type = constants.MODULE_NAME,
operation = "testMikroTikRlBckApi",
log_input = True,
log_output = True,
sensitive_keys = ["sessionToken", "X-Session-Token"]
)
@log_chain_to_mongo(attr_name = "logs_mongo")
@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
@handle_cancelled_request()
async def callback_test(
inbound_headers: dict = None,
inbound_data: dict = None,
inbound_files: dict = None,
**kwargs
):
response = await current_app.mikrotik_hotspot_1000_controller.quick_roll_back(
mikrotik_client = AsyncMikroTik(
config_by = "easyfi-hs-1000",
mikrotik_ip = inbound_data.get("ip", "102.210.173.150"),
username = inbound_data.get("u", "easyfi"),
password = inbound_data.get("p", "easyfi"),
port = None,
use_https = False,
http_client = None,
action_log_conn = current_app.data_mongo
),
use_https = False
)
# Done here:
return ResponseModel(
status_code = StatusCodes.OK if response.success else StatusCodes.FAILED,
http_code = HttpCodes.SUCCESS if response.success else HttpCodes.INTERNAL_SERVER_ERROR,
message = response.message
)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass