85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
"""MinHash 指纹与去重测试。"""
|
|
import _bootstrap # noqa: F401
|
|
import unittest
|
|
|
|
from fault_log_analyzer.fingerprint import Deduplicator, MinHash, shingles
|
|
|
|
|
|
class TestShingles(unittest.TestCase):
|
|
def test_basic(self):
|
|
self.assertEqual(shingles("abcde"), {"abc", "bcd", "cde"})
|
|
|
|
def test_short_text(self):
|
|
self.assertEqual(shingles("ab"), {"ab"})
|
|
|
|
def test_empty(self):
|
|
self.assertEqual(shingles(""), set())
|
|
|
|
def test_whitespace_collapsed(self):
|
|
self.assertEqual(shingles("a b"), {"a b"})
|
|
|
|
|
|
class TestMinHash(unittest.TestCase):
|
|
def test_signature_empty(self):
|
|
mh = MinHash(num_perm=8)
|
|
self.assertEqual(mh.signature([]), [0] * 8)
|
|
|
|
def test_fingerprint_deterministic(self):
|
|
mh = MinHash()
|
|
self.assertEqual(
|
|
mh.fingerprint("no space left on device"),
|
|
mh.fingerprint("no space left on device"),
|
|
)
|
|
|
|
def test_fingerprint_length(self):
|
|
self.assertEqual(len(MinHash().fingerprint("hello world")), 16)
|
|
|
|
def test_fingerprint_empty(self):
|
|
self.assertEqual(MinHash().fingerprint(""), "0" * 16)
|
|
|
|
def test_fingerprint_is_hex(self):
|
|
self.assertRegex(MinHash().fingerprint("hello world"), r"^[0-9a-f]{16}$")
|
|
|
|
def test_similar_text_share_signature(self):
|
|
mh = MinHash(num_perm=16)
|
|
sig1 = mh.signature(shingles("disk full on /dev/sda1"))
|
|
sig2 = mh.signature(shingles("disk full on /dev/sda1"))
|
|
self.assertEqual(sig1, sig2)
|
|
|
|
|
|
class TestDeduplicator(unittest.TestCase):
|
|
def _clock(self):
|
|
state = {"t": 100.0}
|
|
return lambda: state["t"], state
|
|
|
|
def test_first_not_duplicate(self):
|
|
clock, state = self._clock()
|
|
d = Deduplicator(window_seconds=300, clock=clock)
|
|
self.assertFalse(d.is_duplicate("fp1"))
|
|
|
|
def test_duplicate_within_window(self):
|
|
clock, state = self._clock()
|
|
d = Deduplicator(window_seconds=300, clock=clock)
|
|
d.is_duplicate("fp1")
|
|
state["t"] = 200.0
|
|
self.assertTrue(d.is_duplicate("fp1"))
|
|
|
|
def test_expired_after_window(self):
|
|
clock, state = self._clock()
|
|
d = Deduplicator(window_seconds=300, clock=clock)
|
|
d.is_duplicate("fp1")
|
|
state["t"] = 500.0
|
|
self.assertFalse(d.is_duplicate("fp1"))
|
|
|
|
def test_mark_and_clear(self):
|
|
clock, state = self._clock()
|
|
d = Deduplicator(window_seconds=300, clock=clock)
|
|
d.mark("fp1")
|
|
self.assertTrue(d.is_duplicate("fp1"))
|
|
d.clear()
|
|
self.assertFalse(d.is_duplicate("fp1"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|