feat: auto-fill subscription accounting dimensions from plan with item fallback (#57615)

When a plan is selected in the Subscription's Plans table, the Subscription's
accounting dimensions (cost center and any custom dimensions) auto-fill from the
plan, falling back to the plan item's company default (selling cost center for a
Customer, buying for a Supplier). Only empty fields are filled. Stale async
responses are ignored so a quick re-pick of the plan can't be overwritten.

(cherry picked from commit 7febc28ed6)

# Conflicts:
#	erpnext/accounts/doctype/subscription/subscription.js
#	erpnext/accounts/doctype/subscription/subscription.py
#	erpnext/accounts/doctype/subscription/test_subscription.py
This commit is contained in:
Jatin3128
2026-07-30 15:07:15 +05:30
committed by Mergify
parent 9a596594da
commit 3cf47766f2
3 changed files with 590 additions and 0 deletions

View File

@@ -96,3 +96,109 @@ frappe.ui.form.on("Subscription", {
});
},
});
<<<<<<< HEAD
=======
frappe.ui.form.on("Subscription Plan Detail", {
plan: function (frm, cdt, cdn) {
const row = locals[cdt][cdn];
if (!row.plan) return;
const requested_plan = row.plan;
frappe.call({
method: "erpnext.accounts.doctype.subscription.subscription.get_plan_dimensions",
args: {
plan: requested_plan,
company: frm.doc.company,
party_type: frm.doc.party_type,
},
callback: function (r) {
if (!r.message || locals[cdt]?.[cdn]?.plan !== requested_plan) return;
// Only fill dimensions left empty, so a manual entry or an earlier plan is never overwritten.
for (const [dimension, value] of Object.entries(r.message)) {
if (frm.fields_dict[dimension] && !frm.doc[dimension]) {
frm.set_value(dimension, value);
}
}
},
});
},
});
// Status -> colour and label for the calendar heatmap. Keys are Title-case to
// match the value frappe-charts shows in its hover tooltip.
const HEATMAP_COLORS = {
Paid: "#39d353",
Unpaid: "#388bfd",
Overdue: "#f0883e",
Cancelled: "#f85149",
Refunded: "#a371f7",
Planned: "#87ceeb",
};
// Days inside the window but outside the subscription's active span stay faded.
const EMPTY_COLOR = "#ebedf0";
function title_case(status) {
return status.charAt(0).toUpperCase() + status.slice(1);
}
function render_heatmap($wrapper, days, doc) {
const data_points = {};
days.forEach((day) => {
data_points[day.date] = title_case(day.status);
});
$wrapper.empty();
const chart_el = $('<div class="subscription-billing-heatmap"></div>').appendTo($wrapper)[0];
new frappe.Chart(chart_el, {
type: "heatmap",
data: {
dataPoints: data_points,
start: new Date(days[0].date),
end: new Date(days[days.length - 1].date),
},
discreteDomains: 1,
showLegend: 0,
// frappe-charts only does an intensity scale; we recolour each square by
// its own status below, so the scale colours are placeholders.
colors: ["#ebedf0", "#ebedf0", "#ebedf0", "#ebedf0", "#ebedf0"],
});
// Paint every day square with its status colour (data-value holds the status).
// The chart re-renders once for its entry animation, so repaint on each redraw.
const within_subscription = (date) =>
(!doc.start_date || date >= doc.start_date) && (!doc.end_date || date <= doc.end_date);
const paint = () =>
chart_el.querySelectorAll("[data-date]").forEach((square) => {
const status = square.getAttribute("data-value");
if (status === "Planned" && !within_subscription(square.getAttribute("data-date"))) {
// Outside the subscription's span: render blank and drop the status so the
// hover tooltip shows only the date, not "Planned".
square.setAttribute("fill", EMPTY_COLOR);
square.setAttribute("data-value", "");
return;
}
square.setAttribute("fill", HEATMAP_COLORS[status] || EMPTY_COLOR);
});
paint();
new MutationObserver(paint).observe(chart_el, { childList: true, subtree: true });
const legend = Object.keys(HEATMAP_COLORS)
.map(
(status) =>
`<span style="display:inline-flex;align-items:center;gap:4px;margin-right:12px;">
<span style="width:11px;height:11px;border-radius:2px;background:${HEATMAP_COLORS[status]};"></span>
${__(status)}
</span>`
)
.join("");
$(`<div style="margin-top:8px;font-size:11px;color:var(--text-muted);">${legend}</div>`).appendTo(
$wrapper
);
}
>>>>>>> 7febc28ed6 (feat: auto-fill subscription accounting dimensions from plan with item fallback (#57615))

View File

@@ -25,7 +25,11 @@ from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import (
get_accounting_dimensions,
)
from erpnext.accounts.doctype.subscription_plan.subscription_plan import get_plan_rate
<<<<<<< HEAD
from erpnext.accounts.party import get_party_account_currency
=======
from erpnext.stock.doctype.item.item import get_item_defaults
>>>>>>> 7febc28ed6 (feat: auto-fill subscription accounting dimensions from plan with item fallback (#57615))
class InvoiceCancelled(frappe.ValidationError):
@@ -747,6 +751,39 @@ def get_prorata_factor(
return diff / plan_days
@frappe.whitelist()
def get_plan_dimensions(
plan: str, company: str | None = None, party_type: str | None = None
) -> dict[str, str]:
"""Resolve a plan's accounting dimensions, falling back to the plan item's company defaults."""
plan_doc = frappe.get_cached_doc("Subscription Plan", plan)
dimensions = {}
for dimension in ["cost_center", *get_accounting_dimensions()]:
value = plan_doc.get(dimension) or get_item_dimension(plan_doc.item, dimension, company, party_type)
if value:
dimensions[dimension] = value
return dimensions
def get_item_dimension(
item_code: str, dimension: str, company: str | None, party_type: str | None
) -> str | None:
if not company:
return None
item_defaults = get_item_defaults(item_code, company)
if dimension != "cost_center":
return item_defaults.get(dimension)
selling = item_defaults.get("selling_cost_center")
buying = item_defaults.get("buying_cost_center")
if party_type == PARTY_SUPPLIER:
return buying or selling
return selling or buying
def process_all(subscription: list, posting_date: DateTimeLikeObject | None = None) -> None:
"""
Task to updates the status of all `Subscription` apart from those that are cancelled

View File

@@ -17,9 +17,21 @@ from frappe.utils.data import (
nowdate,
)
<<<<<<< HEAD
from erpnext.accounts.doctype.subscription.subscription import get_prorata_factor
test_dependencies = ("UOM", "Item Group", "Item")
=======
from erpnext.accounts.doctype.payment_entry.payment_entry import get_payment_entry
from erpnext.accounts.doctype.subscription.subscription import (
Subscription,
get_plan_dimensions,
get_prorata_factor,
process_all,
)
from erpnext.accounts.utils import update_subscription_on_invoice_update
from erpnext.tests.utils import ERPNextTestSuite
>>>>>>> 7febc28ed6 (feat: auto-fill subscription accounting dimensions from plan with item fallback (#57615))
class TestSubscription(FrappeTestCase):
@@ -583,6 +595,441 @@ class TestSubscription(FrappeTestCase):
subscription.process(nowdate())
self.assertEqual(len(subscription.invoices), 1)
<<<<<<< HEAD
=======
def test_subscription_auto_cancellation(self):
create_plan(
plan_name="_Test plan name 10",
cost=80,
currency="INR",
billing_interval="Day",
billing_interval_count=3,
)
start_date = getdate("2025-01-01")
subscription = create_subscription(
start_date=start_date,
end_date=add_days(start_date, 8),
cancel_at_period_end=1,
generate_new_invoices_past_due_date=1,
generate_invoice_at="Prepaid (bill at period start)",
plans=[{"plan": "_Test plan name 10", "qty": 1}],
)
# Catch-up billing on creation generates every elapsed period and cancels at end
self.assertEqual(len(subscription.invoices), 3)
self.assertEqual(subscription.status, "Cancelled")
def test_subscription_auto_cancellation_uneven_cycle(self):
create_plan(
plan_name="_Test plan name 10",
cost=80,
currency="INR",
billing_interval="Day",
billing_interval_count=3,
)
start_date = getdate("2025-01-01")
subscription = create_subscription(
start_date=start_date,
end_date=add_days(start_date, 6),
cancel_at_period_end=1,
generate_new_invoices_past_due_date=1,
generate_invoice_at="Prepaid (bill at period start)",
plans=[{"plan": "_Test plan name 10", "qty": 1}],
)
# Catch-up billing on creation incl. the partial last cycle, then cancels at end
self.assertEqual(len(subscription.invoices), 3)
self.assertEqual(subscription.status, "Cancelled")
self.assertRaises(frappe.ValidationError, subscription.process, posting_date=add_days(start_date, 7))
def test_invoice_generated_when_scheduler_runs_one_day_late(self):
# The trigger date (period end) is long past, yet catch-up still bills the period
# on creation (Bug 1: the check is `>= trigger`, not `== trigger`).
subscription = create_subscription(start_date="2018-01-01")
self.assertEqual(len(subscription.invoices), 1)
def test_deferred_revenue_applied_for_customer_subscription(self):
item_code = "_Test Non Stock Item"
frappe.db.set_value("Item", item_code, "enable_deferred_revenue", 1)
try:
# Build the period without saving, so on-create billing doesn't try to post an
# invoice (the deferred item has no account configured). This only exercises the
# item-mapping helper.
subscription = create_subscription(start_date="2018-01-01", do_not_save=True)
subscription.update_subscription_period("2018-01-01")
items = subscription.get_items_from_plans(subscription.plans)
self.assertEqual(items[0].get("enable_deferred_revenue"), 1)
self.assertEqual(getdate(items[0]["service_start_date"]), getdate("2018-01-01"))
self.assertEqual(getdate(items[0]["service_end_date"]), getdate("2018-01-31"))
finally:
frappe.db.set_value("Item", item_code, "enable_deferred_revenue", 0)
def test_validate_end_date_with_no_plans_does_not_crash(self):
sub = frappe.new_doc("Subscription")
sub.party_type = "Customer"
sub.party = "_Test Customer"
sub.company = "_Test Company"
sub.start_date = "2018-01-01"
sub.end_date = "2018-03-01"
try:
sub.validate_end_date()
except TypeError as e:
self.fail(f"validate_end_date crashed with no plans: {e}")
def test_process_all_logs_error_when_first_subscription_fails(self):
sub1 = create_subscription(start_date="2018-01-01")
sub2 = create_subscription(start_date="2018-01-02")
processed = []
original_process = Subscription.process
original_rollback = frappe.db.rollback
def patched(self, posting_date=None):
processed.append(self.name)
if self.name == sub1.name:
raise frappe.ValidationError("forced failure")
Subscription.process = patched
# process_all calls frappe.db.rollback() on error which would otherwise wipe
# the test transaction; stub it so we can observe the iteration in isolation.
frappe.db.rollback = lambda *a, **kw: None
try:
process_all([sub1.name, sub2.name])
finally:
Subscription.process = original_process
frappe.db.rollback = original_rollback
self.assertEqual(processed, [sub1.name, sub2.name])
def test_subscription_auto_completion(self):
create_plan(
plan_name="_Test Plan 3 Day",
cost=100,
billing_interval="Day",
billing_interval_count=3,
currency="INR",
)
start_date = getdate("2025-01-01")
end_date = add_days(start_date, 6)
subscription = create_subscription(
start_date=start_date,
end_date=end_date,
party_type="Customer",
party="_Test Customer",
generate_invoice_at="Prepaid (bill at period start)",
generate_new_invoices_past_due_date=1,
plans=[{"plan": "_Test Plan 3 Day", "qty": 1}],
)
for day in range(0, 10):
if subscription.status == "Cancelled":
break
subscription.process(posting_date=add_days(start_date, day))
invoices = frappe.get_all(
"Sales Invoice",
filters={"subscription": subscription.name, "docstatus": 1},
fields=["name", "from_date", "to_date"],
order_by="from_date asc",
)
for invoice in invoices:
pi = get_payment_entry("Sales Invoice", invoice.name)
pi.submit()
# Paying the invoices refreshes the subscription via the Payment Entry hook, so
# reload before processing the stale in-memory copy.
subscription.reload()
# After processing through all days, subscription should be completed
subscription.process(posting_date=add_days(end_date, 1))
self.assertEqual(subscription.status, "Completed")
def test_status_updates_immediately_when_invoice_paid(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date=nowdate())
self.assertEqual(subscription.status, "Unpaid")
invoice = subscription.get_current_invoice()
payment = get_payment_entry("Sales Invoice", invoice.name)
payment.submit()
subscription.reload()
self.assertEqual(subscription.status, "Active")
def test_invoice_update_hook_refreshes_subscription_status(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date=nowdate())
self.assertEqual(subscription.status, "Unpaid")
invoice = subscription.get_current_invoice()
invoice.db_set("outstanding_amount", 0)
invoice.db_set("status", "Paid")
update_subscription_on_invoice_update(invoice)
subscription.reload()
self.assertEqual(subscription.status, "Active")
def test_payment_entry_triggers_subscription_status_update(self):
# Test that payment entry → invoice → subscription status update chain works
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date=nowdate())
self.assertEqual(subscription.status, "Unpaid")
invoice = subscription.get_current_invoice()
self.assertIsNotNone(invoice)
self.assertGreater(invoice.outstanding_amount, 0)
# Create and submit payment entry
payment_entry = get_payment_entry(invoice.doctype, invoice.name, bank_account="_Test Bank - _TC")
payment_entry.reference_no = "12345"
payment_entry.reference_date = nowdate()
payment_entry.submit()
# Subscription status should now be Active (via on_update_after_submit hook)
subscription.reload()
self.assertEqual(subscription.status, "Active")
def test_first_invoice_generated_on_create_for_prepaid(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
)
self.assertEqual(len(subscription.invoices), 1)
def test_current_invoice_dates_reflect_latest_invoice(self):
subscription = create_subscription(
start_date="2018-01-01",
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date="2018-01-01")
invoice = subscription.get_current_invoice()
subscription.reload()
self.assertEqual(getdate(subscription.current_invoice_start), getdate(invoice.from_date))
self.assertEqual(getdate(subscription.current_invoice_end), getdate(invoice.to_date))
# `next_billing_period_start` tracks the next (unbilled) period.
self.assertEqual(
getdate(subscription.next_billing_period_start), getdate(add_days(invoice.to_date, 1))
)
def test_first_invoice_not_generated_on_create_during_trial(self):
subscription = create_subscription(
start_date=nowdate(),
trial_period_start=nowdate(),
trial_period_end=add_days(nowdate(), 30),
generate_invoice_at="Prepaid (bill at period start)",
)
self.assertEqual(len(subscription.invoices), 0)
self.assertEqual(subscription.status, "Trialing")
def test_first_invoice_not_generated_during_bulk_import(self):
frappe.flags.in_import = True
try:
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
)
self.assertEqual(len(subscription.invoices), 0)
finally:
frappe.flags.in_import = False
def test_first_invoice_not_generated_for_future_dated_subscription(self):
subscription = create_subscription(
start_date=add_days(nowdate(), 10),
generate_invoice_at="Prepaid (bill at period start)",
)
self.assertEqual(len(subscription.invoices), 0)
def test_generate_invoice_at_migration_patch(self):
from erpnext.patches.v16_0.migrate_subscription_generate_invoice_at import VALUE_MAP, execute
subscription = create_subscription(start_date=add_days(nowdate(), 10))
for old_value, new_value in VALUE_MAP.items():
frappe.db.set_value("Subscription", subscription.name, "generate_invoice_at", old_value)
execute()
self.assertEqual(
frappe.db.get_value("Subscription", subscription.name, "generate_invoice_at"), new_value
)
def test_next_billing_period_populated_for_prepaid(self):
subscription = create_subscription(
start_date=add_days(nowdate(), 10),
generate_invoice_at="Prepaid (bill at period start)",
)
self.assertEqual(getdate(subscription.next_billing_period_start), getdate(add_days(nowdate(), 10)))
self.assertGreater(
getdate(subscription.next_billing_period_end), getdate(subscription.next_billing_period_start)
)
def test_status_becomes_refunded_when_only_invoice_credited(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date=nowdate())
self.assertEqual(subscription.status, "Unpaid")
make_full_credit_note(subscription.get_current_invoice().name)
subscription.reload()
self.assertEqual(subscription.status, "Refunded")
def test_status_stays_unpaid_when_one_of_two_invoices_credited(self):
subscription = create_subscription(
start_date=add_months(nowdate(), -2),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
generate_new_invoices_past_due_date=1,
)
invoices = frappe.get_all(
"Sales Invoice",
filters={"subscription": subscription.name, "docstatus": 1, "is_return": 0},
pluck="name",
order_by="from_date asc",
)
self.assertGreaterEqual(len(invoices), 2)
make_full_credit_note(invoices[0])
subscription.reload()
self.assertNotEqual(subscription.status, "Refunded")
def test_refunded_reverts_to_active_after_full_settlement(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date=nowdate())
invoice = subscription.get_current_invoice()
make_full_credit_note(invoice.name)
subscription.reload()
self.assertEqual(subscription.status, "Refunded")
invoice.db_set("status", "Paid")
invoice.db_set("outstanding_amount", 0)
subscription.process()
self.assertEqual(subscription.status, "Active")
def test_heatmap_spans_twelve_months_from_start_month(self):
start_date = getdate("2024-03-14")
subscription = create_subscription(start_date=start_date)
heatmap = subscription.get_billing_heatmap()
self.assertEqual(getdate(heatmap[0]["date"]), get_first_day(start_date))
self.assertEqual(
getdate(heatmap[-1]["date"]), get_last_day(add_months(get_first_day(start_date), 11))
)
self.assertIn("status", heatmap[0])
def test_heatmap_marks_paid_days_green(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date=nowdate())
invoice = subscription.get_current_invoice()
invoice.db_set("status", "Paid")
invoice.db_set("outstanding_amount", 0)
subscription.reload()
cells = {cell["date"]: cell for cell in subscription.get_billing_heatmap()}
self.assertEqual(cells[str(getdate(invoice.from_date))]["status"], "paid")
def test_heatmap_marks_future_planned_days(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
)
today = getdate(nowdate())
planned = [
cell
for cell in subscription.get_billing_heatmap()
if cell["status"] == "planned" and getdate(cell["date"]) > today
]
self.assertTrue(planned)
def test_heatmap_marks_refunded_days_for_credited_periods(self):
subscription = create_subscription(
start_date=nowdate(),
generate_invoice_at="Prepaid (bill at period start)",
submit_invoice=1,
)
subscription.process(posting_date=nowdate())
invoice = subscription.get_current_invoice()
make_full_credit_note(invoice.name)
subscription.reload()
cells = {cell["date"]: cell for cell in subscription.get_billing_heatmap()}
self.assertEqual(cells[str(getdate(invoice.from_date))]["status"], "refunded")
def test_plan_dimensions_resolve_from_plan_then_item(self):
from erpnext.stock.doctype.item.test_item import make_item
# Plan-level cost center takes precedence.
create_plan(plan_name="_Test Sub Plan CC", cost=100, currency="INR")
frappe.db.set_value(
"Subscription Plan", "_Test Sub Plan CC", "cost_center", "_Test Cost Center - _TC"
)
self.assertEqual(
get_plan_dimensions("_Test Sub Plan CC", "_Test Company", "Customer").get("cost_center"),
"_Test Cost Center - _TC",
)
# No plan cost center: fall back to the item's company default (selling vs buying by party type).
item = make_item(
"_Test Sub Dimension Item",
{
"is_stock_item": 0,
"item_defaults": [
{
"company": "_Test Company",
"selling_cost_center": "_Test Cost Center - _TC",
"buying_cost_center": "_Test Cost Center 2 - _TC",
}
],
},
)
create_plan(plan_name="_Test Sub Plan No CC", cost=100, currency="INR", item=item.name)
self.assertEqual(
get_plan_dimensions("_Test Sub Plan No CC", "_Test Company", "Customer").get("cost_center"),
"_Test Cost Center - _TC",
)
self.assertEqual(
get_plan_dimensions("_Test Sub Plan No CC", "_Test Company", "Supplier").get("cost_center"),
"_Test Cost Center 2 - _TC",
)
# Without a company the item fallback is skipped.
self.assertNotIn("cost_center", get_plan_dimensions("_Test Sub Plan No CC"))
def make_full_credit_note(invoice_name):
from erpnext.accounts.doctype.sales_invoice.mapper import make_sales_return
credit_note = make_sales_return(invoice_name)
credit_note.insert()
credit_note.submit()
return credit_note
>>>>>>> 7febc28ed6 (feat: auto-fill subscription accounting dimensions from plan with item fallback (#57615))
def make_plans():
create_plan(plan_name="_Test Plan Name", cost=900, currency="INR")