InvenTree/InvenTree/common/tests.py

"""Tests for mechanisms in common."""

import io
import json
import time
from datetime import timedelta
from http import HTTPStatus
from unittest import mock

from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import Client, TestCase
from django.urls import reverse

import PIL

from InvenTree.helpers import str2bool
from InvenTree.unit_test import (InvenTreeAPITestCase, InvenTreeTestCase,
                                 PluginMixin)
from plugin import registry
from plugin.models import NotificationUserSetting

from .api import WebhookView
from .models import (ColorTheme, CustomUnit, InvenTreeSetting,
                     InvenTreeUserSetting, NotesImage, NotificationEntry,
                     NotificationMessage, ProjectCode, WebhookEndpoint,
                     WebhookMessage)

CONTENT_TYPE_JSON = 'application/json'


class SettingsTest(InvenTreeTestCase):
    """Tests for the 'settings' model."""

    fixtures = [
        'settings',
    ]

    def test_settings_objects(self):
        """Test fixture loading and lookup for settings."""
        # There should be two settings objects in the database
        settings = InvenTreeSetting.objects.all()

        self.assertTrue(settings.count() >= 2)

        instance_name = InvenTreeSetting.objects.get(pk=1)
        self.assertEqual(instance_name.key, 'INVENTREE_INSTANCE')
        self.assertEqual(instance_name.value, 'My very first InvenTree Instance')

        # Check object lookup (case insensitive)
        self.assertEqual(InvenTreeSetting.get_setting_object('iNvEnTrEE_inSTanCE').pk, 1)

    def test_settings_functions(self):
        """Test settings functions and properties."""
        # define settings to check
        instance_ref = 'INVENTREE_INSTANCE'
        instance_obj = InvenTreeSetting.get_setting_object(instance_ref, cache=False)

        stale_ref = 'STOCK_STALE_DAYS'
        stale_days = InvenTreeSetting.get_setting_object(stale_ref, cache=False)

        report_size_obj = InvenTreeSetting.get_setting_object('REPORT_DEFAULT_PAGE_SIZE')
        report_test_obj = InvenTreeSetting.get_setting_object('REPORT_ENABLE_TEST_REPORT')

        # check settings base fields
        self.assertEqual(instance_obj.name, 'Server Instance Name')
        self.assertEqual(instance_obj.get_setting_name(instance_ref), 'Server Instance Name')
        self.assertEqual(instance_obj.description, 'String descriptor for the server instance')
        self.assertEqual(instance_obj.get_setting_description(instance_ref), 'String descriptor for the server instance')

        # check units
        self.assertEqual(instance_obj.units, '')
        self.assertEqual(instance_obj.get_setting_units(instance_ref), '')
        self.assertEqual(instance_obj.get_setting_units(stale_ref), 'days')

        # check as_choice
        self.assertEqual(instance_obj.as_choice(), 'My very first InvenTree Instance')
        self.assertEqual(report_size_obj.as_choice(), 'A4')

        # check is_choice
        self.assertEqual(instance_obj.is_choice(), False)
        self.assertEqual(report_size_obj.is_choice(), True)

        # check setting_type
        self.assertEqual(instance_obj.setting_type(), 'string')
        self.assertEqual(report_test_obj.setting_type(), 'boolean')
        self.assertEqual(stale_days.setting_type(), 'integer')

        # check as_int
        self.assertEqual(stale_days.as_int(), 0)
        self.assertEqual(instance_obj.as_int(), 'InvenTree')  # not an int -> return default

        # check as_bool
        self.assertEqual(report_test_obj.as_bool(), True)

        # check to_native_value
        self.assertEqual(stale_days.to_native_value(), 0)

    def test_allValues(self):
        """Make sure that the allValues functions returns correctly."""
        # define testing settings

        # check a few keys
        result = InvenTreeSetting.allValues()
        self.assertIn('INVENTREE_INSTANCE', result)
        self.assertIn('PART_COPY_TESTS', result)
        self.assertIn('STOCK_OWNERSHIP_CONTROL', result)
        self.assertIn('SIGNUP_GROUP', result)
        self.assertIn('SERVER_RESTART_REQUIRED', result)

        result = InvenTreeSetting.allValues(exclude_hidden=True)
        self.assertNotIn('SERVER_RESTART_REQUIRED', result)

    def test_all_settings(self):
        """Make sure that the all_settings function returns correctly"""
        result = InvenTreeSetting.all_settings()
        self.assertIn("INVENTREE_INSTANCE", result)
        self.assertIsInstance(result['INVENTREE_INSTANCE'], InvenTreeSetting)

    @mock.patch("common.models.InvenTreeSetting.get_setting_definition")
    def test_check_all_settings(self, get_setting_definition):
        """Make sure that the check_all_settings function returns correctly"""
        # define partial schema
        settings_definition = {
            "AB": {  # key that's has not already been accessed
                "required": True,
            },
            "CD": {
                "required": True,
                "protected": True,
            },
            "EF": {}
        }

        def mocked(key, **kwargs):
            return settings_definition.get(key, {})
        get_setting_definition.side_effect = mocked

        self.assertEqual(InvenTreeSetting.check_all_settings(settings_definition=settings_definition), (False, ["AB", "CD"]))
        InvenTreeSetting.set_setting('AB', "hello", self.user)
        InvenTreeSetting.set_setting('CD', "world", self.user)
        self.assertEqual(InvenTreeSetting.check_all_settings(), (True, []))

    @mock.patch("common.models.InvenTreeSetting.get_setting_definition")
    def test_settings_validator(self, get_setting_definition):
        """Make sure that the validator function gets called on set setting."""

        def validator(x):
            if x == "hello":
                return x

            raise ValidationError(f"{x} is not valid")

        mock_validator = mock.Mock(side_effect=validator)

        # define partial schema
        settings_definition = {
            "AB": {  # key that's has not already been accessed
                "validator": mock_validator,
            },
        }

        def mocked(key, **kwargs):
            return settings_definition.get(key, {})
        get_setting_definition.side_effect = mocked

        InvenTreeSetting.set_setting("AB", "hello", self.user)
        mock_validator.assert_called_with("hello")

        with self.assertRaises(ValidationError):
            InvenTreeSetting.set_setting("AB", "world", self.user)
        mock_validator.assert_called_with("world")

    def run_settings_check(self, key, setting):
        """Test that all settings are valid.

        - Ensure that a name is set and that it is translated
        - Ensure that a description is set
        - Ensure that every setting key is valid
        - Ensure that a validator is supplied
        """
        self.assertTrue(type(setting) is dict)

        name = setting.get('name', None)

        self.assertIsNotNone(name)
        self.assertIn('django.utils.functional.lazy', str(type(name)))

        description = setting.get('description', None)

        self.assertIsNotNone(description)
        self.assertIn('django.utils.functional.lazy', str(type(description)))

        if key != key.upper():
            raise ValueError(f"Setting key '{key}' is not uppercase")  # pragma: no cover

        # Check that only allowed keys are provided
        allowed_keys = [
            'name',
            'description',
            'default',
            'validator',
            'hidden',
            'choices',
            'units',
            'requires_restart',
            'after_save',
            'before_save',
        ]

        for k in setting.keys():
            self.assertIn(k, allowed_keys)

        # Check default value for boolean settings
        validator = setting.get('validator', None)

        if validator is bool:
            default = setting.get('default', None)

            # Default value *must* be supplied for boolean setting!
            self.assertIsNotNone(default)

            # Default value for boolean must itself be a boolean
            self.assertIn(default, [True, False])

    def test_setting_data(self):
        """Test for settings data.

        - Ensure that every setting has a name, which is translated
        - Ensure that every setting has a description, which is translated
        """
        for key, setting in InvenTreeSetting.SETTINGS.items():

            try:
                self.run_settings_check(key, setting)
            except Exception as exc:  # pragma: no cover
                print(f"run_settings_check failed for global setting '{key}'")
                raise exc

        for key, setting in InvenTreeUserSetting.SETTINGS.items():
            try:
                self.run_settings_check(key, setting)
            except Exception as exc:  # pragma: no cover
                print(f"run_settings_check failed for user setting '{key}'")
                raise exc

    def test_defaults(self):
        """Populate the settings with default values."""
        for key in InvenTreeSetting.SETTINGS.keys():

            value = InvenTreeSetting.get_setting_default(key)

            InvenTreeSetting.set_setting(key, value, self.user)

            self.assertEqual(value, InvenTreeSetting.get_setting(key))

            # Any fields marked as 'boolean' must have a default value specified
            setting = InvenTreeSetting.get_setting_object(key)

            if setting.is_bool():
                if setting.default_value in ['', None]:
                    raise ValueError(f'Default value for boolean setting {key} not provided')  # pragma: no cover

                if setting.default_value not in [True, False]:
                    raise ValueError(f'Non-boolean default value specified for {key}')  # pragma: no cover

    def test_global_setting_caching(self):
        """Test caching operations for the global settings class"""

        key = 'PART_NAME_FORMAT'

        cache_key = InvenTreeSetting.create_cache_key(key)
        self.assertEqual(cache_key, 'InvenTreeSetting:PART_NAME_FORMAT')

        cache.clear()

        self.assertIsNone(cache.get(cache_key))

        # First request should set cache
        val = InvenTreeSetting.get_setting(key)
        self.assertEqual(cache.get(cache_key).value, val)

        for val in ['A', '{{ part.IPN }}', 'C']:
            # Check that the cached value is updated whenever the setting is saved
            InvenTreeSetting.set_setting(key, val, None)
            self.assertEqual(cache.get(cache_key).value, val)
            self.assertEqual(InvenTreeSetting.get_setting(key), val)

    def test_user_setting_caching(self):
        """Test caching operation for the user settings class"""

        cache.clear()

        # Generate a number of new users
        for idx in range(5):
            get_user_model().objects.create(
                username=f"User_{idx}",
                password="hunter42",
                email="email@dot.com",
            )

        key = 'SEARCH_PREVIEW_RESULTS'

        # Check that the settings are correctly cached for each separate user
        for user in get_user_model().objects.all():
            setting = InvenTreeUserSetting.get_setting_object(key, user=user)
            cache_key = setting.cache_key
            self.assertEqual(cache_key, f"InvenTreeUserSetting:SEARCH_PREVIEW_RESULTS_user:{user.username}")
            InvenTreeUserSetting.set_setting(key, user.pk, None, user=user)
            self.assertIsNotNone(cache.get(cache_key))

        # Iterate through a second time, ensure the values have been cached correctly
        for user in get_user_model().objects.all():
            value = InvenTreeUserSetting.get_setting(key, user=user)
            self.assertEqual(value, user.pk)


