Merge remote-tracking branch 'origin/master'

.gitignore

@@ -165,4 +165,5 @@ cython_debug/
 /servers/
 /users/
 unit_test.py
+logs.txt
 !*.py

Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.12-alpine3.20
+LABEL authors="hapso"
+
+WORKDIR /app
+COPY . /app
+RUN pip install --no-cache-dir -r requirements.txt
+EXPOSE 3000/tcp
+EXPOSE 3000/udp
+CMD ["python3", "app.py"]

app.py

@@ -4,6 +4,7 @@ import inspect
 from flask import Flask, Response, jsonify, request
 from flask_cors import CORS
 
+import firebase_manager
 import generic_executor
 
 app = Flask(__name__)
@@ -20,7 +21,7 @@ def generic_response_maker(status_code: http.HTTPStatus, _message: str = None) -
         case http.HTTPStatus.INTERNAL_SERVER_ERROR:
             message = jsonify({'message': 'Internal Server Error.'})
         case http.HTTPStatus.NO_CONTENT:
-            message = jsonify({'message': 'Deletion successful.'})
+            message = jsonify({'message': 'Deletion successful'})
         case http.HTTPStatus.ACCEPTED:
             message = jsonify({'message': 'Action successful.'})
         case http.HTTPStatus.BAD_REQUEST:
@@ -34,26 +35,45 @@ def generic_response_maker(status_code: http.HTTPStatus, _message: str = None) -
         case http.HTTPStatus.METHOD_NOT_ALLOWED:
             message = jsonify({'message': 'This API call does not exist.'})
         case _:
-            status_code = http.HTTPStatus.BAD_GATEWAY
-            message = jsonify({'message': 'Bad Gateway.'})
+            message = jsonify({'message': 'Could not process request.'})
     return message, status_code.value
 
 
-def parse_and_validate_request(parameters: [str]) -> list[str]:
+def authenticate_request(data: dict):
+    if 'token' not in data:
+        raise Exception("Missing 'token' in request body. The API doesn't support anonymous access anymore.")
+    else:
+        #valid, user_id = firebase_manager.verify_jwt_token(data['token']) //TODO : REMOVE THIS when the front is ready
+        valid, user_id = True, data['token']
+        if not valid:
+            raise Exception("Invalid JWT token.")
+        else:
+            user = firebase_manager.get_user_from_id(user_id)
+            if not user:
+                raise Exception("User not found.")
+            if not user.email_verified:
+                raise Exception("Your google account isn't verified yet.")
+    return user
+
+
+def parse_and_validate_request(parameters: [str]) -> list[str] or None:
     args = []
     data = request.get_json()
     if not data:
         raise Exception("Empty request body.")
+    user = authenticate_request(data)
+    data['user'] = user
     for name in parameters:
         if name not in data:
             raise Exception(f"Missing parameter {name}")
         value = data[name]
-        if isinstance(value, str):
         args.append(value)
     return args
 
 
 route_handlers = {
+    'SetSubdomain': generic_executor.set_subdomain,
+    'FetchServers': generic_executor.fetch_servers,
     'AccountCreate': generic_executor.account_create,
     'ServerCreate': generic_executor.server_create,
     'ServerDelete': generic_executor.server_delete,
@@ -73,14 +93,19 @@ def dynamic_route_handler(path):
     route_fn = route_handlers[path]
     parameters = []
     sig = inspect.signature(route_fn)
-
     for param in sig.parameters.values():
         parameters.append(param.name)
-
+    try :
         mapped_parameters = parse_and_validate_request(parameters)
-    status: http.HTTPStatus = route_fn(*mapped_parameters)
-    return generic_response_maker(status)
+        if mapped_parameters is None:
+            return generic_response_maker(http.HTTPStatus.BAD_REQUEST)
+        status, message = route_fn(*mapped_parameters)
+        if isinstance(message, list):
+            return jsonify(message), http.HTTPStatus.OK
+        return generic_response_maker(status, message if message else None)
+    except Exception as e:
+        return generic_response_maker(http.HTTPStatus.BAD_REQUEST, str(e))
 
 
 if __name__ == '__main__':
-    app.run(host='0.0.0.0', port=3005, debug=True)
+    app.run(host='0.0.0.0', port=3000, debug=True)

file_manager.py

@@ -1,6 +1,8 @@
+import logging
 import os
 import shutil
 import re
+import subprocess
 
 supported_versions = ["bukkit", "paper", "spigot"]
 
@@ -61,5 +63,51 @@ def update_server_property(file_path, property_name, new_value):
         file.writelines(content)
 
 
+def reload_nginx() -> bool:
+    try:
+        subprocess.run(['nginx', '-s', 'reload'], check=True)
+        return True
+    except subprocess.CalledProcessError as e:
+        return False
+
+
+def append_stream_config(port, subdomain, config_file_path='/etc/nginx/sites-available/servii.fr'):
+    os.makedirs(os.path.dirname(config_file_path), exist_ok=True)
+    with open(config_file_path, 'r') as file:
+        content = file.read()
+    new_content = f"""stream {{
+    server {{
+        listen 25565;
+        listen [::]:25565;
+        server_name {subdomain}.servii.fr;
+        proxy_pass 127.0.0.1:{port};
+    }}
+}}\n"""
+    final_content = content + new_content
+    with open(config_file_path, 'w') as file:
+        file.write(final_content)
+
+
+def remove_stream_config(port, subdomain, config_file_path='/etc/nginx/sites-available/servii.fr'):
+    pattern = (r'stream {{\s*server {{\s*listen 25565;\s*listen \[\(::\]\):'
+               r'25565;\s*server_name {}.servii\.fr;\s*proxy_pass 127\.0\.0\.1:{};\s*}}}\n'
+               .format(re.escape(subdomain), port))
+    with open(config_file_path, 'r') as file:
+        content = file.read()
+    cleaned_content = re.sub(pattern, '', content, flags=re.MULTILINE | re.DOTALL)
+    with open(config_file_path, 'w') as file:
+        file.write(cleaned_content)
+
+    print(f"Stream configuration removed from {config_file_path}")
+
+
+def log_error(error_type: str, error_message: str):
+    logging.basicConfig(filename='logs.txt', level=logging.INFO,
+                        format='%(asctime)s - %(levelname)s - %(message)s',
+                        datefmt='%Y-%m-%d %H:%M:%S')
+    logger = logging.getLogger(__name__)
+    logger.error(f'{error_type}: {error_message}')
+
+
 if __name__ == "__main__":
     pass

firebase_manager.py

@@ -0,0 +1,153 @@
+from datetime import datetime
+
+import firebase_admin
+import jwt
+from firebase_admin import auth, credentials, firestore
+from google.api_core.exceptions import Aborted, DataLoss, NotFound, OutOfRange, PermissionDenied, ResourceExhausted
+
+cred = credentials.Certificate('servii.json')
+app = firebase_admin.initialize_app(cred)
+firestore_database = firestore.client()
+
+'''
+TODO
+Write a function that launches upon app's startup, it does check in the firestore for any already running servers.
+Fetches the PID's.
+Stops all the current processes,
+Tell the database they have now stopped running.
+Also ensure the program can add an additional argument to avoid this checking for scalability.
+'''
+
+
+def get_user_from_id(user_id):
+    return auth.get_user(user_id)
+
+
+def verify_jwt_token(token):
+    try:
+        decoded_token = jwt.decode(token, options={"verify_signature": False})
+        user_id = decoded_token.get('sub')
+        return True, user_id
+    except jwt.ExpiredSignatureError:
+        return False, None
+    except jwt.InvalidTokenError:
+        return False, None
+
+
+def fetch_port() -> int or None:
+    servers_ref = firestore_database.collection("users")
+    query = servers_ref.order_by("port", direction="DESCENDING").limit(1)
+    highest_port_doc = next(query.stream(), None)
+    if highest_port_doc:
+        return highest_port_doc.get('port')
+    return None
+
+
+def user_field_exists(user_id: str, field: str) -> bool:
+    try:
+        doc = firestore_database.collection('users').document(user_id).get()
+        if doc.exists:
+            return True
+        return False
+    except NotFound:
+        return False
+
+
+def server_name_taken(user_id: str, server_name: str) -> bool:
+    servers = firestore_database.collection('users').document(user_id).collection('servers')
+    query = servers.where('name', '==', server_name)
+    for _ in query.stream():
+        return True
+    return False
+
+
+def get_user_field(user_id, field_name):
+    user_doc_ref = firestore_database.collection('users').document(user_id)
+    user_doc = user_doc_ref.get()
+    if user_doc.exists:
+        field_value = user_doc.to_dict().get(field_name)
+        return field_value
+    else:
+        return None
+
+
+def get_server_port(user_id: str) -> int or None:
+    try:
+        servers_ref = firestore_database.collection('users').document(user_id)
+        server_doc = servers_ref.get()
+        port = server_doc.get('port')
+        return port if port else None
+    except Exception:
+        return None
+
+
+def create_firestore(user_id: str, data: dict) -> bool:
+    doc_ref = firestore_database.collection('users').document(user_id)
+    try:
+        doc_ref.create(data)
+        return True
+    except (NotFound, PermissionDenied, Aborted, ResourceExhausted,
+            OutOfRange, DataLoss, TypeError, Exception, ValueError) as e:
+        log_exception_to_firestore(e, user_id, data)
+        return False
+
+
+def update_firestore(user_id: str, data: dict) -> bool:
+    doc_ref = firestore_database.collection('users').document(user_id)
+    try:
+        doc_ref.update(data)
+        return True
+    except (NotFound, PermissionDenied, Aborted, ResourceExhausted,
+            OutOfRange, DataLoss, TypeError, Exception, ValueError) as e:
+        log_exception_to_firestore(e, user_id, data)
+        return False
+
+
+def set_firestore(user_id: str, data: dict) -> bool:
+    doc_ref = firestore_database.collection('users').document(user_id)
+    try:
+        doc_ref.set(data)
+        return True
+    except (NotFound, PermissionDenied, Aborted, ResourceExhausted,
+            OutOfRange, DataLoss, TypeError, Exception, ValueError) as e:
+        log_exception_to_firestore(e, user_id, data)
+        return False
+
+
+def create_server(user_id: str, server_name: str, version: str, port: str, framework: str = "paper"):
+    port: int = int(port)
+    servers_ref = firestore_database.collection('users').document(user_id).collection('servers')
+    server_doc_ref = servers_ref.document(server_name)
+    server_doc_ref.set(
+        {'name': server_name, 'port': port, 'running': False, 'version': version, 'framework': framework})
+
+
+def delete_server(user_id: str, server_name: str):
+    user_ref = firestore_database.collection('users').document(user_id)
+    servers_ref = user_ref.collection('servers')
+    server_doc_ref = servers_ref.document(server_name)
+    server_doc_ref.delete()
+
+
+def delete_user(user_id: str):
+    user_ref = firestore_database.collection('users').document(user_id)
+    user_ref.delete()
+
+
+def log_exception_to_firestore(exception: Exception = None, user_id: str = None, data: dict = None):
+    new_id: str = datetime.now().strftime('%Y-%m-%d %H:%M:%S %Z%z')
+    log_entry = {
+        'exception_name': str(type(exception).__name__),
+        'exception': str(exception) if exception else 'No exception',
+        'user_id': str(user_id) if user_id else 'No user_id',
+        'data': str(data) if data else 'No data provided',
+    }
+    try:
+        firestore_database.collection('firebase.logs').document(new_id).create(log_entry)
+        print("Log entry added successfully.")
+    except Exception as e:
+        print(f"Failed to add log entry: {e}")
+
+
+if __name__ == "__main__":
+    pass

generic_executor.py

@@ -1,92 +1,190 @@
-from server_mc_manager import MinecraftServerManager
 from http import HTTPStatus
+
+from firebase_admin.auth import UserRecord
+
 import file_manager
+import firebase_manager
+from server_mc_manager import MinecraftServerManager
 
 mc_manager: MinecraftServerManager = MinecraftServerManager()
 
 
-def account_create(port: str) -> HTTPStatus:
+def set_subdomain(user: UserRecord, subdomain: str) -> tuple[HTTPStatus, str or None]:
+    user_id: str = user.uid
+    store = firebase_manager.firestore_database
+    _users = store.collection("users")
+    query = _users.where("subdomain", "==", subdomain)
+    for _ in query.stream():
+        return HTTPStatus.ALREADY_REPORTED, "Subdomain already associated."
     try:
-        file_manager.create_folder("users/" + port)
-        return HTTPStatus.CREATED
+        if firebase_manager.get_user_field(user_id, "subdomain"):
+            return HTTPStatus.OK, "You cannot change your subdomain."
+        exists: bool = firebase_manager.update_firestore(user_id, {'subdomain': subdomain})
+        if not exists:
+            account_create(user)
+            return set_subdomain(user, subdomain)
+        port = firebase_manager.get_server_port(user_id)
+        if port is None:
+            return HTTPStatus.NOT_FOUND, "Port not found on database"
+        try:
+            file_manager.append_stream_config(port, subdomain)
+            file_manager.reload_nginx()
         except Exception as e:
-        print(f"Error creating account folder: {e}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
+            firebase_manager.update_firestore(user_id, {'subdomain': None})
+            return HTTPStatus.INTERNAL_SERVER_ERROR, f"Nginx configuration failed. {str(e)}"
+        return HTTPStatus.OK, "Successfully associated subdomain."
+    except Exception as e:
+        return HTTPStatus.FORBIDDEN, str(e)
 
 
-def server_create(port: str, name: str, version: str) -> HTTPStatus:
-    server_path: str = f"users/{port}/{name}"
-    server_template_path: str = "servers/paper/" + version
+def fetch_servers(user: UserRecord) -> tuple[HTTPStatus, str or list]:
+    user_id: str = user.uid
+    server_ref = firebase_manager.firestore_database.collection('users').document(user_id).collection('servers')
     try:
+        servers = list(server_ref.stream())
+        servers_data = [doc.to_dict() for doc in servers]
+        if not servers_data:
+            account_create(user)
+        return HTTPStatus.OK, servers_data
+    except Exception:
+        return account_create(user)
+
+
+def account_create(user: UserRecord) -> tuple[HTTPStatus, str or None]:
+    if firebase_manager.user_field_exists(user.uid, 'port'):
+        return HTTPStatus.FORBIDDEN, "User already exists."
+    try:
+        file_manager.create_folder("users/" + user.uid)
+        firebase_manager.set_firestore(user.uid, {'mail': user.email,
+                                                  'name': user.display_name,
+                                                  'photo_url': user.photo_url,
+                                                  'subdomain': None,
+                                                  'port': firebase_manager.fetch_port() + 1
+                                                  })
+        return HTTPStatus.CREATED, "Successfully created account."
+    except Exception as e:
+        return HTTPStatus.INTERNAL_SERVER_ERROR, "Error creating account."
+
+
+allowed_frameworks: list[str] = ["paper", "spigot", "bukkit"]
+
+
+def server_create(user: UserRecord, name: str, version: str, framework: str = "paper") -> (
+        tuple)[HTTPStatus, str or None]:
+    if framework not in allowed_frameworks:
+        return HTTPStatus.METHOD_NOT_ALLOWED, f"Framework {framework} not recognized."
+    user_id = user.uid
+    server_path: str = f"users/{user_id}/{name}"
+    server_template_path: str = f"servers/{framework}/{version}"
+    try:
+        port = firebase_manager.get_server_port(user_id)
+        if port is None:
+            return HTTPStatus.NOT_FOUND, "Port not found on database"
+        subdomain = firebase_manager.firestore_database.collection("users").document(user_id).get().get("subdomain")
+        if subdomain is None:
+            firebase_manager.delete_server(user_id, name)
+            return HTTPStatus.NOT_FOUND, f"You haven't associated a subdomain yet."
+        if firebase_manager.server_name_taken(user_id, name):
+            return HTTPStatus.CONFLICT, f"Server name '{name}' already in use."
+        firebase_manager.create_server(user_id, name, version, port, framework)
         file_manager.create_folder(server_path)
         file_manager.copy_folder_contents(server_template_path, server_path)
         file_manager.copy_folder_contents("servers/shared", server_path)
         file_manager.update_server_property(server_path + "/server.properties", "server-port", port)
-        return HTTPStatus.CREATED
+        return HTTPStatus.CREATED, f"Successfully created server '{name}'."
     except Exception as e:
-        print(f"Error creating server: {e}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
+        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Server creation failed | {e}"
 
 
-def server_delete(port: str, name: str) -> HTTPStatus:
-    server_path: str = f"users/{port}/{name}"
+def server_delete(name: str, user: UserRecord) -> tuple[HTTPStatus, str or None]:
+    user_id = user.uid
+    server_path: str = f"users/{user_id}/{name}"
     try:
+        firebase_manager.delete_server(user_id, name)
         file_manager.delete_non_empty_folder(server_path)
-        return HTTPStatus.NO_CONTENT
-    except Exception as e:
-        print(f"Error deleting server: {e}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
+        return HTTPStatus.OK, f"Successfully deleted server {name}."
+    except Exception:
+        return HTTPStatus.INTERNAL_SERVER_ERROR, None
 
 
-def account_delete(port: str) -> HTTPStatus:
+def account_delete(user: UserRecord, subdomain: str) -> tuple[HTTPStatus, str or None]:
+    user_id = user.uid
     try:
-        file_manager.delete_non_empty_folder("users/" + port)
-        return HTTPStatus.NO_CONTENT
-    except Exception as e:
-        print(f"Error deleting account: {e}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
 
-
-def server_run(port: str, name: str) -> HTTPStatus:
+        file_manager.delete_non_empty_folder("users/" + user_id)
         try:
-        server_id = mc_manager.start_server(f"users/{port}/{name}")
+            port = firebase_manager.get_server_port(user_id)
+            if port is None:
+                return HTTPStatus.NOT_FOUND, "User not found on firestore."
+            firebase_manager.delete_user(user_id)
+        except Exception as e:
+            return HTTPStatus.NOT_FOUND, f"Database deletion failed | {e}"
+        try:
+            file_manager.remove_stream_config()
+        except Exception as e:
+            return HTTPStatus.INTERNAL_SERVER_ERROR, f"Nginx deletion failed. {str(e)}"
+        return HTTPStatus.OK, f"Successfully deleted user {user_id}."
+    except Exception as e:
+        file_manager.log_error(type(e).__name__, str(e))
+        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error deleting user '{user_id}' on the server."
+
+
+def server_run(user: UserRecord, name: str) -> tuple[HTTPStatus, str or None]:
+    user_id = user.uid
+    try:
+        port: int = firebase_manager.get_server_port(user_id)
+        server_id = mc_manager.start_server(f"users/{user_id}/{name}", port)
+        if server_id is None:
+            return HTTPStatus.OK, f"You cannot run multiples instances at this time."
         mc_manager.servers[server_id]['port'] = int(port)
-        return HTTPStatus.ACCEPTED
+        return HTTPStatus.ACCEPTED, f"Successfully started server {name}."
     except Exception as e:
-        print(f"Error starting server: {e}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
+        print(f"Error when running server: {e}")
+        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error when running server: {e}"
 
 
-def server_stop(port: str, name: str) -> HTTPStatus:
+def server_stop(user: UserRecord, name: str) -> tuple[HTTPStatus, str or None]:
+    port = None
     try:
-        server_id = mc_manager.get_server_id_by_port(int(port))
-        mc_manager.stop_server(server_id)
-        return HTTPStatus.ACCEPTED
+        port = firebase_manager.get_server_port(user.uid)
+        if port is None:
+            return HTTPStatus.NOT_FOUND, f"Server {name} not found in firestore."
+        exists: bool = mc_manager.stop_server(port)
+        if exists:
+            return HTTPStatus.OK, f"Successfully stopped server {name}."
+        return HTTPStatus.OK, f"Server {name} already stopped."
     except Exception as e:
-        print(f"Error stopping server: {e}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
+        if port:
+            mc_manager.stop_server_forcefully(port)
+        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error occurred when stopping server {name}."
 
 
-def update_property(port: str, name: str, prop: str, value: str) -> HTTPStatus:
-    property_file_path: str = f"users/{port}/{name}/server.properties"
+def update_property(user: UserRecord, name: str, prop: str, value: str) -> tuple[HTTPStatus, str or None]:
+    property_file_path: str = f"users/{user.uid}/{name}/server.properties"
     try:
         file_manager.update_server_property(property_file_path, prop, value)
-        return HTTPStatus.OK
+        return HTTPStatus.OK, f"Successfully set '{prop}' to '{value}'."
     except ValueError as e:
-        return HTTPStatus.BAD_REQUEST
+        file_manager.log_error(type(e).__name__, str(e))
+        return HTTPStatus.BAD_REQUEST, f"Property '{prop}' not found."
+    except FileNotFoundError:
+        return HTTPStatus.NOT_FOUND, f"Requested server '{name}' not found."
     except Exception as e:
         print(f"Unhandled error: {type(e).__name__}, {str(e)}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
+        return HTTPStatus.INTERNAL_SERVER_ERROR, None
 
 
-def run_command(port: str, command: str) -> HTTPStatus:
+def run_command(user: UserRecord, command: str, name: str) -> tuple[HTTPStatus, str or None]:
     try:
-        server_id = mc_manager.get_server_id_by_port(int(port))
-        if server_id is None:
-            return HTTPStatus.NOT_FOUND
-        mc_manager.execute_server_command(server_id, command)
-        return HTTPStatus.ACCEPTED
+        port = firebase_manager.get_server_port(user.uid)
+        if port is None:
+            return HTTPStatus.NOT_FOUND, f"Server {name} not found in firestore."
+        mc_manager.execute_server_command(port, command)
+        return HTTPStatus.OK, f"Command '{command}' executed successfully."
     except Exception as e:
         print(f"Error executing command: {e}")
-        return HTTPStatus.INTERNAL_SERVER_ERROR
+        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error executing command: {command} || {str(e)}"
 
+
+if __name__ == "__main__":
+    pass

requirements.txt

@@ -1,4 +1,6 @@
+firebase_admin==6.5.0
 Flask==3.0.3
 Flask_Cors==4.0.1
-plotly==5.22.0
-psutil==5.9.8
+google-api-core==2.19.0
+PyJWT==2.8.0
+gunicorn==20.1.0

server_mc_manager.py

@@ -4,48 +4,49 @@ import shlex
 
 class MinecraftServerManager:
     def __init__(self):
-        self.servers = {}
+        self.servers: dict = {}
+        self.servers_count: int = 0
+
+    def start_server(self, server_directory: str, port: int,
+                     java_executable='java', jar_file='server.jar', memory_size='2048M') -> int or None:
+
+        if port in self.servers:
+            return None
 
-    def start_server(self, server_directory, java_executable='java', jar_file='server.jar', memory_size='2048M'):
         command = f"{java_executable} -Xmx{memory_size} -Xms{memory_size} -jar {jar_file} > /dev/null"
         process = subprocess.Popen(shlex.split(command), cwd=server_directory, stdin=subprocess.PIPE)
 
-        server_id = len(self.servers) + 1
-        self.servers[server_id] = {
+        self.servers_count = len(self.servers) + 1
+        self.servers[port] = {
             'process': process,
             'directory': server_directory,
-            'port': None
+            'port': port
         }
+        return port
 
-        print(f"Started server {server_id} in directory {server_directory}")
-        return server_id
-
-    def execute_server_command(self, server_id, command):
-        if server_id in self.servers:
-            process = self.servers[server_id]['process']
+    def execute_server_command(self, port, command) -> bool:
+        if port not in self.servers:
+            return False
+        process = self.servers[port]['process']
         process.stdin.write(command.encode('utf-8') + b'\n')
         process.stdin.flush()
-            print(f"Server {server_id} executed command : {command}")
-        else:
-            print(f"No server found with ID {server_id}")
+        return True
 
-    def stop_server(self, server_id):
-        if server_id in self.servers:
-            process = self.servers[server_id]['process']
+    def stop_server(self, port: int) -> bool:
+        if port not in self.servers:
+            return False
+        process = self.servers[port]['process']
         process.communicate(input=b"stop\n")
-            del self.servers[server_id]
-            print(f"Stopped server {server_id}")
-        else:
-            print(f"No server found with ID {server_id}")
+        del self.servers[port]
+        return True
 
-    def stop_server_forcefully(self, server_id):
-        if server_id in self.servers:
+    def stop_server_forcefully(self, server_id) -> bool:
+        if server_id not in self.servers:
+            return False
         process = self.servers[server_id]['process']
         process.terminate()
         del self.servers[server_id]
-            print(f"Stopped server {server_id}")
-        else:
-            print(f"No server found with ID {server_id}")
+        return True
 
     def get_servers(self):
         return self.servers.values()

unit_test.py

@@ -0,0 +1,7 @@
+from firebase_admin import functions
+
+import file_manager
+import firebase_manager
+
+
+file_manager.append_stream_config(3005, "antho")