| |
|
|
| """ |
| 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() |
| |
| |
| 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() |