207 lines
7.7 KiB
Python
207 lines
7.7 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from cryptography.fernet import Fernet
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import Client, override_settings
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from apps.sources.models import MailboxConnection
|
|
from apps.sources.services.mailbox_connections import (
|
|
claim_mailbox_connection,
|
|
decrypt_mailbox_password,
|
|
)
|
|
from apps.sources.tasks import poll_mailbox, schedule_mailbox_polls
|
|
|
|
|
|
def _payload(**overrides):
|
|
payload = {
|
|
"platform": MailboxConnection.Platform.VDAB,
|
|
"provider": MailboxConnection.Provider.GMAIL,
|
|
"custom_host": "",
|
|
"port": 993,
|
|
"username": "vdab-alerts@example.invalid",
|
|
"password": "test-app-password",
|
|
"mailbox": "INBOX",
|
|
"poll_interval_minutes": 15,
|
|
"enabled": "on",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_user_can_link_edit_pause_and_sync_separate_platform_mailboxes(client, user, monkeypatch):
|
|
key = Fernet.generate_key().decode("ascii")
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
"apps.sources.views.poll_mailbox.delay",
|
|
lambda connection_id, force=False: calls.append((connection_id, force)),
|
|
)
|
|
client.force_login(user)
|
|
|
|
with override_settings(MAILBOX_CREDENTIAL_KEYS=[key]):
|
|
response = client.post(reverse("sources:mailbox_add"), _payload())
|
|
assert response.status_code == 302
|
|
vdab = MailboxConnection.objects.get(user=user, platform="vdab")
|
|
assert decrypt_mailbox_password(vdab) == "test-app-password"
|
|
assert "test-app-password" not in vdab.encrypted_password
|
|
|
|
response = client.post(
|
|
reverse("sources:mailbox_add"),
|
|
_payload(
|
|
platform=MailboxConnection.Platform.INDEED,
|
|
provider=MailboxConnection.Provider.OUTLOOK,
|
|
username="indeed-alerts@example.invalid",
|
|
password="second-app-password",
|
|
poll_interval_minutes=30,
|
|
),
|
|
)
|
|
assert response.status_code == 302
|
|
indeed = MailboxConnection.objects.get(user=user, platform="indeed")
|
|
assert indeed.encrypted_password == ""
|
|
|
|
edit_response = client.get(reverse("sources:mailbox_edit", args=[vdab.pk]))
|
|
assert edit_response.status_code == 200
|
|
assert b"test-app-password" not in edit_response.content
|
|
assert vdab.encrypted_password.encode() not in edit_response.content
|
|
|
|
original_ciphertext = vdab.encrypted_password
|
|
response = client.post(
|
|
reverse("sources:mailbox_edit", args=[vdab.pk]),
|
|
_payload(
|
|
username="vdab-updated@example.invalid",
|
|
password="",
|
|
poll_interval_minutes=5,
|
|
),
|
|
)
|
|
assert response.status_code == 302
|
|
vdab.refresh_from_db()
|
|
assert vdab.username == "vdab-updated@example.invalid"
|
|
assert vdab.poll_interval_minutes == 5
|
|
assert vdab.encrypted_password == original_ciphertext
|
|
assert decrypt_mailbox_password(vdab) == "test-app-password"
|
|
|
|
response = client.post(reverse("sources:mailbox_toggle", args=[indeed.pk]))
|
|
assert response.status_code == 302
|
|
indeed.refresh_from_db()
|
|
assert indeed.enabled is False
|
|
|
|
response = client.post(reverse("sources:sync_platform_alerts", args=[vdab.pk]))
|
|
assert response.status_code == 302
|
|
assert calls == [
|
|
(vdab.pk, True),
|
|
(indeed.pk, True),
|
|
(vdab.pk, True),
|
|
(vdab.pk, True),
|
|
]
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_sources_page_offers_all_reviewed_jobboard_mailboxes(client, user):
|
|
client.force_login(user)
|
|
|
|
response = client.get(reverse("sources:list"))
|
|
|
|
assert response.status_code == 200
|
|
for platform in MailboxConnection.Platform:
|
|
assert f'value="{platform.value}"'.encode() in response.content
|
|
assert f"{platform.label}-alert instellen".encode() in response.content
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_mailbox_views_are_user_scoped_and_require_encryption_key(client, user, monkeypatch):
|
|
other = get_user_model().objects.create_user(username="other", password="test-only-password")
|
|
other_connection = MailboxConnection.objects.create(
|
|
user=other,
|
|
platform=MailboxConnection.Platform.VDAB,
|
|
provider=MailboxConnection.Provider.GMAIL,
|
|
username="private@example.invalid",
|
|
encrypted_password="opaque-ciphertext",
|
|
)
|
|
client.force_login(user)
|
|
monkeypatch.setattr("apps.sources.views.poll_mailbox.delay", lambda *args, **kwargs: None)
|
|
|
|
assert (
|
|
client.get(reverse("sources:mailbox_edit", args=[other_connection.pk])).status_code == 404
|
|
)
|
|
assert (
|
|
client.post(reverse("sources:sync_platform_alerts", args=[other_connection.pk])).status_code
|
|
== 404
|
|
)
|
|
assert (
|
|
client.post(reverse("sources:mailbox_toggle", args=[other_connection.pk])).status_code
|
|
== 404
|
|
)
|
|
|
|
with override_settings(MAILBOX_CREDENTIAL_KEYS=[]):
|
|
response = client.post(reverse("sources:mailbox_add"), _payload(password="visible-once"))
|
|
assert response.status_code == 200
|
|
assert MailboxConnection.objects.filter(user=user).count() == 0
|
|
assert b"visible-once" not in response.content
|
|
assert b"Credentialopslag is niet geconfigureerd" in response.content
|
|
|
|
anonymous_response = Client().post(reverse("sources:mailbox_add"), _payload())
|
|
assert anonymous_response.status_code == 302
|
|
csrf_client = Client(enforce_csrf_checks=True)
|
|
csrf_client.force_login(user)
|
|
assert csrf_client.post(reverse("sources:mailbox_add"), _payload()).status_code == 403
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_scheduler_queues_only_due_active_mailboxes_and_claim_is_exclusive(user, monkeypatch):
|
|
now = timezone.now()
|
|
due = MailboxConnection.objects.create(
|
|
user=user,
|
|
platform=MailboxConnection.Platform.VDAB,
|
|
provider=MailboxConnection.Provider.GMAIL,
|
|
username="due@example.invalid",
|
|
encrypted_password="opaque",
|
|
next_poll_at=now - timedelta(minutes=1),
|
|
)
|
|
MailboxConnection.objects.create(
|
|
user=user,
|
|
platform=MailboxConnection.Platform.INDEED,
|
|
provider=MailboxConnection.Provider.OUTLOOK,
|
|
username="later@example.invalid",
|
|
encrypted_password="opaque",
|
|
next_poll_at=now + timedelta(minutes=20),
|
|
)
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
"apps.sources.tasks.poll_mailbox.delay", lambda connection_id: calls.append(connection_id)
|
|
)
|
|
|
|
assert schedule_mailbox_polls() == {"scheduled": 1}
|
|
assert calls == [due.pk]
|
|
assert claim_mailbox_connection(connection_id=due.pk, worker_token="worker-a") is not None
|
|
assert claim_mailbox_connection(connection_id=due.pk, worker_token="worker-b") is None
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.django_db
|
|
def test_mailbox_task_does_not_persist_or_return_secret_on_imap_error(user, monkeypatch):
|
|
connection = MailboxConnection.objects.create(
|
|
user=user,
|
|
platform=MailboxConnection.Platform.VDAB,
|
|
provider=MailboxConnection.Provider.GMAIL,
|
|
username="errors@example.invalid",
|
|
encrypted_password="opaque",
|
|
next_poll_at=timezone.now() - timedelta(minutes=1),
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.sources.tasks.poll_imap_mailbox",
|
|
lambda connection: (_ for _ in ()).throw(RuntimeError("leaked-app-password")),
|
|
)
|
|
|
|
result = poll_mailbox(connection.pk)
|
|
connection.refresh_from_db()
|
|
assert "leaked-app-password" not in str(result)
|
|
assert "leaked-app-password" not in connection.last_error_message
|
|
assert result == {"status": "failed", "error_category": "imap", "imported": 0}
|