forked from DGNum/gestioCOF
Merge branch 'aureplop/kfet-tests_cancel_operations' into 'master'
kfet.tests -- Add tests for cancel_operations view + small things See merge request klub-dev-ens/gestioCOF!315
This commit is contained in:
Martin Pepin
commit
e21154e869
|
|
@ -1,3 +1,6 @@
|
|||
from decimal import Decimal
|
||||
from unittest import mock
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Permission
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
|
|
@ -5,9 +8,16 @@ from django.test import TestCase
|
|||
|
||||
from gestioncof.models import CofProfile
|
||||
|
||||
from ..models import Account
|
||||
from ..models import Account, Article, ArticleCategory, Checkout, Operation
|
||||
from .testcases import TestCaseMixin
|
||||
from .utils import create_root, create_team, create_user, get_perms, user_add_perms
|
||||
from .utils import (
|
||||
create_operation_group,
|
||||
create_root,
|
||||
create_team,
|
||||
create_user,
|
||||
get_perms,
|
||||
user_add_perms,
|
||||
)
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
|
@ -86,3 +96,80 @@ class PermHelpersTest(TestCaseMixin, TestCase):
|
|||
map(repr, [self.perm1, self.perm2, self.perm_team]),
|
||||
ordered=False,
|
||||
)
|
||||
|
||||
|
||||
class OperationHelpersTest(TestCase):
|
||||
def test_create_operation_group(self):
|
||||
operation_group = create_operation_group()
|
||||
|
||||
on_acc = Account.objects.get(cofprofile__user__username="user")
|
||||
checkout = Checkout.objects.get(name="Checkout")
|
||||
self.assertDictEqual(
|
||||
operation_group.__dict__,
|
||||
{
|
||||
"_checkout_cache": checkout,
|
||||
"_on_acc_cache": on_acc,
|
||||
"_state": mock.ANY,
|
||||
"amount": 0,
|
||||
"at": mock.ANY,
|
||||
"checkout_id": checkout.pk,
|
||||
"comment": "",
|
||||
"id": mock.ANY,
|
||||
"is_cof": False,
|
||||
"on_acc_id": on_acc.pk,
|
||||
"valid_by_id": None,
|
||||
},
|
||||
)
|
||||
self.assertFalse(operation_group.opes.all())
|
||||
|
||||
def test_create_operation_group_with_content(self):
|
||||
article_category = ArticleCategory.objects.create(name="Category")
|
||||
article1 = Article.objects.create(
|
||||
category=article_category, name="Article 1", price=Decimal("2.50")
|
||||
)
|
||||
article2 = Article.objects.create(
|
||||
category=article_category, name="Article 2", price=Decimal("4.00")
|
||||
)
|
||||
operation_group = create_operation_group(
|
||||
content=[
|
||||
{
|
||||
"type": Operation.PURCHASE,
|
||||
"amount": Decimal("-3.50"),
|
||||
"article": article1,
|
||||
"article_nb": 2,
|
||||
},
|
||||
{"type": Operation.PURCHASE, "article": article2, "article_nb": 2},
|
||||
{"type": Operation.PURCHASE, "article": article2},
|
||||
{"type": Operation.DEPOSIT, "amount": Decimal("10.00")},
|
||||
{"type": Operation.WITHDRAW, "amount": Decimal("-1.00")},
|
||||
{"type": Operation.EDIT, "amount": Decimal("7.00")},
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(operation_group.amount, Decimal("0.50"))
|
||||
|
||||
operation_list = list(operation_group.opes.all())
|
||||
# Passed args: with purchase, article, article_nb, amount
|
||||
self.assertEqual(operation_list[0].type, Operation.PURCHASE)
|
||||
self.assertEqual(operation_list[0].article, article1)
|
||||
self.assertEqual(operation_list[0].article_nb, 2)
|
||||
self.assertEqual(operation_list[0].amount, Decimal("-3.50"))
|
||||
# Passed args: with purchase, article, article_nb; without amount
|
||||
self.assertEqual(operation_list[1].type, Operation.PURCHASE)
|
||||
self.assertEqual(operation_list[1].article, article2)
|
||||
self.assertEqual(operation_list[1].article_nb, 2)
|
||||
self.assertEqual(operation_list[1].amount, Decimal("-8.00"))
|
||||
# Passed args: with purchase, article; without article_nb, amount
|
||||
self.assertEqual(operation_list[2].type, Operation.PURCHASE)
|
||||
self.assertEqual(operation_list[2].article, article2)
|
||||
self.assertEqual(operation_list[2].article_nb, 1)
|
||||
self.assertEqual(operation_list[2].amount, Decimal("-4.00"))
|
||||
# Passed args: with deposit, amount
|
||||
self.assertEqual(operation_list[3].type, Operation.DEPOSIT)
|
||||
self.assertEqual(operation_list[3].amount, Decimal("10.00"))
|
||||
# Passed args: with withdraw, amount
|
||||
self.assertEqual(operation_list[4].type, Operation.WITHDRAW)
|
||||
self.assertEqual(operation_list[4].amount, Decimal("-1.00"))
|
||||
# Passed args: with edit, amount
|
||||
self.assertEqual(operation_list[5].type, Operation.EDIT)
|
||||
self.assertEqual(operation_list[5].amount, Decimal("7.00"))
|
||||
|
|
|
|||
|
|
@ -28,7 +28,16 @@ from ..models import (
|
|||
TransferGroup,
|
||||
)
|
||||
from .testcases import ViewTestCaseMixin
|
||||
from .utils import create_team, create_user, get_perms, user_add_perms
|
||||
from .utils import (
|
||||
create_checkout,
|
||||
create_checkout_statement,
|
||||
create_inventory_article,
|
||||
create_operation_group,
|
||||
create_team,
|
||||
create_user,
|
||||
get_perms,
|
||||
user_add_perms,
|
||||
)
|
||||
|
||||
|
||||
class AccountListViewTests(ViewTestCaseMixin, TestCase):
|
||||
|
|
@ -2952,6 +2961,21 @@ class KPsulPerformOperationsViewTests(ViewTestCaseMixin, TestCase):
|
|||
|
||||
|
||||
class KPsulCancelOperationsViewTests(ViewTestCaseMixin, TestCase):
|
||||
"""
|
||||
Test cases for kpsul_cancel_operations view.
|
||||
|
||||
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.cancel_operations"
|
||||
url_expected = "/k-fet/k-psul/cancel_operations"
|
||||
|
||||
|
|
@ -2960,8 +2984,790 @@ class KPsulCancelOperationsViewTests(ViewTestCaseMixin, TestCase):
|
|||
auth_user = "team"
|
||||
auth_forbidden = [None, "user"]
|
||||
|
||||
def test_ok(self):
|
||||
pass
|
||||
with_liq = True
|
||||
|
||||
def setUp(self):
|
||||
super(KPsulCancelOperationsViewTests, self).setUp()
|
||||
|
||||
self.checkout = create_checkout(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)
|
||||
|
||||
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 test_invalid_operation_not_int(self):
|
||||
data = {"operations[]": ["a"]}
|
||||
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"], {})
|
||||
|
||||
def test_invalid_operation_not_exist(self):
|
||||
data = {"operations[]": ["1000"]}
|
||||
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"], {"opes_notexisting": [1000]})
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_purchase(self, now_mock):
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{"type": Operation.PURCHASE, "article": self.article, "article_nb": 2}
|
||||
],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(seconds=15)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
group = OperationGroup.objects.get()
|
||||
self.assertDictEqual(
|
||||
group.__dict__,
|
||||
{
|
||||
"_state": mock.ANY,
|
||||
"amount": Decimal("0.00"),
|
||||
"at": mock.ANY,
|
||||
"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": self.now + timedelta(seconds=15),
|
||||
"canceled_by_id": None,
|
||||
"group_id": group.pk,
|
||||
"id": mock.ANY,
|
||||
"type": Operation.PURCHASE,
|
||||
},
|
||||
)
|
||||
|
||||
self.assertDictEqual(
|
||||
json_data, {"canceled": [operation.pk], "errors": {}, "warnings": {}}
|
||||
)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("55.00"))
|
||||
self.article.refresh_from_db()
|
||||
self.assertEqual(self.article.stock, 22)
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("100.00"))
|
||||
|
||||
self.kpsul_consumer_mock.group_send.assert_called_with(
|
||||
"kfet.kpsul",
|
||||
{
|
||||
"opegroups": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": group.pk,
|
||||
"amount": Decimal("0.00"),
|
||||
"is_cof": False,
|
||||
}
|
||||
],
|
||||
"opes": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": operation.pk,
|
||||
"canceled_by__trigramme": None,
|
||||
"canceled_at": self.now + timedelta(seconds=15),
|
||||
}
|
||||
],
|
||||
"checkouts": [],
|
||||
"articles": [{"id": self.article.pk, "stock": 22}],
|
||||
},
|
||||
)
|
||||
|
||||
def test_purchase_with_addcost(self):
|
||||
# TODO(AD): L'état de la balance du compte destinataire de la majoration ne
|
||||
# devrait pas empêcher l'annulation d'une opération.
|
||||
addcost_user = create_user(
|
||||
"addcost", "ADD", account_attrs={"balance": Decimal("10.00")}
|
||||
)
|
||||
addcost_account = addcost_user.profile.account_kfet
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{
|
||||
"type": Operation.PURCHASE,
|
||||
"article": self.article,
|
||||
"article_nb": 2,
|
||||
"amount": Decimal("-6.00"),
|
||||
"addcost_amount": Decimal("1.00"),
|
||||
"addcost_for": addcost_account,
|
||||
}
|
||||
],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
self._assertResponseOk(resp)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("56.00"))
|
||||
addcost_account.refresh_from_db()
|
||||
self.assertEqual(addcost_account.balance, Decimal("9.00"))
|
||||
|
||||
def test_purchase_cash(self):
|
||||
group = create_operation_group(
|
||||
on_acc=self.accounts["liq"],
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{
|
||||
"type": Operation.PURCHASE,
|
||||
"article": self.article,
|
||||
"article_nb": 2,
|
||||
"amount": Decimal("-5.00"),
|
||||
}
|
||||
],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
self._assertResponseOk(resp)
|
||||
|
||||
self.assertEqual(self.accounts["liq"].balance, Decimal("0.00"))
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("95.00"))
|
||||
|
||||
ws_data_checkouts = self.kpsul_consumer_mock.group_send.call_args[0][1][
|
||||
"checkouts"
|
||||
]
|
||||
self.assertListEqual(
|
||||
ws_data_checkouts, [{"id": self.checkout.pk, "balance": Decimal("95.00")}]
|
||||
)
|
||||
|
||||
def test_purchase_cash_with_addcost(self):
|
||||
# TODO(AD): L'état de la balance du compte destinataire de la majoration ne
|
||||
# devrait pas empêcher l'annulation d'une opération.
|
||||
addcost_user = create_user(
|
||||
"addcost", "ADD", account_attrs={"balance": Decimal("10.00")}
|
||||
)
|
||||
addcost_account = addcost_user.profile.account_kfet
|
||||
group = create_operation_group(
|
||||
on_acc=self.accounts["liq"],
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{
|
||||
"type": Operation.PURCHASE,
|
||||
"article": self.article,
|
||||
"article_nb": 2,
|
||||
"amount": Decimal("-6.00"),
|
||||
"addcost_amount": Decimal("1.00"),
|
||||
"addcost_for": addcost_account,
|
||||
}
|
||||
],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
self._assertResponseOk(resp)
|
||||
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("94.00"))
|
||||
addcost_account.refresh_from_db()
|
||||
self.assertEqual(addcost_account.balance, Decimal("9.00"))
|
||||
|
||||
ws_data_checkouts = self.kpsul_consumer_mock.group_send.call_args[0][1][
|
||||
"checkouts"
|
||||
]
|
||||
self.assertListEqual(
|
||||
ws_data_checkouts, [{"id": self.checkout.pk, "balance": Decimal("94.00")}]
|
||||
)
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_deposit(self, now_mock):
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.DEPOSIT, "amount": Decimal("10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(seconds=15)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
group = OperationGroup.objects.get()
|
||||
self.assertDictEqual(
|
||||
group.__dict__,
|
||||
{
|
||||
"_state": mock.ANY,
|
||||
"amount": Decimal("0.00"),
|
||||
"at": mock.ANY,
|
||||
"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": self.now + timedelta(seconds=15),
|
||||
"canceled_by_id": None,
|
||||
"group_id": group.pk,
|
||||
"id": mock.ANY,
|
||||
"type": Operation.DEPOSIT,
|
||||
},
|
||||
)
|
||||
|
||||
self.assertDictEqual(
|
||||
json_data, {"canceled": [operation.pk], "errors": {}, "warnings": {}}
|
||||
)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("39.25"))
|
||||
self.article.refresh_from_db()
|
||||
self.assertEqual(self.article.stock, 20)
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("89.25"))
|
||||
|
||||
self.kpsul_consumer_mock.group_send.assert_called_with(
|
||||
"kfet.kpsul",
|
||||
{
|
||||
"opegroups": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": group.pk,
|
||||
"amount": Decimal("0.00"),
|
||||
"is_cof": False,
|
||||
}
|
||||
],
|
||||
"opes": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": operation.pk,
|
||||
"canceled_by__trigramme": None,
|
||||
"canceled_at": self.now + timedelta(seconds=15),
|
||||
}
|
||||
],
|
||||
"checkouts": [{"id": self.checkout.pk, "balance": Decimal("89.25")}],
|
||||
"articles": [],
|
||||
},
|
||||
)
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_withdraw(self, now_mock):
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.WITHDRAW, "amount": Decimal("-10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(seconds=15)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
group = OperationGroup.objects.get()
|
||||
self.assertDictEqual(
|
||||
group.__dict__,
|
||||
{
|
||||
"_state": mock.ANY,
|
||||
"amount": Decimal("0.00"),
|
||||
"at": mock.ANY,
|
||||
"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": self.now + timedelta(seconds=15),
|
||||
"canceled_by_id": None,
|
||||
"group_id": group.pk,
|
||||
"id": mock.ANY,
|
||||
"type": Operation.WITHDRAW,
|
||||
},
|
||||
)
|
||||
|
||||
self.assertDictEqual(
|
||||
json_data, {"canceled": [operation.pk], "errors": {}, "warnings": {}}
|
||||
)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("60.75"))
|
||||
self.article.refresh_from_db()
|
||||
self.assertEqual(self.article.stock, 20)
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("110.75"))
|
||||
|
||||
self.kpsul_consumer_mock.group_send.assert_called_with(
|
||||
"kfet.kpsul",
|
||||
{
|
||||
"opegroups": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": group.pk,
|
||||
"amount": Decimal("0.00"),
|
||||
"is_cof": False,
|
||||
}
|
||||
],
|
||||
"opes": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": operation.pk,
|
||||
"canceled_by__trigramme": None,
|
||||
"canceled_at": self.now + timedelta(seconds=15),
|
||||
}
|
||||
],
|
||||
"checkouts": [{"id": self.checkout.pk, "balance": Decimal("110.75")}],
|
||||
"articles": [],
|
||||
},
|
||||
)
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_edit(self, now_mock):
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.EDIT, "amount": Decimal("-10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(seconds=15)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
group = OperationGroup.objects.get()
|
||||
self.assertDictEqual(
|
||||
group.__dict__,
|
||||
{
|
||||
"_state": mock.ANY,
|
||||
"amount": Decimal("0.00"),
|
||||
"at": mock.ANY,
|
||||
"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": self.now + timedelta(seconds=15),
|
||||
"canceled_by_id": None,
|
||||
"group_id": group.pk,
|
||||
"id": mock.ANY,
|
||||
"type": Operation.EDIT,
|
||||
},
|
||||
)
|
||||
|
||||
self.assertDictEqual(
|
||||
json_data, {"canceled": [operation.pk], "errors": {}, "warnings": {}}
|
||||
)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("60.75"))
|
||||
self.article.refresh_from_db()
|
||||
self.assertEqual(self.article.stock, 20)
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("100.00"))
|
||||
|
||||
self.kpsul_consumer_mock.group_send.assert_called_with(
|
||||
"kfet.kpsul",
|
||||
{
|
||||
"opegroups": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": group.pk,
|
||||
"amount": Decimal("0.00"),
|
||||
"is_cof": False,
|
||||
}
|
||||
],
|
||||
"opes": [
|
||||
{
|
||||
"cancellation": True,
|
||||
"id": operation.pk,
|
||||
"canceled_by__trigramme": None,
|
||||
"canceled_at": self.now + timedelta(seconds=15),
|
||||
}
|
||||
],
|
||||
"checkouts": [],
|
||||
"articles": [],
|
||||
},
|
||||
)
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_old_operations(self, now_mock):
|
||||
kfet_config.set(cancel_duration=timedelta(minutes=10))
|
||||
user_add_perms(self.users["team"], ["kfet.cancel_old_operations"])
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
at=self.now,
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.WITHDRAW, "amount": Decimal("-10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(minutes=10, seconds=1)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
self.assertEqual(len(json_data["canceled"]), 1)
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_invalid_old_operations_requires_perm(self, now_mock):
|
||||
kfet_config.set(cancel_duration=timedelta(minutes=10))
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
at=self.now,
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.WITHDRAW, "amount": Decimal("-10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(minutes=10, seconds=1)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
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": ["Annuler des commandes non récentes"]},
|
||||
)
|
||||
|
||||
def test_already_canceled(self):
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{
|
||||
"type": Operation.WITHDRAW,
|
||||
"amount": Decimal("-10.75"),
|
||||
"canceled_at": timezone.now(),
|
||||
}
|
||||
],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
self.assertDictEqual(
|
||||
json_data["warnings"], {"already_canceled": [operation.pk]}
|
||||
)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("50.00"))
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("100.00"))
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_checkout_before_last_statement(self, now_mock):
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
at=self.now,
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.WITHDRAW, "amount": Decimal("-10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(seconds=30)
|
||||
create_checkout_statement(checkout=self.checkout)
|
||||
now_mock.return_value += timedelta(seconds=30)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
self.assertEqual(len(json_data["canceled"]), 1)
|
||||
|
||||
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"))
|
||||
|
||||
@mock.patch("django.utils.timezone.now")
|
||||
def test_article_before_last_inventory(self, now_mock):
|
||||
now_mock.return_value = self.now
|
||||
group = create_operation_group(
|
||||
at=self.now,
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{"type": Operation.PURCHASE, "article": self.article, "article_nb": 2}
|
||||
],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
now_mock.return_value += timedelta(seconds=30)
|
||||
create_inventory_article(article=self.article)
|
||||
now_mock.return_value += timedelta(seconds=30)
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
self.assertEqual(len(json_data["canceled"]), 1)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("55.00"))
|
||||
self.article.refresh_from_db()
|
||||
self.assertEqual(self.article.stock, 20)
|
||||
|
||||
def test_negative(self):
|
||||
kfet_config.set(overdraft_amount=Decimal("40.00"))
|
||||
user_add_perms(self.users["team"], ["kfet.perform_negative_operations"])
|
||||
self.account.balance = Decimal("-20.00")
|
||||
self.account.save()
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.DEPOSIT, "amount": Decimal("10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
self.assertEqual(len(json_data["canceled"]), 1)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("-30.75"))
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("89.25"))
|
||||
|
||||
def test_invalid_negative_above_thresholds(self):
|
||||
kfet_config.set(overdraft_amount=Decimal("5.00"))
|
||||
user_add_perms(self.users["team"], ["kfet.perform_negative_operations"])
|
||||
self.account.balance = Decimal("-20.00")
|
||||
self.account.save()
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.DEPOSIT, "amount": Decimal("10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
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": [self.account.trigramme]})
|
||||
|
||||
def test_invalid_negative_requires_perms(self):
|
||||
kfet_config.set(overdraft_amount=Decimal("40.00"))
|
||||
self.account.balance = Decimal("-20.00")
|
||||
self.account.save()
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[{"type": Operation.DEPOSIT, "amount": Decimal("10.75")}],
|
||||
)
|
||||
operation = group.opes.get()
|
||||
|
||||
data = {"operations[]": [str(operation.pk)]}
|
||||
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_partial_0(self):
|
||||
group = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{"type": Operation.PURCHASE, "article": self.article, "article_nb": 2},
|
||||
{"type": Operation.DEPOSIT, "amount": Decimal("10.75")},
|
||||
{"type": Operation.EDIT, "amount": Decimal("-6.00")},
|
||||
{
|
||||
"type": Operation.WITHDRAW,
|
||||
"amount": Decimal("-10.75"),
|
||||
"canceled_at": timezone.now(),
|
||||
},
|
||||
],
|
||||
)
|
||||
operation1 = group.opes.get(type=Operation.PURCHASE)
|
||||
operation2 = group.opes.get(type=Operation.EDIT)
|
||||
operation3 = group.opes.get(type=Operation.WITHDRAW)
|
||||
|
||||
data = {
|
||||
"operations[]": [str(operation1.pk), str(operation2.pk), str(operation3.pk)]
|
||||
}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
group.refresh_from_db()
|
||||
self.assertEqual(group.amount, Decimal("10.75"))
|
||||
self.assertEqual(group.opes.exclude(canceled_at=None).count(), 3)
|
||||
|
||||
self.assertDictEqual(
|
||||
json_data,
|
||||
{
|
||||
"canceled": [operation1.pk, operation2.pk],
|
||||
"warnings": {"already_canceled": [operation3.pk]},
|
||||
"errors": {},
|
||||
},
|
||||
)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("61.00"))
|
||||
self.article.refresh_from_db()
|
||||
self.assertEqual(self.article.stock, 22)
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("100.00"))
|
||||
|
||||
def test_multi_0(self):
|
||||
group1 = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{"type": Operation.PURCHASE, "article": self.article, "article_nb": 2},
|
||||
{"type": Operation.DEPOSIT, "amount": Decimal("10.75")},
|
||||
{"type": Operation.EDIT, "amount": Decimal("-6.00")},
|
||||
],
|
||||
)
|
||||
operation11 = group1.opes.get(type=Operation.PURCHASE)
|
||||
group2 = create_operation_group(
|
||||
on_acc=self.account,
|
||||
checkout=self.checkout,
|
||||
content=[
|
||||
{"type": Operation.PURCHASE, "article": self.article, "article_nb": 5},
|
||||
{"type": Operation.DEPOSIT, "amount": Decimal("3.00")},
|
||||
],
|
||||
)
|
||||
operation21 = group2.opes.get(type=Operation.PURCHASE)
|
||||
operation22 = group2.opes.get(type=Operation.DEPOSIT)
|
||||
|
||||
data = {
|
||||
"operations[]": [
|
||||
str(operation11.pk),
|
||||
str(operation21.pk),
|
||||
str(operation22.pk),
|
||||
]
|
||||
}
|
||||
resp = self.client.post(self.url, data)
|
||||
|
||||
json_data = self._assertResponseOk(resp)
|
||||
|
||||
group1.refresh_from_db()
|
||||
self.assertEqual(group1.amount, Decimal("4.75"))
|
||||
self.assertEqual(group1.opes.exclude(canceled_at=None).count(), 1)
|
||||
group2.refresh_from_db()
|
||||
self.assertEqual(group2.amount, Decimal(0))
|
||||
self.assertEqual(group2.opes.exclude(canceled_at=None).count(), 2)
|
||||
|
||||
self.assertEqual(len(json_data["canceled"]), 3)
|
||||
|
||||
self.account.refresh_from_db()
|
||||
self.assertEqual(self.account.balance, Decimal("64.50"))
|
||||
self.article.refresh_from_db()
|
||||
self.assertEqual(self.article.stock, 27)
|
||||
self.checkout.refresh_from_db()
|
||||
self.assertEqual(self.checkout.balance, Decimal("97.00"))
|
||||
|
||||
|
||||
class KPsulArticlesData(ViewTestCaseMixin, TestCase):
|
||||
|
|
|
|||
|
|
@ -1,7 +1,21 @@
|
|||
from datetime import timedelta
|
||||
from decimal import Decimal
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Permission
|
||||
from django.utils import timezone
|
||||
|
||||
from ..models import Account
|
||||
from ..models import (
|
||||
Account,
|
||||
Article,
|
||||
ArticleCategory,
|
||||
Checkout,
|
||||
CheckoutStatement,
|
||||
Inventory,
|
||||
InventoryArticle,
|
||||
Operation,
|
||||
OperationGroup,
|
||||
)
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
|
@ -184,3 +198,180 @@ def user_add_perms(user, perms_labels):
|
|||
# it to avoid using of the previous permissions cache.
|
||||
# https://docs.djangoproject.com/en/dev/topics/auth/default/#permission-caching
|
||||
return User.objects.get(pk=user.pk)
|
||||
|
||||
|
||||
def create_checkout(**kwargs):
|
||||
"""
|
||||
Factory to create a checkout.
|
||||
See defaults for unpassed arguments in code below.
|
||||
"""
|
||||
if "created_by" not in kwargs or "created_by_id" not in kwargs:
|
||||
try:
|
||||
team_account = Account.objects.get(cofprofile__user__username="team")
|
||||
except Account.DoesNotExist:
|
||||
team_account = create_team().profile.account_kfet
|
||||
kwargs["created_by"] = team_account
|
||||
kwargs.setdefault("name", "Checkout")
|
||||
kwargs.setdefault("valid_from", timezone.now() - timedelta(days=14))
|
||||
kwargs.setdefault("valid_to", timezone.now() - timedelta(days=14))
|
||||
|
||||
return Checkout.objects.create(**kwargs)
|
||||
|
||||
|
||||
def create_operation_group(content=None, **kwargs):
|
||||
"""
|
||||
Factory to create an OperationGroup and a set of related Operation.
|
||||
|
||||
It aims to get objects for testing purposes with minimal setup, and
|
||||
preserving consistency.
|
||||
For this, it uses, and creates if necessary, default objects for unpassed
|
||||
arguments.
|
||||
|
||||
Args:
|
||||
content: list of dict
|
||||
Describe set of Operation to create along the OperationGroup.
|
||||
Each item is passed to the Operation factory.
|
||||
kwargs:
|
||||
Used to control OperationGroup creation.
|
||||
|
||||
"""
|
||||
if content is None:
|
||||
content = []
|
||||
|
||||
# Prepare OperationGroup creation.
|
||||
|
||||
# Set 'checkout' for OperationGroup if unpassed.
|
||||
if "checkout" not in kwargs and "checkout_id" not in kwargs:
|
||||
try:
|
||||
checkout = Checkout.objects.get(name="Checkout")
|
||||
except Checkout.DoesNotExist:
|
||||
checkout = create_checkout()
|
||||
kwargs["checkout"] = checkout
|
||||
|
||||
# Set 'on_acc' for OperationGroup if unpassed.
|
||||
if "on_acc" not in kwargs and "on_acc_id" not in kwargs:
|
||||
try:
|
||||
on_acc = Account.objects.get(cofprofile__user__username="user")
|
||||
except Account.DoesNotExist:
|
||||
on_acc = create_user().profile.account_kfet
|
||||
kwargs["on_acc"] = on_acc
|
||||
|
||||
# Set 'is_cof' for OperationGroup if unpassed.
|
||||
if "is_cof" not in kwargs:
|
||||
# Use current is_cof status of 'on_acc'.
|
||||
kwargs["is_cof"] = kwargs["on_acc"].cofprofile.is_cof
|
||||
|
||||
# Create OperationGroup.
|
||||
group = OperationGroup.objects.create(**kwargs)
|
||||
|
||||
# We can now create objects referencing this OperationGroup.
|
||||
|
||||
# Process set of related Operation.
|
||||
if content:
|
||||
# Create them.
|
||||
operation_list = []
|
||||
for operation_kwargs in content:
|
||||
operation = create_operation(group=group, **operation_kwargs)
|
||||
operation_list.append(operation)
|
||||
|
||||
# Update OperationGroup accordingly, for consistency.
|
||||
for operation in operation_list:
|
||||
if not operation.canceled_at:
|
||||
group.amount += operation.amount
|
||||
group.save()
|
||||
|
||||
return group
|
||||
|
||||
|
||||
def create_operation(**kwargs):
|
||||
"""
|
||||
Factory to create an Operation for testing purposes.
|
||||
|
||||
If you give a 'group' (OperationGroup), it won't update it, you have do
|
||||
this "manually". Prefer using OperationGroup factory to get a consistent
|
||||
group with operations.
|
||||
|
||||
"""
|
||||
if "group" not in kwargs and "group_id" not in kwargs:
|
||||
# To get a consistent OperationGroup (amount...) for the operation
|
||||
# in-creation, prefer using create_operation_group factory with
|
||||
# 'content'.
|
||||
kwargs["group"] = create_operation_group()
|
||||
|
||||
if "type" not in kwargs:
|
||||
raise RuntimeError("Can't create an Operation without 'type'.")
|
||||
|
||||
# Apply defaults for purchase
|
||||
if kwargs["type"] == Operation.PURCHASE:
|
||||
if "article" not in kwargs:
|
||||
raise NotImplementedError(
|
||||
"One could write a create_article factory. Right now, you must"
|
||||
"pass an 'article'."
|
||||
)
|
||||
|
||||
# Unpassed 'article_nb' defaults to 1.
|
||||
kwargs.setdefault("article_nb", 1)
|
||||
|
||||
# Unpassed 'amount' will use current article price and quantity.
|
||||
if "amount" not in kwargs:
|
||||
if "addcost_for" in kwargs or "addcost_amount" in kwargs:
|
||||
raise NotImplementedError(
|
||||
"One could handle the case where 'amount' is missing and "
|
||||
"addcost applies. Right now, please pass an 'amount'."
|
||||
)
|
||||
kwargs["amount"] = -kwargs["article"].price * kwargs["article_nb"]
|
||||
|
||||
return Operation.objects.create(**kwargs)
|
||||
|
||||
|
||||
def create_checkout_statement(**kwargs):
|
||||
if "checkout" not in kwargs:
|
||||
kwargs["checkout"] = create_checkout()
|
||||
if "by" not in kwargs:
|
||||
try:
|
||||
team_account = Account.objects.get(cofprofile__user__username="team")
|
||||
except Account.DoesNotExist:
|
||||
team_account = create_team().profile.account_kfet
|
||||
kwargs["by"] = team_account
|
||||
kwargs.setdefault("balance_new", kwargs["checkout"].balance)
|
||||
kwargs.setdefault("balance_old", kwargs["checkout"].balance)
|
||||
kwargs.setdefault("amount_taken", Decimal(0))
|
||||
|
||||
return CheckoutStatement.objects.create(**kwargs)
|
||||
|
||||
|
||||
def create_article(**kwargs):
|
||||
kwargs.setdefault("name", "Article")
|
||||
kwargs.setdefault("price", Decimal("2.50"))
|
||||
kwargs.setdefault("stock", 20)
|
||||
if "category" not in kwargs:
|
||||
kwargs["category"] = create_article_category()
|
||||
|
||||
return Article.objects.create(**kwargs)
|
||||
|
||||
|
||||
def create_article_category(**kwargs):
|
||||
kwargs.setdefault("name", "Category")
|
||||
return ArticleCategory.objects.create(**kwargs)
|
||||
|
||||
|
||||
def create_inventory(**kwargs):
|
||||
if "by" not in kwargs:
|
||||
try:
|
||||
team_account = Account.objects.get(cofprofile__user__username="team")
|
||||
except Account.DoesNotExist:
|
||||
team_account = create_team().profile.account_kfet
|
||||
kwargs["by"] = team_account
|
||||
|
||||
return Inventory.objects.create(**kwargs)
|
||||
|
||||
|
||||
def create_inventory_article(**kwargs):
|
||||
if "inventory" not in kwargs:
|
||||
kwargs["inventory"] = create_inventory()
|
||||
if "article" not in kwargs:
|
||||
kwargs["article"] = create_article()
|
||||
kwargs.setdefault("stock_old", kwargs["article"].stock)
|
||||
kwargs.setdefault("stock_new", kwargs["article"].stock)
|
||||
|
||||
return InventoryArticle.objects.create(**kwargs)
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ Django==1.11.*
|
|||
django-autocomplete-light==3.1.3
|
||||
django-autoslug==1.9.3
|
||||
django-cas-ng==3.5.7
|
||||
django-djconfig==0.5.3
|
||||
django-djconfig==0.8.0
|
||||
django-recaptcha==1.4.0
|
||||
django-redis-cache==1.8.1
|
||||
icalendar
|
||||
|
|
|
|||
Loading…
Reference in a new issue