247 lines
8.8 KiB
Python
247 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import zipfile
|
|
from datetime import timedelta
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from django.contrib.auth import get_user_model
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from apps.jobs.models import Application, ApplicationTimelineEvent, JobSourceAlias, ScoreRun
|
|
from apps.jobs.services.feedback import record_feedback
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_applied_feedback_creates_sanitized_application_snapshot_and_timeline(
|
|
job, profile, source, user
|
|
):
|
|
JobSourceAlias.objects.create(
|
|
job=job,
|
|
source=source,
|
|
url=job.canonical_url,
|
|
canonical_url=job.canonical_url,
|
|
external_id="ext-applied-1",
|
|
source_title="Applied page",
|
|
source_employer=job.employer_name,
|
|
is_canonical=True,
|
|
extraction_method="manual-test",
|
|
)
|
|
record_feedback(user=user, job=job, action="applied")
|
|
|
|
application = Application.objects.get(user=user, job=job)
|
|
snapshot = application.snapshot
|
|
assert snapshot["schema_version"] == "1.0.0"
|
|
assert snapshot["snapshot_kind"] == "application"
|
|
assert snapshot["title"] == job.original_title
|
|
assert snapshot["job"]["id"] == str(job.pk)
|
|
assert snapshot["job"]["description_html"] == job.description_html_sanitized
|
|
assert snapshot["sources"][0]["url"] == job.canonical_url
|
|
assert snapshot["sources"][0]["is_canonical"] is True
|
|
assert snapshot["scores"] is None
|
|
|
|
event_types = set(
|
|
ApplicationTimelineEvent.objects.filter(application=application).values_list(
|
|
"event_type", flat=True
|
|
)
|
|
)
|
|
assert ApplicationTimelineEvent.EventType.CREATED in event_types
|
|
assert ApplicationTimelineEvent.EventType.SNAPSHOT_CAPTURED in event_types
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_application_update_creates_timeline_events_for_status_notes_and_contact(
|
|
client, job, profile, source, user
|
|
):
|
|
record_feedback(user=user, job=job, action="applied")
|
|
application = Application.objects.get(user=user, job=job)
|
|
|
|
client.force_login(user)
|
|
response = client.post(
|
|
reverse("jobs:application-edit", kwargs={"pk": application.pk}),
|
|
{
|
|
"status": Application.Status.INTERVIEW,
|
|
"contact_name": "Sofie Van Hecke",
|
|
"contact_email": "recruiter@example.invalid",
|
|
"notes": "Eerste follow-up gepland voor vrijdag.",
|
|
},
|
|
)
|
|
assert response.status_code == 302
|
|
|
|
application.refresh_from_db()
|
|
assert application.status == Application.Status.INTERVIEW
|
|
assert application.contact_name == "Sofie Van Hecke"
|
|
assert application.contact_email == "recruiter@example.invalid"
|
|
|
|
event_types = set(
|
|
ApplicationTimelineEvent.objects.filter(application=application).values_list(
|
|
"event_type", flat=True
|
|
)
|
|
)
|
|
assert ApplicationTimelineEvent.EventType.STATUS_CHANGED in event_types
|
|
assert ApplicationTimelineEvent.EventType.NOTES_UPDATED in event_types
|
|
assert ApplicationTimelineEvent.EventType.CONTACT_UPDATED in event_types
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_pipeline_status_post_is_persistent_audited_and_user_scoped(client, job, user):
|
|
record_feedback(user=user, job=job, action="applied")
|
|
application = Application.objects.get(user=user, job=job)
|
|
client.force_login(user)
|
|
|
|
response = client.post(
|
|
reverse("jobs:application-status", kwargs={"pk": application.pk}),
|
|
{"status": Application.Status.INTERVIEW},
|
|
)
|
|
|
|
assert response.status_code == 302
|
|
application.refresh_from_db()
|
|
assert application.status == Application.Status.INTERVIEW
|
|
assert ApplicationTimelineEvent.objects.filter(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.STATUS_CHANGED,
|
|
metadata__from=Application.Status.APPLIED,
|
|
metadata__to=Application.Status.INTERVIEW,
|
|
).exists()
|
|
|
|
other = get_user_model().objects.create_user(username="scoped-other")
|
|
other_application = Application.objects.create(user=other, job=job)
|
|
assert (
|
|
client.post(
|
|
reverse("jobs:application-status", kwargs={"pk": other_application.pk}),
|
|
{"status": Application.Status.OFFER},
|
|
).status_code
|
|
== 404
|
|
)
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_application_export_generates_sanitized_zip_payload(client, job, source, user, profile):
|
|
JobSourceAlias.objects.create(
|
|
job=job,
|
|
source=source,
|
|
url=job.canonical_url,
|
|
canonical_url=job.canonical_url,
|
|
external_id="ext-export-1",
|
|
source_title="Export page",
|
|
source_employer=job.employer_name,
|
|
is_canonical=True,
|
|
extraction_method="manual-test",
|
|
)
|
|
ScoreRun.objects.create(
|
|
job=job,
|
|
profile=profile,
|
|
profile_version=1,
|
|
score=Decimal("84.20"),
|
|
confidence=Decimal("0.71"),
|
|
recommendation=ScoreRun.Recommendation.POSSIBLE,
|
|
components={"content": 0.41},
|
|
positives=["python"],
|
|
concerns=["travel"],
|
|
hard_exclusions=[],
|
|
evidence={"notes": "Voorbeeldscore"},
|
|
model_version="testsuite",
|
|
prompt_version="1.0.0",
|
|
)
|
|
record_feedback(user=user, job=job, action="applied")
|
|
record_feedback(user=user, job=job, action="applied")
|
|
application = Application.objects.get(user=user, job=job)
|
|
|
|
application.notes = "Vaste follow-up notitie."
|
|
application.save(update_fields=["notes", "updated_at"])
|
|
|
|
client.force_login(user)
|
|
response = client.get(reverse("jobs:application-export", kwargs={"pk": application.pk}))
|
|
assert response.status_code == 200
|
|
assert response["Content-Type"] == "application/zip"
|
|
assert 'filename="' in response["Content-Disposition"]
|
|
|
|
filename = response["Content-Disposition"].partition("filename=")[2].strip().strip('"')
|
|
assert filename.startswith(f"application-{application.pk}-")
|
|
assert filename.endswith(".zip")
|
|
assert ".." not in filename and "/" not in filename and "\\" not in filename
|
|
|
|
archive = zipfile.ZipFile(io.BytesIO(response.content))
|
|
members = sorted(archive.namelist())
|
|
assert members == ["application.json", "application_print.html", "timeline.csv"]
|
|
|
|
application_payload = json.loads(archive.read("application.json").decode("utf-8"))
|
|
assert application_payload["application"]["id"] == str(application.pk)
|
|
assert application_payload["application"]["job_id"] == str(job.pk)
|
|
assert application_payload["application"]["status"] == Application.Status.APPLIED
|
|
assert application_payload["snapshot"]["scores"]["score"] == 84.2
|
|
assert application_payload["snapshot"]["sources"][0]["url"] == job.canonical_url
|
|
|
|
csv_rows = list(csv.DictReader(io.StringIO(archive.read("timeline.csv").decode("utf-8"))))
|
|
assert {row["type"] for row in csv_rows} >= {
|
|
ApplicationTimelineEvent.EventType.CREATED,
|
|
ApplicationTimelineEvent.EventType.SNAPSHOT_CAPTURED,
|
|
}
|
|
|
|
html = archive.read("application_print.html").decode("utf-8")
|
|
assert "<html" in html.lower()
|
|
assert "<link " not in html.lower()
|
|
assert "<script" not in html.lower()
|
|
|
|
assert "raw body" not in json.dumps(application_payload)
|
|
|
|
response_json = client.get(reverse("jobs:application-print", kwargs={"pk": application.pk}))
|
|
assert response_json.status_code == 200
|
|
assert response_json["Content-Type"] == "text/html; charset=utf-8"
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_application_export_and_delete_are_scoped_and_idempotent(client, job, user, source):
|
|
other_user = get_user_model().objects.create_user(
|
|
username="other",
|
|
email="other@example.invalid",
|
|
password="correct-horse-battery-staple",
|
|
)
|
|
|
|
record_feedback(user=user, job=job, action="applied")
|
|
owner_application = Application.objects.get(user=user, job=job)
|
|
other_application = Application.objects.create(
|
|
user=other_user,
|
|
job=job,
|
|
status=Application.Status.APPLIED,
|
|
applied_at=timezone.now() - timedelta(days=1),
|
|
follow_up_date=timezone.now().date(),
|
|
)
|
|
|
|
client.force_login(user)
|
|
assert (
|
|
client.get(
|
|
reverse("jobs:application-export", kwargs={"pk": other_application.pk})
|
|
).status_code
|
|
== 404
|
|
)
|
|
assert (
|
|
client.post(
|
|
reverse("jobs:application-delete", kwargs={"pk": other_application.pk})
|
|
).status_code
|
|
== 404
|
|
)
|
|
|
|
delete_response = client.post(
|
|
reverse("jobs:application-delete", kwargs={"pk": owner_application.pk}),
|
|
follow=True,
|
|
)
|
|
assert delete_response.status_code == 200
|
|
assert not Application.objects.filter(pk=owner_application.pk).exists()
|
|
|
|
repeat_delete = client.post(
|
|
reverse("jobs:application-delete", kwargs={"pk": owner_application.pk}),
|
|
follow=True,
|
|
)
|
|
assert repeat_delete.status_code == 200
|