mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-03 06:40:31 +00:00
Trenton H
b457610ffb
* Fix: redirect SHARE_LINK_BUNDLE_DIR to the test temp layout instead of the real media directory * Fix: include f_to in test_filters subTest labels so each of the 8 cases reports distinctly * Fix: run the post_consume error-log assertion after the raising call and match the actual paperless_mail logger name * Fix: rename the blank-password workflow test to match its behavior and add a real wrong-password-fails test * Fix: use the created social account's actual pk and remove an accidental tuple wrapping the mock provider * Fix: assert against the created documents' actual pks instead of hardcoded 1 and 2 * Fix: assert test_compression actually produces a valid LZMA-compressed zip * Fix: clear os.environ when patching PAPERLESS_ADMIN_* vars so a host-set value can't leak into the no-user test * Fix: restore MIDDLEWARE, AUTHENTICATION_BACKENDS and REST_FRAMEWORK auth classes after each remote-user settings test instead of leaking the mutation into later tests * Fix: use a guaranteed-nonexistent temp path instead of hardcoded /tmp/foo/bar in test_export_target_not_exists
474 lines
14 KiB
Python
474 lines
14 KiB
Python
from unittest import mock
|
|
|
|
from allauth.mfa.models import Authenticator
|
|
from allauth.socialaccount.models import SocialAccount
|
|
from allauth.socialaccount.models import SocialApp
|
|
from django.contrib.auth.models import User
|
|
from rest_framework import status
|
|
from rest_framework.authtoken.models import Token
|
|
from rest_framework.test import APITestCase
|
|
|
|
from paperless_testing.dirs import DirectoriesMixin
|
|
from paperless_testing.factories import UserFactory
|
|
|
|
|
|
# see allauth.socialaccount.providers.openid.provider.OpenIDProvider
|
|
class MockOpenIDProvider:
|
|
id = "openid"
|
|
name = "OpenID"
|
|
|
|
def get_brands(self):
|
|
default_servers = [
|
|
dict(id="yahoo", name="Yahoo", openid_url="http://me.yahoo.com"),
|
|
dict(id="hyves", name="Hyves", openid_url="http://hyves.nl"),
|
|
]
|
|
return default_servers
|
|
|
|
def get_login_url(self, request, **kwargs):
|
|
return "openid/login/"
|
|
|
|
|
|
# see allauth.socialaccount.providers.openid_connect.provider.OpenIDConnectProviderAccount
|
|
class MockOpenIDConnectProviderAccount:
|
|
def __init__(self, mock_social_account_dict) -> None:
|
|
self.account = mock_social_account_dict
|
|
|
|
def to_str(self):
|
|
return self.account["name"]
|
|
|
|
|
|
# see allauth.socialaccount.providers.openid_connect.provider.OpenIDConnectProvider
|
|
class MockOpenIDConnectProvider:
|
|
id = "openid_connect"
|
|
name = "OpenID Connect"
|
|
|
|
def __init__(self, app=None) -> None:
|
|
self.app = app
|
|
self.name = app.name
|
|
|
|
def get_login_url(self, request, **kwargs):
|
|
return f"{self.app.provider_id}/login/?process=connect"
|
|
|
|
|
|
class TestApiProfile(DirectoriesMixin, APITestCase):
|
|
ENDPOINT = "/api/profile/"
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
|
|
self.user = UserFactory(
|
|
username="temp_admin",
|
|
first_name="firstname",
|
|
last_name="surname",
|
|
superuser=True,
|
|
)
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def setupSocialAccount(self) -> SocialAccount:
|
|
SocialApp.objects.create(
|
|
name="Keycloak",
|
|
provider="openid_connect",
|
|
provider_id="keycloak-test",
|
|
)
|
|
social_account = SocialAccount(uid="123456789", provider="keycloak-test")
|
|
self.user.socialaccount_set.add(social_account, bulk=False)
|
|
return social_account
|
|
|
|
def test_get_profile(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user
|
|
WHEN:
|
|
- API call is made to get profile
|
|
THEN:
|
|
- Profile is returned
|
|
"""
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
self.assertEqual(response.data["email"], self.user.email)
|
|
self.assertEqual(response.data["first_name"], self.user.first_name)
|
|
self.assertEqual(response.data["last_name"], self.user.last_name)
|
|
|
|
@mock.patch(
|
|
"allauth.socialaccount.models.SocialAccount.get_provider_account",
|
|
)
|
|
@mock.patch(
|
|
"allauth.socialaccount.adapter.DefaultSocialAccountAdapter.list_providers",
|
|
)
|
|
def test_get_profile_w_social(
|
|
self,
|
|
mock_list_providers,
|
|
mock_get_provider_account,
|
|
) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user and setup social account
|
|
WHEN:
|
|
- API call is made to get profile
|
|
THEN:
|
|
- Profile is returned with social accounts
|
|
"""
|
|
social_account = self.setupSocialAccount()
|
|
|
|
openid_provider = MockOpenIDConnectProvider(
|
|
app=SocialApp.objects.get(provider_id="keycloak-test"),
|
|
)
|
|
mock_list_providers.return_value = [
|
|
openid_provider,
|
|
]
|
|
mock_get_provider_account.return_value = MockOpenIDConnectProviderAccount(
|
|
mock_social_account_dict={
|
|
"name": openid_provider.name,
|
|
},
|
|
)
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
self.assertEqual(
|
|
response.data["social_accounts"],
|
|
[
|
|
{
|
|
"id": social_account.pk,
|
|
"provider": "keycloak-test",
|
|
"name": "Keycloak",
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_profile_w_social_removed_app(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user and setup social account
|
|
- Social app has been removed
|
|
WHEN:
|
|
- API call is made to get profile
|
|
THEN:
|
|
- Profile is returned with "Unknown App" as name
|
|
"""
|
|
social_account = self.setupSocialAccount()
|
|
|
|
# Remove the social app
|
|
SocialApp.objects.get(provider_id="keycloak-test").delete()
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
self.assertEqual(
|
|
response.data["social_accounts"],
|
|
[
|
|
{
|
|
"id": social_account.pk,
|
|
"provider": "keycloak-test",
|
|
"name": "Unknown App",
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_update_profile(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user
|
|
WHEN:
|
|
- API call is made to update profile
|
|
THEN:
|
|
- Profile is updated
|
|
"""
|
|
|
|
user_data = {
|
|
"email": "new@email.com",
|
|
"password": "superpassword1234",
|
|
"first_name": "new first name",
|
|
"last_name": "new last name",
|
|
}
|
|
response = self.client.patch(self.ENDPOINT, user_data)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
user = User.objects.get(username=self.user.username)
|
|
self.assertTrue(user.check_password(user_data["password"]))
|
|
self.assertEqual(user.email, user_data["email"])
|
|
self.assertEqual(user.first_name, user_data["first_name"])
|
|
self.assertEqual(user.last_name, user_data["last_name"])
|
|
|
|
def test_update_profile_invalid_password_returns_field_error(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user
|
|
WHEN:
|
|
- API call is made to update profile with weak password
|
|
THEN:
|
|
- Profile update fails with password field error
|
|
"""
|
|
|
|
user_data = {
|
|
"email": "new@email.com",
|
|
"password": "short", # shorter than default validator threshold
|
|
"first_name": "new first name",
|
|
"last_name": "new last name",
|
|
}
|
|
|
|
response = self.client.patch(self.ENDPOINT, user_data)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn("password", response.data)
|
|
self.assertIsInstance(response.data["password"], list)
|
|
self.assertTrue(
|
|
any(
|
|
"too short" in message.lower() for message in response.data["password"]
|
|
),
|
|
)
|
|
|
|
def test_update_profile_placeholder_password_skips_validation(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user with existing password
|
|
WHEN:
|
|
- API call is made with the obfuscated placeholder password value
|
|
THEN:
|
|
- Profile is updated without changing the password or running validators
|
|
"""
|
|
|
|
original_password = "orig-pass-12345"
|
|
self.user.set_password(original_password)
|
|
self.user.save()
|
|
|
|
user_data = {
|
|
"email": "new@email.com",
|
|
"password": "*" * 12, # matches obfuscated value from serializer
|
|
"first_name": "new first name",
|
|
"last_name": "new last name",
|
|
}
|
|
|
|
response = self.client.patch(self.ENDPOINT, user_data)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
user = User.objects.get(username=self.user.username)
|
|
self.assertTrue(user.check_password(original_password))
|
|
self.assertEqual(user.email, user_data["email"])
|
|
self.assertEqual(user.first_name, user_data["first_name"])
|
|
self.assertEqual(user.last_name, user_data["last_name"])
|
|
|
|
def test_update_auth_token(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user
|
|
WHEN:
|
|
- API call is made to generate auth token
|
|
THEN:
|
|
- Token is created the first time, updated the second
|
|
"""
|
|
|
|
self.assertEqual(len(Token.objects.all()), 0)
|
|
|
|
response = self.client.post(f"{self.ENDPOINT}generate_auth_token/")
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
token1 = Token.objects.filter(user=self.user).first()
|
|
self.assertIsNotNone(token1)
|
|
|
|
response = self.client.post(f"{self.ENDPOINT}generate_auth_token/")
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
token2 = Token.objects.filter(user=self.user).first()
|
|
|
|
self.assertNotEqual(token1.key, token2.key)
|
|
|
|
def test_profile_not_logged_in(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- User not logged in
|
|
WHEN:
|
|
- API call is made to get profile and update token
|
|
THEN:
|
|
- Profile is returned
|
|
"""
|
|
|
|
self.client.logout()
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
response = self.client.post(f"{self.ENDPOINT}generate_auth_token/")
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
@mock.patch(
|
|
"allauth.socialaccount.adapter.DefaultSocialAccountAdapter.list_providers",
|
|
)
|
|
def test_get_social_account_providers(
|
|
self,
|
|
mock_list_providers,
|
|
) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user
|
|
WHEN:
|
|
- API call is made to get social account providers
|
|
THEN:
|
|
- Social account providers are returned
|
|
"""
|
|
self.setupSocialAccount()
|
|
|
|
mock_list_providers.return_value = [
|
|
MockOpenIDConnectProvider(
|
|
app=SocialApp.objects.get(provider_id="keycloak-test"),
|
|
),
|
|
]
|
|
|
|
response = self.client.get(f"{self.ENDPOINT}social_account_providers/")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(
|
|
response.data[0]["name"],
|
|
"Keycloak",
|
|
)
|
|
self.assertIn(
|
|
"keycloak-test/login/?process=connect",
|
|
response.data[0]["login_url"],
|
|
)
|
|
|
|
@mock.patch(
|
|
"allauth.socialaccount.adapter.DefaultSocialAccountAdapter.list_providers",
|
|
)
|
|
def test_get_social_account_providers_openid(
|
|
self,
|
|
mock_list_providers,
|
|
) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user and openid social account provider
|
|
WHEN:
|
|
- API call is made to get social account providers
|
|
THEN:
|
|
- Brands for openid provider are returned
|
|
"""
|
|
|
|
mock_list_providers.return_value = [
|
|
MockOpenIDProvider(),
|
|
]
|
|
|
|
response = self.client.get(f"{self.ENDPOINT}social_account_providers/")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(
|
|
len(response.data),
|
|
2,
|
|
)
|
|
|
|
def test_disconnect_social_account(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured user
|
|
WHEN:
|
|
- API call is made to disconnect a social account
|
|
THEN:
|
|
- Social account is deleted from the user or request fails
|
|
"""
|
|
self.setupSocialAccount()
|
|
|
|
# Test with invalid id
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}disconnect_social_account/",
|
|
{"id": -1},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Test with valid id
|
|
social_account_id = self.user.socialaccount_set.all()[0].pk
|
|
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}disconnect_social_account/",
|
|
{"id": social_account_id},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data, social_account_id)
|
|
|
|
self.assertEqual(
|
|
len(self.user.socialaccount_set.filter(pk=social_account_id)),
|
|
0,
|
|
)
|
|
|
|
|
|
class TestApiTOTPViews(APITestCase):
|
|
ENDPOINT = "/api/profile/totp/"
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
|
|
self.user = UserFactory(username="temp_admin", superuser=True)
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def test_get_totp(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Existing user account
|
|
WHEN:
|
|
- API request is made to TOTP endpoint
|
|
THEN:
|
|
- TOTP is generated
|
|
"""
|
|
response = self.client.get(
|
|
self.ENDPOINT,
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertIn("qr_svg", response.data)
|
|
self.assertIn("secret", response.data)
|
|
|
|
@mock.patch("allauth.mfa.totp.internal.auth.validate_totp_code")
|
|
def test_activate_totp(self, mock_validate_totp_code) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Existing user account
|
|
WHEN:
|
|
- API request is made to activate TOTP
|
|
THEN:
|
|
- TOTP is activated, recovery codes are returned
|
|
"""
|
|
mock_validate_totp_code.return_value = True
|
|
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
data={
|
|
"secret": "123",
|
|
"code": "456",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertTrue(Authenticator.objects.filter(user=self.user).exists())
|
|
self.assertIn("recovery_codes", response.data)
|
|
|
|
def test_deactivate_totp(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Existing user account with TOTP enabled
|
|
WHEN:
|
|
- API request is made to deactivate TOTP
|
|
THEN:
|
|
- TOTP is deactivated
|
|
"""
|
|
Authenticator.objects.create(
|
|
user=self.user,
|
|
type=Authenticator.Type.TOTP,
|
|
data={},
|
|
)
|
|
|
|
response = self.client.delete(
|
|
self.ENDPOINT,
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(Authenticator.objects.filter(user=self.user).count(), 0)
|
|
|
|
# test fails
|
|
response = self.client.delete(
|
|
self.ENDPOINT,
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|