class GlobalSettingsApiTest(InvenTreeAPITestCase):
    """Tests for the global settings API."""

    def setUp(self):
        """Ensure cache is cleared as part of test setup"""
        cache.clear()
        return super().setUp()

    def test_global_settings_api_list(self):
        """Test list URL for global settings."""
        url = reverse('api-global-setting-list')

        # Read out each of the global settings value, to ensure they are instantiated in the database
        for key in InvenTreeSetting.SETTINGS:
            InvenTreeSetting.get_setting_object(key, cache=False)

        response = self.get(url, expected_code=200)

        # Number of results should match the number of settings
        self.assertEqual(len(response.data), len(InvenTreeSetting.SETTINGS.keys()))

    def test_company_name(self):
        """Test a settings object lifecycle e2e."""
        setting = InvenTreeSetting.get_setting_object('INVENTREE_COMPANY_NAME')

        # Check default value
        self.assertEqual(setting.value, 'My company name')

        url = reverse('api-global-setting-detail', kwargs={'key': setting.key})

        # Test getting via the API
        for val in ['test', '123', 'My company nam3']:
            setting.value = val
            setting.save()

            response = self.get(url, expected_code=200)

            self.assertEqual(response.data['value'], val)

        # Test setting via the API
        for val in ['cat', 'hat', 'bat', 'mat']:
            response = self.patch(
                url,
                {
                    'value': val,
                },
                expected_code=200
            )

            self.assertEqual(response.data['value'], val)

            setting.refresh_from_db()
            self.assertEqual(setting.value, val)

    def test_api_detail(self):
        """Test that we can access the detail view for a setting based on the <key>."""
        # These keys are invalid, and should return 404
        for key in ["apple", "carrot", "dog"]:
            response = self.get(
                reverse('api-global-setting-detail', kwargs={'key': key}),
                expected_code=404,
            )

        key = 'INVENTREE_INSTANCE'
        url = reverse('api-global-setting-detail', kwargs={'key': key})

        InvenTreeSetting.objects.filter(key=key).delete()

        # Check that we can access a setting which has not previously been created
        self.assertFalse(InvenTreeSetting.objects.filter(key=key).exists())

        # Access via the API, and the default value should be received
        response = self.get(url, expected_code=200)

        self.assertEqual(response.data['value'], 'InvenTree')

        # Now, the object should have been created in the DB
        self.patch(
            url,
            {
                'value': 'My new title',
            },
            expected_code=200,
        )

        setting = InvenTreeSetting.objects.get(key=key)

        self.assertEqual(setting.value, 'My new title')

        # And retrieving via the API now returns the updated value
        response = self.get(url, expected_code=200)

        self.assertEqual(response.data['value'], 'My new title')


