mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-05-28 01:47:32 +08:00
27 lines
655 B
Python
27 lines
655 B
Python
import io
|
|
|
|
from app.logger import LogInterceptor
|
|
|
|
|
|
class FlushErrorBuffer(io.BytesIO):
|
|
def flush(self):
|
|
raise OSError(22, "Invalid argument")
|
|
|
|
|
|
class FlushErrorStream:
|
|
buffer = FlushErrorBuffer()
|
|
encoding = "utf-8"
|
|
line_buffering = False
|
|
|
|
|
|
def test_log_interceptor_flush_ignores_oserror_and_runs_callbacks():
|
|
interceptor = LogInterceptor(FlushErrorStream())
|
|
interceptor._logs_since_flush = [{"m": "message"}]
|
|
flushed_logs = []
|
|
|
|
interceptor.on_flush(lambda logs: flushed_logs.append(list(logs)))
|
|
interceptor.flush()
|
|
|
|
assert flushed_logs == [[{"m": "message"}]]
|
|
assert interceptor._logs_since_flush == []
|