213 lines
7.8 KiB
Python
213 lines
7.8 KiB
Python
import hashlib
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from django.db import connection
|
|
from django.test.utils import CaptureQueriesContext
|
|
from django.utils import timezone
|
|
|
|
from apps.jobs.models import (
|
|
Employer,
|
|
FieldProvenance,
|
|
JobPosting,
|
|
JobSourceAlias,
|
|
JobVersion,
|
|
ScoreRun,
|
|
)
|
|
from apps.jobs.services.pipeline import PersistenceContext, process_raw_document
|
|
from apps.sources.models import RawDocument, Source
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_pipeline_is_idempotent_and_scores(source, profile):
|
|
content = Path("fixtures/pages/sample_jsonld_job.html").read_text(encoding="utf-8")
|
|
document = RawDocument.objects.create(
|
|
source=source,
|
|
url="https://jobs.example.org/vacatures/infrastructure-engineer",
|
|
final_url="https://jobs.example.org/vacatures/infrastructure-engineer",
|
|
kind=RawDocument.Kind.HTML,
|
|
content_type="text/html",
|
|
content_hash=hashlib.sha256(content.encode()).hexdigest(),
|
|
body_text=content,
|
|
byte_length=len(content.encode()),
|
|
retain_until=timezone.now() + timedelta(days=7),
|
|
)
|
|
first = process_raw_document(document)
|
|
second = process_raw_document(document)
|
|
assert first["created"] == 1
|
|
assert second["created"] == 0
|
|
assert JobPosting.objects.count() == 1
|
|
assert JobSourceAlias.objects.count() == 1
|
|
assert ScoreRun.objects.filter(profile=profile).count() == 2
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_unchanged_import_batch_has_bounded_lookup_queries(source, profile):
|
|
content = Path("fixtures/pages/sample_jsonld_job.html").read_text(encoding="utf-8")
|
|
documents = [
|
|
RawDocument.objects.create(
|
|
source=source,
|
|
url=f"https://jobs.example.org/import/{index}",
|
|
final_url=f"https://jobs.example.org/import/{index}",
|
|
kind=RawDocument.Kind.HTML,
|
|
content_type="text/html",
|
|
content_hash=hashlib.sha256(f"{index}{content}".encode()).hexdigest(),
|
|
body_text=content,
|
|
byte_length=len(content.encode()),
|
|
retain_until=timezone.now() + timedelta(days=7),
|
|
)
|
|
for index in range(20)
|
|
]
|
|
context = PersistenceContext()
|
|
|
|
with CaptureQueriesContext(connection) as queries:
|
|
results = [process_raw_document(document, context=context) for document in documents]
|
|
|
|
assert sum(int(result["created"]) for result in results) == 1
|
|
assert sum(int(result["duplicates"]) for result in results) == 19
|
|
assert len(queries) < 300
|
|
assert Employer.objects.count() == 1
|
|
assert JobPosting.objects.count() == 1
|
|
assert JobVersion.objects.count() == 1
|
|
assert FieldProvenance.objects.count() > 0
|
|
assert ScoreRun.objects.filter(profile=profile).count() == 20
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_pipeline_uses_reviewed_employer_identity_for_public_ats_feed(profile):
|
|
source = Source.objects.create(
|
|
name="Deliverect",
|
|
source_type=Source.Type.ATS,
|
|
base_url="https://api.lever.co/v0/postings/deliverect?mode=json",
|
|
domain="api.lever.co",
|
|
status=Source.Status.ACTIVE,
|
|
policy=Source.Policy.ALLOW,
|
|
metadata={"employer_name": "Deliverect", "direct_employer": True},
|
|
)
|
|
content = Path("fixtures/ats/lever-public-postings.json").read_text(encoding="utf-8")
|
|
document = RawDocument.objects.create(
|
|
source=source,
|
|
url=source.base_url,
|
|
final_url=source.base_url,
|
|
kind=RawDocument.Kind.JSON,
|
|
content_type="application/json",
|
|
content_hash=hashlib.sha256(content.encode()).hexdigest(),
|
|
body_text=content,
|
|
byte_length=len(content.encode()),
|
|
retain_until=timezone.now() + timedelta(days=7),
|
|
)
|
|
|
|
result = process_raw_document(document)
|
|
|
|
assert result["created"] == 1
|
|
job = JobPosting.objects.get()
|
|
assert job.employer_name == "Deliverect"
|
|
assert job.direct_employer is True
|
|
evidence = job.source_aliases.get().field_evidence.get(field_name="employer_name")
|
|
assert evidence.extraction_method == "reviewed-source-config"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_pipeline_resolves_recruiter_alias_to_direct_employer():
|
|
source = Source.objects.create(
|
|
name="SmartRecruiters",
|
|
source_type=Source.Type.ATS,
|
|
base_url="https://jobs.smartrecruiters.com/example",
|
|
domain="jobs.smartrecruiters.com",
|
|
status=Source.Status.ACTIVE,
|
|
policy=Source.Policy.ALLOW,
|
|
)
|
|
employer = Employer.objects.create(
|
|
name="Example Public IT",
|
|
normalized_name="example public it",
|
|
domain="jobs.example.org",
|
|
is_direct_employer=True,
|
|
is_recruiter=False,
|
|
confidence=0.9,
|
|
)
|
|
direct_job = JobPosting.objects.create(
|
|
employer=employer,
|
|
original_title="Security Engineer",
|
|
normalized_title="security engineer",
|
|
canonical_url="https://jobs.example.org/vacatures/security-engineer",
|
|
canonical_key="8" * 64,
|
|
content_hash="e" * 64,
|
|
description_text="Beveiliger, hybride werk met support op locatie.",
|
|
raw_location="Antwerpen",
|
|
region="Antwerpen",
|
|
municipality="Antwerpen",
|
|
workplace_type=JobPosting.Workplace.HYBRID,
|
|
employment_types=["full_time"],
|
|
direct_employer=True,
|
|
)
|
|
content = """
|
|
<html lang="nl">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>Security Engineer | Example Public IT</title>
|
|
<link rel="canonical" href="https://jobs.smartrecruiters.com/example/security-engineer">
|
|
<script type="application/ld+json">
|
|
{
|
|
"@context": "https://schema.org",
|
|
"@type": "JobPosting",
|
|
"identifier": {
|
|
"@type": "PropertyValue",
|
|
"name": "Example Public IT",
|
|
"value": "VR-DEMO-SEC-001"
|
|
},
|
|
"title": "Security Engineer",
|
|
"description": "<p>Beveiliging, monitoring en hybride support.</p>",
|
|
"datePosted": "2026-07-20",
|
|
"validThrough": "2026-08-31T23:59:00+02:00",
|
|
"employmentType": ["FULL_TIME"],
|
|
"hiringOrganization": {
|
|
"@type": "Organization",
|
|
"name": "Example Public IT",
|
|
"sameAs": "https://www.example.org"
|
|
},
|
|
"jobLocation": {
|
|
"@type": "Place",
|
|
"address": {
|
|
"@type": "PostalAddress",
|
|
"streetAddress": "Voorbeeldstraat 1",
|
|
"addressLocality": "Antwerpen",
|
|
"addressRegion": "Antwerpen",
|
|
"postalCode": "2000",
|
|
"addressCountry": "BE"
|
|
}
|
|
},
|
|
"jobLocationType": "HYBRID",
|
|
"url": "/example/security-engineer"
|
|
}
|
|
</script>
|
|
</head>
|
|
<body><main><h1>Security Engineer</h1></main></body>
|
|
</html>
|
|
""".strip()
|
|
body_hash = hashlib.sha256(content.encode()).hexdigest()
|
|
document = RawDocument.objects.create(
|
|
source=source,
|
|
url="https://jobs.smartrecruiters.com/example/security-engineer",
|
|
final_url="https://jobs.smartrecruiters.com/example/security-engineer",
|
|
kind=RawDocument.Kind.HTML,
|
|
content_type="text/html",
|
|
content_hash=body_hash,
|
|
body_text=content,
|
|
byte_length=len(content.encode()),
|
|
retain_until=timezone.now() + timedelta(days=7),
|
|
)
|
|
result = process_raw_document(document)
|
|
direct_job.refresh_from_db()
|
|
assert direct_job.canonical_url == "https://jobs.example.org/vacatures/security-engineer"
|
|
assert result["created"] == 0
|
|
assert result["duplicates"] == 1
|
|
alias = JobSourceAlias.objects.get(job=direct_job, source=source)
|
|
payload = alias.payload.get("employer_resolution")
|
|
assert payload is not None
|
|
assert payload["reason"] == "resolved_direct_match"
|
|
assert payload["resolved_direct"] is True
|
|
assert payload["canonical_url"] == "https://jobs.example.org/vacatures/security-engineer"
|