"""Tests for the Order API."""

import base64
import io
import json
from datetime import date, datetime, timedelta
from typing import Optional

from django.core.exceptions import ValidationError
from django.db import connection
from django.test.utils import CaptureQueriesContext
from django.urls import reverse

from djmoney.money import Money
from icalendar import Calendar
from rest_framework import status

from common.currency import currency_codes
from common.models import InvenTreeSetting
from common.settings import set_global_setting
from company.models import Company, SupplierPart, SupplierPriceBreak
from InvenTree.unit_test import InvenTreeAPITestCase
from order import models
from order.status_codes import (
    PurchaseOrderStatus,
    ReturnOrderLineStatus,
    ReturnOrderStatus,
    SalesOrderStatus,
    SalesOrderStatusGroups,
)
from part.models import Part
from stock.models import StockItem, StockLocation
from stock.status_codes import StockStatus
from users.models import Owner


class OrderTest(InvenTreeAPITestCase):
    """Base class for order API unit testing."""

    fixtures = [
        'category',
        'part',
        'company',
        'location',
        'supplier_part',
        'stock',
        'order',
        'sales_order',
    ]

    roles = ['purchase_order.change', 'sales_order.change']

    def filter(self, filters, count):
        """Test API filters."""
        response = self.get(self.LIST_URL, filters)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(len(response.data), count)

        return response


class PurchaseOrderTest(OrderTest):
    """Tests for the PurchaseOrder API."""

    LIST_URL = reverse('api-po-list')

    def test_options(self):
        """Test the PurchaseOrder OPTIONS endpoint."""
        self.assignRole('purchase_order.add')

        response = self.options(self.LIST_URL, expected_code=200)

        data = response.data
        self.assertEqual(data['name'], 'Purchase Order List')

        post = data['actions']['POST']

        def check_options(data, field_name, spec):
            """Helper function to check that the options are configured correctly."""
            field_data = data[field_name]

            for k, v in spec.items():
                self.assertIn(k, field_data)
                self.assertEqual(field_data[k], v)

        # Checks for the 'order_currency' field
        check_options(
            post,
            'order_currency',
            {
                'type': 'choice',
                'required': False,
                'read_only': False,
                'label': 'Order Currency',
                'help_text': 'Currency for this order (leave blank to use company default)',
            },
        )

        # Checks for the 'reference' field
        check_options(
            post,
            'reference',
            {
                'type': 'string',
                'required': True,
                'read_only': False,
                'label': 'Reference',
            },
        )

        # Checks for the 'supplier' field
        check_options(
            post,
            'supplier',
            {'type': 'related field', 'required': True, 'api_url': '/api/company/'},
        )

    def test_po_list(self):
        """Test the PurchaseOrder list API endpoint."""
        # List *ALL* PurchaseOrder items
        self.filter({}, 7)

        # Filter by assigned-to-me
        self.filter({'assigned_to_me': 1}, 0)
        self.filter({'assigned_to_me': 0}, 7)

        # Filter by supplier
        self.filter({'supplier': 1}, 1)
        self.filter({'supplier': 3}, 5)

        # Filter by "outstanding"
        self.filter({'outstanding': True}, 5)
        self.filter({'outstanding': False}, 2)

        # Filter by "status"
        self.filter({'status': PurchaseOrderStatus.PENDING.value}, 3)
        self.filter({'status': PurchaseOrderStatus.CANCELLED.value}, 1)

        # Filter by "reference"
        self.filter({'reference': 'PO-0001'}, 1)
        self.filter({'reference': 'PO-9999'}, 0)

        # Filter by "assigned_to_me"
        self.filter({'assigned_to_me': 1}, 0)
        self.filter({'assigned_to_me': 0}, 7)

        # Filter by "part"
        self.filter({'part': 1}, 2)
        self.filter({'part': 2}, 0)  # Part not assigned to any PO

        # Filter by "supplier_part"
        self.filter({'supplier_part': 1}, 1)
        self.filter({'supplier_part': 3}, 2)
        self.filter({'supplier_part': 4}, 0)

    def test_total_price(self):
        """Unit tests for the 'total_price' field."""
        # Ensure we have exchange rate data
        self.generate_exchange_rates()

        currencies = currency_codes()
        n = len(currencies)

        idx = 0

        new_orders = []

        # Let's generate some more orders
        for supplier in Company.objects.filter(is_supplier=True):
            for _idx in range(10):
                new_orders.append(
                    models.PurchaseOrder(supplier=supplier, reference=f'PO-{idx + 100}')
                )

                idx += 1

        models.PurchaseOrder.objects.bulk_create(new_orders)

        idx = 0

        # Create some purchase order line items
        lines = []

        for po in models.PurchaseOrder.objects.all():
            for sp in po.supplier.supplied_parts.all():
                lines.append(
                    models.PurchaseOrderLineItem(
                        order=po,
                        part=sp,
                        quantity=idx + 1,
                        purchase_price=Money((idx + 1) / 10, currencies[idx % n]),
                    )
                )

                idx += 1

        models.PurchaseOrderLineItem.objects.bulk_create(lines)

        # List all purchase orders
        for limit in [1, 5, 10, 100]:
            with CaptureQueriesContext(connection) as ctx:
                response = self.get(
                    self.LIST_URL, data={'limit': limit}, expected_code=200
                )

                # Total database queries must be below 25, independent of the number of results
                self.assertLess(len(ctx), 25)

                for result in response.data['results']:
                    self.assertIn('total_price', result)
                    self.assertIn('order_currency', result)

    def test_overdue(self):
        """Test "overdue" status."""
        self.filter({'overdue': True}, 0)
        self.filter({'overdue': False}, 7)

        order = models.PurchaseOrder.objects.get(pk=1)
        order.target_date = datetime.now().date() - timedelta(days=10)
        order.save()

        self.filter({'overdue': True}, 1)
        self.filter({'overdue': False}, 6)

    def test_po_detail(self):
        """Test the PurchaseOrder detail API endpoint."""
        url = '/api/order/po/1/'

        response = self.get(url)

        self.assertEqual(response.status_code, 200)

        data = response.data

        self.assertEqual(data['pk'], 1)
        self.assertEqual(data['description'], 'Ordering some screws')

    def test_po_reference(self):
        """Test that a reference with a too big / small reference is handled correctly."""
        # get permissions
        self.assignRole('purchase_order.add')

        url = reverse('api-po-list')
        huge_number = 'PO-92233720368547758089999999999999999'

        response = self.post(
            url,
            {
                'supplier': 1,
                'reference': huge_number,
                'description': 'PO created via the API',
            },
            expected_code=201,
        )

        order = models.PurchaseOrder.objects.get(pk=response.data['pk'])

        # Check that the created_by field is set correctly
        self.assertEqual(order.created_by.username, 'testuser')

        self.assertEqual(order.reference, huge_number)
        self.assertEqual(order.reference_int, 0x7FFFFFFF)

    def test_po_reference_wildcard_default(self):
        """Test that a reference with a wildcard default."""
        # get permissions
        self.assignRole('purchase_order.add')

        # set PO reference setting
        set_global_setting('PURCHASEORDER_REFERENCE_PATTERN', '{?:PO}-{ref:04d}')

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

        # first, check that the default character is suggested by OPTIONS
        options = json.loads(self.options(url).content)
        suggested_reference = options['actions']['POST']['reference']['default']
        self.assertTrue(suggested_reference.startswith('PO-'))

        # next, check that certain variations of a provided reference are accepted
        test_accepted_references = ['PO-9991', 'P-9992', 'T-9993', 'ABC-9994']
        for ref in test_accepted_references:
            response = self.post(
                url,
                {
                    'supplier': 1,
                    'reference': ref,
                    'description': 'PO created via the API',
                },
                expected_code=201,
            )
            order = models.PurchaseOrder.objects.get(pk=response.data['pk'])
            self.assertEqual(order.reference, ref)

        # finally, check that certain provided referencees are rejected (because the wildcard character is required!)
        test_rejected_references = ['9995', '-9996']
        for ref in test_rejected_references:
            response = self.post(
                url,
                {
                    'supplier': 1,
                    'reference': ref,
                    'description': 'PO created via the API',
                },
                expected_code=400,
            )

    def test_po_attachments(self):
        """Test the list endpoint for the PurchaseOrderAttachment model."""
        url = reverse('api-attachment-list')

        response = self.get(url, {'model_type': 'purchaseorder'})

        self.assertEqual(response.status_code, status.HTTP_200_OK)

    def test_output_options(self):
        """Test the various output options for the PurchaseOrder detail endpoint."""
        self.run_output_test(
            reverse('api-po-detail', kwargs={'pk': 1}), ['supplier_detail']
        )

    def test_po_operations(self):
        """Test that we can create / edit and delete a PurchaseOrder via the API."""
        n = models.PurchaseOrder.objects.count()

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

        # Initially we do not have "add" permission for the PurchaseOrder model,
        # so this POST request should return 403
        response = self.post(
            url,
            {
                'supplier': 1,
                'reference': '123456789-xyz',
                'description': 'PO created via the API',
            },
            expected_code=403,
        )

        # And no new PurchaseOrder objects should have been created
        self.assertEqual(models.PurchaseOrder.objects.count(), n)

        # Ok, now let's give this user the correct permission
        self.assignRole('purchase_order.add')

        # Initially we do not have "add" permission for the PurchaseOrder model,
        # so this POST request should return 403
        response = self.post(
            url,
            {
                'supplier': 1,
                'reference': 'PO-123456789',
                'description': 'PO created via the API',
            },
            expected_code=201,
        )

        self.assertEqual(models.PurchaseOrder.objects.count(), n + 1)

        pk = response.data['pk']

        # Try to create a PurchaseOrder with identical reference (should fail!)
        response = self.post(
            url,
            {
                'supplier': 1,
                'reference': '123456789-xyz',
                'description': 'A different description',
            },
            expected_code=400,
        )

        self.assertEqual(models.PurchaseOrder.objects.count(), n + 1)

        url = reverse('api-po-detail', kwargs={'pk': pk})

        # Get detail info!
        response = self.get(url)
        self.assertEqual(response.data['pk'], pk)
        self.assertEqual(response.data['reference'], 'PO-123456789')

        # Try to alter (edit) the PurchaseOrder
        response = self.patch(url, {'reference': 'PO-12345'}, expected_code=200)

        # Reference should have changed
        self.assertEqual(response.data['reference'], 'PO-12345')

        # Now, let's try to delete it!
        # Initially, we do *not* have the required permission!
        response = self.delete(url, expected_code=403)

        # Now, add the "delete" permission!
        self.assignRole('purchase_order.delete')

        response = self.delete(url, expected_code=204)

        # Number of PurchaseOrder objects should have decreased
        self.assertEqual(models.PurchaseOrder.objects.count(), n)

        # And if we try to access the detail view again, it has gone
        response = self.get(url, expected_code=404)

    def test_po_create(self):
        """Test that we can create a new PurchaseOrder via the API."""
        self.assignRole('purchase_order.add')

        setting = 'PURCHASEORDER_REQUIRE_RESPONSIBLE'
        url = reverse('api-po-list')

        InvenTreeSetting.set_setting(setting, False)

        data = {
            'reference': 'PO-12345678',
            'supplier': 1,
            'description': 'A test purchase order',
        }

        self.post(url, data, expected_code=201)

        # Check the 'responsible required' field
        InvenTreeSetting.set_setting(setting, True)

        data['reference'] = 'PO-12345679'
        data['responsible'] = None

        response = self.post(url, data, expected_code=400)

        self.assertIn('Responsible user or group must be specified', str(response.data))

        owner = Owner.objects.first()
        assert owner
        data['responsible'] = owner.pk

        response = self.post(url, data, expected_code=201)

        # Revert the setting to previous value
        InvenTreeSetting.set_setting(setting, False)

    def test_po_creation_date(self):
        """Test that we can create set the creation_date field of PurchaseOrder via the API."""
        self.assignRole('purchase_order.add')

        response = self.post(
            reverse('api-po-list'),
            {
                'reference': 'PO-19881110',
                'supplier': 1,
                'description': 'PO created on 1988-11-10',
                'creation_date': '1988-11-10',
            },
            expected_code=201,
        )

        po = models.PurchaseOrder.objects.get(pk=response.data['pk'])
        self.assertEqual(po.creation_date, datetime(1988, 11, 10).date())

        """Ensure if we do not pass the creation_date field than the current date will be saved"""
        creation_date = datetime.now().date()
        response = self.post(
            reverse('api-po-list'),
            {
                'reference': 'PO-11111111',
                'supplier': 1,
                'description': 'Check that the creation date is today',
            },
            expected_code=201,
        )

        po = models.PurchaseOrder.objects.get(pk=response.data['pk'])
        self.assertEqual(po.creation_date, creation_date)

    def test_po_duplicate(self):
        """Test that we can duplicate a PurchaseOrder via the API."""
        self.assignRole('purchase_order.add')

        po = models.PurchaseOrder.objects.get(pk=1)

        self.assertGreater(po.lines.count(), 0)

        lines = []

        # Add some extra line items to this order
        for idx in range(5):
            lines.append(
                models.PurchaseOrderExtraLine(
                    order=po, quantity=idx + 10, reference='some reference'
                )
            )

        # bulk create orders
        models.PurchaseOrderExtraLine.objects.bulk_create(lines)

        data = self.get(reverse('api-po-detail', kwargs={'pk': 1})).data

        del data['pk']
        del data['reference']

        # Duplicate with non-existent PK to provoke error
        data['duplicate'] = {
            'order_id': 10000001,
            'copy_lines': True,
            'copy_extra_lines': False,
        }

        data['reference'] = 'PO-9999'

        # Duplicate via the API
        response = self.post(reverse('api-po-list'), data, expected_code=400)

        data['duplicate'] = {
            'order_id': 1,
            'copy_lines': True,
            'copy_extra_lines': False,
        }

        data['reference'] = 'PO-9999'

        # Duplicate via the API
        response = self.post(reverse('api-po-list'), data, expected_code=201)

        # Order is for the same supplier
        self.assertEqual(response.data['supplier'], po.supplier.pk)

        po_dup = models.PurchaseOrder.objects.get(pk=response.data['pk'])

        self.assertEqual(po_dup.extra_lines.count(), 0)
        self.assertEqual(po_dup.lines.count(), po.lines.count())

        data['reference'] = 'PO-9998'

        data['duplicate'] = {
            'order_id': 1,
            'copy_lines': False,
            'copy_extra_lines': True,
        }

        response = self.post(reverse('api-po-list'), data, expected_code=201)

        po_dup = models.PurchaseOrder.objects.get(pk=response.data['pk'])

        self.assertEqual(po_dup.extra_lines.count(), po.extra_lines.count())
        self.assertEqual(po_dup.lines.count(), 0)

    def test_po_cancel(self):
        """Test the PurchaseOrderCancel API endpoint."""
        po = models.PurchaseOrder.objects.get(pk=1)

        self.assertEqual(po.status, PurchaseOrderStatus.PENDING)

        url = reverse('api-po-cancel', kwargs={'pk': po.pk})

        # Get an OPTIONS request from the endpoint
        self.options(url, data={'context': True}, expected_code=200)

        # Try to cancel the PO, but without required permissions
        self.post(url, {}, expected_code=403)

        self.assignRole('purchase_order.add')

        self.post(url, {}, expected_code=201)

        po.refresh_from_db()

        self.assertEqual(po.status, PurchaseOrderStatus.CANCELLED)

        # Try to cancel again (should fail)
        self.post(url, {}, expected_code=400)

    def test_po_complete(self):
        """Test the PurchaseOrderComplete API endpoint."""
        po = models.PurchaseOrder.objects.get(pk=3)

        url = reverse('api-po-complete', kwargs={'pk': po.pk})

        self.assertEqual(po.status, PurchaseOrderStatus.PLACED)

        # Try to complete the PO, without required permissions
        self.post(url, {}, expected_code=403)

        self.assignRole('purchase_order.add')

        # Add a line item
        sp = SupplierPart.objects.filter(supplier=po.supplier).first()

        models.PurchaseOrderLineItem.objects.create(part=sp, order=po, quantity=100)

        # Should fail due to incomplete lines
        response = self.post(url, {}, expected_code=400)

        self.assertIn(
            'Order has incomplete line items', str(response.data['accept_incomplete'])
        )

        # Post again, accepting incomplete line items
        self.post(url, {'accept_incomplete': True}, expected_code=201)

        po.refresh_from_db()

        self.assertEqual(po.status, PurchaseOrderStatus.COMPLETE)

    def test_po_issue(self):
        """Test the PurchaseOrderIssue API endpoint."""
        po = models.PurchaseOrder.objects.get(pk=2)

        url = reverse('api-po-issue', kwargs={'pk': po.pk})

        # Try to issue the PO, without required permissions
        self.post(url, {}, expected_code=403)

        self.assignRole('purchase_order.add')

        self.post(url, {}, expected_code=201)

        po.refresh_from_db()

        self.assertEqual(po.status, PurchaseOrderStatus.PLACED)

    def test_po_calendar(self):
        """Test the calendar export endpoint."""
        # Create required purchase orders
        self.assignRole('purchase_order.add')

        for i in range(1, 9):
            self.post(
                reverse('api-po-list'),
                {
                    'reference': f'PO-1100000{i}',
                    'supplier': 1,
                    'description': f'Calendar PO {i}',
                    'target_date': f'2024-12-{i:02d}',
                },
                expected_code=201,
            )

        # Get some of these orders with target date, complete or cancel them
        for po in models.PurchaseOrder.objects.filter(target_date__isnull=False):
            if po.reference in [
                'PO-11000001',
                'PO-11000002',
                'PO-11000003',
                'PO-11000004',
            ]:
                # Set issued status for these POs
                self.post(
                    reverse('api-po-issue', kwargs={'pk': po.pk}), {}, expected_code=201
                )

                if po.reference in ['PO-11000001', 'PO-11000002']:
                    # Set complete status for these POs
                    self.post(
                        reverse('api-po-complete', kwargs={'pk': po.pk}),
                        {'accept_incomplete': True},
                        expected_code=201,
                    )

            elif po.reference in ['PO-11000005', 'PO-11000006']:
                # Set cancel status for these POs
                self.post(
                    reverse('api-po-cancel', kwargs={'pk': po.pk}),
                    {'accept_incomplete': True},
                    expected_code=201,
                )

        url = reverse('api-po-so-calendar', kwargs={'ordertype': 'purchase-order'})

        # Test without completed orders
        response = self.get(url, expected_code=200, format=None)

        number_orders = len(
            models.PurchaseOrder.objects.filter(target_date__isnull=False).filter(
                status__lt=PurchaseOrderStatus.COMPLETE.value
            )
        )

        # Transform content to a Calendar object
        calendar = Calendar.from_ical(response.content)
        n_events = 0
        # Count number of events in calendar
        for component in calendar.walk():
            if component.name == 'VEVENT':
                n_events += 1

        self.assertGreaterEqual(n_events, 1)
        self.assertEqual(number_orders, n_events)

        # Test with completed orders
        response = self.get(
            url, data={'include_completed': 'True'}, expected_code=200, format=None
        )

        number_orders_incl_completed = len(
            models.PurchaseOrder.objects.filter(target_date__isnull=False)
        )

        self.assertGreater(number_orders_incl_completed, number_orders)

        # Transform content to a Calendar object
        calendar = Calendar.from_ical(response.content)
        n_events = 0
        # Count number of events in calendar
        for component in calendar.walk():
            if component.name == 'VEVENT':
                n_events += 1

        self.assertGreaterEqual(n_events, 1)
        self.assertEqual(number_orders_incl_completed, n_events)

    def test_po_calendar_noauth(self):
        """Test accessing calendar without authorization."""
        self.client.logout()
        response = self.client.get(
            reverse('api-po-so-calendar', kwargs={'ordertype': 'purchase-order'}),
            format='json',
        )

        self.assertEqual(response.status_code, 401)

        resp_dict = response.json()
        self.assertEqual(
            resp_dict['detail'], 'Authentication credentials were not provided.'
        )

    def test_po_calendar_auth(self):
        """Test accessing calendar with header authorization."""
        self.client.logout()
        base64_token = base64.b64encode(
            f'{self.username}:{self.password}'.encode('ascii')
        ).decode('ascii')
        response = self.client.get(
            reverse('api-po-so-calendar', kwargs={'ordertype': 'purchase-order'}),
            format='json',
            headers={'authorization': f'basic {base64_token}'},
        )
        self.assertEqual(response.status_code, 200)


