4548642839

Build Type

push

github

Committed by GitHub

Commit Message

[Feature] Add RMA support (#4488)

Run Details

1082 of 1082 new or added lines in 49 files covered. (100.0%)

26469 of 30045 relevant lines covered (88.1%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.08

/InvenTree/InvenTree/api_tester.py

"""Helper functions for performing API unit tests."""

import csv
import io
import re

from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.http.response import StreamingHttpResponse

from djmoney.contrib.exchange.models import ExchangeBackend, Rate
from rest_framework.test import APITestCase

from plugin import registry
from plugin.models import PluginConfig


class UserMixin:
    """Mixin to setup a user and login for tests.

    Use parameters to set username, password, email, roles and permissions.
    """

    # User information
    username = 'testuser'
    password = 'mypassword'
    email = 'test@testing.com'

    superuser = False
    is_staff = True
    auto_login = True

    # Set list of roles automatically associated with the user
    roles = []

    @classmethod
    def setUpTestData(cls):
        """Run setup for all tests in a given class"""
        super().setUpTestData()

        # Create a user to log in with
        cls.user = get_user_model().objects.create_user(
            username=cls.username,
            password=cls.password,
            email=cls.email
        )

        # Create a group for the user
        cls.group = Group.objects.create(name='my_test_group')
        cls.user.groups.add(cls.group)

        if cls.superuser:
            cls.user.is_superuser = True

        if cls.is_staff:
            cls.user.is_staff = True

        cls.user.save()

        # Assign all roles if set
        if cls.roles == 'all':
            cls.assignRole(group=cls.group, assign_all=True)

        # else filter the roles
        else:
            for role in cls.roles:
                cls.assignRole(role=role, group=cls.group)

    def setUp(self):
        """Run setup for individual test methods"""

        if self.auto_login:
            self.client.login(username=self.username, password=self.password)

    @classmethod
    def assignRole(cls, role=None, assign_all: bool = False, group=None):
        """Set the user roles for the registered user.

        Arguments:
            role: Role of the format 'rule.permission' e.g. 'part.add'
            assign_all: Set to True to assign *all* roles
            group: The group to assign roles to (or leave None to use the group assigned to this class)
        """

        if group is None:
            group = cls.group

        if type(assign_all) is not bool:
            # Raise exception if common mistake is made!
            raise TypeError('assignRole: assign_all must be a boolean value')

        if not role and not assign_all:
            raise ValueError('assignRole: either role must be provided, or assign_all must be set')

        if not assign_all and role:
            rule, perm = role.split('.')

        for ruleset in group.rule_sets.all():

            if assign_all or ruleset.name == rule:

                if assign_all or perm == 'view':
                    ruleset.can_view = True
                elif assign_all or perm == 'change':
                    ruleset.can_change = True
                elif assign_all or perm == 'delete':
                    ruleset.can_delete = True
                elif assign_all or perm == 'add':
                    ruleset.can_add = True

                ruleset.save()
                break


class PluginMixin:
    """Mixin to ensure that all plugins are loaded for tests."""

    def setUp(self):
        """Setup for plugin tests."""
        super().setUp()

        # Load plugin configs
        self.plugin_confs = PluginConfig.objects.all()
        # Reload if not present
        if not self.plugin_confs:
            registry.reload_plugins()
            self.plugin_confs = PluginConfig.objects.all()


class ExchangeRateMixin:
    """Mixin class for generating exchange rate data"""

    def generate_exchange_rates(self):
        """Helper function which generates some exchange rates to work with"""

        rates = {
            'AUD': 1.5,
            'CAD': 1.7,
            'GBP': 0.9,
            'USD': 1.0,
        }

        # Create a dummy backend
        ExchangeBackend.objects.create(
            name='InvenTreeExchange',
            base_currency='USD',
        )

        backend = ExchangeBackend.objects.get(name='InvenTreeExchange')

        items = []

        for currency, rate in rates.items():
            items.append(
                Rate(
                    currency=currency,
                    value=rate,
                    backend=backend,
                )
            )

        Rate.objects.bulk_create(items)


class InvenTreeAPITestCase(ExchangeRateMixin, UserMixin, APITestCase):
    """Base class for running InvenTree API tests."""

    def checkResponse(self, url, method, expected_code, response):
        """Debug output for an unexpected response"""

        # No expected code, return
        if expected_code is None:
            return

        if expected_code != response.status_code:

            print(f"Unexpected {method} response at '{url}': status_code = {response.status_code}")

            if hasattr(response, 'data'):
                print('data:', response.data)
            if hasattr(response, 'body'):
                print('body:', response.body)
            if hasattr(response, 'content'):
                print('content:', response.content)

        self.assertEqual(expected_code, response.status_code)

    def getActions(self, url):
        """Return a dict of the 'actions' available at a given endpoint.

        Makes use of the HTTP 'OPTIONS' method to request this.
        """
        response = self.client.options(url)
        self.assertEqual(response.status_code, 200)

        actions = response.data.get('actions', None)

        if not actions:
            actions = {}

        return actions

    def get(self, url, data=None, expected_code=200, format='json'):
        """Issue a GET request."""
        # Set default - see B006
        if data is None:
            data = {}

        response = self.client.get(url, data, format=format)

        self.checkResponse(url, 'GET', expected_code, response)

        return response

    def post(self, url, data=None, expected_code=None, format='json'):
        """Issue a POST request."""

        # Set default value - see B006
        if data is None:
            data = {}

        response = self.client.post(url, data=data, format=format)

        self.checkResponse(url, 'POST', expected_code, response)

        return response

    def delete(self, url, data=None, expected_code=None, format='json'):
        """Issue a DELETE request."""

        if data is None:
            data = {}

        response = self.client.delete(url, data=data, format=format)

        self.checkResponse(url, 'DELETE', expected_code, response)

        return response

    def patch(self, url, data, expected_code=None, format='json'):
        """Issue a PATCH request."""
        response = self.client.patch(url, data=data, format=format)

        self.checkResponse(url, 'PATCH', expected_code, response)

        return response

    def put(self, url, data, expected_code=None, format='json'):
        """Issue a PUT request."""
        response = self.client.put(url, data=data, format=format)

        self.checkResponse(url, 'PUT', expected_code, response)

        return response

    def options(self, url, expected_code=None):
        """Issue an OPTIONS request."""
        response = self.client.options(url, format='json')

        self.checkResponse(url, 'OPTIONS', expected_code, response)

        return response

    def download_file(self, url, data, expected_code=None, expected_fn=None, decode=True):
        """Download a file from the server, and return an in-memory file."""
        response = self.client.get(url, data=data, format='json')

        if expected_code is not None:
            self.assertEqual(response.status_code, expected_code)

        # Check that the response is of the correct type
        if not isinstance(response, StreamingHttpResponse):
            raise ValueError("Response is not a StreamingHttpResponse object as expected")

        # Extract filename
        disposition = response.headers['Content-Disposition']

        result = re.search(r'attachment; filename="([\w.]+)"', disposition)

        fn = result.groups()[0]

        if expected_fn is not None:
            self.assertEqual(expected_fn, fn)

        if decode:
            # Decode data and return as StringIO file object
            fo = io.StringIO()
            fo.name = fo
            fo.write(response.getvalue().decode('UTF-8'))
        else:
            # Return a a BytesIO file object
            fo = io.BytesIO()
            fo.name = fn
            fo.write(response.getvalue())

        fo.seek(0)

        return fo

    def process_csv(self, fo, delimiter=',', required_cols=None, excluded_cols=None, required_rows=None):
        """Helper function to process and validate a downloaded csv file."""
        # Check that the correct object type has been passed
        self.assertTrue(isinstance(fo, io.StringIO))

        fo.seek(0)

        reader = csv.reader(fo, delimiter=delimiter)

        headers = []
        rows = []

        for idx, row in enumerate(reader):
            if idx == 0:
                headers = row
            else:
                rows.append(row)

        if required_cols is not None:
            for col in required_cols:
                self.assertIn(col, headers)

        if excluded_cols is not None:
            for col in excluded_cols:
                self.assertNotIn(col, headers)

        if required_rows is not None:
            self.assertEqual(len(rows), required_rows)

        # Return the file data as a list of dict items, based on the headers
        data = []

        for row in rows:
            entry = {}

            for idx, col in enumerate(headers):
                entry[col] = row[idx]

            data.append(entry)

        return data

1	"""Helper functions for performing API unit tests."""
2
3	import csv	1✔
4	import io	1✔
5	import re	1✔
6
7	from django.contrib.auth import get_user_model	1✔
8	from django.contrib.auth.models import Group	1✔
9	from django.http.response import StreamingHttpResponse	1✔
10
11	from djmoney.contrib.exchange.models import ExchangeBackend, Rate	1✔
12	from rest_framework.test import APITestCase	1✔
13
14	from plugin import registry	1✔
15	from plugin.models import PluginConfig	1✔
16
17
18	class UserMixin:	1✔
19	"""Mixin to setup a user and login for tests.
20
21	Use parameters to set username, password, email, roles and permissions.
22	"""
23
24	# User information
25	username = 'testuser'	1✔
26	password = 'mypassword'	1✔
27	email = 'test@testing.com'	1✔
28
29	superuser = False	1✔
30	is_staff = True	1✔
31	auto_login = True	1✔
32
33	# Set list of roles automatically associated with the user
34	roles = []	1✔
35
36	@classmethod	1✔
37	def setUpTestData(cls):	1✔
38	"""Run setup for all tests in a given class"""
39	super().setUpTestData()	1✔
40
41	# Create a user to log in with
42	cls.user = get_user_model().objects.create_user(	1✔
43	username=cls.username,
44	password=cls.password,
45	email=cls.email
46	)
47
48	# Create a group for the user
49	cls.group = Group.objects.create(name='my_test_group')	1✔
50	cls.user.groups.add(cls.group)	1✔
51
52	if cls.superuser:	1✔
53	cls.user.is_superuser = True	1✔
54
55	if cls.is_staff:	1✔
56	cls.user.is_staff = True	1✔
57
58	cls.user.save()	1✔
59
60	# Assign all roles if set
61	if cls.roles == 'all':	1✔
62	cls.assignRole(group=cls.group, assign_all=True)	1✔
63
64	# else filter the roles
65	else:
66	for role in cls.roles:	1✔
67	cls.assignRole(role=role, group=cls.group)	1✔
68
69	def setUp(self):	1✔
70	"""Run setup for individual test methods"""
71
72	if self.auto_login:	1✔
73	self.client.login(username=self.username, password=self.password)	1✔
74
75	@classmethod	1✔
76	def assignRole(cls, role=None, assign_all: bool = False, group=None):	1✔
77	"""Set the user roles for the registered user.
78
79	Arguments:
80	role: Role of the format 'rule.permission' e.g. 'part.add'
81	assign_all: Set to True to assign all roles
82	group: The group to assign roles to (or leave None to use the group assigned to this class)
83	"""
84
85	if group is None:	1✔
86	group = cls.group	1✔
87
88	if type(assign_all) is not bool:	1✔
89	# Raise exception if common mistake is made!
90	raise TypeError('assignRole: assign_all must be a boolean value')	×
91
92	if not role and not assign_all:	1✔
93	raise ValueError('assignRole: either role must be provided, or assign_all must be set')	×
94
95	if not assign_all and role:	1✔
96	rule, perm = role.split('.')	1✔
97
98	for ruleset in group.rule_sets.all():	1✔
99
100	if assign_all or ruleset.name == rule:	1✔
101
102	if assign_all or perm == 'view':	1✔
103	ruleset.can_view = True	1✔
104	elif assign_all or perm == 'change':	1✔
105	ruleset.can_change = True	1✔
106	elif assign_all or perm == 'delete':	1✔
107	ruleset.can_delete = True	1✔
108	elif assign_all or perm == 'add':	1✔
109	ruleset.can_add = True	1✔
110
111	ruleset.save()	1✔
112	break	1✔
113
114
115	class PluginMixin:	1✔
116	"""Mixin to ensure that all plugins are loaded for tests."""
117
118	def setUp(self):	1✔
119	"""Setup for plugin tests."""
120	super().setUp()	1✔
121
122	# Load plugin configs
123	self.plugin_confs = PluginConfig.objects.all()	1✔
124	# Reload if not present
125	if not self.plugin_confs:	1✔
126	registry.reload_plugins()	1✔
127	self.plugin_confs = PluginConfig.objects.all()	1✔
128
129
130	class ExchangeRateMixin:	1✔
131	"""Mixin class for generating exchange rate data"""
132
133	def generate_exchange_rates(self):	1✔
134	"""Helper function which generates some exchange rates to work with"""
135
136	rates = {	1✔
137	'AUD': 1.5,
138	'CAD': 1.7,
139	'GBP': 0.9,
140	'USD': 1.0,
141	}
142
143	# Create a dummy backend
144	ExchangeBackend.objects.create(	1✔
145	name='InvenTreeExchange',
146	base_currency='USD',
147	)
148
149	backend = ExchangeBackend.objects.get(name='InvenTreeExchange')	1✔
150
151	items = []	1✔
152
153	for currency, rate in rates.items():	1✔
154	items.append(	1✔
155	Rate(
156	currency=currency,
157	value=rate,
158	backend=backend,
159	)
160	)
161
162	Rate.objects.bulk_create(items)	1✔
163
164
165	class InvenTreeAPITestCase(ExchangeRateMixin, UserMixin, APITestCase):	1✔
166	"""Base class for running InvenTree API tests."""
167
168	def checkResponse(self, url, method, expected_code, response):	1✔
169	"""Debug output for an unexpected response"""
170
171	# No expected code, return
172	if expected_code is None:	1✔
173	return	1✔
174
175	if expected_code != response.status_code:	1✔
176
177	print(f"Unexpected {method} response at '{url}': status_code = {response.status_code}")	×
178
179	if hasattr(response, 'data'):	×
180	print('data:', response.data)	×
181	if hasattr(response, 'body'):	×
182	print('body:', response.body)	×
183	if hasattr(response, 'content'):	×
184	print('content:', response.content)	×
185
186	self.assertEqual(expected_code, response.status_code)	1✔
187
188	def getActions(self, url):	1✔
189	"""Return a dict of the 'actions' available at a given endpoint.
190
191	Makes use of the HTTP 'OPTIONS' method to request this.
192	"""
193	response = self.client.options(url)	1✔
194	self.assertEqual(response.status_code, 200)	1✔
195
196	actions = response.data.get('actions', None)	1✔
197
198	if not actions:	1✔
199	actions = {}	×
200
201	return actions	1✔
202
203	def get(self, url, data=None, expected_code=200, format='json'):	1✔
204	"""Issue a GET request."""
205	# Set default - see B006
206	if data is None:	1✔
207	data = {}	1✔
208
209	response = self.client.get(url, data, format=format)	1✔
210
211	self.checkResponse(url, 'GET', expected_code, response)	1✔
212
213	return response	1✔
214
215	def post(self, url, data=None, expected_code=None, format='json'):	1✔
216	"""Issue a POST request."""
217
218	# Set default value - see B006
219	if data is None:	1✔
220	data = {}	1✔
221
222	response = self.client.post(url, data=data, format=format)	1✔
223
224	self.checkResponse(url, 'POST', expected_code, response)	1✔
225
226	return response	1✔
227
228	def delete(self, url, data=None, expected_code=None, format='json'):	1✔
229	"""Issue a DELETE request."""
230
231	if data is None:	1✔
232	data = {}	1✔
233
234	response = self.client.delete(url, data=data, format=format)	1✔
235
236	self.checkResponse(url, 'DELETE', expected_code, response)	1✔
237
238	return response	1✔
239
240	def patch(self, url, data, expected_code=None, format='json'):	1✔
241	"""Issue a PATCH request."""
242	response = self.client.patch(url, data=data, format=format)	1✔
243
244	self.checkResponse(url, 'PATCH', expected_code, response)	1✔
245
246	return response	1✔
247
248	def put(self, url, data, expected_code=None, format='json'):	1✔
249	"""Issue a PUT request."""
250	response = self.client.put(url, data=data, format=format)	1✔
251
252	self.checkResponse(url, 'PUT', expected_code, response)	1✔
253
254	return response	1✔
255
256	def options(self, url, expected_code=None):	1✔
257	"""Issue an OPTIONS request."""
258	response = self.client.options(url, format='json')	1✔
259
260	self.checkResponse(url, 'OPTIONS', expected_code, response)	1✔
261
262	return response	1✔
263
264	def download_file(self, url, data, expected_code=None, expected_fn=None, decode=True):	1✔
265	"""Download a file from the server, and return an in-memory file."""
266	response = self.client.get(url, data=data, format='json')	1✔
267
268	if expected_code is not None:	1✔
269	self.assertEqual(response.status_code, expected_code)	1✔
270
271	# Check that the response is of the correct type
272	if not isinstance(response, StreamingHttpResponse):	1✔
273	raise ValueError("Response is not a StreamingHttpResponse object as expected")	1✔
274
275	# Extract filename
276	disposition = response.headers['Content-Disposition']	1✔
277
278	result = re.search(r'attachment; filename="([\w.]+)"', disposition)	1✔
279
280	fn = result.groups()[0]	1✔
281
282	if expected_fn is not None:	1✔
283	self.assertEqual(expected_fn, fn)	1✔
284
285	if decode:	1✔
286	# Decode data and return as StringIO file object
287	fo = io.StringIO()	1✔
288	fo.name = fo	1✔
289	fo.write(response.getvalue().decode('UTF-8'))	1✔
290	else:
291	# Return a a BytesIO file object
292	fo = io.BytesIO()	1✔
293	fo.name = fn	1✔
294	fo.write(response.getvalue())	1✔
295
296	fo.seek(0)	1✔
297
298	return fo	1✔
299
300	def process_csv(self, fo, delimiter=',', required_cols=None, excluded_cols=None, required_rows=None):	1✔
301	"""Helper function to process and validate a downloaded csv file."""
302	# Check that the correct object type has been passed
303	self.assertTrue(isinstance(fo, io.StringIO))	1✔
304
305	fo.seek(0)	1✔
306
307	reader = csv.reader(fo, delimiter=delimiter)	1✔
308
309	headers = []	1✔
310	rows = []	1✔
311
312	for idx, row in enumerate(reader):	1✔
313	if idx == 0:	1✔
314	headers = row	1✔
315	else:
316	rows.append(row)	1✔
317
318	if required_cols is not None:	1✔
319	for col in required_cols:	1✔
320	self.assertIn(col, headers)	1✔
321
322	if excluded_cols is not None:	1✔
323	for col in excluded_cols:	1✔
324	self.assertNotIn(col, headers)	1✔
325
326	if required_rows is not None:	1✔
327	self.assertEqual(len(rows), required_rows)	1✔
328
329	# Return the file data as a list of dict items, based on the headers
330	data = []	1✔
331
332	for row in rows:	1✔
333	entry = {}	1✔
334
335	for idx, col in enumerate(headers):	1✔
336	entry[col] = row[idx]	1✔
337
338	data.append(entry)	1✔
339
340	return data	1✔

inventree / InvenTree / 4548642839

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous