import json from datetime import timedelta from decimal import Decimal from unittest import mock from django.contrib.auth.models import Group, User from django.test import Client, TestCase from django.urls import reverse from django.utils import timezone from .. import KFET_DELETED_TRIGRAMME from ..auth import KFET_GENERIC_TRIGRAMME from ..config import kfet_config from ..models import ( Account, AccountNegative, Article, ArticleCategory, Checkout, CheckoutStatement, Inventory, InventoryArticle, Operation, OperationGroup, Order, OrderArticle, Supplier, SupplierArticle, Transfer, TransferGroup, ) from .testcases import ViewTestCaseMixin 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): url_name = "kfet.account" url_expected = "/k-fet/accounts/" auth_user = "team" auth_forbidden = [None, "user"] def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class AccountValidFreeTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.is_validandfree.ajax" url_expected = "/k-fet/accounts/is_validandfree" auth_user = "team" auth_forbidden = [None, "user"] def test_ok_isvalid_isfree(self): """Upper case trigramme not taken is valid and free.""" r = self.client.get(self.url, {"trigramme": "AAA"}) self.assertDictEqual( json.loads(r.content.decode("utf-8")), {"is_valid": True, "is_free": True} ) def test_ok_isvalid_notfree(self): """Already taken trigramme is not free, but valid.""" r = self.client.get(self.url, {"trigramme": "000"}) self.assertDictEqual( json.loads(r.content.decode("utf-8")), {"is_valid": True, "is_free": False} ) def test_ok_notvalid_isfree(self): """Lower case if forbidden but free.""" r = self.client.get(self.url, {"trigramme": "aaa"}) self.assertDictEqual( json.loads(r.content.decode("utf-8")), {"is_valid": False, "is_free": True} ) class AccountCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.create" url_expected = "/k-fet/accounts/new" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] post_data = { "trigramme": "AAA", "username": "plopplopplop", "first_name": "first", "last_name": "last", "email": "email@domain.net", } def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.add_account"])} def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.account.create")) account = Account.objects.get(trigramme="AAA") self.assertInstanceExpected( account, {"username": "plopplopplop", "first_name": "first", "last_name": "last"}, ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class AccountCreateAjaxViewTests(ViewTestCaseMixin, TestCase): urls_conf = [ { "name": "kfet.account.create.fromuser", "kwargs": {"username": "user"}, "expected": "/k-fet/accounts/new/user/user", }, { "name": "kfet.account.create.fromclipper", "kwargs": {"login_clipper": "myclipper", "fullname": "first last1 last2"}, "expected": ("/k-fet/accounts/new/clipper/myclipper/first%20last1%20last2"), }, {"name": "kfet.account.create.empty", "expected": "/k-fet/accounts/new/empty"}, ] auth_user = "team" auth_forbidden = [None, "user"] def test_fromuser(self): r = self.client.get(self.t_urls[0]) self.assertEqual(r.status_code, 200) user = self.users["user"] self.assertEqual(r.context["user_form"].instance, user) self.assertEqual(r.context["cof_form"].instance, user.profile) self.assertIn("account_form", r.context) def test_fromclipper(self): r = self.client.get(self.t_urls[1]) self.assertEqual(r.status_code, 200) self.assertIn("user_form", r.context) self.assertIn("cof_form", r.context) self.assertIn("account_form", r.context) def test_empty(self): r = self.client.get(self.t_urls[2]) self.assertEqual(r.status_code, 200) self.assertIn("user_form", r.context) self.assertIn("cof_form", r.context) self.assertIn("account_form", r.context) class AccountCreateAutocompleteViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.create.autocomplete" url_expected = "/k-fet/autocomplete/account_new" auth_user = "team" auth_forbidden = [None, "user"] def test_ok(self): r = self.client.get(self.url, {"q": "first"}) self.assertEqual(r.status_code, 200) self.assertEqual(len(r.context["users_notcof"]), 0) self.assertEqual(len(r.context["users_cof"]), 0) self.assertSetEqual( set(r.context["kfet"]), set([(self.accounts["user"], self.users["user"])]) ) class AccountSearchViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.search.autocomplete" url_expected = "/k-fet/autocomplete/account_search" auth_user = "team" auth_forbidden = [None, "user"] def test_ok(self): r = self.client.get(self.url, {"q": "first"}) self.assertEqual(r.status_code, 200) self.assertSetEqual(set(r.context["accounts"]), set([("000", "first last")])) class AccountReadViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.read" url_kwargs = {"trigramme": "001"} url_expected = "/k-fet/accounts/001" auth_user = "team" auth_forbidden = [None, "user"] # Users with forbidden access users should get a 404 here, to avoid leaking trigrams # See issue #224 def test_forbidden(self): for user in self.auth_forbidden: self.assertRedirectsToLoginOr404(user, self.url_expected) self.assertRedirectsToLoginOr404(user, "/k-fet/accounts/NEX") def assertRedirectsToLoginOr404(self, user, url): client = Client() if user is None: response = client.get(url) self.assertRedirects( response, "/login?next={}".format(url), fetch_redirect_response=False ) else: client.login(username=user, password=user) response = client.get(url) self.assertEqual(response.status_code, 404) def get_users_extra(self): return {"user1": create_user("user1", "001")} def setUp(self): super().setUp() user1_acc = self.accounts["user1"] team_acc = self.accounts["team"] # Dummy operations and operation groups checkout = Checkout.objects.create( created_by=team_acc, name="checkout", valid_from=timezone.now(), valid_to=timezone.now() + timezone.timedelta(days=365), ) opeg_data = [ (timezone.now(), Decimal("10")), (timezone.now() - timezone.timedelta(days=3), Decimal("3")), ] OperationGroup.objects.bulk_create( [ OperationGroup( on_acc=user1_acc, checkout=checkout, at=at, is_cof=False, amount=amount, ) for (at, amount) in opeg_data ] ) self.operation_groups = OperationGroup.objects.order_by("-amount") Operation.objects.create( group=self.operation_groups[0], type=Operation.PURCHASE, amount=Decimal("10"), ) Operation.objects.create( group=self.operation_groups[1], type=Operation.PURCHASE, amount=Decimal("3") ) def test_ok(self): """We can query the "Account - Read" page.""" r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_ok_self(self): client = Client() client.login(username="user1", password="user1") r = client.get(self.url) self.assertEqual(r.status_code, 200) class AccountUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.update" url_kwargs = {"trigramme": "001"} url_expected = "/k-fet/accounts/001/edit" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] post_data = { # User "first_name": "The first", "last_name": "The last", "email": "", # Group "groups[]": [], # Account "trigramme": "051", "nickname": "", "promo": "", # 'is_frozen': not checked # Account password "pwd1": "", "pwd2": "", } def get_users_extra(self): return { "user1": create_user("user1", "001"), "team1": create_team("team1", "101", perms=["kfet.change_account"]), } # Users with forbidden access users should get a 404 here, to avoid leaking trigrams # See issue #224 def test_forbidden(self): for method in ["get", "post"]: for user in self.auth_forbidden: self.assertRedirectsToLoginOr404(user, method, self.url_expected) self.assertRedirectsToLoginOr404( user, method, "/k-fet/accounts/NEX/edit" ) def assertRedirectsToLoginOr404(self, user, method, url): client = Client() meth = getattr(client, method) if user is None: response = meth(url) self.assertRedirects( response, "/login?next={}".format(url), fetch_redirect_response=False ) else: client.login(username=user, password=user) response = meth(url) self.assertEqual(response.status_code, 404) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_get_ok_self(self): client = Client() client.login(username="user1", password="user1") r = client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.account.read", args=["051"])) self.accounts["user1"].refresh_from_db() self.users["user1"].refresh_from_db() self.assertInstanceExpected( self.accounts["user1"], {"first_name": "first", "last_name": "last", "trigramme": "051"}, ) def test_post_ok_self(self): client = Client() client.login(username="user1", password="user1") post_data = {"first_name": "The first", "last_name": "The last"} r = client.post(self.url, post_data) self.assertRedirects(r, reverse("kfet.account.read", args=["001"])) self.accounts["user1"].refresh_from_db() self.users["user1"].refresh_from_db() self.assertInstanceExpected( self.accounts["user1"], {"first_name": "first", "last_name": "last"} ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class AccountDeleteViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.delete" url_kwargs = {"trigramme": "001"} url_expected = "/k-fet/accounts/001/delete" auth_user = "team1" auth_forbidden = [None, "user", "team"] http_methods = ["GET", "POST"] with_liq = True def get_users_extra(self): return { "user1": create_user("user1", "001"), "team1": create_team("team1", "101", perms=["kfet.delete_account"]), "trez": create_user("trez", "#13"), } def test_get_405(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 405) def test_post_ok(self): r = self.client.post(self.url, {}) self.assertRedirects(r, reverse("kfet.account")) with self.assertRaises(Account.DoesNotExist): self.accounts["user1"].refresh_from_db() def test_protected_accounts(self): for trigramme in ["LIQ", "#13", KFET_GENERIC_TRIGRAMME, KFET_DELETED_TRIGRAMME]: if Account.objects.get(trigramme=trigramme).readable: expected_code = 200 else: expected_code = 404 r = self.client.post( reverse(self.url_name, kwargs={"trigramme": trigramme}), {} ) self.assertRedirects( r, reverse("kfet.account.read", kwargs={"trigramme": trigramme}), target_status_code=expected_code, ) # Devrait être redondant avec le précédent, mais on sait jamais self.assertTrue(Account.objects.filter(trigramme=trigramme).exists()) def test_nonempty_accounts(self): self.accounts["user1"].balance = 1 self.accounts["user1"].save() r = self.client.post(self.url, {}) self.assertRedirects(r, reverse("kfet.account.read", kwargs=self.url_kwargs)) # Shouldn't throw an error self.accounts["user1"].refresh_from_db() class AccountGroupListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.group" url_expected = "/k-fet/accounts/groups" auth_user = "team1" auth_forbidden = [None, "user", "team"] def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.manage_perms"])} def setUp(self): super().setUp() self.group1 = Group.objects.create(name="K-Fêt - Group1") self.group2 = Group.objects.create(name="K-Fêt - Group2") def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) self.assertQuerysetEqual( r.context["groups"], map(repr, [self.group1, self.group2]), ordered=False ) class AccountGroupCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.group.create" url_expected = "/k-fet/accounts/groups/new" http_methods = ["GET", "POST"] auth_user = "team1" auth_forbidden = [None, "user", "team"] def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.manage_perms"])} @property def post_data(self): return { "name": "The Group", "permissions": [ str(self.perms["kfet.is_team"].pk), str(self.perms["kfet.manage_perms"].pk), ], } def setUp(self): super().setUp() self.perms = get_perms("kfet.is_team", "kfet.manage_perms") def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): r = self.client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.account.group")) group = Group.objects.get(name="K-Fêt The Group") self.assertQuerysetEqual( group.permissions.all(), map(repr, [self.perms["kfet.is_team"], self.perms["kfet.manage_perms"]]), ordered=False, ) class AccountGroupUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.group.update" http_methods = ["GET", "POST"] auth_user = "team1" auth_forbidden = [None, "user", "team"] @property def url_kwargs(self): return {"pk": self.group.pk} @property def url_expected(self): return "/k-fet/accounts/groups/{}/edit".format(self.group.pk) def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.manage_perms"])} @property def post_data(self): return { "name": "The Group", "permissions": [ str(self.perms["kfet.is_team"].pk), str(self.perms["kfet.manage_perms"].pk), ], } def setUp(self): super().setUp() self.perms = get_perms("kfet.is_team", "kfet.manage_perms") self.group = Group.objects.create(name="K-Fêt - Group") self.group.permissions.set(self.perms.values()) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): r = self.client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.account.group")) self.group.refresh_from_db() self.assertEqual(self.group.name, "K-Fêt The Group") self.assertQuerysetEqual( self.group.permissions.all(), map(repr, [self.perms["kfet.is_team"], self.perms["kfet.manage_perms"]]), ordered=False, ) class AccountNegativeListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.negative" url_expected = "/k-fet/accounts/negatives" auth_user = "team1" auth_forbidden = [None, "user", "team"] def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.view_negs"])} def setUp(self): super().setUp() account = self.accounts["user"] account.balance = -5 account.save() account.update_negative() def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) self.assertQuerysetEqual( r.context["negatives"], map(repr, [self.accounts["user"].negative]), ordered=False, ) class AccountStatOperationListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.stat.operation.list" url_kwargs = {"trigramme": "001"} url_expected = "/k-fet/accounts/001/stat/operations/list" auth_user = "user1" auth_forbidden = [None, "user", "team"] def get_users_extra(self): return {"user1": create_user("user1", "001")} # Users with forbidden access users should get a 404 here, to avoid leaking trigrams # See issue #224 def test_forbidden(self): for user in self.auth_forbidden: self.assertRedirectsToLoginOr404(user, self.url_expected) self.assertRedirectsToLoginOr404( user, "/k-fet/accounts/NEX/stat/operations/list" ) def assertRedirectsToLoginOr404(self, user, url): client = Client() if user is None: response = client.get(url) self.assertRedirects( response, "/login?next={}".format(url), fetch_redirect_response=False ) else: client.login(username=user, password=user) response = client.get(url) self.assertEqual(response.status_code, 404) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) base_url = reverse("kfet.account.stat.operation", args=["001"]) expected_stats = [ { "label": "Tout le temps", "url": { "path": base_url, "query": { "types": ["['purchase']"], "scale_name": ["month"], "scale_last": ["True"], "scale_begin": [ self.accounts["user1"].created_at.isoformat(" ") ], }, }, }, { "label": "1 an", "url": { "path": base_url, "query": { "types": ["['purchase']"], "scale_n_steps": ["12"], "scale_name": ["month"], "scale_last": ["True"], }, }, }, { "label": "3 mois", "url": { "path": base_url, "query": { "types": ["['purchase']"], "scale_n_steps": ["13"], "scale_name": ["week"], "scale_last": ["True"], }, }, }, { "label": "2 semaines", "url": { "path": base_url, "query": { "types": ["['purchase']"], "scale_n_steps": ["14"], "scale_name": ["day"], "scale_last": ["True"], }, }, }, ] for stat, expected in zip(content["stats"], expected_stats): expected_url = expected.pop("url") self.assertUrlsEqual(stat["url"], expected_url) self.assertDictContainsSubset(expected, stat) class AccountStatOperationViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.stat.operation" url_kwargs = {"trigramme": "001"} url_expected = "/k-fet/accounts/001/stat/operations" auth_user = "user1" auth_forbidden = [None, "user", "team"] # Users with forbidden access users should get a 404 here, to avoid leaking trigrams # See issue #224 def test_forbidden(self): for user in self.auth_forbidden: self.assertRedirectsToLoginOr404(user, self.url_expected) self.assertRedirectsToLoginOr404( user, "/k-fet/accounts/NEX/stat/operations" ) def assertRedirectsToLoginOr404(self, user, url): client = Client() if user is None: response = client.get(url) self.assertRedirects( response, "/login?next={}".format(url), fetch_redirect_response=False ) else: client.login(username=user, password=user) response = client.get(url) self.assertEqual(response.status_code, 404) def get_users_extra(self): return {"user1": create_user("user1", "001")} def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class AccountStatBalanceListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.stat.balance.list" url_kwargs = {"trigramme": "001"} url_expected = "/k-fet/accounts/001/stat/balance/list" auth_user = "user1" auth_forbidden = [None, "user", "team"] # Users with forbidden access users should get a 404 here, to avoid leaking trigrams # See issue #224 def test_forbidden(self): for user in self.auth_forbidden: self.assertRedirectsToLoginOr404(user, self.url_expected) self.assertRedirectsToLoginOr404( user, "/k-fet/accounts/NEX/stat/balance/list" ) def assertRedirectsToLoginOr404(self, user, url): client = Client() if user is None: response = client.get(url) self.assertRedirects( response, "/login?next={}".format(url), fetch_redirect_response=False ) else: client.login(username=user, password=user) response = client.get(url) self.assertEqual(response.status_code, 404) def get_users_extra(self): return {"user1": create_user("user1", "001")} def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) base_url = reverse("kfet.account.stat.balance", args=["001"]) expected_stats = [ {"label": "Tout le temps", "url": base_url}, { "label": "1 an", "url": {"path": base_url, "query": {"last_days": ["365"]}}, }, { "label": "6 mois", "url": {"path": base_url, "query": {"last_days": ["183"]}}, }, { "label": "3 mois", "url": {"path": base_url, "query": {"last_days": ["90"]}}, }, { "label": "30 jours", "url": {"path": base_url, "query": {"last_days": ["30"]}}, }, ] for stat, expected in zip(content["stats"], expected_stats): expected_url = expected.pop("url") self.assertUrlsEqual(stat["url"], expected_url) self.assertDictContainsSubset(expected, stat) class AccountStatBalanceViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.stat.balance" url_kwargs = {"trigramme": "001"} url_expected = "/k-fet/accounts/001/stat/balance" auth_user = "user1" auth_forbidden = [None, "user", "team"] # Users with forbidden access users should get a 404 here, to avoid leaking trigrams # See issue #224 def test_forbidden(self): for user in self.auth_forbidden: self.assertRedirectsToLoginOr404(user, self.url_expected) self.assertRedirectsToLoginOr404(user, "/k-fet/accounts/NEX/stat/balance") def assertRedirectsToLoginOr404(self, user, url): client = Client() if user is None: response = client.get(url) self.assertRedirects( response, "/login?next={}".format(url), fetch_redirect_response=False ) else: client.login(username=user, password=user) response = client.get(url) self.assertEqual(response.status_code, 404) def get_users_extra(self): return {"user1": create_user("user1", "001")} def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class CheckoutListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.checkout" url_expected = "/k-fet/checkouts/" auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() self.checkout1 = Checkout.objects.create( name="Checkout 1", created_by=self.accounts["team"], valid_from=self.now, valid_to=self.now + timedelta(days=5), ) self.checkout2 = Checkout.objects.create( name="Checkout 2", created_by=self.accounts["team"], valid_from=self.now + timedelta(days=10), valid_to=self.now + timedelta(days=15), ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) self.assertQuerysetEqual( r.context["checkouts"], map(repr, [self.checkout1, self.checkout2]), ordered=False, ) class CheckoutCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.checkout.create" url_expected = "/k-fet/checkouts/new" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] post_data = { "name": "Checkout", "valid_from": "2017-10-08 17:45:00", "valid_to": "2017-11-08 16:00:00", "balance": "3.14", # 'is_protected': not checked } def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.add_checkout"])} def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) checkout = Checkout.objects.get(name="Checkout") self.assertRedirects(r, checkout.get_absolute_url()) self.assertInstanceExpected( checkout, { "name": "Checkout", "valid_from": timezone.make_aware(datetime(2017, 10, 8, 17, 45)), "valid_to": timezone.make_aware(datetime(2017, 11, 8, 16, 00)), "balance": Decimal("3.14"), "is_protected": False, }, ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class CheckoutReadViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.checkout.read" auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.checkout.pk} @property def url_expected(self): return "/k-fet/checkouts/{}".format(self.checkout.pk) def setUp(self): super().setUp() with mock.patch("django.utils.timezone.now") as mock_now: mock_now.return_value = self.now self.checkout = Checkout.objects.create( name="Checkout", balance=Decimal("10"), created_by=self.accounts["team"], valid_from=self.now, valid_to=self.now + timedelta(days=1), ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) self.assertEqual(r.context["checkout"], self.checkout) class CheckoutUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.checkout.update" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] post_data = { "name": "Checkout updated", "valid_from": "2018-01-01 08:00:00", "valid_to": "2018-07-01 16:00:00", } @property def url_kwargs(self): return {"pk": self.checkout.pk} @property def url_expected(self): return "/k-fet/checkouts/{}/edit".format(self.checkout.pk) def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.change_checkout"])} def setUp(self): super().setUp() self.checkout = Checkout.objects.create( name="Checkout", valid_from=self.now, valid_to=self.now + timedelta(days=5), balance=Decimal("3.14"), is_protected=False, created_by=self.accounts["team"], ) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, self.checkout.get_absolute_url()) self.checkout.refresh_from_db() self.assertInstanceExpected( self.checkout, { "name": "Checkout updated", "valid_from": timezone.make_aware(datetime(2018, 1, 1, 8, 0, 0)), "valid_to": timezone.make_aware(datetime(2018, 7, 1, 16, 0, 0)), }, ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class CheckoutStatementListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.checkoutstatement" url_expected = "/k-fet/checkouts/statements/" auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() self.checkout1 = Checkout.objects.create( created_by=self.accounts["team"], name="Checkout 1", valid_from=self.now, valid_to=self.now + timedelta(days=5), ) self.checkout2 = Checkout.objects.create( created_by=self.accounts["team"], name="Checkout 2", valid_from=self.now + timedelta(days=10), valid_to=self.now + timedelta(days=15), ) self.statement1 = CheckoutStatement.objects.create( checkout=self.checkout1, by=self.accounts["team"], balance_old=5, balance_new=0, amount_taken=5, ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) expected_statements = list(self.checkout1.statements.all()) + list( self.checkout2.statements.all() ) self.assertQuerysetEqual( r.context["checkoutstatements"], map(repr, expected_statements), ordered=False, ) class CheckoutStatementCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.checkoutstatement.create" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] post_data = { # Let "balance_001": 0, "balance_002": 0, "balance_005": 0, "balance_01": 0, "balance_02": 0, "balance_05": 0, "balance_1": 1, "balance_2": 0, "balance_5": 0, "balance_10": 1, "balance_20": 0, "balance_50": 0, "balance_100": 1, "balance_200": 0, "balance_500": 0, # Taken "taken_001": 0, "taken_002": 0, "taken_005": 0, "taken_01": 0, "taken_02": 0, "taken_05": 0, "taken_1": 2, "taken_2": 0, "taken_5": 0, "taken_10": 2, "taken_20": 0, "taken_50": 0, "taken_100": 2, "taken_200": 0, "taken_500": 0, "taken_cheque": 0, # 'not_count': not checked } @property def url_kwargs(self): return {"pk_checkout": self.checkout.pk} @property def url_expected(self): return "/k-fet/checkouts/{}/statements/add".format(self.checkout.pk) def get_users_extra(self): return { "team1": create_team("team1", "001", perms=["kfet.add_checkoutstatement"]) } def setUp(self): super().setUp() self.checkout = Checkout.objects.create( name="Checkout", created_by=self.accounts["team"], balance=5, valid_from=self.now, valid_to=self.now + timedelta(days=5), ) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) @mock.patch("django.utils.timezone.now") def test_post_ok(self, mock_now): self.now += timedelta(days=2) mock_now.return_value = self.now client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, self.checkout.get_absolute_url()) statement = CheckoutStatement.objects.get(at=self.now) self.assertInstanceExpected( statement, { "by": self.accounts["team1"], "checkout": self.checkout, "balance_old": Decimal("5"), "balance_new": Decimal("111"), "amount_taken": Decimal("222"), "amount_error": Decimal("328"), "at": self.now, "not_count": False, }, ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class CheckoutStatementUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.checkoutstatement.update" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] post_data = { "amount_taken": 3, "amount_error": 2, "balance_old": 8, "balance_new": 5, # Taken "taken_001": 0, "taken_002": 0, "taken_005": 0, "taken_01": 0, "taken_02": 0, "taken_05": 0, "taken_1": 1, "taken_2": 1, "taken_5": 0, "taken_10": 0, "taken_20": 0, "taken_50": 0, "taken_100": 0, "taken_200": 0, "taken_500": 0, "taken_cheque": 0, } @property def url_kwargs(self): return {"pk_checkout": self.checkout.pk, "pk": self.statement.pk} @property def url_expected(self): return "/k-fet/checkouts/{pk_checkout}/statements/{pk}/edit".format( pk_checkout=self.checkout.pk, pk=self.statement.pk ) def get_users_extra(self): return { "team1": create_team( "team1", "101", perms=["kfet.change_checkoutstatement"] ) } def setUp(self): super().setUp() self.checkout = Checkout.objects.create( name="Checkout", created_by=self.accounts["team"], balance=5, valid_from=self.now, valid_to=self.now + timedelta(days=5), ) self.statement = CheckoutStatement.objects.create( by=self.accounts["team"], checkout=self.checkout, balance_new=5, balance_old=8, amount_error=2, amount_taken=5, ) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) @mock.patch("django.utils.timezone.now") def test_post_ok(self, mock_now): self.now += timedelta(days=2) mock_now.return_value = self.now client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, self.checkout.get_absolute_url()) self.statement.refresh_from_db() self.assertInstanceExpected( self.statement, { "taken_1": 1, "taken_2": 1, "balance_new": 5, "balance_old": 8, "amount_error": 0, "amount_taken": 3, }, ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class ArticleCategoryListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.category" url_expected = "/k-fet/categories/" auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() self.category1 = ArticleCategory.objects.create(name="Category 1") self.category2 = ArticleCategory.objects.create(name="Category 2") def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) self.assertQuerysetEqual( r.context["categories"], map(repr, [self.category1, self.category2]) ) class ArticleCategoryUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.category.update" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.category.pk} @property def url_expected(self): return "/k-fet/categories/{}/edit".format(self.category.pk) def get_users_extra(self): return { "team1": create_team("team1", "101", perms=["kfet.change_articlecategory"]) } @property def post_data(self): return { "name": "The Category", # 'has_addcost': not checked } def setUp(self): super().setUp() self.category = ArticleCategory.objects.create(name="Category") def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.category")) self.category.refresh_from_db() self.assertInstanceExpected( self.category, {"name": "The Category", "has_addcost": False} ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class ArticleListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.article" url_expected = "/k-fet/articles/" auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() category = ArticleCategory.objects.create(name="Category") self.article1 = Article.objects.create(name="Article 1", category=category) self.article2 = Article.objects.create(name="Article 2", category=category) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) self.assertQuerysetEqual( r.context["articles"], map(repr, [self.article1, self.article2]) ) class ArticleCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.article.create" url_expected = "/k-fet/articles/new" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.add_article"])} @property def post_data(self): return { "name": "Article", "category": self.category.pk, "stock": 5, "price": "2.5", "volume": "33", "abv": "6.3", } def setUp(self): super().setUp() self.category = ArticleCategory.objects.create(name="Category") def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) article = Article.objects.get(name="Article") self.assertRedirects(r, article.get_absolute_url()) self.assertInstanceExpected( article, {"name": "Article", "category": self.category} ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class ArticleReadViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.article.read" auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.article.pk} @property def url_expected(self): return "/k-fet/articles/{}".format(self.article.pk) def setUp(self): super().setUp() self.article = Article.objects.create( name="Article", category=ArticleCategory.objects.create(name="Category"), stock=5, price=Decimal("2.5"), ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) self.assertEqual(r.context["article"], self.article) class ArticleUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.article.update" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.article.pk} @property def url_expected(self): return "/k-fet/articles/{}/edit".format(self.article.pk) def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.change_article"])} @property def post_data(self): return { "name": "The Article", "category": self.article.category.pk, "is_sold": "1", "price": "3.5", "box_type": "carton", # 'hidden': not checked "volume": "33", "abv": "6.3", } def setUp(self): super().setUp() self.category = ArticleCategory.objects.create(name="Category") self.article = Article.objects.create( name="Article", category=self.category, stock=5, price=Decimal("2.5") ) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, self.article.get_absolute_url()) self.article.refresh_from_db() self.assertInstanceExpected( self.article, {"name": "The Article", "price": Decimal("3.5")} ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class ArticleDeleteViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.article.delete" auth_user = "team1" auth_forbidden = [None, "user", "team"] @property def url_kwargs(self): return {"pk": self.article.pk} @property def url_expected(self): return "/k-fet/articles/{}/delete".format(self.article.pk) def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.delete_article"])} def setUp(self): super().setUp() self.category = ArticleCategory.objects.create(name="Category") self.article = Article.objects.create( name="Article", category=self.category, stock=5, price=Decimal("2.5") ) def test_get_redirects(self): r = self.client.get(self.url) self.assertRedirects(r, reverse("kfet.article.read", kwargs=self.url_kwargs)) def test_post_ok(self): r = self.client.post(self.url, {}) self.assertRedirects(r, reverse("kfet.article")) with self.assertRaises(Article.DoesNotExist): self.article.refresh_from_db() class ArticleStatSalesListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.article.stat.sales.list" auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.article.pk} @property def url_expected(self): return "/k-fet/articles/{}/stat/sales/list".format(self.article.pk) def setUp(self): super().setUp() self.article = Article.objects.create( name="Article", category=ArticleCategory.objects.create(name="Category") ) checkout = Checkout.objects.create( name="Checkout", created_by=self.accounts["team"], balance=5, valid_from=self.now, valid_to=self.now + timedelta(days=5), ) self.opegroup = create_operation_group( on_acc=self.accounts["user"], checkout=checkout, content=[ {"type": Operation.PURCHASE, "article": self.article, "article_nb": 2}, ], ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) base_url = reverse("kfet.article.stat.sales", args=[self.article.pk]) expected_stats = [ { "label": "Tout le temps", "url": { "path": base_url, "query": { "scale_name": ["month"], "scale_last": ["True"], "scale_begin": [self.opegroup.at.isoformat(" ")], }, }, }, { "label": "1 an", "url": { "path": base_url, "query": { "scale_n_steps": ["12"], "scale_name": ["month"], "scale_last": ["True"], }, }, }, { "label": "3 mois", "url": { "path": base_url, "query": { "scale_n_steps": ["13"], "scale_name": ["week"], "scale_last": ["True"], }, }, }, { "label": "2 semaines", "url": { "path": base_url, "query": { "scale_n_steps": ["14"], "scale_name": ["day"], "scale_last": ["True"], }, }, }, ] for stat, expected in zip(content["stats"], expected_stats): expected_url = expected.pop("url") self.assertUrlsEqual(stat["url"], expected_url) self.assertDictContainsSubset(expected, stat) class ArticleStatSalesViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.article.stat.sales" auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.article.pk} @property def url_expected(self): return "/k-fet/articles/{}/stat/sales".format(self.article.pk) def setUp(self): super().setUp() self.article = Article.objects.create( name="Article", category=ArticleCategory.objects.create(name="Category") ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class KPsulViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.kpsul" url_expected = "/k-fet/k-psul/" auth_user = "team" auth_forbidden = [None, "user"] def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class KPsulCheckoutDataViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.kpsul.checkout_data" url_expected = "/k-fet/k-psul/checkout_data" http_methods = ["POST"] auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() self.checkout = Checkout.objects.create( name="Checkout", balance=Decimal("10"), created_by=self.accounts["team"], valid_from=self.now, valid_to=self.now + timedelta(days=5), ) def test_ok(self): r = self.client.post(self.url, {"pk": self.checkout.pk}) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) expected = {"name": "Checkout", "balance": "10.00"} self.assertDictContainsSubset(expected, content) self.assertSetEqual( set(content.keys()), set( [ "balance", "id", "name", "valid_from", "valid_to", "last_statement_at", "last_statement_balance", "last_statement_by_first_name", "last_statement_by_last_name", "last_statement_by_trigramme", ] ), ) 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" http_methods = ["POST"] auth_user = "team" auth_forbidden = [None, "user"] 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, ) # Another Article, price=2.5, stock=20, no COF reduction self.article_no_reduction = Article.objects.create( category=ArticleCategory.objects.create( name="Category_no_reduction", has_reduction=False, ), name="Article_no_reduction", 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"], ["[kfet] 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"], ["[kfet] 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", { "groups": [ { "add": True, "type": "operation", "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, "entries": [ { "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_purchase_no_reduction(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": "2", "form-0-type": "purchase", "form-0-amount": "", "form-0-article": str(self.article_no_reduction.pk), "form-0-article_nb": "1", "form-1-type": "purchase", "form-1-amount": "", "form-1-article": str(self.article.pk), "form-1-article_nb": "1", } ) resp = self.client.post(self.url, data) self._assertResponseOk(resp) operation_group = OperationGroup.objects.get() self.assertEqual(operation_group.amount, Decimal("-4.50")) operation = Operation.objects.get(article=self.article) self.assertEqual(operation.amount, Decimal("-2.00")) operation = Operation.objects.get(article=self.article_no_reduction) self.assertEqual(operation.amount, Decimal("-2.50")) 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", { "groups": [ { "add": True, "type": "operation", "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", "entries": [ { "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"], ["[kfet] 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", { "groups": [ { "add": True, "type": "operation", "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, "entries": [ { "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", { "groups": [ { "add": True, "type": "operation", "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", "entries": [ { "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"], ["[kfet] 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]["groups"][0][ "entries" ][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]["groups"][0][ "entries" ][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]["groups"][0][ "entries" ][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]["groups"][0][ "entries" ][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]["groups"][0][ "entries" ][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": ["[kfet] 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", { "groups": [ { "add": True, "type": "operation", "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, "entries": [ { "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): """ 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.operations.cancel" url_expected = "/k-fet/k-psul/cancel_operations" http_methods = ["POST"] auth_user = "team" auth_forbidden = [None, "user"] 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": [ { "id": operation.id, # l'encodage des dates en JSON est relou... "canceled_at": mock.ANY, "canceled_by__trigramme": None, } ], "errors": {}, "warnings": {}, "opegroups_to_update": [ { "id": group.pk, "amount": str(group.amount), "is_cof": group.is_cof, } ], }, ) 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", {"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": [ { "id": operation.id, # l'encodage des dates en JSON est relou... "canceled_at": mock.ANY, "canceled_by__trigramme": None, } ], "errors": {}, "warnings": {}, "opegroups_to_update": [ { "id": group.pk, "amount": str(group.amount), "is_cof": group.is_cof, } ], }, ) 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", { "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": [ { "id": operation.id, # l'encodage des dates en JSON est relou... "canceled_at": mock.ANY, "canceled_by__trigramme": None, } ], "errors": {}, "warnings": {}, "opegroups_to_update": [ { "id": group.pk, "amount": str(group.amount), "is_cof": group.is_cof, } ], }, ) 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", { "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": [ { "id": operation.id, # l'encodage des dates en JSON est relou... "canceled_at": mock.ANY, "canceled_by__trigramme": None, } ], "errors": {}, "warnings": {}, "opegroups_to_update": [ { "id": group.pk, "amount": str(group.amount), "is_cof": group.is_cof, } ], }, ) 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", {"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": ["[kfet] 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": ["[kfet] 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.maxDiff = None self.assertDictEqual( json_data, { "canceled": [ { "id": operation1.id, # l'encodage des dates en JSON est relou... "canceled_at": mock.ANY, "canceled_by__trigramme": None, }, { "id": operation2.id, # l'encodage des dates en JSON est relou... "canceled_at": mock.ANY, "canceled_by__trigramme": None, }, ], "errors": {}, "warnings": {"already_canceled": [operation3.pk]}, "opegroups_to_update": [ { "id": group.pk, "amount": str(group.amount), "is_cof": group.is_cof, } ], }, ) 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): url_name = "kfet.kpsul.articles_data" url_expected = "/k-fet/k-psul/articles_data" auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() category = ArticleCategory.objects.create(name="Catégorie") self.article1 = Article.objects.create(category=category, name="Article 1") self.article2 = Article.objects.create( category=category, name="Article 2", price=Decimal("2.5") ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) articles = content["articles"] expected_list = [ {"category__name": "Catégorie", "name": "Article 1", "price": "0.00"}, {"category__name": "Catégorie", "name": "Article 2", "price": "2.50"}, ] for expected, article in zip(expected_list, articles): self.assertDictContainsSubset(expected, article) self.assertSetEqual( set(article.keys()), set( [ "id", "name", "price", "stock", "category_id", "category__name", "category__has_addcost", "category__has_reduction", ] ), ) class KPsulUpdateAddcost(ViewTestCaseMixin, TestCase): url_name = "kfet.kpsul.update_addcost" url_expected = "/k-fet/k-psul/update_addcost" http_methods = ["POST"] auth_user = "team" auth_forbidden = [None, "user"] post_data = {"trigramme": "000", "amount": "0.5"} def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.manage_addcosts"])} def test_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertEqual(r.status_code, 200) self.assertEqual(kfet_config.addcost_for, Account.objects.get(trigramme="000")) self.assertEqual(kfet_config.addcost_amount, Decimal("0.5")) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbidden(r) class KPsulGetSettings(ViewTestCaseMixin, TestCase): url_name = "kfet.kpsul.get_settings" url_expected = "/k-fet/k-psul/get_settings" auth_user = "team" auth_forbidden = [None, "user"] def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class HistoryJSONViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.history.json" url_expected = "/k-fet/history.json" auth_user = "user" auth_forbidden = [None, "noaccount"] def test_ok(self): r = self.client.post(self.url) self.assertEqual(r.status_code, 200) def get_users_extra(self): noaccount = User.objects.create(username="noaccount") noaccount.set_password("noaccount") noaccount.save() return {"noaccount": noaccount} class AccountReadJSONViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.account.read.json" http_methods = ["GET"] auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"trigramme": self.accounts["user"].trigramme} @property def url_expected(self): return "/k-fet/accounts/{}/.json".format(self.accounts["user"].trigramme) def setUp(self): super().setUp() # A Checkout, curently usable, balance=100 self.checkout = Checkout.objects.create( created_by=self.accounts["team"], name="Checkout", valid_from=self.now - timedelta(days=7), valid_to=self.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, volume=Decimal("33"), abv=Decimal("7.3"), ) now = timezone.localtime(self.now) ope_party1 = OperationGroup.objects.create( on_acc=self.accounts["user"], checkout=self.checkout, at=now.replace(hour=1), ) ope_party2 = OperationGroup.objects.create( on_acc=self.accounts["user"], checkout=self.checkout, at=now.replace(day=now.day - 1, hour=22), ) ope_noparty = OperationGroup.objects.create( on_acc=self.accounts["user"], checkout=self.checkout, at=now.replace(day=now.day - 1, hour=19), ) Operation.objects.create( group=ope_party1, type=Operation.PURCHASE, article=self.article, article_nb=2, ) Operation.objects.create( group=ope_party2, type=Operation.PURCHASE, article=self.article, article_nb=1, ) Operation.objects.create( group=ope_party2, type=Operation.PURCHASE, article=self.article, article_nb=1, ) Operation.objects.create( group=ope_noparty, type=Operation.PURCHASE, article=self.article, article_nb=3, ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) expected = {"name": "first last", "trigramme": "000", "balance": "0.00"} self.assertDictContainsSubset(expected, content) self.assertSetEqual( set(content.keys()), set( [ "balance", "departement", "email", "id", "is_cof", "is_frozen", "name", "nickname", "promo", "trigramme", "n_units", ] ), ) @mock.patch("django.utils.timezone.now") def test_without_party(self, mock_now): now = timezone.localtime(self.now) mock_now.return_value = now.replace(hour=20).astimezone(timezone.utc) r = self.client.get(self.url) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) self.assertEqual(content["n_units"], 0) @mock.patch("django.utils.timezone.now") def test_with_party(self, mock_now): now = timezone.localtime(self.now) mock_now.return_value = now.replace(hour=5).astimezone(timezone.utc) r = self.client.get(self.url) self.assertEqual(r.status_code, 200) content = json.loads(r.content.decode("utf-8")) self.assertAlmostEqual(float(content["n_units"]), 3 * 33 * 7.3 * 0.8 / 100) class SettingsListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.settings" url_expected = "/k-fet/settings/" auth_user = "team1" auth_forbidden = [None, "user", "team"] def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.see_config"])} def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class SettingsUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.settings.update" url_expected = "/k-fet/settings/edit" http_methods = ["GET", "POST"] auth_user = "team1" auth_forbidden = [None, "user", "team"] @property def post_data(self): return { "kfet_reduction_cof": "25", "kfet_addcost_amount": "0.5", "kfet_addcost_for": self.accounts["user"].pk, "kfet_overdraft_duration": "2 00:00:00", "kfet_overdraft_amount": "25", "kfet_cancel_duration": "00:20:00", } def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.change_config"])} def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): r = self.client.post(self.url, self.post_data) # Redirect is skipped because client may lack permissions. self.assertRedirects(r, reverse("kfet.settings"), fetch_redirect_response=False) expected_config = { "reduction_cof": Decimal("25"), "addcost_amount": Decimal("0.5"), "addcost_for": self.accounts["user"], "overdraft_duration": timedelta(days=2), "overdraft_amount": Decimal("25"), "cancel_duration": timedelta(minutes=20), } for key, expected in expected_config.items(): self.assertEqual(getattr(kfet_config, key), expected) class TransferListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.transfers" url_expected = "/k-fet/transfers/" auth_user = "team" auth_forbidden = [None, "user"] def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class TransferCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.transfers.create" url_expected = "/k-fet/transfers/new" auth_user = "team" auth_forbidden = [None, "user"] def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class TransferPerformViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.transfers.perform" url_expected = "/k-fet/transfers/perform" http_methods = ["POST"] auth_user = "team1" auth_forbidden = [None, "user", "team"] def get_users_extra(self): return { "team1": create_team( "team1", "101", perms=[ # Required "kfet.add_transfer", # Convenience "kfet.perform_negative_operations", ], ) } @property def post_data(self): return { # General "comment": "", # Formset management "form-TOTAL_FORMS": "10", "form-INITIAL_FORMS": "0", "form-MIN_NUM_FORMS": "1", "form-MAX_NUM_FORMS": "1000", # Transfer 1 "form-0-from_acc": str(self.accounts["user"].pk), "form-0-to_acc": str(self.accounts["team"].pk), "form-0-amount": "3.5", # Transfer 2 "form-1-from_acc": str(self.accounts["team"].pk), "form-1-to_acc": str(self.accounts["team1"].pk), "form-1-amount": "2.4", } def test_ok(self): r = self.client.post(self.url, self.post_data) self.assertEqual(r.status_code, 200) user = self.accounts["user"] user.refresh_from_db() self.assertEqual(user.balance, Decimal("-3.5")) team = self.accounts["team"] team.refresh_from_db() self.assertEqual(team.balance, Decimal("1.1")) team1 = self.accounts["team1"] team1.refresh_from_db() self.assertEqual(team1.balance, Decimal("2.4")) class TransferCancelViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.transfers.cancel" url_expected = "/k-fet/transfers/cancel" http_methods = ["POST"] auth_user = "team1" auth_forbidden = [None, "user", "team"] def get_users_extra(self): return { "team1": create_team( "team1", "101", perms=[ # Convenience "kfet.perform_negative_operations" ], ) } @property def post_data(self): return {"transfers[]": [self.transfer1.pk, self.transfer2.pk]} def setUp(self): super().setUp() group = TransferGroup.objects.create() self.transfer1 = Transfer.objects.create( group=group, from_acc=self.accounts["user"], to_acc=self.accounts["team"], amount="3.5", ) self.transfer2 = Transfer.objects.create( group=group, from_acc=self.accounts["team"], to_acc=self.accounts["root"], amount="2.4", ) def test_ok(self): r = self.client.post(self.url, self.post_data) self.assertEqual(r.status_code, 200) user = self.accounts["user"] user.refresh_from_db() self.assertEqual(user.balance, Decimal("3.5")) team = self.accounts["team"] team.refresh_from_db() self.assertEqual(team.balance, Decimal("-1.1")) root = self.accounts["root"] root.refresh_from_db() self.assertEqual(root.balance, Decimal("-2.4")) class InventoryListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.inventory" url_expected = "/k-fet/inventaires/" auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() self.inventory = Inventory.objects.create(by=self.accounts["team"]) category = ArticleCategory.objects.create(name="Category") article = Article.objects.create(name="Article", category=category) InventoryArticle.objects.create( inventory=self.inventory, article=article, stock_old=5, stock_new=0 ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) inventories = r.context["inventories"] self.assertQuerysetEqual(inventories, map(repr, [self.inventory])) class InventoryCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.inventory.create" url_expected = "/k-fet/inventaires/new" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.add_inventory"])} @property def post_data(self): return { # Formset management "form-TOTAL_FORMS": "2", "form-INITIAL_FORMS": "2", "form-MIN_NUM_FORMS": "0", "form-MAX_NUM_FORMS": "1000", # Article 1 "form-0-article": str(self.article1.pk), "form-0-stock_new": "5", # Article 2 "form-1-article": str(self.article2.pk), "form-1-stock_new": "10", } def setUp(self): super().setUp() category = ArticleCategory.objects.create(name="Category") self.article1 = Article.objects.create(category=category, name="Article 1") self.article2 = Article.objects.create(category=category, name="Article 2") def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.inventory")) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class InventoryReadViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.inventory.read" auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.inventory.pk} @property def url_expected(self): return "/k-fet/inventaires/{}".format(self.inventory.pk) def setUp(self): super().setUp() self.inventory = Inventory.objects.create(by=self.accounts["team"]) category = ArticleCategory.objects.create(name="Category") article = Article.objects.create(name="Article", category=category) InventoryArticle.objects.create( inventory=self.inventory, article=article, stock_old=5, stock_new=0 ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class OrderListViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.order" url_expected = "/k-fet/orders/" auth_user = "team" auth_forbidden = [None, "user"] def setUp(self): super().setUp() category = ArticleCategory.objects.create(name="Category") article = Article.objects.create(name="Article", category=category) supplier = Supplier.objects.create(name="Supplier") SupplierArticle.objects.create(supplier=supplier, article=article) self.order = Order.objects.create(supplier=supplier) OrderArticle.objects.create( order=self.order, article=article, quantity_ordered=24 ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) orders = r.context["orders"] self.assertQuerysetEqual(orders, map(repr, [self.order])) class OrderReadViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.order.read" auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.order.pk} @property def url_expected(self): return "/k-fet/orders/{}".format(self.order.pk) def setUp(self): super().setUp() category = ArticleCategory.objects.create(name="Category") article = Article.objects.create(name="Article", category=category) supplier = Supplier.objects.create(name="Supplier") SupplierArticle.objects.create(supplier=supplier, article=article) self.order = Order.objects.create(supplier=supplier) OrderArticle.objects.create( order=self.order, article=article, quantity_ordered=24 ) def test_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) class SupplierUpdateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.order.supplier.update" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.supplier.pk} @property def url_expected(self): return "/k-fet/orders/suppliers/{}/edit".format(self.supplier.pk) def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.change_supplier"])} @property def post_data(self): return { "name": "The Supplier", "phone": "", "comment": "", "address": "", "email": "", } def setUp(self): super().setUp() self.supplier = Supplier.objects.create(name="Supplier") def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) def test_post_ok(self): client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.order")) self.supplier.refresh_from_db() self.assertEqual(self.supplier.name, "The Supplier") def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class OrderCreateViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.order.new" auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.supplier.pk} @property def url_expected(self): return "/k-fet/orders/suppliers/{}/new-order".format(self.supplier.pk) def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.add_order"])} @property def post_data(self): return { # Formset management "form-TOTAL_FORMS": "1", "form-INITIAL_FORMS": "1", "form-MIN_NUM_FORMS": "0", "form-MAX_NUM_FORMS": "1000", # Article "form-0-article": self.article.pk, "form-0-quantity_ordered": "20", } def setUp(self): super().setUp() category = ArticleCategory.objects.create(name="Category") self.article = Article.objects.create(name="Article", category=category) self.supplier = Supplier.objects.create(name="Supplier") SupplierArticle.objects.create(supplier=self.supplier, article=self.article) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) @mock.patch("django.utils.timezone.now") def test_post_ok(self, mock_now): mock_now.return_value = self.now client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) order = Order.objects.get(at=self.now) self.assertRedirects(r, reverse("kfet.order.read", args=[order.pk])) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r) class OrderToInventoryViewTests(ViewTestCaseMixin, TestCase): url_name = "kfet.order.to_inventory" http_methods = ["GET", "POST"] auth_user = "team" auth_forbidden = [None, "user"] @property def url_kwargs(self): return {"pk": self.order.pk} @property def url_expected(self): return "/k-fet/orders/{}/to_inventory".format(self.order.pk) def get_users_extra(self): return {"team1": create_team("team1", "101", perms=["kfet.order_to_inventory"])} @property def post_data(self): return { # Formset mangaement "form-TOTAL_FORMS": "1", "form-INITIAL_FORMS": "1", "form-MIN_NUM_FORMS": "0", "form-MAX_NUM_FORMS": "1000", # Article 1 "form-0-article": self.article.pk, "form-0-quantity_received": "20", "form-0-price_HT": "", "form-0-TVA": "", "form-0-rights": "", } def setUp(self): super().setUp() category = ArticleCategory.objects.create(name="Category") self.article = Article.objects.create(name="Article", category=category) supplier = Supplier.objects.create(name="Supplier") SupplierArticle.objects.create(supplier=supplier, article=self.article) self.order = Order.objects.create(supplier=supplier) OrderArticle.objects.create( order=self.order, article=self.article, quantity_ordered=24 ) def test_get_ok(self): r = self.client.get(self.url) self.assertEqual(r.status_code, 200) @mock.patch("django.utils.timezone.now") def test_post_ok(self, mock_now): mock_now.return_value = self.now client = Client() client.login(username="team1", password="team1") r = client.post(self.url, self.post_data) self.assertRedirects(r, reverse("kfet.order")) inventory = Inventory.objects.first() self.assertInstanceExpected( inventory, {"by": self.accounts["team1"], "at": self.now, "order": self.order}, ) self.assertQuerysetEqual(inventory.articles.all(), map(repr, [self.article])) compte = InventoryArticle.objects.get(article=self.article) self.assertInstanceExpected( compte, {"stock_old": 0, "stock_new": 20, "stock_error": 0} ) def test_post_forbidden(self): r = self.client.post(self.url, self.post_data) self.assertForbiddenKfet(r)