class PurchaseOrderLineItemTest(OrderTest):
    """Unit tests for PurchaseOrderLineItems."""

    LIST_URL = reverse('api-po-line-list')

    def test_po_line_list(self):
        """Test the PurchaseOrderLine list API endpoint."""
        # List *ALL* PurchaseOrderLine items
        self.filter({}, 5)

        # Filter by pending status
        self.filter({'pending': 1}, 5)
        self.filter({'pending': 0}, 0)

        # Filter by received status
        self.filter({'received': 1}, 0)
        self.filter({'received': 0}, 5)

        # Filter by has_pricing status
        self.filter({'has_pricing': 1}, 0)
        self.filter({'has_pricing': 0}, 5)

    def test_po_line_bulk_delete(self):
        """Test that we can bulk delete multiple PurchaseOrderLineItems via the API."""
        n = models.PurchaseOrderLineItem.objects.count()

        self.assignRole('purchase_order.delete')

        url = reverse('api-po-line-list')

        # Try to delete a set of line items via their IDs
        self.delete(url, {'items': [1, 2]}, expected_code=200)

        # We should have 2 less PurchaseOrderLineItems after deletign them
        self.assertEqual(models.PurchaseOrderLineItem.objects.count(), n - 2)

    def test_po_line_merge_pricing(self):
        """Test that we can create a new PurchaseOrderLineItem via the API."""
        self.assignRole('purchase_order.add')
        self.generate_exchange_rates()

        su = Company.objects.get(pk=1)
        sp = SupplierPart.objects.get(pk=1)
        po = models.PurchaseOrder.objects.create(supplier=su, reference='PO-1234567890')
        SupplierPriceBreak.objects.create(part=sp, quantity=1, price=Money(1, 'USD'))
        SupplierPriceBreak.objects.create(part=sp, quantity=10, price=Money(0.5, 'USD'))

        li1 = self.post(
            reverse('api-po-line-list'),
            {
                'order': po.pk,
                'part': sp.pk,
                'quantity': 1,
                'auto_pricing': True,
                'merge_items': False,
            },
            expected_code=201,
        ).json()
        self.assertEqual(float(li1['purchase_price']), 1)

        li2 = self.post(
            reverse('api-po-line-list'),
            {
                'order': po.pk,
                'part': sp.pk,
                'quantity': 10,
                'auto_pricing': True,
                'merge_items': False,
            },
            expected_code=201,
        ).json()
        self.assertEqual(float(li2['purchase_price']), 0.5)

        # test that items where not merged
        self.assertNotEqual(li1['pk'], li2['pk'])

        li3 = self.post(
            reverse('api-po-line-list'),
            {
                'order': po.pk,
                'part': sp.pk,
                'quantity': 9,
                'auto_pricing': True,
                'merge_items': True,
            },
            expected_code=201,
        ).json()

        # test that items where merged
        self.assertEqual(li1['pk'], li3['pk'])

        # test that price was recalculated
        self.assertEqual(float(li3['purchase_price']), 0.5)

        # test that pricing will be not recalculated if auto_pricing is False
        li4 = self.post(
            reverse('api-po-line-list'),
            {
                'order': po.pk,
                'part': sp.pk,
                'quantity': 1,
                'auto_pricing': False,
                'purchase_price': 0.5,
                'merge_items': False,
            },
            expected_code=201,
        ).json()
        self.assertEqual(float(li4['purchase_price']), 0.5)

        # test that pricing is correctly recalculated if auto_pricing is True for update
        li5 = self.patch(
            reverse('api-po-line-detail', kwargs={'pk': li4['pk']}),
            {**li4, 'quantity': 5, 'auto_pricing': False},
            expected_code=200,
        ).json()
        self.assertEqual(float(li5['purchase_price']), 0.5)

        li5 = self.patch(
            reverse('api-po-line-detail', kwargs={'pk': li4['pk']}),
            {**li4, 'quantity': 5, 'auto_pricing': True},
            expected_code=200,
        ).json()
        self.assertEqual(float(li5['purchase_price']), 1)

    def test_output_options(self):
        """Test PurchaseOrderLineItem output option endpoint."""
        self.run_output_test(
            reverse('api-po-line-detail', kwargs={'pk': 1}),
            ['part_detail', 'order_detail'],
        )


class PurchaseOrderDownloadTest(OrderTest):
    """Unit tests for downloading PurchaseOrder data via the API endpoint."""

    required_cols = [
        'ID',
        'Line Items',
        'Description',
        'Issue Date',
        'Order Currency',
        'Reference',
        'Order Status',
        'Supplier Reference',
    ]

    excluded_cols = ['metadata']

    def test_download_wrong_format(self):
        """Incorrect format should default raise an error."""
        url = reverse('api-po-list')

        response = self.export_data(url, export_format='xyz', expected_code=400)
        self.assertIn('is not a valid choice', str(response['export_format']))

    def test_download_csv(self):
        """Download PurchaseOrder data as .csv."""
        with self.export_data(
            reverse('api-po-list'),
            expected_code=200,
            expected_fn=r'InvenTree_PurchaseOrder_.+\.csv',
        ) as file:
            data = self.process_csv(
                file,
                required_cols=self.required_cols,
                excluded_cols=self.excluded_cols,
                required_rows=models.PurchaseOrder.objects.count(),
            )

            for row in data:
                order = models.PurchaseOrder.objects.get(pk=row['ID'])

                self.assertEqual(order.description, row['Description'])
                self.assertEqual(order.reference, row['Reference'])

    def test_download_line_items(self):
        """Test that the PurchaseOrderLineItems can be downloaded to a file."""
        with self.export_data(
            reverse('api-po-line-list'),
            export_format='xlsx',
            expected_code=200,
            expected_fn=r'InvenTree_PurchaseOrderLineItem.+\.xlsx',
            decode=False,
        ) as file:
            self.assertIsInstance(file, io.BytesIO)


