148 lines
5.0 KiB
Python
148 lines
5.0 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.utils import timezone
|
|
|
|
from apps.jobs.models import Application, Feedback, JobPosting, JobSourceAlias
|
|
from apps.jobs.services.dedupe import candidate_similarity, find_existing_job, text_similarity
|
|
from apps.jobs.services.feedback import record_feedback
|
|
from apps.jobs.services.lifecycle import update_lifecycle
|
|
from apps.jobs.services.normalization import CanonicalJobDraft
|
|
from apps.sources.models import RawDocument
|
|
|
|
|
|
def _draft(job, **overrides):
|
|
values = {
|
|
"source_url": job.canonical_url,
|
|
"canonical_url": job.canonical_url,
|
|
"external_id": "external-1",
|
|
"title": job.original_title,
|
|
"normalized_title": job.normalized_title,
|
|
"job_family": job.job_family,
|
|
"employer_name": job.employer_name,
|
|
"employer_domain": job.employer.domain,
|
|
"location_text": job.raw_location,
|
|
"region": job.region,
|
|
"municipality": job.municipality,
|
|
"postal_code": "3500",
|
|
"country": "BE",
|
|
"workplace_type": job.workplace_type,
|
|
"employment_types": job.employment_types,
|
|
"language": "nl",
|
|
"description_html": "",
|
|
"description_text": job.description_text,
|
|
"date_posted": None,
|
|
"valid_through": None,
|
|
"compensation": {},
|
|
"skills_required": job.skills_required,
|
|
"skills_preferred": job.skills_preferred,
|
|
"content_hash": job.content_hash,
|
|
"canonical_key": job.canonical_key,
|
|
}
|
|
values.update(overrides)
|
|
return CanonicalJobDraft(**values)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_dedupe_exact_fuzzy_and_new(job, source):
|
|
document = RawDocument.objects.create(
|
|
source=source,
|
|
kind=RawDocument.Kind.HTML,
|
|
content_hash="e" * 64,
|
|
)
|
|
JobSourceAlias.objects.create(
|
|
job=job,
|
|
source=source,
|
|
raw_document=document,
|
|
url=job.canonical_url,
|
|
canonical_url=job.canonical_url,
|
|
external_id="external-1",
|
|
)
|
|
exact = find_existing_job(_draft(job))
|
|
assert exact.job == job
|
|
assert exact.reason == "exact_external_id"
|
|
|
|
fuzzy_draft = _draft(
|
|
job,
|
|
external_id="",
|
|
canonical_url="https://jobs.example.org/other-id",
|
|
canonical_key="f" * 64,
|
|
)
|
|
fuzzy = find_existing_job(fuzzy_draft, threshold=0.8)
|
|
assert fuzzy.job == job
|
|
assert fuzzy.reason == "fuzzy_strong"
|
|
assert candidate_similarity(job, fuzzy_draft) >= 0.8
|
|
assert text_similarity("", "x") == 0
|
|
|
|
new = find_existing_job(
|
|
_draft(
|
|
job,
|
|
external_id="",
|
|
canonical_url="https://new.example.org/job",
|
|
canonical_key="1" * 64,
|
|
title="Chef kok",
|
|
normalized_title="chef kok",
|
|
employer_name="Restaurant",
|
|
employer_domain="restaurant.example",
|
|
description_text="Koken in een restaurant",
|
|
)
|
|
)
|
|
assert new.job is None
|
|
assert new.reason == "new"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_applied_feedback_creates_application_and_follow_up(user, profile, job):
|
|
feedback = record_feedback(user=user, job=job, action=Feedback.Action.APPLIED, reason="goed")
|
|
assert feedback.profile == profile
|
|
application = Application.objects.get(user=user, job=job)
|
|
assert application.status == Application.Status.APPLIED
|
|
assert application.snapshot["title"] == job.original_title
|
|
assert application.follow_up_date == timezone.localdate() + timedelta(days=7)
|
|
|
|
application.status = Application.Status.PREPARING
|
|
application.save(update_fields=["status"])
|
|
record_feedback(user=user, job=job, action=Feedback.Action.APPLIED)
|
|
application.refresh_from_db()
|
|
assert application.status == Application.Status.APPLIED
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_lifecycle_transitions(employer):
|
|
now = timezone.now()
|
|
expired = JobPosting.objects.create(
|
|
employer=employer,
|
|
original_title="Expired",
|
|
normalized_title="expired",
|
|
canonical_key="2" * 64,
|
|
content_hash="3" * 64,
|
|
valid_through=now - timedelta(minutes=1),
|
|
status=JobPosting.Status.ACTIVE,
|
|
)
|
|
uncertain = JobPosting.objects.create(
|
|
employer=employer,
|
|
original_title="Uncertain",
|
|
normalized_title="uncertain",
|
|
canonical_key="4" * 64,
|
|
content_hash="5" * 64,
|
|
last_seen=now - timedelta(days=5),
|
|
status=JobPosting.Status.ACTIVE,
|
|
)
|
|
removed = JobPosting.objects.create(
|
|
employer=employer,
|
|
original_title="Removed",
|
|
normalized_title="removed",
|
|
canonical_key="6" * 64,
|
|
content_hash="7" * 64,
|
|
last_seen=now - timedelta(days=20),
|
|
status=JobPosting.Status.UNCERTAIN,
|
|
)
|
|
result = update_lifecycle()
|
|
assert result == {"expired": 1, "uncertain": 1, "removed": 1}
|
|
expired.refresh_from_db()
|
|
uncertain.refresh_from_db()
|
|
removed.refresh_from_db()
|
|
assert expired.status == JobPosting.Status.EXPIRED
|
|
assert uncertain.status == JobPosting.Status.UNCERTAIN
|
|
assert removed.status == JobPosting.Status.REMOVED
|