From 337b7d21d45201b2a11a20e8941a0a25a36bc8ee Mon Sep 17 00:00:00 2001 From: Yurii Balandiuk Date: Thu, 11 Sep 2025 18:21:57 +0300 Subject: [PATCH 1/2] docs, chore, refactor: update documentation, Docker Compose, and rename files/folders - Updated project documentation - Modified Docker Compose configuration - Renamed folders and files for better structure --- TelegramBotAlertsPack/alerts.py | 143 ------------ compose.yml | 6 +- {LoggerPack => logger_setup}/__init__.py | 0 .../logger.py | 14 +- main.py | 22 +- {MonitorPack => monitoring}/__init__.py | 0 .../setup_monitoring.py | 73 ++++-- requirements.txt | 3 + .../__init__.py | 0 telegram_bot_alerts/setup_alerts.py | 215 ++++++++++++++++++ tests/python_tests.py | 0 11 files changed, 301 insertions(+), 175 deletions(-) delete mode 100644 TelegramBotAlertsPack/alerts.py rename {LoggerPack => logger_setup}/__init__.py (100%) rename LoggerPack/log_container.py => logger_setup/logger.py (51%) rename {MonitorPack => monitoring}/__init__.py (100%) rename MonitorPack/monitor.py => monitoring/setup_monitoring.py (50%) rename {TelegramBotAlertsPack => telegram_bot_alerts}/__init__.py (100%) create mode 100644 telegram_bot_alerts/setup_alerts.py create mode 100644 tests/python_tests.py diff --git a/TelegramBotAlertsPack/alerts.py b/TelegramBotAlertsPack/alerts.py deleted file mode 100644 index 297ab5c..0000000 --- a/TelegramBotAlertsPack/alerts.py +++ /dev/null @@ -1,143 +0,0 @@ -import sys -import threading -from concurrent.futures import ThreadPoolExecutor -import LoggerPack.log_container as lc -from dotenv import load_dotenv -import os -import time -import telebot -from telebot.types import ReplyKeyboardMarkup, KeyboardButton - -try: - from MonitorPack.monitor import get_container_metrics as gcm - from MonitorPack.monitor import get_container_status_real_time as gcs -except FileNotFoundError as excerr: - lc.logger.error("Docker isn't working!\n" + str(excerr)) - print("Docker isn't working!") - -if __name__ == "__main__": - lc.logger.error("This file cannot be run as main!") - print("\nThis file cannot be run as main!") - sys.exit() - -# Load environment variables from the .env file -try: - load_dotenv() -except NameError as nerr: - lc.logger.error("Unable to load the environment!\n" + str(nerr)) - print("Unable to load the environment!") - -try: - API_TOKEN = os.getenv("API_TOKEN") - if not API_TOKEN: raise ValueError - bot = telebot.TeleBot(API_TOKEN) -except ValueError as verr: - lc.logger.error("The API-TOKEN hasn't been found!\n" + str(verr)) - print("The API-TOKEN hasn't been found!") - -def main_keyboard(): - markup = ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=False) - btn1 = KeyboardButton("▶️ Start") - btn2 = KeyboardButton("🔍 Help") - btn3 = KeyboardButton("📊 Status") - btn4 = KeyboardButton("ℹ️ Clear") - markup.add(btn1, btn2, btn3, btn4) - return markup - -# executor = ThreadPoolExecutor(max_workers=1) -# -# def stream_function(message): -# executor.submit(update_containers_status_real_time, message.chat.id) - # threading.Thread(target=update_containers_status_real_time, args=(message.chat.id,)).start() - -# Handle '/start' -@bot.message_handler(commands=['start']) # The ‘bot’ highlighting occurs due to the possible failure to launch the bot when checking in the try/except block ^^^ -def send_start(message): - bot.send_message(message.chat.id, "Hi! I'm the monitor container telegram bot", reply_markup=main_keyboard()) - # stream_function(message) - -@bot.message_handler(func=lambda message: message.text == "▶️ Start") -def send_start_instruction(message): - send_start(message) - -# Handle '/help' -@bot.message_handler(func=lambda message: message.text == "🔍 Help") -def send_help(message): - bot.send_message(message.chat.id, "Help information: Use Status to get container stats, Clear to clear chat.", reply_markup=main_keyboard()) - -@bot.message_handler(commands=['help']) -def send_help_instruction(message): - send_help(message) - -# Handle '/status' -@bot.message_handler(func=lambda message: message.text == "📊 Status") -def sen_containers_stats(message): - count = 0 - - for number in range(5): - try: - container_stats_text = "\n".join(gcm()) - bot.reply_to(message, container_stats_text) - count += 1 - if count == 5: - bot.send_message(message.chat.id, "You have received five information containers", reply_markup=main_keyboard()) - break - except Exception as err: - lc.logger.error("The containers is not running now!\n" + str(err)) - bot.reply_to(message, "The docker containers is not running now!") - break - -@bot.message_handler(commands=['status']) -def send_status_instruction(message): - sen_containers_stats(message) - -# Handle '/clear' -@bot.message_handler(func=lambda message: message.text == "ℹ️ Clear") -def clear_chat(message): - chat_id = message.chat.id - last_message_id = message.message_id - - for msg_id in range(last_message_id, last_message_id - 100, -1): - try: - bot.delete_message(chat_id, msg_id) - time.sleep(0.2) - except Exception as exc: - pass - bot.send_message(message.chat.id, "The chat is cleared! ", reply_markup=main_keyboard()) - -@bot.message_handler(commands=['clear']) -def send_clear_instruction(message): - clear_chat(message) - -# Handle all other messages with content_type 'text' (content_types defaults to ['text']) -@bot.message_handler(func=lambda message: True) -def echo_message(message): - bot.reply_to(message, message.text) - bot.send_message(message.chat.id, "Make your choice ->", reply_markup=main_keyboard()) - -# Function for parallel container checking to avoid resource overload -def update_containers_status_real_time(chat_id): - count = 0 - - while True: - status = gcs() - if status == 0: - count += 1 - if count % 10 == 0: - print("Done:", count) - else: - print("Done:", count, end=" | ") - time.sleep(15) - continue - bot.send_message(chat_id, status) - time.sleep(15) - -# The start telegram bot function -def start_telegram_bot(): - try: - bot.infinity_polling(none_stop=True, timeout=60) - except Exception as exc: - lc.logger.error("The telegram bot is not running now!" + str(exc)) - time.sleep(15) - - diff --git a/compose.yml b/compose.yml index 717fcac..beed02b 100644 --- a/compose.yml +++ b/compose.yml @@ -3,7 +3,7 @@ services: my-app: image: containers-monitoring-app - container_name: my-python-container + container_name: main-runner restart: always environment: - DOCKER_HOST=tcp://host.docker.internal:2375 @@ -16,7 +16,7 @@ services: ports: - "9090:9090" volumes: - - D:/ContainerMonitoringProgram/container-monitoring/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml + - D:/own_projects/Containers-monitoring-application/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml command: - "--config.file=/etc/prometheus/prometheus.yml" restart: always @@ -48,7 +48,7 @@ services: volumes: - grafana-data:/var/lib/grafana env_file: - - D:/ContainerMonitoringProgram/container-monitoring/.env + - D:/own_projects/Containers-monitoring-application/.env restart: always volumes: diff --git a/LoggerPack/__init__.py b/logger_setup/__init__.py similarity index 100% rename from LoggerPack/__init__.py rename to logger_setup/__init__.py diff --git a/LoggerPack/log_container.py b/logger_setup/logger.py similarity index 51% rename from LoggerPack/log_container.py rename to logger_setup/logger.py index e3a2ddf..cde9daa 100644 --- a/LoggerPack/log_container.py +++ b/logger_setup/logger.py @@ -1,16 +1,24 @@ +""" +The module is responsible for creating logs for monitoring execution +""" + +import sys import logging + from logging.handlers import RotatingFileHandler -import sys + logger = logging.getLogger("ContainerMonitor") logger.setLevel(logging.DEBUG) -formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') +formatter = logging.Formatter('%(asctime)s - %(name)s - \ + %(levelname)s - %(message)s') console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) -file_handler = RotatingFileHandler("../container_monitor.log", maxBytes=1024 * 1024, backupCount=5) +file_handler = RotatingFileHandler("../container_monitor.log", + maxBytes=1024 * 1024, backupCount=5) file_handler.setFormatter(formatter) logger.addHandler(file_handler) diff --git a/main.py b/main.py index 63b94f8..58cbded 100644 --- a/main.py +++ b/main.py @@ -1,15 +1,19 @@ -import LoggerPack.log_container as lc -from TelegramBotAlertsPack import alerts +""" -def main_block(): +""" +import logger_setup.logger as lc +from telegram_bot_alerts import setup_alerts + + +""" + +""" +def main() -> None: lc.logger.info("The app has started!") - alerts.start_telegram_bot() + setup_alerts.start_telegram_bot() return 0 if __name__ == "__main__": - ending_code = main_block() - if ending_code == 0: + END = main() + if END == 0: lc.logger.info("The app is complete!") - - - diff --git a/MonitorPack/__init__.py b/monitoring/__init__.py similarity index 100% rename from MonitorPack/__init__.py rename to monitoring/__init__.py diff --git a/MonitorPack/monitor.py b/monitoring/setup_monitoring.py similarity index 50% rename from MonitorPack/monitor.py rename to monitoring/setup_monitoring.py index bf92311..9e8baab 100644 --- a/MonitorPack/monitor.py +++ b/monitoring/setup_monitoring.py @@ -1,7 +1,9 @@ -import LoggerPack.log_container as lc +import sys import docker from docker import errors -import sys +from typing import Optional + +import logger_setup.logger as lc if __name__ == "__main__": lc.logger.error("This file cannot be run as main!") @@ -13,8 +15,23 @@ except errors.DockerException: lc.logger.error("Docker is not working right now!") -def get_container_metrics(): +def get_container_metrics() -> list[str]: + """ + Retrieve metrics for all running Docker containers. + + For each container, this function collects: + - container name + - container status + - parent image + - creation time + - CPU usage percentage + - RAM usage in MB + + Returns: + list[str]: A list of formatted strings containing metrics + for each container. + """ containers = client.containers.list() containers_stats = [] for container in containers: @@ -33,15 +50,41 @@ def get_container_metrics(): f"~ RAM use: {mem_usage_mb:.2f} MB\n") return containers_stats -def calculate_cpu_percent(stats): - cpu_delta = stats["cpu_stats"]["cpu_usage"]["total_usage"] - stats["precpu_stats"]["cpu_usage"]["total_usage"] - system_delta = stats["cpu_stats"]["system_cpu_usage"] - stats["precpu_stats"]["system_cpu_usage"] + +def calculate_cpu_percent(stats) -> float: + """ + Calculate CPU usage percentage for a Docker container. + + Args: + stats (dict): The container statistics dictionary + obtained from `container.stats(stream=False)`. + + Returns: + float: The CPU usage percentage. + """ + cpu_delta = stats["cpu_stats"]["cpu_usage"]["total_usage"] - \ + stats["precpu_stats"]["cpu_usage"]["total_usage"] + system_delta = stats["cpu_stats"]["system_cpu_usage"] - \ + stats["precpu_stats"]["system_cpu_usage"] if system_delta > 0: - return (cpu_delta / system_delta) * len(stats["cpu_stats"]["cpu_usage"]["percpu_usage"]) * 100.0 + return (cpu_delta / system_delta) * \ + len(stats["cpu_stats"]["cpu_usage"]["percpu_usage"]) * 100.0 return 0.0 -def get_container_status_real_time(): + +def get_container_status_real_time() -> Optional[str]: + """ + Check running containers for high resource usage in real time. + + This function monitors all containers and reports if: + - CPU usage exceeds 50% + - Memory usage exceeds 500 MB + + Returns: + str or int: Warning message string if any container exceeds + thresholds; otherwise, returns None. + """ containers = client.containers.list() for container in containers: @@ -49,15 +92,11 @@ def get_container_status_real_time(): cpu_percent = calculate_cpu_percent(stats) if float(cpu_percent) > 50.00: - return f"The container {container.name} uses more than 50% of the available processor resources on the host" + return f"The container {container.name} uses more than 50% \ + of the available processor resources on the host" mem_usage_mb = stats["memory_stats"]["usage"] / (1024 * 1024) if float(mem_usage_mb) > 500.00: - return f"The container {container.name} uses 500 MB of the available RAM resources on the host" - - return 0 - - - - - + return f"The container {container.name} uses \ + 500 MB of the available RAM resources on the host" + return None \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index af09339..2fb28ac 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,6 @@ docker pyTelegramBotAPI python-dotenv +black +pylint +flake8 \ No newline at end of file diff --git a/TelegramBotAlertsPack/__init__.py b/telegram_bot_alerts/__init__.py similarity index 100% rename from TelegramBotAlertsPack/__init__.py rename to telegram_bot_alerts/__init__.py diff --git a/telegram_bot_alerts/setup_alerts.py b/telegram_bot_alerts/setup_alerts.py new file mode 100644 index 0000000..66f88a1 --- /dev/null +++ b/telegram_bot_alerts/setup_alerts.py @@ -0,0 +1,215 @@ +import sys +import os +import time +import logger_setup.logger as lc +import telebot + +from telebot.types import ReplyKeyboardMarkup, KeyboardButton +from dotenv import load_dotenv + +try: + from monitoring.setup_monitoring import get_container_metrics as gcm + from monitoring.setup_monitoring import get_container_status_real_time as gcs +except FileNotFoundError as excerr: + lc.logger.error("Docker isn't working now!\n" + str(excerr)) + print("Docker isn't working now!") + +if __name__ == "__main__": + lc.logger.error("This file cannot be run as main!") + print("\nThis file cannot be run as main!") + sys.exit() + +try: + load_dotenv() +except NameError as nerr: + lc.logger.error("Unable to load the environment!\n" + str(nerr)) + print("Unable to load the environment!") + +try: + API_TOKEN = os.getenv("API_TOKEN") + if not API_TOKEN: + raise ValueError + bot = telebot.TeleBot(API_TOKEN) +except ValueError as verr: + lc.logger.error("The API-TOKEN hasn't been found!\n" + str(verr)) + print("The API-TOKEN hasn't been found!") + + +def main_keyboard() -> ReplyKeyboardMarkup: + """ + Create the main keyboard for the Telegram bot. + + Returns: + ReplyKeyboardMarkup: Keyboard markup with buttons for Start, Help, + Status, and Clear actions. + """ + markup = ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=False) + btn1 = KeyboardButton("▶️ Start") + btn2 = KeyboardButton("🔍 Help") + btn3 = KeyboardButton("📊 Status") + btn4 = KeyboardButton("ℹ️ Clear") + markup.add(btn1, btn2, btn3, btn4) + return markup + + +@bot.message_handler(commands=['start']) +def send_start(message) -> None: + """ + Send the start message to the user with the main keyboard. + + Args: + message: Telegram message object containing chat and user info. + """ + bot.send_message(message.chat.id, "Hi! I\'m a Docker container monitoring bot", + reply_markup=main_keyboard()) + + +@bot.message_handler(func=lambda message: message.text == "▶️ Start") +def send_start_instruction(message) -> None: + """ + Trigger the start command via button press. + + Args: + message: Telegram message object. + """ + send_start(message) + + +@bot.message_handler(func=lambda message: message.text == "🔍 Help") +def send_help(message) -> None: + """ + Send help information to the user. + + Args: + message: Telegram message object. + """ + bot.send_message(message.chat.id, "Help information: Use Status to " + "get container stats, Clear to clear chat.", reply_markup=main_keyboard()) + + +@bot.message_handler(commands=['help']) +def send_help_instruction(message) -> None: + """ + Trigger the help command via /help command. + + Args: + message: Telegram message object. + """ + send_help(message) + + +@bot.message_handler(func=lambda message: message.text == "📊 Status") +def sen_containers_stats(message) -> None: + """ + Send real-time Docker container statistics to the user. + + Args: + message: Telegram message object. + """ + count = 0 + + for number in range(5): + try: + container_stats_text = "\n".join(gcm()) + bot.reply_to(message, container_stats_text) + count += 1 + if count == 5: + bot.send_message(message.chat.id, + "You have received five information containers", + reply_markup=main_keyboard()) + break + except Exception as err: + lc.logger.error("The docker containers is not running now!\n" + str(err)) + bot.reply_to(message, "The docker containers is not running now!") + break + + +@bot.message_handler(commands=['status']) +def send_status_instruction(message) -> None: + """ + Trigger the status command via /status command. + + Args: + message: Telegram message object. + """ + sen_containers_stats(message) + + +@bot.message_handler(func=lambda message: message.text == "ℹ️ Clear") +def clear_chat(message) -> None: + """ + Clear the last 100 messages in the chat. + + Args: + message: Telegram message object. + """ + chat_id = message.chat.id + last_message_id = message.message_id + + for msg_id in range(last_message_id, last_message_id - 100, -1): + try: + bot.delete_message(chat_id, msg_id) + time.sleep(0.2) + except Exception as exc: + pass + bot.send_message(message.chat.id, "The chat has been cleared!", reply_markup=main_keyboard()) + + +@bot.message_handler(commands=['clear']) +def send_clear_instruction(message) -> None: + """ + Trigger the clear command via /clear command. + + Args: + message: Telegram message object. + """ + clear_chat(message) + + +@bot.message_handler(func=lambda message: True) +def echo_message(message) -> None: + """ + Echo back any message and show the main keyboard. + + Args: + message: Telegram message object. + """ + bot.reply_to(message, message.text) + bot.send_message(message.chat.id, "Make your choice ->", reply_markup=main_keyboard()) + + +def update_containers_status_real_time(chat_id: int) -> None: + """ + Continuously send real-time container status updates to the user. + + Args: + chat_id (int): Telegram chat ID where updates will be sent. + """ + count = 0 + + while True: + status = gcs() + if status is None: + count += 1 + if count % 10 == 0: + print("Done:", count) + else: + print("Done:", count, end=" | ") + time.sleep(15) + continue + bot.send_message(chat_id, status) + time.sleep(15) + + +def start_telegram_bot() -> None: + """ + Start the Telegram bot with infinite polling. + + Handles exceptions and retries after 15 seconds if the bot fails. + + """ + try: + bot.infinity_polling(none_stop=True, timeout=60) + except Exception as exc: + lc.logger.error("The telegram bot is not running now!" + str(exc)) + time.sleep(15) \ No newline at end of file diff --git a/tests/python_tests.py b/tests/python_tests.py new file mode 100644 index 0000000..e69de29 From b4f094e3592c762e1d93f261065f25dbd280393e Mon Sep 17 00:00:00 2001 From: Yurii Balandiuk Date: Mon, 15 Sep 2025 16:12:43 +0300 Subject: [PATCH 2/2] test: extend test suite with coverage for monitoring and bot modules --- main.py | 18 +++-- monitoring/setup_monitoring.py | 1 + requirements.txt | 3 +- tests/{python_tests.py => __init__.py} | 0 tests/setup_alerts_test.py | 57 +++++++++++++ tests/setup_monitoring_test.py | 106 +++++++++++++++++++++++++ 6 files changed, 178 insertions(+), 7 deletions(-) rename tests/{python_tests.py => __init__.py} (100%) create mode 100644 tests/setup_alerts_test.py create mode 100644 tests/setup_monitoring_test.py diff --git a/main.py b/main.py index 58cbded..0d5dcd6 100644 --- a/main.py +++ b/main.py @@ -1,18 +1,24 @@ -""" - -""" import logger_setup.logger as lc from telegram_bot_alerts import setup_alerts -""" +def main() -> int: + """ + Main entry point of the application. + + This function initializes the logger, starts the Telegram bot for alerts, + and returns 0 upon successful execution. -""" -def main() -> None: + Steps: + 1. Logs that the application has started. + 2. Starts the Telegram bot using `setup_alerts.start_telegram_bot()`. + 3. Returns 0 when execution completes successfully. + """ lc.logger.info("The app has started!") setup_alerts.start_telegram_bot() return 0 + if __name__ == "__main__": END = main() if END == 0: diff --git a/monitoring/setup_monitoring.py b/monitoring/setup_monitoring.py index 9e8baab..ec27a81 100644 --- a/monitoring/setup_monitoring.py +++ b/monitoring/setup_monitoring.py @@ -13,6 +13,7 @@ try: client = docker.from_env() except errors.DockerException: + client = None lc.logger.error("Docker is not working right now!") diff --git a/requirements.txt b/requirements.txt index 2fb28ac..572327b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,4 +3,5 @@ pyTelegramBotAPI python-dotenv black pylint -flake8 \ No newline at end of file +flake8 +pytest \ No newline at end of file diff --git a/tests/python_tests.py b/tests/__init__.py similarity index 100% rename from tests/python_tests.py rename to tests/__init__.py diff --git a/tests/setup_alerts_test.py b/tests/setup_alerts_test.py new file mode 100644 index 0000000..62d0865 --- /dev/null +++ b/tests/setup_alerts_test.py @@ -0,0 +1,57 @@ +import pytest +from unittest.mock import Mock, MagicMock, patch +import telegram_bot_alerts.setup_alerts as tba + +# Фікстура для мок-повідомлення +@pytest.fixture +def mock_message(): + mock_chat = MagicMock() + mock_chat.id = 123 + message = MagicMock() + message.chat = mock_chat + message.message_id = 456 + message.text = "▶️ Start" + return message + + +@patch("telegram_bot_alerts.setup_alerts.gcm") +@patch("telegram_bot_alerts.setup_alerts.bot") +def test_sen_containers_stats_success(bot_mock, gcm_mock, mock_message): + gcm_mock.return_value = ["Container stats line 1", "Container stats line 2"] + tba.sen_containers_stats(mock_message) + bot_mock.reply_to.assert_called() + called_text = bot_mock.reply_to.call_args[0][1] + assert "Container stats line 1" in called_text + assert "Container stats line 2" in called_text + + +@patch("telegram_bot_alerts.setup_alerts.gcs") +@patch("telegram_bot_alerts.setup_alerts.bot") +def test_update_containers_status_real_time_none(bot_mock, gcs_mock): + gcs_mock.return_value = None + with patch("time.sleep", return_value=None): + count = 0 + status = gcs_mock() + if status is None: + count += 1 + assert count == 1 + bot_mock.send_message.assert_not_called() + + +@patch("telegram_bot_alerts.setup_alerts.gcs") +@patch("telegram_bot_alerts.setup_alerts.bot") +def test_update_containers_status_real_time_status(bot_mock, gcs_mock): + gcs_mock.return_value = "High CPU usage" + with patch("time.sleep", return_value=None): + chat_id = 123 + status = gcs_mock() + if status is not None: + bot_mock.send_message(chat_id, status) + bot_mock.send_message.assert_called_once_with(chat_id, "High CPU usage") + + +@patch("telegram_bot_alerts.setup_alerts.bot") +def test_echo_message(bot_mock, mock_message): + tba.echo_message(mock_message) + bot_mock.reply_to.assert_called_once_with(mock_message, mock_message.text) + bot_mock.send_message.assert_called() diff --git a/tests/setup_monitoring_test.py b/tests/setup_monitoring_test.py new file mode 100644 index 0000000..e10d15b --- /dev/null +++ b/tests/setup_monitoring_test.py @@ -0,0 +1,106 @@ +import pytest +from unittest.mock import MagicMock, patch +import monitoring.setup_monitoring + + +def test_calculate_cpu_percent_basic(): + stats = { + "cpu_stats": { + "cpu_usage": {"total_usage": 200, "percpu_usage": [100, 100]}, + "system_cpu_usage": 400, + }, + "precpu_stats": { + "cpu_usage": {"total_usage": 100, "percpu_usage": [50, 50]}, + "system_cpu_usage": 200, + } + } + result = monitoring.setup_monitoring.calculate_cpu_percent(stats) + assert result == 100.0 # (100 / 200) * 2 * 100 = 100% + +def test_calculate_cpu_percent_zero_system_delta(): + stats = { + "cpu_stats": {"cpu_usage": {"total_usage": 200, "percpu_usage": [100,100]}, "system_cpu_usage": 400}, + "precpu_stats": {"cpu_usage": {"total_usage": 100, "percpu_usage": [50,50]}, "system_cpu_usage": 400} + } + result = monitoring.setup_monitoring.calculate_cpu_percent(stats) + assert result == 0.0 + + +@patch("monitoring.setup_monitoring.client") +def test_get_container_metrics(mock_client): + mock_container = MagicMock() + mock_container.name = "test_container" + mock_container.status = "running" + mock_container.image.tags = ["test_image:latest"] + mock_container.attrs = {"Created": "2025-01-01T00:00:00"} + mock_container.stats.return_value = { + "memory_stats": {"usage": 104857600}, + "cpu_stats": { + "cpu_usage": {"total_usage": 200, "percpu_usage": [100,100]}, + "system_cpu_usage": 400 + }, + "precpu_stats": { + "cpu_usage": {"total_usage": 100, "percpu_usage": [50,50]}, + "system_cpu_usage": 200 + } + } + mock_client.containers.list.return_value = [mock_container] + + metrics = monitoring.setup_monitoring.get_container_metrics() + assert len(metrics) == 1 + assert "test_container" in metrics[0] + assert "test_image:latest" in metrics[0] + assert "RAM use: 100.00 MB" in metrics[0] + + +@patch("monitoring.setup_monitoring.client") +def test_get_container_status_real_time_cpu(mock_client): + mock_container = MagicMock() + mock_container.name = "high_cpu" + mock_container.stats.return_value = { + "memory_stats": {"usage": 1048576}, + "cpu_stats": { + "cpu_usage": {"total_usage": 200, "percpu_usage": [100,100]}, + "system_cpu_usage": 200 + }, + "precpu_stats": { + "cpu_usage": {"total_usage": 0, "percpu_usage": [0,0]}, + "system_cpu_usage": 100 + } + } + mock_client.containers.list.return_value = [mock_container] + + status = monitoring.setup_monitoring.get_container_status_real_time() + assert "high_cpu" in status + assert "50%" in status + + +@patch("monitoring.setup_monitoring.client") +def test_get_container_status_real_time_mem(mock_client): + mock_container = MagicMock() + mock_container.name = "high_mem" + mock_container.stats.return_value = { + "memory_stats": {"usage": 600 * 1024 * 1024}, + "cpu_stats": {"cpu_usage": {"total_usage": 0, "percpu_usage": [0,0]}, "system_cpu_usage": 100}, + "precpu_stats": {"cpu_usage": {"total_usage": 0, "percpu_usage": [0,0]}, "system_cpu_usage": 100} + } + mock_client.containers.list.return_value = [mock_container] + + status = monitoring.setup_monitoring.get_container_status_real_time() + assert "high_mem" in status + assert "500 MB" in status + + +@patch("monitoring.setup_monitoring.client") +def test_get_container_status_real_time_none(mock_client): + mock_container = MagicMock() + mock_container.name = "normal" + mock_container.stats.return_value = { + "memory_stats": {"usage": 100 * 1024 * 1024}, + "cpu_stats": {"cpu_usage": {"total_usage": 0, "percpu_usage": [0,0]}, "system_cpu_usage": 100}, + "precpu_stats": {"cpu_usage": {"total_usage": 0, "percpu_usage": [0,0]}, "system_cpu_usage": 100} + } + mock_client.containers.list.return_value = [mock_container] + + status = monitoring.setup_monitoring.get_container_status_real_time() + assert status is None \ No newline at end of file