class PurchaseOrderReceiveTest(OrderTest):
    """Unit tests for receiving items against a PurchaseOrder."""

    def setUp(self):
        """Init routines for this unit test class."""
        super().setUp()

        self.assignRole('purchase_order.add')

        self.url = reverse('api-po-receive', kwargs={'pk': 1})

        # Number of stock items which exist at the start of each test
        self.n = StockItem.objects.count()

        # Mark the order as "placed" so we can receive line items
        order = models.PurchaseOrder.objects.get(pk=1)
        order.status = PurchaseOrderStatus.PLACED.value
        order.save()

    def test_empty(self):
        """Test without any POST data."""
        data = self.post(self.url, {}, expected_code=400).data

        self.assertIn('This field is required', str(data['items']))

        # No new stock items have been created
        self.assertEqual(self.n, StockItem.objects.count())

    def test_no_items(self):
        """Test with an empty list of items."""
        data = self.post(
            self.url, {'items': [], 'location': None}, expected_code=400
        ).data

        self.assertIn('Line items must be provided', str(data))

        # No new stock items have been created
        self.assertEqual(self.n, StockItem.objects.count())

    def test_invalid_items(self):
        """Test than errors are returned as expected for invalid data."""
        data = self.post(
            self.url,
            {'items': [{'line_item': 12345, 'location': 12345}]},
            expected_code=400,
        ).data

        items = data['items'][0]

        self.assertIn('Invalid pk "12345"', str(items['line_item']))
        self.assertIn('object does not exist', str(items['location']))

        # No new stock items have been created
        self.assertEqual(self.n, StockItem.objects.count())

    def test_invalid_status(self):
        """Test with an invalid StockStatus value."""
        data = self.post(
            self.url,
            {
                'items': [
                    {'line_item': 22, 'location': 1, 'status': 99999, 'quantity': 5}
                ]
            },
            expected_code=400,
        ).data

        self.assertIn('"99999" is not a valid choice.', str(data))

        # No new stock items have been created
        self.assertEqual(self.n, StockItem.objects.count())

    def test_mismatched_items(self):
        """Test for supplier parts which *do* exist but do not match the order supplier."""
        data = self.post(
            self.url,
            {
                'items': [{'line_item': 22, 'quantity': 123, 'location': 1}],
                'location': None,
            },
            expected_code=400,
        ).data

        self.assertIn('Line item does not match purchase order', str(data))

        # No new stock items have been created
        self.assertEqual(self.n, StockItem.objects.count())

    def test_null_barcode(self):
        """Test than a "null" barcode field can be provided."""
        # Set stock item barcode
        item = StockItem.objects.get(pk=1)
        item.save()

        # Test with "null" value
        self.post(
            self.url,
            {
                'items': [{'line_item': 1, 'quantity': 50, 'barcode': None}],
                'location': 1,
            },
            expected_code=201,
        )

    def test_invalid_barcodes(self):
        """Tests for checking in items with invalid barcodes.

        - Cannot check in "duplicate" barcodes
        - Barcodes cannot match 'barcode_hash' field for existing StockItem
        """
        # Set stock item barcode
        item = StockItem.objects.get(pk=1)
        item.assign_barcode(barcode_data='MY-BARCODE-HASH')

        response = self.post(
            self.url,
            {
                'items': [
                    {'line_item': 1, 'quantity': 50, 'barcode': 'MY-BARCODE-HASH'}
                ],
                'location': 1,
            },
            expected_code=400,
        )

        self.assertIn('Barcode is already in use', str(response.data))

        response = self.post(
            self.url,
            {
                'items': [
                    {'line_item': 1, 'quantity': 5, 'barcode': 'MY-BARCODE-HASH-1'},
                    {'line_item': 1, 'quantity': 5, 'barcode': 'MY-BARCODE-HASH-1'},
                ],
                'location': 1,
            },
            expected_code=400,
        )

        self.assertIn('barcode values must be unique', str(response.data))

        # No new stock items have been created
        self.assertEqual(self.n, StockItem.objects.count())

    def test_valid(self):
        """Test receipt of valid data."""
        line_1 = models.PurchaseOrderLineItem.objects.get(pk=1)
        line_2 = models.PurchaseOrderLineItem.objects.get(pk=2)

        self.assertEqual(StockItem.objects.filter(supplier_part=line_1.part).count(), 0)
        self.assertEqual(StockItem.objects.filter(supplier_part=line_2.part).count(), 0)

        self.assertEqual(line_1.received, 0)
        self.assertEqual(line_2.received, 50)

        one_week_from_today = date.today() + timedelta(days=7)

        valid_data = {
            'items': [
                {
                    'line_item': 1,
                    'quantity': 50,
                    'expiry_date': one_week_from_today.strftime(r'%Y-%m-%d'),
                    'barcode': 'MY-UNIQUE-BARCODE-123',
                },
                {
                    'line_item': 2,
                    'quantity': 200,
                    'expiry_date': one_week_from_today.strftime(r'%Y-%m-%d'),
                    'location': 2,  # Explicit location
                    'barcode': 'MY-UNIQUE-BARCODE-456',
                },
            ],
            'location': 1,  # Default location
        }

        # Before posting "valid" data, we will mark the purchase order as "pending"
        # In this case we do expect an error!
        order = models.PurchaseOrder.objects.get(pk=1)
        order.status = PurchaseOrderStatus.PENDING.value
        order.save()

        response = self.post(self.url, valid_data, expected_code=400)

        self.assertIn('can only be received against', str(response.data))

        # Now, set the PurchaseOrder back to "PLACED" so the items can be received
        order.status = PurchaseOrderStatus.PLACED.value
        order.save()

        # Receive two separate line items against this order
        self.post(self.url, valid_data, expected_code=201)

        # There should be two newly created stock items
        self.assertEqual(self.n + 2, StockItem.objects.count())

        line_1 = models.PurchaseOrderLineItem.objects.get(pk=1)
        line_2 = models.PurchaseOrderLineItem.objects.get(pk=2)

        self.assertEqual(line_1.received, 50)
        self.assertEqual(line_2.received, 250)

        stock_1 = StockItem.objects.filter(supplier_part=line_1.part)
        stock_2 = StockItem.objects.filter(supplier_part=line_2.part)

        # 1 new stock item created for each supplier part
        self.assertEqual(stock_1.count(), 1)
        self.assertEqual(stock_2.count(), 1)

        # Check received locations
        self.assertEqual(stock_1.last().location.pk, 1)
        self.assertEqual(stock_2.last().location.pk, 2)

        # Expiry dates should be set
        self.assertEqual(stock_1.last().expiry_date, one_week_from_today)
        self.assertEqual(stock_2.last().expiry_date, one_week_from_today)

        # Barcodes should have been assigned to the stock items
        self.assertTrue(
            StockItem.objects.filter(barcode_data='MY-UNIQUE-BARCODE-123').exists()
        )
        self.assertTrue(
            StockItem.objects.filter(barcode_data='MY-UNIQUE-BARCODE-456').exists()
        )

    def test_batch_code(self):
        """Test that we can supply a 'batch code' when receiving items."""
        line_1 = models.PurchaseOrderLineItem.objects.get(pk=1)
        line_2 = models.PurchaseOrderLineItem.objects.get(pk=2)

        self.assertEqual(StockItem.objects.filter(supplier_part=line_1.part).count(), 0)
        self.assertEqual(StockItem.objects.filter(supplier_part=line_2.part).count(), 0)

        data = {
            'items': [
                {'line_item': 1, 'quantity': 10, 'batch_code': 'B-abc-123'},
                {'line_item': 2, 'quantity': 10, 'batch_code': 'B-xyz-789'},
            ],
            'location': 1,
        }

        n = StockItem.objects.count()

        self.post(self.url, data, expected_code=201)

        # Check that two new stock items have been created!
        self.assertEqual(n + 2, StockItem.objects.count())

        item_1 = StockItem.objects.filter(supplier_part=line_1.part).first()
        item_2 = StockItem.objects.filter(supplier_part=line_2.part).first()

        self.assertEqual(item_1.batch, 'B-abc-123')
        self.assertEqual(item_2.batch, 'B-xyz-789')

    def test_serial_numbers(self):
        """Test that we can supply a 'serial number' when receiving items."""
        line_1 = models.PurchaseOrderLineItem.objects.get(pk=1)
        line_2 = models.PurchaseOrderLineItem.objects.get(pk=2)

        self.assertEqual(StockItem.objects.filter(supplier_part=line_1.part).count(), 0)
        self.assertEqual(StockItem.objects.filter(supplier_part=line_2.part).count(), 0)

        data = {
            'items': [
                {
                    'line_item': 1,
                    'quantity': 10,
                    'batch_code': 'B-abc-123',
                    'serial_numbers': '100+',
                },
                {'line_item': 2, 'quantity': 10, 'batch_code': 'B-xyz-789'},
            ],
            'location': 1,
        }

        n = StockItem.objects.count()

        self.post(self.url, data, expected_code=201, max_query_count=275)

        # Check that the expected number of stock items has been created
        self.assertEqual(n + 11, StockItem.objects.count())

        # 10 serialized stock items created for the first line item
        self.assertEqual(
            StockItem.objects.filter(supplier_part=line_1.part).count(), 10
        )

        # Check that the correct serial numbers have been allocated
        for i in range(100, 110):
            item = StockItem.objects.get(serial_int=i)
            self.assertEqual(item.serial, str(i))
            self.assertEqual(item.quantity, 1)
            self.assertEqual(item.batch, 'B-abc-123')

        # A single stock item (quantity 10) created for the second line item
        items = StockItem.objects.filter(supplier_part=line_2.part)
        self.assertEqual(items.count(), 1)

        item = items.first()

        self.assertEqual(item.quantity, 10)
        self.assertEqual(item.batch, 'B-xyz-789')

    def test_receive_large_quantity(self):
        """Test receipt of a large number of items."""
        from stock.status_codes import StockStatus

        sp = SupplierPart.objects.first()

        # Create a new order
        po = models.PurchaseOrder.objects.create(
            reference='PO-9999', supplier=sp.supplier
        )

        N_LINES = 250

        # Create some line items
        models.PurchaseOrderLineItem.objects.bulk_create([
            models.PurchaseOrderLineItem(order=po, part=sp, quantity=1000 + i)
            for i in range(N_LINES)
        ])

        # Place the order
        po.place_order()

        url = reverse('api-po-receive', kwargs={'pk': po.pk})

        lines = po.lines.all()
        location = StockLocation.objects.filter(structural=False).first()

        N_ITEMS = StockItem.objects.count()

        # Receive all items in a single request
        response = self.post(
            url,
            {
                'items': [
                    {
                        'line_item': line.pk,
                        'quantity': line.quantity,
                        'status': StockStatus.QUARANTINED.value,
                    }
                    for line in lines
                ],
                'location': location.pk,
            },
            max_query_count=104 + 3 * N_LINES,
        ).data

        # Check for expected response
        self.assertEqual(len(response), N_LINES)
        self.assertEqual(N_ITEMS + N_LINES, StockItem.objects.count())

        for item in response:
            self.assertEqual(item['purchase_order'], po.pk)
            self.assertEqual(item['status'], StockStatus.QUARANTINED)

        # Check that the order has been completed
        po.refresh_from_db()
        self.assertEqual(po.status, PurchaseOrderStatus.COMPLETE)

        for line in lines:
            line.refresh_from_db()
            self.assertEqual(line.received, line.quantity)

    def test_packaging(self):
        """Test that we can supply a 'packaging' value when receiving items."""
        line_1 = models.PurchaseOrderLineItem.objects.get(pk=1)
        line_2 = models.PurchaseOrderLineItem.objects.get(pk=2)

        line_1.part.packaging = 'Reel'
        line_1.part.save()

        line_2.part.packaging = 'Tube'
        line_2.part.save()

        # Receive items without packaging data
        data = {
            'items': [
                {'line_item': line_1.pk, 'quantity': 1},
                {'line_item': line_2.pk, 'quantity': 1},
            ],
            'location': 1,
        }

        n = StockItem.objects.count()

        self.post(self.url, data, expected_code=201)

        item_1 = StockItem.objects.filter(supplier_part=line_1.part).first()
        self.assertEqual(item_1.packaging, 'Reel')

        item_2 = StockItem.objects.filter(supplier_part=line_2.part).first()
        self.assertEqual(item_2.packaging, 'Tube')

        # Receive items and override packaging data
        data = {
            'items': [
                {'line_item': line_1.pk, 'quantity': 1, 'packaging': 'Bag'},
                {'line_item': line_2.pk, 'quantity': 1, 'packaging': 'Box'},
            ],
            'location': 1,
        }

        self.post(self.url, data, expected_code=201)

        item_1 = StockItem.objects.filter(supplier_part=line_1.part).last()
        self.assertEqual(item_1.packaging, 'Bag')

        item_2 = StockItem.objects.filter(supplier_part=line_2.part).last()
        self.assertEqual(item_2.packaging, 'Box')

        # Check that the expected number of stock items has been created
        self.assertEqual(n + 4, StockItem.objects.count())

    def test_virtual(self):
        """Test receipt of "virtual" items (i.e. items which do not create a StockItem)."""
        line = models.PurchaseOrderLineItem.objects.get(pk=1)
        base_part = line.part.part
        base_part.virtual = True
        base_part.save()

        self.assertEqual(line.received, 0)

        N_ITEMS = base_part.stock_entries().count()
        N_STOCK = base_part.get_stock_count()

        # Try with serial numbers (expect to fail)
        data = {
            'items': [{'line_item': line.pk, 'quantity': 1, 'serial_numbers': '999'}],
            'location': 1,
        }

        response = self.post(self.url, data, expected_code=400)

        self.assertIn(
            'Serial numbers cannot be assigned to virtual parts',
            str(response.data['non_field_errors']),
        )

        # Try without serial numbers (expect to succeed)
        data = {
            'items': [{'line_item': line.pk, 'quantity': line.quantity}],
            'location': 1,
        }

        self.post(self.url, data, expected_code=201)

        # No new stock items should have been created
        self.assertEqual(base_part.stock_entries().count(), N_ITEMS)
        self.assertEqual(base_part.get_stock_count(), N_STOCK)

        # Check that the line item has been fully received
        line.refresh_from_db()
        self.assertEqual(line.received, line.quantity)