class UserSettingsApiTest(InvenTreeAPITestCase):
    """Tests for the user settings API."""

    def test_user_settings_api_list(self):
        """Test list URL for user settings."""
        url = reverse('api-user-setting-list')

        self.get(url, expected_code=200)

    def test_user_setting_invalid(self):
        """Test a user setting with an invalid key."""
        url = reverse('api-user-setting-detail', kwargs={'key': 'DONKEY'})

        self.get(url, expected_code=404)

    def test_user_setting_init(self):
        """Test we can retrieve a setting which has not yet been initialized."""
        key = 'HOMEPAGE_PART_LATEST'

        # Ensure it does not actually exist in the database
        self.assertFalse(InvenTreeUserSetting.objects.filter(key=key).exists())

        url = reverse('api-user-setting-detail', kwargs={'key': key})

        response = self.get(url, expected_code=200)

        self.assertEqual(response.data['value'], 'True')

        self.patch(url, {'value': 'False'}, expected_code=200)

        setting = InvenTreeUserSetting.objects.get(key=key, user=self.user)

        self.assertEqual(setting.value, 'False')
        self.assertEqual(setting.to_native_value(), False)

    def test_user_setting_boolean(self):
        """Test a boolean user setting value."""
        # Ensure we have a boolean setting available
        setting = InvenTreeUserSetting.get_setting_object(
            'SEARCH_PREVIEW_SHOW_PARTS',
            user=self.user
        )

        # Check default values
        self.assertEqual(setting.to_native_value(), True)

        # Fetch via API
        url = reverse('api-user-setting-detail', kwargs={'key': setting.key})

        response = self.get(url, expected_code=200)

        self.assertEqual(response.data['pk'], setting.pk)
        self.assertEqual(response.data['key'], 'SEARCH_PREVIEW_SHOW_PARTS')
        self.assertEqual(response.data['description'], 'Display parts in search preview window')
        self.assertEqual(response.data['type'], 'boolean')
        self.assertEqual(len(response.data['choices']), 0)
        self.assertTrue(str2bool(response.data['value']))

        # Assign some truthy values
        for v in ['true', True, 1, 'y', 'TRUE']:
            self.patch(
                url,
                {
                    'value': str(v),
                },
                expected_code=200,
            )

            response = self.get(url, expected_code=200)

            self.assertTrue(str2bool(response.data['value']))

        # Assign some false(ish) values
        for v in ['false', False, '0', 'n', 'FalSe']:
            self.patch(
                url,
                {
                    'value': str(v),
                },
                expected_code=200,
            )

            response = self.get(url, expected_code=200)

            self.assertFalse(str2bool(response.data['value']))

        # Assign some invalid values
        for v in ['x', '', 'invalid', None, '-1', 'abcde']:
            response = self.patch(
                url,
                {
                    'value': str(v),
                },
                expected_code=200
            )

            # Invalid values evaluate to False
            self.assertFalse(str2bool(response.data['value']))

    def test_user_setting_choice(self):
        """Test a user setting with choices."""
        setting = InvenTreeUserSetting.get_setting_object(
            'DATE_DISPLAY_FORMAT',
            user=self.user
        )

        url = reverse('api-user-setting-detail', kwargs={'key': setting.key})

        # Check default value
        self.assertEqual(setting.value, 'YYYY-MM-DD')

        # Check that a valid option can be assigned via the API
        for opt in ['YYYY-MM-DD', 'DD-MM-YYYY', 'MM/DD/YYYY']:

            self.patch(
                url,
                {
                    'value': opt,
                },
                expected_code=200,
            )

            setting.refresh_from_db()
            self.assertEqual(setting.value, opt)

        # Send an invalid option
        for opt in ['cat', 'dog', 12345]:

            response = self.patch(
                url,
                {
                    'value': opt,
                },
                expected_code=400,
            )

            self.assertIn('Chosen value is not a valid option', str(response.data))

    def test_user_setting_integer(self):
        """Test a integer user setting value."""
        setting = InvenTreeUserSetting.get_setting_object(
            'SEARCH_PREVIEW_RESULTS',
            user=self.user,
            cache=False,
        )

        url = reverse('api-user-setting-detail', kwargs={'key': setting.key})

        # Check default value for this setting
        self.assertEqual(setting.value, 10)

        for v in [1, 9, 99]:
            setting.value = v
            setting.save()

            response = self.get(url)

            self.assertEqual(response.data['value'], str(v))

        # Set valid options via the api
        for v in [5, 15, 25]:
            self.patch(
                url,
                {
                    'value': v,
                },
                expected_code=200,
            )

            setting.refresh_from_db()
            self.assertEqual(setting.to_native_value(), v)

        # Set invalid options via the API
        # Note that this particular setting has a MinValueValidator(1) associated with it
        for v in [0, -1, -5]:

            response = self.patch(
                url,
                {
                    'value': v,
                },
                expected_code=400,
            )


