This commit is contained in:
+28
-70
@@ -1,87 +1,45 @@
|
|||||||
import logging
|
import logging
|
||||||
import logzero
|
|
||||||
import pytest
|
|
||||||
import tempfile
|
|
||||||
import subprocess
|
|
||||||
import os
|
|
||||||
import sys
|
import sys
|
||||||
import textwrap
|
import pytest
|
||||||
|
import logzero
|
||||||
from logzero import logger as log
|
from logzero import logger as log
|
||||||
from slic.utils.logcfg import add_log_Level, logcfg, setup_import_logging
|
from slic.utils.logcfg import add_log_Level, logcfg
|
||||||
|
from slic.utils.logign import ignore_log_msg
|
||||||
|
|
||||||
# Set for all the tests
|
@pytest.mark.parametrize("levelname,msg_to_ignore,msg_to_keep", [
|
||||||
@pytest.fixture(scope="module", autouse=True)
|
("WARNING", "This should be ignored", "This should appear"),
|
||||||
def setup_custom_levels():
|
("ENLARGE", "ENLARGE this", "Keep this ENLARGE"),
|
||||||
add_log_Level(log, "LONG", logging.INFO + 5, func_name="long")
|
|
||||||
add_log_Level(log, "ENLARGE", logging.INFO + 6, func_name="enlarge")
|
|
||||||
logcfg("LONG")
|
|
||||||
setup_import_logging()
|
|
||||||
|
|
||||||
# Test simple custom log
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("levelname, logfunc, message", [
|
|
||||||
("LONG", lambda msg: log.long(msg), "This is a LONG message"),
|
|
||||||
("ENLARGE", lambda msg: log.enlarge(msg), "Please ENLARGE this!"),
|
|
||||||
])
|
])
|
||||||
|
def test_ignore_log_msg_behavior(levelname, msg_to_ignore, msg_to_keep, capsys):
|
||||||
def test_custom_log_outputs(levelname, logfunc, message, capsys):
|
# Remove all handlers
|
||||||
# Remove all handlers to avoid duplicate logs
|
|
||||||
for h in log.handlers[:]:
|
for h in log.handlers[:]:
|
||||||
log.removeHandler(h)
|
log.removeHandler(h)
|
||||||
|
|
||||||
# Create a new handler that logs to stderr
|
# Set handler manually like your working test
|
||||||
handler = logging.StreamHandler(sys.stderr)
|
handler = logging.StreamHandler(sys.stderr)
|
||||||
formatter = logzero.LogFormatter()
|
formatter = logzero.LogFormatter()
|
||||||
handler.setFormatter(formatter)
|
handler.setFormatter(formatter)
|
||||||
log.addHandler(handler)
|
log.addHandler(handler)
|
||||||
|
|
||||||
logcfg(levelname)
|
# Add custom level and set config
|
||||||
|
add_log_Level(log, "ENLARGE", logging.INFO + 6, func_name="enlarge")
|
||||||
|
logcfg("DEBUG")
|
||||||
|
|
||||||
# Call and capture
|
# Clear any prior capsys capture
|
||||||
logfunc(message)
|
capsys.readouterr()
|
||||||
|
|
||||||
|
with ignore_log_msg(log, lvl=levelname, msg=msg_to_ignore):
|
||||||
|
if levelname == "WARNING":
|
||||||
|
log.warning(msg_to_ignore)
|
||||||
|
log.warning(msg_to_keep)
|
||||||
|
elif levelname == "ENLARGE":
|
||||||
|
log.enlarge(msg_to_ignore)
|
||||||
|
log.enlarge(msg_to_keep)
|
||||||
|
|
||||||
|
# Read captured stderr
|
||||||
captured = capsys.readouterr().err
|
captured = capsys.readouterr().err
|
||||||
|
|
||||||
print("Captured stderr:", repr(captured))
|
print("Captured stderr:", repr(captured))
|
||||||
|
|
||||||
assert message in captured, f"Expected message '{message}' not found in log"
|
# Assert the message to ignore is gone, the other one is present
|
||||||
assert levelname in captured, f"Expected level '{levelname}' not present in log"
|
assert msg_to_ignore not in captured, f"'{msg_to_ignore}' should have been filtered"
|
||||||
|
assert msg_to_keep in captured, f"'{msg_to_keep}' should have appeared"
|
||||||
|
|
||||||
# Test subprocess: top-level import logs only once
|
|
||||||
|
|
||||||
def test_import_logging_once_per_module():
|
|
||||||
code = textwrap.dedent("""
|
|
||||||
import sys
|
|
||||||
from slic.utils.logcfg import setup_import_logging, add_log_Level, logcfg
|
|
||||||
from logzero import logger as log
|
|
||||||
import logging
|
|
||||||
|
|
||||||
add_log_Level(log, "TRACE", logging.DEBUG - 1, func_name="trace")
|
|
||||||
logcfg("TRACE")
|
|
||||||
setup_import_logging()
|
|
||||||
|
|
||||||
import json
|
|
||||||
import math
|
|
||||||
import io
|
|
||||||
import random
|
|
||||||
""")
|
|
||||||
|
|
||||||
with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as tmp:
|
|
||||||
tmp.write(code)
|
|
||||||
tmp_path = tmp.name
|
|
||||||
|
|
||||||
env = os.environ.copy()
|
|
||||||
root_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
|
|
||||||
env["PYTHONPATH"] = root_path + os.pathsep + env.get("PYTHONPATH", "")
|
|
||||||
|
|
||||||
result = subprocess.run([sys.executable, tmp_path], capture_output=True, text=True, env=env)
|
|
||||||
os.remove(tmp_path)
|
|
||||||
|
|
||||||
assert result.returncode == 0, f"Script failed:\n{result.stderr}"
|
|
||||||
|
|
||||||
stderr = result.stderr
|
|
||||||
lines = stderr.splitlines()
|
|
||||||
|
|
||||||
for mod in ["json", "math", "io", "random"]:
|
|
||||||
count = sum(1 for line in lines if f"importing: {mod}" in line)
|
|
||||||
assert count == 1, f"Expected 1 import log for '{mod}', found {count}"
|
|
||||||
|
|||||||
Reference in New Issue
Block a user