class SalesOrderTest(OrderTest):
    """Tests for the SalesOrder API."""

    LIST_URL = reverse('api-so-list')

    def test_so_list(self):
        """Test the SalesOrder list API endpoint."""
        # All orders
        self.filter({}, 5)

        # Filter by customer
        self.filter({'customer': 4}, 3)
        self.filter({'customer': 5}, 2)

        # Filter by outstanding
        self.filter({'outstanding': True}, 3)
        self.filter({'outstanding': False}, 2)

        # Filter by status
        self.filter({'status': SalesOrderStatus.PENDING.value}, 3)  # PENDING
        self.filter({'status': SalesOrderStatus.SHIPPED.value}, 1)  # SHIPPED
        self.filter({'status': 99}, 0)  # Invalid

        # Filter by "reference"
        self.filter({'reference': 'ABC123'}, 1)
        self.filter({'reference': 'XXX999'}, 0)

        # Filter by "assigned_to_me"
        self.filter({'assigned_to_me': 1}, 0)
        self.filter({'assigned_to_me': 0}, 5)

    def test_total_price(self):
        """Unit tests for the 'total_price' field."""
        # Ensure we have exchange rate data
        self.generate_exchange_rates()

        currencies = currency_codes()
        n = len(currencies)

        idx = 0
        new_orders = []

        # Generate some new SalesOrders
        for customer in Company.objects.filter(is_customer=True):
            for _idx in range(10):
                new_orders.append(
                    models.SalesOrder(customer=customer, reference=f'SO-{idx + 100}')
                )

                idx += 1

        models.SalesOrder.objects.bulk_create(new_orders)

        idx = 0

        # Create some new SalesOrderLineItem objects

        lines = []
        extra_lines = []

        for so in models.SalesOrder.objects.all():
            for p in Part.objects.filter(salable=True):
                lines.append(
                    models.SalesOrderLineItem(
                        order=so,
                        part=p,
                        quantity=idx + 1,
                        sale_price=Money((idx + 1) / 5, currencies[idx % n]),
                    )
                )

                idx += 1

            # Create some extra lines against this order
            for _ in range(3):
                extra_lines.append(
                    models.SalesOrderExtraLine(
                        order=so, quantity=(idx + 2) % 10, price=Money(10, 'CAD')
                    )
                )

        models.SalesOrderLineItem.objects.bulk_create(lines)
        models.SalesOrderExtraLine.objects.bulk_create(extra_lines)

        # List all SalesOrder objects and count queries
        for limit in [1, 5, 10, 100]:
            with CaptureQueriesContext(connection) as ctx:
                response = self.get(
                    self.LIST_URL, data={'limit': limit}, expected_code=200
                )

                # Total database queries must be less than 25
                self.assertLess(len(ctx), 25)

                n = len(response.data['results'])

                for result in response.data['results']:
                    self.assertIn('total_price', result)
                    self.assertIn('order_currency', result)

    def test_overdue(self):
        """Test "overdue" status."""
        self.filter({'overdue': True}, 0)
        self.filter({'overdue': False}, 5)

        for pk in [1, 2]:
            order = models.SalesOrder.objects.get(pk=pk)
            order.target_date = datetime.now().date() - timedelta(days=10)
            order.save()

        self.filter({'overdue': True}, 2)
        self.filter({'overdue': False}, 3)

    def test_so_detail(self):
        """Test the SalesOrder detail endpoint."""
        url = '/api/order/so/1/'

        response = self.get(url)

        data = response.data

        self.assertEqual(data['pk'], 1)

    def test_so_attachments(self):
        """Test the list endpoint for the SalesOrderAttachment model."""
        url = reverse('api-attachment-list')

        # Filter by 'salesorder'
        self.get(
            url, data={'model_type': 'salesorder', 'model_id': 1}, expected_code=200
        )

    def test_so_operations(self):
        """Test that we can create / edit and delete a SalesOrder via the API."""
        n = models.SalesOrder.objects.count()

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

        # Initially we do not have "add" permission for the SalesOrder model,
        # so this POST request should return 403 (denied)
        response = self.post(
            url,
            {'customer': 4, 'reference': '12345', 'description': 'Sales order'},
            expected_code=403,
        )

        self.assignRole('sales_order.add')

        # Now we should be able to create a SalesOrder via the API
        response = self.post(
            url,
            {'customer': 4, 'reference': 'SO-12345', 'description': 'Sales order'},
            expected_code=201,
        )

        # Check that the new order has been created
        self.assertEqual(models.SalesOrder.objects.count(), n + 1)

        # Grab the PK for the newly created SalesOrder
        pk = response.data['pk']

        # Basic checks against the newly created SalesOrder
        so = models.SalesOrder.objects.get(pk=pk)
        self.assertEqual(so.reference, 'SO-12345')
        self.assertEqual(so.created_by.username, 'testuser')

        # Try to create a SO with identical reference (should fail)
        response = self.post(
            url,
            {
                'customer': 4,
                'reference': 'SO-12345',
                'description': 'Another sales order',
            },
            expected_code=400,
        )

        url = reverse('api-so-detail', kwargs={'pk': pk})

        # Extract detail info for the SalesOrder
        response = self.get(url)
        self.assertEqual(response.data['reference'], 'SO-12345')

        # Try to alter (edit) the SalesOrder
        # Initially try with an invalid reference field value
        response = self.patch(url, {'reference': 'SO-12345-a'}, expected_code=400)

        response = self.patch(url, {'reference': 'SO-12346'}, expected_code=200)

        # Reference should have changed
        self.assertEqual(response.data['reference'], 'SO-12346')

        # Now, let's try to delete this SalesOrder
        # Initially, we do not have the required permission
        response = self.delete(url, expected_code=403)

        self.assignRole('sales_order.delete')

        response = self.delete(url, expected_code=204)

        # Check that the number of sales orders has decreased
        self.assertEqual(models.SalesOrder.objects.count(), n)

        # And the resource should no longer be available
        response = self.get(url, expected_code=404)

    def test_so_create(self):
        """Test that we can create a new SalesOrder via the API."""
        self.assignRole('sales_order.add')

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

        # Will fail due to invalid reference field
        response = self.post(
            url,
            {
                'reference': '1234566778',
                'customer': 4,
                'description': 'A test sales order',
            },
            expected_code=400,
        )

        self.assertIn(
            'Reference must match required pattern', str(response.data['reference'])
        )

        self.post(
            url,
            {
                'reference': 'SO-12345',
                'customer': 4,
                'description': 'A better test sales order',
            },
            expected_code=201,
        )

    def test_so_duplicate(self):
        """Test SalesOrder duplication via the API."""
        from common.models import Parameter, ParameterTemplate

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

        self.assignRole('sales_order.add')

        so = models.SalesOrder.objects.get(pk=1)
        self.assertEqual(so.status, SalesOrderStatus.PENDING)

        # Add some parameters to the sales order
        for idx in range(5):
            template = ParameterTemplate.objects.create(name=f'Template {idx}')

            Parameter.objects.create(
                template=template,
                model_type=so.get_content_type(),
                model_id=so.pk,
                data=f'Value {idx}',
            )

        self.assertEqual(so.parameters.count(), 5)

        # Create a duplicate of this sales order
        # We explicitly specify "copy_parameters" as False, so the duplicated sales order should not have any parameters
        response = self.post(
            url,
            {
                'reference': 'SO-12345',
                'customer': so.customer.pk,
                'duplicate': {'order_id': so.pk, 'copy_parameters': False},
            },
        )

        duplicate_id = response.data['pk']
        duplicate_so = models.SalesOrder.objects.get(pk=duplicate_id)

        self.assertEqual(duplicate_so.reference, 'SO-12345')
        self.assertEqual(duplicate_so.customer, so.customer)
        self.assertEqual(duplicate_so.parameters.count(), 0)

        # Duplicate again, with default values for the "duplicate" options (which should result in parameters being copied)
        response = self.post(
            url,
            {
                'reference': 'SO-12346',
                'customer': so.customer.pk,
                'duplicate': {'order_id': so.pk},
            },
        )

        duplicate_id = response.data['pk']
        duplicate_so = models.SalesOrder.objects.get(pk=duplicate_id)

        self.assertEqual(duplicate_so.reference, 'SO-12346')
        self.assertEqual(duplicate_so.customer, so.customer)
        self.assertEqual(duplicate_so.parameters.count(), 5)

    def test_so_cancel(self):
        """Test API endpoint for cancelling a SalesOrder."""
        so = models.SalesOrder.objects.get(pk=1)

        self.assertEqual(so.status, SalesOrderStatus.PENDING)

        url = reverse('api-so-cancel', kwargs={'pk': so.pk})

        # Try to cancel, without permission
        self.post(url, {}, expected_code=403)

        self.assignRole('sales_order.add')

        self.post(url, {}, expected_code=201)

        so.refresh_from_db()

        self.assertEqual(so.status, SalesOrderStatus.CANCELLED)

    def test_so_calendar(self):
        """Test the calendar export endpoint."""
        # Create required sales orders
        self.assignRole('sales_order.add')

        for i in range(1, 9):
            self.post(
                reverse('api-so-list'),
                {
                    'reference': f'SO-1100000{i}',
                    'customer': 4,
                    'description': f'Calendar SO {i}',
                    'target_date': f'2024-12-{i:02d}',
                },
                expected_code=201,
            )

        # Cancel a few orders - these will not show in incomplete view below
        for so in models.SalesOrder.objects.filter(target_date__isnull=False):
            if so.reference in [
                'SO-11000006',
                'SO-11000007',
                'SO-11000008',
                'SO-11000009',
            ]:
                self.post(
                    reverse('api-so-cancel', kwargs={'pk': so.pk}), expected_code=201
                )

        url = reverse('api-po-so-calendar', kwargs={'ordertype': 'sales-order'})

        # Test without completed orders
        response = self.get(url, expected_code=200, format=None)

        number_orders = len(
            models.SalesOrder.objects.filter(target_date__isnull=False).filter(
                status__lt=SalesOrderStatus.SHIPPED.value
            )
        )

        # Transform content to a Calendar object
        calendar = Calendar.from_ical(response.content)
        n_events = 0
        # Count number of events in calendar
        for component in calendar.walk():
            if component.name == 'VEVENT':
                n_events += 1

        self.assertGreaterEqual(n_events, 1)
        self.assertEqual(number_orders, n_events)

        # Test with completed orders
        response = self.get(
            url, data={'include_completed': 'True'}, expected_code=200, format=None
        )

        number_orders_incl_complete = len(
            models.SalesOrder.objects.filter(target_date__isnull=False)
        )
        self.assertGreater(number_orders_incl_complete, number_orders)

        # Transform content to a Calendar object
        calendar = Calendar.from_ical(response.content)
        n_events = 0
        # Count number of events in calendar
        for component in calendar.walk():
            if component.name == 'VEVENT':
                n_events += 1

        self.assertGreaterEqual(n_events, 1)
        self.assertEqual(number_orders_incl_complete, n_events)

    def test_export(self):
        """Test we can export the SalesOrder list."""
        set_global_setting(models.SalesOrder.UNLOCK_SETTING, True)

        n = models.SalesOrder.objects.count()

        # Check there are some sales orders
        self.assertGreater(n, 0)

        for order in models.SalesOrder.objects.all():
            # Reconstruct the total price
            order.save()

        # Download file, check we get a 200 response
        for fmt in ['csv', 'xlsx', 'tsv']:
            self.export_data(
                reverse('api-so-list'),
                export_format=fmt,
                decode=fmt == 'csv',
                expected_code=200,
                expected_fn=r'InvenTree_SalesOrder_.+',
            )

    def test_sales_order_complete(self):
        """Tests for marking a SalesOrder as complete."""
        self.assignRole('sales_order.add')

        # Let's create a SalesOrder
        customer = Company.objects.filter(is_customer=True).first()
        so = models.SalesOrder.objects.create(
            customer=customer, reference='SO-12345', description='Test SO'
        )

        self.assertEqual(so.status, SalesOrderStatus.PENDING.value)

        # Create a line item
        part = Part.objects.filter(salable=True).first()

        line = models.SalesOrderLineItem.objects.create(
            order=so, part=part, quantity=10, sale_price=Money(10, 'USD')
        )

        shipment = so.shipments.first()

        if shipment is None:
            shipment = models.SalesOrderShipment.objects.create(
                order=so, reference='SHIP-12345'
            )
        assert shipment

        # Allocate some stock
        item = StockItem.objects.create(part=part, quantity=100, location=None)
        models.SalesOrderAllocation.objects.create(
            quantity=10, line=line, item=item, shipment=shipment
        )

        # Ship the shipment
        shipment.complete_shipment(self.user)

        # Ok, now we should be able to "complete" the shipment via the API
        # The 'SALESORDER_SHIP_COMPLETE' setting determines if the outcome is "SHIPPED" or "COMPLETE"
        InvenTreeSetting.set_setting('SALESORDER_SHIP_COMPLETE', False)

        url = reverse('api-so-complete', kwargs={'pk': so.pk})
        self.post(url, {}, expected_code=201)

        so.refresh_from_db()
        self.assertEqual(so.status, SalesOrderStatus.SHIPPED.value)
        self.assertIsNotNone(so.shipment_date)
        self.assertIsNotNone(so.shipped_by)

        # Now, let's try to "complete" the shipment again
        # This time it should get marked as "COMPLETE"
        self.post(url, {}, expected_code=201)

        so.refresh_from_db()
        self.assertEqual(so.status, SalesOrderStatus.COMPLETE.value)

        # Now, let's try *again* (it should fail as the order is already complete)
        response = self.post(url, {}, expected_code=400)

        self.assertIn('Order is already complete', str(response.data))

        # Next, we'll change the setting so that the order status jumps straight to "complete"
        so.status = SalesOrderStatus.PENDING.value
        so.shipment_date = None
        so.shipped_by = None
        so.save()
        so.refresh_from_db()

        self.assertEqual(so.status, SalesOrderStatus.PENDING.value)
        self.assertIsNone(so.shipped_by)
        self.assertIsNone(so.shipment_date)

        InvenTreeSetting.set_setting('SALESORDER_SHIP_COMPLETE', True)

        self.post(url, {}, expected_code=201)

        # The orders status should now be "complete" (not "shipped")
        so.refresh_from_db()
        self.assertEqual(so.status, SalesOrderStatus.COMPLETE.value)

        self.assertIsNotNone(so.shipment_date)
        self.assertIsNotNone(so.shipped_by)

    def test_output_options(self):
        """Test the output options for the SalesOrder detail endpoint."""
        self.run_output_test(
            reverse('api-so-detail', kwargs={'pk': 1}), ['customer_detail']
        )


