"""Unit tests for the 'users' app."""

from django.apps import apps

from django.contrib.auth.models import Group

from django.test import TestCase

from django.urls import reverse

from common.settings import set_global_setting

from InvenTree.helpers_mfa import get_codes

from InvenTree.unit_test import AdminTestCase, InvenTreeAPITestCase, InvenTreeTestCase

from users.models import ApiToken, Owner

from users.oauth2_scopes import _roles

from users.ruleset import (

RULESET_CHOICES,

RULESET_NAMES,

get_ruleset_ignore,

get_ruleset_models,

)

G_RULESETS = get_ruleset_models()

G_RULESETS_IG = get_ruleset_ignore()

G_SCOPES = _roles.keys()

class RuleSetModelTest(TestCase):

"""Some simplistic tests to ensure the RuleSet model is setup correctly."""

def test_ruleset_models(self):

"""Test that the role rulesets work as intended."""

keys = G_RULESETS.keys()

# Check if there are any rulesets which do not have models defined

missing = [name for name in RULESET_NAMES if name not in keys]

if len(missing) > 0: # pragma: no cover

print('INVE-E5: The following rulesets do not have models assigned:')

for m in missing:

print('-', m)

# Check if models have been defined for a ruleset which is incorrect

extra = [name for name in keys if name not in RULESET_NAMES]

if len(extra) > 0: # pragma: no cover

print(

'INVE-E5: The following rulesets have been improperly added to get_ruleset_models():'

)

for e in extra:

print('-', e)

# Check that each ruleset has models assigned

empty = [key for key in keys if len(G_RULESETS[key]) == 0]

if len(empty) > 0: # pragma: no cover

print(

'INVE-E5: The following rulesets have empty entries in get_ruleset_models():'

)

for e in empty:

print('-', e)

self.assertEqual(len(missing), 0, 'See INVE-E5 in the docs')

self.assertEqual(len(extra), 0, 'See INVE-E5 in the docs')

self.assertEqual(len(empty), 0, 'See INVE-E5 in the docs')

def test_model_names(self):

"""Test that each model defined in the rulesets is valid, based on the database schema!"""

available_models = apps.get_models()

available_tables = set()

# Extract each available database model and construct a formatted string

for model in available_models:

label = model.objects.model._meta.label

label = label.replace('.', '_').lower()

available_tables.add(label)

assigned_models = set()

# Now check that each defined model is a valid table name

for key in G_RULESETS:

models = G_RULESETS[key]

for m in models:

assigned_models.add(m)

missing_models = set()

for model in available_tables:

if (

model not in assigned_models and model not in G_RULESETS_IG

): # pragma: no cover

missing_models.add(model)

if len(missing_models) > 0: # pragma: no cover

print(

'INVE-E5: The following database models are not covered by the defined RuleSet permissions:'

)

for m in missing_models:

print('-', m)

extra_models = set()

defined_models = set()

for model in assigned_models:

defined_models.add(model)

for model in G_RULESETS_IG:

defined_models.add(model)

for model in defined_models: # pragma: no cover

if model not in available_tables:

extra_models.add(model)

if len(extra_models) > 0: # pragma: no cover

print(

'INVE-E5: The following RuleSet permissions do not match a database model:'

)

for m in extra_models:

print('-', m)

self.assertEqual(len(missing_models), 0, 'See INVE-E5 in the docs')

self.assertEqual(len(extra_models), 0, 'See INVE-E5 in the docs')

def test_scope_names(self):

"""Ensure that the rulesets map to scopes and vice versa."""

# Check that each scope has a corresponding ruleset

missing = [scope for scope in G_SCOPES if scope not in G_RULESETS]

if len(missing) > 0: # pragma: no cover

print('INVE-E6: The following scopes do not have corresponding rulesets:')

for m in missing:

print('-', m)

# Check that each ruleset has a corresponding scope

