diff --git a/kfet/tests/test_views.py b/kfet/tests/test_views.py index 40f895a1..28599937 100644 --- a/kfet/tests/test_views.py +++ b/kfet/tests/test_views.py @@ -10,12 +10,12 @@ from django.utils import timezone from ..config import kfet_config from ..models import ( - Account, Article, ArticleCategory, Checkout, CheckoutStatement, Inventory, - InventoryArticle, Operation, OperationGroup, Order, OrderArticle, Supplier, - SupplierArticle, Transfer, TransferGroup, + Account, AccountNegative, Article, ArticleCategory, Checkout, + CheckoutStatement, Inventory, InventoryArticle, Operation, OperationGroup, + Order, OrderArticle, Supplier, SupplierArticle, Transfer, TransferGroup, ) from .testcases import ViewTestCaseMixin -from .utils import create_team, create_user, get_perms +from .utils import create_team, create_user, get_perms, user_add_perms class AccountListViewTests(ViewTestCaseMixin, TestCase): @@ -1452,6 +1452,42 @@ class KPsulCheckoutDataViewTests(ViewTestCaseMixin, TestCase): class KPsulPerformOperationsViewTests(ViewTestCaseMixin, TestCase): + """ + Test cases for kpsul_perform_operations view. + + Below is the test ordering, try to keep this organized ;-) + + * OperationGroup: + - test_group... + - test_invalid_group... + * Operation: + - test_purchase... + - test_invalid_purchase... + - test_deposit... + - test_invalid_deposit... + - test_withdraw... + - test_invalid_withdraw... + - test_edit... + - test_invalid_edit... + * Addcost: + - test_addcost... + * Negative: + - test_negative... + - test_invalid_negative... + * More concrete examples: + - test_multi... + + To test valid requests, one should use '_assertResponseOk(response)' to get + hints about failure reasons, if any. + + At least one test per operation type should test the complete response and + behavior (HTTP, WebSocket, object updates, and object creations) + Other tests of the same operation type can only assert the specific + behavior differences. + + For invalid requests, response errors should be tested. + + """ url_name = 'kfet.kpsul.perform_operations' url_expected = '/k-fet/k-psul/perform_operations' @@ -1460,8 +1496,1394 @@ class KPsulPerformOperationsViewTests(ViewTestCaseMixin, TestCase): auth_user = 'team' auth_forbidden = [None, 'user'] - def test_ok(self): - pass + with_liq = True + + def setUp(self): + super(KPsulPerformOperationsViewTests, self).setUp() + + # A Checkout, curently usable, balance=100 + self.checkout = Checkout.objects.create( + created_by=self.accounts["team"], + name="Checkout", + valid_from=timezone.now() - timedelta(days=7), + valid_to=timezone.now() + timedelta(days=7), + balance=Decimal("100.00"), + ) + # An Article, price=2.5, stock=20 + self.article = Article.objects.create( + category=ArticleCategory.objects.create(name="Category"), + name="Article", + price=Decimal("2.5"), + stock=20, + ) + # An Account, trigramme=000, balance=50 + # Do not assume user is cof, nor not cof. + self.account = self.accounts['user'] + self.account.balance = Decimal("50.00") + self.account.save() + + # Mock consumer of K-Psul websocket to catch what we're sending + kpsul_consumer_patcher = mock.patch("kfet.consumers.KPsul") + self.kpsul_consumer_mock = kpsul_consumer_patcher.start() + self.addCleanup(kpsul_consumer_patcher.stop) + + # Reset cache of kfet config + kfet_config._conf_init = False + + def _assertResponseOk(self, response): + """ + Asserts that status code of 'response' is 200, and returns the + deserialized content of the JSONResponse. + + In case status code is not 200, it prints the content of "errors" of + the response. + + """ + json_data = ( + json.loads(getattr(response, "content", b"{}").decode("utf-8")) + ) + try: + self.assertEqual(response.status_code, 200) + except AssertionError as exc: + msg = ( + "Expected response is 200, got {}. Errors: {}" + .format(response.status_code, json_data.get("errors")) + ) + raise AssertionError(msg) from exc + return json_data + + def get_base_post_data(self): + return { + # OperationGroup form + 'on_acc': str(self.account.pk), + 'checkout': str(self.checkout.pk), + # Operation formset + 'form-TOTAL_FORMS': '0', + 'form-INITIAL_FORMS': '0', + 'form-MIN_NUM_FORMS': '1', + 'form-MAX_NUM_FORMS': '1000', + } + + base_post_data = property(get_base_post_data) + + def test_invalid_group_on_acc(self): + data = dict(self.base_post_data, **{"on_acc": "GNR"}) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"]["operation_group"], ["on_acc"]) + + def test_group_on_acc_expects_comment(self): + user_add_perms( + self.users["team"], ["kfet.perform_commented_operations"] + ) + self.account.trigramme = "#13" + self.account.save() + self.assertTrue(self.account.need_comment) + + data = dict(self.base_post_data, **{ + "comment": "A comment to explain it", + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + def test_invalid_group_on_acc_expects_comment(self): + user_add_perms( + self.users["team"], ["kfet.perform_commented_operations"] + ) + self.account.trigramme = "#13" + self.account.save() + self.assertTrue(self.account.need_comment) + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"]["need_comment"], True) + + def test_invalid_group_on_acc_needs_comment_requires_perm(self): + self.account.trigramme = "#13" + self.account.save() + self.assertTrue(self.account.need_comment) + + data = dict(self.base_post_data, **{ + "comment": "A comment to explain it", + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["missing_perms"], + ["Enregistrer des commandes avec commentaires"], + ) + + def test_group_on_acc_frozen(self): + user_add_perms( + self.users["team"], ["kfet.override_frozen_protection"] + ) + self.account.is_frozen = True + self.account.save() + + data = dict(self.base_post_data, **{ + "comment": "A comment to explain it", + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + def test_invalid_group_on_acc_frozen_requires_perm(self): + self.account.is_frozen = True + self.account.save() + + data = dict(self.base_post_data, **{ + "comment": "A comment to explain it", + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["missing_perms"], + ["Forcer le gel d'un compte"], + ) + + def test_invalid_group_checkout(self): + self.checkout.valid_from -= timedelta(days=300) + self.checkout.valid_to -= timedelta(days=300) + self.checkout.save() + + data = dict(self.base_post_data) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"]["operation_group"], ["checkout"]) + + def test_invalid_group_expects_one_operation(self): + data = dict(self.base_post_data) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"]["operations"], []) + + def test_purchase_with_user_is_nof_cof(self): + self.account.cofprofile.is_cof = False + self.account.cofprofile.save() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + # Check response status + json_data = self._assertResponseOk(resp) + + # Check object creations + operation_group = OperationGroup.objects.get() + self.assertDictEqual(operation_group.__dict__, { + "_state": mock.ANY, + "at": mock.ANY, + "amount": Decimal("-5.00"), + "checkout_id": self.checkout.pk, + "comment": "", + "id": mock.ANY, + "is_cof": False, + "on_acc_id": self.account.pk, + "valid_by_id": None, + }) + operation = Operation.objects.get() + self.assertDictEqual(operation.__dict__, { + "_state": mock.ANY, + "addcost_amount": None, + "addcost_for_id": None, + "amount": Decimal("-5.00"), + "article_id": self.article.pk, + "article_nb": 2, + "canceled_at": None, + "canceled_by_id": None, + "group_id": operation_group.pk, + "id": mock.ANY, + "type": "purchase", + }) + + # Check response content + self.assertDictEqual(json_data, { + "operationgroup": operation_group.pk, + "operations": [operation.pk], + "warnings": {}, + "errors": {}, + }) + + # Check object updates + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("45.00")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("100.00")) + self.article.refresh_from_db() + self.assertEqual(self.article.stock, 18) + + # Check websocket data + self.kpsul_consumer_mock.group_send.assert_called_once_with( + "kfet.kpsul", + { + "opegroups": [ + { + "add": True, + "at": mock.ANY, + "amount": Decimal("-5.00"), + "checkout__name": "Checkout", + "comment": "", + "id": operation_group.pk, + "is_cof": False, + "on_acc__trigramme": "000", + "valid_by__trigramme": None, + "opes": [ + { + "id": operation.pk, + "addcost_amount": None, + "addcost_for__trigramme": None, + "amount": Decimal("-5.00"), + "article__name": "Article", + "article_nb": 2, + "canceled_at": None, + "canceled_by__trigramme": None, + "group_id": operation_group.pk, + "type": "purchase", + }, + ], + }, + ], + "checkouts": [ + { + "id": self.checkout.pk, + "balance": Decimal("100.00"), + }, + ], + "articles": [ + { + "id": self.article.pk, + "stock": 18, + }, + ], + }, + ) + + def test_purchase_with_user_is_cof(self): + kfet_config.set(kfet_reduction_cof=Decimal("20")) + self.account.cofprofile.is_cof = True + self.account.cofprofile.save() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertEqual(operation_group.amount, Decimal("-4.00")) + self.assertEqual(operation_group.is_cof, True) + operation = Operation.objects.get() + self.assertEqual(operation.amount, Decimal("-4.00")) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("46.00")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("100.00")) + self.article.refresh_from_db() + self.assertEqual(self.article.stock, 18) + + def test_purchase_with_cash(self): + data = dict(self.base_post_data, **{ + "on_acc": str(self.accounts["liq"].pk), + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertEqual(operation_group.on_acc, self.accounts["liq"]) + self.assertEqual(operation_group.is_cof, False) + + self.accounts["liq"].refresh_from_db() + self.assertEqual(self.accounts["liq"].balance, 0) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("105.00")) + self.article.refresh_from_db() + self.assertEqual(self.article.stock, 18) + + def test_invalid_purchase_expects_article(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": "", + "form-0-article_nb": "1", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], + [ + {"__all__": ["Un achat nécessite un article et une quantité"]}, + ], + ) + + def test_invalid_purchase_expects_article_nb(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], + [ + {"__all__": ["Un achat nécessite un article et une quantité"]}, + ], + ) + + def test_invalid_purchase_expects_article_nb_greater_than_1( + self + ): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "-1", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], + [ + { + "__all__": [ + "Un achat nécessite un article et une quantité", + ], + "article_nb": [ + "Assurez-vous que cette valeur est supérieure ou " + "égale à 1.", + ], + }, + ], + ) + + def test_invalid_operation_not_purchase_with_cash(self): + data = dict(self.base_post_data, **{ + "on_acc": str(self.accounts["liq"].pk), + "form-TOTAL_FORMS": "1", + "form-0-type": "deposit", + "form-0-amount": "10.00", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"]["account"], "LIQ") + + def test_deposit(self): + user_add_perms(self.users["team"], ["kfet.perform_deposit"]) + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "deposit", + "form-0-amount": "10.75", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + json_data = self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertDictEqual(operation_group.__dict__, { + "_state": mock.ANY, + "at": mock.ANY, + "amount": Decimal("10.75"), + "checkout_id": self.checkout.pk, + "comment": "", + "id": mock.ANY, + "is_cof": False, + "on_acc_id": self.account.pk, + "valid_by_id": self.accounts["team"].pk, + }) + operation = Operation.objects.get() + self.assertDictEqual(operation.__dict__, { + "_state": mock.ANY, + "addcost_amount": None, + "addcost_for_id": None, + "amount": Decimal("10.75"), + "article_id": None, + "article_nb": None, + "canceled_at": None, + "canceled_by_id": None, + "group_id": operation_group.pk, + "id": mock.ANY, + "type": "deposit", + }) + + self.assertDictEqual(json_data, { + "operationgroup": operation_group.pk, + "operations": [operation.pk], + "warnings": {}, + "errors": {}, + }) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("60.75")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("110.75")) + + self.kpsul_consumer_mock.group_send.assert_called_once_with( + "kfet.kpsul", + { + "opegroups": [ + { + "add": True, + "at": mock.ANY, + "amount": Decimal("10.75"), + "checkout__name": "Checkout", + "comment": "", + "id": operation_group.pk, + "is_cof": False, + "on_acc__trigramme": "000", + "valid_by__trigramme": "100", + "opes": [ + { + "id": operation.pk, + "addcost_amount": None, + "addcost_for__trigramme": None, + "amount": Decimal("10.75"), + "article__name": None, + "article_nb": None, + "canceled_at": None, + "canceled_by__trigramme": None, + "group_id": operation_group.pk, + "type": "deposit", + }, + ], + }, + ], + "checkouts": [ + { + "id": self.checkout.pk, + "balance": Decimal("110.75"), + }, + ], + "articles": [], + }, + ) + + def test_invalid_deposit_expects_amount(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "deposit", + "form-0-amount": "", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], [{"__all__": ["Bad request"]}] + ) + + def test_invalid_deposit_too_many_params(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "deposit", + "form-0-amount": "10", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "3", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], [{"__all__": ["Bad request"]}] + ) + + def test_invalid_deposit_expects_positive_amount(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "deposit", + "form-0-amount": "-10", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], + [{"__all__": ["Charge non positive"]}] + ) + + def test_invalid_deposit_requires_perm(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "deposit", + "form-0-amount": "10.75", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["missing_perms"], ["Effectuer une charge"] + ) + + def test_withdraw(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "withdraw", + "form-0-amount": "-10.75", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + json_data = self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertDictEqual(operation_group.__dict__, { + "_state": mock.ANY, + "at": mock.ANY, + "amount": Decimal("-10.75"), + "checkout_id": self.checkout.pk, + "comment": "", + "id": mock.ANY, + "is_cof": False, + "on_acc_id": self.account.pk, + "valid_by_id": None, + }) + operation = Operation.objects.get() + self.assertDictEqual(operation.__dict__, { + "_state": mock.ANY, + "addcost_amount": None, + "addcost_for_id": None, + "amount": Decimal("-10.75"), + "article_id": None, + "article_nb": None, + "canceled_at": None, + "canceled_by_id": None, + "group_id": operation_group.pk, + "id": mock.ANY, + "type": "withdraw", + }) + + self.assertDictEqual(json_data, { + "operationgroup": operation_group.pk, + "operations": [operation.pk], + "warnings": {}, + "errors": {}, + }) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("39.25")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("89.25")) + + self.kpsul_consumer_mock.group_send.assert_called_once_with( + "kfet.kpsul", + { + "opegroups": [ + { + "add": True, + "at": mock.ANY, + "amount": Decimal("-10.75"), + "checkout__name": "Checkout", + "comment": "", + "id": operation_group.pk, + "is_cof": False, + "on_acc__trigramme": "000", + "valid_by__trigramme": None, + "opes": [ + { + "id": operation.pk, + "addcost_amount": None, + "addcost_for__trigramme": None, + "amount": Decimal("-10.75"), + "article__name": None, + "article_nb": None, + "canceled_at": None, + "canceled_by__trigramme": None, + "group_id": operation_group.pk, + "type": "withdraw", + }, + ], + }, + ], + "checkouts": [ + { + "id": self.checkout.pk, + "balance": Decimal("89.25"), + }, + ], + "articles": [], + }, + ) + + def test_invalid_withdraw_expects_amount(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "withdraw", + "form-0-amount": "", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], [{"__all__": ["Bad request"]}] + ) + + def test_invalid_withdraw_too_many_params(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "withdraw", + "form-0-amount": "-10", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "3", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], [{"__all__": ["Bad request"]}] + ) + + def test_invalid_withdraw_expects_negative_amount(self): + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "withdraw", + "form-0-amount": "10", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["operations"], + [{"__all__": ["Retrait non négatif"]}] + ) + + def test_edit(self): + user_add_perms(self.users["team"], ["kfet.edit_balance_account"]) + + data = dict(self.base_post_data, **{ + "comment": "A comment to explain it", + "form-TOTAL_FORMS": "1", + "form-0-type": "edit", + "form-0-amount": "10.75", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + json_data = self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertDictEqual(operation_group.__dict__, { + "_state": mock.ANY, + "at": mock.ANY, + "amount": Decimal("10.75"), + "checkout_id": self.checkout.pk, + "comment": "A comment to explain it", + "id": mock.ANY, + "is_cof": False, + "on_acc_id": self.account.pk, + "valid_by_id": self.accounts["team"].pk, + }) + operation = Operation.objects.get() + self.assertDictEqual(operation.__dict__, { + "_state": mock.ANY, + "addcost_amount": None, + "addcost_for_id": None, + "amount": Decimal("10.75"), + "article_id": None, + "article_nb": None, + "canceled_at": None, + "canceled_by_id": None, + "group_id": operation_group.pk, + "id": mock.ANY, + "type": "edit", + }) + + self.assertDictEqual(json_data, { + "operationgroup": operation_group.pk, + "operations": [operation.pk], + "warnings": {}, + "errors": {}, + }) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("60.75")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("100.00")) + + self.kpsul_consumer_mock.group_send.assert_called_once_with( + "kfet.kpsul", + { + "opegroups": [ + { + "add": True, + "at": mock.ANY, + "amount": Decimal("10.75"), + "checkout__name": "Checkout", + "comment": "A comment to explain it", + "id": operation_group.pk, + "is_cof": False, + "on_acc__trigramme": "000", + "valid_by__trigramme": "100", + "opes": [ + { + "id": operation.pk, + "addcost_amount": None, + "addcost_for__trigramme": None, + "amount": Decimal("10.75"), + "article__name": None, + "article_nb": None, + "canceled_at": None, + "canceled_by__trigramme": None, + "group_id": operation_group.pk, + "type": "edit", + }, + ], + }, + ], + "checkouts": [ + { + "id": self.checkout.pk, + "balance": Decimal("100.00"), + }, + ], + "articles": [], + }, + ) + + def test_invalid_edit_requires_perm(self): + data = dict(self.base_post_data, **{ + "comment": "A comment to explain it", + "form-TOTAL_FORMS": "1", + "form-0-type": "edit", + "form-0-amount": "10.75", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"]["missing_perms"], + ["Modifier la balance d'un compte"], + ) + + def test_invalid_edit_expects_comment(self): + user_add_perms(self.users["team"], ["kfet.edit_balance_account"]) + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "edit", + "form-0-amount": "10.75", + "form-0-article": "", + "form-0-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 400) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"]["need_comment"], True) + + def _setup_addcost(self): + self.register_user("addcost", create_user("addcost", "ADD")) + kfet_config.set( + addcost_amount=Decimal("0.50"), + addcost_for=self.accounts["addcost"], + ) + + def test_addcost_user_is_not_cof(self): + self.account.cofprofile.is_cof = False + self.account.cofprofile.save() + self._setup_addcost() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertEqual(operation_group.amount, Decimal("-6.00")) + operation = Operation.objects.get() + self.assertEqual(operation.addcost_for, self.accounts["addcost"]) + self.assertEqual(operation.addcost_amount, Decimal("1.00")) + self.assertEqual(operation.amount, Decimal("-6.00")) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("44.00")) + self.accounts["addcost"].refresh_from_db() + self.assertEqual(self.accounts["addcost"].balance, Decimal("1.00")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("100.00")) + + ws_data_ope = ( + self.kpsul_consumer_mock.group_send + .call_args[0][1]["opegroups"][0]["opes"][0] + ) + self.assertEqual(ws_data_ope["addcost_amount"], Decimal("1.00")) + self.assertEqual(ws_data_ope["addcost_for__trigramme"], "ADD") + + def test_addcost_user_is_cof(self): + kfet_config.set(reduction_cof=Decimal("20")) + self.account.cofprofile.is_cof = True + self.account.cofprofile.save() + self._setup_addcost() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertEqual(operation_group.amount, Decimal("-4.80")) + operation = Operation.objects.get() + self.assertEqual(operation.addcost_for, self.accounts["addcost"]) + self.assertEqual(operation.addcost_amount, Decimal("0.80")) + self.assertEqual(operation.amount, Decimal("-4.80")) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("45.20")) + self.accounts["addcost"].refresh_from_db() + self.assertEqual(self.accounts["addcost"].balance, Decimal("0.80")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("100.00")) + + ws_data_ope = ( + self.kpsul_consumer_mock.group_send + .call_args[0][1]["opegroups"][0]["opes"][0] + ) + self.assertEqual(ws_data_ope["addcost_amount"], Decimal("0.80")) + self.assertEqual(ws_data_ope["addcost_for__trigramme"], "ADD") + + def test_addcost_user_is_cash(self): + self.account.cofprofile.is_cof = True + self.account.cofprofile.save() + self._setup_addcost() + + data = dict(self.base_post_data, **{ + "on_acc": str(self.accounts["liq"].pk), + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertEqual(operation_group.amount, Decimal("-6.00")) + operation = Operation.objects.get() + self.assertEqual(operation.addcost_for, self.accounts["addcost"]) + self.assertEqual(operation.addcost_amount, Decimal("1.00")) + self.assertEqual(operation.amount, Decimal("-6.00")) + + self.accounts["addcost"].refresh_from_db() + self.assertEqual(self.accounts["addcost"].balance, Decimal("1.00")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("106.00")) + + ws_data_ope = ( + self.kpsul_consumer_mock.group_send + .call_args[0][1]["opegroups"][0]["opes"][0] + ) + self.assertEqual(ws_data_ope["addcost_amount"], Decimal("1.00")) + self.assertEqual(ws_data_ope["addcost_for__trigramme"], "ADD") + + def test_addcost_to_self(self): + self._setup_addcost() + self.accounts["addcost"].balance = Decimal("20.00") + self.accounts["addcost"].save() + + data = dict(self.base_post_data, **{ + "on_acc": str(self.accounts["addcost"].pk), + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertEqual(operation_group.amount, Decimal("-5.00")) + operation = Operation.objects.get() + self.assertEqual(operation.addcost_for, None) + self.assertEqual(operation.addcost_amount, None) + self.assertEqual(operation.amount, Decimal("-5.00")) + + self.accounts["addcost"].refresh_from_db() + self.assertEqual(self.accounts["addcost"].balance, Decimal("15.00")) + + ws_data_ope = ( + self.kpsul_consumer_mock.group_send + .call_args[0][1]["opegroups"][0]["opes"][0] + ) + self.assertEqual(ws_data_ope["addcost_amount"], None) + self.assertEqual(ws_data_ope["addcost_for__trigramme"], None) + + def test_addcost_category_disabled(self): + self._setup_addcost() + self.article.category.has_addcost = False + self.article.category.save() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + operation_group = OperationGroup.objects.get() + self.assertEqual(operation_group.amount, Decimal("-5.00")) + operation = Operation.objects.get() + self.assertEqual(operation.addcost_for, None) + self.assertEqual(operation.addcost_amount, None) + self.assertEqual(operation.amount, Decimal("-5.00")) + + self.accounts["addcost"].refresh_from_db() + self.assertEqual(self.accounts["addcost"].balance, Decimal("0.00")) + + ws_data_ope = ( + self.kpsul_consumer_mock.group_send + .call_args[0][1]["opegroups"][0]["opes"][0] + ) + self.assertEqual(ws_data_ope["addcost_amount"], None) + self.assertEqual(ws_data_ope["addcost_for__trigramme"], None) + + def test_negative_new(self): + user_add_perms( + self.users["team"], ["kfet.perform_negative_operations"] + ) + self.account.balance = Decimal("1.00") + self.account.save() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("-4.00")) + + def test_negative_exists(self): + user_add_perms( + self.users["team"], ["kfet.perform_negative_operations"] + ) + self.account.balance = Decimal("-10.00") + self.account.save() + self.account.update_negative() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("-15.00")) + + def test_negative_exists_balance_higher_than_initial(self): + user_add_perms(self.users["team"], ["kfet.perform_deposit"]) + self.account.balance = Decimal("-10.00") + self.account.save() + self.account.update_negative() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "2", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "1", + "form-1-type": "deposit", + "form-1-amount": "5.00", + "form-1-article": "", + "form-1-article_nb": "", + }) + resp = self.client.post(self.url, data) + + self._assertResponseOk(resp) + + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("-7.50")) + + def test_invalid_negative_new_requires_perm(self): + self.account.balance = Decimal("1.00") + self.account.save() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual( + json_data["errors"], + { + "missing_perms": ["Enregistrer des commandes en négatif"], + }, + ) + + def test_invalid_negative_exceeds_allowed_duration_from_config(self): + user_add_perms( + self.users["team"], ["kfet.perform_negative_operations"] + ) + kfet_config.set(overdraft_duration=timedelta(days=5)) + self.account.balance = Decimal("1.00") + self.account.save() + self.account.negative = AccountNegative.objects.create( + account=self.account, + start=timezone.now() - timedelta(days=5, minutes=1), + ) + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"], {"negative": ["000"]}) + + def test_invalid_negative_exceeds_allowed_duration_from_account(self): + user_add_perms( + self.users["team"], ["kfet.perform_negative_operations"] + ) + kfet_config.set(overdraft_duration=timedelta(days=5)) + self.account.balance = Decimal("1.00") + self.account.save() + self.account.negative = AccountNegative.objects.create( + account=self.account, + start=timezone.now() - timedelta(days=3), + authz_overdraft_until=timezone.now() - timedelta(seconds=1), + ) + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"], {"negative": ["000"]}) + + def test_invalid_negative_exceeds_amount_allowed_from_config(self): + user_add_perms( + self.users["team"], ["kfet.perform_negative_operations"] + ) + kfet_config.set(overdraft_amount=Decimal("-1.00")) + self.account.balance = Decimal("1.00") + self.account.save() + self.account.update_negative() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"], {"negative": ["000"]}) + + def test_invalid_negative_exceeds_amount_allowed_from_account(self): + user_add_perms( + self.users["team"], ["kfet.perform_negative_operations"] + ) + kfet_config.set(overdraft_amount=Decimal("10.00")) + self.account.balance = Decimal("1.00") + self.account.save() + self.account.update_negative() + self.account.negative = AccountNegative.objects.create( + account=self.account, + start=timezone.now() - timedelta(days=3), + authz_overdraft_amount=Decimal("1.00"), + ) + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "1", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + }) + resp = self.client.post(self.url, data) + + self.assertEqual(resp.status_code, 403) + json_data = json.loads(resp.content.decode("utf-8")) + self.assertEqual(json_data["errors"], {"negative": ["000"]}) + + def test_multi_0(self): + article2 = Article.objects.create( + name="Article 2", + price=Decimal("4"), + stock=-5, + category=ArticleCategory.objects.first(), + ) + self.account.cofprofile.is_cof = False + self.account.cofprofile.save() + + data = dict(self.base_post_data, **{ + "form-TOTAL_FORMS": "2", + "form-0-type": "purchase", + "form-0-amount": "", + "form-0-article": str(self.article.pk), + "form-0-article_nb": "2", + "form-1-type": "purchase", + "form-1-amount": "", + "form-1-article": str(article2.pk), + "form-1-article_nb": "1", + }) + resp = self.client.post(self.url, data) + + # Check response status + json_data = self._assertResponseOk(resp) + + # Check object creations + operation_group = OperationGroup.objects.get() + self.assertDictEqual(operation_group.__dict__, { + "_state": mock.ANY, + "at": mock.ANY, + "amount": Decimal("-9.00"), + "checkout_id": self.checkout.pk, + "comment": "", + "id": mock.ANY, + "is_cof": False, + "on_acc_id": self.account.pk, + "valid_by_id": None, + }) + operation_list = Operation.objects.all() + self.assertEqual(len(operation_list), 2) + self.assertDictEqual(operation_list[0].__dict__, { + "_state": mock.ANY, + "addcost_amount": None, + "addcost_for_id": None, + "amount": Decimal("-5.00"), + "article_id": self.article.pk, + "article_nb": 2, + "canceled_at": None, + "canceled_by_id": None, + "group_id": operation_group.pk, + "id": mock.ANY, + "type": "purchase", + }) + self.assertDictEqual(operation_list[1].__dict__, { + "_state": mock.ANY, + "addcost_amount": None, + "addcost_for_id": None, + "amount": Decimal("-4.00"), + "article_id": article2.pk, + "article_nb": 1, + "canceled_at": None, + "canceled_by_id": None, + "group_id": operation_group.pk, + "id": mock.ANY, + "type": "purchase", + }) + + # Check response content + self.assertDictEqual(json_data, { + "operationgroup": operation_group.pk, + "operations": [operation_list[0].pk, operation_list[1].pk], + "warnings": {}, + "errors": {}, + }) + + # Check object updates + self.account.refresh_from_db() + self.assertEqual(self.account.balance, Decimal("41.00")) + self.checkout.refresh_from_db() + self.assertEqual(self.checkout.balance, Decimal("100.00")) + self.article.refresh_from_db() + self.assertEqual(self.article.stock, 18) + article2.refresh_from_db() + self.assertEqual(article2.stock, -6) + + # Check websocket data + self.kpsul_consumer_mock.group_send.assert_called_once_with( + "kfet.kpsul", + { + "opegroups": [ + { + "add": True, + "at": mock.ANY, + "amount": Decimal("-9.00"), + "checkout__name": "Checkout", + "comment": "", + "id": operation_group.pk, + "is_cof": False, + "on_acc__trigramme": "000", + "valid_by__trigramme": None, + "opes": [ + { + "id": operation_list[0].pk, + "addcost_amount": None, + "addcost_for__trigramme": None, + "amount": Decimal("-5.00"), + "article__name": "Article", + "article_nb": 2, + "canceled_at": None, + "canceled_by__trigramme": None, + "group_id": operation_group.pk, + "type": "purchase", + }, + { + "id": operation_list[1].pk, + "addcost_amount": None, + "addcost_for__trigramme": None, + "amount": Decimal("-4.00"), + "article__name": "Article 2", + "article_nb": 1, + "canceled_at": None, + "canceled_by__trigramme": None, + "group_id": operation_group.pk, + "type": "purchase", + }, + ], + }, + ], + "checkouts": [ + { + "id": self.checkout.pk, + "balance": Decimal("100.00"), + }, + ], + "articles": [ + { + "id": self.article.pk, + "stock": 18, + }, + { + "id": article2.pk, + "stock": -6, + }, + ], + }, + ) class KPsulCancelOperationsViewTests(ViewTestCaseMixin, TestCase): diff --git a/kfet/tests/testcases.py b/kfet/tests/testcases.py index aa2fb1b6..3a69e9ca 100644 --- a/kfet/tests/testcases.py +++ b/kfet/tests/testcases.py @@ -178,7 +178,9 @@ class ViewTestCaseMixin(TestCaseMixin): During setup, three users are created with their kfet account: - 'user': a basic user without any permission, account trigramme: 000, - 'team': a user with kfet.is_team permission, account trigramme: 100, - - 'root': a superuser, account trigramme: 200. + - 'root': a superuser, account trigramme: 200, + - 'liq': if class attribute 'with_liq' is 'True', account + trigramme: LIQ. Their password is their username. One can create additionnal users with 'get_users_extra' method, or prevent @@ -221,6 +223,8 @@ class ViewTestCaseMixin(TestCaseMixin): auth_user = None auth_forbidden = [] + with_liq = False + def setUp(self): """ Warning: Do not forget to call super().setUp() in subclasses. @@ -262,7 +266,7 @@ class ViewTestCaseMixin(TestCaseMixin): """ # Format desc: username, password, trigramme - return { + users_base = { # user, user, 000 'user': create_user(), # team, team, 100 @@ -270,6 +274,9 @@ class ViewTestCaseMixin(TestCaseMixin): # root, root, 200 'root': create_root(), } + if self.with_liq: + users_base['liq'] = create_user('liq', 'LIQ') + return users_base @cached_property def users_base(self):