Admin Stripe invoices + recurring subscriptions (customers pay us) #24

Merged
westfarn merged 19 commits from feature/23-admin-stripe-invoices into master 2026-07-31 12:02:41 -07:00
19 changed files with 1403 additions and 2 deletions
+7
View File
@@ -33,6 +33,13 @@ EMAIL_HOST_USER=
EMAIL_HOST_PASSWORD=
EMAIL_PORT=2525
EMAIL_USE_TLS=true
# DEFAULT_FROM_EMAIL=AI ML Operations, LLC <info@aimloperations.com>
# Stripe (test keys from https://dashboard.stripe.com/test/apikeys)
STRIPE_SECRET_KEY=
STRIPE_PUBLISHABLE_KEY=
# From `stripe listen` or Dashboard → Developers → Webhooks
STRIPE_WEBHOOK_SECRET=
# Gunicorn
GUNICORN_WORKERS=2
+12
View File
@@ -34,6 +34,18 @@ EMAIL_HOST_USER=replace-with-smtp-user
EMAIL_HOST_PASSWORD=replace-with-smtp-password
EMAIL_PORT=2525
EMAIL_USE_TLS=true
# DEFAULT_FROM_EMAIL=AI ML Operations, LLC <info@aimloperations.com>
# Stripe (live keys from https://dashboard.stripe.com/apikeys)
# Secret key starts with sk_live_; publishable with pk_live_
STRIPE_SECRET_KEY=replace-with-stripe-secret-key
STRIPE_PUBLISHABLE_KEY=replace-with-stripe-publishable-key
# Dashboard → Developers → Webhooks → endpoint
# URL: https://aimloperations.com/financial/stripe/webhook/
# Events: invoice.paid, invoice.payment_failed, invoice.finalized,
# customer.subscription.updated, customer.subscription.deleted,
# checkout.session.completed
STRIPE_WEBHOOK_SECRET=replace-with-stripe-webhook-signing-secret
# Gunicorn
GUNICORN_WORKERS=2
@@ -200,6 +200,15 @@ EMAIL_HOST_USER = env("EMAIL_HOST_USER", "")
EMAIL_HOST_PASSWORD = env("EMAIL_HOST_PASSWORD", "")
EMAIL_PORT = int(env("EMAIL_PORT", "2525"))
EMAIL_USE_TLS = env_bool("EMAIL_USE_TLS", True)
DEFAULT_FROM_EMAIL = env(
"DEFAULT_FROM_EMAIL",
"AI ML Operations, LLC <info@aimloperations.com>",
)
# Stripe (invoices / subscriptions — customers pay us)
STRIPE_SECRET_KEY = env("STRIPE_SECRET_KEY", "")
STRIPE_PUBLISHABLE_KEY = env("STRIPE_PUBLISHABLE_KEY", "")
STRIPE_WEBHOOK_SECRET = env("STRIPE_WEBHOOK_SECRET", "")
LOGIN_REDIRECT_URL = "/"
LOGOUT_REDIRECT_URL = "/"
+27 -1
View File
@@ -1,5 +1,15 @@
from django.contrib import admin
from .models import Contract, Employee, ChargeNumber, TimeCard, TimeCardCell, UserProfile
from .models import (
Contract,
Employee,
ChargeNumber,
TimeCard,
TimeCardCell,
UserProfile,
BillingCustomer,
Invoice,
RecurringSubscription,
)
class ContractAdmin(admin.ModelAdmin):
pass
@@ -20,9 +30,25 @@ class TimeCardAdmin(admin.ModelAdmin):
class TimeCardCellAdmin(admin.ModelAdmin):
pass
class BillingCustomerAdmin(admin.ModelAdmin):
list_display = ("name", "email", "company", "stripe_customer_id")
search_fields = ("name", "email", "company")
class InvoiceAdmin(admin.ModelAdmin):
list_display = ("id", "customer", "description", "amount_cents", "status", "created")
list_filter = ("status",)
search_fields = ("description", "stripe_invoice_id", "customer__email")
class RecurringSubscriptionAdmin(admin.ModelAdmin):
list_display = ("id", "customer", "description", "amount_cents", "interval", "status")
list_filter = ("status", "interval")
admin.site.register(Contract, ContractAdmin)
admin.site.register(Employee, EmployeeAdmin)
admin.site.register(UserProfile, UserProfileAdmin)
admin.site.register(ChargeNumber, ChargeNumberAdmin)
admin.site.register(TimeCard, TimeCardAdmin)
admin.site.register(TimeCardCell, TimeCardCellAdmin)
admin.site.register(BillingCustomer, BillingCustomerAdmin)
admin.site.register(Invoice, InvoiceAdmin)
admin.site.register(RecurringSubscription, RecurringSubscriptionAdmin)
+154 -1
View File
@@ -2,7 +2,19 @@ import datetime
from django import forms
from django.contrib.auth.models import User
from django.forms import ModelForm
from .models import Employee, Contract, ChargeNumber, TimeCardCell, AddressModel, UserProfile, set_user_type
from .models import (
Employee,
Contract,
ChargeNumber,
TimeCardCell,
AddressModel,
UserProfile,
set_user_type,
BillingCustomer,
Invoice,
RecurringSubscription,
)
from .stripe_billing import dollars_to_cents
class NewEmployeeForm(ModelForm):
first_name = forms.CharField(max_length=30, required=False, label="First Name")
@@ -99,3 +111,144 @@ class TimeLogForm(ModelForm):
cleaned_data['hour'] = duration
return cleaned_data
class BillingCustomerForm(ModelForm):
class Meta:
model = BillingCustomer
fields = ["name", "email", "company", "notes"]
class OneOffInvoiceForm(forms.Form):
customer = forms.ModelChoiceField(
queryset=BillingCustomer.objects.all(),
required=False,
help_text="Pick an existing customer, or fill New customer fields below.",
)
new_customer_name = forms.CharField(max_length=200, required=False, label="New customer name")
new_customer_email = forms.EmailField(required=False, label="New customer email")
new_customer_company = forms.CharField(
max_length=200, required=False, label="New customer company"
)
description = forms.CharField(max_length=500)
amount = forms.DecimalField(
max_digits=10,
decimal_places=2,
min_value=0.50,
label="Amount (USD)",
help_text="Minimum $0.50",
)
due_date = forms.DateField(
required=False,
widget=forms.DateInput(attrs={"type": "date"}),
)
notes = forms.CharField(required=False, widget=forms.Textarea(attrs={"rows": 3}))
send_email = forms.BooleanField(
required=False,
initial=True,
label="Email payment link to customer",
)
def clean(self):
cleaned = super().clean()
customer = cleaned.get("customer")
name = (cleaned.get("new_customer_name") or "").strip()
email = (cleaned.get("new_customer_email") or "").strip()
if not customer and not (name and email):
raise forms.ValidationError(
"Select an existing customer or provide new customer name and email."
)
return cleaned
def resolve_customer(self, user):
customer = self.cleaned_data.get("customer")
if customer:
return customer
customer = BillingCustomer(
name=self.cleaned_data["new_customer_name"].strip(),
email=self.cleaned_data["new_customer_email"].strip(),
company=(self.cleaned_data.get("new_customer_company") or "").strip(),
created_by=user,
last_modified_BY=user,
)
customer.save()
return customer
def build_invoice(self, user) -> Invoice:
customer = self.resolve_customer(user)
return Invoice(
customer=customer,
description=self.cleaned_data["description"],
amount_cents=dollars_to_cents(self.cleaned_data["amount"]),
due_date=self.cleaned_data.get("due_date"),
notes=self.cleaned_data.get("notes") or "",
created_by=user,
last_modified_BY=user,
)
class RecurringSubscriptionForm(forms.Form):
customer = forms.ModelChoiceField(
queryset=BillingCustomer.objects.all(),
required=False,
help_text="Pick an existing customer, or fill New customer fields below.",
)
new_customer_name = forms.CharField(max_length=200, required=False, label="New customer name")
new_customer_email = forms.EmailField(required=False, label="New customer email")
new_customer_company = forms.CharField(
max_length=200, required=False, label="New customer company"
)
description = forms.CharField(max_length=500)
amount = forms.DecimalField(
max_digits=10,
decimal_places=2,
min_value=0.50,
label="Amount per period (USD)",
)
interval = forms.ChoiceField(
choices=RecurringSubscription.Interval.choices,
initial=RecurringSubscription.Interval.MONTH,
)
notes = forms.CharField(required=False, widget=forms.Textarea(attrs={"rows": 3}))
send_email = forms.BooleanField(
required=False,
initial=True,
label="Email checkout / payment link to customer",
)
def clean(self):
cleaned = super().clean()
customer = cleaned.get("customer")
name = (cleaned.get("new_customer_name") or "").strip()
email = (cleaned.get("new_customer_email") or "").strip()
if not customer and not (name and email):
raise forms.ValidationError(
"Select an existing customer or provide new customer name and email."
)
return cleaned
def resolve_customer(self, user):
customer = self.cleaned_data.get("customer")
if customer:
return customer
customer = BillingCustomer(
name=self.cleaned_data["new_customer_name"].strip(),
email=self.cleaned_data["new_customer_email"].strip(),
company=(self.cleaned_data.get("new_customer_company") or "").strip(),
created_by=user,
last_modified_BY=user,
)
customer.save()
return customer
def build_subscription(self, user) -> RecurringSubscription:
customer = self.resolve_customer(user)
return RecurringSubscription(
customer=customer,
description=self.cleaned_data["description"],
amount_cents=dollars_to_cents(self.cleaned_data["amount"]),
interval=self.cleaned_data["interval"],
notes=self.cleaned_data.get("notes") or "",
created_by=user,
last_modified_BY=user,
)
+262
View File
@@ -0,0 +1,262 @@
"""Admin invoice / subscription views and Stripe webhook."""
import json
import logging
from django.conf import settings
from django.contrib import messages
from django.http import HttpResponse, HttpResponseBadRequest
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils import timezone
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from .forms import OneOffInvoiceForm, RecurringSubscriptionForm
from .models import BillingCustomer, Invoice, RecurringSubscription
from .permissions import financial_admin_required
from .stripe_billing import (
StripeAPIError,
StripeNotConfigured,
apply_checkout_session_completed,
apply_stripe_invoice_event,
apply_subscription_event,
construct_webhook_event,
create_one_off_invoice,
create_subscription_checkout,
send_pay_link_email,
)
logger = logging.getLogger(__name__)
@financial_admin_required
def invoice_list(request):
invoices = Invoice.objects.select_related("customer").all()[:100]
subscriptions = RecurringSubscription.objects.select_related("customer").all()[:100]
customers = BillingCustomer.objects.all()[:50]
return render(
request,
"financial/invoices.html",
{
"invoices": invoices,
"subscriptions": subscriptions,
"customers": customers,
"stripe_configured": bool(getattr(settings, "STRIPE_SECRET_KEY", "")),
},
)
@financial_admin_required
def invoice_new(request):
if request.method == "POST":
form = OneOffInvoiceForm(request.POST)
if form.is_valid():
invoice = form.build_invoice(request.user)
invoice.save()
try:
create_one_off_invoice(invoice)
if form.cleaned_data.get("send_email") and invoice.hosted_invoice_url:
send_pay_link_email(
to_email=invoice.customer.email,
customer_name=invoice.customer.name,
description=invoice.description,
amount_dollars=invoice.amount_dollars,
pay_url=invoice.hosted_invoice_url,
kind="invoice",
)
invoice.pay_link_emailed_at = timezone.now()
invoice.save(update_fields=["pay_link_emailed_at"])
messages.success(
request,
f"Invoice created and payment link emailed to {invoice.customer.email}.",
)
else:
messages.success(
request,
"Invoice created in Stripe. Payment link saved on the detail page.",
)
return redirect("invoice_detail", invoice_id=invoice.pk)
except StripeNotConfigured as exc:
messages.error(request, str(exc))
except StripeAPIError as exc:
messages.error(request, f"Stripe error: {exc.user_message}")
else:
form = OneOffInvoiceForm()
return render(request, "financial/invoice_new.html", {"form": form})
@financial_admin_required
def invoice_detail(request, invoice_id):
invoice = get_object_or_404(
Invoice.objects.select_related("customer"), pk=invoice_id
)
return render(request, "financial/invoice_detail.html", {"invoice": invoice})
@financial_admin_required
def invoice_resend_email(request, invoice_id):
invoice = get_object_or_404(
Invoice.objects.select_related("customer"), pk=invoice_id
)
if request.method != "POST":
return redirect("invoice_detail", invoice_id=invoice.pk)
if not invoice.hosted_invoice_url:
messages.error(request, "No payment link available for this invoice.")
return redirect("invoice_detail", invoice_id=invoice.pk)
try:
send_pay_link_email(
to_email=invoice.customer.email,
customer_name=invoice.customer.name,
description=invoice.description,
amount_dollars=invoice.amount_dollars,
pay_url=invoice.hosted_invoice_url,
kind="invoice",
)
invoice.pay_link_emailed_at = timezone.now()
invoice.save(update_fields=["pay_link_emailed_at"])
messages.success(request, f"Payment link resent to {invoice.customer.email}.")
except Exception as exc: # noqa: BLE001 — surface mail errors to admin UI
messages.error(request, f"Email failed: {exc}")
return redirect("invoice_detail", invoice_id=invoice.pk)
@financial_admin_required
def subscription_new(request):
if request.method == "POST":
form = RecurringSubscriptionForm(request.POST)
if form.is_valid():
subscription = form.build_subscription(request.user)
subscription.save()
success_url = (
request.build_absolute_uri(
reverse("subscription_detail", args=[subscription.pk])
)
+ "?checkout=success"
)
cancel_url = (
request.build_absolute_uri(
reverse("subscription_detail", args=[subscription.pk])
)
+ "?checkout=cancel"
)
try:
create_subscription_checkout(
subscription,
success_url=success_url,
cancel_url=cancel_url,
)
if form.cleaned_data.get("send_email") and subscription.checkout_url:
send_pay_link_email(
to_email=subscription.customer.email,
customer_name=subscription.customer.name,
description=subscription.description,
amount_dollars=subscription.amount_dollars,
pay_url=subscription.checkout_url,
kind="subscription",
)
subscription.pay_link_emailed_at = timezone.now()
subscription.save(update_fields=["pay_link_emailed_at"])
messages.success(
request,
f"Subscription checkout created and emailed to {subscription.customer.email}.",
)
else:
messages.success(
request,
"Subscription checkout created. Share the link from the detail page.",
)
return redirect("subscription_detail", subscription_id=subscription.pk)
except StripeNotConfigured as exc:
messages.error(request, str(exc))
except StripeAPIError as exc:
messages.error(request, f"Stripe error: {exc.user_message}")
else:
form = RecurringSubscriptionForm()
return render(request, "financial/subscription_new.html", {"form": form})
@financial_admin_required
def subscription_detail(request, subscription_id):
subscription = get_object_or_404(
RecurringSubscription.objects.select_related("customer"),
pk=subscription_id,
)
return render(
request,
"financial/subscription_detail.html",
{"subscription": subscription},
)
@financial_admin_required
def subscription_resend_email(request, subscription_id):
subscription = get_object_or_404(
RecurringSubscription.objects.select_related("customer"),
pk=subscription_id,
)
if request.method != "POST":
return redirect("subscription_detail", subscription_id=subscription.pk)
if not subscription.checkout_url:
messages.error(request, "No checkout link available for this subscription.")
return redirect("subscription_detail", subscription_id=subscription.pk)
try:
send_pay_link_email(
to_email=subscription.customer.email,
customer_name=subscription.customer.name,
description=subscription.description,
amount_dollars=subscription.amount_dollars,
pay_url=subscription.checkout_url,
kind="subscription",
)
subscription.pay_link_emailed_at = timezone.now()
subscription.save(update_fields=["pay_link_emailed_at"])
messages.success(
request, f"Checkout link resent to {subscription.customer.email}."
)
except Exception as exc: # noqa: BLE001
messages.error(request, f"Email failed: {exc}")
return redirect("subscription_detail", subscription_id=subscription.pk)
@csrf_exempt
@require_POST
def stripe_webhook(request):
"""Receive Stripe webhooks; verify signature when secret is configured."""
payload = request.body
sig_header = request.META.get("HTTP_STRIPE_SIGNATURE", "")
webhook_secret = getattr(settings, "STRIPE_WEBHOOK_SECRET", "") or ""
try:
if webhook_secret:
event = construct_webhook_event(payload, sig_header, webhook_secret)
else:
logger.warning(
"STRIPE_WEBHOOK_SECRET unset — accepting unsigned webhook (dev only)"
)
event = json.loads(payload.decode("utf-8"))
except (ValueError, json.JSONDecodeError) as exc:
return HttpResponseBadRequest(str(exc))
event_type = event.get("type")
data_object = event.get("data", {}).get("object", {})
if event_type in {
"invoice.paid",
"invoice.payment_failed",
"invoice.finalized",
"invoice.voided",
"invoice.marked_uncollectible",
}:
apply_stripe_invoice_event(data_object)
elif event_type in {
"customer.subscription.updated",
"customer.subscription.deleted",
}:
apply_subscription_event(data_object)
elif event_type == "checkout.session.completed":
apply_checkout_session_completed(data_object)
return HttpResponse(status=200)
@@ -0,0 +1,88 @@
# Generated by Django 5.0 on 2026-07-31 11:31
import django.db.models.deletion
import django.utils.timezone
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('financial', '0015_userprofile'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='BillingCustomer',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created', models.DateTimeField(default=django.utils.timezone.now)),
('last_modified', models.DateTimeField(default=django.utils.timezone.now)),
('slug', models.SlugField(blank=True, null=True, unique=True)),
('name', models.CharField(max_length=200)),
('email', models.EmailField(max_length=254)),
('company', models.CharField(blank=True, max_length=200)),
('stripe_customer_id', models.CharField(blank=True, default='', max_length=255)),
('notes', models.TextField(blank=True)),
('created_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
('last_modified_BY', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['name'],
},
),
migrations.CreateModel(
name='Invoice',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created', models.DateTimeField(default=django.utils.timezone.now)),
('last_modified', models.DateTimeField(default=django.utils.timezone.now)),
('slug', models.SlugField(blank=True, null=True, unique=True)),
('description', models.CharField(max_length=500)),
('amount_cents', models.PositiveIntegerField(help_text='Amount in cents (USD)')),
('currency', models.CharField(default='usd', max_length=3)),
('status', models.CharField(choices=[('draft', 'Draft'), ('open', 'Open'), ('paid', 'Paid'), ('void', 'Void'), ('uncollectible', 'Uncollectible'), ('failed', 'Failed')], default='draft', max_length=20)),
('due_date', models.DateField(blank=True, null=True)),
('notes', models.TextField(blank=True)),
('stripe_invoice_id', models.CharField(blank=True, default='', max_length=255)),
('hosted_invoice_url', models.URLField(blank=True, default='', max_length=500)),
('invoice_pdf_url', models.URLField(blank=True, default='', max_length=500)),
('pay_link_emailed_at', models.DateTimeField(blank=True, null=True)),
('created_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
('customer', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='invoices', to='financial.billingcustomer')),
('last_modified_BY', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-created'],
},
),
migrations.CreateModel(
name='RecurringSubscription',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created', models.DateTimeField(default=django.utils.timezone.now)),
('last_modified', models.DateTimeField(default=django.utils.timezone.now)),
('slug', models.SlugField(blank=True, null=True, unique=True)),
('description', models.CharField(max_length=500)),
('amount_cents', models.PositiveIntegerField(help_text='Amount per period in cents (USD)')),
('currency', models.CharField(default='usd', max_length=3)),
('interval', models.CharField(choices=[('month', 'Monthly'), ('year', 'Yearly'), ('week', 'Weekly')], default='month', max_length=10)),
('status', models.CharField(choices=[('incomplete', 'Incomplete'), ('active', 'Active'), ('past_due', 'Past due'), ('canceled', 'Canceled'), ('unpaid', 'Unpaid'), ('trialing', 'Trialing'), ('paused', 'Paused')], default='incomplete', max_length=20)),
('notes', models.TextField(blank=True)),
('stripe_product_id', models.CharField(blank=True, default='', max_length=255)),
('stripe_price_id', models.CharField(blank=True, default='', max_length=255)),
('stripe_subscription_id', models.CharField(blank=True, default='', max_length=255)),
('checkout_session_id', models.CharField(blank=True, default='', max_length=255)),
('checkout_url', models.URLField(blank=True, default='', max_length=500)),
('pay_link_emailed_at', models.DateTimeField(blank=True, null=True)),
('created_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
('customer', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='subscriptions', to='financial.billingcustomer')),
('last_modified_BY', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-created'],
},
),
]
+102
View File
@@ -302,6 +302,108 @@ class TimeCardCell(IdMixin, TimeMixin):
charge_number = models.ForeignKey(ChargeNumber, on_delete=models.CASCADE, null=True, blank=True)
class BillingCustomer(IdMixin, TimeMixin):
"""Customer who pays us (one-off invoices or subscriptions)."""
name = models.CharField(max_length=200)
email = models.EmailField()
company = models.CharField(max_length=200, blank=True)
stripe_customer_id = models.CharField(max_length=255, blank=True, default="")
notes = models.TextField(blank=True)
class Meta:
ordering = ["name"]
def __str__(self):
return f"{self.name} <{self.email}>"
class Invoice(IdMixin, TimeMixin):
"""One-off invoice — customer pays us via Stripe hosted invoice URL."""
class Status(models.TextChoices):
DRAFT = "draft", "Draft"
OPEN = "open", "Open"
PAID = "paid", "Paid"
VOID = "void", "Void"
UNCOLLECTIBLE = "uncollectible", "Uncollectible"
FAILED = "failed", "Failed"
customer = models.ForeignKey(
BillingCustomer, on_delete=models.PROTECT, related_name="invoices"
)
description = models.CharField(max_length=500)
amount_cents = models.PositiveIntegerField(help_text="Amount in cents (USD)")
currency = models.CharField(max_length=3, default="usd")
status = models.CharField(
max_length=20, choices=Status.choices, default=Status.DRAFT
)
due_date = models.DateField(null=True, blank=True)
notes = models.TextField(blank=True)
stripe_invoice_id = models.CharField(max_length=255, blank=True, default="")
hosted_invoice_url = models.URLField(max_length=500, blank=True, default="")
invoice_pdf_url = models.URLField(max_length=500, blank=True, default="")
pay_link_emailed_at = models.DateTimeField(null=True, blank=True)
class Meta:
ordering = ["-created"]
def __str__(self):
return f"Invoice {self.pk}{self.customer}{self.status}"
@property
def amount_dollars(self):
return self.amount_cents / 100.0
class RecurringSubscription(IdMixin, TimeMixin):
"""Monthly (or custom) recurring charge — customer pays us via Stripe Subscription."""
class Status(models.TextChoices):
INCOMPLETE = "incomplete", "Incomplete"
ACTIVE = "active", "Active"
PAST_DUE = "past_due", "Past due"
CANCELED = "canceled", "Canceled"
UNPAID = "unpaid", "Unpaid"
TRIALING = "trialing", "Trialing"
PAUSED = "paused", "Paused"
class Interval(models.TextChoices):
MONTH = "month", "Monthly"
YEAR = "year", "Yearly"
WEEK = "week", "Weekly"
customer = models.ForeignKey(
BillingCustomer, on_delete=models.PROTECT, related_name="subscriptions"
)
description = models.CharField(max_length=500)
amount_cents = models.PositiveIntegerField(help_text="Amount per period in cents (USD)")
currency = models.CharField(max_length=3, default="usd")
interval = models.CharField(
max_length=10, choices=Interval.choices, default=Interval.MONTH
)
status = models.CharField(
max_length=20, choices=Status.choices, default=Status.INCOMPLETE
)
notes = models.TextField(blank=True)
stripe_product_id = models.CharField(max_length=255, blank=True, default="")
stripe_price_id = models.CharField(max_length=255, blank=True, default="")
stripe_subscription_id = models.CharField(max_length=255, blank=True, default="")
checkout_session_id = models.CharField(max_length=255, blank=True, default="")
checkout_url = models.URLField(max_length=500, blank=True, default="")
pay_link_emailed_at = models.DateTimeField(null=True, blank=True)
class Meta:
ordering = ["-created"]
def __str__(self):
return f"Subscription {self.pk}{self.customer}{self.status}"
@property
def amount_dollars(self):
return self.amount_cents / 100.0
def set_user_type(user, user_type):
"""Set user type and sync the Employee record (mutually exclusive types)."""
user.__dict__.pop("profile", None)
+383
View File
@@ -0,0 +1,383 @@
"""Stripe REST helpers (stdlib HTTP) for invoices and subscriptions.
Customers pay us — one-off hosted invoices and Checkout subscription sessions.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import logging
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from decimal import Decimal
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.template.loader import get_template
from django.utils import timezone
from .models import BillingCustomer, Invoice, RecurringSubscription
logger = logging.getLogger(__name__)
STRIPE_API_BASE = "https://api.stripe.com/v1"
class StripeNotConfigured(Exception):
"""Raised when Stripe secret key is missing."""
class StripeAPIError(Exception):
"""Raised when Stripe API returns an error."""
def __init__(self, message: str, status: int | None = None):
super().__init__(message)
self.status = status
self.user_message = message
def _secret_key() -> str:
key = getattr(settings, "STRIPE_SECRET_KEY", "") or ""
if not key:
raise StripeNotConfigured(
"STRIPE_SECRET_KEY is not set. Add it to the environment."
)
return key
def _stripe_request(method: str, path: str, data: dict | None = None) -> dict:
"""Call Stripe API with form-encoded body (Stripe default)."""
url = f"{STRIPE_API_BASE}{path}"
body = None
headers = {
"Authorization": f"Bearer {_secret_key()}",
}
if data is not None:
body = urllib.parse.urlencode(_flatten_params(data)).encode("utf-8")
headers["Content-Type"] = "application/x-www-form-urlencoded"
request = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
raw = exc.read().decode("utf-8", errors="replace")
try:
payload = json.loads(raw)
message = payload.get("error", {}).get("message") or raw
except json.JSONDecodeError:
message = raw or str(exc)
raise StripeAPIError(message, status=exc.code) from exc
except urllib.error.URLError as exc:
raise StripeAPIError(f"Could not reach Stripe: {exc.reason}") from exc
def _flatten_params(data: dict, prefix: str = "") -> list[tuple[str, str]]:
"""Flatten nested dicts/lists into Stripe form keys (a[b]=1, items[0][x]=y)."""
items: list[tuple[str, str]] = []
for key, value in data.items():
full_key = f"{prefix}[{key}]" if prefix else str(key)
if value is None:
continue
if isinstance(value, dict):
items.extend(_flatten_params(value, full_key))
elif isinstance(value, (list, tuple)):
for index, entry in enumerate(value):
indexed = f"{full_key}[{index}]"
if isinstance(entry, dict):
items.extend(_flatten_params(entry, indexed))
else:
items.append((indexed, str(entry)))
elif isinstance(value, bool):
items.append((full_key, "true" if value else "false"))
else:
items.append((full_key, str(value)))
return items
def construct_webhook_event(payload: bytes, sig_header: str, secret: str) -> dict:
"""Verify Stripe-Signature and return event dict."""
if not sig_header:
raise ValueError("Missing Stripe-Signature header")
elements = {part.split("=", 1)[0]: part.split("=", 1)[1] for part in sig_header.split(",") if "=" in part}
timestamp = elements.get("t")
signatures = [part.split("=", 1)[1] for part in sig_header.split(",") if part.startswith("v1=")]
if not timestamp or not signatures:
raise ValueError("Invalid Stripe-Signature header")
signed_payload = f"{timestamp}.{payload.decode('utf-8')}".encode("utf-8")
expected = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
if not any(hmac.compare_digest(expected, candidate) for candidate in signatures):
raise ValueError("Webhook signature verification failed")
# Reject stale timestamps (>5 minutes)
if abs(time.time() - int(timestamp)) > 300:
raise ValueError("Webhook timestamp outside tolerance")
return json.loads(payload.decode("utf-8"))
def ensure_stripe_customer(customer: BillingCustomer) -> str:
"""Create Stripe Customer if needed; return stripe_customer_id."""
if customer.stripe_customer_id:
return customer.stripe_customer_id
stripe_customer = _stripe_request(
"POST",
"/customers",
{
"email": customer.email,
"name": customer.name,
"metadata": {
"local_customer_id": str(customer.pk),
"company": customer.company or "",
},
},
)
customer.stripe_customer_id = stripe_customer["id"]
customer.save(update_fields=["stripe_customer_id", "last_modified"])
return customer.stripe_customer_id
def create_one_off_invoice(invoice: Invoice) -> Invoice:
"""Create + finalize a Stripe Invoice; store hosted pay URL."""
stripe_customer_id = ensure_stripe_customer(invoice.customer)
_stripe_request(
"POST",
"/invoiceitems",
{
"customer": stripe_customer_id,
"amount": invoice.amount_cents,
"currency": invoice.currency,
"description": invoice.description,
},
)
create_data: dict = {
"customer": stripe_customer_id,
"collection_method": "send_invoice",
"auto_advance": True,
"metadata": {"local_invoice_id": str(invoice.pk)},
}
if invoice.due_date:
due_dt = datetime.combine(invoice.due_date, datetime.min.time())
if timezone.is_naive(due_dt):
due_dt = timezone.make_aware(due_dt, timezone.get_current_timezone())
create_data["due_date"] = int(due_dt.timestamp())
else:
create_data["days_until_due"] = 30
stripe_invoice = _stripe_request("POST", "/invoices", create_data)
stripe_invoice = _stripe_request(
"POST", f"/invoices/{stripe_invoice['id']}/finalize", {}
)
invoice.stripe_invoice_id = stripe_invoice["id"]
invoice.hosted_invoice_url = stripe_invoice.get("hosted_invoice_url") or ""
invoice.invoice_pdf_url = stripe_invoice.get("invoice_pdf") or ""
status = stripe_invoice.get("status") or Invoice.Status.OPEN
if status in dict(Invoice.Status.choices):
invoice.status = status
invoice.last_modified = timezone.now()
invoice.save(
update_fields=[
"stripe_invoice_id",
"hosted_invoice_url",
"invoice_pdf_url",
"status",
"last_modified",
]
)
return invoice
def create_subscription_checkout(
subscription: RecurringSubscription,
*,
success_url: str,
cancel_url: str,
) -> RecurringSubscription:
"""Create Stripe Product/Price + Checkout Session (subscription mode)."""
stripe_customer_id = ensure_stripe_customer(subscription.customer)
product = _stripe_request(
"POST",
"/products",
{
"name": subscription.description,
"metadata": {"local_subscription_id": str(subscription.pk)},
},
)
price = _stripe_request(
"POST",
"/prices",
{
"product": product["id"],
"unit_amount": subscription.amount_cents,
"currency": subscription.currency,
"recurring": {"interval": subscription.interval},
},
)
session = _stripe_request(
"POST",
"/checkout/sessions",
{
"mode": "subscription",
"customer": stripe_customer_id,
"line_items": [{"price": price["id"], "quantity": 1}],
"success_url": success_url,
"cancel_url": cancel_url,
"metadata": {"local_subscription_id": str(subscription.pk)},
"subscription_data": {
"metadata": {"local_subscription_id": str(subscription.pk)},
},
},
)
subscription.stripe_product_id = product["id"]
subscription.stripe_price_id = price["id"]
subscription.checkout_session_id = session["id"]
subscription.checkout_url = session.get("url") or ""
subscription.status = RecurringSubscription.Status.INCOMPLETE
subscription.last_modified = timezone.now()
subscription.save(
update_fields=[
"stripe_product_id",
"stripe_price_id",
"checkout_session_id",
"checkout_url",
"status",
"last_modified",
]
)
return subscription
def dollars_to_cents(amount: Decimal | float | str) -> int:
return int(round(Decimal(str(amount)) * 100))
def send_pay_link_email(
*,
to_email: str,
customer_name: str,
description: str,
amount_dollars: float,
pay_url: str,
kind: str = "invoice",
) -> None:
"""Email customer a Stripe payment / checkout link."""
subject = (
f"Payment request from AI ML Operations — {description}"
if kind == "invoice"
else f"Subscription signup — {description}"
)
from_email = getattr(
settings,
"DEFAULT_FROM_EMAIL",
"AI ML Operations, LLC <info@aimloperations.com>",
)
context = {
"customer_name": customer_name,
"description": description,
"amount_dollars": f"{amount_dollars:.2f}",
"pay_url": pay_url,
"kind": kind,
"subject": subject,
}
html_content = get_template("emails/invoice_pay_link.html").render(context)
text_content = get_template("emails/invoice_pay_link.txt").render(context)
msg = EmailMultiAlternatives(subject, text_content, from_email, [to_email])
msg.attach_alternative(html_content, "text/html")
msg.send(fail_silently=False)
def apply_stripe_invoice_event(stripe_invoice: dict) -> Invoice | None:
"""Sync local Invoice from Stripe invoice object dict."""
local_id = (stripe_invoice.get("metadata") or {}).get("local_invoice_id")
stripe_id = stripe_invoice.get("id")
invoice = None
if local_id:
invoice = Invoice.objects.filter(pk=local_id).first()
if invoice is None and stripe_id:
invoice = Invoice.objects.filter(stripe_invoice_id=stripe_id).first()
if invoice is None:
return None
status = stripe_invoice.get("status") or invoice.status
if status in dict(Invoice.Status.choices):
invoice.status = status
invoice.hosted_invoice_url = (
stripe_invoice.get("hosted_invoice_url") or invoice.hosted_invoice_url
)
invoice.invoice_pdf_url = (
stripe_invoice.get("invoice_pdf") or invoice.invoice_pdf_url
)
if stripe_id:
invoice.stripe_invoice_id = stripe_id
invoice.last_modified = timezone.now()
invoice.save(
update_fields=[
"status",
"hosted_invoice_url",
"invoice_pdf_url",
"stripe_invoice_id",
"last_modified",
]
)
return invoice
def apply_subscription_event(stripe_sub: dict) -> RecurringSubscription | None:
"""Sync local RecurringSubscription from Stripe subscription object."""
local_id = (stripe_sub.get("metadata") or {}).get("local_subscription_id")
stripe_id = stripe_sub.get("id")
sub = None
if local_id:
sub = RecurringSubscription.objects.filter(pk=local_id).first()
if sub is None and stripe_id:
sub = RecurringSubscription.objects.filter(
stripe_subscription_id=stripe_id
).first()
if sub is None:
return None
status = stripe_sub.get("status") or sub.status
if status in dict(RecurringSubscription.Status.choices):
sub.status = status
if stripe_id:
sub.stripe_subscription_id = stripe_id
sub.last_modified = timezone.now()
sub.save(update_fields=["status", "stripe_subscription_id", "last_modified"])
return sub
def apply_checkout_session_completed(session: dict) -> RecurringSubscription | None:
"""Attach subscription id after Checkout completes."""
if session.get("mode") != "subscription":
return None
local_id = (session.get("metadata") or {}).get("local_subscription_id")
sub = None
if local_id:
sub = RecurringSubscription.objects.filter(pk=local_id).first()
if sub is None and session.get("id"):
sub = RecurringSubscription.objects.filter(
checkout_session_id=session["id"]
).first()
if sub is None:
return None
stripe_sub_id = session.get("subscription")
if stripe_sub_id:
sub.stripe_subscription_id = stripe_sub_id
sub.status = RecurringSubscription.Status.ACTIVE
sub.last_modified = timezone.now()
sub.save(update_fields=["stripe_subscription_id", "status", "last_modified"])
return sub
@@ -0,0 +1,48 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{{ subject }}</title>
<style>
body { margin: 0; padding: 0; font-family: Arial, sans-serif; background-color: #f4f4f4; }
.email-container { max-width: 600px; margin: 0 auto; background-color: #ffffff; border: 1px solid #dddddd; }
.header { background-color: #0b3d5c; color: #ffffff; padding: 20px; text-align: center; }
.content { padding: 24px; color: #333333; font-size: 15px; line-height: 1.5; }
.btn {
display: inline-block; margin: 20px 0; padding: 12px 24px;
background-color: #0b3d5c; color: #ffffff !important; text-decoration: none;
border-radius: 4px; font-weight: bold;
}
.footer { background-color: #f4f4f4; color: #777777; text-align: center; padding: 12px; font-size: 12px; }
.amount { font-size: 22px; font-weight: bold; margin: 12px 0; }
</style>
</head>
<body>
<div class="email-container">
<div class="header">
<h1>AI ML Operations</h1>
</div>
<div class="content">
<p>Hello {{ customer_name }},</p>
{% if kind == "subscription" %}
<p>Please complete checkout to start your recurring payment:</p>
{% else %}
<p>You have a new invoice ready for payment:</p>
{% endif %}
<p><strong>{{ description }}</strong></p>
<p class="amount">${{ amount_dollars }} USD</p>
<p>
<a class="btn" href="{{ pay_url }}">
{% if kind == "subscription" %}Complete checkout{% else %}Pay invoice{% endif %}
</a>
</p>
<p style="font-size: 13px; color: #666;">Or open this link:<br>{{ pay_url }}</p>
<p>Thank you,<br>AI ML Operations, LLC</p>
</div>
<div class="footer">
<p>This is an automated message. Please do not reply to this email.</p>
</div>
</div>
</body>
</html>
@@ -0,0 +1,12 @@
Hello {{ customer_name }},
{% if kind == "subscription" %}Please complete checkout to start your recurring payment:{% else %}You have a new invoice ready for payment:{% endif %}
{{ description }}
Amount: ${{ amount_dollars }} USD
{% if kind == "subscription" %}Checkout link:{% else %}Pay invoice:{% endif %}
{{ pay_url }}
Thank you,
AI ML Operations, LLC
@@ -29,6 +29,11 @@
<span class="card-title">Manage Users</span>
<p class="card-text">Set Employee or Client type for user accounts.</p>
</a>
<a href="{% url 'invoice_list' %}" class="card"
data-tianji-event="financial_nav" data-tianji-event-destination="invoices">
<span class="card-title">Invoices</span>
<p class="card-text">Create one-off invoices and monthly subscriptions (Stripe).</p>
</a>
{% endif %}
{% if can_write_financials %}
<a href="{% url 'Timekeeping' %}" class="card"
@@ -0,0 +1,38 @@
{% extends "base.html" %}
{% load static %}
{% block title %}Invoice {{ invoice.pk }} - AI ML Operations{% endblock %}
{% block content %}
<div class="section">
<div class="container">
<div style="margin-bottom: 2rem;">
<a href="{% url 'invoice_list' %}" class="btn">Back to Invoices</a>
</div>
<h1 class="section-title" style="text-align: left;">Invoice #{{ invoice.pk }}</h1>
<div class="card" style="max-width: 720px;">
<p><strong>Customer:</strong> {{ invoice.customer.name }} &lt;{{ invoice.customer.email }}&gt;</p>
<p><strong>Description:</strong> {{ invoice.description }}</p>
<p><strong>Amount:</strong> ${{ invoice.amount_dollars|floatformat:2 }} {{ invoice.currency|upper }}</p>
<p><strong>Status:</strong> {{ invoice.get_status_display }}</p>
<p><strong>Due:</strong> {{ invoice.due_date|default:"—" }}</p>
<p><strong>Stripe invoice:</strong> {{ invoice.stripe_invoice_id|default:"—" }}</p>
{% if invoice.hosted_invoice_url %}
<p><strong>Pay link:</strong> <a href="{{ invoice.hosted_invoice_url }}" target="_blank" rel="noopener">Open hosted invoice</a></p>
{% endif %}
{% if invoice.invoice_pdf_url %}
<p><strong>PDF:</strong> <a href="{{ invoice.invoice_pdf_url }}" target="_blank" rel="noopener">Download</a></p>
{% endif %}
<p><strong>Email sent:</strong> {{ invoice.pay_link_emailed_at|default:"Not yet" }}</p>
{% if invoice.notes %}<p><strong>Notes:</strong> {{ invoice.notes }}</p>{% endif %}
{% if invoice.hosted_invoice_url %}
<form method="post" action="{% url 'invoice_resend_email' invoice.pk %}" style="margin-top: 1.5rem;">
{% csrf_token %}
<button type="submit" class="btn">Resend payment email</button>
</form>
{% endif %}
</div>
</div>
</div>
{% endblock %}
@@ -0,0 +1,25 @@
{% extends "base.html" %}
{% load static %}
{% block title %}New Invoice - AI ML Operations{% endblock %}
{% block content %}
<div class="section">
<div class="container">
<div style="margin-bottom: 2rem;">
<a href="{% url 'invoice_list' %}" class="btn">Back to Invoices</a>
</div>
<h1 class="section-title" style="text-align: left;">New one-off invoice</h1>
<p style="color: var(--text-muted); margin-bottom: 1.5rem;">
Creates a Stripe invoice (customer pays us) and optionally emails the pay link.
</p>
<div class="card" style="max-width: 640px;">
<form method="post" action="{% url 'invoice_new' %}">
{% csrf_token %}
{{ form.as_p }}
<button type="submit" class="btn" style="margin-top: 1rem;">Create &amp; send</button>
</form>
</div>
</div>
</div>
{% endblock %}
@@ -0,0 +1,92 @@
{% extends "base.html" %}
{% load static %}
{% block title %}Invoices - AI ML Operations{% endblock %}
{% block content %}
<div class="section">
<div class="container">
<div style="margin-bottom: 2rem;">
<a href="{% url 'financial_index' %}" class="btn">Back to Dashboard</a>
</div>
<h1 class="section-title" style="text-align: left;">Invoices &amp; Subscriptions</h1>
{% if not stripe_configured %}
<p style="color: #c0392b; margin-bottom: 1.5rem;">
Stripe is not configured. Set <code>STRIPE_SECRET_KEY</code> (and webhook secret) in the environment before creating invoices.
</p>
{% endif %}
<div class="card-grid" style="margin-bottom: 2rem;">
<a href="{% url 'invoice_new' %}" class="card">
<span class="card-title">New one-off invoice</span>
<p class="card-text">Create a Stripe invoice and email the customer a pay link.</p>
</a>
<a href="{% url 'subscription_new' %}" class="card">
<span class="card-title">New monthly subscription</span>
<p class="card-text">Customer pays us on a recurring schedule via Checkout.</p>
</a>
</div>
<h2 class="section-title" style="font-size: 1.5rem; margin-bottom: 1rem;">One-off invoices</h2>
<div class="table-responsive" style="margin-bottom: 3rem;">
{% if invoices %}
<table class="table">
<thead>
<tr>
<th>Customer</th>
<th>Description</th>
<th>Amount</th>
<th>Status</th>
<th>Created</th>
</tr>
</thead>
<tbody>
{% for inv in invoices %}
<tr>
<td><a href="{% url 'invoice_detail' inv.pk %}">{{ inv.customer.name }}</a></td>
<td>{{ inv.description }}</td>
<td>${{ inv.amount_dollars|floatformat:2 }}</td>
<td>{{ inv.get_status_display }}</td>
<td>{{ inv.created|date:"Y-m-d" }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p style="color: var(--text-muted);">No invoices yet.</p>
{% endif %}
</div>
<h2 class="section-title" style="font-size: 1.5rem; margin-bottom: 1rem;">Recurring subscriptions</h2>
<div class="table-responsive">
{% if subscriptions %}
<table class="table">
<thead>
<tr>
<th>Customer</th>
<th>Description</th>
<th>Amount</th>
<th>Interval</th>
<th>Status</th>
</tr>
</thead>
<tbody>
{% for sub in subscriptions %}
<tr>
<td><a href="{% url 'subscription_detail' sub.pk %}">{{ sub.customer.name }}</a></td>
<td>{{ sub.description }}</td>
<td>${{ sub.amount_dollars|floatformat:2 }}</td>
<td>{{ sub.get_interval_display }}</td>
<td>{{ sub.get_status_display }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p style="color: var(--text-muted);">No subscriptions yet.</p>
{% endif %}
</div>
</div>
</div>
{% endblock %}
@@ -0,0 +1,34 @@
{% extends "base.html" %}
{% load static %}
{% block title %}Subscription {{ subscription.pk }} - AI ML Operations{% endblock %}
{% block content %}
<div class="section">
<div class="container">
<div style="margin-bottom: 2rem;">
<a href="{% url 'invoice_list' %}" class="btn">Back to Invoices</a>
</div>
<h1 class="section-title" style="text-align: left;">Subscription #{{ subscription.pk }}</h1>
<div class="card" style="max-width: 720px;">
<p><strong>Customer:</strong> {{ subscription.customer.name }} &lt;{{ subscription.customer.email }}&gt;</p>
<p><strong>Description:</strong> {{ subscription.description }}</p>
<p><strong>Amount:</strong> ${{ subscription.amount_dollars|floatformat:2 }} / {{ subscription.get_interval_display }}</p>
<p><strong>Status:</strong> {{ subscription.get_status_display }}</p>
<p><strong>Stripe subscription:</strong> {{ subscription.stripe_subscription_id|default:"—" }}</p>
{% if subscription.checkout_url %}
<p><strong>Checkout link:</strong> <a href="{{ subscription.checkout_url }}" target="_blank" rel="noopener">Open Checkout</a></p>
{% endif %}
<p><strong>Email sent:</strong> {{ subscription.pay_link_emailed_at|default:"Not yet" }}</p>
{% if subscription.notes %}<p><strong>Notes:</strong> {{ subscription.notes }}</p>{% endif %}
{% if subscription.checkout_url %}
<form method="post" action="{% url 'subscription_resend_email' subscription.pk %}" style="margin-top: 1.5rem;">
{% csrf_token %}
<button type="submit" class="btn">Resend checkout email</button>
</form>
{% endif %}
</div>
</div>
</div>
{% endblock %}
@@ -0,0 +1,25 @@
{% extends "base.html" %}
{% load static %}
{% block title %}New Subscription - AI ML Operations{% endblock %}
{% block content %}
<div class="section">
<div class="container">
<div style="margin-bottom: 2rem;">
<a href="{% url 'invoice_list' %}" class="btn">Back to Invoices</a>
</div>
<h1 class="section-title" style="text-align: left;">New recurring subscription</h1>
<p style="color: var(--text-muted); margin-bottom: 1.5rem;">
Customer pays us on a schedule via Stripe Checkout. Default interval is monthly.
</p>
<div class="card" style="max-width: 640px;">
<form method="post" action="{% url 'subscription_new' %}">
{% csrf_token %}
{{ form.as_p }}
<button type="submit" class="btn" style="margin-top: 1rem;">Create &amp; send</button>
</form>
</div>
</div>
</div>
{% endblock %}
+59
View File
@@ -157,3 +157,62 @@ class FinancialAccessTests(TestCase):
usernames = [e.user.username for e in employees]
self.assertIn("employee", usernames)
self.assertNotIn("extra", usernames)
class InvoiceAccessTests(TestCase):
def setUp(self):
self.admin = User.objects.create_superuser(
username="billing_admin", password="pass", email="admin@example.com"
)
self.employee_user = User.objects.create_user(username="emp2", password="pass")
set_user_type(self.employee_user, UserProfile.UserType.EMPLOYEE)
self.client_http = Client()
def test_non_admin_cannot_view_invoices(self):
self.client_http.login(username="emp2", password="pass")
response = self.client_http.get(reverse("invoice_list"))
self.assertIn(response.status_code, (302, 403))
def test_admin_can_view_invoices(self):
self.client_http.login(username="billing_admin", password="pass")
response = self.client_http.get(reverse("invoice_list"))
self.assertEqual(response.status_code, 200)
def test_webhook_rejects_bad_signature_when_secret_set(self):
from django.test import override_settings
with override_settings(STRIPE_WEBHOOK_SECRET="whsec_test"):
response = self.client_http.post(
reverse("stripe_webhook"),
data=b'{"type":"invoice.paid","data":{"object":{}}}',
content_type="application/json",
HTTP_STRIPE_SIGNATURE="t=1,v1=bad",
)
self.assertEqual(response.status_code, 400)
def test_apply_invoice_webhook_updates_status(self):
from financial.models import BillingCustomer, Invoice
from financial.stripe_billing import apply_stripe_invoice_event
customer = BillingCustomer.objects.create(
name="Acme", email="acme@example.com"
)
invoice = Invoice.objects.create(
customer=customer,
description="Work",
amount_cents=5000,
stripe_invoice_id="in_test_123",
status=Invoice.Status.OPEN,
)
apply_stripe_invoice_event(
{
"id": "in_test_123",
"status": "paid",
"metadata": {"local_invoice_id": str(invoice.pk)},
"hosted_invoice_url": "https://pay.stripe.com/test",
}
)
invoice.refresh_from_db()
self.assertEqual(invoice.status, Invoice.Status.PAID)
self.assertEqual(invoice.hosted_invoice_url, "https://pay.stripe.com/test")
+21
View File
@@ -1,6 +1,7 @@
from django.urls import path
from . import views
from . import invoice_views
urlpatterns = [
path("", views.financial_home, name="financial_home"),
@@ -21,4 +22,24 @@ urlpatterns = [
path("profile", views.profile, name="profile"),
path("manage_users", views.manage_users, name="manage_users"),
path("client_reports", views.client_reports, name="client_reports"),
path("invoices", invoice_views.invoice_list, name="invoice_list"),
path("invoices/new", invoice_views.invoice_new, name="invoice_new"),
path("invoices/<int:invoice_id>", invoice_views.invoice_detail, name="invoice_detail"),
path(
"invoices/<int:invoice_id>/resend",
invoice_views.invoice_resend_email,
name="invoice_resend_email",
),
path("subscriptions/new", invoice_views.subscription_new, name="subscription_new"),
path(
"subscriptions/<int:subscription_id>",
invoice_views.subscription_detail,
name="subscription_detail",
),
path(
"subscriptions/<int:subscription_id>/resend",
invoice_views.subscription_resend_email,
name="subscription_resend_email",
),
path("stripe/webhook/", invoice_views.stripe_webhook, name="stripe_webhook"),
]