extra = [scope for scope in G_RULESETS if scope not in G_SCOPES]

if len(extra) > 0: # pragma: no cover

print('INVE-E6: The following rulesets do not have corresponding scopes:')

for m in extra:

print('-', m)

self.assertEqual(len(missing), 0, 'See INVE-E6 in the docs')

self.assertEqual(len(extra), 0, 'See INVE-E6 in the docs')

def test_permission_assign(self):

"""Test that the permission assigning works!"""

# Create a new group

group = Group.objects.create(name='Test group')

rulesets = group.rule_sets.all()

# Rulesets should have been created automatically for this group

self.assertEqual(rulesets.count(), len(RULESET_CHOICES))

# Check that all permissions have been assigned permissions?

permission_set = set()

for models in G_RULESETS.values():

for model in models:

permission_set.add(model)

# By default no permissions should be assigned

self.assertEqual(group.permissions.count(), 0)

# Add some more rules

for rule in rulesets:

rule.can_view = True

rule.can_add = True

rule.can_change = True

rule.save()

# update_fields is required to trigger permissions update

group.save(update_fields=['name'])

# There should now be three permissions for each rule set

self.assertEqual(group.permissions.count(), 3 * len(permission_set))

# Now remove *all* permissions

for rule in rulesets:

rule.can_view = False

rule.can_add = False

rule.can_change = False

rule.can_delete = False

rule.save()

# update_fields is required to trigger permissions update

group.save(update_fields=['name'])

# There should now not be any permissions assigned to this group

self.assertEqual(group.permissions.count(), 0)

class OwnerModelTest(InvenTreeTestCase):

"""Some simplistic tests to ensure the Owner model is setup correctly."""

def do_request(self, endpoint, filters, status_code=200):

"""Perform an API request."""

response = self.client.get(endpoint, filters, format='json')

self.assertEqual(response.status_code, status_code)

return response.data

def test_owner(self):

"""Tests for the 'owner' model."""

# Check that owner was created for user

user_as_owner = Owner.get_owner(self.user)

self.assertEqual(type(user_as_owner), Owner)

# Check that owner was created for group

group_as_owner = Owner.get_owner(self.group)

self.assertEqual(type(group_as_owner), Owner)

# Check name

self.assertEqual(str(user_as_owner), 'testuser (user)')

# Get related owners (user + group)

related_owners = group_as_owner.get_related_owners(include_group=True)

self.assertIn(user_as_owner, related_owners)

self.assertIn(group_as_owner, related_owners)

# Get related owners (only user)

related_owners = group_as_owner.get_related_owners(include_group=False)

self.assertIn(user_as_owner, related_owners)

self.assertNotIn(group_as_owner, related_owners)

# Get related owners on user

related_owners = user_as_owner.get_related_owners()

self.assertEqual(related_owners, [user_as_owner])

# Check owner matching

owners = Owner.get_owners_matching_user(self.user)

self.assertEqual(owners, [user_as_owner, group_as_owner])

# Delete user and verify owner was deleted too

self.user.delete()

user_as_owner = Owner.get_owner(self.user)

self.assertEqual(user_as_owner, None)

# Delete group and verify owner was deleted too

self.group.delete()

group_as_owner = Owner.get_owner(self.group)

self.assertEqual(group_as_owner, None)

def test_api(self):

"""Test user APIs."""

self.client.logout()

# not authed

self.do_request(reverse('api-owner-list'), {}, 401)

self.do_request(

reverse('api-owner-detail', kwargs={'pk': self.user.id}), {}, 401

)

self.client.login(username=self.username, password=self.password)

# user list

self.do_request(reverse('api-owner-list'), {})

# user list with 'is_active' filter

self.do_request(reverse('api-owner-list'), {'is_active': False})

# user list with search

self.do_request(reverse('api-owner-list'), {'search': 'user'})

# # owner detail - user

# response = self.do_request(reverse('api-owner-detail', kwargs={'pk': 1}), {})

