113 lines
4.4 KiB
Python
113 lines
4.4 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from apps.sources.models import Source
|
|
from apps.sources.services.discovery import (
|
|
SourceCandidate,
|
|
discover_career_links,
|
|
discover_from_email,
|
|
discover_from_feed,
|
|
discover_from_html,
|
|
discover_from_sitemap,
|
|
persist_discovery_candidates,
|
|
)
|
|
|
|
|
|
def test_discovery_finds_career_links_and_skips_denied_domains():
|
|
html = """
|
|
<a href="/werken-bij">Werken bij ons</a>
|
|
<a href="https://careers.partner.example/jobs">Jobs partner</a>
|
|
<a href="https://www.linkedin.com/jobs">Jobs op LinkedIn</a>
|
|
<a href="/about">Over ons</a>
|
|
"""
|
|
result = discover_career_links(html, base_url="https://example.org/")
|
|
assert [item.url for item in result] == [
|
|
"https://example.org/werken-bij",
|
|
"https://careers.partner.example/jobs",
|
|
]
|
|
assert result[0].confidence > result[1].confidence
|
|
|
|
|
|
def test_discovery_from_html_includes_jsonld_and_feed_links():
|
|
html = Path("fixtures/discovery/html_candidate_discovery.html").read_text(encoding="utf-8")
|
|
result = discover_from_html(html, base_url="https://example.org/")
|
|
|
|
candidate_urls = {candidate.url: candidate.discovered_from for candidate in result}
|
|
|
|
assert candidate_urls["https://example.org/werken-bij"] == "html"
|
|
assert candidate_urls["https://jobs.example.org/vacatures/"] == "html"
|
|
assert candidate_urls["https://jobs.example.org/vacatures/data-engineer"] == "html-jsonld"
|
|
assert (
|
|
candidate_urls["https://jobs.example.org/vacatures/data-engineer/solliciteer"]
|
|
== "html-jsonld"
|
|
)
|
|
assert candidate_urls["https://example.org/feed/jobs.xml"] == "html"
|
|
assert candidate_urls["https://jobs.example.org/vacatures/data-engineer"].startswith(
|
|
"html-jsonld"
|
|
)
|
|
assert any(c.source_type == Source.Type.RSS and c.reason == "feed" for c in result)
|
|
assert any(
|
|
c.source_type == Source.Type.EMPLOYER and c.url == "https://example.org/werken-bij"
|
|
for c in result
|
|
)
|
|
|
|
|
|
def test_discovery_from_feed_detects_internal_links_only():
|
|
content = Path("fixtures/discovery/feed_jobs.xml").read_text(encoding="utf-8")
|
|
result = discover_from_feed(content, base_url="https://jobs.example.org/feed/jobs")
|
|
|
|
assert len(result) == 1
|
|
assert result[0].url == "https://jobs.example.org/vacatures/cloud-architect"
|
|
assert result[0].source_type == Source.Type.RSS
|
|
|
|
|
|
def test_discovery_from_sitemap_prefers_candidates_and_filters_private_or_denied():
|
|
index_content = Path("fixtures/discovery/sitemap_index.xml").read_text(encoding="utf-8")
|
|
urlset_content = Path("fixtures/discovery/sitemap_urlset.xml").read_text(encoding="utf-8")
|
|
|
|
index_candidates = discover_from_sitemap(index_content, base_url="https://jobs.example.org/")
|
|
urlset_candidates = discover_from_sitemap(urlset_content, base_url="https://jobs.example.org/")
|
|
|
|
assert len(index_candidates) == 1
|
|
assert index_candidates[0].url == "https://jobs.example.org/sitemap-jobs.xml"
|
|
assert len(urlset_candidates) >= 2
|
|
assert any(
|
|
item.url == "https://jobs.example.org/vacatures/data-engineer" for item in urlset_candidates
|
|
)
|
|
assert all("linkedin.com" not in item.url for item in index_candidates + urlset_candidates)
|
|
|
|
|
|
def test_discovery_from_email_extracts_employer_domain_root():
|
|
raw = Path("fixtures/emails/sample_alert.eml").read_bytes()
|
|
result = discover_from_email(raw)
|
|
assert [item.url for item in result] == ["https://jobs.example.org/"]
|
|
assert result[0].source_type == Source.Type.EMPLOYER
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_discovery_persistence_is_idempotent():
|
|
candidates = [
|
|
SourceCandidate(
|
|
url="https://jobs.example.org/vacatures/data-engineer",
|
|
domain="jobs.example.org",
|
|
source_type=Source.Type.EMPLOYER,
|
|
label="Testbron",
|
|
confidence=0.95,
|
|
reason="test",
|
|
discovered_from="unit-test",
|
|
)
|
|
]
|
|
|
|
created, updated, skipped = persist_discovery_candidates(candidates)
|
|
assert (created, updated, skipped) == (1, 0, 0)
|
|
|
|
created, updated, skipped = persist_discovery_candidates(candidates)
|
|
assert (created, updated, skipped) == (0, 0, 1)
|
|
|
|
source = Source.objects.get(domain="jobs.example.org", source_type=Source.Type.EMPLOYER)
|
|
assert source.status == Source.Status.CANDIDATE
|
|
assert source.policy == Source.Policy.REVIEW
|
|
assert source.name == "Testbron"
|
|
assert len(source.discovery_evidence) == 1
|