netbox/netbox/core/tests/test_api.py

import uuid

from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.utils import timezone
from django_rq import get_queue
from django_rq.workers import get_worker
from rest_framework import status
from rq.job import Job as RQ_Job
from rq.job import JobStatus
from rq.registry import FailedJobRegistry, StartedJobRegistry

from users.constants import TOKEN_PREFIX
from users.models import Token
from utilities.testing import APITestCase, APIViewTestCases, GraphQLQueryTest, TestCase
from utilities.testing.mixins import RQQueueTestMixin
from utilities.testing.utils import disable_logging

from ..models import *


class AppTestCase(APITestCase):

    def test_root(self):
        url = reverse('core-api:api-root')
        response = self.client.get('{}?format=api'.format(url), **self.header)

        self.assertEqual(response.status_code, 200)


class DataSourceTestCase(APIViewTestCases.APIViewTestCase):
    model = DataSource
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    bulk_update_data = {
        'enabled': False,
        'description': 'foo bar baz',
    }

    @classmethod
    def setUpTestData(cls):
        data_sources = (
            DataSource(
                name='Data Source 1', type='local', source_url='file:///var/tmp/source1/',
                parameters={
                    'sync_date': '2024-01-01',
                    'sync_datetime': '2024-01-01T12:30:00+00:00',
                    'sync_time': '12:30:00',
                },
            ),
            DataSource(name='Data Source 2', type='local', source_url='file:///var/tmp/source2/'),
            DataSource(name='Data Source 3', type='local', source_url='file:///var/tmp/source3/'),
        )
        DataSource.objects.bulk_create(data_sources)

        cls.graphql_query_tests = (
            GraphQLQueryTest(
                name='parameters_json_date_lookup',
                query=(
                    '{ data_source_list(filters: {parameters: '
                    '{path: "sync_date", lookup: {date_lookup: {exact: "2024-01-01"}}}}) '
                    '{ id } }'
                ),
                assert_result=cls.assert_only_source_1,
            ),
            GraphQLQueryTest(
                name='parameters_json_datetime_lookup',
                query=(
                    '{ data_source_list(filters: {parameters: '
                    '{path: "sync_datetime", lookup: {datetime_lookup: {exact: "2024-01-01T12:30:00+00:00"}}}}) '
                    '{ id } }'
                ),
                assert_result=cls.assert_only_source_1,
            ),
            GraphQLQueryTest(
                name='parameters_json_time_lookup',
                query=(
                    '{ data_source_list(filters: {parameters: '
                    '{path: "sync_time", lookup: {time_lookup: {exact: "12:30:00"}}}}) '
                    '{ id } }'
                ),
                assert_result=cls.assert_only_source_1,
            ),
        )

        cls.create_data = [
            {
                'name': 'Data Source 4',
                'type': 'git',
                'source_url': 'https://example.com/git/source4'
            },
            {
                'name': 'Data Source 5',
                'type': 'git',
                'source_url': 'https://example.com/git/source5'
            },
            {
                'name': 'Data Source 6',
                'type': 'git',
                'source_url': 'https://example.com/git/source6'
            },
        ]

    def assert_only_source_1(self, data):
        """The JSON lookup returns exactly the source carrying the matching value."""
        ids = sorted(result['id'] for result in data['data_source_list'])
        self.assertEqual(ids, [str(DataSource.objects.get(name='Data Source 1').pk)])


class DataFileTestCase(
    APIViewTestCases.GetObjectViewTestCase,
    APIViewTestCases.ListObjectsViewTestCase,
    APIViewTestCases.GraphQLTestCase
):
    model = DataFile
    brief_fields = ['display', 'id', 'path', 'url']
    user_permissions = ('core.view_datasource', )

    @classmethod
    def setUpTestData(cls):
        datasource = DataSource.objects.create(
            name='Data Source 1',
            type='local',
            source_url='file:///var/tmp/source1/'
        )

        data_files = (
            DataFile(
                source=datasource,
                path='dir1/file1.txt',
                last_updated=timezone.now(),
                size=1000,
                hash='442da078f0111cbdf42f21903724f6597c692535f55bdfbbea758a1ae99ad9e1'
            ),
            DataFile(
                source=datasource,
                path='dir1/file2.txt',
                last_updated=timezone.now(),
                size=2000,
                hash='a78168c7c97115bafd96450ed03ea43acec495094c5caa28f0d02e20e3a76cc2'
            ),
            DataFile(
                source=datasource,
                path='dir1/file3.txt',
                last_updated=timezone.now(),
                size=3000,
                hash='12b8827a14c4d5a2f30b6c6e2b7983063988612391c6cbe8ee7493b59054827a'
            ),
        )
        DataFile.objects.bulk_create(data_files)


class ObjectTypeTestCase(APITestCase):
    model = ObjectType

    def test_list_objects(self):
        object_type_count = ObjectType.objects.count()

        response = self.client.get(reverse('core-api:objecttype-list'), **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data['count'], object_type_count)

    def test_get_object(self):
        object_type = ObjectType.objects.first()

        url = reverse('core-api:objecttype-detail', kwargs={'pk': object_type.pk})
        self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_200_OK)


class JobTestCase(
    APIViewTestCases.GetObjectViewTestCase,
    APIViewTestCases.ListObjectsViewTestCase,
):
    model = Job
    brief_fields = ['completed', 'created', 'status', 'url', 'user']

    @classmethod
    def setUpTestData(cls):
        datasource = DataSource.objects.create(
            name='Data Source 1',
            type='local',
            source_url='file:///var/tmp/source1/',
        )
        ct = ContentType.objects.get_for_model(DataSource)
        Job.objects.bulk_create(
            [
                Job(
                    name='Job 1',
                    object_type=ct,
                    object_id=datasource.pk,
                    status='pending',
                    queue_name='default',
                    job_id=uuid.uuid4(),
                ),
                Job(
                    name='Job 2',
                    object_type=ct,
                    object_id=datasource.pk,
                    status='running',
                    queue_name='default',
                    job_id=uuid.uuid4(),
                ),
                Job(
                    name='Job 3',
                    object_type=ct,
                    object_id=datasource.pk,
                    status='completed',
                    queue_name='default',
                    job_id=uuid.uuid4(),
                    execution_time=timezone.timedelta(seconds=90),
                ),
            ]
        )

    def test_list_objects_by_execution_time(self):
        """The Job list endpoint supports filtering and ordering by execution_time."""
        self.add_permissions('core.view_job')
        url = reverse('core-api:job-list')

        # Filter: only the completed job has a (90s) execution_time
        response = self.client.get(f'{url}?execution_time__gte=60', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data['count'], 1)

        # Ordering by execution_time should be accepted (NULLs sort to one end)
        response = self.client.get(f'{url}?ordering=execution_time', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data['count'], 3)