# self.assertEqual(response['name'], self.username)

# self.assertEqual(response['label'], 'user')

# self.assertEqual(response['owner_id'], self.user.id)

# # owner detail - group

# group = self.user.groups.first()

# response = self.do_request(reverse('api-owner-detail', kwargs={'pk': 2}), {})

# self.assertEqual(response['name'], group.name)

# self.assertEqual(response['label'], 'group')

# self.assertEqual(response['owner_id'], group.pk)

# own user detail

response_detail = self.do_request(

reverse('api-user-detail', kwargs={'pk': self.user.id}), {}, 200

)

self.assertEqual(response_detail['username'], self.username)

response_me = self.do_request(reverse('api-user-me'), {}, 200)

self.assertIn('language', response_me['profile'])

self.assertIn('theme', response_me['profile'])

def test_token(self):

"""Test token mechanisms."""

self.client.logout()

token = ApiToken.objects.filter(user=self.user)

# not authed

self.do_request(reverse('api-token'), {}, 401)

self.client.login(username=self.username, password=self.password)

# token get

response = self.do_request(reverse('api-token'), {})

self.assertEqual(response['token'], token.first().key)

# test user is associated with token

response = self.do_request(

reverse('api-user-me'), {'name': 'another-token'}, 200

)

self.assertEqual(response['username'], self.username)

def test_display_name(self):

"""Test the display name for the owner."""

owner = Owner.get_owner(self.user)

self.assertEqual(owner.name(), 'testuser')

self.assertEqual(str(owner), 'testuser (user)')

# Change setting

set_global_setting('DISPLAY_FULL_NAMES', True)

self.user.first_name = 'first'

self.user.last_name = 'last'

self.user.save()

owner = Owner.get_owner(self.user)

# Now first / last should be used

self.assertEqual(owner.name(), 'first last')

self.assertEqual(str(owner), 'first last (user)')

# Reset

set_global_setting('DISPLAY_FULL_NAMES', False)

self.user.first_name = ''

self.user.last_name = ''

self.user.save()

class MFALoginTest(InvenTreeAPITestCase):

"""Some simplistic tests to ensure that MFA is working."""

def test_api(self):

"""Test that the API is working."""

auth_data = {'username': self.username, 'password': self.password}

login_url = reverse('browser:account:login')

# Double login is not allowed

self.post(login_url, auth_data, expected_code=409)

# Normal login - no mfa

self.client.logout()

response = self.post(login_url, auth_data, expected_code=200)

self._helper_meta_val(response)

# Add MFA - trying in a limited loop in case of timing issues

rc_code = get_codes(user=self.user)[1][0]

# There must be a TOTP device now - success

self.get(reverse('browser:mfa:manage_totp'), expected_code=200)

self.get(reverse('api-token'), expected_code=200)

# Login with MFA enabled but not provided

self.client.logout()

response = self.post(login_url, auth_data, expected_code=401)

self._helper_meta_val(response, val=False)

self.assertEqual(self._helper_get_flow(response)['is_pending'], True)

self.get(reverse('api-token'), expected_code=401)

# Login with MFA enabled and provided - second api call an success

self.client.logout()

response = self.post(login_url, auth_data, expected_code=401)

# MFA not finished - no access allowed

self.get(reverse('api-token'), expected_code=401)

# Complete MFA (with recovery code to avoid timing issues)

self.post(

reverse('browser:mfa:authenticate'), {'code': rc_code}, expected_code=401

)

self.post(reverse('browser:mfa:trust'), {'trust': False}, expected_code=200)

# and run through trust

self.get(reverse('api-token'), expected_code=200)

# Login with MFA enabled and provided - but incorrect pwd

self.client.logout()

auth_data['password'] = 'wrong'

response = self.post(login_url, auth_data, expected_code=400)

self.assertContains(

response,

'The username and/or password you specified are not correct',

status_code=400,

)

