mirror of
https://github.com/suitenumerique/people
synced 2026-04-25 17:15:13 +02:00
♻️(refacto) refacto responses
WIP
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
"""Tests for the authentication process of the resource server."""
|
||||
|
||||
import base64
|
||||
import json
|
||||
|
||||
import pytest
|
||||
import responses
|
||||
@@ -84,8 +83,7 @@ def test_resource_server_authentication_class(client, settings):
|
||||
settings.OIDC_OP_JWKS_ENDPOINT = "https://oidc.example.com/jwks"
|
||||
settings.OIDC_OP_INTROSPECTION_ENDPOINT = "https://oidc.example.com/introspect"
|
||||
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
"https://oidc.example.com/introspect",
|
||||
json={
|
||||
"iss": "https://oidc.example.com",
|
||||
@@ -181,23 +179,20 @@ def test_jwt_resource_server_authentication_class( # pylint: disable=unused-arg
|
||||
|
||||
# Mock the JWKS endpoint
|
||||
public_numbers = private_key.public_key().public_numbers()
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
settings.OIDC_OP_JWKS_ENDPOINT,
|
||||
body=json.dumps(
|
||||
{
|
||||
"keys": [
|
||||
{
|
||||
"kty": settings.OIDC_RS_ENCRYPTION_KEY_TYPE,
|
||||
"alg": settings.OIDC_RS_SIGNING_ALGO,
|
||||
"use": "sig",
|
||||
"kid": "1234567890",
|
||||
"n": to_base64url_uint(public_numbers.n).decode("ascii"),
|
||||
"e": to_base64url_uint(public_numbers.e).decode("ascii"),
|
||||
}
|
||||
]
|
||||
}
|
||||
),
|
||||
json={
|
||||
"keys": [
|
||||
{
|
||||
"kty": settings.OIDC_RS_ENCRYPTION_KEY_TYPE,
|
||||
"alg": settings.OIDC_RS_SIGNING_ALGO,
|
||||
"use": "sig",
|
||||
"kid": "1234567890",
|
||||
"n": to_base64url_uint(public_numbers.n).decode("ascii"),
|
||||
"e": to_base64url_uint(public_numbers.e).decode("ascii"),
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
def encrypt_jwt(json_data):
|
||||
@@ -225,8 +220,7 @@ def test_jwt_resource_server_authentication_class( # pylint: disable=unused-arg
|
||||
],
|
||||
)
|
||||
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
"https://oidc.example.com/introspect",
|
||||
body=encrypt_jwt(
|
||||
{
|
||||
|
||||
@@ -37,7 +37,7 @@ def test_matrix_webhook__search_user_unknown(caplog):
|
||||
# Mock successful responses
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_empty()["message"]),
|
||||
json=matrix.mock_search_empty()["message"],
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
@@ -62,7 +62,7 @@ def test_matrix_webhook__search_multiple_ids(caplog):
|
||||
# Mock successful responses
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful_multiple(user)["message"]),
|
||||
json=matrix.mock_search_successful_multiple(user)["message"],
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
@@ -92,7 +92,7 @@ def test_matrix_webhook__invite_user_to_room_forbidden(caplog):
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful(user)["message"]),
|
||||
json=matrix.mock_search_successful(user)["message"],
|
||||
status=matrix.mock_search_successful(user)["status_code"],
|
||||
)
|
||||
responses.post(
|
||||
@@ -123,7 +123,7 @@ def test_matrix_webhook__invite_user_to_room_already_in_room(caplog):
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful(user)["message"]),
|
||||
json=matrix.mock_search_successful(user)["message"],
|
||||
status=matrix.mock_search_successful(user)["status_code"],
|
||||
)
|
||||
responses.post(
|
||||
@@ -165,7 +165,7 @@ def test_matrix_webhook__invite_user_to_room_success(caplog):
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful(user)["message"]),
|
||||
json=matrix.mock_search_successful(user)["message"],
|
||||
status=matrix.mock_search_successful(user)["status_code"],
|
||||
)
|
||||
responses.post(
|
||||
@@ -216,7 +216,7 @@ def test_matrix_webhook__override_secret_for_tchap():
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful(user)["message"]),
|
||||
json=matrix.mock_search_successful(user)["message"],
|
||||
status=matrix.mock_search_successful(user)["status_code"],
|
||||
)
|
||||
responses.post(
|
||||
@@ -250,17 +250,17 @@ def test_matrix_webhook__kick_user_from_room_not_in_room(caplog):
|
||||
# Mock successful responses
|
||||
responses.post(
|
||||
re.compile(r".*/join"),
|
||||
body=str(matrix.mock_join_room_successful),
|
||||
json=matrix.mock_join_room_successful,
|
||||
status=status.HTTP_200_OK,
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful(user)["message"]),
|
||||
json=matrix.mock_search_successful(user)["message"],
|
||||
status=matrix.mock_search_successful(user)["status_code"],
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/kick"),
|
||||
body=str(matrix.mock_kick_user_not_in_room()["message"]),
|
||||
json=matrix.mock_kick_user_not_in_room()["message"],
|
||||
status=matrix.mock_kick_user_not_in_room()["status_code"],
|
||||
)
|
||||
webhooks_synchronizer.remove_user_from_group(team=webhook.team, user=user)
|
||||
@@ -296,7 +296,7 @@ def test_matrix_webhook__kick_user_from_room_success(caplog):
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful(user)["message"]),
|
||||
json=matrix.mock_search_successful(user)["message"],
|
||||
status=matrix.mock_search_successful(user)["status_code"],
|
||||
)
|
||||
responses.post(
|
||||
@@ -345,7 +345,7 @@ def test_matrix_webhook__kick_user_from_room_forbidden(caplog):
|
||||
)
|
||||
responses.post(
|
||||
re.compile(r".*/search"),
|
||||
body=json.dumps(matrix.mock_search_successful(user)["message"]),
|
||||
json=matrix.mock_search_successful(user)["message"],
|
||||
status=matrix.mock_search_successful(user)["status_code"],
|
||||
)
|
||||
responses.post(
|
||||
|
||||
@@ -4,7 +4,6 @@ Focus on "create" action.
|
||||
"""
|
||||
# pylint: disable=W0613
|
||||
|
||||
import json
|
||||
import re
|
||||
|
||||
import pytest
|
||||
@@ -94,10 +93,9 @@ def test_api_aliases_create__async_alias_bad_request(dimail_token_ok):
|
||||
client = APIClient()
|
||||
client.force_login(access.user)
|
||||
# Mock dimail response
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(r".*/aliases/"),
|
||||
body=json.dumps({"detail": "Alias already exists"}),
|
||||
json={"detail": "Alias already exists"},
|
||||
status=status.HTTP_409_CONFLICT,
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
@@ -68,8 +68,7 @@ def test_api_mail_domains__create_authenticated():
|
||||
|
||||
domain_name = "test.domain.fr"
|
||||
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(r".*/domains/"),
|
||||
body=str(
|
||||
{
|
||||
@@ -81,17 +80,15 @@ def test_api_mail_domains__create_authenticated():
|
||||
)
|
||||
body_content_domain1 = CHECK_DOMAIN_BROKEN.copy()
|
||||
body_content_domain1["name"] = domain_name
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain_name}/check/"),
|
||||
body=json.dumps(body_content_domain1),
|
||||
json=body_content_domain1,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain_name}/spec/"),
|
||||
body=json.dumps(DOMAIN_SPEC),
|
||||
json=DOMAIN_SPEC,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
@@ -155,30 +152,25 @@ def test_api_mail_domains__create_dimail_domain(caplog):
|
||||
client.force_login(user)
|
||||
domain_name = "test.fr"
|
||||
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(r".*/domains/"),
|
||||
body=str(
|
||||
{
|
||||
"name": domain_name,
|
||||
}
|
||||
),
|
||||
json={
|
||||
"name": domain_name,
|
||||
}
|
||||
status=status.HTTP_201_CREATED,
|
||||
content_type="application/json",
|
||||
)
|
||||
body_content_domain1 = CHECK_DOMAIN_OK.copy()
|
||||
body_content_domain1["name"] = domain_name
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain_name}/check/"),
|
||||
body=json.dumps(body_content_domain1),
|
||||
json=body_content_domain1,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain_name}/spec/"),
|
||||
body=json.dumps(DOMAIN_SPEC),
|
||||
json=DOMAIN_SPEC,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
@@ -213,8 +205,7 @@ def test_api_mail_domains__no_creation_when_dimail_duplicate(caplog):
|
||||
"status_code": status.HTTP_409_CONFLICT,
|
||||
"detail": "Domain already exists",
|
||||
}
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(r".*/domains/"),
|
||||
body=str({"detail": dimail_error["detail"]}),
|
||||
status=dimail_error["status_code"],
|
||||
|
||||
@@ -104,14 +104,12 @@ def test_api_mail_domains__fetch_from_dimail_admin_successful(role):
|
||||
assert domain.expected_config is None
|
||||
assert domain.last_check_details is None
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain.name}/check/"),
|
||||
json=dimail_fixtures.CHECK_DOMAIN_OK,
|
||||
status=200,
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain.name}/spec/"),
|
||||
json=dimail_fixtures.DOMAIN_SPEC,
|
||||
status=200,
|
||||
|
||||
@@ -122,8 +122,7 @@ def test_api_mailboxes__create_display_name_no_constraint_on_different_domains(
|
||||
|
||||
# ensure response
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{new_mailbox_data['local_part']}@{access.domain.name}"
|
||||
@@ -160,8 +159,7 @@ def test_api_mailboxes__create_roles_success(role, dimail_token_ok, mailbox_data
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{mail_domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{mailbox_data['local_part']}@{mail_domain.name}"
|
||||
@@ -210,8 +208,7 @@ def test_api_mailboxes__create_with_accent_success(role, dimail_token_ok):
|
||||
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{mail_domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{mailbox_values['local_part']}@{mail_domain.name}"
|
||||
@@ -285,8 +282,7 @@ def test_api_mailboxes__create_without_secondary_email(role, caplog, dimail_toke
|
||||
del mailbox_values["secondary_email"]
|
||||
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{mail_domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{mailbox_values['local_part']}@{mail_domain.name}"
|
||||
@@ -386,8 +382,7 @@ def test_api_mailboxes__same_local_part_on_different_domains(dimail_token_ok):
|
||||
factories.MailboxFactory.build(local_part=existing_mailbox.local_part)
|
||||
).data
|
||||
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{mailbox_values['local_part']}@{access.domain.name}"
|
||||
@@ -520,8 +515,7 @@ def test_api_mailboxes__async_dimail_unauthorized(
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(
|
||||
rf".*/domains/{access.domain.name}/mailboxes/{mailbox_data['local_part']}"
|
||||
),
|
||||
@@ -562,8 +556,7 @@ def test_api_mailboxes__domain_owner_or_admin_successful_creation_and_provisioni
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{mailbox_data['local_part']}@{access.domain.name}"
|
||||
@@ -620,8 +613,7 @@ def test_api_mailboxes__domain_owner_or_admin_successful_creation_sets_password(
|
||||
client = APIClient()
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{mailbox_data['local_part']}@{access.domain.name}"
|
||||
@@ -663,8 +655,7 @@ def test_api_mailboxes__dimail_token_permission_denied(caplog, mailbox_data):
|
||||
client = APIClient()
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(r".*/token/"),
|
||||
body='{"details": "Permission denied"}',
|
||||
status=status.HTTP_403_FORBIDDEN,
|
||||
@@ -708,8 +699,7 @@ def test_api_mailboxes__user_unrelated_to_domain(dimail_token_ok, mailbox_data):
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body='{"details": "Permission denied"}',
|
||||
status=status.HTTP_403_FORBIDDEN,
|
||||
@@ -744,15 +734,13 @@ def test_api_mailboxes__duplicate_display_name(dimail_token_ok, mailbox_data):
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body='{"detail": "Internal server error"}',
|
||||
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(
|
||||
rf".*/domains/{access.domain.name}/address/{mailbox_data['local_part']}"
|
||||
),
|
||||
@@ -801,15 +789,13 @@ def test_api_mailboxes__handling_dimail_unexpected_error(
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body='{"detail": "Internal server error"}',
|
||||
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(
|
||||
rf".*/domains/{access.domain.name}/address/{mailbox_data['local_part']}/"
|
||||
),
|
||||
@@ -852,15 +838,13 @@ def test_api_mailboxes__display_name_duplicate_error(dimail_token_ok, mailbox_da
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body='{"detail": "Internal server error"}',
|
||||
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(
|
||||
rf".*/domains/{access.domain.name}/address/{mailbox_data['local_part']}/"
|
||||
),
|
||||
@@ -911,8 +895,7 @@ def test_api_mailboxes__send_correct_logger_infos(
|
||||
client.force_login(access.user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(
|
||||
f"{mailbox_data['local_part']}@{access.domain.name}"
|
||||
@@ -962,8 +945,7 @@ def test_api_mailboxes__sends_new_mailbox_notification(
|
||||
client.force_login(user)
|
||||
# Ensure successful response using "responses":
|
||||
# token response in fixtures
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{access.domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(f"{mailbox_data['local_part']}@{access.domain}"),
|
||||
status=status.HTTP_201_CREATED,
|
||||
|
||||
@@ -113,7 +113,7 @@ Please add a valid secondary email before trying again."
|
||||
],
|
||||
)
|
||||
@responses.activate
|
||||
def test_api_mailboxes__reset_password_admin_successful(role):
|
||||
def test_api_mailboxes__reset_password_admin_successful(role, dimail_token_ok):
|
||||
"""Owner and admin users should be able to reset password on mailboxes.
|
||||
New password should be sent to secondary email."""
|
||||
mail_domain = factories.MailDomainEnabledFactory()
|
||||
@@ -123,15 +123,7 @@ def test_api_mailboxes__reset_password_admin_successful(role):
|
||||
client = APIClient()
|
||||
client.force_login(access.user)
|
||||
dimail_url = settings.MAIL_PROVISIONING_API_URL
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
f"{dimail_url}/token/",
|
||||
body=dimail.TOKEN_OK,
|
||||
status=200,
|
||||
)
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
f"{dimail_url}/domains/{mail_domain.name}/mailboxes/{mailbox.local_part}/reset_password/",
|
||||
body=dimail.response_mailbox_created(str(mailbox)),
|
||||
status=200,
|
||||
@@ -161,7 +153,7 @@ def test_api_mailboxes__reset_password_non_existing():
|
||||
|
||||
|
||||
@responses.activate
|
||||
def test_api_mailboxes__reset_password_connexion_failed():
|
||||
def test_api_mailboxes__reset_password_connexion_failed(dimail_token_ok):
|
||||
"""
|
||||
No mail is sent when password reset failed because of connexion error.
|
||||
"""
|
||||
@@ -173,16 +165,8 @@ def test_api_mailboxes__reset_password_connexion_failed():
|
||||
)
|
||||
client = APIClient()
|
||||
client.force_login(access.user)
|
||||
|
||||
dimail_url = settings.MAIL_PROVISIONING_API_URL
|
||||
responses.add(
|
||||
responses.GET,
|
||||
f"{dimail_url}/token/",
|
||||
body=dimail.TOKEN_OK,
|
||||
status=200,
|
||||
)
|
||||
responses.add(
|
||||
responses.POST,
|
||||
responses.post(
|
||||
f"{dimail_url}/domains/{mail_domain.name}/mailboxes/{mailbox.local_part}/reset_password/",
|
||||
body=ConnectionError(),
|
||||
)
|
||||
|
||||
@@ -48,8 +48,7 @@ def test_fetch_domain_status():
|
||||
(domain_failed, body_content_ok3),
|
||||
]:
|
||||
# mock dimail API
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain.name}/check/"),
|
||||
body=json.dumps(body_content),
|
||||
status=status.HTTP_200_OK,
|
||||
|
||||
@@ -9,17 +9,15 @@ import responses
|
||||
from rest_framework import status
|
||||
|
||||
from mailbox_manager import factories
|
||||
from mailbox_manager.tests.fixtures.dimail import TOKEN_OK
|
||||
|
||||
|
||||
## DIMAIL RESPONSES
|
||||
@pytest.fixture(name="dimail_token_ok")
|
||||
def fixture_dimail_token_ok():
|
||||
"""Mock dimail response when /token/ endpoit is given valid credentials."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(r".*/token/"),
|
||||
body=TOKEN_OK,
|
||||
json={"access_token": "token", "token_type": "bearer"},
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
@@ -287,10 +287,6 @@ DOMAIN_SPEC = [
|
||||
]
|
||||
|
||||
|
||||
## TOKEN
|
||||
TOKEN_OK = json.dumps({"access_token": "token", "token_type": "bearer"})
|
||||
|
||||
|
||||
## ALLOWS
|
||||
def response_allows_created(user_name, domain_name):
|
||||
"""mimic dimail response upon successful allows creation.
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
Unit tests for admin actions
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
|
||||
from django.urls import reverse
|
||||
@@ -19,7 +18,6 @@ from .fixtures.dimail import (
|
||||
CHECK_DOMAIN_BROKEN,
|
||||
CHECK_DOMAIN_OK,
|
||||
DOMAIN_SPEC,
|
||||
TOKEN_OK,
|
||||
response_mailbox_created,
|
||||
)
|
||||
|
||||
@@ -76,17 +74,15 @@ def test_fetch_domain_status__should_switch_to_failed_when_domain_broken(client)
|
||||
body_content_domain1["name"] = domain1.name
|
||||
body_content_domain2 = CHECK_DOMAIN_BROKEN.copy()
|
||||
body_content_domain2["name"] = domain2.name
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain1.name}/check/"),
|
||||
body=json.dumps(body_content_domain1),
|
||||
json=body_content_domain1,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain2.name}/check/"),
|
||||
body=json.dumps(body_content_domain2),
|
||||
json=body_content_domain2,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
@@ -101,7 +97,9 @@ def test_fetch_domain_status__should_switch_to_failed_when_domain_broken(client)
|
||||
|
||||
@responses.activate
|
||||
@pytest.mark.django_db
|
||||
def test_fetch_domain_status__should_switch_to_enabled_when_domain_ok(client):
|
||||
def test_fetch_domain_status__should_switch_to_enabled_when_domain_ok(
|
||||
client, dimail_token_ok
|
||||
):
|
||||
"""Test admin action should switch domain state to ENABLED
|
||||
when dimail's response is "ok". It should also activate any pending mailbox."""
|
||||
admin = core_factories.UserFactory(is_staff=True, is_superuser=True)
|
||||
@@ -119,22 +117,14 @@ def test_fetch_domain_status__should_switch_to_enabled_when_domain_ok(client):
|
||||
body_content_domain1 = CHECK_DOMAIN_OK.copy()
|
||||
body_content_domain1["name"] = domain1.name
|
||||
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain1.name}/check/"),
|
||||
body=json.dumps(body_content_domain1),
|
||||
json=body_content_domain1,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
# we need to get a token to create mailboxes
|
||||
responses.add(
|
||||
responses.GET,
|
||||
re.compile(r".*/token/"),
|
||||
body=TOKEN_OK,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
# token call in fixtures
|
||||
responses.post(
|
||||
responses.POST,
|
||||
re.compile(rf".*/domains/{domain1.name}/mailboxes/"),
|
||||
body=response_mailbox_created(f"truc@{domain1.name}"),
|
||||
@@ -172,10 +162,9 @@ def test_fetch_domain_expected_config(client, domain_status):
|
||||
"action": "fetch_domain_expected_config_from_dimail",
|
||||
"_selected_action": [domain.id],
|
||||
}
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain.name}/spec/"),
|
||||
body=json.dumps(DOMAIN_SPEC),
|
||||
json=DOMAIN_SPEC,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
@@ -208,7 +197,7 @@ def test_fetch_domain_expected_config__should_not_fetch_for_disabled_domain(clie
|
||||
|
||||
@responses.activate
|
||||
@pytest.mark.django_db
|
||||
def test_send_pending_mailboxes(client):
|
||||
def test_send_pending_mailboxes(client, dimail_token_ok):
|
||||
"""Test admin action to send pending mailboxes to dimail."""
|
||||
admin = core_factories.UserFactory(is_staff=True, is_superuser=True)
|
||||
client.force_login(admin)
|
||||
@@ -223,15 +212,8 @@ def test_send_pending_mailboxes(client):
|
||||
|
||||
url = reverse("admin:mailbox_manager_maildomain_changelist")
|
||||
for mailbox in mailboxes:
|
||||
responses.add(
|
||||
responses.GET,
|
||||
re.compile(r".*/token/"),
|
||||
body=TOKEN_OK,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.POST,
|
||||
# token call in fixture
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(f"{mailbox.local_part}@{domain.name}"),
|
||||
status=status.HTTP_201_CREATED,
|
||||
@@ -247,7 +229,7 @@ def test_send_pending_mailboxes(client):
|
||||
|
||||
@responses.activate
|
||||
@pytest.mark.django_db
|
||||
def test_send_pending_mailboxes__listing_failed_mailboxes(client):
|
||||
def test_send_pending_mailboxes__listing_failed_mailboxes(client, dimail_token_ok):
|
||||
"""Test admin action to send pending mailboxes to dimail."""
|
||||
admin = core_factories.UserFactory(is_staff=True, is_superuser=True)
|
||||
client.force_login(admin)
|
||||
@@ -261,15 +243,8 @@ def test_send_pending_mailboxes__listing_failed_mailboxes(client):
|
||||
}
|
||||
|
||||
url = reverse("admin:mailbox_manager_maildomain_changelist")
|
||||
responses.add(
|
||||
responses.GET,
|
||||
re.compile(r".*/token/"),
|
||||
body=TOKEN_OK,
|
||||
status=status.HTTP_200_OK,
|
||||
content_type="application/json",
|
||||
)
|
||||
responses.add(
|
||||
responses.POST,
|
||||
# token call in fixtures
|
||||
responses.post(
|
||||
re.compile(rf".*/domains/{domain.name}/mailboxes/"),
|
||||
body=response_mailbox_created(f"{mailbox.local_part}@{domain.name}"),
|
||||
status=status.HTTP_409_CONFLICT,
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
Unit tests for mailbox manager tasks.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from unittest import mock
|
||||
|
||||
@@ -63,18 +62,16 @@ def test_fetch_domain_status_task_success(): # pylint: disable=too-many-locals
|
||||
(domain_failed, body_content_ok3),
|
||||
]:
|
||||
# Mock dimail API with success response
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain.name}/check/"),
|
||||
body=json.dumps(body_content),
|
||||
json=body_content,
|
||||
status=200,
|
||||
content_type="application/json",
|
||||
)
|
||||
# domain_enabled2 is broken with internal error, we try to fix it
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain_enabled2.name}/fix/"),
|
||||
body=json.dumps(body_content_broken_internal),
|
||||
json=body_content_broken_internal,
|
||||
status=200,
|
||||
content_type="application/json",
|
||||
)
|
||||
@@ -170,10 +167,9 @@ def test_fetch_domains_status_error_handling(caplog):
|
||||
domain = factories.MailDomainEnabledFactory()
|
||||
|
||||
# Mock dimail API with error response
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
re.compile(rf".*/domains/{domain.name}/check/"),
|
||||
body=json.dumps({"error": "Internal Server Error"}),
|
||||
json={"error": "Internal Server Error"},
|
||||
status=500,
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
@@ -61,8 +61,7 @@ def test_organization_plugins_run_after_create(
|
||||
hook_settings, nature_juridique, is_commune, is_public_service
|
||||
):
|
||||
"""Test the run_after_create method of the organization plugins for nominal case."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json={
|
||||
"results": [
|
||||
@@ -106,8 +105,7 @@ def test_organization_plugins_run_after_create(
|
||||
@responses.activate
|
||||
def test_organization_plugins_run_after_create_api_fail(hook_settings):
|
||||
"""Test the plugin when the API call fails."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json={"error": "Internal Server Error"},
|
||||
status=500,
|
||||
@@ -139,8 +137,7 @@ def test_organization_plugins_run_after_create_api_fail(hook_settings):
|
||||
)
|
||||
def test_organization_plugins_run_after_create_missing_data(hook_settings, results):
|
||||
"""Test the plugin when the API call returns missing data."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json=results,
|
||||
status=200,
|
||||
@@ -168,8 +165,7 @@ def test_organization_plugins_run_after_create_no_list_enseignes(
|
||||
hook_settings,
|
||||
):
|
||||
"""Test the run_after_create method of the organization plugins for nominal case."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
responses.get(
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json={
|
||||
"results": [
|
||||
|
||||
Reference in New Issue
Block a user