from datetime import timedelta from io import BytesIO from django.apps import apps from django.contrib import admin from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase from django.urls import reverse from django.utils import timezone from openpyxl import Workbook, load_workbook from apps.shop.import_models import ProductImportBatch from apps.shop.models import Product, ProductCategory from apps.shop.product_import import parse_workbook, import_products, HEADERS from apps.workspace.models import Workspace class ExcelImportTests(TestCase): @classmethod def setUpTestData(cls): cls.workspace = Workspace.objects.create(name='Main', username='main') cls.other = Workspace.objects.create(name='Other', username='other') cls.user = get_user_model().objects.create_user(username='staff', mobile=9123456789, password='secret', is_active=True, is_staff=True, is_superuser=True) def setUp(self): self.client.force_login(self.user) self.options = {'mode': 'upsert', 'create_categories': True, 'errors': 'atomic', 'ordering': 'preserve'} def row(self, code='001', name='Product', category='Parent/Child', **values): return {'row': 2, 'values': dict(acc_code=code, name=name, category=category, **values), 'formulas': []} def upload(self, headers=None, data=None, name='products.xlsx'): workbook = Workbook() sheet = workbook.active sheet.title = 'Products' sheet.append(headers or ['name', 'category', 'acc_code', 'price']) for row in data or [['Product', 'Parent/Child', '001', 1200]]: sheet.append(row) content = BytesIO() workbook.save(content) return SimpleUploadedFile(name, content.getvalue()) def preview(self, **overrides): data = dict(workspace=self.workspace.pk, file=self.upload(), header_row=1, mode='upsert', create_categories='on', errors='atomic', ordering='preserve') data.update(overrides) return self.client.post(reverse('admin:shop_product_import'), data) def confirm(self, batch): return self.client.post(reverse('admin:shop_product_import'), {'confirm': '1', 'batch_id': batch.pk}) def test_parse_aliases_persian_digits_and_text_codes(self): rows = parse_workbook(self.upload(['نام محصول', 'دسته', 'کد محصول', 'قیمت'], [['P', 'C', '0001', '۱٬۲۰۰']])) result = import_products(self.workspace, rows, self.options) self.assertTrue(result['committed']) self.assertEqual(Product.objects.get().acc_code, '0001') self.assertEqual(Product.objects.get().price, 1200) def test_manual_mapping_sheet_and_header_row(self): rows = parse_workbook(self.upload(['Title', 'Group', 'Code']), 'Products', 1, {'name': 'Title', 'category': 'Group', 'acc_code': 'Code'}) self.assertEqual(rows[0]['values']['acc_code'], '001') with self.assertRaises(ValidationError): parse_workbook(self.upload(), 'Absent') def test_duplicate_headers_mapping_and_wrong_file_rejected(self): for file, mapping in ((self.upload(['name', 'name', 'category', 'acc_code']), {}), (self.upload(), {'unknown': 'X'}), (self.upload(name='bad.xls'), {}), (SimpleUploadedFile('bad.xlsx', b'broken'), {})): with self.assertRaises(ValidationError): parse_workbook(file, mapping=mapping) def test_formula_is_row_error(self): rows = parse_workbook(self.upload(data=[['=1+1', 'C', 'X', 1]])) result = import_products(self.workspace, rows, self.options) self.assertEqual(result['errors'], 1) self.assertEqual(Product.objects.count(), 0) def test_dry_run_creates_no_products_or_categories(self): result = import_products(self.workspace, [self.row()], self.options, dry_run=True) self.assertEqual(result['created'], 1) self.assertFalse(result['committed']) self.assertEqual(ProductCategory.objects.count(), 0) self.assertEqual(Product.objects.count(), 0) def test_nested_categories_and_independent_orders(self): result = import_products(self.workspace, [self.row(order_prd='3', excel_order='8', order_cat='6', category_excel_order='9', unit='عدد', view_only='بله', attr='{"color": "blue"}')], self.options) self.assertTrue(result['committed']) product = Product.objects.get() self.assertEqual(product.category.parent.name, 'Parent') self.assertEqual((product.order_prd, product.excel_order, product.category.order_cat, product.category.excel_order), (3, 8, 6, 9)) self.assertEqual(product.attr, {'color': 'blue', 'acc_vahed': 'عدد'}) self.assertTrue(product.view_only) def test_atomic_errors_rollback_everything(self): result = import_products(self.workspace, [self.row(), self.row('002', price='-1')], self.options) self.assertFalse(result['committed']) self.assertEqual(ProductCategory.objects.count(), 0) self.assertEqual(Product.objects.count(), 0) self.assertIn('price', result['rows'][1]['errors']) def test_skip_errors_rollback_categories_of_rejected_row(self): self.options['errors'] = 'skip' result = import_products(self.workspace, [self.row(), self.row('002', category='Rejected', name='X'*201)], self.options) self.assertTrue(result['committed']) self.assertEqual(Product.objects.count(), 1) self.assertFalse(ProductCategory.objects.filter(name='Rejected').exists()) def test_duplicate_code_in_file_and_database_are_errors(self): result = import_products(self.workspace, [self.row(), self.row()], self.options) self.assertEqual(result['errors'], 1) category = ProductCategory.objects.create(sys_ws=self.workspace, name='C') for _ in range(2): Product.objects.create(sys_ws=self.workspace, category=category, name='P', acc_code='001', price=0) result = import_products(self.workspace, [self.row()], self.options) self.assertEqual(result['errors'], 1) def test_matching_is_scoped_to_workspace(self): import_products(self.other, [self.row()], self.options) import_products(self.workspace, [self.row(name='Mine')], self.options) self.assertEqual(Product.objects.count(), 2) self.assertEqual(Product.objects.get(sys_ws=self.other).name, 'Product') def test_update_create_modes_and_preserve_blank(self): import_products(self.workspace, [self.row(price='80', order_prd='4')], self.options) self.options['mode'] = 'create' self.assertEqual(import_products(self.workspace, [self.row()], self.options)['errors'], 1) self.options['mode'] = 'update' self.assertEqual(import_products(self.workspace, [self.row('missing')], self.options)['errors'], 1) result = import_products(self.workspace, [self.row(name='Updated', price='')], self.options) self.assertTrue(result['committed']) self.assertEqual(Product.objects.get().price, 80) self.assertEqual(Product.objects.get().order_prd, 4) def test_category_creation_disabled_and_ambiguous(self): self.options['create_categories'] = False self.assertEqual(import_products(self.workspace, [self.row()], self.options)['errors'], 1) self.options['create_categories'] = True for _ in range(2): ProductCategory.objects.create(sys_ws=self.workspace, name='C') self.assertEqual(import_products(self.workspace, [self.row(category='C')], self.options)['errors'], 1) def test_conflicting_category_orders(self): result = import_products(self.workspace, [self.row(order_cat='1'), self.row('002', order_cat='2')], self.options) self.assertEqual(result['errors'], 1) self.assertFalse(result['committed']) def test_rows_order_and_append(self): self.options['ordering'] = 'rows' import_products(self.workspace, [self.row('1'), self.row('2')], self.options) self.assertEqual(list(Product.objects.order_by('acc_code').values_list('order_prd', flat=True)), [1, 2]) self.options['ordering'] = 'append' import_products(self.workspace, [self.row('3')], self.options) self.assertEqual(Product.objects.get(acc_code='3').order_prd, 3) def test_bad_numbers_booleans_attributes_and_category_path(self): for values in ({'price': '1.5'}, {'price': 'NaN'}, {'view_only': 'maybe'}, {'attr': '[]'}, {'category': 'A//B'}, {'price': '999999999999999999999'}): self.assertEqual(import_products(self.workspace, [self.row(**values)], self.options)['errors'], 1) def test_admin_preview_confirm_replay_and_audit(self): response = self.preview() self.assertEqual(response.status_code, 200) batch = ProductImportBatch.objects.get() self.assertEqual(Product.objects.count(), 0) self.assertContains(response, 'ثبت نهایی') self.assertEqual(self.confirm(batch).status_code, 200) batch.refresh_from_db() self.assertIsNotNone(batch.confirmed_at) self.assertEqual(Product.objects.count(), 1) self.assertContains(self.confirm(batch), 'ثبت نهایی انجام شد') self.assertEqual(Product.objects.count(), 1) from django.contrib.admin.models import LogEntry self.assertEqual(LogEntry.objects.filter(object_repr=f'Excel import #{batch.pk}').count(), 1) def test_admin_expired_preview_rejected(self): self.preview() batch = ProductImportBatch.objects.get() batch.expires_at = timezone.now()-timedelta(seconds=1) batch.save() self.assertContains(self.confirm(batch), 'منقضی') self.assertEqual(Product.objects.count(), 0) def test_admin_changed_catalog_requires_new_preview(self): self.preview() batch = ProductImportBatch.objects.get() ProductCategory.objects.create(sys_ws=self.workspace, name='Changed') self.assertContains(self.confirm(batch), 'تغییر کرده') self.assertEqual(Product.objects.count(), 0) def test_other_admin_cannot_confirm_or_download_batch(self): self.preview() batch = ProductImportBatch.objects.get() other = get_user_model().objects.create_user(username='otherstaff', mobile=9123456790, password='secret', is_active=True, is_staff=True, is_superuser=True) self.client.force_login(other) self.assertEqual(self.confirm(batch).status_code, 404) self.assertEqual(self.client.get(reverse('admin:shop_product_import_report', args=[batch.pk])).status_code, 404) def test_template_and_formula_safe_report(self): response = self.client.get(reverse('admin:shop_product_import_template')) self.assertEqual(response.status_code, 200) sheet = load_workbook(BytesIO(response.content)).active self.assertEqual([cell.value for cell in sheet[1]], list(HEADERS)) self.preview(file=self.upload(data=[['=1+1', 'C', 'X', 1]])) batch = ProductImportBatch.objects.get() response = self.client.get(reverse('admin:shop_product_import_report', args=[batch.pk])) sheet = load_workbook(BytesIO(response.content)).active self.assertEqual(sheet.cell(2, 2).data_type, 's') self.assertEqual(sheet.cell(2, 2).value, '=1+1') def test_nonstaff_and_staff_without_permissions_cannot_import(self): self.user.is_superuser = False self.user.save() self.assertEqual(self.preview().status_code, 403) self.assertEqual(self.client.get(reverse('admin:shop_product_import_template')).status_code, 403) def test_atomic_preview_cannot_be_forced_to_commit_errors(self): self.preview(file=self.upload(data=[['Valid', 'C', '1', 10], ['Invalid', 'D', '2', -1]])) batch = ProductImportBatch.objects.get() self.confirm(batch) batch.refresh_from_db() self.assertIsNone(batch.confirmed_at) self.assertEqual(Product.objects.count(), 0) def test_all_application_models_have_working_admin_lists_and_forms(self): labels = {'shop', 'workspace', 'customer', 'cuser', 'jwt_auth', 'lead', 'ticket', 'customform', 'reminder'} for model in apps.get_models(): if model._meta.app_label not in labels: continue with self.subTest(model=model._meta.label): self.assertIn(model, admin.site._registry) url = reverse(f'admin:{model._meta.app_label}_{model._meta.model_name}_changelist') self.assertEqual(self.client.get(url).status_code, 200) model_admin = admin.site._registry[model] from django.test import RequestFactory request = RequestFactory().get('/') request.user = self.user if model_admin.has_add_permission(request): url = reverse(f'admin:{model._meta.app_label}_{model._meta.model_name}_add') self.assertEqual(self.client.get(url).status_code, 200) def test_cross_workspace_relation_and_category_cycle_rejected(self): category = ProductCategory.objects.create(sys_ws=self.other, name='C') response = self.client.post(reverse('admin:shop_product_add'), {'name': 'P', 'price': '1', 'category': category.pk, 'sys_ws': self.workspace.pk, 'acc_code': '1', 'view_only': '2'}) self.assertContains(response, 'فضای کاری دیگری') self.assertEqual(Product.objects.count(), 0) category.parent = category from utils.admin_base import RelatedWorkspaceForm from django.forms import modelform_factory form = modelform_factory(ProductCategory, form=RelatedWorkspaceForm, fields=['name', 'parent', 'sys_ws'])(data={'name': 'C', 'parent': category.pk, 'sys_ws': self.other.pk}, instance=category) self.assertFalse(form.is_valid()) self.assertIn('parent', form.errors) def test_product_export_can_be_reimported_with_category_path_and_attributes(self): import_products(self.workspace, [self.row(price='70', attr='{"color":"blue"}', view_only='true')], self.options) from apps.shop.admin import export_products response = export_products(Product.objects.all()) rows = parse_workbook(SimpleUploadedFile('export.xlsx', response.content)) self.assertEqual(rows[0]['values']['category'], 'Parent/Child') result = import_products(self.workspace, rows, self.options) self.assertEqual(result['updated'], 1) self.assertEqual(ProductCategory.objects.count(), 2) self.assertEqual(Product.objects.get().attr, {'color': 'blue'}) def test_skip_rejected_category_order_does_not_poison_later_rows(self): self.options['errors'] = 'skip' rows = [self.row('1', name='X'*201, order_cat='1'), self.row('2', order_cat='2')] result = import_products(self.workspace, rows, self.options) self.assertEqual(result['errors'], 1) self.assertEqual(Product.objects.get().category.order_cat, 2) def test_permissions_rechecked_after_preview(self): self.preview() batch = ProductImportBatch.objects.get() self.user.is_superuser = False self.user.save() self.assertEqual(self.confirm(batch).status_code, 403) self.assertEqual(Product.objects.count(), 0) def test_secret_tokens_and_otps_are_not_in_admin_forms_or_exports(self): from apps.jwt_auth.models import JwtToken, CUserRoleLoginOpt from apps.cuser.models import SmsOpt from django.test import RequestFactory request = RequestFactory().get('/') request.user = self.user for model, secrets in ((JwtToken, ('refresh', 'access')), (CUserRoleLoginOpt, ('uuid',)), (SmsOpt, ('opt',))): model_admin = admin.site._registry[model] form = model_admin.get_form(request) self.assertFalse(set(secrets).intersection(form.base_fields)) response = model_admin.export_selected(request, model.objects.none()) headers = list(load_workbook(BytesIO(response.content)).active.values)[0] self.assertFalse(set(secrets).intersection(headers)) from apps.jwt_auth.test_regressions import RegressionFixtures class LegacyProductImportTests(RegressionFixtures): def admin_login(self): from apps.cuser.models import CuserRole 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) def workbook_content(self, data): book = Workbook() sheet = book.active sheet.append(['name', 'cate_name', 'acc_code', 'size', 'vahed']) for row in data: sheet.append(row) output = BytesIO() book.save(output) return output.getvalue() def test_legacy_api_uses_shared_validation_and_rolls_back(self): from unittest.mock import patch self.admin_login() content = self.workbook_content([['P', 'C', '1', 1, 'عدد'], ['Bad', 'D', '2', -1, 'عدد']]) with patch('apps.shop.views.excel_import.open', return_value=BytesIO(content), create=True): response = self.client.get('/shop/import/') self.assertEqual(response.status_code, 400) self.assertEqual(Product.objects.count(), 0) self.assertEqual(ProductCategory.objects.count(), 0) def test_legacy_api_import_still_works_and_scopes_workspace(self): from unittest.mock import patch self.admin_login() content = self.workbook_content([['P', 'C', '0001', 1, 'عدد']]) with patch('apps.shop.views.excel_import.open', return_value=BytesIO(content), create=True): response = self.client.get('/shop/import/') self.assertEqual(response.status_code, 200) self.assertEqual(Product.objects.get().acc_code, '0001') self.assertEqual(Product.objects.get().sys_ws, self.workspace) self.assertTrue(Product.objects.get().view_only) def test_legacy_api_requires_admin_role(self): self.use_role() self.assertEqual(self.client.get('/shop/import/').status_code, 403)