class NotificationUserSettingsApiTest(InvenTreeAPITestCase):
    """Tests for the notification user settings API."""

    def test_api_list(self):
        """Test list URL."""
        url = reverse('api-notification-setting-list')

        self.get(url, expected_code=200)

    def test_setting(self):
        """Test the string name for NotificationUserSetting."""

        NotificationUserSetting.set_setting('NOTIFICATION_METHOD_MAIL', True, change_user=self.user, user=self.user)
        test_setting = NotificationUserSetting.get_setting_object('NOTIFICATION_METHOD_MAIL', user=self.user)
        self.assertEqual(str(test_setting), 'NOTIFICATION_METHOD_MAIL (for testuser): True')


class PluginSettingsApiTest(PluginMixin, InvenTreeAPITestCase):
    """Tests for the plugin settings API."""

    def test_plugin_list(self):
        """List installed plugins via API."""
        url = reverse('api-plugin-list')

        # Simple request
        self.get(url, expected_code=200)

        # Request with filter
        self.get(url, expected_code=200, data={'mixin': 'settings'})

    def test_api_list(self):
        """Test list URL."""
        url = reverse('api-plugin-setting-list')

        self.get(url, expected_code=200)

    def test_valid_plugin_slug(self):
        """Test that an valid plugin slug runs through."""
        # Activate plugin
        registry.set_plugin_state('sample', True)

        # get data
        url = reverse('api-plugin-setting-detail', kwargs={'plugin': 'sample', 'key': 'API_KEY'})
        response = self.get(url, expected_code=200)

        # check the right setting came through
        self.assertTrue(response.data['key'], 'API_KEY')
        self.assertTrue(response.data['plugin'], 'sample')
        self.assertTrue(response.data['type'], 'string')
        self.assertTrue(response.data['description'], 'Key required for accessing external API')

        # Failure mode tests

        # Non-existent plugin
        url = reverse('api-plugin-setting-detail', kwargs={'plugin': 'doesnotexist', 'key': 'doesnotmatter'})
        response = self.get(url, expected_code=404)
        self.assertIn("Plugin 'doesnotexist' not installed", str(response.data))

        # Wrong key
        url = reverse('api-plugin-setting-detail', kwargs={'plugin': 'sample', 'key': 'doesnotexsist'})
        response = self.get(url, expected_code=404)
        self.assertIn("Plugin 'sample' has no setting matching 'doesnotexsist'", str(response.data))

    def test_invalid_setting_key(self):
        """Test that an invalid setting key returns a 404."""
        ...

    def test_uninitialized_setting(self):
        """Test that requesting an uninitialized setting creates the setting."""
        ...


