Home » Как писать юнит-тесты в Python с использованием unittest

Как писать юнит-тесты в Python с использованием unittest

Если ты админ сервера и пишешь скрипты для автоматизации, то тестирование кода — это не просто хорошая практика, а необходимость. Особенно когда твои скрипты управляют критической инфраструктурой. Представь: развернул новый сервис, запустил скрипт мониторинга, а он падает в продакшене из-за банальной ошибки в логике. Знакомо? Юнит-тесты в Python помогут избежать таких ситуаций и сделают твой код более надёжным. В этой статье разберём unittest — встроенный фреймворк для тестирования Python, который не требует дополнительных зависимостей и отлично подходит для серверных скриптов.

Как работает unittest и зачем он нужен

unittest — это стандартный модуль Python, вдохновлённый JUnit из мира Java. Основная идея: каждая функция или метод должны быть покрыты отдельными тестами, которые проверяют их поведение в различных сценариях. Это особенно важно для серверного кода, где ошибка может привести к простою сервиса.

Фреймворк работает по принципу “test fixtures” — создаёт изолированную среду для каждого теста, выполняет его и очищает за собой. Это гарантирует, что тесты не влияют друг на друга.

Быстрая настройка и первые шаги

Хорошая новость — unittest уже встроен в Python, никаких pip install не нужно. Создай файл test_server_utils.py и начнём с базового примера:

import unittest
import sys
import os

# Добавляем путь к модулю, если он не в PYTHONPATH
sys.path.insert(0, os.path.abspath('.'))

from server_utils import check_port_availability, parse_config_file

class TestServerUtils(unittest.TestCase):
    
    def setUp(self):
        """Выполняется перед каждым тестом"""
        self.test_config = {
            'port': 8080,
            'host': '127.0.0.1',
            'timeout': 30
        }
    
    def tearDown(self):
        """Выполняется после каждого теста"""
        # Очистка временных файлов, закрытие соединений и т.д.
        pass
    
    def test_port_availability_success(self):
        """Тест проверки доступности порта"""
        result = check_port_availability('127.0.0.1', 22)
        self.assertTrue(result)
    
    def test_port_availability_failure(self):
        """Тест недоступного порта"""
        result = check_port_availability('127.0.0.1', 99999)
        self.assertFalse(result)
    
    def test_config_parsing_valid(self):
        """Тест парсинга корректного config файла"""
        config = parse_config_file('valid_config.json')
        self.assertIsNotNone(config)
        self.assertIn('port', config)
        self.assertEqual(config['port'], 8080)
    
    def test_config_parsing_invalid(self):
        """Тест обработки некорректного config файла"""
        with self.assertRaises(ValueError):
            parse_config_file('invalid_config.json')

if __name__ == '__main__':
    unittest.main()

Запуск тестов:

# Запуск всех тестов в файле
python test_server_utils.py

# Запуск конкретного теста
python -m unittest test_server_utils.TestServerUtils.test_port_availability_success

# Запуск с подробным выводом
python -m unittest -v test_server_utils.py

# Запуск всех тестов в директории
python -m unittest discover -s tests -p "test_*.py"

Практические примеры и кейсы

Вот реальные примеры тестов для серверных задач:

import unittest
import tempfile
import json
import subprocess
import socket
from unittest.mock import patch, MagicMock

class TestServerManagement(unittest.TestCase):
    
    def test_service_status_check(self):
        """Проверка статуса сервиса"""
        def get_service_status(service_name):
            try:
                result = subprocess.run(['systemctl', 'is-active', service_name], 
                                      capture_output=True, text=True)
                return result.stdout.strip() == 'active'
            except:
                return False
        
        # Тестируем существующий сервис
        status = get_service_status('sshd')
        self.assertIsInstance(status, bool)
    
    def test_disk_space_monitoring(self):
        """Тест мониторинга дискового пространства"""
        def check_disk_space(path, threshold=90):
            import shutil
            total, used, free = shutil.disk_usage(path)
            percent_used = (used / total) * 100
            return percent_used < threshold
        
        result = check_disk_space('/')
        self.assertIsInstance(result, bool)
        
        # Тест с нереальным порогом
        result = check_disk_space('/', threshold=0)
        self.assertFalse(result)
    
    @patch('socket.socket')
    def test_port_scanner_mock(self, mock_socket):
        """Тест сканера портов с моками"""
        mock_sock = MagicMock()
        mock_socket.return_value = mock_sock
        mock_sock.connect_ex.return_value = 0  # Успешное соединение
        
        def scan_port(host, port):
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            result = sock.connect_ex((host, port))
            sock.close()
            return result == 0
        
        result = scan_port('127.0.0.1', 80)
        self.assertTrue(result)
        mock_sock.connect_ex.assert_called_with(('127.0.0.1', 80))
    
    def test_config_validation(self):
        """Тест валидации конфигурационных файлов"""
        def validate_server_config(config):
            required_fields = ['host', 'port', 'workers']
            for field in required_fields:
                if field not in config:
                    raise ValueError(f"Missing required field: {field}")
            
            if not isinstance(config['port'], int) or config['port'] <= 0:
                raise ValueError("Port must be a positive integer")
            
            return True
        
        # Валидный config
        valid_config = {'host': '0.0.0.0', 'port': 8080, 'workers': 4}
        self.assertTrue(validate_server_config(valid_config))
        
        # Невалидный config
        invalid_config = {'host': '0.0.0.0', 'port': 'invalid'}
        with self.assertRaises(ValueError):
            validate_server_config(invalid_config)
    
    def test_log_parsing(self):
        """Тест парсинга логов"""
        def parse_access_log(log_line):
            import re
            pattern = r'(\d+\.\d+\.\d+\.\d+) - - \[(.*?)\] "(.*?)" (\d+) (\d+)'
            match = re.match(pattern, log_line)
            if match:
                return {
                    'ip': match.group(1),
                    'timestamp': match.group(2),
                    'request': match.group(3),
                    'status': int(match.group(4)),
                    'size': int(match.group(5))
                }
            return None
        
        log_line = '192.168.1.1 - - [01/Jan/2024:12:00:00 +0000] "GET /index.html HTTP/1.1" 200 1234'
        result = parse_access_log(log_line)
        
        self.assertIsNotNone(result)
        self.assertEqual(result['ip'], '192.168.1.1')
        self.assertEqual(result['status'], 200)
        self.assertEqual(result['size'], 1234)

Assertion методы и их использование

unittest предоставляет множество assertion методов для разных типов проверок:

Метод	Проверяет	Пример использования
assertEqual(a, b)	a == b	self.assertEqual(response.status_code, 200)
assertNotEqual(a, b)	a != b	self.assertNotEqual(old_config, new_config)
assertTrue(x)	bool(x) is True	self.assertTrue(is_service_running('nginx'))
assertFalse(x)	bool(x) is False	self.assertFalse(has_errors)
assertIn(a, b)	a in b	self.assertIn('ERROR', log_content)
assertIsNone(x)	x is None	self.assertIsNone(get_user_by_id(999))
assertRaises(exc, fun, *args)	fun(*args) raises exc	self.assertRaises(ConnectionError, connect_to_db)
assertGreater(a, b)	a > b	self.assertGreater(free_memory, 1024)

Мокирование и тестирование внешних зависимостей

При тестировании серверных скриптов часто нужно имитировать внешние системы — базы данных, API, файловую систему. Для этого используй модуль mock:

import unittest
from unittest.mock import patch, MagicMock, mock_open
import requests

class TestExternalServices(unittest.TestCase):
    
    @patch('requests.get')
    def test_api_health_check(self, mock_get):
        """Тест проверки здоровья API"""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {'status': 'ok'}
        mock_get.return_value = mock_response
        
        def check_api_health(url):
            response = requests.get(url)
            return response.status_code == 200 and response.json().get('status') == 'ok'
        
        result = check_api_health('http://api.example.com/health')
        self.assertTrue(result)
        mock_get.assert_called_once_with('http://api.example.com/health')
    
    @patch('builtins.open', new_callable=mock_open, read_data='log line 1\nlog line 2\n')
    def test_log_reader(self, mock_file):
        """Тест чтения лог-файла"""
        def count_log_lines(filename):
            with open(filename, 'r') as f:
                return len(f.readlines())
        
        result = count_log_lines('/var/log/app.log')
        self.assertEqual(result, 2)
        mock_file.assert_called_once_with('/var/log/app.log', 'r')
    
    @patch('subprocess.run')
    def test_system_command_execution(self, mock_run):
        """Тест выполнения системных команд"""
        mock_run.return_value.returncode = 0
        mock_run.return_value.stdout = 'nginx is running'
        
        def check_service_status(service):
            import subprocess
            result = subprocess.run(['systemctl', 'status', service], 
                                  capture_output=True, text=True)
            return result.returncode == 0
        
        result = check_service_status('nginx')
        self.assertTrue(result)
        mock_run.assert_called_once_with(['systemctl', 'status', 'nginx'], 
                                        capture_output=True, text=True)

Организация тестов и best practices

Для больших проектов создай структуру:

project/
├── server_utils/
│   ├── __init__.py
│   ├── config.py
│   ├── monitoring.py
│   └── deployment.py
├── tests/
│   ├── __init__.py
│   ├── test_config.py
│   ├── test_monitoring.py
│   └── test_deployment.py
├── requirements.txt
└── setup.py

Создай базовый класс для общих настроек:

import unittest
import tempfile
import os

class BaseServerTest(unittest.TestCase):
    """Базовый класс для всех серверных тестов"""
    
    def setUp(self):
        """Общие настройки для всех тестов"""
        self.temp_dir = tempfile.mkdtemp()
        self.original_cwd = os.getcwd()
        os.chdir(self.temp_dir)
    
    def tearDown(self):
        """Очистка после тестов"""
        os.chdir(self.original_cwd)
        import shutil
        shutil.rmtree(self.temp_dir)
    
    def create_temp_file(self, content, filename='test_file.txt'):
        """Утилита для создания временных файлов"""
        filepath = os.path.join(self.temp_dir, filename)
        with open(filepath, 'w') as f:
            f.write(content)
        return filepath

class TestMonitoring(BaseServerTest):
    """Тесты системы мониторинга"""
    
    def test_cpu_usage_monitoring(self):
        """Тест мониторинга CPU"""
        def get_cpu_usage():
            import psutil
            return psutil.cpu_percent(interval=1)
        
        cpu_usage = get_cpu_usage()
        self.assertGreaterEqual(cpu_usage, 0)
        self.assertLessEqual(cpu_usage, 100)

Интеграция с CI/CD и автоматизация

Для автоматического запуска тестов при деплое создай скрипт:

#!/bin/bash
# test_runner.sh

echo "Запуск юнит-тестов..."

# Установка зависимостей
pip install -r requirements.txt

# Запуск тестов с отчётом о покрытии
python -m unittest discover -s tests -p "test_*.py" -v

# Проверка кода статическими анализаторами
if command -v flake8 &> /dev/null; then
    echo "Проверка стиля кода..."
    flake8 server_utils tests
fi

# Запуск тестов производительности
if [ -f "performance_tests.py" ]; then
    echo "Запуск performance тестов..."
    python performance_tests.py
fi

echo "Все тесты завершены!"

Сравнение с другими фреймворками

Фреймворк	Плюсы	Минусы	Когда использовать
unittest	Встроен в Python, OOP-подход	Многословный синтаксис	Корпоративные проекты, строгая структура
pytest	Простой синтаксис, мощные фикстуры	Дополнительная зависимость	Современная разработка, TDD
nose2	Расширение unittest	Менее популярен	Миграция с nose
doctest	Тесты в документации	Ограниченные возможности	Простые примеры в документации

Продвинутые техники и интеграции

Для серверных задач особенно полезны тесты с временными ограничениями и параллельным выполнением:

import unittest
import time
import threading
from concurrent.futures import ThreadPoolExecutor

class TestServerPerformance(unittest.TestCase):
    
    def test_function_timeout(self):
        """Тест с ограничением времени выполнения"""
        def slow_function():
            time.sleep(2)
            return "completed"
        
        start_time = time.time()
        result = slow_function()
        execution_time = time.time() - start_time
        
        self.assertEqual(result, "completed")
        self.assertLess(execution_time, 3.0)  # Не более 3 секунд
    
    def test_concurrent_connections(self):
        """Тест обработки множественных соединений"""
        def handle_connection(connection_id):
            time.sleep(0.1)  # Имитация обработки
            return f"processed_{connection_id}"
        
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [executor.submit(handle_connection, i) for i in range(100)]
            results = [f.result() for f in futures]
        
        self.assertEqual(len(results), 100)
        self.assertTrue(all(r.startswith("processed_") for r in results))
    
    def test_memory_usage(self):
        """Тест контроля памяти"""
        import psutil
        import os
        
        process = psutil.Process(os.getpid())
        initial_memory = process.memory_info().rss
        
        # Выполняем операцию, которая может потреблять много памяти
        big_list = [i for i in range(10000)]
        
        current_memory = process.memory_info().rss
        memory_increase = current_memory - initial_memory
        
        # Проверяем, что прирост памяти не критичен
        self.assertLess(memory_increase, 50 * 1024 * 1024)  # Менее 50MB
        
        del big_list  # Освобождаем память

Интеграция с мониторингом и логированием

Для серверных приложений важно тестировать интеграцию с системами мониторинга:

import unittest
import logging
from unittest.mock import patch
import json

class TestLoggingIntegration(unittest.TestCase):
    
    def setUp(self):
        """Настройка логирования для тестов"""
        self.logger = logging.getLogger('test_logger')
        self.logger.setLevel(logging.DEBUG)
        
        # Создаём in-memory handler для тестов
        self.log_capture = []
        
        class ListHandler(logging.Handler):
            def __init__(self, log_list):
                super().__init__()
                self.log_list = log_list
                
            def emit(self, record):
                self.log_list.append(self.format(record))
        
        handler = ListHandler(self.log_capture)
        handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
        self.logger.addHandler(handler)
    
    def test_error_logging(self):
        """Тест логирования ошибок"""
        def process_request(data):
            try:
                result = json.loads(data)
                self.logger.info(f"Processed request: {result}")
                return result
            except json.JSONDecodeError as e:
                self.logger.error(f"Invalid JSON: {e}")
                raise
        
        # Тест успешной обработки
        result = process_request('{"key": "value"}')
        self.assertEqual(result, {"key": "value"})
        self.assertTrue(any("INFO: Processed request" in log for log in self.log_capture))
        
        # Тест обработки ошибки
        with self.assertRaises(json.JSONDecodeError):
            process_request('invalid json')
        
        self.assertTrue(any("ERROR: Invalid JSON" in log for log in self.log_capture))
    
    @patch('requests.post')
    def test_metrics_reporting(self, mock_post):
        """Тест отправки метрик"""
        mock_post.return_value.status_code = 200
        
        def send_metrics(metric_name, value):
            import requests
            payload = {
                'metric': metric_name,
                'value': value,
                'timestamp': time.time()
            }
            response = requests.post('http://metrics.example.com/api', json=payload)
            return response.status_code == 200
        
        result = send_metrics('cpu_usage', 75.5)
        self.assertTrue(result)
        mock_post.assert_called_once()

Тестирование в контейнерах и изолированных средах

Если развёртываешь на VPS или выделенных серверах, полезно тестировать в изолированных средах:

import unittest
import docker
import tempfile
import os

class TestDockerIntegration(unittest.TestCase):
    
    def setUp(self):
        """Настройка Docker для тестов"""
        try:
            self.docker_client = docker.from_env()
            self.container = None
        except Exception as e:
            self.skipTest(f"Docker недоступен: {e}")
    
    def tearDown(self):
        """Очистка Docker контейнеров"""
        if self.container:
            try:
                self.container.remove(force=True)
            except:
                pass
    
    def test_service_in_container(self):
        """Тест сервиса в Docker контейнере"""
        # Создаём простой Dockerfile
        dockerfile_content = """
        FROM python:3.9-slim
        WORKDIR /app
        COPY . .
        RUN pip install flask
        CMD ["python", "app.py"]
        """
        
        with tempfile.TemporaryDirectory() as temp_dir:
            dockerfile_path = os.path.join(temp_dir, 'Dockerfile')
            app_path = os.path.join(temp_dir, 'app.py')
            
            with open(dockerfile_path, 'w') as f:
                f.write(dockerfile_content)
            
            with open(app_path, 'w') as f:
                f.write("""
from flask import Flask
app = Flask(__name__)

@app.route('/health')
def health():
    return {'status': 'ok'}

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)
""")
            
            # Собираем образ
            image, logs = self.docker_client.images.build(path=temp_dir, tag='test-app')
            
            # Запускаем контейнер
            self.container = self.docker_client.containers.run(
                image.id,
                ports={'5000/tcp': 5000},
                detach=True
            )
            
            # Ждём запуска
            import time
            time.sleep(5)
            
            # Проверяем доступность
            import requests
            try:
                response = requests.get('http://localhost:5000/health', timeout=10)
                self.assertEqual(response.status_code, 200)
                self.assertEqual(response.json()['status'], 'ok')
            except requests.exceptions.RequestException:
                self.fail("Сервис недоступен в контейнере")

Полезные факты и нестандартные применения

Несколько интересных возможностей unittest, которые могут пригодиться:

Условные тесты: используй @unittest.skipIf для пропуска тестов в зависимости от условий
Параметризованные тесты: создавай тесты с разными входными данными через subTest
Тестирование CLI: unittest отлично подходит для тестирования command-line утилит
Интеграция с IDE: большинство IDE поддерживают unittest из коробки

import unittest
import sys
import subprocess

class TestAdvancedFeatures(unittest.TestCase):
    
    @unittest.skipIf(sys.platform == "win32", "Не поддерживается на Windows")
    def test_unix_specific_feature(self):
        """Тест только для Unix-систем"""
        result = subprocess.run(['ps', 'aux'], capture_output=True, text=True)
        self.assertEqual(result.returncode, 0)
    
    def test_multiple_scenarios(self):
        """Тест с несколькими сценариями"""
        test_cases = [
            ('127.0.0.1', True),
            ('256.256.256.256', False),
            ('localhost', True),
            ('', False)
        ]
        
        def is_valid_host(host):
            import socket
            try:
                socket.gethostbyname(host)
                return True
            except socket.gaierror:
                return False
        
        for host, expected in test_cases:
            with self.subTest(host=host):
                result = is_valid_host(host)
                self.assertEqual(result, expected)
    
    def test_cli_tool(self):
        """Тест command-line утилиты"""
        # Создаём простую CLI утилиту
        cli_script = """
import sys
import argparse

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--port', type=int, default=8080)
    parser.add_argument('--host', default='localhost')
    
    args = parser.parse_args()
    print(f"Server: {args.host}:{args.port}")
    return 0

if __name__ == '__main__':
    sys.exit(main())
"""
        
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(cli_script)
            script_path = f.name
        
        try:
            # Тестируем CLI
            result = subprocess.run([sys.executable, script_path, '--port', '9000'], 
                                  capture_output=True, text=True)
            
            self.assertEqual(result.returncode, 0)
            self.assertIn('Server: localhost:9000', result.stdout)
        finally:
            os.unlink(script_path)

Заключение и рекомендации

unittest — это мощный инструмент, который должен быть в арсенале каждого системного администратора и DevOps-инженера. Особенно полезен для:

Тестирования deployment скриптов — убедись, что твои скрипты работают корректно перед продакшеном
Валидации конфигураций — проверяй корректность настроек сервисов
Мониторинга системы — создавай тесты для проверки состояния сервисов
API интеграций — тестируй взаимодействие с внешними сервисами

Начни с простых тестов для критически важных функций и постепенно расширяй покрытие. Помни: лучше иметь несколько хорошо написанных тестов, чем много плохих. Интегрируй тесты в CI/CD pipeline и запускай их при каждом деплое.

Для серверных приложений особенно важно тестировать edge cases — некорректные входные данные, сетевые ошибки, нехватку ресурсов. unittest с его assertion методами и возможностями мокирования отлично подходит для таких задач.

Дополнительные ресурсы:

В этой статье собрана информация и материалы из различных интернет-источников. Мы признаем и ценим работу всех оригинальных авторов, издателей и веб-сайтов. Несмотря на то, что были приложены все усилия для надлежащего указания исходного материала, любая непреднамеренная оплошность или упущение не являются нарушением авторских прав. Все упомянутые товарные знаки, логотипы и изображения являются собственностью соответствующих владельцев. Если вы считаете, что какой-либо контент, использованный в этой статье, нарушает ваши авторские права, немедленно свяжитесь с нами для рассмотрения и принятия оперативных мер.

Данная статья предназначена исключительно для ознакомительных и образовательных целей и не ущемляет права правообладателей. Если какой-либо материал, защищенный авторским правом, был использован без должного упоминания или с нарушением законов об авторском праве, это непреднамеренно, и мы исправим это незамедлительно после уведомления. Обратите внимание, что переиздание, распространение или воспроизведение части или всего содержимого в любой форме запрещено без письменного разрешения автора и владельца веб-сайта. Для получения разрешений или дополнительных запросов, пожалуйста, свяжитесь с нами.