kir / tests /unit /test_debug_logger.py
jwadow
refactor: update kiro_gateway naming to kiro
f1668d0
Raw
History Blame Contribute Delete
32.2 kB
# -*- coding: utf-8 -*-
"""
Unit-тесты для DebugLogger.
Проверяет логику буферизации и записи debug логов в разных режимах.
"""
import json
import pytest
from pathlib import Path
from unittest.mock import patch, MagicMock
class TestDebugLoggerModeOff:
"""Тесты для режима DEBUG_MODE=off."""
def test_prepare_new_request_does_nothing(self, tmp_path):
"""
Что он делает: Проверяет, что prepare_new_request ничего не делает в режиме off.
Цель: Убедиться, что в режиме off директория не создаётся.
"""
print("Настройка: Режим off...")
with patch('kiro.debug_logger.DEBUG_MODE', 'off'):
with patch('kiro.debug_logger.DEBUG_DIR', str(tmp_path / "debug_logs")):
# Пересоздаём экземпляр с новыми настройками
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = tmp_path / "debug_logs"
print("Действие: Вызов prepare_new_request...")
logger.prepare_new_request()
print(f"Проверяем, что директория не создана...")
assert not (tmp_path / "debug_logs").exists()
def test_log_request_body_does_nothing(self, tmp_path):
"""
Что он делает: Проверяет, что log_request_body ничего не делает в режиме off.
Цель: Убедиться, что данные не записываются.
"""
print("Настройка: Режим off...")
with patch('kiro.debug_logger.DEBUG_MODE', 'off'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = tmp_path / "debug_logs"
print("Действие: Вызов log_request_body...")
logger.log_request_body(b'{"test": "data"}')
print(f"Проверяем, что файл не создан...")
assert not (tmp_path / "debug_logs" / "request_body.json").exists()
class TestDebugLoggerModeAll:
"""Тесты для режима DEBUG_MODE=all."""
def test_prepare_new_request_clears_directory(self, tmp_path):
"""
Что он делает: Проверяет, что prepare_new_request очищает директорию в режиме all.
Цель: Убедиться, что старые логи удаляются.
"""
print("Настройка: Режим all, создаём старый файл...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
old_file = debug_dir / "old_file.txt"
old_file.write_text("old content")
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов prepare_new_request...")
logger.prepare_new_request()
print(f"Проверяем, что старый файл удалён...")
assert not old_file.exists()
print(f"Проверяем, что директория существует...")
assert debug_dir.exists()
def test_log_request_body_writes_immediately(self, tmp_path):
"""
Что он делает: Проверяет, что log_request_body пишет сразу в файл в режиме all.
Цель: Убедиться, что данные записываются немедленно.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_request_body...")
test_data = b'{"model": "test", "messages": []}'
logger.log_request_body(test_data)
print(f"Проверяем, что файл создан...")
file_path = debug_dir / "request_body.json"
assert file_path.exists()
print(f"Проверяем содержимое файла...")
content = json.loads(file_path.read_text())
assert content["model"] == "test"
def test_log_kiro_request_body_writes_immediately(self, tmp_path):
"""
Что он делает: Проверяет, что log_kiro_request_body пишет сразу в файл в режиме all.
Цель: Убедиться, что Kiro payload записывается немедленно.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_kiro_request_body...")
test_data = b'{"conversationState": {}}'
logger.log_kiro_request_body(test_data)
print(f"Проверяем, что файл создан...")
file_path = debug_dir / "kiro_request_body.json"
assert file_path.exists()
def test_log_raw_chunk_appends_to_file(self, tmp_path):
"""
Что он делает: Проверяет, что log_raw_chunk дописывает в файл в режиме all.
Цель: Убедиться, что чанки накапливаются.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_raw_chunk дважды...")
logger.log_raw_chunk(b'chunk1')
logger.log_raw_chunk(b'chunk2')
print(f"Проверяем содержимое файла...")
file_path = debug_dir / "response_stream_raw.txt"
content = file_path.read_bytes()
assert content == b'chunk1chunk2'
class TestDebugLoggerModeErrors:
"""Тесты для режима DEBUG_MODE=errors."""
def test_log_request_body_buffers_data(self, tmp_path):
"""
Что он делает: Проверяет, что log_request_body буферизует данные в режиме errors.
Цель: Убедиться, что данные не записываются сразу.
"""
print("Настройка: Режим errors...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_request_body...")
test_data = b'{"test": "buffered"}'
logger.log_request_body(test_data)
print(f"Проверяем, что файл НЕ создан...")
assert not debug_dir.exists()
print(f"Проверяем, что данные в буфере...")
assert logger._request_body_buffer == test_data
def test_flush_on_error_writes_buffers(self, tmp_path):
"""
Что он делает: Проверяет, что flush_on_error записывает буферы в файлы.
Цель: Убедиться, что при ошибке данные сохраняются.
"""
print("Настройка: Режим errors, заполняем буферы...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
# Заполняем буферы
logger.log_request_body(b'{"request": "body"}')
logger.log_kiro_request_body(b'{"kiro": "request"}')
logger.log_raw_chunk(b'raw_chunk')
logger.log_modified_chunk(b'modified_chunk')
print("Действие: Вызов flush_on_error...")
logger.flush_on_error(400, "Bad Request")
print(f"Проверяем, что все файлы созданы...")
assert (debug_dir / "request_body.json").exists()
assert (debug_dir / "kiro_request_body.json").exists()
assert (debug_dir / "response_stream_raw.txt").exists()
assert (debug_dir / "response_stream_modified.txt").exists()
assert (debug_dir / "error_info.json").exists()
print(f"Проверяем error_info.json...")
error_info = json.loads((debug_dir / "error_info.json").read_text())
assert error_info["status_code"] == 400
assert error_info["error_message"] == "Bad Request"
def test_flush_on_error_clears_buffers(self, tmp_path):
"""
Что он делает: Проверяет, что flush_on_error очищает буферы после записи.
Цель: Убедиться, что буферы не накапливаются между запросами.
"""
print("Настройка: Режим errors...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
logger.log_request_body(b'{"test": "data"}')
print("Действие: Вызов flush_on_error...")
logger.flush_on_error(500, "Error")
print(f"Проверяем, что буферы очищены...")
assert logger._request_body_buffer is None
assert logger._kiro_request_body_buffer is None
assert len(logger._raw_chunks_buffer) == 0
assert len(logger._modified_chunks_buffer) == 0
def test_discard_buffers_clears_without_writing(self, tmp_path):
"""
Что он делает: Проверяет, что discard_buffers очищает буферы без записи.
Цель: Убедиться, что успешные запросы не оставляют логов.
"""
print("Настройка: Режим errors, заполняем буферы...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
logger.log_request_body(b'{"test": "data"}')
logger.log_raw_chunk(b'chunk')
print("Действие: Вызов discard_buffers...")
logger.discard_buffers()
print(f"Проверяем, что директория НЕ создана...")
assert not debug_dir.exists()
print(f"Проверяем, что буферы очищены...")
assert logger._request_body_buffer is None
assert len(logger._raw_chunks_buffer) == 0
def test_flush_on_error_writes_error_info_in_mode_all(self, tmp_path):
"""
Что он делает: Проверяет, что flush_on_error записывает error_info.json в режиме all.
Цель: Убедиться, что информация об ошибке сохраняется в обоих режимах.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов flush_on_error...")
logger.flush_on_error(400, "Bad Request")
print(f"Проверяем, что error_info.json создан...")
assert (debug_dir / "error_info.json").exists()
print(f"Проверяем содержимое error_info.json...")
error_info = json.loads((debug_dir / "error_info.json").read_text())
assert error_info["status_code"] == 400
assert error_info["error_message"] == "Bad Request"
class TestDebugLoggerLogErrorInfo:
"""Тесты для метода log_error_info()."""
def test_log_error_info_writes_in_mode_all(self, tmp_path):
"""
Что он делает: Проверяет, что log_error_info записывает файл в режиме all.
Цель: Убедиться, что error_info.json создаётся при ошибках.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_error_info...")
logger.log_error_info(500, "Internal Server Error")
print(f"Проверяем, что error_info.json создан...")
error_file = debug_dir / "error_info.json"
assert error_file.exists()
print(f"Проверяем содержимое...")
error_info = json.loads(error_file.read_text())
assert error_info["status_code"] == 500
assert error_info["error_message"] == "Internal Server Error"
def test_log_error_info_writes_in_mode_errors(self, tmp_path):
"""
Что он делает: Проверяет, что log_error_info записывает файл в режиме errors.
Цель: Убедиться, что метод работает в обоих режимах.
"""
print("Настройка: Режим errors...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_error_info...")
logger.log_error_info(404, "Not Found")
print(f"Проверяем, что error_info.json создан...")
error_file = debug_dir / "error_info.json"
assert error_file.exists()
def test_log_error_info_does_nothing_in_mode_off(self, tmp_path):
"""
Что он делает: Проверяет, что log_error_info ничего не делает в режиме off.
Цель: Убедиться, что в режиме off файлы не создаются.
"""
print("Настройка: Режим off...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'off'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_error_info...")
logger.log_error_info(500, "Error")
print(f"Проверяем, что директория НЕ создана...")
assert not debug_dir.exists()
class TestDebugLoggerHelperMethods:
"""Тесты для вспомогательных методов DebugLogger."""
def test_is_enabled_returns_true_for_errors(self):
"""
Что он делает: Проверяет _is_enabled() для режима errors.
Цель: Убедиться, что режим errors считается включённым.
"""
print("Настройка: Режим errors...")
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
print(f"Проверяем _is_enabled()...")
assert logger._is_enabled() is True
def test_is_enabled_returns_true_for_all(self):
"""
Что он делает: Проверяет _is_enabled() для режима all.
Цель: Убедиться, что режим all считается включённым.
"""
print("Настройка: Режим all...")
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
print(f"Проверяем _is_enabled()...")
assert logger._is_enabled() is True
def test_is_enabled_returns_false_for_off(self):
"""
Что он делает: Проверяет _is_enabled() для режима off.
Цель: Убедиться, что режим off считается выключенным.
"""
print("Настройка: Режим off...")
with patch('kiro.debug_logger.DEBUG_MODE', 'off'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
print(f"Проверяем _is_enabled()...")
assert logger._is_enabled() is False
def test_is_immediate_write_returns_true_for_all(self):
"""
Что он делает: Проверяет _is_immediate_write() для режима all.
Цель: Убедиться, что режим all пишет сразу.
"""
print("Настройка: Режим all...")
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
print(f"Проверяем _is_immediate_write()...")
assert logger._is_immediate_write() is True
def test_is_immediate_write_returns_false_for_errors(self):
"""
Что он делает: Проверяет _is_immediate_write() для режима errors.
Цель: Убедиться, что режим errors буферизует.
"""
print("Настройка: Режим errors...")
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
print(f"Проверяем _is_immediate_write()...")
assert logger._is_immediate_write() is False
class TestDebugLoggerJsonHandling:
"""Тесты для обработки JSON в DebugLogger."""
def test_log_request_body_formats_json_pretty(self, tmp_path):
"""
Что он делает: Проверяет, что JSON форматируется красиво.
Цель: Убедиться, что JSON читаем в файле.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_request_body с JSON...")
logger.log_request_body(b'{"key":"value"}')
print(f"Проверяем форматирование...")
content = (debug_dir / "request_body.json").read_text()
# Должен быть отформатирован с отступами
assert " " in content or "\n" in content
def test_log_request_body_handles_invalid_json(self, tmp_path):
"""
Что он делает: Проверяет обработку невалидного JSON.
Цель: Убедиться, что невалидный JSON записывается как есть.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
logger = DebugLogger.__new__(DebugLogger)
logger._initialized = False
logger.__init__()
logger.debug_dir = debug_dir
print("Действие: Вызов log_request_body с невалидным JSON...")
invalid_data = b'not a json {{'
logger.log_request_body(invalid_data)
print(f"Проверяем, что данные записаны как есть...")
content = (debug_dir / "request_body.json").read_bytes()
assert content == invalid_data
class TestDebugLoggerAppLogsCapture:
"""Тесты для захвата логов приложения (app_logs.txt)."""
def test_prepare_new_request_sets_up_log_capture(self, tmp_path):
"""
Что он делает: Проверяет, что prepare_new_request настраивает захват логов.
Цель: Убедиться, что sink для логов создаётся.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
dbg_logger = DebugLogger.__new__(DebugLogger)
dbg_logger._initialized = False
dbg_logger.__init__()
dbg_logger.debug_dir = debug_dir
print("Действие: Вызов prepare_new_request...")
dbg_logger.prepare_new_request()
print(f"Проверяем, что sink создан...")
assert dbg_logger._loguru_sink_id is not None
# Очистка
dbg_logger._clear_app_logs_buffer()
def test_flush_on_error_writes_app_logs_in_mode_errors(self, tmp_path):
"""
Что он делает: Проверяет, что flush_on_error записывает app_logs.txt в режиме errors.
Цель: Убедиться, что логи приложения сохраняются при ошибках.
"""
print("Настройка: Режим errors...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
from loguru import logger as loguru_logger
dbg_logger = DebugLogger.__new__(DebugLogger)
dbg_logger._initialized = False
dbg_logger.__init__()
dbg_logger.debug_dir = debug_dir
# Настраиваем захват логов
dbg_logger.prepare_new_request()
# Добавляем данные в буфер чтобы flush сработал
dbg_logger.log_request_body(b'{"test": "data"}')
# Пишем тестовый лог напрямую в буфер (имитация)
dbg_logger._app_logs_buffer.write("Test log message\n")
print("Действие: Вызов flush_on_error...")
dbg_logger.flush_on_error(500, "Test Error")
print(f"Проверяем, что app_logs.txt создан...")
app_logs_file = debug_dir / "app_logs.txt"
assert app_logs_file.exists()
print(f"Проверяем содержимое...")
content = app_logs_file.read_text()
assert "Test log message" in content
def test_discard_buffers_saves_logs_in_mode_all(self, tmp_path):
"""
Что он делает: Проверяет, что discard_buffers сохраняет логи в режиме all.
Цель: Убедиться, что даже успешные запросы сохраняют логи в режиме all.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
dbg_logger = DebugLogger.__new__(DebugLogger)
dbg_logger._initialized = False
dbg_logger.__init__()
dbg_logger.debug_dir = debug_dir
# Настраиваем захват логов
dbg_logger.prepare_new_request()
# Пишем тестовый лог напрямую в буфер
dbg_logger._app_logs_buffer.write("Success log message\n")
print("Действие: Вызов discard_buffers...")
dbg_logger.discard_buffers()
print(f"Проверяем, что app_logs.txt создан...")
app_logs_file = debug_dir / "app_logs.txt"
assert app_logs_file.exists()
print(f"Проверяем содержимое...")
content = app_logs_file.read_text()
assert "Success log message" in content
def test_discard_buffers_does_not_save_logs_in_mode_errors(self, tmp_path):
"""
Что он делает: Проверяет, что discard_buffers НЕ сохраняет логи в режиме errors.
Цель: Убедиться, что успешные запросы не оставляют логов в режиме errors.
"""
print("Настройка: Режим errors...")
debug_dir = tmp_path / "debug_logs"
with patch('kiro.debug_logger.DEBUG_MODE', 'errors'):
from kiro.debug_logger import DebugLogger
dbg_logger = DebugLogger.__new__(DebugLogger)
dbg_logger._initialized = False
dbg_logger.__init__()
dbg_logger.debug_dir = debug_dir
# Настраиваем захват логов
dbg_logger.prepare_new_request()
# Пишем тестовый лог напрямую в буфер
dbg_logger._app_logs_buffer.write("Should not be saved\n")
print("Действие: Вызов discard_buffers...")
dbg_logger.discard_buffers()
print(f"Проверяем, что директория НЕ создана...")
assert not debug_dir.exists()
def test_clear_app_logs_buffer_removes_sink(self, tmp_path):
"""
Что он делает: Проверяет, что _clear_app_logs_buffer удаляет sink.
Цель: Убедиться, что sink корректно удаляется.
"""
print("Настройка: Режим all...")
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
dbg_logger = DebugLogger.__new__(DebugLogger)
dbg_logger._initialized = False
dbg_logger.__init__()
dbg_logger.debug_dir = tmp_path / "debug_logs"
# Настраиваем захват логов
dbg_logger.prepare_new_request()
sink_id = dbg_logger._loguru_sink_id
assert sink_id is not None
print("Действие: Вызов _clear_app_logs_buffer...")
dbg_logger._clear_app_logs_buffer()
print(f"Проверяем, что sink_id сброшен...")
assert dbg_logger._loguru_sink_id is None
def test_app_logs_not_saved_when_empty(self, tmp_path):
"""
Что он делает: Проверяет, что пустые логи не создают файл.
Цель: Убедиться, что app_logs.txt не создаётся если логов нет.
"""
print("Настройка: Режим all...")
debug_dir = tmp_path / "debug_logs"
debug_dir.mkdir()
with patch('kiro.debug_logger.DEBUG_MODE', 'all'):
from kiro.debug_logger import DebugLogger
dbg_logger = DebugLogger.__new__(DebugLogger)
dbg_logger._initialized = False
dbg_logger.__init__()
dbg_logger.debug_dir = debug_dir
# НЕ пишем ничего в буфер
print("Действие: Вызов _write_app_logs_to_file...")
dbg_logger._write_app_logs_to_file()
print(f"Проверяем, что app_logs.txt НЕ создан...")
app_logs_file = debug_dir / "app_logs.txt"
assert not app_logs_file.exists()