class WebhookMessageTests(TestCase):
    """Tests for webhooks."""

    def setUp(self):
        """Setup for all tests."""
        self.endpoint_def = WebhookEndpoint.objects.create()
        self.url = f'/api/webhook/{self.endpoint_def.endpoint_id}/'
        self.client = Client(enforce_csrf_checks=True)

    def test_bad_method(self):
        """Test that a wrong HTTP method does not work."""
        response = self.client.get(self.url)
        assert response.status_code == HTTPStatus.METHOD_NOT_ALLOWED

    def test_missing_token(self):
        """Tests that token checks work."""
        response = self.client.post(
            self.url,
            content_type=CONTENT_TYPE_JSON,
        )

        assert response.status_code == HTTPStatus.FORBIDDEN
        assert (
            json.loads(response.content)['detail'] == WebhookView.model_class.MESSAGE_TOKEN_ERROR
        )

    def test_bad_token(self):
        """Test that a wrong token is not working."""
        response = self.client.post(
            self.url,
            content_type=CONTENT_TYPE_JSON,
            **{'HTTP_TOKEN': '1234567fghj'},
        )

        assert response.status_code == HTTPStatus.FORBIDDEN
        assert (json.loads(response.content)['detail'] == WebhookView.model_class.MESSAGE_TOKEN_ERROR)

    def test_bad_url(self):
        """Test that a wrongly formed url is not working."""
        response = self.client.post(
            '/api/webhook/1234/',
            content_type=CONTENT_TYPE_JSON,
        )

        assert response.status_code == HTTPStatus.NOT_FOUND

    def test_bad_json(self):
        """Test that malformed JSON is not accepted."""
        response = self.client.post(
            self.url,
            data="{'this': 123}",
            content_type=CONTENT_TYPE_JSON,
            **{'HTTP_TOKEN': str(self.endpoint_def.token)},
        )

        assert response.status_code == HTTPStatus.NOT_ACCEPTABLE
        assert (
            json.loads(response.content)['detail'] == 'Expecting property name enclosed in double quotes'
        )

    def test_success_no_token_check(self):
        """Test that a endpoint without a token set does not require one."""
        # delete token
        self.endpoint_def.token = ''
        self.endpoint_def.save()

        # check
        response = self.client.post(
            self.url,
            content_type=CONTENT_TYPE_JSON,
        )

        assert response.status_code == HTTPStatus.OK
        assert str(response.content, 'utf-8') == WebhookView.model_class.MESSAGE_OK

    def test_bad_hmac(self):
        """Test that a malformed HMAC does not pass."""
        # delete token
        self.endpoint_def.token = ''
        self.endpoint_def.secret = '123abc'
        self.endpoint_def.save()

        # check
        response = self.client.post(
            self.url,
            content_type=CONTENT_TYPE_JSON,
        )

        assert response.status_code == HTTPStatus.FORBIDDEN
        assert (json.loads(response.content)['detail'] == WebhookView.model_class.MESSAGE_TOKEN_ERROR)

    def test_success_hmac(self):
        """Test with a valid HMAC provided."""
        # delete token
        self.endpoint_def.token = ''
        self.endpoint_def.secret = '123abc'
        self.endpoint_def.save()

        # check
        response = self.client.post(
            self.url,
            content_type=CONTENT_TYPE_JSON,
            **{'HTTP_TOKEN': str('68MXtc/OiXdA5e2Nq9hATEVrZFpLb3Zb0oau7n8s31I=')},
        )

        assert response.status_code == HTTPStatus.OK
        assert str(response.content, 'utf-8') == WebhookView.model_class.MESSAGE_OK

    def test_success(self):
        """Test full e2e webhook call.

        The message should go through and save the json payload.
        """
        response = self.client.post(
            self.url,
            data={"this": "is a message"},
            content_type=CONTENT_TYPE_JSON,
            **{'HTTP_TOKEN': str(self.endpoint_def.token)},
        )

        assert response.status_code == HTTPStatus.OK
        assert str(response.content, 'utf-8') == WebhookView.model_class.MESSAGE_OK
        message = WebhookMessage.objects.get()
        assert message.body == {"this": "is a message"}


