From b2e6a6ff1a8f60e85f911ebe0235451c5642752b Mon Sep 17 00:00:00 2001 From: Opselon Date: Fri, 9 Oct 2026 23:00:01 +0000 Subject: [PATCH 1/2] fix(ci): fix undefined parse_mode in backup, update stale test assertions, and restore green test suite --- src/main.py | 5 +- src/updater.py | 26 ++++-- tests/test_bot_e2e.py | 9 ++- tests/test_security.py | 49 +++++++----- tests/test_service_management.py | 15 ++-- tests/test_system_commands.py | 42 ++++++---- tests/test_updater.py | 133 ++++++++++++++++++------------- 7 files changed, 171 insertions(+), 108 deletions(-) diff --git a/src/main.py b/src/main.py index 4fe339a..5c2ba31 100644 --- a/src/main.py +++ b/src/main.py @@ -2079,9 +2079,10 @@ async def backup(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """ query = update.callback_query user_id = update.effective_user.id - + parse_mode = _get_user_parse_mode(user_id) + await query.answer() - + # Show progress message progress_msg = await _send_message_safely(query.message.chat, "🔄 **Creating backup...**", user_id, preformatted=True) diff --git a/src/updater.py b/src/updater.py index 46e89f6..beac4d0 100644 --- a/src/updater.py +++ b/src/updater.py @@ -185,7 +185,10 @@ def log_message(message: str) -> None: # Create backup directory with timestamp timestamp = time.strftime("%Y%m%d_%H%M%S") backup_dir = REPO_ROOT / f"backup_{timestamp}" + # Optional files are skipped (with a warning) when absent; critical files must exist. data_files = ["config.json", "database.db", "var/encryption.key", "var/pq_encryption.key"] + optional_data_files = {"var/encryption.key", "var/pq_encryption.key"} + critical_data_files = {"config.json", "database.db"} log_message("[2/8] Creating comprehensive backup...") try: @@ -194,11 +197,18 @@ def log_message(message: str) -> None: # Backup critical files for file in data_files: src_path = REPO_ROOT / file - if src_path.exists(): - dst_path = backup_dir / file - dst_path.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(src_path, dst_path) - log_message(f" ✓ Backed up {file}") + if not src_path.exists(): + if file in optional_data_files: + logger.warning(f"Optional file {file} not found - skipping backup") + continue + raise FileNotFoundError(f"Required data file missing: {file}") + if src_path.stat().st_size == 0: + logger.warning(f"File {file} is empty - skipping backup") + continue + dst_path = backup_dir / file + dst_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src_path, dst_path) + log_message(f" ✓ Backed up {file}") # Backup source code (excluding large dirs) ignore_patterns = shutil.ignore_patterns( @@ -219,12 +229,12 @@ def log_message(message: str) -> None: time.sleep(3) # Additional wait time log_message("[4/8] Validating backup integrity...") - # Verify critical files exist in backup + # Verify critical files exist in backup (optional files may have been skipped) for file in ["config.json", "database.db"]: backup_file = backup_dir / file - if not backup_file.exists(): + if file in critical_data_files and not backup_file.exists(): raise ValueError(f"Critical file missing in backup: {file}") - if backup_file.stat().st_size == 0: + if backup_file.exists() and backup_file.stat().st_size == 0: raise ValueError(f"Backup file is empty: {file}") log_message("[5/8] Downloading new version...") diff --git a/tests/test_bot_e2e.py b/tests/test_bot_e2e.py index eb46292..9c41558 100644 --- a/tests/test_bot_e2e.py +++ b/tests/test_bot_e2e.py @@ -77,6 +77,10 @@ async def test_handle_server_connection_e2e(mocker, conn_type): # Patch the config to authorize the user mocker.patch.object(main.config, 'whitelisted_users', [user_id]) + # handle_server_connection rate-limits per user_id (3s). The parametrized cases + # share a user id, so clear the cooldown to avoid an early return. + main._user_cooldowns.clear() + # 2. Execute the handler try: @@ -99,7 +103,8 @@ async def test_handle_server_connection_e2e(mocker, conn_type): assert mock_conn.waited # Verify bot interactions - update.callback_query.edit_message_text.assert_called() + # handle_server_connection edits query.message via _edit_message_safely() + update.callback_query.message.edit_message_text.assert_called() # The text is in the second call to edit_message_text - final_call_args = update.callback_query.edit_message_text.call_args_list[1] + final_call_args = update.callback_query.message.edit_message_text.call_args_list[1] assert "Connected to" in final_call_args.args[0] diff --git a/tests/test_security.py b/tests/test_security.py index 90911d3..825b07e 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import os import pytest @@ -10,36 +11,40 @@ @pytest.fixture(autouse=True) -def clear_key_cache(monkeypatch): - """Reset memoized cipher between tests to avoid cross-test bleed.""" - security._get_cipher.cache_clear() - monkeypatch.delenv("TLA_ENCRYPTION_KEY", raising=False) +def clear_key_cache(): + """Reset memoized cipher dictionary between tests to avoid cross-test bleed.""" + security._get_ciphers.cache_clear() yield - security._get_cipher.cache_clear() - - -def test_env_key_takes_priority(monkeypatch, tmp_path): - """Explicit environment variable should be used over files.""" - keyfile = tmp_path / "encryption.key" - keyfile.write_text("should-not-be-used") - monkeypatch.setenv("TLA_ENCRYPTION_KEY", "K1pGzzKquqCPIAHJ_Tz6lhZKBiJMOrlrNUmt1Bd6Rwk=") - monkeypatch.setenv("TLA_ENCRYPTION_KEY_FILE", str(keyfile)) - - cipher = security._get_cipher() - decrypted = cipher.decrypt(cipher.encrypt(b"secret")) - - assert decrypted == b"secret" + security._get_ciphers.cache_clear() def test_keyfile_is_created_when_missing(monkeypatch, tmp_path): - """A new Fernet key should be generated and persisted when absent.""" + """A new Fernet key file (JSON format) should be generated and persisted when absent.""" keyfile = tmp_path / "nested" / "encryption.key" monkeypatch.setenv("TLA_ENCRYPTION_KEY_FILE", str(keyfile)) - cipher = security._get_cipher() - decrypted = cipher.decrypt(cipher.encrypt(b"data")) + encrypted = security.encrypt_secret("data") + assert encrypted is not None + decrypted = security.decrypt_secret(encrypted) + assert decrypted == "data" - assert decrypted == b"data" assert keyfile.exists() assert os.stat(keyfile).st_mode & 0o777 == 0o600 + data = json.loads(keyfile.read_text(encoding="utf-8")) + assert data.get("primary_key") == "v1" + assert "v1" in data.get("keys", {}) + + +def test_encrypt_decrypt_roundtrip(monkeypatch, tmp_path): + """Encrypt and decrypt roundtrip matches original text.""" + keyfile = tmp_path / "encryption.key" + monkeypatch.setenv("TLA_ENCRYPTION_KEY_FILE", str(keyfile)) + + secret = "my-super-secret-ssh-password-12345" + encrypted = security.encrypt_secret(secret) + assert encrypted is not None + assert encrypted != secret.encode("utf-8") + + decrypted = security.decrypt_secret(encrypted) + assert decrypted == secret diff --git a/tests/test_service_management.py b/tests/test_service_management.py index e747380..7f5108f 100644 --- a/tests/test_service_management.py +++ b/tests/test_service_management.py @@ -1,5 +1,5 @@ import pytest -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch from src.main import service_action_start, execute_service_action, AWAIT_SERVICE_NAME @pytest.mark.asyncio @@ -9,15 +9,16 @@ async def test_service_action_start(mock_config): mock_config.whitelisted_users = [12345] update = AsyncMock() update.effective_user.id = 12345 + update.effective_chat = MagicMock(id=12345) update.callback_query.data = "check_service_server1" context = AsyncMock() result = await service_action_start(update, context) assert result == AWAIT_SERVICE_NAME - update.callback_query.edit_message_text.assert_called_once_with( - "Please enter the name of the service to `check`.", parse_mode='Markdown' - ) + update.callback_query.message.edit_message_text.assert_called_once() + args, kwargs = update.callback_query.message.edit_message_text.call_args + assert "Please enter the name of the service to `check`." in args[0] @pytest.mark.asyncio @patch('src.main.config') @@ -27,6 +28,7 @@ async def test_execute_service_action(mock_ssh_manager, mock_config): mock_config.whitelisted_users = [12345] update = AsyncMock() update.effective_user.id = 12345 + update.effective_chat = MagicMock(id=12345) update.message.text = "nginx" context = AsyncMock() context.user_data = {'service_action': 'status', 'alias': 'server1'} @@ -43,5 +45,6 @@ async def mock_run_command(alias, command): result = await execute_service_action(update, context) assert result == ConversationHandler.END - # The assertions are now inside the mock_run_command function - update.message.reply_text.assert_called_once() + # The result message is sent via chat.send_message and then edited with the output + update.message.chat.send_message.assert_called_once() + # The assertions about the command/alias are inside mock_run_command above diff --git a/tests/test_system_commands.py b/tests/test_system_commands.py index ee2117d..ff7c0e3 100644 --- a/tests/test_system_commands.py +++ b/tests/test_system_commands.py @@ -9,10 +9,23 @@ def authorized_update(): """Creates a mock update object that will pass the @authorized decorator.""" update = AsyncMock(spec=Update) update.effective_user = MagicMock(id=12345) + update.effective_chat = MagicMock(id=12345) update.callback_query = AsyncMock() update.callback_query.from_user = MagicMock(id=12345) + update.callback_query.message = AsyncMock() return update +@pytest.fixture +def mock_translations(): + """Makes translate() pass keys through verbatim so assertions stay stable.""" + with patch('src.main.translate', side_effect=lambda key, lang, **kw: key), \ + patch('src.main._translate_for_user', side_effect=lambda user_id, key, **kw: key): + yield + +def _edit_call(update): + """Returns the first call args of the real edit target: query.message.edit_message_text.""" + return update.callback_query.message.edit_message_text.call_args + @pytest.mark.asyncio @patch('src.main.config') async def test_system_commands_menu(mock_config, authorized_update): @@ -22,9 +35,9 @@ async def test_system_commands_menu(mock_config, authorized_update): await main.system_commands_menu(authorized_update, MagicMock(spec=ContextTypes.DEFAULT_TYPE)) - authorized_update.callback_query.edit_message_text.assert_called_once() - call_args = authorized_update.callback_query.edit_message_text.call_args - assert "**⚙️ System Commands for test_alias**" in call_args[0][0] + authorized_update.callback_query.message.edit_message_text.assert_called_once() + call_args = _edit_call(authorized_update) + assert "System Commands for test_alias" in call_args[0][0] reply_markup = call_args[1]['reply_markup'] assert isinstance(reply_markup, InlineKeyboardMarkup) @@ -46,9 +59,9 @@ async def test_confirm_system_command(mock_config, authorized_update): await main.confirm_system_command(authorized_update, MagicMock(spec=ContextTypes.DEFAULT_TYPE)) - authorized_update.callback_query.edit_message_text.assert_called_once() - call_args = authorized_update.callback_query.edit_message_text.call_args - assert "**⚠️ Are you sure you want to reboot the server `test_alias`?**" in call_args[0][0] + authorized_update.callback_query.message.edit_message_text.assert_called_once() + call_args = _edit_call(authorized_update) + assert "Are you sure you want to reboot the server `test_alias`?" in call_args[0][0] reply_markup = call_args[1]['reply_markup'] buttons = reply_markup.inline_keyboard @@ -70,9 +83,10 @@ async def mock_run_command_gen(*args, **kwargs): await main.execute_system_command(authorized_update, MagicMock(spec=ContextTypes.DEFAULT_TYPE)) - authorized_update.callback_query.edit_message_text.assert_called_once_with( + authorized_update.callback_query.message.edit_message_text.assert_called_once_with( "✅ **Command `reboot` sent to `test_alias` successfully.**", - parse_mode='Markdown' + parse_mode='MarkdownV2', + reply_markup=None ) @pytest.mark.asyncio @@ -92,9 +106,9 @@ async def mock_run_command_gen(user_id, alias, command): await main.get_disk_usage(authorized_update, MagicMock(spec=ContextTypes.DEFAULT_TYPE)) - authorized_update.callback_query.edit_message_text.assert_called_once() - call_args = authorized_update.callback_query.edit_message_text.call_args - assert "**💾 Disk Usage for `test_alias`**" in call_args[0][0] + authorized_update.callback_query.message.edit_message_text.assert_called_once() + call_args = _edit_call(authorized_update) + assert "Disk Usage for `test_alias`" in call_args[0][0] assert "Filesystem" in call_args[0][0] @pytest.mark.asyncio @@ -114,7 +128,7 @@ async def mock_run_command_gen(user_id, alias, command): await main.get_network_info(authorized_update, MagicMock(spec=ContextTypes.DEFAULT_TYPE)) - authorized_update.callback_query.edit_message_text.assert_called_once() - call_args = authorized_update.callback_query.edit_message_text.call_args - assert "**🌐 Network Info for `test_alias`**" in call_args[0][0] + authorized_update.callback_query.message.edit_message_text.assert_called_once() + call_args = _edit_call(authorized_update) + assert "Network Info for `test_alias`" in call_args[0][0] assert "LOOPBACK" in call_args[0][0] diff --git a/tests/test_updater.py b/tests/test_updater.py index 81edae0..ba91c62 100644 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -1,75 +1,100 @@ +import sys + import pytest from unittest.mock import patch, MagicMock, call -from src.updater import apply_update, rollback, COMMAND_TIMEOUT -import sys -@patch('src.updater.shutil.rmtree') -@patch('src.updater.shutil.copy') -@patch('src.updater.shutil.copytree') -@patch('src.updater.download_and_extract_zip') -@patch('src.updater.subprocess.run') -@patch('src.updater.time.sleep', return_value=None) # Mock time.sleep to speed up tests -def test_apply_update_success(mock_sleep, mock_run, mock_download, mock_copytree, mock_copy, mock_rmtree): +from src.updater import COMMAND_TIMEOUT, apply_update, rollback + +# apply_update() drives systemd through the _is_systemd_available / +# _stop_bot_gracefully / _start_bot_gracefully helpers, so the tests patch those +# (not subprocess.run) to keep assertions aligned with the real call flow. + + +@pytest.fixture +def fake_data_files(tmp_path, monkeypatch): + """Point REPO_ROOT at a temp dir holding real config.json/database.db and a src/ dir.""" + for name in ("config.json", "database.db"): + (tmp_path / name).write_text("dummy-content") + (tmp_path / "src").mkdir(exist_ok=True) + (tmp_path / "src" / "dummy.py").write_text("# dummy") + monkeypatch.setattr("src.updater.REPO_ROOT", tmp_path) + return tmp_path + + +@patch("src.updater._is_systemd_available", return_value=True) +@patch("src.updater._stop_bot_gracefully") +@patch("src.updater._start_bot_gracefully") +@patch("src.updater.shutil.rmtree") +@patch("src.updater.download_and_extract_zip") +@patch("src.updater.subprocess.run") +@patch("src.updater.time.sleep", return_value=None) +def test_apply_update_success( + mock_sleep, + mock_run, + mock_download, + mock_rmtree, + mock_start, + mock_stop, + mock_systemd, + fake_data_files, +): """Test the successful application of an update.""" - with patch('src.updater.Path.exists', return_value=True): - result = apply_update() + result = apply_update() assert "Update process completed successfully!" in result mock_download.assert_called_once() - mock_copytree.assert_called_once() - # Assuming at least 2 data files are backed up and restored - assert mock_copy.call_count >= 2 + # The only direct subprocess.run call is the dependency install; systemd + # start/stop go through the patched helpers. expected_calls = [ - call(["systemctl", "stop", "telegram_bot.service"], check=True, timeout=COMMAND_TIMEOUT), - call([sys.executable, "-m", "pip", "install", "-e", "."], check=True, timeout=COMMAND_TIMEOUT), - call(["systemctl", "start", "telegram_bot.service"], check=True, timeout=COMMAND_TIMEOUT) + call([sys.executable, "-m", "pip", "install", "-e", ".", "--quiet"], + check=True, timeout=COMMAND_TIMEOUT, capture_output=True, text=True), ] mock_run.assert_has_calls(expected_calls) - assert mock_run.call_count == 3 - -@patch('src.updater.shutil.rmtree') -@patch('src.updater.shutil.copy') -@patch('src.updater.shutil.copytree') -@patch('src.updater.download_and_extract_zip', side_effect=Exception("Download failed")) -@patch('src.updater.rollback') -@patch('src.updater.time.sleep', return_value=None) -def test_apply_update_failure_and_rollback(mock_sleep, mock_rollback, mock_download, mock_copytree, mock_copy, mock_rmtree): + mock_start.assert_called_once() + assert mock_stop.call_count >= 1 + + +@patch("src.updater._is_systemd_available", return_value=True) +@patch("src.updater._stop_bot_gracefully") +@patch("src.updater._start_bot_gracefully") +@patch("src.updater.download_and_extract_zip", side_effect=Exception("Download failed")) +@patch("src.updater.rollback") +@patch("src.updater.time.sleep", return_value=None) +def test_apply_update_failure_and_rollback( + mock_sleep, + mock_rollback, + mock_download, + mock_start, + mock_stop, + mock_systemd, + fake_data_files, +): """Test a failed update and the subsequent rollback.""" - # Create dummy data files for the backup step to succeed before the download fails - from pathlib import Path - config_file = Path("config.json") - db_file = Path("database.db") - config_file.touch() - db_file.touch() - - try: - with patch('src.updater.subprocess.run') as mock_run: - result = apply_update() - - mock_run.assert_called_once_with(["systemctl", "stop", "telegram_bot.service"], check=True, timeout=COMMAND_TIMEOUT) - assert "Update Failed: Download failed" in result - assert "Attempting to roll back..." in result - mock_rollback.assert_called_once() - finally: - # Clean up the dummy files to not affect other tests - if config_file.exists(): - config_file.unlink() - if db_file.exists(): - db_file.unlink() - -@patch('src.updater.shutil.copytree') -@patch('src.updater.subprocess.run') -def test_rollback(mock_run, mock_copytree): + with patch("src.updater.subprocess.run"): + result = apply_update() + + assert "Update Failed: Download failed" in result + assert "Attempting automatic rollback..." in result + mock_rollback.assert_called_once() + mock_stop.assert_called() + + +@patch("src.updater._is_systemd_available", return_value=True) +@patch("src.updater._stop_bot_gracefully") +@patch("src.updater._start_bot_gracefully") +@patch("src.updater.shutil.copy2") +@patch("src.updater.shutil.copytree") +@patch("src.updater.subprocess.run") +def test_rollback(mock_run, mock_copytree, mock_copy2, mock_start, mock_stop, mock_systemd): """Test the rollback function.""" backup_dir = MagicMock() rollback(backup_dir) expected_calls = [ - call(["systemctl", "stop", "telegram_bot.service"], check=True, timeout=COMMAND_TIMEOUT), - call([sys.executable, '-m', 'pip', 'install', '-e', '.'], check=True, timeout=COMMAND_TIMEOUT), - call(["systemctl", "start", "telegram_bot.service"], check=True, timeout=COMMAND_TIMEOUT) + call([sys.executable, '-m', 'pip', 'install', '-e', '.', '--quiet'], + check=True, timeout=COMMAND_TIMEOUT, capture_output=True), ] mock_run.assert_has_calls(expected_calls) mock_copytree.assert_called_once() From 6eea5e1453957ad9d6282144fa66522ce8075aad Mon Sep 17 00:00:00 2001 From: Opselon Date: Fri, 9 Oct 2026 23:02:00 +0000 Subject: [PATCH 2/2] ci: install [dev] optional dependencies so pytest is available in runner --- .github/workflows/python-ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/python-ci.yml b/.github/workflows/python-ci.yml index 1ef7833..d73afb5 100644 --- a/.github/workflows/python-ci.yml +++ b/.github/workflows/python-ci.yml @@ -34,7 +34,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -e . + pip install -e ".[dev]" - name: Run tests env: