import random import string import json import datetime from urllib.parse import quote from django.utils import timezone from django.test import TestCase from django.core import mail from rest_framework.test import APITestCase from rest_framework import status from companies.factories import CompanyFactory from products.factories import ProductFactory, ActiveProductFactory from products.models import Product, CategoryTag from core.factories import CustomUserFactory from core.utils import get_tokens_for_user # Create your tests here. class ProductViewSetTest(APITestCase): """ProductViewSet tests """ def setUp(self): """Tests setup """ self.endpoint = '/api/v1/products/' self.factory = ActiveProductFactory self.model = Product # 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) def tearDown(self): self.model.objects.all().delete() # anon user def test_anon_user_cannot_create_instance(self): """Not logged-in user cannot create new instance """ instances = [self.factory() for n in range(random.randint(1,5))] # Query endpoint response = self.client.post(self.endpoint, data={}) # Assert access is forbidden self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_anon_user_cannot_modify_existing_instance(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) def test_anon_user_cannot_delete_existing_instance(self): """Not logged-in user cannot delete existing instance """ # Create instances instance = self.factory() # Query endpoint url = self.endpoint + f'{instance.pk}/' response = self.client.delete(url) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) # Assert instance still exists on db self.assertTrue(self.model.objects.get(id=instance.pk)) def test_anon_user_can_list_instance(self): """Not logged-in user can list instance """ # Request list response = self.client.get(self.endpoint) # Assert access is forbidden self.assertEqual(response.status_code, status.HTTP_200_OK) def test_anon_user_can_filter_name(self): # create instances expected = self.factory(name='product1', tags="zapatos, verdes") unexpected = self.factory(name='product2', tags="rojos") url = f"{self.endpoint}?name=product1" # Request list response = self.client.get(url) payload = response.json() # Assert access is granted self.assertEqual(response.status_code, status.HTTP_200_OK) # Assert number of instnaces in response self.assertEquals(1, len(payload)) self.assertEquals(payload[0]['tags'], expected.tags) def test_anon_user_can_filter_tags(self): # create instances expected_instance = [ self.factory(name='product1', tags="zapatos, rojos"), self.factory(name='product2', tags="rojos"), self.factory(name='product2', tags="colores, rojos") ] unexpected_instance = [ self.factory(name='sadfdsa', tags="zapatos, azules"), self.factory(name='qwerw', tags="xxl") ] # prepare url url = f"{self.endpoint}?tags=rojos" # Request list response = self.client.get(url) payload = response.json() # Assert access is granted self.assertEqual(response.status_code, status.HTTP_200_OK) # Assert number of instnaces in response self.assertEquals(len(expected_instance), len(payload)) def test_anon_user_can_filter_attributes(self): # create instances expected_instance = [ self.factory(attributes='xxl', tags="zapatos, rojos"), self.factory(attributes='blue, xxl', tags="rojos"), self.factory(attributes='sizes xxl', tags="rojos") ] unexpected_instance = [ self.factory(name='sadfdsa', tags="zapatos, azules"), self.factory(name='qwerw', tags="xxl") ] # prepare url url = f"{self.endpoint}?attributes=xxl" # Request list response = self.client.get(url) payload = response.json() # Assert access is granted self.assertEqual(response.status_code, status.HTTP_200_OK) # Assert number of instnaces in response self.assertEquals(len(expected_instance), len(payload)) def test_anon_user_can_filter_category(self): # create instances expected_instance = [ self.factory(category='ropa', tags="zapatos, rojos"), self.factory(category='ropa', tags="rojos"), self.factory(category='ropa', tags="colores/rojos"), ] unexpected_instance = [ self.factory(category='roperos', tags="zapatos, azules"), self.factory(category='enropados', tags="xxl") ] # prepare url url = f"{self.endpoint}?category=ropa" # Request list response = self.client.get(url) payload = response.json() # Assert access is granted self.assertEqual(response.status_code, status.HTTP_200_OK) # Assert number of instnaces in response self.assertEquals(len(expected_instance), len(payload)) def test_anon_user_can_filter_company(self): # create instances company = CompanyFactory() expected_instance = [ self.factory(category='ropa', tags="zapatos, rojos", company=company), self.factory(category='ropa', tags="rojos", company=company), self.factory(category='ropa', tags="colores/rojos", company=company) ] unexpected_instance = [ self.factory(category='roperos', tags="zapatos, azules"), self.factory(category='enropados', tags="xxl") ] # prepare url url = f"{self.endpoint}?company={company.id}" # Request list response = self.client.get(url) payload = response.json() # Assert access is granted self.assertEqual(response.status_code, status.HTTP_200_OK) # Assert number of instnaces in response self.assertEquals(len(expected_instance), len(payload)) def test_anon_can_get_related_products(self): tag = 'cosa' company = CompanyFactory() # Create instances instance = self.factory() # make our user the creator instance.creator = self.user instance.save() instances = [self.factory(tags=tag, company=company) for i in range(10)] url = f"{self.endpoint}{instances[0].id}/related/" response = self.client.get(url) self.assertEquals(response.status_code, 200) payload= response.json() self.assertTrue(len(payload) <= 10) # authenticated user def test_auth_user_can_paginate_instances(self): """authenticated user can paginate instances """ # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # create instances instances = [self.factory(creator=self.user) for n in range(12)] # Request list url = f"{self.endpoint}?limit=5&offset=10" response = self.client.get(url) # Assert access is allowed self.assertEqual(response.status_code, status.HTTP_200_OK) # assert only 2 instances in response payload = response.json() self.assertEquals(2, len(payload['results'])) def test_auth_user_can_list_instances(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_get_instance(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_create_instance(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'])) def test_auth_user_can_modify_own_instance(self): """Regular logged-in user can modify existing instance """ # Create instances instance = self.factory() # make our user the creator instance.creator = self.user instance.save() # Define request data company = CompanyFactory() data = { 'company': company.id, 'sku': 'qwerewq', 'name': 'qwerewq', 'description': 'qwerewq', 'image': None, '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': ['tag1x, tag2x'], 'category': 'mayortagcategory2', 'attributes': ['color/blue', 'size/m'], 'identifiers': '34rf34f43c43', } # 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, data, format='json') # Assert endpoint returns OK code self.assertEqual(response.status_code, status.HTTP_200_OK) # Assert instance has been modified for key in data: if key == 'company': continue self.assertEqual(data[key], response.data[key]) def test_auth_user_cannot_modify_other_users_instance(self): """Regular logged-in user cannot modify other user's instance """ # Create instances instance = self.factory() # 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 url = self.endpoint + f'{instance.pk}/' response = self.client.put(url, data=data, format='json') # Assert endpoint returns OK code self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_auth_user_cannot_delete_other_users_instance(self): """Regular logged-in user cannot delete other user's instance """ # 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, format='json') # Assert endpoint returns OK code self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_auth_user_can_delete_own_instance(self): """Regular logged-in user can delete existing instance """ # Create instances instance = self.factory() # make our user the creator instance.creator = self.user instance.save() # 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()) class LoadCoopProductsTestCase(APITestCase): def setUp(self): """Tests setup """ self.endpoint = '/api/v1/load_products/' self.model = Product self.factory = ActiveProductFactory # create admin user self.admin_email = f"admin_user@mail.com" self.password = ''.join(random.choices(string.ascii_uppercase, k = 10)) self.admin_user = CustomUserFactory(email=self.admin_email, password=self.password, is_staff=True, is_active=True) # create regular user self.reg_email = f"user@mail.com" self.user = CustomUserFactory(email=self.reg_email, is_active=True) self.user.set_password(self.password) self.user.save() # test CSV file path self.csv_path = 'datasets/test_products.csv' def test_auth_user_with_company_can_load_csv(self): # delete existing instances self.model.objects.all().delete() # create company company = CompanyFactory(creator=self.user) # link with user self.user.company = company self.user.save() # read csv file files = {'csv_file': open(self.csv_path,'rt')} # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # send in request response = self.client.post(self.endpoint, files) # check response self.assertEqual(response.status_code, 200) # check for object creation self.assertEquals(9, self.model.objects.count()) def test_auth_user_with_no_company_cannot_load_csv(self): # delete existing instances self.model.objects.all().delete() # read csv file files = {'csv_file': open(self.csv_path,'r')} # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # send in request response = self.client.post(self.endpoint, files) # check response self.assertEqual(response.status_code, 406) # check for object creation self.assertEqual(0, self.model.objects.count()) def test_anon_user_cannot_load_csv(self): # delete existing instances self.model.objects.all().delete() # read csv file files = {'csv_file': open(self.csv_path,'r')} # send in request response = self.client.post(self.endpoint, files) # check response self.assertEqual(response.status_code, 401) # check for object creation self.assertEqual(0, self.model.objects.count()) class ProductSearchTest(TestCase): def setUp(self): """Tests setup """ self.endpoint = '/api/v1/search_products/' self.model = Product self.factory = ActiveProductFactory # create admin user self.admin_email = "admin_user@mail.com" self.password = ''.join(random.choices(string.ascii_uppercase, k = 10)) self.admin_user = CustomUserFactory(email=self.admin_email, password=self.password, is_staff=True, is_active=True) # create regular user self.reg_email = "user@mail.com" self.user = CustomUserFactory(email=self.reg_email, is_active=True) self.user.set_password(self.password) self.user.save() def tearDown(self): self.model.objects.all().delete() def test_anon_user_can_search(self): company = CompanyFactory(company_name='Zapatos Rojos') expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes"), self.factory(tags="colores/rojos, tono/brillante"), self.factory(tags="lunares/azules", description="zapatos rojos"), self.factory(tags="lunares/rojos", description="zapatos"), self.factory(tags="lunares/verdes", company=company), ] unexpected_instances = [ self.factory(description="chanclas"), self.factory(tags="azules"), ] q = quote("zapatos rojos") url = f"{self.endpoint}?q={q}" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() # check for expected fields in payload self.assertIsNotNone(payload.get('filters')) self.assertIsNotNone(payload.get('count')) self.assertIsNotNone(payload.get('products')) self.assertIsNotNone(payload.get('prices')) # check for object creation self.assertEquals(len(payload['products']), len(expected_instances)) # check for filters self.assertTrue(len(payload['filters']['tags']) >= 2 ) # check prices self.assertTrue(payload['prices']['min'] <= payload['prices']['max']) def test_anon_user_can_search_no_querystring(self): expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes"), self.factory(tags="colores/rojos, tono/brillante"), self.factory(tags="lunares/azules", description="zapatos rojos"), self.factory(tags="lunares/rojos", description="zapatos"), self.factory(attributes='"zapatos de campo", tono/oscuro'), ] unexpected_instances = [ self.factory(description="chanclas"), self.factory(tags="azules"), ] q = quote("zapatos rojos") url = f"{self.endpoint}?q=" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() # check for expected fields in payload self.assertIsNotNone(payload.get('filters')) self.assertIsNotNone(payload.get('count')) self.assertIsNotNone(payload.get('products')) self.assertIsNotNone(payload.get('prices')) # check for object creation self.assertEquals(len(payload['products']), len(expected_instances) + len(unexpected_instances)) def test_anon_user_can_paginate_search(self): expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes"), self.factory(tags="colores/rojos, tono/brillante"), self.factory(tags="lunares/azules", description="zapatos rojos"), self.factory(tags="lunares/rojos", description="zapatos"), ] unexpected_instances = [ self.factory(description="chanclas"), self.factory(tags="azules"), ] q = quote("zapatos rojos") # test limit less than available limit = 3 url = f"{self.endpoint}?q={q}&limit={limit}" response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), limit) self.assertEquals(payload['count'], len(expected_instances)) # test limit less than available limit = 10 url = f"{self.endpoint}?q={q}&limit={limit}" response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) self.assertEquals(payload['count'], len(expected_instances)) # test limit equal to available, offset zero limit = len(expected_instances) offset = 0 url = f"{self.endpoint}?q={q}&limit={limit}&offset={offset}" response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) self.assertEquals(payload['count'], len(expected_instances)) # test limit and offset equal to available limit = len(expected_instances) offset = limit url = f"{self.endpoint}?q={q}&limit={limit}&offset={offset}" response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), 0) self.assertEquals(payload['count'], len(expected_instances)) def test_anon_user_can_filter_shipping_cost_true(self): expected_instances = [ self.factory(tags="colores/rojos, tono/brillante", shipping_cost=100.00), self.factory(tags="lunares/azules", description="zapatos rojos", shipping_cost=12.00), ] unexpected_instances = [ self.factory(description="chanclas"), self.factory(tags="azules"), self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes", shipping_cost=None), self.factory(tags="lunares/rojos", description="zapatos", shipping_cost=0.00), ] q = quote("zapatos rojos") # shipping_cost=true url = f"{self.endpoint}?q={q}&shipping_cost=true" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) def test_anon_user_can_filter_shipping_cost_false(self): expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes", shipping_cost=None), self.factory(tags="lunares/rojos", description="zapatos", shipping_cost=0.00), ] unexpected_instances = [ self.factory(description="chanclas", shipping_cost=100.00), self.factory(tags="azules", shipping_cost=10.00), ] q = quote("zapatos rojos") # shipping_cost=false url = f"{self.endpoint}?q={q}&shipping_cost=false" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) def test_anon_user_can_filter_discount_true(self): expected_instances = [ self.factory(tags="colores/rojos, tono/brillante", discount=100.00), self.factory(tags="lunares/azules", description="zapatos rojos", discount=12.00), self.factory(attributes='"zapatos de campo", tono/rojo', discount=9.00), ] unexpected_instances = [ self.factory(description="chanclas", discount=0.00), self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes", discount=None), ] q = quote("zapatos rojos") # discount=true url = f"{self.endpoint}?q={q}&discount=true" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) def test_anon_user_can_filter_discount_false(self): expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes", discount=None), self.factory(tags="lunares/azules", description="zapatos rojos", discount=0.00), self.factory(tags="lunares/rojos", description="zapatos", discount=0.00), ] unexpected_instances = [ self.factory(description="chanclas", discount=100.00), self.factory(tags="azules", discount=9.00), ] q = quote("zapatos rojos") # discount=true url = f"{self.endpoint}?q={q}&discount=false" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) def test_anon_user_can_search_empty_string(self): expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes", discount=None), self.factory(tags="colores/rojos, tono/brillante", discount=100.00), self.factory(tags="lunares/azules", description="zapatos rojos", discount=12.00), self.factory(tags="lunares/rojos", description="zapatos", discount=0.00), self.factory(attributes='"zapatos de campo", tono/oscuro', discount=9.00), self.factory(description="chanclas"), self.factory(tags="azules"), ] # discount=true url = f"{self.endpoint}?q=" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) def test_anon_user_can_filter_by_category(self): expected_instances = [ self.factory(tags="lunares/rojos", category='ropa/nueva', description="zapatos verdes", discount=None), self.factory(tags="lunares/rojos", category="ropa/nueva", discount=0.00), self.factory(attributes='"zapatos de campo", tono/rojo', category="ropa/nueva", discount=9.00), ] unexpected_instances = [ self.factory(description="chanclas", tags='rojos'), self.factory(tags="zapatos/azules", category="deporte", description='rojos', discount=12.00), ] q = quote("zapatos rojos") url = f"{self.endpoint}?q={q}&category=ropa/nueva" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) def test_anon_user_can_filter_by_tags(self): expected_instances = [ self.factory(tags="lunares/rojos, deporte", description="zapatos verdes", discount=None), self.factory(tags="lunares/rojos, deporte", discount=0.00), self.factory(tags="deporte", attributes='"zapatos de campo", tono/rojos', category="ropa", discount=9.00), ] unexpected_instances = [ self.factory(description="chanclas", tags='rojos'), self.factory(tags="zapatos/azules", category="deporte", description='rojos', discount=12.00), ] q = quote("zapatos rojos") # discount=true url = f"{self.endpoint}?q={q}&tags=deporte" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) def test_anon_user_can_filter_by_min_price(self): expected_instances = [ self.factory(tags="colores/rojos, tono/brillante", price=100.00), self.factory(tags="lunares/azules", description="zapatos rojos", price=12.00), ] unexpected_instances = [ self.factory(attributes='ropa/zapatos, tono/oscuro', price=9.00), self.factory(description="zapatos", price=4.00), self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes", price=None), ] price_min = 5.00 q = quote("zapatos rojos") url = f"{self.endpoint}?q={q}&price_min={price_min}" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) for product in payload['products']: self.assertTrue(float(product['price']) >= price_min) def test_anon_user_can_filter_by_max_price(self): expected_instances = [ self.factory(tags="colores/rojos, tono/brillante", price=10.00), self.factory(tags="lunares/azules", description="zapatos rojos", price=12.00), self.factory(attributes='"zapatos de campo", tono/rojos', price=49.00), ] unexpected_instances = [ self.factory(description="zapatos", price=60.00), self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes", price=100.00), ] price_max = 50.00 q = quote("zapatos rojos") url = f"{self.endpoint}?q={q}&price_max={price_max}" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() self.assertEquals(len(payload['products']), len(expected_instances)) for product in payload['products']: self.assertTrue(float(product['price']) <= price_max) def test_anon_user_can_search_newest(self): expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes"), self.factory(tags="colores/rojos, tono/brillante"), self.factory(tags="lunares/azules", description="zapatos rojos"), self.factory(tags="lunares/rojos", description="zapatos"), self.factory(attributes='"zapatos de campo", tono/rojo'), ] unexpected_instances = [ self.factory(description="chanclas"), self.factory(tags="azules"), ] q = quote("zapatos rojos") url = f"{self.endpoint}?q={q}&order=newest" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() # check for object creation self.assertEquals(len(payload['products']), len(expected_instances)) # check results ordered by newest # assert correct order dates = [d['created'][:-1] for d in payload['products']] for i in range(len(dates)-1): # first instance should be most recent self.assertTrue(dates[i] > dates[i+1]) def test_anon_user_can_search_oldest(self): expected_instances = [ self.factory(tags="lunares/rojos", category='zapatos', description="zapatos verdes"), self.factory(tags="colores/rojos, tono/brillante"), self.factory(tags="lunares/azules", description="zapatos rojos"), self.factory(tags="lunares/rojos", description="zapatos"), self.factory(attributes='"zapatos de campo", tono/rojo'), ] unexpected_instances = [ self.factory(description="chanclas"), self.factory(tags="azules"), ] q = quote("zapatos rojos") url = f"{self.endpoint}?q={q}&order=oldest" # send in request response = self.client.get(url) # check response self.assertEqual(response.status_code, 200) # load response data payload = response.json() # check for object creation self.assertEquals(len(payload['products']), len(expected_instances)) # check results ordered by newest # assert correct order dates = [d['created'][:-1] for d in payload['products']] for i in range(len(dates)-1): # first instance should be most recent self.assertTrue(dates[i] < dates[i+1]) class MyProductsViewTest(APITestCase): """my_products tests """ def setUp(self): """Tests setup """ self.endpoint = '/api/v1/my_products/' self.factory = ActiveProductFactory self.model = Product # create user self.email = f"user@mail.com" self.password = ''.join(random.choices(string.ascii_uppercase, k = 10)) self.user = CustomUserFactory(email=self.email, is_active=True) self.user.set_password(self.password) self.user.save() def test_auth_user_gets_data(self): # create instance company = CompanyFactory() self.user.company = company self.user.save() user_instances = [ self.factory(company=company), self.factory(company=company), ] # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # Query endpoint response = self.client.get(self.endpoint) payload = response.json() # Assert forbidden code self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEquals(len(user_instances), len(payload)) def test_auth_user_can_paginate_instances(self): """authenticated user can paginate instances """ # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # create instances company = CompanyFactory() self.user.company = company self.user.save() instances = [self.factory(company=company) for n in range(12)] # Request list url = f"{self.endpoint}?limit=5&offset=10" response = self.client.get(url) # Assert access is allowed self.assertEqual(response.status_code, status.HTTP_200_OK) # assert only 2 instances in response payload = response.json() self.assertEquals(payload['count'], self.model.objects.count()) self.assertEquals(2, len(payload['results'])) def test_anon_user_cannot_access(self): # send in request response = self.client.get(self.endpoint) # check response self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_auth_user_without_company(self): # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # Query endpoint response = self.client.get(self.endpoint) payload = response.json() # Assert forbidden code self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEquals([], payload) class AdminProductViewSetTest(APITestCase): def setUp(self): """Tests setup """ self.endpoint = '/api/v1/admin_products/' self.factory = ProductFactory self.model = Product # create user self.email = f"user@mail.com" self.password = ''.join(random.choices(string.ascii_uppercase, k = 10)) self.user = CustomUserFactory(email=self.email, is_active=True) self.user.set_password(self.password) # self.user.role = 'SITE_ADMIN' self.user.save() def test_anon_user_cannot_access(self): instance = self.factory() url = f"{self.endpoint}{instance.id}/" # GET response = self.client.get(self.endpoint) # check response self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) # POST response = self.client.post(self.endpoint, data={}) # check response self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) # PUT response = self.client.get(url, data={}) # check response self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) # delete response = self.client.get(url) # check response self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_auth_user_cannot_access(self): # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") instance = self.factory() url = f"{self.endpoint}{instance.id}/" # GET response = self.client.get(self.endpoint) # check response self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) # POST response = self.client.post(self.endpoint, data={}) # check response self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) # PUT response = self.client.get(url, data={}) # check response self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) # delete response = self.client.get(url) # check response self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_admin_user_can_list(self): # make user site amdin self.user.role = 'SITE_ADMIN' self.user.save() # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # create instances instance = [self.factory() for i in range(random.randint(1,5))] # query endpoint response = self.client.get(self.endpoint) # assertions self.assertEquals(response.status_code, 200) payload = response.json() self.assertEquals(len(instance), len(payload)) def test_admin_user_can_get_details(self): # make user site amdin self.user.role = 'SITE_ADMIN' self.user.save() # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # create instances instance = self.factory() url = f"{self.endpoint}{instance.id}/" # query endpoint response = self.client.get(url) # assertions self.assertEquals(response.status_code, 200) payload = response.json() self.assertEquals(instance.id, payload['id']) def test_admin_can_create_instance(self): # make user site amdin self.user.role = 'SITE_ADMIN' self.user.save() # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # create instances data = { 'name': 'test_product_name', } # query endpoint response = self.client.post(self.endpoint, data=data) # assertions self.assertEquals(response.status_code, 201) payload = response.json() self.assertEquals(data['name'], payload['name']) def test_admin_can_update_instance(self): # make user site amdin self.user.role = 'SITE_ADMIN' self.user.save() # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # create instance instance = self.factory() url = f"{self.endpoint}{instance.id}/" # data data = { 'name': 'test_product_name', } # query endpoint response = self.client.put(url, data=data) # assertions self.assertEquals(response.status_code, 200) payload = response.json() self.assertEquals(data['name'], payload['name']) def test_admin_can_delete_instance(self): # make user site amdin self.user.role = 'SITE_ADMIN' self.user.save() # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") # create instance instance = self.factory() url = f"{self.endpoint}{instance.id}/" # query endpoint response = self.client.delete(url) # assertions self.assertEquals(response.status_code, 204) class PurchaseEmailTest(APITestCase): def setUp(self): """Tests setup """ self.endpoint = '/api/v1/purchase_email/' self.factory = ProductFactory self.model = Product # create user self.email = f"user@mail.com" self.password = ''.join(random.choices(string.ascii_uppercase, k = 10)) self.user = CustomUserFactory(email=self.email, is_active=True) self.user.set_password(self.password) self.user.save() def test_anon_user_can_use(self): company = CompanyFactory() self.user.role = 'COOP_MANAGER' self.user.company = company self.user.save() product = ProductFactory(company=company) data = { 'email': self.email, 'telephone': '123123123', 'company': company.id, 'product': product.id, 'comment': '', } response = self.client.post(self.endpoint, data=data, format='json') # assertions self.assertEquals(response.status_code, 200) self.assertEquals(2, len(mail.outbox)) def test_auth_user_can_use(self): # required instances company = CompanyFactory() product = ProductFactory(company=company) # make user the manager self.user.company = company self.user.role = 'COOP_MANAGER' self.user.save() data = { 'telephone': '123123123', 'company': company.id, 'product': product.id, 'comment': '', } # Authenticate token = get_tokens_for_user(self.user) self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token['access']}") response = self.client.post(self.endpoint, data=data, format='json') # assertions self.assertEquals(response.status_code, 200) self.assertEquals(2, len(mail.outbox)) def test_anon_user_bad_email(self): company = CompanyFactory() product = ProductFactory(company=company) data = { 'email': '324r@qwer', 'telephone': '123123123', 'company': company.id, 'product': product.id, 'comment': '', } response = self.client.post(self.endpoint, data=data, format='json') # assertions self.assertEquals(response.status_code, 406) payload = response.json() self.assertTrue( 'email' in payload['error']) class AllCategoriesTest(APITestCase): def setUp(self): """Tests setup """ self.endpoint = '/api/v1/products/all_categories/' # self.factory = ProductFactory self.model = CategoryTag # create user self.email = f"user@mail.com" self.password = ''.join(random.choices(string.ascii_uppercase, k = 10)) self.user = CustomUserFactory(email=self.email, is_active=True) self.user.set_password(self.password) self.user.save() def test_get_all_categories(self): # create instances instances = [ self.model.objects.create(name='A'), self.model.objects.create(name='B'), self.model.objects.create(name='C'), self.model.objects.create(name='D'), self.model.objects.create(name='E'), ] response = self.client.get(self.endpoint) # assertions self.assertEquals(response.status_code, 200) payload = response.json() self.assertEquals(len(instances), len(payload))