class NotificationTest(InvenTreeAPITestCase):
    """Tests for NotificationEntry."""

    fixtures = [
        'users',
    ]

    def test_check_notification_entries(self):
        """Test that notification entries can be created."""
        # Create some notification entries

        self.assertEqual(NotificationEntry.objects.count(), 0)

        NotificationEntry.notify('test.notification', 1)

        self.assertEqual(NotificationEntry.objects.count(), 1)

        delta = timedelta(days=1)

        self.assertFalse(NotificationEntry.check_recent('test.notification', 2, delta))
        self.assertFalse(NotificationEntry.check_recent('test.notification2', 1, delta))

        self.assertTrue(NotificationEntry.check_recent('test.notification', 1, delta))

    def test_api_list(self):
        """Test list URL."""

        url = reverse('api-notifications-list')

        self.get(url, expected_code=200)

        # Test the OPTIONS endpoint for the 'api-notification-list'
        # Ref: https://github.com/inventree/InvenTree/pull/3154
        response = self.options(url)

        self.assertIn('DELETE', response.data['actions'])
        self.assertIn('GET', response.data['actions'])
        self.assertNotIn('POST', response.data['actions'])

        self.assertEqual(response.data['description'], 'List view for all notifications of the current user.')

        # POST action should fail (not allowed)
        response = self.post(url, {}, expected_code=405)

    def test_bulk_delete(self):
        """Tests for bulk deletion of user notifications"""

        from error_report.models import Error

        # Create some notification messages by throwing errors
        for _ii in range(10):
            Error.objects.create()

        # Check that messages have been created
        messages = NotificationMessage.objects.all()

        # As there are three staff users (including the 'test' user) we expect 30 notifications
        # However, one user is marked as inactive
        self.assertEqual(messages.count(), 20)

        # Only 10 messages related to *this* user
        my_notifications = messages.filter(user=self.user)
        self.assertEqual(my_notifications.count(), 10)

        # Get notification via the API
        url = reverse('api-notifications-list')
        response = self.get(url, {}, expected_code=200)
        self.assertEqual(len(response.data), 10)

        # Mark some as read
        for ntf in my_notifications[0:3]:
            ntf.read = True
            ntf.save()

        # Read out via API again
        response = self.get(
            url,
            {
                'read': True,
            },
            expected_code=200
        )

        # Check validity of returned data
        self.assertEqual(len(response.data), 3)
        for ntf in response.data:
            self.assertTrue(ntf['read'])

        # Now, let's bulk delete all 'unread' notifications via the API,
        # but only associated with the logged in user
        response = self.delete(
            url,
            {
                'filters': {
                    'read': False,
                }
            },
            expected_code=204,
        )

        # Only 7 notifications should have been deleted,
        # as the notifications associated with other users must remain untouched
        self.assertEqual(NotificationMessage.objects.count(), 13)
        self.assertEqual(NotificationMessage.objects.filter(user=self.user).count(), 3)


