70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
import os, sys
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from minhash_dedup import *
|
|
|
|
def test_shingles_basic():
|
|
sh = shingles("abcde", 3)
|
|
assert " abc" in sh or "abc" in sh or len(sh) > 0
|
|
assert len(shingles("hello world", 4)) > 0
|
|
|
|
def test_identical_lines_high_similarity():
|
|
a = "the quick brown fox jumps over the lazy dog"
|
|
b = "the quick brown fox jumps over the lazy dog"
|
|
sa = minhash_signature(shingles(a))
|
|
sb = minhash_signature(shingles(b))
|
|
assert jaccard_est(sa, sb) == 1.0
|
|
|
|
def test_distinct_lines_low_similarity():
|
|
a = "the quick brown fox jumps over the lazy dog"
|
|
b = "completely unrelated sentence about databases and servers"
|
|
sa = minhash_signature(shingles(a))
|
|
sb = minhash_signature(shingles(b))
|
|
assert jaccard_est(sa, sb) < 0.5
|
|
|
|
def test_find_duplicates_detects_near_dup():
|
|
lines = [
|
|
"error: connection refused to host 10.0.0.1 port 5432",
|
|
"error: connection refused to host 10.0.0.2 port 5432",
|
|
"totally different line",
|
|
]
|
|
dups = find_duplicates(lines, threshold=0.3)
|
|
assert any((i,j)==(0,1) for i,j,_ in dups), dups
|
|
|
|
def test_candidate_pairs_self():
|
|
sigs = [minhash_signature(shingles("abc"*10)), minhash_signature(shingles("abc"*10))]
|
|
pairs = candidate_pairs(sigs, bands=8)
|
|
assert (0,1) in pairs
|
|
|
|
def test_empty_shingle_set():
|
|
sig = minhash_signature(set())
|
|
assert all(x == 2**64-1 for x in sig)
|
|
|
|
def test_jaccard_length_mismatch():
|
|
try:
|
|
jaccard_est([1,2],[1,2,3])
|
|
raise AssertionError("expected ValueError")
|
|
except ValueError:
|
|
pass
|
|
|
|
def test_threshold_filter():
|
|
lines = ["aaaa", "bbbb", "cccc"]
|
|
dups = find_duplicates(lines, threshold=0.99)
|
|
# distinct short strings may or may not collide; just ensure no crash and type
|
|
assert isinstance(dups, list)
|
|
|
|
if __name__ == "__main__":
|
|
# minimal runner without pytest
|
|
import traceback
|
|
tests = [v for k,v in globals().items() if k.startswith("test_")]
|
|
fails = 0
|
|
for t in tests:
|
|
try:
|
|
t()
|
|
print("PASS", t.__name__)
|
|
except Exception as e:
|
|
fails += 1
|
|
print("FAIL", t.__name__, repr(e))
|
|
traceback.print_exc()
|
|
print("FAILS:", fails)
|
|
sys.exit(1 if fails else 0)
|