freedak/Scrapling
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: update tests accordingly

Browse Source
This commit is contained in:
Karim shoair
2026-01-20 20:22:47 +02:00
parent 122923c80e
commit d90189e675
2 changed files with 47 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/spiders/test_request.py
+31 -13
View File
@@ -80,15 +80,24 @@ class TestRequestProperties:
request = Request("https://api.v2.example.com/endpoint")
assert request.domain == "api.v2.example.com"
def test_fingerprint_includes_session_and_url(self):
"""Test fingerprint generation."""
request = Request("https://example.com", sid="session1")
assert request._fp == "session1:https://example.com"
def test_fingerprint_empty_session(self):
"""Test fingerprint with empty session ID."""
def test_fingerprint_returns_bytes(self):
"""Test fingerprint generation returns bytes."""
request = Request("https://example.com")
assert request._fp == ":https://example.com"
fp = request.update_fingerprint()
assert isinstance(fp, bytes)
assert len(fp) == 20 # SHA1 produces 20 bytes
def test_fingerprint_is_deterministic(self):
"""Test same request produces same fingerprint."""
r1 = Request("https://example.com", data={"key": "value"})
r2 = Request("https://example.com", data={"key": "value"})
assert r1.update_fingerprint() == r2.update_fingerprint()
def test_fingerprint_different_urls(self):
"""Test different URLs produce different fingerprints."""
r1 = Request("https://example.com/page1")
r2 = Request("https://example.com/page2")
assert r1.update_fingerprint() != r2.update_fingerprint()
class TestRequestCopy:
@@ -160,17 +169,26 @@ class TestRequestComparison:
def test_equality_by_fingerprint(self):
"""Test equality comparison by fingerprint."""
r1 = Request("https://example.com", sid="session1")
r2 = Request("https://example.com", sid="session1")
r3 = Request("https://example.com", sid="session2")
r1 = Request("https://example.com")
r2 = Request("https://example.com")
r3 = Request("https://example.com/other")
# Generate fingerprints first (required for equality)
r1.update_fingerprint()
r2.update_fingerprint()
r3.update_fingerprint()
assert r1 == r2
assert r1 != r3
def test_equality_different_priorities_same_fingerprint(self):
"""Test requests with same fingerprint are equal despite different priorities."""
r1 = Request("https://example.com", sid="s1", priority=1)
r2 = Request("https://example.com", sid="s1", priority=100)
r1 = Request("https://example.com", priority=1)
r2 = Request("https://example.com", priority=100)
# Generate fingerprints first
r1.update_fingerprint()
r2.update_fingerprint()
assert r1 == r2 # Same fingerprint means equal
tests/spiders/test_scheduler.py
+16 -20
View File
@@ -75,12 +75,12 @@ class TestSchedulerEnqueue:
assert len(scheduler) == 2
@pytest.mark.asyncio
async def test_enqueue_different_sessions_not_duplicate(self):
"""Test that same URL with different sessions are not duplicates."""
async def test_enqueue_different_methods_not_duplicate(self):
"""Test that same URL with different methods are not duplicates."""
scheduler = Scheduler()
request1 = Request("https://example.com", sid="session1")
request2 = Request("https://example.com", sid="session2")
request1 = Request("https://example.com", method="GET")
request2 = Request("https://example.com", method="POST")
result1 = await scheduler.enqueue(request1)
result2 = await scheduler.enqueue(request2)
@@ -195,17 +195,19 @@ class TestSchedulerSnapshot:
@pytest.mark.asyncio
async def test_snapshot_captures_seen_set(self):
"""Test snapshot captures seen URLs."""
"""Test snapshot captures seen fingerprints."""
scheduler = Scheduler()
await scheduler.enqueue(Request("https://example.com/1", sid="s1"))
await scheduler.enqueue(Request("https://example.com/2", sid="s1"))
await scheduler.enqueue(Request("https://example.com/1"))
await scheduler.enqueue(Request("https://example.com/2"))
requests, seen = scheduler.snapshot()
assert len(seen) == 2
assert "s1:https://example.com/1" in seen
assert "s1:https://example.com/2" in seen
# Fingerprints are now bytes (SHA1 hashes)
for fp in seen:
assert isinstance(fp, bytes)
assert len(fp) == 20 # SHA1 produces 20 bytes
@pytest.mark.asyncio
async def test_snapshot_returns_copies(self):
@@ -218,12 +220,12 @@ class TestSchedulerSnapshot:
# Modifying snapshot shouldn't affect scheduler
requests.append(Request("https://modified.com"))
seen.add("new_fingerprint")
seen.add(b"new_fingerprint_bytes")
original_requests, original_seen = scheduler.snapshot()
assert len(original_requests) == 1
assert "new_fingerprint" not in original_seen
assert b"new_fingerprint_bytes" not in original_seen
@pytest.mark.asyncio
async def test_snapshot_excludes_dequeued_requests(self):
@@ -257,7 +259,7 @@ class TestSchedulerRestore:
Request("https://example.com/1", priority=10),
Request("https://example.com/2", priority=5),
]
checkpoint_seen = {"fp1", "fp2", "fp3"}
checkpoint_seen = {b"fp1_bytes_padded!", b"fp2_bytes_padded!", b"fp3_bytes_padded!"}
data = CheckpointData(requests=checkpoint_requests, seen=checkpoint_seen)
@@ -272,20 +274,14 @@ class TestSchedulerRestore:
data = CheckpointData(
requests=[],
seen={"fp1", "fp2"},
seen={b"fp1_bytes_here_pad", b"fp2_bytes_here_pad"}, # Bytes fingerprints
)
scheduler.restore(data)
# Now try to enqueue a request with matching fingerprint
request = Request("https://example.com")
request.sid = "" # Empty sid
# Manually set fingerprint that matches seen
# Since fingerprint is sid:url, we need to create matching ones
# Verify seen set was restored
_, seen = scheduler.snapshot()
assert seen == {"fp1", "fp2"}
assert seen == {b"fp1_bytes_here_pad", b"fp2_bytes_here_pad"}
@pytest.mark.asyncio
async def test_restore_maintains_priority_order(self):