class SalesOrderLineItemTest(OrderTest):
    """Tests for the SalesOrderLineItem API."""

    LIST_URL = reverse('api-so-line-list')

    @classmethod
    def setUpTestData(cls):
        """Init routine for this unit test class."""
        super().setUpTestData()

        # List of salable parts
        parts = Part.objects.filter(salable=True)

        lines = []

        # Create a bunch of SalesOrderLineItems for each order
        for idx, so in enumerate(models.SalesOrder.objects.all()):
            for part in parts:
                lines.append(
                    models.SalesOrderLineItem(
                        order=so,
                        part=part,
                        quantity=(idx + 1) * 5,
                        reference=f'Order {so.reference} - line {idx}',
                    )
                )

        # Bulk create
        models.SalesOrderLineItem.objects.bulk_create(lines)

        cls.url = reverse('api-so-line-list')

    def test_so_line_list(self):
        """Test list endpoint."""
        response = self.get(self.url, {}, expected_code=200)

        n = models.SalesOrderLineItem.objects.count()

        # We should have received *all* lines
        self.assertEqual(len(response.data), n)

        # List *all* lines, but paginate
        response = self.get(self.url, {'limit': 5}, expected_code=200)

        self.assertEqual(response.data['count'], n)
        self.assertEqual(len(response.data['results']), 5)

        n_orders = models.SalesOrder.objects.count()
        n_parts = Part.objects.filter(salable=True).count()

        # List by part
        for part in Part.objects.filter(salable=True)[:3]:
            response = self.get(self.url, {'part': part.pk, 'limit': 10})

            self.assertEqual(response.data['count'], n_orders)

        # List by order
        for order in models.SalesOrder.objects.all()[:3]:
            response = self.get(self.url, {'order': order.pk, 'limit': 10})

            self.assertEqual(response.data['count'], n_parts)

        # Filter by has_pricing status
        self.filter({'has_pricing': 1}, 0)
        self.filter({'has_pricing': 0}, n)

        # Filter by 'completed' status
        self.filter({'completed': 1}, 0)
        self.filter({'completed': 0}, n)

        # Filter by 'allocated' status
        self.filter({'allocated': 'true'}, 1)
        self.filter({'allocated': 'false'}, n - 1)

    def test_so_line_allocated_filters(self):
        """Test filtering by allocation status for a SalesOrderLineItem."""
        self.assignRole('sales_order.add')

        # Crete a new SalesOrder via the API
        company = Company.objects.filter(is_customer=True).first()
        assert company

        response = self.post(
            reverse('api-so-list'),
            {
                'customer': company.pk,
                'reference': 'SO-12345',
                'description': 'Test Sales Order',
            },
        )

        order_id = response.data['pk']
        order = models.SalesOrder.objects.get(pk=order_id)

        so_line_url = reverse('api-so-line-list')

        # Initially, there should be no line items against this order
        response = self.get(so_line_url, {'order': order_id})

        self.assertEqual(len(response.data), 0)

        parts = [25, 50, 100]

        # Let's create some new line items
        for part_id in parts:
            self.post(so_line_url, {'order': order_id, 'part': part_id, 'quantity': 10})

        # Should be three items now
        response = self.get(so_line_url, {'order': order_id})

        self.assertEqual(len(response.data), 3)

        for item in response.data:
            # Check that the line item has been created
            self.assertEqual(item['order'], order_id)

            # Check that the line quantities are correct
            self.assertEqual(item['quantity'], 10)
            self.assertEqual(item['allocated'], 0)
            self.assertEqual(item['shipped'], 0)

        # Initial API filters should return no results
        self.filter({'order': order_id, 'allocated': 1}, 0)
        self.filter({'order': order_id, 'completed': 1}, 0)

        # Create a new shipment against this SalesOrder
        shipment = models.SalesOrderShipment.objects.create(
            order=order, reference='SHIP-12345'
        )

        # Next, allocate stock against 2 line items
        for item in parts[:2]:
            p = Part.objects.get(pk=item)
            s = StockItem.objects.create(part=p, quantity=100)
            l = models.SalesOrderLineItem.objects.filter(order=order, part=p).first()
            assert l

            # Allocate against the API
            self.post(
                reverse('api-so-allocate', kwargs={'pk': order.pk}),
                {
                    'items': [{'line_item': l.pk, 'stock_item': s.pk, 'quantity': 10}],
                    'shipment': shipment.pk,
                },
            )

        # Filter by 'fully allocated' status
        self.filter({'order': order_id, 'allocated': 1}, 2)
        self.filter({'order': order_id, 'allocated': 0}, 1)

        self.filter({'order': order_id, 'completed': 1}, 0)
        self.filter({'order': order_id, 'completed': 0}, 3)

        # Finally, mark this shipment as 'shipped'
        self.post(
            reverse('api-so-shipment-ship', kwargs={'pk': shipment.pk}),
            {},
            expected_code=200,
        )

        # Filter by 'completed' status
        self.filter({'order': order_id, 'completed': 1}, 2)
        self.filter({'order': order_id, 'completed': 0}, 1)

    def test_output_options(self):
        """Test the various output options for the SalesOrderLineItem detail endpoint."""
        self.run_output_test(
            reverse('api-so-line-detail', kwargs={'pk': 1}),
            ['part_detail', 'order_detail', 'customer_detail'],
        )


class SalesOrderDownloadTest(OrderTest):
    """Unit tests for downloading SalesOrder data via the API endpoint."""

    def test_download_fail(self):
        """Test that downloading without the 'export' option fails."""
        url = reverse('api-so-list')

        response = self.export_data(url, export_plugin='no-plugin', expected_code=400)
        self.assertIn('is not a valid choice', str(response['export_plugin']))

    def test_download_xlsx(self):
        """Test xlsx file download."""
        url = reverse('api-so-list')

        # Download .xls file
        with self.export_data(
            url, export_format='xlsx', expected_code=200, decode=False
        ) as file:
            self.assertIsInstance(file, io.BytesIO)

    def test_download_csv(self):
        """Test that the list of sales orders can be downloaded as a .csv file."""
        url = reverse('api-so-list')

        required_cols = [
            'Line Items',
            'ID',
            'Reference',
            'Customer',
            'Order Status',
            'Shipment Date',
            'Description',
            'Project Code',
            'Responsible',
        ]

        excluded_cols = ['metadata']

        # Download .xls file
        with self.export_data(url, export_format='csv') as file:
            data = self.process_csv(
                file,
                required_cols=required_cols,
                excluded_cols=excluded_cols,
                required_rows=models.SalesOrder.objects.count(),
            )

            for line in data:
                order = models.SalesOrder.objects.get(pk=line['ID'])

                self.assertEqual(line['Description'], order.description)
                self.assertEqual(line['Order Status'], str(order.status))

        # Download only outstanding sales orders
        with self.export_data(url, {'outstanding': True}, export_format='tsv') as file:
            self.process_csv(
                file,
                required_cols=required_cols,
                excluded_cols=excluded_cols,
                required_rows=models.SalesOrder.objects.filter(
                    status__in=SalesOrderStatusGroups.OPEN
                ).count(),
                delimiter='\t',
            )


class SalesOrderAllocateTest(OrderTest):
    """Unit tests for allocating stock items against a SalesOrder."""

    @classmethod
    def setUpTestData(cls):
        """Init routine for this unit test class."""
        super().setUpTestData()

    def setUp(self):
        """Init routines for this unit testing class."""
        super().setUp()

        self.assignRole('sales_order.add')

        self.url = reverse('api-so-allocate', kwargs={'pk': 1})

        self.order = models.SalesOrder.objects.get(pk=1)

        # Create some line items for this purchase order
        parts = Part.objects.filter(salable=True)

        for part in parts:
            # Create a new line item
            models.SalesOrderLineItem.objects.create(
                order=self.order, part=part, quantity=5
            )

            # Ensure we have stock!
            StockItem.objects.create(part=part, quantity=100)

        # Create a new shipment against this SalesOrder
        self.shipment = models.SalesOrderShipment.objects.create(order=self.order)

    def test_invalid(self):
        """Test POST with invalid data."""
        # No data
        response = self.post(self.url, {}, expected_code=400)

        self.assertIn('This field is required', str(response.data['items']))

        # Test with a single line items
        line = self.order.lines.first()
        part = line.part

        # Valid stock_item, but quantity is invalid
        data = {
            'items': [
                {
                    'line_item': line.pk,
                    'stock_item': part.stock_items.last().pk,
                    'quantity': 0,
                }
            ]
        }

        response = self.post(self.url, data, expected_code=400)

        self.assertIn('Quantity must be positive', str(response.data['items']))

        # Valid stock item, too much quantity
        data['items'][0]['quantity'] = 250

        response = self.post(self.url, data, expected_code=400)

        self.assertIn('Available quantity (100) exceeded', str(response.data['items']))

        # Valid stock item, valid quantity
        data['items'][0]['quantity'] = 50

        # Invalid shipment!
        data['shipment'] = 9999

        response = self.post(self.url, data, expected_code=400)

        self.assertIn('does not exist', str(response.data['shipment']))

        # Valid shipment, but points to the wrong order
        shipment = models.SalesOrderShipment.objects.create(
            order=models.SalesOrder.objects.get(pk=2)
        )

        data['shipment'] = shipment.pk

        response = self.post(self.url, data, expected_code=400)

        self.assertIn(
            'Shipment is not associated with this order', str(response.data['shipment'])
        )

    def test_allocate(self):
        """Test that the allocation endpoint acts as expected, when provided with valid data!"""
        # First, check that there are no line items allocated against this SalesOrder
        self.assertEqual(self.order.stock_allocations.count(), 0)

        data = {'items': [], 'shipment': self.shipment.pk}

        for line in self.order.lines.all():
            for stock_item in line.part.stock_items.all():
                # Find a non-serialized stock item to allocate
                if not stock_item.serialized:
                    break

            # Fully-allocate each line
            data['items'].append({
                'line_item': line.pk,
                'stock_item': stock_item.pk,
                'quantity': 5,
            })

        self.post(self.url, data, expected_code=201)

        # There should have been 1 stock item allocated against each line item
        n_lines = self.order.lines.count()

        self.assertEqual(self.order.stock_allocations.count(), n_lines)

        for line in self.order.lines.all():
            self.assertEqual(line.allocations.count(), 1)

    def test_allocate_variant(self):
        """Test that the allocation endpoint acts as expected, when provided with variant."""
        # First, check that there are no line items allocated against this SalesOrder
        self.assertEqual(self.order.stock_allocations.count(), 0)

        data = {'items': [], 'shipment': self.shipment.pk}

        def check_template(line_item):
            return line_item.part.is_template

        for line in filter(check_template, self.order.lines.all()):
            stock_item: Optional[StockItem] = None

            stock_item = None

            # Allocate a matching variant
            parts: list[Part] = Part.objects.filter(salable=True).filter(
                variant_of=line.part.pk
            )
            for part in parts:
                stock_item = part.stock_items.last()

                for item in part.stock_items.all():
                    if item.serialized:
                        continue

                    stock_item = item
                    break

                if stock_item is not None:
                    break

            if stock_item is None:
                raise self.fail('No stock item found for part')  # pragma: no cover

            # Fully-allocate each line
            data['items'].append({
                'line_item': line.pk,
                'stock_item': stock_item.pk,
                'quantity': 5,
            })

        self.post(self.url, data, expected_code=201)

        # At least one item should be allocated, and all should be variants
        self.assertGreater(self.order.stock_allocations.count(), 0)
        for allocation in self.order.stock_allocations.all():
            self.assertNotEqual(allocation.item.part.pk, allocation.line.part.pk)

    def test_shipment_complete(self):
        """Test that we can complete a shipment via the API."""
        url = reverse('api-so-shipment-ship', kwargs={'pk': self.shipment.pk})

        self.assertFalse(self.shipment.is_complete())
        self.assertFalse(self.shipment.check_can_complete(raise_error=False))

        with self.assertRaises(ValidationError):
            self.shipment.check_can_complete()

        # Attempting to complete this shipment via the API should fail
        response = self.post(url, {}, expected_code=400)

        self.assertIn('Shipment has no allocated stock items', str(response.data))

        # Allocate stock against this shipment
        line = self.order.lines.first()
        part = line.part

        models.SalesOrderAllocation.objects.create(
            shipment=self.shipment, line=line, item=part.stock_items.last(), quantity=5
        )

        # Shipment should now be able to be completed
        self.assertTrue(self.shipment.check_can_complete())

        # Attempt with an invalid date
        response = self.post(url, {'shipment_date': 'asfasd'}, expected_code=400)

        self.assertIn('Date has wrong format', str(response.data))

        response = self.post(
            url,
            {
                'invoice_number': 'INV01234',
                'link': 'http://test.com/link.html',
                'tracking_number': 'TRK12345',
                'shipment_date': '2020-12-05',
                'delivery_date': '2023-12-05',
            },
            expected_code=200,
        )

        self.shipment.refresh_from_db()

        self.assertTrue(self.shipment.is_complete())
        self.assertEqual(self.shipment.tracking_number, 'TRK12345')
        self.assertEqual(self.shipment.invoice_number, 'INV01234')
        self.assertEqual(self.shipment.link, 'http://test.com/link.html')
        self.assertEqual(self.shipment.delivery_date, datetime(2023, 12, 5).date())
        self.assertTrue(self.shipment.is_delivered())

    def test_shipment_delivery_date(self):
        """Test delivery date functions via API."""
        url = reverse('api-so-shipment-detail', kwargs={'pk': self.shipment.pk})

        # Attempt remove delivery_date from shipment
        response = self.patch(url, {'delivery_date': None}, expected_code=200)

        # Shipment should not be marked as delivered
        self.assertFalse(self.shipment.is_delivered())

        # Attempt to set delivery date
        response = self.patch(url, {'delivery_date': 'asfasd'}, expected_code=400)

        self.assertIn('Date has wrong format', str(response.data))

        response = self.patch(url, {'delivery_date': '2023-05-15'}, expected_code=200)
        self.shipment.refresh_from_db()

        # Shipment should now be marked as delivered
        self.assertTrue(self.shipment.is_delivered())
        self.assertEqual(self.shipment.delivery_date, datetime(2023, 5, 15).date())

    def test_sales_order_shipment_list(self):
        """Test the SalesOrderShipment list API endpoint."""
        url = reverse('api-so-shipment-list')

        # Count before creation
        count_before = models.SalesOrderShipment.objects.count()

        # Create some new shipments via the API
        for order in models.SalesOrder.objects.all():
            for idx in range(3):
                self.post(
                    url,
                    {
                        'order': order.pk,
                        'reference': f'SH{idx + 1}',
                        'tracking_number': f'TRK_{order.pk}_{idx}',
                    },
                    expected_code=201,
                )

            # Filter API by order
            response = self.get(url, {'order': order.pk}, expected_code=200)

            # 3 shipments returned for each SalesOrder instance
            self.assertGreaterEqual(len(response.data), 3)

        # List *all* shipments
        response = self.get(url, expected_code=200)

        self.assertEqual(
            len(response.data), count_before + 3 * models.SalesOrder.objects.count()
        )

    def test_output_options(self):
        """Test the various output options for the SalesOrderAllocation detail endpoint."""
        self.run_output_test(
            reverse('api-so-allocation-list'),
            [
                'part_detail',
                'item_detail',
                'order_detail',
                'location_detail',
                'customer_detail',
            ],
            assert_subset=True,
        )


