351 lines
11 KiB
Python
351 lines
11 KiB
Python
import random
|
|
import string
|
|
|
|
from rest_framework.test import APITestCase
|
|
from rest_framework import status
|
|
|
|
from core.factories import CustomUserFactory
|
|
from core.utils import get_tokens_for_user
|
|
|
|
from products.factories import ProductFactory
|
|
from companies.factories import CompanyFactory
|
|
|
|
from .models import StatsLog
|
|
from .factories import StatsLogFactory
|
|
|
|
|
|
# Create your tests here.
|
|
class TrackUserViewTest(APITestCase):
|
|
"""ProductViewSet tests
|
|
"""
|
|
|
|
def setUp(self):
|
|
"""Tests setup
|
|
"""
|
|
self.endpoint = '/api/v1/stats/me/'
|
|
self.factory = StatsLogFactory
|
|
self.model = StatsLog
|
|
# create user
|
|
self.password = ''.join(random.choices(string.ascii_uppercase, k = 10))
|
|
self.user = CustomUserFactory(email="test@mail.com", password=self.password, is_active=True)
|
|
|
|
# anon user
|
|
def test_anon_user_can_only_post(self):
|
|
"""Not logged-in user can only POST data
|
|
"""
|
|
# Create instance
|
|
product = ProductFactory()
|
|
|
|
data = {
|
|
'action': 'VIEW',
|
|
'action_object': {
|
|
'model': 'product',
|
|
'id': product.id,
|
|
},
|
|
}
|
|
|
|
# Query endpoint
|
|
response = self.client.get(self.endpoint)
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
|
|
|
|
# Query endpoint
|
|
response = self.client.put(self.endpoint, data={})
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
|
|
|
|
# Query endpoint
|
|
response = self.client.delete(self.endpoint, data={})
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
|
|
|
|
# Query endpoint
|
|
response = self.client.post(self.endpoint, data=data, format='json')
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
def test_anon_user_can_register_product_action(self):
|
|
"""Not logged-in user cannot modify existing instance
|
|
"""
|
|
# Create instance
|
|
product = ProductFactory()
|
|
|
|
data = {
|
|
'action': 'VIEW',
|
|
'action_object': {
|
|
'model': 'product',
|
|
'id': product.id,
|
|
},
|
|
}
|
|
|
|
# Query endpoint
|
|
response = self.client.post(self.endpoint, data=data, format='json')
|
|
|
|
# Assert forbidden code
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
def test_anon_user_can_register_company_action(self):
|
|
"""Not logged-in user cannot modify existing instance
|
|
"""
|
|
# Create instance
|
|
instance = self.factory()
|
|
|
|
# Query endpoint
|
|
url = self.endpoint + f'{instance.pk}/'
|
|
response = self.client.put(url, {}, format='json')
|
|
|
|
# Assert forbidden code
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
# authenticated user
|
|
def test_auth_user_can_only_post(self):
|
|
"""Regular logged-in user can list instance
|
|
"""
|
|
# Create instances
|
|
instances = [self.factory() for n in range(random.randint(1,5))]
|
|
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Request list
|
|
response = self.client.get(self.endpoint)
|
|
|
|
# Assert access is allowed
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
# Assert all instances are returned
|
|
self.assertEqual(len(instances), len(response.data))
|
|
|
|
def test_auth_user_can_register_product_action(self):
|
|
"""Regular logged-in user can list instance
|
|
"""
|
|
# Create instances
|
|
instance = self.factory()
|
|
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Request list
|
|
url = f"{self.endpoint}{instance.id}/"
|
|
response = self.client.get(url)
|
|
|
|
# Assert access is allowed
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
data = json.loads(response.content)
|
|
self.assertEquals(instance.id, data['id'])
|
|
|
|
def test_auth_user_can_register_company_action(self):
|
|
"""Regular logged-in user can create new instance
|
|
"""
|
|
# Define request data
|
|
company = CompanyFactory()
|
|
data = {
|
|
'company': company.id,
|
|
'sku': 'qwerewq',
|
|
'name': 'qwerewq',
|
|
'description': 'qwerewq',
|
|
'url': 'http://qwerewq.com',
|
|
'price': '12.21',
|
|
'shipping_cost': '21.12',
|
|
'shipping_terms': 'qwerewq',
|
|
'source': 'SYNCHRONIZED',
|
|
'sourcing_date': datetime.datetime.now().isoformat()+'Z',
|
|
'update_date': datetime.datetime.now().isoformat()+'Z',
|
|
'discount': '0.05',
|
|
'stock': 22,
|
|
'tags': ['tag1, tag2'],
|
|
'category': 'Mr',
|
|
'attributes': ['color/red', 'size/xxl'],
|
|
'identifiers': '34rf34f43c43',
|
|
}
|
|
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Query endpoint
|
|
response = self.client.post(self.endpoint, data=data, format='json')
|
|
# Assert endpoint returns created status
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
# Assert instance exists on db
|
|
self.assertTrue(self.model.objects.get(id=response.data['id']))
|
|
|
|
|
|
class StatsLogViewSetTest(APITestCase):
|
|
"""ProductViewSet tests
|
|
"""
|
|
|
|
def setUp(self):
|
|
"""Tests setup
|
|
"""
|
|
self.endpoint = '/api/v1/stats/'
|
|
self.factory = StatsLogFactory
|
|
self.model = StatsLog
|
|
# create user
|
|
self.password = ''.join(random.choices(string.ascii_uppercase, k = 10))
|
|
self.user = CustomUserFactory(email="test@mail.com", password=self.password, is_active=True)
|
|
self.instance_data = {
|
|
'action_object': None,
|
|
'user': self.user.id,
|
|
'anonymous': False,
|
|
'ip_address': None,
|
|
'geo': None,
|
|
'contact': None,
|
|
'shop': True,
|
|
|
|
}
|
|
self.alt_data = {
|
|
'action_object': None,
|
|
'user': self.user.id,
|
|
'anonymous': True,
|
|
'ip_address': None,
|
|
'geo': None,
|
|
'contact': None,
|
|
'shop': False,
|
|
}
|
|
|
|
# anonymous user
|
|
def test_anon_user_cannot_crud(self):
|
|
"""Not logged-in user cannot access endpoint at all
|
|
"""
|
|
|
|
# Query endpoint
|
|
response = self.client.get(self.endpoint)
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
# Query endpoint
|
|
response = self.client.post(self.endpoint, data={})
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
# Query endpoint
|
|
response = self.client.put(self.endpoint, data={})
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
# Query endpoint
|
|
response = self.client.delete(self.endpoint)
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
# authenticated user
|
|
def test_auth_user_cannot_crud(self):
|
|
"""Authenticated user cannot access endpoint at all
|
|
"""
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Query endpoint
|
|
response = self.client.get(self.endpoint)
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
# Query endpoint
|
|
response = self.client.post(self.endpoint, data={})
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
# Query endpoint
|
|
response = self.client.put(self.endpoint, data={})
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
# Query endpoint
|
|
response = self.client.delete(self.endpoint)
|
|
# Assert access is forbidden
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
# admin user
|
|
def test_admin_user_can_list_instance(self):
|
|
"""Admin user can list instance
|
|
"""
|
|
# make user be admin
|
|
self.user.is_staff = True
|
|
self.user.save()
|
|
|
|
# Create instances
|
|
instances = [self.factory() for n in range(random.randint(1,5))]
|
|
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Request list
|
|
response = self.client.get(self.endpoint)
|
|
|
|
# Assert access is allowed
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
# Assert all instances are returned
|
|
self.assertEqual(len(instances), len(response.data))
|
|
|
|
def test_admin_user_can_create_instance(self):
|
|
"""Admin user can create new instance
|
|
"""
|
|
# make user be admin
|
|
self.user.is_staff = True
|
|
self.user.save()
|
|
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Query endpoint
|
|
response = self.client.post(self.endpoint, data=self.instance_data, format='json')
|
|
|
|
# Assert endpoint returns created status
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
# Assert instance exists on db
|
|
self.assertTrue(self.model.objects.get(id=response.data['id']))
|
|
|
|
def test_admin_user_can_modify_instance(self):
|
|
"""Admin user can modify existing instance
|
|
"""
|
|
# make user be admin
|
|
self.user.is_staff = True
|
|
self.user.save()
|
|
|
|
# Create instances
|
|
instance = self.factory()
|
|
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Query endpoint
|
|
url = self.endpoint + f'{instance.pk}/'
|
|
response = self.client.put(url, self.alt_data, format='json')
|
|
|
|
# Assert endpoint returns OK code
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
def test_admin_user_can_delete_instance(self):
|
|
"""Admin user can delete existing instance
|
|
"""
|
|
# make user be admin
|
|
self.user.is_staff = True
|
|
self.user.save()
|
|
|
|
# Create instances
|
|
instance = self.factory()
|
|
|
|
# Authenticate user
|
|
token = get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}")
|
|
|
|
# Query endpoint
|
|
url = self.endpoint + f'{instance.pk}/'
|
|
response = self.client.delete(url)
|
|
|
|
# assert 204 no content
|
|
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
|
# Assert instance doesn't exists anymore on db
|
|
self.assertFalse(self.model.objects.filter(id=instance.pk).exists())
|
|
|