diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..4f5f125 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,6 @@ +[run] +omit = + *__init__* + tests/* + application/debugger.py + application/alembic/* \ No newline at end of file diff --git a/.github/workflows/develop-branch-actions.yml b/.github/workflows/develop-branch-actions.yml index 0b311b7..1e4c8b4 100644 --- a/.github/workflows/develop-branch-actions.yml +++ b/.github/workflows/develop-branch-actions.yml @@ -45,6 +45,11 @@ jobs: exclude: "tests/*,doc/*,scripts/*" max-line-length: "100" + - name: pytest + run: | + coverage run -m pytest + coverage report --fail-under 40 + build_develop_windows_exe: runs-on: windows-latest diff --git a/.github/workflows/every-other-branch-actions.yml b/.github/workflows/every-other-branch-actions.yml index 3f9b0ab..c5f5155 100644 --- a/.github/workflows/every-other-branch-actions.yml +++ b/.github/workflows/every-other-branch-actions.yml @@ -49,6 +49,11 @@ jobs: exclude: "tests/*,doc/*,scripts/*" max-line-length: "100" + - name: pytest + run: | + coverage run -m pytest + coverage report --fail-under 40 + build_other_windows_exe: runs-on: windows-latest diff --git a/.github/workflows/main-branch-after-merge-actions.yml b/.github/workflows/main-branch-after-merge-actions.yml index 209d8ff..69611e4 100644 --- a/.github/workflows/main-branch-after-merge-actions.yml +++ b/.github/workflows/main-branch-after-merge-actions.yml @@ -43,6 +43,11 @@ jobs: exclude: "tests/*,doc/*,scripts/*" max-line-length: "100" + - name: pytest + run: | + coverage run -m pytest + coverage report --fail-under 40 + - name: Run PyInstaller run: | python package.py diff --git a/.github/workflows/main-branch-pr-actions.yml b/.github/workflows/main-branch-pr-actions.yml index f7b2be8..4a9ed62 100644 --- a/.github/workflows/main-branch-pr-actions.yml +++ b/.github/workflows/main-branch-pr-actions.yml @@ -43,6 +43,11 @@ jobs: exclude: "tests/*,doc/*,scripts/*" max-line-length: "100" + - name: pytest + run: | + coverage run -m pytest + coverage report --fail-under 40 + build_main_pr_windows_exe: runs-on: windows-latest diff --git a/.gitignore b/.gitignore index 0e37d97..c30de19 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Hidden folder/file types. .coverage +htmlcov/ venv/* .env *__pycache__* diff --git a/application/api/v1/blueprints/game.py b/application/api/v1/blueprints/game.py index 8f70b84..f0ffce7 100644 --- a/application/api/v1/blueprints/game.py +++ b/application/api/v1/blueprints/game.py @@ -186,7 +186,7 @@ def _get_all(self): @authorization_required def get(self, game_name=None, game_arg_id=None, argument_name=None): page = request.args.get("page", 1, type=int) - per_page = min(request.args.get("per_page", 10, type=int), 10000) + per_page = min(request.args.get("per_page", 1000, type=int), 10000) if game_arg_id: qry = self._get_argument(game_arg_id) diff --git a/application/common/authorization.py b/application/common/authorization.py index ff49a5d..47db174 100644 --- a/application/common/authorization.py +++ b/application/common/authorization.py @@ -7,6 +7,18 @@ from application.models.tokens import Tokens +def _get_token(bearer_token: str) -> Tokens: + token_lookup = Tokens.query.filter_by( + token_active=True, token_value=bearer_token + ).first() + return token_lookup + + +def _get_setting(setting_name: str) -> Settings: + setting_lookup = Settings.query.filter_by(setting_name=setting_name).first() + return setting_lookup + + def _verify_bearer_token(request: Request) -> int: """This is the bearer token gauntlet. The requests only goal is to get through all of the checks. @@ -25,17 +37,13 @@ def _verify_bearer_token(request: Request) -> int: bearer_token = auth.split("Bearer")[-1].strip() # Make sure it's there first... - token_lookup = Tokens.query.filter_by( - token_active=True, token_value=bearer_token - ).first() + token_lookup = _get_token(bearer_token) if token_lookup is None: return 403 # Next decode this bad thing... - secret_obj = Settings.query.filter_by( - setting_name=constants.SETTING_NAME_APP_SECRET - ).first() + secret_obj = _get_setting(constants.SETTING_NAME_APP_SECRET) try: decoded_token = jwt.decode( diff --git a/application/common/game_base.py b/application/common/game_base.py index 1ac18c3..dead478 100644 --- a/application/common/game_base.py +++ b/application/common/game_base.py @@ -146,12 +146,15 @@ def _get_argument_list(self) -> []: def _get_argument_dict(self) -> []: return self._game_args - def _get_command_str(self) -> str: + def _get_command_str(self, args_only=False) -> str: arg_string = "" for _, arg in self._game_args.items(): arg_string += str(arg) + " " - return f"{self._game_executable} {arg_string}" + if args_only: + return arg_string + else: + return f"{self._game_executable} {arg_string}" def _rebuild_arguments_dict(self) -> None: game_qry = Games.query.filter_by(game_name=self._game_name) diff --git a/application/common/steam_manifest_parser.py b/application/common/steam_manifest_parser.py index 8067433..5871a77 100644 --- a/application/common/steam_manifest_parser.py +++ b/application/common/steam_manifest_parser.py @@ -6,7 +6,9 @@ def read_dir(dir: str): for file in os.listdir(dir): if file.endswith(".acf"): - acf = read_acf(dir + file) + # acf = read_dir(dir) + acf = read_acf(os.path.join(dir, file)) + # acf = read_acf(dir + file) steamapps[acf["appid"]] = acf return steamapps diff --git a/application/config/config.py b/application/config/config.py index 579b3c1..d50a986 100644 --- a/application/config/config.py +++ b/application/config/config.py @@ -24,7 +24,7 @@ class DefaultConfig: # NGINX Settings NGINX_DEFAULT_HOSTNAME = "localhost" - NGINX_DEFAULT_PORT = "5312" + NGINX_DEFAULT_PORT = "53128" NGINX_DEFAULT_ENABLED = True # Designate where the database file is stored based on platform. diff --git a/application/games/ark_game.py b/application/games/ark_game.py new file mode 100644 index 0000000..8b90e04 --- /dev/null +++ b/application/games/ark_game.py @@ -0,0 +1,177 @@ +import os +import time + +from jinja2 import Environment, FileSystemLoader + +from application.common import logger, constants +from application.common.game_argument import GameArgument +from application.common.game_base import BaseGame +from application.common.toolbox import _get_proc_by_name, get_resources_dir +from application.extensions import DATABASE +from application.models.games import Games + + +class ArkGame(BaseGame): + def __init__(self, defaults_dict: dict = {}) -> None: + super(ArkGame, self).__init__(defaults_dict) + + self._game_name = "ark" + self._game_pretty_name = "Ark: Survival Evolved" + self._game_executable = "ShooterGameServer.exe" + self._game_steam_id = "376030" + self._game_info_url = "https://ark.fandom.com/wiki/Dedicated_server_setup" + + """ + Reference: + + start ShooterGameServer.exe TheIsland?listen?SessionName= + ?ServerPassword= + ?ServerAdminPassword=?Port= + ?QueryPort=?MaxPlayers= + exit + """ + + # Add Args here, can update later. + self._add_argument( + GameArgument( + "server_name", + value="MyArkServer", + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "join_password", + value="abc123", + required=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "admin_password", + value="abc123", + required=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "port", + value=7777, + required=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "query_port", + value=27015, + required=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "max_players", + value=4, + required=True, + is_permanent=True, + ) + ) + + def startup(self) -> None: + # Run base class checks + super().startup() + + # Get individual arguments for this game + arguments = self._get_argument_dict() + + server_name = arguments["server_name"]._value + join_password = arguments["join_password"]._value + admin_password = arguments["admin_password"]._value + port = arguments["port"]._value + query_port = arguments["query_port"]._value + max_players = arguments["max_players"]._value + + # Create a formatted batch file. + env = Environment(loader=FileSystemLoader(get_resources_dir(__file__))) + template = env.get_template("start_ark_server_template.bat.j2") + output_from_parsed_template = template.render( + GAME_STEAM_ID=self._game_steam_id, + GAME_NAME=self._game_name, + server_name=server_name, + join_password=join_password, + admin_password=admin_password, + port=port, + query_port=query_port, + max_players=max_players, + ) + + # Print the formatted jinja + logger.debug(output_from_parsed_template) + + # In theory, the software has already check that the game is installed, so no check/guard + # needed. + game_qry = Games.query.filter_by(game_steam_id=self._game_steam_id) + game_obj = game_qry.first() + game_install_dir = game_obj.game_install_dir + + # Need game install location to write batch file. + full_path_startup_script = os.path.join( + game_install_dir, constants.STARTUP_BATCH_FILE_NAME + ) + + game_working_dir = os.path.join( + game_install_dir, "ShooterGame", "Binaries", "Win64" + ) + + # If file exists, remove it. + if os.path.exists(full_path_startup_script): + os.remove(full_path_startup_script) + + # Write the batch file. + with open(full_path_startup_script, "w") as myfile: + myfile.write(output_from_parsed_template) + + # Call the batch file on another process as to not block this one. + command = f'START /MIN CMD.EXE /C "{full_path_startup_script}"' + result = self._run_game(command, game_working_dir) + + time.sleep(1) + + process = _get_proc_by_name(self._game_executable) + + logger.info(result) + logger.info("Process:") + logger.info(process) + + update_dict = {"game_pid": int(process.pid)} + + game_qry.update(update_dict) + DATABASE.session.commit() + + def shutdown(self) -> None: + game_qry = Games.query.filter_by(game_steam_id=self._game_steam_id) + game_obj = game_qry.first() + game_pid = game_obj.game_pid + + process = _get_proc_by_name(self._game_executable) + + if process: + logger.info(process) + logger.info(game_pid) + + process.terminate() + process.wait() + + update_dict = {"game_pid": None} + game_qry.update(update_dict) + DATABASE.session.commit() diff --git a/application/games/resources/start_ark_server_template.bat.j2 b/application/games/resources/start_ark_server_template.bat.j2 new file mode 100644 index 0000000..b41d80d --- /dev/null +++ b/application/games/resources/start_ark_server_template.bat.j2 @@ -0,0 +1,6 @@ +@echo off +set SteamAppId=346110 +echo "Starting {{GAME_NAME}} Dedicated Server - PRESS CTRL-C to exit" + +start ShooterGameServer.exe TheIsland?listen?SessionName={{server_name}}?ServerPassword={{join_password}}?ServerAdminPassword={{admin_password}}?Port={{port}}?QueryPort={{query_port}}?MaxPlayers={{max_players}} -server -log +exit diff --git a/application/games/resources/start_satisfactory_server_template.bat.j2 b/application/games/resources/start_satisfactory_server_template.bat.j2 new file mode 100644 index 0000000..9ef235c --- /dev/null +++ b/application/games/resources/start_satisfactory_server_template.bat.j2 @@ -0,0 +1,6 @@ +@echo off +set SteamAppId=526870 +echo "Starting {{GAME_NAME}} Dedicated Server - PRESS CTRL-C to exit" + +@echo on +{{GAME_COMMAND}} diff --git a/application/games/resources/start_valheim_server_template.bat.j2 b/application/games/resources/start_valheim_server_template.bat.j2 new file mode 100644 index 0000000..5b88071 --- /dev/null +++ b/application/games/resources/start_valheim_server_template.bat.j2 @@ -0,0 +1,8 @@ +@echo off +set SteamAppId=892970 +echo "Starting {{GAME_NAME}} Dedicated Server - PRESS CTRL-C to exit"\ + +REM Tip: Make a local copy of this script to avoid it being overwritten by steam. +REM NOTE: Minimum password length is 5 characters & Password cant be in the server name. +REM NOTE: You need to make sure the ports 2456-2458 is being forwarded to your server through your local router & firewall. +valheim_server -nographics -batchmode {{GAME_ARGUMENTS}} \ No newline at end of file diff --git a/application/games/satisfactory_game.py b/application/games/satisfactory_game.py new file mode 100644 index 0000000..dd43016 --- /dev/null +++ b/application/games/satisfactory_game.py @@ -0,0 +1,177 @@ +import os +import time + +from jinja2 import Environment, FileSystemLoader + +from application.common import logger, constants +from application.common.game_argument import GameArgument +from application.common.game_base import BaseGame +from application.common.toolbox import _get_proc_by_name, get_resources_dir +from application.extensions import DATABASE +from application.models.games import Games + + +class Satisfactory(BaseGame): + def __init__(self, defaults_dict: dict = {}) -> None: + super(Satisfactory, self).__init__(defaults_dict) + + self._game_name = "satisfactory" + self._game_pretty_name = "Satisfactory" + self._game_executable = "FactoryServer.exe" + self._game_steam_id = "1690800" + self._game_info_url = "https://satisfactory.fandom.com/wiki/Dedicated_servers" + + # Add Args here, can update later. + # Default is 2456 + self._add_argument( + GameArgument( + "-multihome", + value="0.0.0.0", + required=True, + use_quotes=False, + use_equals=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-Port", + value=15002, + required=True, + use_quotes=False, + use_equals=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-ServerQueryPort", + value=15000, + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-BeaconPort", + value=15001, + required=True, + use_quotes=False, + use_equals=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-log", + value=" ", + required=False, + use_quotes=False, + is_permanent=False, + ) + ) + + self._add_argument( + GameArgument( + "-unattended", + value=" ", + required=False, + use_quotes=False, + is_permanent=False, + ) + ) + + self._add_argument( + GameArgument( + "-DisablePacketRouting", + value=" ", + required=False, + use_quotes=False, + is_permanent=False, + ) + ) + + def startup(self) -> None: + # Run base class checks + super().startup() + + # Format command string. + command = self._get_command_str(args_only=False) + + # Create a formatted batch file. + env = Environment(loader=FileSystemLoader(get_resources_dir(__file__))) + template = env.get_template("start_satisfactory_server_template.bat.j2") + output_from_parsed_template = template.render( + GAME_STEAM_ID=self._game_steam_id, + GAME_NAME=self._game_name, + GAME_COMMAND=command, + ) + + # Print the formatted jinja + logger.debug(output_from_parsed_template) + + # In theory, the software has already check that the game is installed, so no check/guard + # needed. + game_qry = Games.query.filter_by(game_steam_id=self._game_steam_id) + game_obj = game_qry.first() + game_install_dir = game_obj.game_install_dir + + # Need game install location to write batch file. + full_path_startup_script = os.path.join( + game_install_dir, constants.STARTUP_BATCH_FILE_NAME + ) + + # If file exists, remove it. + if os.path.exists(full_path_startup_script): + os.remove(full_path_startup_script) + + # Write the batch file. + with open(full_path_startup_script, "w") as myfile: + myfile.write(output_from_parsed_template) + + # Call the batch file on another process as to not block this one. + command = f'START /MIN CMD.EXE /C "{full_path_startup_script}"' + result = self._run_game(command, game_install_dir) + + time.sleep(1) + + process = _get_proc_by_name(self._game_executable) + + logger.info(result) + logger.info("Process:") + logger.info(process) + + update_dict = {"game_pid": int(process.pid)} + + game_qry.update(update_dict) + DATABASE.session.commit() + + def shutdown(self) -> None: + game_qry = Games.query.filter_by(game_steam_id=self._game_steam_id) + game_obj = game_qry.first() + game_pid = game_obj.game_pid + + process = _get_proc_by_name(self._game_executable) + + if process: + logger.info(process) + logger.info(game_pid) + + process.terminate() + process.wait() + + update_dict = {"game_pid": None} + game_qry.update(update_dict) + DATABASE.session.commit() + + process_2 = _get_proc_by_name("UnrealServer-Win64-Shipping.exe") + + if process: + logger.info(process_2) + process_2.terminate() + process_2.wait() diff --git a/application/games/valheim_game.py b/application/games/valheim_game.py new file mode 100644 index 0000000..840ce14 --- /dev/null +++ b/application/games/valheim_game.py @@ -0,0 +1,249 @@ +import os +import time + +from jinja2 import Environment, FileSystemLoader + +from application.common import logger, constants +from application.common.game_argument import GameArgument +from application.common.game_base import BaseGame +from application.common.toolbox import _get_proc_by_name, get_resources_dir +from application.extensions import DATABASE +from application.models.games import Games + + +class ValheimGame(BaseGame): + def __init__(self, defaults_dict: dict = {}) -> None: + super(ValheimGame, self).__init__(defaults_dict) + + self._game_name = "valheim" + self._game_pretty_name = "Valheim" + self._game_executable = "valheim_server.exe" + self._game_steam_id = "896660" + self._game_info_url = ( + "https://valheim.com/support/a-guide-to-dedicated-servers/" + ) + + if self._game_default_install_dir: + default_persistent_data_path = os.path.join( + self._game_default_install_dir, + constants.GAME_INSTALL_FOLDER, + self._game_name, + "saves", + ) + default_log_path = os.path.join( + self._game_default_install_dir, + constants.GAME_INSTALL_FOLDER, + self._game_name, + "logs", + "valheim.txt", + ) + else: + default_persistent_data_path = None + default_log_path = None + + # Add Args here, can update later. + # Default is 2456 + self._add_argument( + GameArgument( + "-name", + value="Valheim", + required=True, + use_quotes=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-port", + value=2456, + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-world", + value="badlands", + required=True, + use_quotes=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-password", + value="abc123", + required=True, + use_quotes=True, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-savedir", + value=default_persistent_data_path, + required=True, + use_quotes=True, + is_permanent=True, + file_mode=constants.FileModes.DIRECTORY.value, + ) + ) + + self._add_argument( + GameArgument( + "-public", + value="0", + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-logFile", + value=default_log_path, + required=True, + use_quotes=True, + is_permanent=True, + file_mode=constants.FileModes.FILE.value, + ) + ) + + self._add_argument( + GameArgument( + "-saveinterval", + value="1800", + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-backups", + value="4", + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-backupshort", + value="7200", + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-backuplong", + value="43200", + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + self._add_argument( + GameArgument( + "-crossplay", + value=" ", + required=False, + use_quotes=False, + is_permanent=False, + ) + ) + + self._add_argument( + GameArgument( + "-preset", + value="normal", + required=True, + use_quotes=False, + is_permanent=True, + ) + ) + + def startup(self) -> None: + # Run base class checks + super().startup() + + # Format command string. + command_args = self._get_command_str(args_only=True) + + # Create a formatted batch file. + env = Environment(loader=FileSystemLoader(get_resources_dir(__file__))) + template = env.get_template("start_valheim_server_template.bat.j2") + output_from_parsed_template = template.render( + GAME_STEAM_ID=self._game_steam_id, + GAME_NAME=self._game_name, + GAME_ARGUMENTS=command_args, + ) + + # Print the formatted jinja + logger.debug(output_from_parsed_template) + + # In theory, the software has already check that the game is installed, so no check/guard + # needed. + game_qry = Games.query.filter_by(game_steam_id=self._game_steam_id) + game_obj = game_qry.first() + game_install_dir = game_obj.game_install_dir + + # Need game install location to write batch file. + full_path_startup_script = os.path.join( + game_install_dir, constants.STARTUP_BATCH_FILE_NAME + ) + + # If file exists, remove it. + if os.path.exists(full_path_startup_script): + os.remove(full_path_startup_script) + + # Write the batch file. + with open(full_path_startup_script, "w") as myfile: + myfile.write(output_from_parsed_template) + + # Call the batch file on another process as to not block this one. + command = f'START /MIN CMD.EXE /C "{full_path_startup_script}"' + result = self._run_game(command, game_install_dir) + + time.sleep(1) + + process = _get_proc_by_name(self._game_executable) + + logger.info(result) + logger.info("Process:") + logger.info(process) + + update_dict = {"game_pid": int(process.pid)} + + game_qry.update(update_dict) + DATABASE.session.commit() + + def shutdown(self) -> None: + game_qry = Games.query.filter_by(game_steam_id=self._game_steam_id) + game_obj = game_qry.first() + game_pid = game_obj.game_pid + + process = _get_proc_by_name(self._game_executable) + + if process: + logger.info(process) + logger.info(game_pid) + + process.terminate() + process.wait() + + update_dict = {"game_pid": None} + game_qry.update(update_dict) + DATABASE.session.commit() diff --git a/application/gui/widgets/nginx_widget.py b/application/gui/widgets/nginx_widget.py index a639b71..ff12ab4 100644 --- a/application/gui/widgets/nginx_widget.py +++ b/application/gui/widgets/nginx_widget.py @@ -40,11 +40,15 @@ def __init__( self._nginx_enable_checkbox: QCheckBox = None self._nginx_hostname: QLineEdit = None self._nginx_port: QLineEdit = None + self._nginx_port_save_btn: QPushButton = None self._regenerate_certificate: QPushButton = None self._view_public_cert: QPushButton = None self._viewer_window = NginxCertViewer(self._clipboard, self._nginx_manager) + self.MIN_PORT_NUMBER = 10000 + self.MAX_PORT_NUMBER = 65535 + self.init_ui() def init_ui(self): @@ -97,10 +101,13 @@ def _create_nginx_controls(self) -> QVBoxLayout: h3_layout = QHBoxLayout() label = QLabel("Nginx Port: ") self._nginx_port = QLineEdit(nginx_proxy_port) + self._nginx_port_save_btn = QPushButton("Save") + + self._nginx_port_save_btn.clicked.connect(self._handle_nginx_port_save_btn) - self._nginx_port.textChanged.connect(self._handle_nginx_port_edit) h3_layout.addWidget(label) h3_layout.addWidget(self._nginx_port) + h3_layout.addWidget(self._nginx_port_save_btn) self._regenerate_certificate = QPushButton("Reset SSL Certificate") self._regenerate_certificate.clicked.connect(self._handle_regen_button) @@ -159,16 +166,30 @@ def _handle_nginx_hostname_edit(self, text): constants.SETTING_NGINX_PROXY_HOSTNAME, text ) - def _handle_nginx_port_edit(self, text): + def _handle_nginx_port_save_btn(self): + text = self._nginx_port.text() num_chars = len(text) - if text == "5000": + if int(text) <= self.MIN_PORT_NUMBER: + message = QMessageBox() + message.setText("Please use a port number greater than 10000.") + message.exec() + return + + if int(text) > self.MAX_PORT_NUMBER: + message = QMessageBox() + message.setText("Please use a port number less than 65535.") + message.exec() + return + + # Make sure every character is a unicode digit. + if len(text) > len(set(text)): message = QMessageBox() - message.setText("The port 5000 is a reserved port. Please use another one.") + message.setText("Please make sure all digits in the port are unique.") message.exec() return - if num_chars >= 4: + if num_chars >= 5: self._client.app.update_setting_by_name( constants.SETTING_NGINX_PROXY_PORT, text ) diff --git a/coverage.ps1 b/coverage.ps1 new file mode 100644 index 0000000..c18087b --- /dev/null +++ b/coverage.ps1 @@ -0,0 +1,2 @@ +coverage run -m pytest +coverage report \ No newline at end of file diff --git a/test-requirements.txt b/test-requirements.txt index fee3880..81008d5 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -2,4 +2,5 @@ black coverage flake8 -pytest \ No newline at end of file +pytest +pytest-mock diff --git a/tests/functional/test_settings_api.py b/tests/functional/test_settings_api.py index dd8bcf1..2368ef7 100644 --- a/tests/functional/test_settings_api.py +++ b/tests/functional/test_settings_api.py @@ -1,6 +1,6 @@ import os -from application.source.models.settings import Settings +from application.models.settings import Settings class TestAppSettingsApi: diff --git a/tests/unit/resources/appmanifest_1829350.acf b/tests/unit/resources/appmanifest_1829350.acf new file mode 100644 index 0000000..06e1f67 --- /dev/null +++ b/tests/unit/resources/appmanifest_1829350.acf @@ -0,0 +1,43 @@ +"AppState" +{ + "appid" "1829350" + "Universe" "1" + "LauncherPath" "C:\\AgentSmith\\steam\\steamcmd.exe" + "name" "V Rising Dedicated Server" + "StateFlags" "4" + "installdir" "VRisingDedicatedServer" + "LastUpdated" "1710615232" + "LastPlayed" "0" + "SizeOnDisk" "1453745726" + "StagingSize" "0" + "buildid" "12520643" + "LastOwner" "76561200388861161" + "UpdateResult" "0" + "BytesToDownload" "213631952" + "BytesDownloaded" "213631952" + "BytesToStage" "1453745726" + "BytesStaged" "1453745726" + "TargetBuildID" "12520643" + "AutoUpdateBehavior" "0" + "AllowOtherDownloadsWhileRunning" "0" + "ScheduledAutoUpdate" "0" + "InstalledDepots" + { + "1004" + { + "manifest" "1923798258558304932" + "size" "44194776" + } + "1829351" + { + "manifest" "5850011680124575304" + "size" "1409550950" + } + } + "UserConfig" + { + } + "MountedConfig" + { + } +} diff --git a/tests/unit/resources/steam_manifest/appmanifest_1829350.acf b/tests/unit/resources/steam_manifest/appmanifest_1829350.acf new file mode 100644 index 0000000..06e1f67 --- /dev/null +++ b/tests/unit/resources/steam_manifest/appmanifest_1829350.acf @@ -0,0 +1,43 @@ +"AppState" +{ + "appid" "1829350" + "Universe" "1" + "LauncherPath" "C:\\AgentSmith\\steam\\steamcmd.exe" + "name" "V Rising Dedicated Server" + "StateFlags" "4" + "installdir" "VRisingDedicatedServer" + "LastUpdated" "1710615232" + "LastPlayed" "0" + "SizeOnDisk" "1453745726" + "StagingSize" "0" + "buildid" "12520643" + "LastOwner" "76561200388861161" + "UpdateResult" "0" + "BytesToDownload" "213631952" + "BytesDownloaded" "213631952" + "BytesToStage" "1453745726" + "BytesStaged" "1453745726" + "TargetBuildID" "12520643" + "AutoUpdateBehavior" "0" + "AllowOtherDownloadsWhileRunning" "0" + "ScheduledAutoUpdate" "0" + "InstalledDepots" + { + "1004" + { + "manifest" "1923798258558304932" + "size" "44194776" + } + "1829351" + { + "manifest" "5850011680124575304" + "size" "1409550950" + } + } + "UserConfig" + { + } + "MountedConfig" + { + } +} diff --git a/tests/unit/test_authorization.py b/tests/unit/test_authorization.py new file mode 100644 index 0000000..475199c --- /dev/null +++ b/tests/unit/test_authorization.py @@ -0,0 +1,45 @@ +from application.common import authorization + + +class FakeRequest: + def __init__(self, headers=None): + self.headers = headers or {} + + +class FakeToken: + def __init__(self) -> None: + self.token_name = "foo" + + +class FakeSetting: + def __init__(self) -> None: + self.setting_value = "bar" + + +class TestAuthorization: + @classmethod + def setup_class(cls): + pass + + @classmethod + def teardown_class(cls): + pass + + def test_verify_bearer_token(self, mocker): + # Arrange + fake_headers = {"Authorization": "Bearer 1234"} + fake_request = FakeRequest(headers=fake_headers) + fake_token = FakeToken() + fake_setting = FakeSetting() + + mocker.patch( + "application.common.authorization._get_token", return_value=fake_token + ) + mocker.patch( + "application.common.authorization._get_setting", return_value=fake_setting + ) + mocker.patch("jwt.decode", return_value={"token_name": "foo"}) + + return_code = authorization._verify_bearer_token(fake_request) + + assert return_code == 200 diff --git a/tests/unit/test_steam_parser.py b/tests/unit/test_steam_parser.py new file mode 100644 index 0000000..7c49a3a --- /dev/null +++ b/tests/unit/test_steam_parser.py @@ -0,0 +1,35 @@ +import os + +from application.common.steam_manifest_parser import read_acf, parser, read_dir + + +class TestSteamManifestParser: + ACF_FILE_NAME = "appmanifest_1829350.acf" + ACF_FILE_PATH = "" + ACF_FOLDER = "" + + @classmethod + def setup_class(cls): + # Get the current working directory + current_location = os.path.abspath(__file__) + current_folder = os.path.dirname(current_location) + cls.ACF_FOLDER = os.path.join(current_folder, "resources", "steam_manifest") + cls.ACF_FILE_PATH = os.path.join( + current_folder, "resources", "steam_manifest", cls.ACF_FILE_NAME + ) + + @classmethod + def teardown_class(cls): + pass + + def test_read_dir(self): + steamapps = read_dir(self.ACF_FOLDER) + assert len(steamapps) == 1 + assert "1829350" in steamapps + + def test_read_acf(self): + acf = read_acf(self.ACF_FILE_PATH) + assert acf["appid"] == "1829350" + assert acf["name"] == "V Rising Dedicated Server" + assert acf["StateFlags"] == "4" + assert acf["installdir"] == "VRisingDedicatedServer"