class BackgroundTaskTestCase(RQQueueTestMixin, TestCase):
    user_permissions = ()

    @staticmethod
    def dummy_job_default():
        return "Job finished"

    @staticmethod
    def dummy_job_failing():
        raise Exception("Job failed")

    def setUp(self):
        super().setUp()

        # The base TestCase creates self.user; make it active and create a token for API calls.
        self.user.is_active = True
        self.user.save()
        self.token = Token.objects.create(user=self.user)
        self.header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{self.token.key}.{self.token.token}'}

    def test_background_queue_list(self):
        url = reverse('core-api:rqqueue-list')

        # Attempt to load view without permission
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Load view with permission
        self.user.is_superuser = True
        self.user.save()
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 200)
        self.assertIn('default', str(response.content))
        self.assertIn('high', str(response.content))
        self.assertIn('low', str(response.content))

    def test_background_queue(self):
        url = reverse('core-api:rqqueue-detail', args=['default'])

        # Attempt to load view without permission
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Load view with permission
        self.user.is_superuser = True
        self.user.save()
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 200)
        self.assertIn('default', str(response.content))
        self.assertIn('oldest_job_timestamp', str(response.content))
        self.assertIn('scheduled_jobs', str(response.content))

    def test_background_task_list(self):
        queue = get_queue('default')
        queue.enqueue(self.dummy_job_default)
        url = reverse('core-api:rqtask-list')

        # Attempt to load view without permission
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Load view with permission
        self.user.is_superuser = True
        self.user.save()
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 200)
        self.assertIn('origin', str(response.content))
        self.assertIn('core.tests.test_api.BackgroundTaskTestCase.dummy_job_default()', str(response.content))

    def test_background_task(self):
        queue = get_queue('default')
        job = queue.enqueue(self.dummy_job_default)
        url = reverse('core-api:rqtask-detail', args=[job.id])

        # Attempt to load view without permission
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Load view with permission
        self.user.is_superuser = True
        self.user.save()
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 200)
        self.assertIn(str(job.id), str(response.content))
        self.assertIn('origin', str(response.content))
        self.assertIn('meta', str(response.content))
        self.assertIn('kwargs', str(response.content))

    def test_background_task_delete(self):
        queue = get_queue('default')
        job = queue.enqueue(self.dummy_job_default)
        url = reverse('core-api:rqtask-delete', args=[job.id])

        # Attempt to load view without permission
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Load view with permission
        self.user.is_superuser = True
        self.user.save()
        response = self.client.post(url, **self.header)
        self.assertEqual(response.status_code, 200)
        self.assertFalse(RQ_Job.exists(job.id, connection=queue.connection))
        queue = get_queue('default')
        self.assertNotIn(job.id, queue.job_ids)

    def test_background_task_requeue(self):
        # Enqueue & run a job that will fail
        queue = get_queue('default')
        job = queue.enqueue(self.dummy_job_failing)
        with disable_logging():
            self.run_rq_jobs('default')
        self.assertTrue(job.is_failed)
        url = reverse('core-api:rqtask-requeue', args=[job.id])

        # Attempt to requeue the job without permission
        response = self.client.post(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Re-enqueue the failed job and check that its status has been reset
        self.user.is_superuser = True
        self.user.save()
        response = self.client.post(url, **self.header)
        self.assertEqual(response.status_code, 200)
        job = RQ_Job.fetch(job.id, queue.connection)
        self.assertFalse(job.is_failed)

    def test_background_task_enqueue(self):
        # Enqueue some jobs that each depends on its predecessor
        queue = get_queue('default')
        job = previous_job = None
        for _ in range(0, 3):
            job = queue.enqueue(self.dummy_job_default, depends_on=previous_job)
            previous_job = job
        url = reverse('core-api:rqtask-enqueue', args=[job.id])

        # Check that the last job to be enqueued has a status of deferred
        self.assertIsNotNone(job)
        self.assertEqual(job.get_status(), JobStatus.DEFERRED)
        self.assertIsNone(job.enqueued_at)

        # Attempt to force-enqueue the job without permission
        response = self.client.post(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Force-enqueue the deferred job
        self.user.is_superuser = True
        self.user.save()
        response = self.client.post(url, **self.header)
        self.assertEqual(response.status_code, 200)

        # Check that job's status is updated correctly
        job = queue.fetch_job(job.id)
        self.assertEqual(job.get_status(), JobStatus.QUEUED)
        self.assertIsNotNone(job.enqueued_at)

    def test_background_task_stop(self):
        queue = get_queue('default')
        worker = get_worker('default')
        job = queue.enqueue(self.dummy_job_default)
        worker.prepare_job_execution(job)
        url = reverse('core-api:rqtask-stop', args=[job.id])
        self.assertEqual(job.get_status(), JobStatus.STARTED)

        # Attempt to stop the task without permission
        response = self.client.post(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Stop the task
        self.user.is_superuser = True
        self.user.save()
        response = self.client.post(url, **self.header)
        self.assertEqual(response.status_code, 200)
        with disable_logging():
            worker.monitor_work_horse(job, queue)  # Sets the job as Failed and removes from Started
        started_job_registry = StartedJobRegistry(queue.name, connection=queue.connection)
        self.assertEqual(len(started_job_registry), 0)

        # Verify that the task was cancelled
        canceled_job_registry = FailedJobRegistry(queue.name, connection=queue.connection)
        self.assertEqual(len(canceled_job_registry), 1)
        self.assertIn(job.id, canceled_job_registry)

    def test_worker_list(self):
        worker1 = get_worker('default', name=uuid.uuid4().hex)
        worker1.register_birth()
        worker2 = get_worker('high')
        worker2.register_birth()
        url = reverse('core-api:rqworker-list')

        # Attempt to fetch the worker list without permission
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Fetch the worker list
        self.user.is_superuser = True
        self.user.save()
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 200)
        self.assertIn(str(worker1.name), str(response.content))

    def test_worker(self):
        worker1 = get_worker('default', name=uuid.uuid4().hex)
        worker1.register_birth()
        url = reverse('core-api:rqworker-detail', args=[worker1.name])

        # Attempt to fetch a worker without permission
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 403)

        # Fetch the worker
        self.user.is_superuser = True
        self.user.save()
        response = self.client.get(url, **self.header)
        self.assertEqual(response.status_code, 200)
        self.assertIn(str(worker1.name), str(response.content))
        self.assertIn('birth_date', str(response.content))
        self.assertIn('total_working_time', str(response.content))