from datetime import timedelta from io import BytesIO from django.core.files.uploadedfile import SimpleUploadedFile from django.utils import timezone from openpyxl import Workbook, load_workbook from apps.jwt_auth.test_regressions import RegressionFixtures from apps.cuser.models import CuserRole, RoleType from apps.shop.models import Product, ProductCategory, Order, OrderItem, Cart from apps.shop.import_models import ProductImportBatch ROOT = '/shop/manage/' class ProductManagementTests(RegressionFixtures): @classmethod def setUpTestData(cls): super().setUpTestData() cls.admin_role = CuserRole.objects.create(cuser=cls.user, role_type=cls.admin_type, sys_ws=cls.workspace, name_family='Admin', mobile=cls.user.mobile, is_active=True, mobile_verify=True) cls.category = ProductCategory.objects.create(sys_ws=cls.workspace, name='C', order_cat=1) cls.foreign_category = ProductCategory.objects.create(sys_ws=cls.other_workspace, name='Foreign') cls.product = Product.objects.create(sys_ws=cls.workspace, category=cls.category, name='P', acc_code='P1', price=10) cls.foreign_product = Product.objects.create(sys_ws=cls.other_workspace, category=cls.foreign_category, name='Foreign', acc_code='F1', price=20) def setUp(self): self.use_role(self.admin_role) def payload(self, **values): return dict(name='New product', price='100', category=self.category.pk, acc_code='NEW', **values) def patch(self, path, data): return self.client.patch(ROOT+path, data, format='json') def upload(self, rows=None): book = Workbook() sheet = book.active sheet.append(['name', 'category', 'acc_code', 'price']) for row in rows or [['Imported', 'Root/Child', 'EXCEL1', 50]]: sheet.append(row) output = BytesIO() book.save(output) return SimpleUploadedFile('products.xlsx', output.getvalue()) def preview(self, **values): data = dict(file=self.upload(), mode='upsert', create_categories='true', errors='atomic', ordering='preserve') data.update(values) return self.client.post(ROOT+'import/preview/', data, format='multipart') def test_operator_customer_even_flagged_admin_denied_every_management_endpoint(self): paths = [('get', 'products/'), ('post', 'products/'), ('get', f'products/{self.product.pk}/'), ('patch', f'products/{self.product.pk}/'), ('delete', f'products/{self.product.pk}/'), ('post', 'products/bulk/'), ('get', 'products/export/'), ('get', 'categories/'), ('post', 'categories/'), ('patch', f'categories/{self.category.pk}/'), ('delete', f'categories/{self.category.pk}/'), ('post', 'import/preview/'), ('get', 'import/template/'), ('get', 'imports/'), ('get', 'imports/999/'), ('post', 'imports/999/confirm/'), ('get', 'imports/999/report/')] for role in (self.role, self.operator): role.is_admin = True role.save() self.use_role(role) for method, path in paths: with self.subTest(role=role.role_type.role_name_en, method=method, path=path): response = getattr(self.client, method)(ROOT+path) self.assertEqual(response.status_code, 403) def test_unauthenticated_and_revoked_session_cannot_manage(self): self.client.credentials() self.assertEqual(self.client.get(ROOT+'products/').status_code, 401) self.use_role(self.admin_role) from apps.jwt_auth.models import JwtToken JwtToken.objects.update(is_blocked=True) self.assertEqual(self.client.get(ROOT+'products/').status_code, 401) def test_capabilities_match_current_role_and_owner(self): self.assertTrue(self.client.get('/cuser/info/').data['permissions']['manage_products']) self.use_role(self.operator) self.assertFalse(self.client.get('/cuser/info/').data['permissions']['manage_products']) owner_type = RoleType.objects.create(role_name='Owner', role_name_en='workspace_owner') owner = CuserRole.objects.create(cuser=self.user, role_type=owner_type, sys_ws=self.workspace, name_family='Owner', mobile=self.user.mobile, is_active=True, mobile_verify=True) self.use_role(owner) self.assertEqual(self.client.get(ROOT+'products/').status_code, 200) def test_product_list_paginated_filtered_and_workspace_scoped(self): result = self.client.get(ROOT+'products/').data self.assertEqual(result['count'], 1) self.assertEqual(result['results'][0]['id'], self.product.pk) self.assertEqual(self.client.get(ROOT+'products/?search=Foreign').data['count'], 0) self.assertEqual(self.client.get(ROOT+'products/?search=P1&price_min=10&price_max=10').data['count'], 1) self.assertEqual(self.client.get(ROOT+'products/?view_only=true').data['count'], 0) self.assertEqual(self.client.get(ROOT+'products/?price_min=20&price_max=1').status_code, 400) self.assertEqual(self.client.get(ROOT+'products/?ordering=unknown').status_code, 400) def test_categories_include_empty_and_counts(self): empty = ProductCategory.objects.create(sys_ws=self.workspace, name='Empty', parent=self.category) response = self.client.get(ROOT+'categories/') self.assertEqual(response.status_code, 200) rows = {row['id']: row for row in response.data} self.assertNotIn(self.foreign_category.pk, rows) self.assertEqual(rows[self.category.pk]['product_count'], 1) self.assertEqual(rows[self.category.pk]['child_count'], 1) self.assertEqual(rows[empty.pk]['product_count'], 0) def test_product_crud_attributes_orders_and_big_price_precision(self): data = self.payload() data.update(name='N'*150, price='9223372036854775807', attr={'unit': 'عدد'}, order_prd=3, excel_order=7) response = self.client.post(ROOT+'products/', data, format='json') self.assertEqual(response.status_code, 201, response.data) self.assertEqual(response.data['price'], '9223372036854775807') product_id = response.data['id'] response = self.patch(f'products/{product_id}/', {'price': '0', 'order_prd': None, 'attr': None, 'view_only': True}) self.assertEqual(response.status_code, 200) self.assertIsNone(response.data['order_prd']) self.assertIsNone(response.data['attr']) self.assertTrue(response.data['view_only']) self.assertEqual(self.client.delete(ROOT+f'products/{product_id}/').status_code, 204) def test_product_duplicate_code_invalid_values_and_foreign_category(self): for values in ({'acc_code': 'P1'}, {'category': self.foreign_category.pk}, {'price': -1}, {'price': '1.5'}, {'price': '9223372036854775808'}, {'attr': []}, {'order_prd': -1}): data = self.payload() data.update(values) response = self.client.post(ROOT+'products/', data, format='json') self.assertEqual(response.status_code, 400, response.data) self.assertEqual(Product.objects.count(), 2) def test_foreign_records_not_read_changed_or_deleted(self): for path in (f'products/{self.foreign_product.pk}/', f'categories/{self.foreign_category.pk}/'): self.assertEqual(self.client.get(ROOT+path).status_code, 404) self.assertEqual(self.patch(path, {'name': 'Stolen'}).status_code, 404) self.assertEqual(self.client.delete(ROOT+path).status_code, 404) self.foreign_product.refresh_from_db() self.assertEqual(self.foreign_product.name, 'Foreign') def test_category_crud_duplicate_and_foreign_parent(self): response = self.client.post(ROOT+'categories/', {'name': 'New category', 'parent': self.category.pk, 'order_cat': 2, 'excel_order': 9}, format='json') self.assertEqual(response.status_code, 201) pk = response.data['id'] self.assertEqual(self.client.post(ROOT+'categories/', {'name': 'New category', 'parent': self.category.pk}, format='json').status_code, 400) self.assertEqual(self.patch(f'categories/{pk}/', {'parent': self.foreign_category.pk}).status_code, 400) response = self.patch(f'categories/{pk}/', {'name': 'Renamed', 'parent': None, 'order_cat': None}) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['name'], 'Renamed') self.assertEqual(self.client.delete(ROOT+f'categories/{pk}/').status_code, 204) def test_category_cycles_and_nonempty_deletion_blocked(self): child = ProductCategory.objects.create(sys_ws=self.workspace, name='Child', parent=self.category) self.assertEqual(self.patch(f'categories/{self.category.pk}/', {'parent': child.pk}).status_code, 400) self.assertEqual(self.patch(f'categories/{child.pk}/', {'parent': child.pk}).status_code, 400) self.assertEqual(self.client.delete(ROOT+f'categories/{self.category.pk}/').status_code, 409) self.assertTrue(Product.objects.filter(pk=self.product.pk).exists()) def test_product_with_order_or_cart_cannot_delete(self): order = Order.objects.create(sys_ws=self.workspace, customer=self.customer, total_price=10, factor_type=1) OrderItem.objects.create(order=order, product=self.product, product_name='P', price=10, amount=1) self.assertEqual(self.client.delete(ROOT+f'products/{self.product.pk}/').status_code, 409) self.assertTrue(OrderItem.objects.filter(order=order).exists()) OrderItem.objects.all().delete() Cart.objects.create(sys_ws=self.workspace, product=self.product, cuser_role=self.role, quantity=1) self.assertEqual(self.client.delete(ROOT+f'products/{self.product.pk}/').status_code, 409) def test_bulk_move_toggle_and_reorder(self): second = Product.objects.create(sys_ws=self.workspace, category=self.category, name='Second', price=0) target = ProductCategory.objects.create(sys_ws=self.workspace, name='Target') for action, fields in (('move', {'category': target.pk}), ('view_only', {'view_only': True}), ('excel_order', {})): response = self.client.post(ROOT+'products/bulk/', dict(ids=[second.pk, self.product.pk], action=action, **fields), format='json') self.assertEqual(response.status_code, 200, response.data) self.product.refresh_from_db() self.assertEqual(self.product.category, target) self.assertTrue(self.product.view_only) self.assertEqual(self.product.excel_order, 2) def test_bulk_foreign_or_duplicate_ids_roll_back(self): for ids in ([self.product.pk, self.foreign_product.pk], [self.product.pk, self.product.pk]): response = self.client.post(ROOT+'products/bulk/', {'ids': ids, 'action': 'view_only', 'view_only': True}, format='json') self.assertEqual(response.status_code, 400) self.product.refresh_from_db() self.assertFalse(self.product.view_only) def test_bulk_deletion_all_or_nothing(self): removable = Product.objects.create(sys_ws=self.workspace, category=self.category, name='Removable', price=0) Cart.objects.create(sys_ws=self.workspace, product=self.product, cuser_role=self.role, quantity=1) response = self.client.post(ROOT+'products/bulk/', {'ids': [removable.pk, self.product.pk], 'action': 'delete'}, format='json') self.assertEqual(response.status_code, 409) self.assertTrue(Product.objects.filter(pk=removable.pk).exists()) def test_legacy_category_update_fixed_and_cross_workspace_product_rejected(self): response = self.client.put(f'/shop/product-category/update/{self.category.pk}/', {'name': 'Fixed'}, format='json') self.assertEqual(response.status_code, 200, response.data) self.assertEqual(response.data['name'], 'Fixed') response = self.client.post('/shop/product/create/', {'name': 'X', 'price': 1, 'category': self.foreign_category.pk}, format='json') self.assertEqual(response.status_code, 400) self.assertEqual(self.client.delete(f'/shop/product-category/delete/{self.category.pk}/').status_code, 409) def test_excel_template_export_is_workspace_scoped_and_formula_safe(self): self.product.name = '=1+1' self.product.save() response = self.client.get(ROOT+'products/export/?search=P1') self.assertEqual(response.status_code, 200) sheet = load_workbook(BytesIO(response.content)).active self.assertEqual(sheet.cell(2, 3).data_type, 's') self.assertEqual(sheet.max_row, 2) self.assertEqual(self.client.get(ROOT+'import/template/').status_code, 200) def test_import_preview_confirm_replay_and_report(self): response = self.preview() self.assertEqual(response.status_code, 201, response.data) batch_id = response.data['id'] self.assertFalse(response.data['committed']) self.assertFalse(Product.objects.filter(acc_code='EXCEL1').exists()) self.assertEqual(self.client.get(ROOT+'imports/').data['count'], 1) self.assertEqual(self.client.get(ROOT+f'imports/{batch_id}/report/').status_code, 200) for _ in range(2): response = self.client.post(ROOT+f'imports/{batch_id}/confirm/', {}, format='json') self.assertEqual(response.status_code, 200, response.data) self.assertTrue(response.data['committed']) self.assertEqual(Product.objects.filter(acc_code='EXCEL1').count(), 1) def test_import_atomic_and_skip_modes(self): bad = [['Good', 'Valid', 'GOOD', 1], ['Bad', 'Invalid', 'BAD', -1]] response = self.preview(file=self.upload(bad)) self.assertEqual(response.data['errors'], 1) self.assertFalse(response.data['can_confirm']) batch_id = response.data['id'] self.client.post(ROOT+f'imports/{batch_id}/confirm/') self.assertFalse(Product.objects.filter(acc_code='GOOD').exists()) response = self.preview(file=self.upload(bad), errors='skip') batch_id = response.data['id'] result = self.client.post(ROOT+f'imports/{batch_id}/confirm/').data self.assertTrue(result['committed']) self.assertEqual(result['errors'], 1) self.assertTrue(Product.objects.filter(acc_code='GOOD').exists()) self.assertFalse(ProductCategory.objects.filter(name='Invalid').exists()) def test_import_wrong_options_mapping_or_file_returns_error(self): for values in ({'mode': 'unknown'}, {'mapping': '[]'}, {'header_row': 0}, {'file': SimpleUploadedFile('bad.xlsx', b'bad')}): response = self.preview(**values) self.assertEqual(response.status_code, 400, response.data) def test_import_expired_or_changed_catalog_cannot_confirm(self): for expired in (True, False): batch_id = self.preview().data['id'] if expired: ProductImportBatch.objects.filter(pk=batch_id).update(expires_at=timezone.now()-timedelta(seconds=1)) else: self.product.name = 'Changed' self.product.save() self.assertEqual(self.client.post(ROOT+f'imports/{batch_id}/confirm/').status_code, 400) self.assertFalse(Product.objects.filter(acc_code='EXCEL1').exists()) def test_import_other_workspace_and_other_owner_hidden(self): batch_id = self.preview().data['id'] self.admin_role.sys_ws = self.other_workspace self.admin_role.save() self.use_role(self.admin_role) for suffix in ('', 'report/', 'confirm/'): method = self.client.post if suffix == 'confirm/' else self.client.get self.assertEqual(method(ROOT+f'imports/{batch_id}/'+suffix).status_code, 404) self.assertEqual(self.client.get(ROOT+'imports/').data['count'], 0) def test_import_rows_paginate_without_losing_report_rows(self): rows = [[f'P{i}', 'C', f'CODE{i}', 1] for i in range(101)] response = self.preview(file=self.upload(rows)) self.assertEqual(len(response.data['rows']), 100) self.assertEqual(response.data['row_count'], 101) batch_id = response.data['id'] self.assertEqual(len(self.client.get(ROOT+f'imports/{batch_id}/?row_page=2').data['rows']), 1) self.assertEqual(self.client.get(ROOT+f'imports/{batch_id}/?row_page=0').status_code, 400) def test_category_multipart_image_and_clearing_nullable_fields(self): from tempfile import TemporaryDirectory from django.test import override_settings from PIL import Image content = BytesIO() Image.new('RGB', (2, 2), color='blue').save(content, format='PNG') with TemporaryDirectory(dir='../../reports/product-management') as directory, override_settings(MEDIA_ROOT=directory): upload = SimpleUploadedFile('category.png', content.getvalue(), content_type='image/png') response = self.client.post(ROOT+'categories/', {'name': 'Image', 'parent': '', 'order_cat': '', 'excel_order': '', 'image': upload}, format='multipart') self.assertEqual(response.status_code, 201, response.data) self.assertIsNone(response.data['order_cat']) pk = response.data['id'] self.assertTrue(response.data['image'].endswith('.png')) response = self.patch(f'categories/{pk}/', {'image': None, 'order_cat': 4}) self.assertEqual(response.status_code, 200) self.assertIsNone(response.data['image']) def test_category_negative_and_out_of_range_orders_are_rejected(self): for values in ({'order_cat': -1}, {'excel_order': -1}, {'order_cat': 2147483648}): data = {'name': 'Invalid'} data.update(values) self.assertEqual(self.client.post(ROOT+'categories/', data, format='json').status_code, 400)