class CommonTest(InvenTreeAPITestCase):
    """Tests for the common config."""

    def test_restart_flag(self):
        """Test that the restart flag is reset on start."""
        import common.models
        from plugin import registry

        # set flag true
        common.models.InvenTreeSetting.set_setting('SERVER_RESTART_REQUIRED', True, None)

        # reload the app
        registry.reload_plugins()

        # now it should be false again
        self.assertFalse(common.models.InvenTreeSetting.get_setting('SERVER_RESTART_REQUIRED'))

    def test_config_api(self):
        """Test config URLs."""
        # Not superuser
        self.get(reverse('api-config-list'), expected_code=403)

        # Turn into superuser
        self.user.is_superuser = True
        self.user.save()

        # Successful checks
        data = [
            self.get(reverse('api-config-list'), expected_code=200).data[0],                                    # list endpoint
            self.get(reverse('api-config-detail', kwargs={'key': 'INVENTREE_DEBUG'}), expected_code=200).data,  # detail endpoint
        ]

        for item in data:
            self.assertEqual(item['key'], 'INVENTREE_DEBUG')
            self.assertEqual(item['env_var'], 'INVENTREE_DEBUG')
            self.assertEqual(item['config_key'], 'debug')

        # Turn into normal user again
        self.user.is_superuser = False
        self.user.save()

    def test_flag_api(self):
        """Test flag URLs."""
        # Not superuser
        response = self.get(reverse('api-flag-list'), expected_code=200)
        self.assertEqual(len(response.data), 2)
        self.assertEqual(response.data[0]['key'], 'EXPERIMENTAL')

        # Turn into superuser
        self.user.is_superuser = True
        self.user.save()

        # Successful checks
        response = self.get(reverse('api-flag-list'), expected_code=200)
        self.assertEqual(len(response.data), 2)
        self.assertEqual(response.data[0]['key'], 'EXPERIMENTAL')
        self.assertTrue(response.data[0]['conditions'])

        response = self.get(reverse('api-flag-detail', kwargs={'key': 'EXPERIMENTAL'}), expected_code=200)
        self.assertEqual(len(response.data), 3)
        self.assertEqual(response.data['key'], 'EXPERIMENTAL')
        self.assertTrue(response.data['conditions'])

        # Try without param -> false
        response = self.get(reverse('api-flag-detail', kwargs={'key': 'NEXT_GEN'}), expected_code=200)
        self.assertFalse(response.data['state'])

        # Try with param -> true
        response = self.get(reverse('api-flag-detail', kwargs={'key': 'NEXT_GEN'}), {'ngen': ''}, expected_code=200)
        self.assertTrue(response.data['state'])

        # Try non existent flag
        response = self.get(reverse('api-flag-detail', kwargs={'key': 'NON_EXISTENT'}), expected_code=404)

        # Turn into normal user again
        self.user.is_superuser = False
        self.user.save()


class ColorThemeTest(TestCase):
    """Tests for ColorTheme."""

    def test_choices(self):
        """Test that default choices are returned."""
        result = ColorTheme.get_color_themes_choices()

        # skip
        if not result:
            return
        self.assertIn(('default', 'Default'), result)

    def test_valid_choice(self):
        """Check that is_valid_choice works correctly."""
        result = ColorTheme.get_color_themes_choices()

        # skip
        if not result:
            return

        # check wrong reference
        self.assertFalse(ColorTheme.is_valid_choice('abcdd'))

        # create themes
        aa = ColorTheme.objects.create(user='aa', name='testname')
        ab = ColorTheme.objects.create(user='ab', name='darker')

        # check valid theme
        self.assertFalse(ColorTheme.is_valid_choice(aa))
        self.assertTrue(ColorTheme.is_valid_choice(ab))


class CurrencyAPITests(InvenTreeAPITestCase):
    """Unit tests for the currency exchange API endpoints"""

    def test_exchange_endpoint(self):
        """Test that the currency exchange endpoint works as expected"""

        response = self.get(reverse('api-currency-exchange'), expected_code=200)

        self.assertIn('base_currency', response.data)
        self.assertIn('exchange_rates', response.data)

    def test_refresh_endpoint(self):
        """Call the 'refresh currencies' endpoint"""

        from djmoney.contrib.exchange.models import Rate

        # Delete any existing exchange rate data
        Rate.objects.all().delete()

        # Updating via the external exchange may not work every time
        for _idx in range(5):
            self.post(reverse('api-currency-refresh'))

            # There should be some new exchange rate objects now
            if Rate.objects.all().exists():
                # Exit early
                return

            # Delay and try again
            time.sleep(10)

        raise TimeoutError("Could not refresh currency exchange data after 5 attempts")


class NotesImageTest(InvenTreeAPITestCase):
    """Tests for uploading images to be used in markdown notes."""

    def test_invalid_files(self):
        """Test that invalid files are rejected."""

        n = NotesImage.objects.count()

        # Test upload of a simple text file
        response = self.post(
            reverse('api-notes-image-list'),
            data={
                'image': SimpleUploadedFile('test.txt', b"this is not an image file", content_type='text/plain'),
            },
            format='multipart',
            expected_code=400
        )

        self.assertIn("Upload a valid image", str(response.data['image']))

        # Test upload of an invalid image file
        response = self.post(
            reverse('api-notes-image-list'),
            data={
                'image': SimpleUploadedFile('test.png', b"this is not an image file", content_type='image/png'),
            },
            format='multipart',
            expected_code=400,
        )

        self.assertIn("Upload a valid image", str(response.data['image']))

        # Check that no extra database entries have been created
        self.assertEqual(NotesImage.objects.count(), n)

    def test_valid_image(self):
        """Test upload of a valid image file"""

        n = NotesImage.objects.count()

        # Construct a simple image file
        image = PIL.Image.new('RGB', (100, 100), color='red')

        with io.BytesIO() as output:
            image.save(output, format='PNG')
            contents = output.getvalue()

        self.post(
            reverse('api-notes-image-list'),
            data={
                'image': SimpleUploadedFile('test.png', contents, content_type='image/png'),
            },
            format='multipart',
            expected_code=201
        )

        # Check that a new file has been created
        self.assertEqual(NotesImage.objects.count(), n + 1)


