Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
- Make ``logger.catch()`` usable as an asynchronous context manager (`#1084 <https://github.com/Delgan/loguru/issues/1084>`_).
- Make ``logger.catch()`` compatible with asynchronous generators (`#1302 <https://github.com/Delgan/loguru/issues/1302>`_).
- Improve feedback for invalid format keys in logger format strings (`#1450 <https://github.com/Delgan/loguru/issues/1450>`_, thanks `@Krishnachaitanyakc <https://github.com/Krishnachaitanyakc>`_).
- Fix unhandled PermissionError raise on Windows when deleting files with open file handles during file retention deletion policy (`#1495 <https://github.com/Delgan/loguru/issues/1495>`_).


`0.7.3`_ (2024-12-06)
Expand Down
12 changes: 9 additions & 3 deletions loguru/_file_sink.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,14 +74,20 @@ def key_log(log):
return (-os.stat(log).st_mtime, log)

for log in sorted(logs, key=key_log)[number:]:
os.remove(log)
try:
os.remove(log)
except PermissionError:
pass

@staticmethod
def retention_age(logs, seconds):
t = datetime.datetime.now().timestamp()
for log in logs:
if os.stat(log).st_mtime <= t - seconds:
os.remove(log)
try:
if os.stat(log).st_mtime <= t - seconds:
os.remove(log)
except PermissionError:
pass


class Rotation:
Expand Down
102 changes: 102 additions & 0 deletions tests/test_filesink_retention.py
Original file line number Diff line number Diff line change
Expand Up @@ -343,6 +343,108 @@ def test_exception_during_retention_at_remove(tmp_path, capsys, delay):
assert out == err == ""


def make_remove_failing_for(filename):
real_remove = os.remove

def remove(path):
if os.path.basename(path) == filename:
raise PermissionError(
"The process cannot access the file because it is being used by another process"
)
real_remove(path)

return remove


def test_permission_error_ignored_during_retention_count(tmp_path, monkeypatch, capsys):
tmp_path.joinpath("test.log.1").write_text("A")
tmp_path.joinpath("test.log.2").write_text("B")

monkeypatch.setattr(os, "remove", make_remove_failing_for("test.log.1"))

i = logger.add(tmp_path / "test.log", format="{message}", retention=0, catch=False)
logger.debug("test")
logger.remove(i)

check_dir(tmp_path, files=[("test.log.1", "A")])

out, err = capsys.readouterr()
assert out == err == ""


def test_permission_error_ignored_during_retention_age(tmp_path, monkeypatch, capsys):
past = datetime.datetime.now().timestamp() - 7200

for name, content in [("test.log.1", "A"), ("test.log.2", "B")]:
filepath = tmp_path / name
filepath.write_text(content)
os.utime(str(filepath), (past, past))

monkeypatch.setattr(os, "remove", make_remove_failing_for("test.log.1"))

i = logger.add(tmp_path / "test.log", format="{message}", retention="1 hour", catch=False)
logger.debug("test")
logger.remove(i)

check_dir(tmp_path, files=[("test.log", "test\n"), ("test.log.1", "A")])

out, err = capsys.readouterr()
assert out == err == ""


@pytest.mark.parametrize("retention", [0, "1 hour"])
def test_error_during_retention_not_ignored(tmp_path, monkeypatch, retention):
past = datetime.datetime.now().timestamp() - 7200
filepath = tmp_path / "test.log.1"
filepath.write_text("A")
os.utime(str(filepath), (past, past))

monkeypatch.setattr(os, "remove", Mock(side_effect=OSError("Removal error")))

i = logger.add(tmp_path / "test.log", format="{message}", retention=retention, catch=False)
logger.debug("test")

with pytest.raises(OSError, match=r"^Removal error$"):
logger.remove(i)


@pytest.mark.skipif(os.name != "nt", reason="Windows can't delete file in use")
def test_file_in_use_ignored_during_retention_count(tmp_path, capsys):
filepath = tmp_path / "test.log.1"
filepath.write_text("A")
tmp_path.joinpath("test.log.2").write_text("B")

with filepath.open("r"):
i = logger.add(tmp_path / "test.log", format="{message}", retention=0, catch=False)
logger.debug("test")
logger.remove(i)

check_dir(tmp_path, files=[("test.log.1", "A")])

out, err = capsys.readouterr()
assert out == err == ""


@pytest.mark.skipif(os.name != "nt", reason="Windows can't delete file in use")
def test_file_in_use_ignored_during_retention_age(tmp_path, capsys):
past = datetime.datetime.now().timestamp() - 7200

for name, content in [("test.log.1", "A"), ("test.log.2", "B")]:
filepath = tmp_path / name
filepath.write_text(content)
os.utime(str(filepath), (past, past))

with tmp_path.joinpath("test.log.1").open("r"):
i = logger.add(tmp_path / "test.log", format="{message}", retention="1 hour", catch=False)
logger.debug("test")
logger.remove(i)

check_dir(tmp_path, files=[("test.log", "test\n"), ("test.log.1", "A")])

out, err = capsys.readouterr()
assert out == err == ""


@pytest.mark.parametrize("retention", [datetime.time(12, 12, 12), os, object()])
def test_invalid_retention_type(retention):
with pytest.raises(TypeError):
Expand Down