FilesExpand file tree

master

/

test_logger.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

185 lines (146 loc) · 6.34 KB

master

/

test_logger.py

Copy path

Top

File metadata and controls

• Code
• Blame

185 lines (146 loc) · 6.34 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

"""Unit tests for logger.py."""

from __future__ import annotations

import logging

from unittest.mock import call, patch

from zeroconf import _logger

from zeroconf._logger import _MAX_SEEN_LOGS, QuietLogger, _mark_seen, set_logger_level_if_unset

def test_loading_logger():

"""Test loading logger does not change level unless it is unset."""

log = logging.getLogger("zeroconf")

log.setLevel(logging.CRITICAL)

set_logger_level_if_unset()

log = logging.getLogger("zeroconf")

assert log.level == logging.CRITICAL

log = logging.getLogger("zeroconf")

log.setLevel(logging.NOTSET)

set_logger_level_if_unset()

log = logging.getLogger("zeroconf")

assert log.level == logging.WARNING

def test_log_warning_once():

"""Test we only log with warning level once."""

_logger._seen_logs.clear()

quiet_logger = QuietLogger()

with (

patch("zeroconf._logger.log.warning") as mock_log_warning,

patch("zeroconf._logger.log.debug") as mock_log_debug,

):

quiet_logger.log_warning_once("the warning")

assert mock_log_warning.mock_calls

assert not mock_log_debug.mock_calls

with (

patch("zeroconf._logger.log.warning") as mock_log_warning,

patch("zeroconf._logger.log.debug") as mock_log_debug,

):

quiet_logger.log_warning_once("the warning")

assert not mock_log_warning.mock_calls

assert mock_log_debug.mock_calls

def test_log_exception_warning():

"""Test we only log with warning level once."""

_logger._seen_logs.clear()

quiet_logger = QuietLogger()

with (

patch("zeroconf._logger.log.warning") as mock_log_warning,

patch("zeroconf._logger.log.debug") as mock_log_debug,

):

quiet_logger.log_exception_warning("the exception warning")

assert mock_log_warning.mock_calls

assert not mock_log_debug.mock_calls

with (

patch("zeroconf._logger.log.warning") as mock_log_warning,

patch("zeroconf._logger.log.debug") as mock_log_debug,

):

quiet_logger.log_exception_warning("the exception warning")

assert not mock_log_warning.mock_calls

assert mock_log_debug.mock_calls

def test_llog_exception_debug():

"""Test we only log with a trace once."""

_logger._seen_logs.clear()

quiet_logger = QuietLogger()

with patch("zeroconf._logger.log.debug") as mock_log_debug:

quiet_logger.log_exception_debug("the exception")

assert mock_log_debug.mock_calls == [call("the exception", exc_info=True)]

with patch("zeroconf._logger.log.debug") as mock_log_debug:

quiet_logger.log_exception_debug("the exception")

assert mock_log_debug.mock_calls == [call("the exception", exc_info=False)]

def test_mark_seen_absorbs_runtime_error_during_eviction() -> None:

"""Concurrent mutation can make ``iter(seen)`` raise ``RuntimeError``.

Free-threaded (3.14t) and multi-instance sync callers share

``_seen_logs``; if another thread mutates it between ``iter()``

and ``next()`` the iterator raises ``RuntimeError``.

``_mark_seen`` must absorb that and still insert the new key.

"""

class RacyDict(dict[str, None]):

def __iter__(self): # type: ignore[override]

raise RuntimeError("dictionary changed size during iteration")

seen: dict[str, None] = RacyDict()

for i in range(_MAX_SEEN_LOGS):

seen[f"k-{i}"] = None

assert _mark_seen(seen, "new-key") is True

assert "new-key" in seen

def test_mark_seen_drains_drift_above_cap() -> None:

"""``_mark_seen`` drains a drifted-over-cap dict back to the cap.

Concurrent inserts on the free-threaded build can leave the dict

transiently above ``_MAX_SEEN_LOGS`` (e.g. two threads both passed

the ``len < cap`` check and both inserted). The next non-racing

call must drain the accumulated overshoot, not just evict one

entry — otherwise the cap silently inflates with thread count.

"""

seen: dict[str, None] = {}

drift = 10

for i in range(_MAX_SEEN_LOGS + drift):

seen[f"k-{i}"] = None

assert len(seen) == _MAX_SEEN_LOGS + drift

assert _mark_seen(seen, "new-key") is True

assert len(seen) == _MAX_SEEN_LOGS

assert "new-key" in seen

for i in range(drift + 1):

assert f"k-{i}" not in seen

def test_mark_seen_drains_drift_on_hit_path() -> None:

"""``_mark_seen`` drains drift even when ``key`` is already cached.

A hit-heavy workload after a contention burst (e.g. the same

exception text deduplicated repeatedly) must still correct the

overshoot — otherwise the dict can sit permanently above the cap

until a miss happens to come along.

"""

seen: dict[str, None] = {}

drift = 10

for i in range(_MAX_SEEN_LOGS + drift):

seen[f"k-{i}"] = None

# Hit on a non-oldest key — survives the drift drain.

hit_key = f"k-{_MAX_SEEN_LOGS}"

assert _mark_seen(seen, hit_key) is False

assert len(seen) == _MAX_SEEN_LOGS

assert hit_key in seen

for i in range(drift):

assert f"k-{i}" not in seen

def test_seen_logs_is_bounded() -> None:

"""``_seen_logs`` stays at the cap and evicts oldest-first (FIFO)."""

_logger._seen_logs.clear()

overflow = 5

with patch("zeroconf._logger.log.warning"), patch("zeroconf._logger.log.debug"):

for i in range(_MAX_SEEN_LOGS + overflow):

QuietLogger.log_warning_once(f"warning-{i}")

assert len(_logger._seen_logs) == _MAX_SEEN_LOGS

for i in range(overflow):

assert f"warning-{i}" not in _logger._seen_logs

for i in range(_MAX_SEEN_LOGS, _MAX_SEEN_LOGS + overflow):

assert f"warning-{i}" in _logger._seen_logs

def test_log_exception_once():

"""Test we only log with warning level once."""

_logger._seen_logs.clear()

quiet_logger = QuietLogger()

exc = Exception()

with (

patch("zeroconf._logger.log.warning") as mock_log_warning,

patch("zeroconf._logger.log.debug") as mock_log_debug,

):

quiet_logger.log_exception_once(exc, "the exceptional exception warning")

assert mock_log_warning.mock_calls

assert not mock_log_debug.mock_calls

with (

patch("zeroconf._logger.log.warning") as mock_log_warning,

patch("zeroconf._logger.log.debug") as mock_log_debug,

):

quiet_logger.log_exception_once(exc, "the exceptional exception warning")

assert not mock_log_warning.mock_calls

assert mock_log_debug.mock_calls