auth_data['password'] = self.password

def _helper_meta_val(

self, response, key: str = 'is_authenticated', val: bool = True

):

"""Helper to run a test on meta response."""

self.assertEqual(response.json()['meta'][key], val)

def _helper_get_flow(self, response, flow_id: str = 'mfa_authenticate'):

"""Helper to run a test on flow response."""

flows = response.json()['data']['flows']

return next(a for a in flows if a['id'] == flow_id)

class AdminTest(AdminTestCase):

"""Tests for the admin interface integration."""

def test_admin(self):

"""Test the admin URL."""

my_token = self.helper(

model=ApiToken, model_kwargs={'user': self.user, 'name': 'test-token'}

)

# Additionally test str fnc

self.assertEqual(str(my_token), my_token.token)

class UserProfileTest(InvenTreeAPITestCase):

"""Tests for the user profile API endpoints."""

def test_profile_retrieve(self):

"""Test retrieving the user profile."""

response = self.client.get(reverse('api-user-profile'))

self.assertEqual(response.status_code, 200)

self.assertIn('language', response.data)

self.assertIn('theme', response.data)

self.assertIn('widgets', response.data)

self.assertIn('displayname', response.data)

self.assertIn('position', response.data)

self.assertIn('status', response.data)

self.assertIn('location', response.data)

self.assertIn('active', response.data)

self.assertIn('contact', response.data)

self.assertIn('type', response.data)

self.assertIn('organisation', response.data)

self.assertIn('primary_group', response.data)

def test_profile_update(self):

"""Test updating the user profile."""

data = {

'language': 'en',

'theme': {'color': 'blue'},

'widgets': {'widget1': 'value1'},

'displayname': 'Test User',

'status': 'Active',

'location': 'Test Location',

'active': True,

'contact': 'test@example.com',

'type': 'internal',

'organisation': 'Test Organisation',

'primary_group': self.group.pk,

}

response = self.patch(reverse('api-user-profile'), data)

self.assertEqual(response.status_code, 200)

self.assertEqual(response.data['language'], data['language'])

self.assertEqual(response.data['theme'], data['theme'])

self.assertEqual(response.data['widgets'], data['widgets'])

self.assertEqual(response.data['displayname'], data['displayname'])

self.assertEqual(response.data['status'], data['status'])

self.assertEqual(response.data['location'], data['location'])

self.assertEqual(response.data['active'], data['active'])

self.assertEqual(response.data['contact'], data['contact'])

self.assertEqual(response.data['type'], data['type'])

self.assertEqual(response.data['organisation'], data['organisation'])

self.assertEqual(response.data['primary_group'], data['primary_group'])

def test_primary_group_validation(self):

"""Test that primary_group is a group that the user is a member of."""

new_group = Group.objects.create(name='New Group')

profile = self.user.profile

profile.primary_group = new_group

profile.save()

self.assertIsNone(profile.primary_group)

def test_validate_primary_group_on_save(self):

"""Test validate_primary_group_on_save signal handler."""

group = Group.objects.create(name='Test Group')

self.user.groups.add(group)

profile = self.user.profile

profile.primary_group = group

profile.save()

# Ensure primary_group is set correctly

self.assertEqual(profile.primary_group, group)

# Remove user from group and save group

self.user.groups.remove(group)

# Ensure primary_group is set to None

profile.refresh_from_db()

self.assertIsNone(profile.primary_group)

def test_validate_primary_group_on_delete(self):

"""Test validate_primary_group_on_delete signal handler."""

group = Group.objects.create(name='Test Group')

self.user.groups.add(group)

profile = self.user.profile

profile.primary_group = group

profile.save()

# Ensure primary_group is set correctly

self.assertEqual(profile.primary_group, group)

# Delete group

group.delete()

# Ensure primary_group is set to None

profile.refresh_from_db()

self.assertIsNone(profile.primary_group)