454 lines
26 KiB
Python
454 lines
26 KiB
Python
"""Authentication, duplicate mobiles and ordering integration regressions."""
|
|
from datetime import timedelta
|
|
from io import BytesIO
|
|
from unittest.mock import Mock, patch
|
|
|
|
from django.contrib import admin
|
|
from django.core.exceptions import ValidationError
|
|
from django.db import IntegrityError, transaction
|
|
from django.utils import timezone
|
|
from openpyxl import load_workbook
|
|
from rest_framework.test import APITestCase
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
|
|
from apps.cuser.cuser_model import CUser
|
|
from apps.cuser.models import CuserRole, RoleType, SmsOpt
|
|
from apps.customer.models import Customer, ResponsibleGroup, ShiveAshenaei
|
|
from apps.jwt_auth.models import CUserRoleLoginOpt, JwtToken
|
|
from apps.jwt_auth.session import issue_tokens
|
|
from apps.shop.admin import ProductAdmin, ProductCategoryAdmin, excel_response
|
|
from apps.shop.models import Product, ProductCategory, Order, OrderItem
|
|
from apps.workspace.models import City, Workspace
|
|
from utils.services.sms import DosefrNavadHashtSMS
|
|
from utils.validations.mobile import normalize_mobile
|
|
|
|
|
|
class RegressionFixtures(APITestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.workspace = Workspace.objects.create(name='Test', username='test')
|
|
cls.other_workspace = Workspace.objects.create(name='Other', username='other')
|
|
cls.user = CUser.objects.create_user('client', 'password', mobile=9123456789, name_family='Client', is_active=True)
|
|
cls.customer_type = RoleType.objects.create(role_name='Customer', role_name_en='workspace_customer')
|
|
cls.operator_type = RoleType.objects.create(role_name='Operator', role_name_en='workspace_operator')
|
|
cls.admin_type = RoleType.objects.create(role_name='Admin', role_name_en='workspace_admin')
|
|
cls.role = CuserRole.objects.create(cuser=cls.user, role_type=cls.customer_type, sys_ws=cls.workspace,
|
|
name_family='Client', mobile=cls.user.mobile, is_active=True,
|
|
mobile_verify=True, customer_panel=True)
|
|
cls.operator = CuserRole.objects.create(cuser=cls.user, role_type=cls.operator_type, sys_ws=cls.workspace,
|
|
name_family='Operator', mobile=cls.user.mobile, is_active=True, mobile_verify=True)
|
|
cls.customer = Customer.objects.create(name_family='Client', mobile=cls.user.mobile, cuser_role=cls.role, sys_ws=cls.workspace)
|
|
cls.city = City.objects.create(name='City')
|
|
cls.group = ResponsibleGroup.objects.create(name='Group', sys_ws=cls.workspace)
|
|
cls.shive = ShiveAshenaei.objects.create(name='Web', sys_ws=cls.workspace)
|
|
|
|
def use_role(self, role=None):
|
|
tokens = issue_tokens(role or self.role)
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + tokens['access'])
|
|
return tokens
|
|
|
|
def post(self, path, data):
|
|
return self.client.post('/auth/' + path, data, format='json')
|
|
|
|
def otp(self, expired=False):
|
|
return SmsOpt.objects.create(mobile=self.user.mobile, opt=123456,
|
|
expire=timezone.now() + timedelta(minutes=-1 if expired else 3))
|
|
|
|
|
|
class AuthRegressionTests(RegressionFixtures):
|
|
def test_unknown_mobile_has_useful_400(self):
|
|
response = self.post('login/mobile/', {'mobile': '09999999999'})
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn('ثبت نشده', str(response.data))
|
|
|
|
def test_invalid_mobile_has_400(self):
|
|
self.assertEqual(self.post('login/mobile/', {'mobile': '123'}).status_code, 400)
|
|
|
|
def test_disabled_customer_panel_has_403_and_sends_no_sms(self):
|
|
self.role.customer_panel = False
|
|
self.role.save()
|
|
with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms') as send:
|
|
response = self.post('login/mobile/', {'mobile': self.user.mobile})
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertIn('پنل مشتریان', str(response.data))
|
|
send.assert_not_called()
|
|
|
|
def test_inactive_user_has_403(self):
|
|
self.user.is_active = False
|
|
self.user.save()
|
|
self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 403)
|
|
|
|
def test_sms_failure_has_503_and_preserves_old_otp(self):
|
|
old = self.otp()
|
|
with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=False):
|
|
response = self.post('login/mobile/', {'mobile': self.user.mobile})
|
|
self.assertEqual(response.status_code, 503)
|
|
self.assertIn('ارسال پیامک', str(response.data))
|
|
self.assertTrue(SmsOpt.objects.filter(pk=old.pk).exists())
|
|
|
|
def test_successful_sms_saves_only_one_code(self):
|
|
self.otp()
|
|
with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True):
|
|
response = self.post('login/mobile/', {'mobile': '+98 912-345-6789'})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data['result']['mobile'], self.user.mobile)
|
|
self.assertEqual(SmsOpt.objects.count(), 1)
|
|
self.assertGreaterEqual(SmsOpt.objects.get().opt, 100000)
|
|
|
|
def test_wrong_otp_preserves_valid_code(self):
|
|
old = self.otp()
|
|
self.assertEqual(self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': 654321}).status_code, 400)
|
|
self.assertTrue(SmsOpt.objects.filter(pk=old.pk).exists())
|
|
|
|
def test_expired_otp_rejected(self):
|
|
self.otp(expired=True)
|
|
self.assertEqual(self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': 123456}).status_code, 400)
|
|
|
|
def test_otp_is_single_use_and_roles_have_distinct_options(self):
|
|
CuserRole.objects.create(cuser=self.user, role_type=self.customer_type, sys_ws=self.other_workspace,
|
|
name_family='Other', mobile=self.user.mobile, is_active=True, mobile_verify=True, customer_panel=True)
|
|
self.otp()
|
|
response = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': 123456})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.data), 2)
|
|
self.assertNotEqual(response.data[0]['uuid'], response.data[1]['uuid'])
|
|
self.assertEqual(self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': 123456}).status_code, 400)
|
|
|
|
def test_bad_password_has_400(self):
|
|
self.assertEqual(self.post('login/username/', {'username': 'client', 'password': 'wrong'}).status_code, 400)
|
|
|
|
def test_public_login_ignores_stale_authorization(self):
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer invalid')
|
|
response = self.post('login/username/', {'username': 'client', 'password': 'password'})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.data), 2)
|
|
|
|
def test_option_expiry_and_replay_have_401(self):
|
|
option = CUserRoleLoginOpt.objects.create(cuser=self.user, cuser_role=self.role, expire=timezone.now()-timedelta(seconds=1))
|
|
self.assertEqual(self.post('token/', {'uuid_opt': str(option.uuid)}).status_code, 401)
|
|
option.expire = timezone.now()+timedelta(minutes=5)
|
|
option.save()
|
|
result = self.post('token/', {'uuid_opt': str(option.uuid)})
|
|
self.assertEqual(result.status_code, 200)
|
|
self.assertEqual(self.post('token/', {'uuid_opt': str(option.uuid)}).status_code, 401)
|
|
self.assertEqual(JwtToken.objects.count(), 1)
|
|
|
|
def test_anonymous_info_has_401(self):
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
|
|
def test_signed_but_unrecorded_access_is_rejected(self):
|
|
refresh = RefreshToken.for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + str(refresh.access_token))
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
|
|
def test_blocked_access_and_refresh_are_rejected(self):
|
|
tokens = self.use_role()
|
|
JwtToken.objects.update(is_blocked=True)
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
|
|
def test_missing_session_rejects_access_and_refresh(self):
|
|
tokens = self.use_role()
|
|
JwtToken.objects.all().delete()
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
|
|
def test_refresh_rotates_and_retains_workspace(self):
|
|
tokens = self.use_role(self.operator)
|
|
response = self.post('token/refresh/', {'refresh': tokens['refresh']})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertNotEqual(response.data['refresh'], tokens['refresh'])
|
|
self.assertEqual(RefreshToken(response.data['refresh'])['workspace_uuid'], str(self.workspace.uuid))
|
|
self.assertEqual(JwtToken.objects.count(), 1)
|
|
self.assertGreater(JwtToken.objects.get().expire, timezone.now())
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + response.data['access'])
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 200)
|
|
|
|
def test_disabling_panel_revokes_both_tokens(self):
|
|
tokens = self.use_role()
|
|
self.role.customer_panel = False
|
|
self.role.save()
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
|
|
def test_expired_workspace_revokes_both_tokens(self):
|
|
tokens = self.use_role()
|
|
self.workspace.expire = timezone.now()-timedelta(seconds=1)
|
|
self.workspace.save()
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
|
|
def test_logout_revokes_session_without_valid_access(self):
|
|
tokens = self.use_role()
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer expired')
|
|
self.assertEqual(self.post('logout/', {'refresh': tokens['refresh']}).status_code, 204)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + tokens['access'])
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.assertEqual(self.post('logout/', {'refresh': tokens['refresh']}).status_code, 204)
|
|
|
|
def test_user_duplicate_formats_are_rejected(self):
|
|
for mobile in ['09123456789', '+98 912-345-6789', '۰۹۱۲۳۴۵۶۷۸۹', '00989123456789']:
|
|
with self.subTest(mobile=mobile), self.assertRaises(ValidationError):
|
|
CUser.objects.create_user('duplicate', 'password', mobile=mobile)
|
|
self.assertEqual(CUser.objects.count(), 1)
|
|
|
|
def test_customer_duplicate_has_database_constraint(self):
|
|
with self.assertRaises(IntegrityError), transaction.atomic():
|
|
Customer.objects.create(name_family='Duplicate', mobile=self.user.mobile, sys_ws=self.workspace)
|
|
Customer.objects.create(name_family='Other', mobile=self.user.mobile, sys_ws=self.other_workspace)
|
|
Customer.objects.create(name_family='Null1', sys_ws=self.workspace)
|
|
Customer.objects.create(name_family='Null2', sys_ws=self.workspace)
|
|
|
|
def test_create_customer_v2_duplicate_has_mobile_error(self):
|
|
self.use_role(self.operator)
|
|
response = self.client.post('/customer/v2/create/', {
|
|
'name_family': 'Duplicate', 'mobile': '+98 912-345-6789', 'phone': 2133333333,
|
|
'address': 'Address', 'city': self.city.pk, 'shive_ashenaei': self.shive.pk, 'responsible_group': self.group.pk,
|
|
}, format='json')
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn('mobile', response.data)
|
|
self.assertEqual(Customer.objects.count(), 1)
|
|
|
|
def test_partial_customer_update_rejects_duplicate_and_allows_same_mobile(self):
|
|
other = Customer.objects.create(name_family='Other', mobile=9333333333, sys_ws=self.workspace)
|
|
self.use_role(self.operator)
|
|
response = self.client.patch(f'/customer/update/{other.pk}/', {'mobile': '09123456789'}, format='json')
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn('mobile', response.data)
|
|
response = self.client.patch(f'/customer/update/{self.customer.pk}/', {'mobile': self.user.mobile}, format='json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_panel_link_is_scoped_to_workspace(self):
|
|
other_operator = CuserRole.objects.create(cuser=self.user, role_type=self.operator_type, sys_ws=self.other_workspace,
|
|
name_family='Operator2', mobile=self.user.mobile, is_active=True, mobile_verify=True)
|
|
other_customer = Customer.objects.create(name_family='Other', mobile=self.user.mobile, sys_ws=self.other_workspace)
|
|
self.use_role(other_operator)
|
|
response = self.client.post('/customer/customer-panel/', {'customer_id': other_customer.pk, 'mobile': self.user.mobile, 'status': True}, format='json')
|
|
self.assertEqual(response.status_code, 200)
|
|
other_customer.refresh_from_db()
|
|
self.assertNotEqual(other_customer.cuser_role_id, self.role.pk)
|
|
self.assertEqual(other_customer.cuser_role.sys_ws_id, self.other_workspace.pk)
|
|
|
|
def test_disable_panel_without_link_does_not_create_user(self):
|
|
other = Customer.objects.create(name_family='Other', mobile=9333333333, sys_ws=self.workspace)
|
|
self.use_role(self.operator)
|
|
response = self.client.post('/customer/customer-panel/', {'customer_id': other.pk, 'mobile': other.mobile, 'status': False}, format='json')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIsNone(response.data['cuser_role_id'])
|
|
self.assertEqual(CUser.objects.count(), 1)
|
|
|
|
def test_provider_checks_body_and_handles_timeout(self):
|
|
import requests
|
|
with patch('utils.services.sms.requests.get', return_value=Mock(text='0')) as send:
|
|
self.assertTrue(DosefrNavadHashtSMS().send_sms(str(self.user.mobile), 'Test'))
|
|
self.assertEqual(send.call_args.kwargs['timeout'], 10)
|
|
with patch('utils.services.sms.requests.get', return_value=Mock(text='error')):
|
|
self.assertFalse(DosefrNavadHashtSMS().send_sms(str(self.user.mobile), 'Test'))
|
|
with patch('utils.services.sms.requests.get', side_effect=requests.Timeout):
|
|
self.assertFalse(DosefrNavadHashtSMS().send_sms(str(self.user.mobile), 'Test'))
|
|
|
|
def test_iranian_mobile_prefix_is_not_accidentally_removed(self):
|
|
self.assertEqual(normalize_mobile('9891234567'), 9891234567)
|
|
|
|
def test_logout_during_refresh_rotation_revokes_new_tokens(self):
|
|
tokens = self.use_role()
|
|
rotated = self.post('token/refresh/', {'refresh': tokens['refresh']})
|
|
self.assertEqual(rotated.status_code, 200)
|
|
self.assertEqual(self.post('logout/', {'refresh': tokens['refresh']}).status_code, 204)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': rotated.data['refresh']}).status_code, 401)
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + rotated.data['access'])
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
|
|
def test_mobile_change_disconnects_old_panel_and_revokes_old_session(self):
|
|
tokens = issue_tokens(self.role)
|
|
self.use_role(self.operator)
|
|
response = self.client.patch(f'/customer/update/{self.customer.pk}/', {'mobile': '09333333333'}, format='json')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.customer.refresh_from_db()
|
|
self.role.refresh_from_db()
|
|
self.assertIsNone(self.customer.cuser_role_id)
|
|
self.assertFalse(self.role.customer_panel)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
|
|
def test_duplicate_migration_audits_before_changing_any_data(self):
|
|
import importlib
|
|
from django.apps import apps
|
|
from django.db import connection
|
|
duplicate = Customer.objects.create(name_family='Legacy', mobile=989123456789, sys_ws=self.workspace)
|
|
migration = importlib.import_module('apps.customer.migrations.0008_customer_mobile_unique')
|
|
with self.assertRaises(RuntimeError):
|
|
migration.check_and_normalize_mobiles(apps, Mock(connection=connection))
|
|
duplicate.refresh_from_db()
|
|
self.assertEqual(duplicate.mobile, 989123456789)
|
|
|
|
def test_migration_normalizes_legacy_country_prefix(self):
|
|
import importlib
|
|
from django.apps import apps
|
|
from django.db import connection
|
|
legacy = Customer.objects.create(name_family='Legacy', mobile=989333333333, sys_ws=self.workspace)
|
|
migration = importlib.import_module('apps.customer.migrations.0008_customer_mobile_unique')
|
|
migration.check_and_normalize_mobiles(apps, Mock(connection=connection))
|
|
legacy.refresh_from_db()
|
|
self.assertEqual(legacy.mobile, 9333333333)
|
|
|
|
def test_customer_excel_import_reports_duplicate_and_rolls_back_all_rows(self):
|
|
import pandas as pd
|
|
self.city.ostan = City.objects.create(name='Province', is_ostan=True)
|
|
self.city.save()
|
|
role = CuserRole.objects.create(cuser=self.user, role_type=self.admin_type, sys_ws=self.workspace,
|
|
name_family='Admin', mobile=self.user.mobile, is_active=True, mobile_verify=True)
|
|
self.use_role(role)
|
|
rows = [{'name': 'New', 'mobile': 9333333333, 'phone': None, 'address': None,
|
|
'ostan': 'Province', 'shar': 'City', 'id_mantaghe': self.group.pk},
|
|
{'name': 'Duplicate', 'mobile': self.user.mobile, 'phone': None, 'address': None,
|
|
'ostan': 'Province', 'shar': 'City', 'id_mantaghe': self.group.pk}]
|
|
with patch('apps.shop.views.excel_import.pd.read_excel', return_value=pd.DataFrame(rows)):
|
|
response = self.client.get('/shop/import/customer/')
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn('mobile', response.data['errors'])
|
|
self.assertEqual(Customer.objects.count(), 1)
|
|
|
|
def test_anonymous_import_requests_are_rejected(self):
|
|
self.assertEqual(self.client.get('/shop/import/customer/').status_code, 401)
|
|
self.assertEqual(self.client.get('/shop/import/').status_code, 401)
|
|
|
|
def test_recorded_session_expiry_is_enforced(self):
|
|
tokens = self.use_role()
|
|
JwtToken.objects.update(expire=timezone.now()-timedelta(seconds=1))
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
|
|
def test_inactive_role_and_workspace_reject_access(self):
|
|
self.use_role()
|
|
self.role.is_active = False
|
|
self.role.save()
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.role.is_active = True
|
|
self.role.save()
|
|
self.workspace.is_active = False
|
|
self.workspace.save()
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
|
|
|
|
def test_admin_user_creation_hashes_password_and_normalizes_mobile(self):
|
|
self.user.is_staff = True
|
|
self.user.is_superuser = True
|
|
self.user.save()
|
|
self.client.force_login(self.user)
|
|
response = self.client.post('/admin/cuser/cuser/add/', {
|
|
'username': 'new-user', 'name_family': 'New User', 'mobile': '+98 933-333-3333',
|
|
'password1': 'Valid-Password-9320!', 'password2': 'Valid-Password-9320!',
|
|
'is_active': 'on', 'is_staff': 'on', '_save': 'Save',
|
|
})
|
|
self.assertEqual(response.status_code, 302)
|
|
user = CUser.objects.get(username='new-user')
|
|
self.assertEqual(user.mobile, 9333333333)
|
|
self.assertTrue(user.check_password('Valid-Password-9320!'))
|
|
self.assertNotEqual(user.password, 'Valid-Password-9320!')
|
|
|
|
def test_admin_duplicate_user_mobile_is_a_form_error(self):
|
|
self.user.is_staff = True
|
|
self.user.is_superuser = True
|
|
self.user.save()
|
|
self.client.force_login(self.user)
|
|
response = self.client.post('/admin/cuser/cuser/add/', {
|
|
'username': 'duplicate', 'mobile': '+98 912-345-6789',
|
|
'password1': 'Valid-Password-9320!', 'password2': 'Valid-Password-9320!', '_save': 'Save',
|
|
})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('mobile', response.context['adminform'].form.errors)
|
|
self.assertEqual(CUser.objects.count(), 1)
|
|
|
|
|
|
class OrderingRegressionTests(RegressionFixtures):
|
|
def setUp(self):
|
|
self.first = ProductCategory.objects.create(name='First', sys_ws=self.workspace, order_cat=1, excel_order=20)
|
|
self.second = ProductCategory.objects.create(name='Second', sys_ws=self.workspace, order_cat=2, excel_order=10)
|
|
self.unordered = ProductCategory.objects.create(name='Unordered', sys_ws=self.workspace)
|
|
self.a = Product.objects.create(name='A', category=self.first, sys_ws=self.workspace, price=10, order_prd=2, excel_order=1, attr={'acc_vahed': 'Unit'})
|
|
self.b = Product.objects.create(name='B', category=self.first, sys_ws=self.workspace, price=20, order_prd=1, excel_order=2)
|
|
self.c = Product.objects.create(name='C', category=self.second, sys_ws=self.workspace, price=30, order_prd=1)
|
|
|
|
def test_display_order_groups_categories_and_products(self):
|
|
self.assertEqual(list(ProductCategory.objects.values_list('name', flat=True)), ['First', 'Second', 'Unordered'])
|
|
self.assertEqual(list(Product.objects.values_list('name', flat=True)), ['B', 'A', 'C'])
|
|
|
|
def test_admin_excel_independent_order_and_current_column_order(self):
|
|
model_admin = ProductAdmin(Product, admin.site)
|
|
sheet = load_workbook(BytesIO(model_admin.export_excel_order(None, Product.objects.all()).content)).active
|
|
self.assertEqual([row[2] for row in list(sheet.values)[1:]], ['C', 'A', 'B'])
|
|
sheet = load_workbook(BytesIO(model_admin.export_display_order(None, Product.objects.order_by('-name')).content)).active
|
|
self.assertEqual([row[2] for row in list(sheet.values)[1:]], ['C', 'B', 'A'])
|
|
category_admin = ProductCategoryAdmin(ProductCategory, admin.site)
|
|
sheet = load_workbook(BytesIO(category_admin.export_excel_order(None, ProductCategory.objects.all()).content)).active
|
|
self.assertEqual([row[1] for row in list(sheet.values)[1:]], ['Second', 'First', 'Unordered'])
|
|
|
|
def test_empty_excel_order_falls_back_to_display(self):
|
|
ProductCategory.objects.update(excel_order=None)
|
|
Product.objects.update(excel_order=None)
|
|
response = ProductAdmin(Product, admin.site).export_excel_order(None, Product.objects.all())
|
|
sheet = load_workbook(BytesIO(response.content)).active
|
|
self.assertEqual([row[2] for row in list(sheet.values)[1:]], ['B', 'A', 'C'])
|
|
|
|
def test_excel_keeps_formula_like_names_as_text(self):
|
|
response = excel_response(('Name',), [('=1+1',)], 'test.xlsx')
|
|
cell = load_workbook(BytesIO(response.content)).active['A2']
|
|
self.assertEqual(cell.data_type, 's')
|
|
self.assertEqual(cell.value, '=1+1')
|
|
|
|
def test_category_api_is_ordered_and_does_not_leak_other_workspace(self):
|
|
foreign = ProductCategory.objects.create(name='Foreign', sys_ws=self.other_workspace, order_cat=0)
|
|
Product.objects.create(name='Foreign product', sys_ws=self.other_workspace, category=foreign, price=1)
|
|
self.use_role()
|
|
response = self.client.get('/shop/v2/product-category/list/')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual([row['name'] for row in response.data], ['First', 'Second'])
|
|
response = self.client.get(f'/shop/product/list/{self.first.pk}/')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIsNone(response.data['category_img'])
|
|
self.assertEqual([row['name'] for row in response.data['products']], ['B', 'A'])
|
|
|
|
def test_order_excel_matches_independent_order_and_handles_empty_order(self):
|
|
order = Order.objects.create(customer=self.customer, total_price=60, factor_type=1, sys_ws=self.workspace)
|
|
for product in [self.b, self.a, self.c]:
|
|
OrderItem.objects.create(order=order, product=product, product_name=product.name, price=product.price, amount=1)
|
|
self.use_role(self.operator)
|
|
response = self.client.get(f'/shop/order/{order.pk}/operator/excel/')
|
|
self.assertEqual(response.status_code, 200)
|
|
sheet = load_workbook(BytesIO(response.content)).active
|
|
self.assertEqual([row[0] for row in list(sheet.values)[1:]], ['SecondC', 'FirstA', 'FirstB'])
|
|
OrderItem.objects.all().delete()
|
|
response = self.client.get(f'/shop/order/{order.pk}/operator/excel/')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(load_workbook(BytesIO(response.content)).active.max_row, 1)
|
|
|
|
def test_order_excel_requires_operator_and_matching_workspace(self):
|
|
order = Order.objects.create(customer=self.customer, total_price=0, factor_type=1, sys_ws=self.other_workspace)
|
|
self.use_role()
|
|
self.assertEqual(self.client.get(f'/shop/order/{order.pk}/operator/excel/').status_code, 403)
|
|
self.use_role(self.operator)
|
|
self.assertEqual(self.client.get(f'/shop/order/{order.pk}/operator/excel/').status_code, 400)
|
|
|
|
def test_admin_bulk_edit_category_and_product_orders(self):
|
|
self.user.is_staff = True
|
|
self.user.is_superuser = True
|
|
self.user.save()
|
|
self.client.force_login(self.user)
|
|
for model, field, objects in [('productcategory', 'order_cat', [self.first, self.second, self.unordered]),
|
|
('product', 'order_prd', [self.a, self.b, self.c])]:
|
|
path = f'/admin/shop/{model}/'
|
|
response = self.client.get(path)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = {'_save': 'Save', 'form-TOTAL_FORMS': len(objects), 'form-INITIAL_FORMS': len(objects),
|
|
'form-MIN_NUM_FORMS': 0, 'form-MAX_NUM_FORMS': 1000}
|
|
for index, obj in enumerate(objects):
|
|
data.update({f'form-{index}-id': obj.pk, f'form-{index}-{field}': 10+index,
|
|
f'form-{index}-excel_order': 20-index})
|
|
self.assertEqual(self.client.post(path, data).status_code, 302)
|
|
for index, obj in enumerate(objects):
|
|
obj.refresh_from_db()
|
|
self.assertEqual(getattr(obj, field), 10+index)
|
|
self.assertEqual(obj.excel_order, 20-index)
|