class ReturnOrderTests(InvenTreeAPITestCase):
    """Unit tests for ReturnOrder API endpoints."""

    fixtures = [
        'category',
        'company',
        'return_order',
        'part',
        'location',
        'supplier_part',
        'stock',
    ]
    roles = ['return_order.view']

    def test_options(self):
        """Test the OPTIONS endpoint."""
        self.assignRole('return_order.add')
        data = self.options(reverse('api-return-order-list'), expected_code=200).data

        self.assertEqual(data['name'], 'Return Order List')

        # Some checks on the 'reference' field
        post = data['actions']['POST']
        reference = post['reference']

        self.assertEqual(reference['default'], 'RMA-0007')
        self.assertEqual(reference['label'], 'Reference')
        self.assertEqual(reference['help_text'], 'Return Order reference')
        self.assertEqual(reference['required'], True)
        self.assertEqual(reference['type'], 'string')

    def test_project_code(self):
        """Test the 'project_code' serializer field."""
        self.assignRole('return_order.add')
        response = self.options(reverse('api-return-order-list'), expected_code=200)
        project_code = response.data['actions']['POST']['project_code']

        self.assertFalse(project_code['required'])
        self.assertFalse(project_code['read_only'])
        self.assertEqual(project_code['type'], 'related field')
        self.assertEqual(project_code['label'], 'Project Code')
        self.assertEqual(project_code['model'], 'projectcode')

    def test_list(self):
        """Tests for the list endpoint."""
        url = reverse('api-return-order-list')

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

        self.assertEqual(len(response.data), 6)

        # Paginated query
        data = self.get(
            url,
            {'limit': 1, 'ordering': 'reference', 'customer_detail': True},
            expected_code=200,
        ).data

        self.assertEqual(data['count'], 6)
        self.assertEqual(len(data['results']), 1)
        result = data['results'][0]
        self.assertEqual(result['reference'], 'RMA-001')
        self.assertEqual(result['customer_detail']['name'], 'A customer')

        # Reverse ordering
        data = self.get(url, {'ordering': '-reference'}, expected_code=200).data

        self.assertEqual(data[0]['reference'], 'RMA-006')

        # Filter by customer
        for cmp_id in [4, 5]:
            data = self.get(url, {'customer': cmp_id}, expected_code=200).data

            self.assertEqual(len(data), 3)

            for result in data:
                self.assertEqual(result['customer'], cmp_id)

        # Filter by status
        data = self.get(
            url, {'status': ReturnOrderStatus.IN_PROGRESS.value}, expected_code=200
        ).data

        self.assertEqual(len(data), 2)

        for result in data:
            self.assertEqual(result['status'], 20)

    def test_create(self):
        """Test creation of ReturnOrder via the API."""
        url = reverse('api-return-order-list')

        # Do not have required permissions yet
        self.post(
            url, {'customer': 1, 'description': 'a return order'}, expected_code=403
        )

        self.assignRole('return_order.add')

        data = self.post(
            url,
            {
                'customer': 4,
                'customer_reference': 'cr',
                'description': 'a return order',
            },
            expected_code=201,
        ).data

        # Reference automatically generated
        self.assertEqual(data['reference'], 'RMA-0007')
        self.assertEqual(data['customer_reference'], 'cr')

    def test_update(self):
        """Test that we can update a ReturnOrder via the API."""
        url = reverse('api-return-order-detail', kwargs={'pk': 1})

        # Test detail endpoint
        data = self.get(url, expected_code=200).data

        self.assertEqual(data['reference'], 'RMA-001')

        # Attempt to update, incorrect permissions
        self.patch(
            url, {'customer_reference': 'My customer reference'}, expected_code=403
        )

        self.assignRole('return_order.change')

        self.patch(url, {'customer_reference': 'customer ref'}, expected_code=200)

        rma = models.ReturnOrder.objects.get(pk=1)
        self.assertEqual(rma.customer_reference, 'customer ref')

    def test_ro_issue(self):
        """Test the 'issue' order for a ReturnOrder."""
        order = models.ReturnOrder.objects.get(pk=1)
        self.assertEqual(order.status, ReturnOrderStatus.PENDING)
        self.assertIsNone(order.issue_date)

        url = reverse('api-return-order-issue', kwargs={'pk': 1})

        # POST without required permissions
        self.post(url, expected_code=403)

        self.assignRole('return_order.add')

        self.post(url, expected_code=201)
        order.refresh_from_db()
        self.assertEqual(order.status, ReturnOrderStatus.IN_PROGRESS)
        self.assertIsNotNone(order.issue_date)

    def test_receive(self):
        """Test that we can receive items against a ReturnOrder."""
        customer = Company.objects.get(pk=4)

        # Create an order
        rma = models.ReturnOrder.objects.create(
            customer=customer, description='A return order'
        )

        self.assertEqual(rma.reference, 'RMA-0007')

        # Create some line items
        part = Part.objects.get(pk=25)
        for idx in range(3):
            stock_item = StockItem.objects.create(
                part=part, customer=customer, quantity=1, serial=idx
            )

            line_item = models.ReturnOrderLineItem.objects.create(
                order=rma, item=stock_item
            )

            self.assertEqual(line_item.outcome, ReturnOrderLineStatus.PENDING)
            self.assertIsNone(line_item.received_date)
            self.assertFalse(line_item.received)

        self.assertEqual(rma.lines.count(), 3)

        def receive(items, location=None, expected_code=400):
            """Helper function to receive items against this ReturnOrder."""
            url = reverse('api-return-order-receive', kwargs={'pk': rma.pk})

            response = self.post(
                url, {'items': items, 'location': location}, expected_code=expected_code
            )

            return response.data

        # Receive without required permissions
        receive([], expected_code=403)

        self.assignRole('return_order.add')

        # Receive, without any location
        data = receive([], expected_code=400)
        self.assertIn('This field may not be null', str(data['location']))

        # Receive, with incorrect order code
        data = receive([], 1, expected_code=400)
        self.assertIn(
            'Items can only be received against orders which are in progress', str(data)
        )

        # Issue the order (via the API)
        self.assertIsNone(rma.issue_date)
        self.post(
            reverse('api-return-order-issue', kwargs={'pk': rma.pk}), expected_code=201
        )

        rma.refresh_from_db()
        self.assertIsNotNone(rma.issue_date)
        self.assertEqual(rma.status, ReturnOrderStatus.IN_PROGRESS)

        # Receive, without any items
        data = receive([], 1, expected_code=400)
        self.assertIn('Line items must be provided', str(data))

        # Get a reference to one of the stock items
        stock_item = rma.lines.first().item

        n_tracking = stock_item.tracking_info.count()

        # Receive items successfully
        data = receive(
            [{'item': line.pk} for line in rma.lines.all()], 1, expected_code=201
        )

        # Check that all line items have been received
        for line in rma.lines.all():
            self.assertTrue(line.received)
            self.assertIsNotNone(line.received_date)

        # A single tracking entry should have been added to the item
        self.assertEqual(stock_item.tracking_info.count(), n_tracking + 1)

        tracking_entry = stock_item.tracking_info.last()
        deltas = tracking_entry.deltas

        self.assertEqual(deltas['status'], StockStatus.QUARANTINED)
        self.assertEqual(deltas['customer'], customer.pk)
        self.assertEqual(deltas['location'], 1)
        self.assertEqual(deltas['returnorder'], rma.pk)

    def test_receive_untracked(self):
        """Test that we can receive untracked items against a ReturnOrder.

        Ref: https://github.com/inventree/InvenTree/pull/8590
        """
        self.assignRole('return_order.add')
        company = Company.objects.get(pk=4)

        # Create a new ReturnOrder
        rma = models.ReturnOrder.objects.create(
            customer=company, description='A return order'
        )

        rma.issue_order()

        # Create some new line items
        part = Part.objects.get(pk=25)

        n_items = part.stock_entries().count()

        for idx in range(2):
            stock_item = StockItem.objects.create(
                part=part, customer=company, quantity=10
            )

            models.ReturnOrderLineItem.objects.create(
                order=rma, item=stock_item, quantity=(idx + 1) * 5
            )

        self.assertEqual(part.stock_entries().count(), n_items + 2)

        line_items = rma.lines.all()

        # Receive items against the order
        url = reverse('api-return-order-receive', kwargs={'pk': rma.pk})

        LOCATION_ID = 1

        self.post(
            url,
            {
                'items': [
                    {'item': line.pk, 'status': StockStatus.DAMAGED.value}
                    for line in line_items
                ],
                'location': LOCATION_ID,
            },
            expected_code=201,
        )

        # Due to the quantities received, we should have created 1 new stock item
        self.assertEqual(part.stock_entries().count(), n_items + 3)

        rma.refresh_from_db()

        for line in rma.lines.all():
            self.assertTrue(line.received)
            self.assertIsNotNone(line.received_date)

            # Check that the associated StockItem has been updated correctly
            self.assertEqual(line.item.status, StockStatus.DAMAGED)
            self.assertIsNone(line.item.customer)
            self.assertIsNone(line.item.sales_order)
            self.assertEqual(line.item.location.pk, LOCATION_ID)

    def test_ro_calendar(self):
        """Test the calendar export endpoint."""
        # Full test is in test_po_calendar. Since these use the same backend, test only
        # that the endpoint is available
        url = reverse('api-po-so-calendar', kwargs={'ordertype': 'return-order'})

        # Test without completed orders
        response = self.get(url, expected_code=200, format=None)
        calendar = Calendar.from_ical(response.content)
        self.assertIsInstance(calendar, Calendar)

    def test_export(self):
        """Test data export for the ReturnOrder API endpoints."""
        # Export return orders
        data = self.export_data(
            reverse('api-return-order-list'),
            export_format='csv',
            decode=True,
            expected_code=200,
        )

        self.process_csv(
            data,
            required_cols=['Reference', 'Customer'],
            required_rows=models.ReturnOrder.objects.count(),
        )

        N = models.ReturnOrderLineItem.objects.count()
        self.assertGreater(N, 0, 'No ReturnOrderLineItems found!')

        # Export return order lines
        data = self.export_data(
            reverse('api-return-order-line-list'),
            export_format='csv',
            decode=True,
            expected_code=200,
        )

        self.process_csv(
            data, required_rows=N, required_cols=['Order', 'Reference', 'Target Date']
        )

        # Export again, with a search term
        data = self.export_data(
            reverse('api-return-order-line-list'),
            params={'search': 'xyz'},
            export_format='csv',
            decode=True,
            expected_code=200,
        )

        self.process_csv(
            data, required_rows=0, required_cols=['Order', 'Reference', 'Target Date']
        )

    def test_output_options(self):
        """Test the various output options for the ReturnOrder detail endpoint."""
        self.run_output_test(
            reverse('api-return-order-detail', kwargs={'pk': 1}), ['customer_detail']
        )


class ReturnOrderLineItemTests(InvenTreeAPITestCase):
    """Unit tests for ReturnOrderLineItem API endpoints."""

    fixtures = [
        'category',
        'company',
        'return_order',
        'part',
        'location',
        'supplier_part',
        'stock',
    ]
    roles = ['return_order.view']

    def test_options(self):
        """Test the OPTIONS endpoint."""
        self.assignRole('return_order.add')
        data = self.options(
            reverse('api-return-order-line-list'), expected_code=200
        ).data

        self.assertEqual(data['name'], 'Return Order Line Item List')

        # Check POST fields
        post = data['actions']['POST']
        self.assertIn('order', post)
        self.assertIn('item', post)
        self.assertIn('quantity', post)
        self.assertIn('outcome', post)

    def test_list(self):
        """Test list endpoint."""
        url = reverse('api-return-order-line-list')

        response = self.get(url, expected_code=200)
        self.assertGreater(len(response.data), 0)

        # Test with pagination
        data = self.get(
            url, {'limit': 1, 'ordering': 'reference'}, expected_code=200
        ).data

        self.assertIn('count', data)
        self.assertIn('results', data)
        self.assertEqual(len(data['results']), 1)

    def test_detail(self):
        """Test detail endpoint."""
        url = reverse('api-return-order-line-detail', kwargs={'pk': 1})

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

        self.assertIn('order', data)
        self.assertIn('item', data)
        self.assertIn('quantity', data)
        self.assertIn('outcome', data)

    def test_output_options(self):
        """Test output options for detail endpoint."""
        self.run_output_test(
            reverse('api-return-order-line-detail', kwargs={'pk': 1}),
            ['part_detail', 'item_detail', 'order_detail'],
        )

    def test_update(self):
        """Test updating ReturnOrderLineItem."""
        url = reverse('api-return-order-line-detail', kwargs={'pk': 1})

        # Without permissions
        self.patch(url, {'price': '10.50'}, expected_code=403)

        self.assignRole('return_order.change')

        self.patch(url, {'price': '15.75'}, expected_code=200)

        line = models.ReturnOrderLineItem.objects.get(pk=1)
        self.assertEqual(float(line.price.amount), 15.75)
