import json
from datetime import datetime, 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.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",
}
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
}
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 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",
]
),
)
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)