class ProjectCodesTest(InvenTreeAPITestCase):
    """Units tests for the ProjectCodes model and API endpoints"""

    @property
    def url(self):
        """Return the URL for the project code list endpoint"""
        return reverse('api-project-code-list')

    @classmethod
    def setUpTestData(cls):
        """Create some initial project codes"""
        super().setUpTestData()

        codes = [
            ProjectCode(code='PRJ-001', description='Test project code'),
            ProjectCode(code='PRJ-002', description='Test project code'),
            ProjectCode(code='PRJ-003', description='Test project code'),
            ProjectCode(code='PRJ-004', description='Test project code'),
        ]

        ProjectCode.objects.bulk_create(codes)

    def test_list(self):
        """Test that the list endpoint works as expected"""

        response = self.get(self.url, expected_code=200)
        self.assertEqual(len(response.data), ProjectCode.objects.count())

    def test_delete(self):
        """Test we can delete a project code via the API"""

        n = ProjectCode.objects.count()

        # Get the first project code
        code = ProjectCode.objects.first()

        # Delete it
        self.delete(
            reverse('api-project-code-detail', kwargs={'pk': code.pk}),
            expected_code=204
        )

        # Check it is gone
        self.assertEqual(ProjectCode.objects.count(), n - 1)

    def test_duplicate_code(self):
        """Test that we cannot create two project codes with the same code"""

        # Create a new project code
        response = self.post(
            self.url,
            data={
                'code': 'PRJ-001',
                'description': 'Test project code',
            },
            expected_code=400
        )

        self.assertIn('project code with this Project Code already exists', str(response.data['code']))

    def test_write_access(self):
        """Test that non-staff users have read-only access"""

        # By default user has staff access, can create a new project code
        response = self.post(
            self.url,
            data={
                'code': 'PRJ-xxx',
                'description': 'Test project code',
            },
            expected_code=201
        )

        pk = response.data['pk']

        # Test we can edit, also
        response = self.patch(
            reverse('api-project-code-detail', kwargs={'pk': pk}),
            data={
                'code': 'PRJ-999',
            },
            expected_code=200
        )

        self.assertEqual(response.data['code'], 'PRJ-999')

        # Restrict user access to non-staff
        self.user.is_staff = False
        self.user.save()

        # As user does not have staff access, should return 403 for list endpoint
        response = self.post(
            self.url,
            data={
                'code': 'PRJ-123',
                'description': 'Test project code'
            },
            expected_code=403
        )

        # Should also return 403 for detail endpoint
        response = self.patch(
            reverse('api-project-code-detail', kwargs={'pk': pk}),
            data={
                'code': 'PRJ-999',
            },
            expected_code=403
        )


class CustomUnitAPITest(InvenTreeAPITestCase):
    """Unit tests for the CustomUnit API"""

    @property
    def url(self):
        """Return the API endpoint for the CustomUnit list"""
        return reverse('api-custom-unit-list')

    @classmethod
    def setUpTestData(cls):
        """Construct some initial test fixture data"""
        super().setUpTestData()

        units = [
            CustomUnit(name='metres_per_amp', definition='meter / ampere', symbol='m/A'),
            CustomUnit(name='hectares_per_second', definition='hectares per second', symbol='ha/s'),
        ]

        CustomUnit.objects.bulk_create(units)

    def test_list(self):
        """Test API list functionality"""

        response = self.get(self.url, expected_code=200)
        self.assertEqual(len(response.data), CustomUnit.objects.count())

    def test_edit(self):
        """Test edit permissions for CustomUnit model"""

        unit = CustomUnit.objects.first()

        # Try to edit without permission
        self.user.is_staff = False
        self.user.save()

        self.patch(
            reverse('api-custom-unit-detail', kwargs={'pk': unit.pk}),
            {
                'name': 'new_unit_name',
            },
            expected_code=403
        )

        # Ok, what if we have permission?
        self.user.is_staff = True
        self.user.save()

        self.patch(
            reverse('api-custom-unit-detail', kwargs={'pk': unit.pk}),
            {
                'name': 'new_unit_name',
            },
            # expected_code=200
        )

        unit.refresh_from_db()
        self.assertEqual(unit.name, 'new_unit_name')

    def test_validation(self):
        """Test that validation works as expected"""

        unit = CustomUnit.objects.first()

        self.user.is_staff = True
        self.user.save()

        # Test invalid 'name' values (must be valid identifier)
        invalid_name_values = [
            '1',
            '1abc',
            'abc def',
            'abc-def',
            'abc.def',
        ]

        url = reverse('api-custom-unit-detail', kwargs={'pk': unit.pk})

        for name in invalid_name_values:
            self.patch(
                url,
                {
                    'name': name,
                },
                expected_code=400
            )