0e13ee5d51
- Connector protocol with fetch(query: SearchQuery) -> list[RawPosting] - ArbetsformedlingenConnector: official Platsbanken API, field mapping (headline->title, employer.name->company, webpage_url->url, description.text->description, id->external_id), region mapping, polite User-Agent - GenericUrlConnector: readability extraction (prefers main/article, strips nav/footer/script/style), title/company guess, Cloudflare challenge detection -> UnsupportedSite - dedupe helper: (source, url) + external_id keying - normalizer: RawPosting -> JobPosting (job_posting table shape) - 37 tests, all passing, recorded fixtures (no live network) - README with usage and mock examples
119 lines
No EOL
4.3 KiB
Python
119 lines
No EOL
4.3 KiB
Python
"""Tests for the deduplication helper."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from connectors.dedupe import dedupe, dedupe_job_postings
|
|
from connectors.models import JobPosting, RawPosting
|
|
|
|
|
|
def _make_raw(
|
|
source: str = "arbetsformedlingen",
|
|
external_id: str | None = "1",
|
|
url: str = "https://example.com/1",
|
|
company: str = "Corp",
|
|
title: str = "Dev",
|
|
) -> RawPosting:
|
|
return RawPosting(
|
|
source=source,
|
|
external_id=external_id,
|
|
url=url,
|
|
company=company,
|
|
title=title,
|
|
)
|
|
|
|
|
|
class TestDedupeRawPostings:
|
|
"""Test dedupe function with RawPosting objects."""
|
|
|
|
def test_no_duplicates_unchanged(self):
|
|
"""Test that a list with no duplicates is unchanged."""
|
|
postings = [
|
|
_make_raw(url="https://a.com/1", external_id="1"),
|
|
_make_raw(url="https://a.com/2", external_id="2"),
|
|
_make_raw(url="https://a.com/3", external_id="3"),
|
|
]
|
|
result = dedupe(postings)
|
|
assert len(result) == 3
|
|
|
|
def test_dedupe_by_url(self):
|
|
"""Test deduplication by (source, url)."""
|
|
postings = [
|
|
_make_raw(url="https://a.com/1", external_id="1"),
|
|
_make_raw(url="https://a.com/1", external_id="2"), # same URL, diff ext_id
|
|
_make_raw(url="https://a.com/2", external_id="3"),
|
|
]
|
|
result = dedupe(postings)
|
|
assert len(result) == 2
|
|
assert result[0].external_id == "1" # first occurrence kept
|
|
assert result[1].external_id == "3"
|
|
|
|
def test_dedupe_by_external_id(self):
|
|
"""Test deduplication by (source, external_id) when URL differs."""
|
|
postings = [
|
|
_make_raw(url="https://a.com/1", external_id="100"),
|
|
_make_raw(url="https://a.com/2", external_id="100"), # same ext_id
|
|
_make_raw(url="https://a.com/3", external_id="200"),
|
|
]
|
|
result = dedupe(postings)
|
|
assert len(result) == 2
|
|
assert result[0].url == "https://a.com/1"
|
|
assert result[1].url == "https://a.com/3"
|
|
|
|
def test_different_sources_same_url_not_deduped(self):
|
|
"""Test that same URL from different sources are NOT deduped."""
|
|
postings = [
|
|
_make_raw(source="arbetsformedlingen", url="https://a.com/1", external_id="1"),
|
|
_make_raw(source="generic_url", url="https://a.com/1", external_id=None),
|
|
]
|
|
result = dedupe(postings)
|
|
assert len(result) == 2
|
|
|
|
def test_none_external_id_ignored(self):
|
|
"""Test that None external_id does not cause dedup by ext_id."""
|
|
postings = [
|
|
_make_raw(url="https://a.com/1", external_id=None),
|
|
_make_raw(url="https://a.com/2", external_id=None),
|
|
_make_raw(url="https://a.com/3", external_id=None),
|
|
]
|
|
result = dedupe(postings)
|
|
assert len(result) == 3
|
|
|
|
def test_empty_list(self):
|
|
"""Test that an empty list returns empty."""
|
|
assert dedupe([]) == []
|
|
|
|
def test_single_item(self):
|
|
"""Test that a single item list is unchanged."""
|
|
postings = [_make_raw()]
|
|
result = dedupe(postings)
|
|
assert len(result) == 1
|
|
|
|
def test_order_preserved(self):
|
|
"""Test that first-occurrence order is preserved."""
|
|
postings = [
|
|
_make_raw(url="https://a.com/3", external_id="3", title="Third"),
|
|
_make_raw(url="https://a.com/1", external_id="1", title="First"),
|
|
_make_raw(url="https://a.com/3", external_id="3", title="Third-Dup"),
|
|
_make_raw(url="https://a.com/2", external_id="2", title="Second"),
|
|
]
|
|
result = dedupe(postings)
|
|
assert len(result) == 3
|
|
assert result[0].title == "Third"
|
|
assert result[1].title == "First"
|
|
assert result[2].title == "Second"
|
|
|
|
|
|
class TestDedupeJobPostings:
|
|
"""Test dedupe_job_postings function with JobPosting objects."""
|
|
|
|
def test_dedupe_job_postings_by_url(self):
|
|
"""Test deduplication of JobPosting objects."""
|
|
postings = [
|
|
JobPosting(source="af", external_id="1", url="https://a.com/1",
|
|
company="C", title="T"),
|
|
JobPosting(source="af", external_id="2", url="https://a.com/1",
|
|
company="C", title="T"),
|
|
]
|
|
result = dedupe_job_postings(postings)
|
|
assert len(result) == 1
|
|
assert result[0].external_id == "1" |