api.py 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723
  1. import copy
  2. import inspect
  3. import json
  4. import strawberry_django
  5. from django.conf import settings
  6. from django.contrib.contenttypes.models import ContentType
  7. from django.test import override_settings
  8. from django.urls import reverse
  9. from rest_framework import status
  10. from rest_framework.test import APIClient
  11. from strawberry.types.base import StrawberryList, StrawberryOptional
  12. from strawberry.types.lazy_type import LazyType
  13. from strawberry.types.union import StrawberryUnion
  14. from core.choices import ObjectChangeActionChoices
  15. from core.models import ObjectChange, ObjectType
  16. from ipam.graphql.types import IPAddressFamilyType
  17. from netbox.models.features import ChangeLoggingMixin
  18. from users.models import ObjectPermission, Token, User
  19. from utilities.api import get_graphql_type_for_model
  20. from .base import ModelTestCase
  21. from .utils import disable_logging, disable_warnings, get_random_string
  22. __all__ = (
  23. 'APITestCase',
  24. 'APIViewTestCases',
  25. )
  26. #
  27. # REST/GraphQL API Tests
  28. #
  29. class APITestCase(ModelTestCase):
  30. """
  31. Base test case for API requests.
  32. client_class: Test client class
  33. view_namespace: Namespace for API views. If None, the model's app_label will be used.
  34. """
  35. client_class = APIClient
  36. view_namespace = None
  37. def setUp(self):
  38. """
  39. Create a user and token for API calls.
  40. """
  41. # Create the test user and assign permissions
  42. self.user = User.objects.create_user(username='testuser')
  43. self.add_permissions(*self.user_permissions)
  44. self.token = Token.objects.create(user=self.user)
  45. self.header = {'HTTP_AUTHORIZATION': f'Token {self.token.key}'}
  46. def _get_view_namespace(self):
  47. return f'{self.view_namespace or self.model._meta.app_label}-api'
  48. def _get_detail_url(self, instance):
  49. viewname = f'{self._get_view_namespace()}:{instance._meta.model_name}-detail'
  50. return reverse(viewname, kwargs={'pk': instance.pk})
  51. def _get_list_url(self):
  52. viewname = f'{self._get_view_namespace()}:{self.model._meta.model_name}-list'
  53. return reverse(viewname)
  54. class APIViewTestCases:
  55. class GetObjectViewTestCase(APITestCase):
  56. @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], LOGIN_REQUIRED=False)
  57. def test_get_object_anonymous(self):
  58. """
  59. GET a single object as an unauthenticated user.
  60. """
  61. url = self._get_detail_url(self._get_queryset().first())
  62. if (self.model._meta.app_label, self.model._meta.model_name) in settings.EXEMPT_EXCLUDE_MODELS:
  63. # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
  64. with disable_warnings('django.request'):
  65. self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
  66. else:
  67. response = self.client.get(url, **self.header)
  68. self.assertHttpStatus(response, status.HTTP_200_OK)
  69. def test_get_object_without_permission(self):
  70. """
  71. GET a single object as an authenticated user without the required permission.
  72. """
  73. url = self._get_detail_url(self._get_queryset().first())
  74. # Try GET without permission
  75. with disable_warnings('django.request'):
  76. self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
  77. def test_get_object(self):
  78. """
  79. GET a single object as an authenticated user with permission to view the object.
  80. """
  81. self.assertGreaterEqual(self._get_queryset().count(), 2,
  82. f"Test requires the creation of at least two {self.model} instances")
  83. instance1, instance2 = self._get_queryset()[:2]
  84. # Add object-level permission
  85. obj_perm = ObjectPermission(
  86. name='Test permission',
  87. constraints={'pk': instance1.pk},
  88. actions=['view']
  89. )
  90. obj_perm.save()
  91. obj_perm.users.add(self.user)
  92. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  93. # Try GET to permitted object
  94. url = self._get_detail_url(instance1)
  95. self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_200_OK)
  96. # Try GET to non-permitted object
  97. url = self._get_detail_url(instance2)
  98. self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_404_NOT_FOUND)
  99. @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
  100. def test_options_object(self):
  101. """
  102. Make an OPTIONS request for a single object.
  103. """
  104. url = self._get_detail_url(self._get_queryset().first())
  105. response = self.client.options(url, **self.header)
  106. self.assertHttpStatus(response, status.HTTP_200_OK)
  107. class ListObjectsViewTestCase(APITestCase):
  108. brief_fields = []
  109. @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], LOGIN_REQUIRED=False)
  110. def test_list_objects_anonymous(self):
  111. """
  112. GET a list of objects as an unauthenticated user.
  113. """
  114. url = self._get_list_url()
  115. if (self.model._meta.app_label, self.model._meta.model_name) in settings.EXEMPT_EXCLUDE_MODELS:
  116. # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
  117. with disable_warnings('django.request'):
  118. self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
  119. else:
  120. response = self.client.get(url, **self.header)
  121. self.assertHttpStatus(response, status.HTTP_200_OK)
  122. self.assertEqual(len(response.data['results']), self._get_queryset().count())
  123. def test_list_objects_brief(self):
  124. """
  125. GET a list of objects using the "brief" parameter.
  126. """
  127. self.add_permissions(f'{self.model._meta.app_label}.view_{self.model._meta.model_name}')
  128. url = f'{self._get_list_url()}?brief=1'
  129. response = self.client.get(url, **self.header)
  130. self.assertEqual(len(response.data['results']), self._get_queryset().count())
  131. self.assertEqual(sorted(response.data['results'][0]), self.brief_fields)
  132. def test_list_objects_without_permission(self):
  133. """
  134. GET a list of objects as an authenticated user without the required permission.
  135. """
  136. url = self._get_list_url()
  137. # Try GET without permission
  138. with disable_warnings('django.request'):
  139. self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
  140. def test_list_objects(self):
  141. """
  142. GET a list of objects as an authenticated user with permission to view the objects.
  143. """
  144. self.assertGreaterEqual(self._get_queryset().count(), 3,
  145. f"Test requires the creation of at least three {self.model} instances")
  146. instance1, instance2 = self._get_queryset()[:2]
  147. # Add object-level permission
  148. obj_perm = ObjectPermission(
  149. name='Test permission',
  150. constraints={'pk__in': [instance1.pk, instance2.pk]},
  151. actions=['view']
  152. )
  153. obj_perm.save()
  154. obj_perm.users.add(self.user)
  155. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  156. # Try GET to permitted objects
  157. response = self.client.get(self._get_list_url(), **self.header)
  158. self.assertHttpStatus(response, status.HTTP_200_OK)
  159. self.assertEqual(len(response.data['results']), 2)
  160. @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
  161. def test_options_objects(self):
  162. """
  163. Make an OPTIONS request for a list endpoint.
  164. """
  165. response = self.client.options(self._get_list_url(), **self.header)
  166. self.assertHttpStatus(response, status.HTTP_200_OK)
  167. class CreateObjectViewTestCase(APITestCase):
  168. create_data = []
  169. validation_excluded_fields = []
  170. def test_create_object_without_permission(self):
  171. """
  172. POST a single object without permission.
  173. """
  174. url = self._get_list_url()
  175. # Try POST without permission
  176. with disable_warnings('django.request'):
  177. response = self.client.post(url, self.create_data[0], format='json', **self.header)
  178. self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
  179. def test_create_object(self):
  180. """
  181. POST a single object with permission.
  182. """
  183. # Add object-level permission
  184. obj_perm = ObjectPermission(
  185. name='Test permission',
  186. actions=['add']
  187. )
  188. obj_perm.save()
  189. obj_perm.users.add(self.user)
  190. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  191. data = copy.deepcopy(self.create_data[0])
  192. # If supported, add a changelog message
  193. if issubclass(self.model, ChangeLoggingMixin):
  194. data['changelog_message'] = get_random_string(10)
  195. initial_count = self._get_queryset().count()
  196. response = self.client.post(self._get_list_url(), data, format='json', **self.header)
  197. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  198. self.assertEqual(self._get_queryset().count(), initial_count + 1)
  199. instance = self._get_queryset().get(pk=response.data['id'])
  200. self.assertInstanceEqual(
  201. instance,
  202. self.create_data[0],
  203. exclude=self.validation_excluded_fields,
  204. api=True
  205. )
  206. # Verify ObjectChange creation
  207. if issubclass(self.model, ChangeLoggingMixin):
  208. objectchange = ObjectChange.objects.get(
  209. changed_object_type=ContentType.objects.get_for_model(instance),
  210. changed_object_id=instance.pk
  211. )
  212. self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_CREATE)
  213. self.assertEqual(objectchange.message, data['changelog_message'])
  214. def test_bulk_create_objects(self):
  215. """
  216. POST a set of objects in a single request.
  217. """
  218. # Add object-level permission
  219. obj_perm = ObjectPermission(
  220. name='Test permission',
  221. actions=['add']
  222. )
  223. obj_perm.save()
  224. obj_perm.users.add(self.user)
  225. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  226. # If supported, add a changelog message
  227. changelog_message = get_random_string(10)
  228. if issubclass(self.model, ChangeLoggingMixin):
  229. for obj_data in self.create_data:
  230. obj_data['changelog_message'] = changelog_message
  231. initial_count = self._get_queryset().count()
  232. response = self.client.post(self._get_list_url(), self.create_data, format='json', **self.header)
  233. self.assertHttpStatus(response, status.HTTP_201_CREATED)
  234. self.assertEqual(len(response.data), len(self.create_data))
  235. self.assertEqual(self._get_queryset().count(), initial_count + len(self.create_data))
  236. for i, obj in enumerate(response.data):
  237. for field in self.create_data[i]:
  238. if field == 'changelog_message':
  239. # Write-only field
  240. continue
  241. if field not in self.validation_excluded_fields:
  242. self.assertIn(field, obj, f"Bulk create field '{field}' missing from object {i} in response")
  243. for i, obj in enumerate(response.data):
  244. self.assertInstanceEqual(
  245. self._get_queryset().get(pk=obj['id']),
  246. self.create_data[i],
  247. exclude=self.validation_excluded_fields,
  248. api=True
  249. )
  250. # Verify ObjectChange creation
  251. if issubclass(self.model, ChangeLoggingMixin):
  252. id_list = [
  253. obj['id'] for obj in response.data
  254. ]
  255. objectchanges = ObjectChange.objects.filter(
  256. changed_object_type=ContentType.objects.get_for_model(self.model),
  257. changed_object_id__in=id_list
  258. )
  259. self.assertEqual(len(objectchanges), len(self.create_data))
  260. for oc in objectchanges:
  261. self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE)
  262. self.assertEqual(oc.message, changelog_message)
  263. class UpdateObjectViewTestCase(APITestCase):
  264. update_data = {}
  265. bulk_update_data = None
  266. validation_excluded_fields = []
  267. def test_update_object_without_permission(self):
  268. """
  269. PATCH a single object without permission.
  270. """
  271. url = self._get_detail_url(self._get_queryset().first())
  272. update_data = self.update_data or getattr(self, 'create_data')[0]
  273. # Try PATCH without permission
  274. with disable_warnings('django.request'):
  275. response = self.client.patch(url, update_data, format='json', **self.header)
  276. self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
  277. def test_update_object(self):
  278. """
  279. PATCH a single object identified by its numeric ID.
  280. """
  281. instance = self._get_queryset().first()
  282. url = self._get_detail_url(instance)
  283. update_data = self.update_data or getattr(self, 'create_data')[0]
  284. # Add object-level permission
  285. obj_perm = ObjectPermission(
  286. name='Test permission',
  287. actions=['change']
  288. )
  289. obj_perm.save()
  290. obj_perm.users.add(self.user)
  291. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  292. data = copy.deepcopy(update_data)
  293. # If supported, add a changelog message
  294. if issubclass(self.model, ChangeLoggingMixin):
  295. data['changelog_message'] = get_random_string(10)
  296. response = self.client.patch(url, data, format='json', **self.header)
  297. self.assertHttpStatus(response, status.HTTP_200_OK)
  298. instance.refresh_from_db()
  299. self.assertInstanceEqual(
  300. instance,
  301. data,
  302. exclude=self.validation_excluded_fields,
  303. api=True
  304. )
  305. # Verify ObjectChange creation
  306. if hasattr(self.model, 'to_objectchange'):
  307. objectchange = ObjectChange.objects.get(
  308. changed_object_type=ContentType.objects.get_for_model(instance),
  309. changed_object_id=instance.pk
  310. )
  311. self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_UPDATE)
  312. self.assertEqual(objectchange.message, data['changelog_message'])
  313. def test_bulk_update_objects(self):
  314. """
  315. PATCH a set of objects in a single request.
  316. """
  317. if self.bulk_update_data is None:
  318. self.skipTest("Bulk update data not set")
  319. # Add object-level permission
  320. obj_perm = ObjectPermission(
  321. name='Test permission',
  322. actions=['change']
  323. )
  324. obj_perm.save()
  325. obj_perm.users.add(self.user)
  326. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  327. id_list = list(self._get_queryset().values_list('id', flat=True)[:3])
  328. self.assertEqual(len(id_list), 3, "Insufficient number of objects to test bulk update")
  329. data = [
  330. {'id': id, **self.bulk_update_data} for id in id_list
  331. ]
  332. # If supported, add a changelog message
  333. changelog_message = get_random_string(10)
  334. if issubclass(self.model, ChangeLoggingMixin):
  335. for obj_data in data:
  336. obj_data['changelog_message'] = changelog_message
  337. response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
  338. self.assertHttpStatus(response, status.HTTP_200_OK)
  339. for i, obj in enumerate(response.data):
  340. for field in self.bulk_update_data:
  341. if field == 'changelog_data':
  342. # Write-only field
  343. continue
  344. self.assertIn(field, obj, f"Bulk update field '{field}' missing from object {i} in response")
  345. for instance in self._get_queryset().filter(pk__in=id_list):
  346. self.assertInstanceEqual(instance, self.bulk_update_data, api=True)
  347. # Verify ObjectChange creation
  348. if issubclass(self.model, ChangeLoggingMixin):
  349. objectchanges = ObjectChange.objects.filter(
  350. changed_object_type=ContentType.objects.get_for_model(self.model),
  351. changed_object_id__in=id_list
  352. )
  353. self.assertEqual(len(objectchanges), len(data))
  354. for oc in objectchanges:
  355. self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
  356. self.assertEqual(oc.message, changelog_message)
  357. class DeleteObjectViewTestCase(APITestCase):
  358. def test_delete_object_without_permission(self):
  359. """
  360. DELETE a single object without permission.
  361. """
  362. url = self._get_detail_url(self._get_queryset().first())
  363. # Try DELETE without permission
  364. with disable_warnings('django.request'):
  365. response = self.client.delete(url, **self.header)
  366. self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
  367. def test_delete_object(self):
  368. """
  369. DELETE a single object identified by its numeric ID.
  370. """
  371. instance = self._get_queryset().first()
  372. url = self._get_detail_url(instance)
  373. # Add object-level permission
  374. obj_perm = ObjectPermission(
  375. name='Test permission',
  376. actions=['delete']
  377. )
  378. obj_perm.save()
  379. obj_perm.users.add(self.user)
  380. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  381. data = {}
  382. # If supported, add a changelog message
  383. if issubclass(self.model, ChangeLoggingMixin):
  384. data['changelog_message'] = get_random_string(10)
  385. response = self.client.delete(url, data, **self.header)
  386. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  387. self.assertFalse(self._get_queryset().filter(pk=instance.pk).exists())
  388. # Verify ObjectChange creation
  389. if hasattr(self.model, 'to_objectchange'):
  390. objectchange = ObjectChange.objects.get(
  391. changed_object_type=ContentType.objects.get_for_model(instance),
  392. changed_object_id=instance.pk
  393. )
  394. self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_DELETE)
  395. self.assertEqual(objectchange.message, data['changelog_message'])
  396. def test_bulk_delete_objects(self):
  397. """
  398. DELETE a set of objects in a single request.
  399. """
  400. # Add object-level permission
  401. obj_perm = ObjectPermission(
  402. name='Test permission',
  403. actions=['delete']
  404. )
  405. obj_perm.save()
  406. obj_perm.users.add(self.user)
  407. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  408. # Target the three most recently created objects to avoid triggering recursive deletions
  409. # (e.g. with MPTT objects)
  410. id_list = list(self._get_queryset().order_by('-id').values_list('id', flat=True)[:3])
  411. self.assertEqual(len(id_list), 3, "Insufficient number of objects to test bulk deletion")
  412. data = [{"id": id} for id in id_list]
  413. # If supported, add a changelog message
  414. changelog_message = get_random_string(10)
  415. if issubclass(self.model, ChangeLoggingMixin):
  416. for obj_data in data:
  417. obj_data['changelog_message'] = changelog_message
  418. initial_count = self._get_queryset().count()
  419. response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
  420. self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
  421. self.assertEqual(self._get_queryset().count(), initial_count - 3)
  422. # Verify ObjectChange creation
  423. if issubclass(self.model, ChangeLoggingMixin):
  424. objectchanges = ObjectChange.objects.filter(
  425. changed_object_type=ContentType.objects.get_for_model(self.model),
  426. changed_object_id__in=id_list
  427. )
  428. self.assertEqual(len(objectchanges), len(data))
  429. for oc in objectchanges:
  430. self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
  431. self.assertEqual(oc.message, changelog_message)
  432. class GraphQLTestCase(APITestCase):
  433. def _get_graphql_base_name(self):
  434. """
  435. Return graphql_base_name, if set. Otherwise, construct the base name for the query
  436. field from the model's verbose name.
  437. """
  438. base_name = self.model._meta.verbose_name.lower().replace(' ', '_')
  439. return getattr(self, 'graphql_base_name', base_name)
  440. def _build_query_with_filter(self, name, filter_string):
  441. """
  442. Called by either _build_query or _build_filtered_query - construct the actual
  443. query given a name and filter string
  444. """
  445. type_class = get_graphql_type_for_model(self.model)
  446. # Compile list of fields to include
  447. fields_string = ''
  448. file_fields = (
  449. strawberry_django.fields.types.DjangoFileType,
  450. strawberry_django.fields.types.DjangoImageType,
  451. )
  452. for field in type_class.__strawberry_definition__.fields:
  453. if (
  454. field.type in file_fields or (
  455. type(field.type) is StrawberryOptional and field.type.of_type in file_fields
  456. )
  457. ):
  458. # image / file fields nullable or not...
  459. fields_string += f'{field.name} {{ name }}\n'
  460. elif type(field.type) is StrawberryList and type(field.type.of_type) is LazyType:
  461. # List of related objects (queryset)
  462. fields_string += f'{field.name} {{ id }}\n'
  463. elif type(field.type) is StrawberryList and type(field.type.of_type) is StrawberryUnion:
  464. # this would require a fragment query
  465. continue
  466. elif type(field.type) is StrawberryUnion:
  467. # this would require a fragment query
  468. continue
  469. elif type(field.type) is StrawberryOptional and type(field.type.of_type) is StrawberryUnion:
  470. # this would require a fragment query
  471. continue
  472. elif type(field.type) is StrawberryOptional and type(field.type.of_type) is LazyType:
  473. fields_string += f'{field.name} {{ id }}\n'
  474. elif hasattr(field, 'is_relation') and field.is_relation:
  475. # Ignore private fields
  476. if field.name.startswith('_'):
  477. continue
  478. # Note: StrawberryField types do not have is_relation
  479. fields_string += f'{field.name} {{ id }}\n'
  480. elif inspect.isclass(field.type) and issubclass(field.type, IPAddressFamilyType):
  481. fields_string += f'{field.name} {{ value, label }}\n'
  482. else:
  483. fields_string += f'{field.name}\n'
  484. query = f"""
  485. {{
  486. {name}{filter_string} {{
  487. {fields_string}
  488. }}
  489. }}
  490. """
  491. return query
  492. def _build_filtered_query(self, name, **filters):
  493. """
  494. Create a filtered query: i.e. device_list(filters: {name: {i_contains: "akron"}}){.
  495. """
  496. # TODO: This should be extended to support AND, OR multi-lookups
  497. if filters:
  498. for field_name, params in filters.items():
  499. lookup = params['lookup']
  500. value = params['value']
  501. if lookup:
  502. query = f'{{{lookup}: "{value}"}}'
  503. filter_string = f'{field_name}: {query}'
  504. else:
  505. filter_string = f'{field_name}: "{value}"'
  506. filter_string = f'(filters: {{{filter_string}}})'
  507. else:
  508. filter_string = ''
  509. return self._build_query_with_filter(name, filter_string)
  510. def _build_query(self, name, **filters):
  511. """
  512. Create a normal query - unfiltered or with a string query: i.e. site(name: "aaa"){.
  513. """
  514. if filters:
  515. filter_string = ', '.join(f'{k}:{v}' for k, v in filters.items())
  516. filter_string = f'({filter_string})'
  517. else:
  518. filter_string = ''
  519. return self._build_query_with_filter(name, filter_string)
  520. @override_settings(LOGIN_REQUIRED=True)
  521. def test_graphql_get_object(self):
  522. url = reverse('graphql')
  523. field_name = self._get_graphql_base_name()
  524. object_id = self._get_queryset().first().pk
  525. query = self._build_query(field_name, id=object_id)
  526. # Non-authenticated requests should fail
  527. header = {
  528. 'HTTP_ACCEPT': 'application/json',
  529. }
  530. with disable_warnings('django.request'):
  531. response = self.client.post(url, data={'query': query}, format="json", **header)
  532. self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
  533. # Add constrained permission
  534. obj_perm = ObjectPermission(
  535. name='Test permission',
  536. actions=['view'],
  537. constraints={'id': 0} # Impossible constraint
  538. )
  539. obj_perm.save()
  540. obj_perm.users.add(self.user)
  541. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  542. # Request should succeed but return empty result
  543. with disable_logging():
  544. response = self.client.post(url, data={'query': query}, format="json", **self.header)
  545. self.assertHttpStatus(response, status.HTTP_200_OK)
  546. data = json.loads(response.content)
  547. self.assertIn('errors', data)
  548. self.assertIsNone(data['data'])
  549. # Remove permission constraint
  550. obj_perm.constraints = None
  551. obj_perm.save()
  552. # Request should return requested object
  553. response = self.client.post(url, data={'query': query}, format="json", **self.header)
  554. self.assertHttpStatus(response, status.HTTP_200_OK)
  555. data = json.loads(response.content)
  556. self.assertNotIn('errors', data)
  557. self.assertIsNotNone(data['data'])
  558. @override_settings(LOGIN_REQUIRED=True)
  559. def test_graphql_list_objects(self):
  560. url = reverse('graphql')
  561. field_name = f'{self._get_graphql_base_name()}_list'
  562. query = self._build_query(field_name)
  563. # Non-authenticated requests should fail
  564. header = {
  565. 'HTTP_ACCEPT': 'application/json',
  566. }
  567. with disable_warnings('django.request'):
  568. response = self.client.post(url, data={'query': query}, format="json", **header)
  569. self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
  570. # Add constrained permission
  571. obj_perm = ObjectPermission(
  572. name='Test permission',
  573. actions=['view'],
  574. constraints={'id': 0} # Impossible constraint
  575. )
  576. obj_perm.save()
  577. obj_perm.users.add(self.user)
  578. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  579. # Request should succeed but return empty results list
  580. response = self.client.post(url, data={'query': query}, format="json", **self.header)
  581. self.assertHttpStatus(response, status.HTTP_200_OK)
  582. data = json.loads(response.content)
  583. self.assertNotIn('errors', data)
  584. self.assertEqual(len(data['data'][field_name]), 0)
  585. # Remove permission constraint
  586. obj_perm.constraints = None
  587. obj_perm.save()
  588. # Request should return all objects
  589. response = self.client.post(url, data={'query': query}, format="json", **self.header)
  590. self.assertHttpStatus(response, status.HTTP_200_OK)
  591. data = json.loads(response.content)
  592. self.assertNotIn('errors', data)
  593. self.assertEqual(len(data['data'][field_name]), self.model.objects.count())
  594. @override_settings(LOGIN_REQUIRED=True)
  595. def test_graphql_filter_objects(self):
  596. if not hasattr(self, 'graphql_filter'):
  597. return
  598. url = reverse('graphql')
  599. field_name = f'{self._get_graphql_base_name()}_list'
  600. query = self._build_filtered_query(field_name, **self.graphql_filter)
  601. # Add object-level permission
  602. obj_perm = ObjectPermission(
  603. name='Test permission',
  604. actions=['view']
  605. )
  606. obj_perm.save()
  607. obj_perm.users.add(self.user)
  608. obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
  609. response = self.client.post(url, data={'query': query}, format="json", **self.header)
  610. self.assertHttpStatus(response, status.HTTP_200_OK)
  611. data = json.loads(response.content)
  612. self.assertNotIn('errors', data)
  613. self.assertGreater(len(data['data'][field_name]), 0)
  614. class APIViewTestCase(
  615. GetObjectViewTestCase,
  616. ListObjectsViewTestCase,
  617. CreateObjectViewTestCase,
  618. UpdateObjectViewTestCase,
  619. DeleteObjectViewTestCase,
  620. GraphQLTestCase
  621. ):
  622. pass