feat(web-app-erpnext): set up the service provider's company and its accounting basics

The wizard bypass left ERPNext without a Company, so no app could create quotes, orders or invoices. The deploy now creates the Company from the service provider's name and country, with the currency from ERPNext's country data and an abbreviation that defaults to the name's initials, plus a fiscal year, five tax accounts, a cash account and the default company. It never rewrites an existing Company and fails loudly on a country ERPNext does not know. The README generator marks roles that read the service provider country, and the CI replay fills the placeholder.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Opus 5.5 committed 2026-10-22 20:52:30 +02:00
1 parent 286004349a
commit 558784e6ca
18 files changed
+1082 -19

No files matched your search

+30 -3
View File
@@ -71,6 +71,21 @@ def _app_name(preamble: str, role_name: str) -> str:
return entity.replace("-", " ").title()
def _role_yaml_mentions(role_dir, needle: str, *subdirs: str) -> bool:
"""Whether a YAML file below one of the role's ``subdirs`` contains ``needle``.
Args:
role_dir: the role's directory.
needle: the text to look for, typically a variable path.
subdirs: role subdirectories to search, e.g. ``"tasks"``.
"""
return any(
needle in read_text(str(path))
for subdir in subdirs
for path in sorted((role_dir / subdir).rglob("*.yml"))
)
def _needs_clearnet_resolver(role_dir) -> bool:
"""Whether the role refuses to deploy without ``networks.internet.dns``.
@@ -78,9 +93,18 @@ def _needs_clearnet_resolver(role_dir) -> bool:
role asserting it has to name it in the block an operator copies, or the
documented deploy aborts on the assert.
"""
return any(
"networks.internet.dns" in read_text(str(path))
for path in sorted((role_dir / "tasks").rglob("*.yml"))
return _role_yaml_mentions(role_dir, "networks.internet.dns", "tasks")
def _needs_service_provider_country(role_dir) -> bool:
"""Whether the role deploys with ``service_provider.company.address.country``.
The group default is a placeholder no country list knows, so a role whose
tasks or vars consume it has to name it in the block an operator copies, or
the documented deploy aborts on the unknown country.
"""
return _role_yaml_mentions(
role_dir, "service_provider.company.address.country", "tasks", "vars"
)
@@ -99,6 +123,9 @@ def _base_context(role_dir, role_name: str, app_name: str, *, invokable: bool) -
"application_author_url": author_urls().get(_role_author(role_dir)),
"application_is_host": not role_has_stack(role_dir),
"application_needs_clearnet_resolver": _needs_clearnet_resolver(role_dir),
"application_needs_service_provider_country": _needs_service_provider_country(
role_dir
),
"cosmos_mermaid": derive_cosmos_mermaid(role_dir, role_name),
"cosmos_legend": _cosmos_legend(),
}
+5
View File
@@ -37,6 +37,11 @@ design:
colors:
base: "#FFA500"
service_provider:
company:
address:
country: "Germany"
API:
openai:
api_key: "{{ lookup('env', 'INFINITO_API_OPENAI_KEY') }}"
+11 -9
View File
@@ -68,12 +68,13 @@ Solid `1:1` edges are fixed relationships; dashed `0..1` edges are conditional (
## Features
- **ERP / CRM / inventory desk** — full ERPNext frontend at `next.erp.{{ DOMAIN_PRIMARY }}`.
- **Direct OIDC SSO** — Frappe's Social Login Key talks to the shared Keycloak client; redirect URI auto-registered via the `redirect_uris` filter (no per-app Keycloak entry needed).
- **Keycloak group → Frappe role mapping** — three tiers: `roles/web-app-erpnext/administrator` → `System Manager`, `roles/web-app-erpnext/manager` → `{Sales, Purchase, Stock, Accounts} User`, default → `Customer`. Reconciled on each login by a hook reading the persisted map.
- **LDAP federation** — when `svc-db-openldap` is present, Frappe's LDAP Settings doctype is auto-configured. End-to-end LDAP login (auto-create Frappe user from LDAP bind) is deferred to a follow-up requirement; v1 ships the Settings doctype only. The LDAP-login Playwright specs are intentionally absent for v1.
- **Outbound mail (v1)** — when `web-app-stalwart` is present, Frappe's outbound Email Account is auto-configured against the central SMTP endpoint. **IMAP inbound (mail-to-Communication) is deferred to a follow-up requirement** (no precedent in the repo for auto-provisioning role-owned Stalwart mailboxes yet).
- **Wizard bypass** — first deploy seeds the API-bot user, marks `setup_complete=1` in System Settings, and skips the Frappe setup wizard entirely.
- **ERP / CRM / inventory desk:** full ERPNext frontend at `next.erp.{{ DOMAIN_PRIMARY }}`.
- **Direct OIDC SSO:** Frappe's Social Login Key talks to the shared Keycloak client; redirect URI auto-registered via the `redirect_uris` filter (no per-app Keycloak entry needed).
- **Keycloak group → Frappe role mapping:** three tiers: `roles/web-app-erpnext/administrator` → `System Manager`, `roles/web-app-erpnext/manager` → `{Sales, Purchase, Stock, Accounts} User`, default → `Customer`. Reconciled on each login by a hook reading the persisted map.
- **LDAP federation:** when `svc-db-openldap` is present, Frappe's LDAP Settings doctype is auto-configured. End-to-end LDAP login (auto-create Frappe user from LDAP bind) is deferred to a follow-up requirement; v1 ships the Settings doctype only. The LDAP-login Playwright specs are intentionally absent for v1.
- **Outbound mail (v1):** when `web-app-stalwart` is present, Frappe's outbound Email Account is auto-configured against the central SMTP endpoint. **IMAP inbound (mail-to-Communication) is deferred to a follow-up requirement** (no precedent in the repo for auto-provisioning role-owned Stalwart mailboxes yet).
- **Wizard bypass:** first deploy seeds the API-bot user, marks `setup_complete=1` in System Settings, fills its empty country and currency with the Company's, and skips the Frappe setup wizard entirely.
- **Company ready to invoice:** every deploy ensures the Company named `service_provider.company.titel` in the country `service_provider.company.address.country`, with the currency ERPNext's country data assigns to that country, plus a fiscal year for today, the tax accounts `VAT 19%`, `VAT 7%`, `VAT 20%`, `Reverse Charge` and `Export` under `Duties and Taxes` and a default cash account, so quotes, orders and invoices work right after the deploy. The abbreviation ERPNext appends to every account is `applications.web-app-erpnext.services.erpnext.company.abbr`, the initials of the company name unless the inventory sets it. A Company that already exists keeps its own values, and a country ERPNext does not know, such as the placeholder default, fails the deploy.
## Quick Setup
@@ -96,19 +97,20 @@ Run the published image to provision the inventory and deploy ERPNext to a manag
APP=web-app-erpnext
HOST=<your-server>
DOMAIN=<your-domain>
COUNTRY=<your-country>
TLS_MODE=self_signed
SSH_PUBLIC_KEY="<your-ssh-public-key>"
docker run --rm -it \
-v "$PWD/inventories:/etc/infinito.nexus/inventories" \
-e APP="$APP" -e HOST="$HOST" -e DOMAIN="$DOMAIN" -e TLS_MODE="$TLS_MODE" -e SSH_PUBLIC_KEY="$SSH_PUBLIC_KEY" \
-e APP="$APP" -e HOST="$HOST" -e DOMAIN="$DOMAIN" -e COUNTRY="$COUNTRY" -e TLS_MODE="$TLS_MODE" -e SSH_PUBLIC_KEY="$SSH_PUBLIC_KEY" \
ghcr.io/infinito-nexus/core/debian bash -c '
INVENTORY=/etc/infinito.nexus/inventories/production
infinito administration inventory provision "$INVENTORY" \
--inventory-file "$INVENTORY/devices.yml" \
--host "$HOST" \
--include "$APP" \
--vars "{\"TLS_MODE\": \"$TLS_MODE\", \"DOMAIN_PRIMARY\": \"$DOMAIN\", \"users\": {\"administrator\": {\"authorized_keys\": [\"$SSH_PUBLIC_KEY\"]}}}" &&
--vars "{\"TLS_MODE\": \"$TLS_MODE\", \"DOMAIN_PRIMARY\": \"$DOMAIN\", \"service_provider\": {\"company\": {\"address\": {\"country\": \"$COUNTRY\"}}}, \"users\": {\"administrator\": {\"authorized_keys\": [\"$SSH_PUBLIC_KEY\"]}}}" &&
infinito administration deploy dedicated "$INVENTORY/devices.yml" \
--password-file "$INVENTORY/.password" \
--diff -vv'
@@ -163,7 +165,7 @@ make compose-exec service=erpnext-backend \
- Variant matrix lives in [variants.yml](./meta/variants.yml). Service flags and image pins in [services.yml](./meta/services.yml). Credentials declared in [secrets.yml](./meta/secrets.yml).
- Site name MUST match the HTTP Host header. The Frappe site is created with `bench new-site next.erp.<DOMAIN_PRIMARY>` and `FRAPPE_SITE_NAME_HEADER` on the frontend nginx maps Host → site.
- All post-bootstrap configuration (OIDC Social Login Key, LDAP Settings, outbound Email Account, group-role mapping) runs as Python scripts piped to `python` inside the backend container (see `files/scripts/`).
- All post-bootstrap configuration (Company, API bot, OIDC Social Login Key, LDAP Settings, outbound Email Account, group-role mapping) runs as Python scripts piped to `python` inside the backend container (see `files/python/apply/`).
## Further Resources
@@ -1,5 +1,3 @@
# nocheck: mirrored-unit-test - imports frappe at module level and runs inside `bench
# execute`; the framework is not installable outside the ERPNext image
import os
import frappe
@@ -11,6 +9,8 @@ API_BOT_FIRSTNAME = os.environ["API_BOT_FIRSTNAME"]
API_BOT_LASTNAME = os.environ["API_BOT_LASTNAME"]
API_BOT_PASSWORD = os.environ["API_BOT_PASSWORD"]
API_KEY_SECRET = os.environ["API_KEY_SECRET"]
COUNTRY = os.environ["COUNTRY"]
CURRENCY = os.environ["CURRENCY"]
API_KEY, API_SECRET = (
API_KEY_SECRET[: len(API_KEY_SECRET) // 2],
@@ -51,9 +51,9 @@ def bypass_setup_wizard():
if not ss.time_zone:
ss.time_zone = "UTC"
if not ss.country:
ss.country = "United States"
ss.country = COUNTRY
if not ss.currency:
ss.currency = "USD"
ss.currency = CURRENCY
ss.flags.ignore_permissions = True
ss.flags.ignore_mandatory = True
ss.save()
@@ -0,0 +1,231 @@
"""Ensure the service provider's Company and the records ERPNext needs to quote,
order and invoice for it.
The site skips the setup wizard, so nothing else installs the warehouse type,
territory, unit and opportunity type the wizard would create. An existing
Company keeps its abbreviation, country and currency and only gains the records
it lacks; the global default company is set only while none is.
Runs on the stdin of the bench Python inside the ERPNext backend container and
prints one JSON line naming the Company and every record it created.
Environment:
SITE_NAME: Frappe site to configure.
COMPANY_NAME: name of the Company.
COMPANY_ABBR: abbreviation ERPNext appends to the Company's account names.
COMPANY_COUNTRY: the Company's country as ERPNext's country data names it;
the Company's currency is the one that data assigns to it.
"""
import json
import os
import frappe
from frappe.geo.country_info import get_country_info
from frappe.utils import getdate
SITES_PATH = "/home/frappe/frappe-bench/sites"
TAX_GROUP = "Duties and Taxes"
TAX_ACCOUNTS = ("VAT 19%", "VAT 7%", "VAT 20%", "Reverse Charge", "Export")
FIXTURES = (
("Warehouse Type", "Transit", {"__newname": "Transit"}),
(
"Territory",
"All Territories",
{"territory_name": "All Territories", "is_group": 1},
),
("UOM", "Unit", {"uom_name": "Unit"}),
("Opportunity Type", "Sales", {"__newname": "Sales"}),
)
INVENTORY_KEYS = {
"COMPANY_NAME": "service_provider.company.titel",
"COMPANY_ABBR": "applications.web-app-erpnext.services.erpnext.company.abbr",
"COMPANY_COUNTRY": "service_provider.company.address.country",
}
def required(name):
"""Return a non-empty operator input or stop the run.
Args:
name: environment variable carrying the input.
Returns:
The value without surrounding whitespace.
"""
value = os.environ.get(name, "").strip()
if not value:
raise SystemExit(
f"{name} is empty: set {INVENTORY_KEYS[name]} in the inventory"
)
return value
def currency_of(country):
"""Return the currency ERPNext's country data assigns to a country.
Args:
country: country name as ERPNext's country data spells it.
Returns:
The ISO currency code.
"""
currency = get_country_info(country).get("currency")
if not currency:
raise SystemExit(
f"ERPNext knows no country {country!r}: set "
f"{INVENTORY_KEYS['COMPANY_COUNTRY']} in the inventory to a country "
"name from ERPNext's country list, such as 'Germany'"
)
return currency
def insert(doctype, fields):
frappe.get_doc({"doctype": doctype, **fields}).insert(ignore_permissions=True)
def ensure_fixtures():
created = []
for doctype, name, fields in FIXTURES:
if not frappe.db.exists(doctype, name):
insert(doctype, fields)
created.append(f"{doctype}: {name}")
return created
def ensure_company(name, abbr, country, currency):
if frappe.db.exists("Company", name):
return []
insert(
"Company",
{
"company_name": name,
"abbr": abbr,
"country": country,
"default_currency": currency,
},
)
return [f"Company: {name}"]
def ensure_fiscal_year(company):
today = getdate()
covering = frappe.get_all(
"Fiscal Year",
filters={
"disabled": 0,
"year_start_date": ["<=", today],
"year_end_date": [">=", today],
},
pluck="name",
)
for name in covering:
companies = [
row.company for row in frappe.get_doc("Fiscal Year", name).companies
]
if not companies or company in companies:
return []
year = str(today.year)
insert(
"Fiscal Year",
{
"year": year,
"year_start_date": f"{year}-01-01",
"year_end_date": f"{year}-12-31",
},
)
return [f"Fiscal Year: {year}"]
def ensure_tax_accounts(company, abbr):
group = f"{TAX_GROUP} - {abbr}"
if not frappe.db.exists("Account", group):
raise SystemExit(
f"Company {company!r} has no account group {group!r} for its taxes"
)
created = []
for account_name in TAX_ACCOUNTS:
account = f"{account_name} - {abbr}"
if frappe.db.exists("Account", account):
continue
insert(
"Account",
{
"account_name": account_name,
"parent_account": group,
"company": company,
"account_type": "Tax",
"is_group": 0,
},
)
created.append(f"Account: {account}")
return created
def ensure_default_cash_account(company):
if frappe.db.get_value("Company", company, "default_cash_account"):
return []
for account_type in ("Cash", "Bank"):
account = frappe.db.get_value(
"Account",
{"company": company, "account_type": account_type, "is_group": 0},
"name",
order_by="lft",
)
if account:
frappe.db.set_value("Company", company, "default_cash_account", account)
return [f"Default Cash Account: {account}"]
raise SystemExit(f"Company {company!r} has neither a Cash nor a Bank account")
def ensure_default_company(company):
defaults = frappe.get_single("Global Defaults")
if defaults.default_company:
return []
country, currency = frappe.db.get_value(
"Company", company, ["country", "default_currency"]
)
defaults.default_company = company
defaults.country = country
defaults.default_currency = currency
defaults.save(ignore_permissions=True)
return [f"Default Company: {company}"]
def main():
site = os.environ["SITE_NAME"]
name = required("COMPANY_NAME")
abbr = required("COMPANY_ABBR")
country = required("COMPANY_COUNTRY")
currency = currency_of(country)
frappe.init(site=site, sites_path=SITES_PATH)
frappe.connect()
try:
created = ensure_fixtures()
created += ensure_company(name, abbr, country, currency)
company_abbr = frappe.db.get_value("Company", name, "abbr")
created += ensure_fiscal_year(name)
created += ensure_tax_accounts(name, company_abbr)
created += ensure_default_cash_account(name)
created += ensure_default_company(name)
frappe.db.commit()
frappe.clear_cache()
finally:
frappe.destroy()
print(
json.dumps(
{
"company": name,
"abbr": company_abbr,
"country": country,
"currency": currency,
"created": created,
}
)
)
if __name__ == "__main__":
main()
+2
View File
@@ -77,6 +77,8 @@ erpnext:
websocket: 9000
proxy:
service_key: frontend
company:
abbr: "{{ service_provider.company.titel.split() | map('first') | join | upper }}"
run_after:
- web-app-keycloak
- web-app-stalwart
@@ -38,12 +38,30 @@
delegate_to: "{{ ERPNEXT_BACKEND_HOST_NODE }}"
changed_when: false
- name: "🏢 ensure the service provider's Company and its accounting basics"
ansible.builtin.shell:
cmd: > # nocheck: container-exec-resolver runtime-resolved CID, delegated to the hosting node
container exec -i
-e SITE_NAME -e COMPANY_NAME -e COMPANY_ABBR -e COMPANY_COUNTRY
{{ ERPNEXT_BACKEND_LOCAL_CID | quote }}
{{ ERPNEXT_BENCH_PYTHON }}
stdin: "{{ lookup('ansible.builtin.file', role_path ~ '/files/python/apply/company.py') }}"
delegate_to: "{{ ERPNEXT_BACKEND_HOST_NODE }}"
environment:
SITE_NAME: "{{ ERPNEXT_SITE_NAME }}"
COMPANY_NAME: "{{ ERPNEXT_COMPANY_NAME }}"
COMPANY_ABBR: "{{ ERPNEXT_COMPANY_ABBR }}"
COMPANY_COUNTRY: "{{ ERPNEXT_COMPANY_COUNTRY }}"
register: erpnext_company_apply
changed_when: erpnext_company_apply.rc == 0 and (erpnext_company_apply.stdout_lines | last | from_json).created | length > 0
- name: "📌 apply API-bot user + bypass setup wizard"
no_log: "{{ MASK_CREDENTIALS_IN_LOGS | bool }}"
ansible.builtin.shell:
cmd: > # nocheck: container-exec-resolver runtime-resolved CID, delegated to the hosting node
container exec -i
-e SITE_NAME -e API_BOT_LOGIN -e API_BOT_EMAIL -e API_BOT_FIRSTNAME -e API_BOT_LASTNAME -e API_BOT_PASSWORD -e API_KEY_SECRET
-e COUNTRY -e CURRENCY
{{ ERPNEXT_BACKEND_LOCAL_CID | quote }}
{{ ERPNEXT_BENCH_PYTHON }}
stdin: "{{ lookup('ansible.builtin.file', role_path ~ '/files/python/apply/api_bot.py') }}"
@@ -56,5 +74,7 @@
API_BOT_LASTNAME: "{{ ERPNEXT_API_BOT_USER.lastname }}"
API_BOT_PASSWORD: "{{ ERPNEXT_API_BOT_PASSWORD }}"
API_KEY_SECRET: "{{ ERPNEXT_API_KEY_SECRET }}"
COUNTRY: "{{ ERPNEXT_COMPANY_COUNTRY }}"
CURRENCY: "{{ (erpnext_company_apply.stdout_lines | last | from_json).currency }}"
register: erpnext_api_bot_apply
changed_when: false
+4
View File
@@ -25,6 +25,10 @@ ERPNEXT_API_KEY_SECRET: "{{ lookup('config', application_id, 'secrets.credential
ERPNEXT_API_BOT_USER: "{{ lookup('users', 'erpnext-api-bot') }}"
ERPNEXT_COMPANY_NAME: "{{ service_provider.company.titel }}"
ERPNEXT_COMPANY_ABBR: "{{ lookup('config', application_id, 'services.erpnext.company.abbr') }}"
ERPNEXT_COMPANY_COUNTRY: "{{ service_provider.company.address.country }}"
ERPNEXT_OIDC_ENABLED: "{{ lookup('config', application_id, 'services.sso.enabled') | bool }}"
ERPNEXT_LDAP_ENABLED: "{{ lookup('config', application_id, 'services.ldap.enabled') | bool }}"
ERPNEXT_EMAIL_ENABLED: "{{ lookup('config', application_id, 'services.email.enabled') | bool }}"
+1
View File
@@ -17,4 +17,5 @@ source scripts/meta/env/load.sh
sed -i "s#<your-ssh-public-key>#ssh-ed25519 AAAA_TEST_DUMMY_KEY github-ci-dummy@infinito#" /tmp/run.sh
sed -i "s#<your-domain>#${INFINITO_DOMAIN}#" /tmp/run.sh
sed -i "s#<your-clearnet-resolver>#${INFINITO_DNS_IP}#" /tmp/run.sh
sed -i "s#<your-country>#Germany#" /tmp/run.sh
docker exec -i "${INFINITO_CONTAINER}" bash -s </tmp/run.sh
+1
View File
@@ -18,6 +18,7 @@ test -s /tmp/host-deploy.sh
source scripts/meta/env/load.sh
sed -i "s#<your-ssh-public-key>#ssh-ed25519 AAAA_TEST_DUMMY_KEY github-ci-dummy@infinito#" /tmp/host-deploy.sh
sed -i "s#<your-domain>#${INFINITO_DOMAIN}#" /tmp/host-deploy.sh
sed -i "s#<your-country>#Germany#" /tmp/host-deploy.sh
docker pull "${GUIDE_RUNTIME_IMAGE}"
+9 -3
View File
@@ -87,6 +87,9 @@ DOMAIN=<your-domain>
{% if application_needs_clearnet_resolver -%}
DNS=<your-clearnet-resolver>
{% endif -%}
{% if application_needs_service_provider_country -%}
COUNTRY=<your-country>
{% endif -%}
TLS_MODE=self_signed
SSH_PUBLIC_KEY="<your-ssh-public-key>"
INVENTORY=inventories/production
@@ -94,7 +97,7 @@ infinito administration inventory provision "$INVENTORY" \
--inventory-file "$INVENTORY/devices.yml" \
--host localhost \
--include "$APP" \
--vars "{\"TLS_MODE\": \"$TLS_MODE\", \"DOMAIN_PRIMARY\": \"$DOMAIN\", {% if application_needs_clearnet_resolver %}\"networks\": {\"internet\": {\"dns\": \"$DNS\"}}, {% endif %}\"users\": {\"administrator\": {\"authorized_keys\": [\"$SSH_PUBLIC_KEY\"]}}}"
--vars "{\"TLS_MODE\": \"$TLS_MODE\", \"DOMAIN_PRIMARY\": \"$DOMAIN\", {% if application_needs_clearnet_resolver %}\"networks\": {\"internet\": {\"dns\": \"$DNS\"}}, {% endif %}{% if application_needs_service_provider_country %}\"service_provider\": {\"company\": {\"address\": {\"country\": \"$COUNTRY\"}}}, {% endif %}\"users\": {\"administrator\": {\"authorized_keys\": [\"$SSH_PUBLIC_KEY\"]}}}"
infinito administration deploy dedicated "$INVENTORY/devices.yml" \
--password-file "$INVENTORY/.password" \
--diff -vv
@@ -109,19 +112,22 @@ DOMAIN=<your-domain>
{% if application_needs_clearnet_resolver -%}
DNS=<your-clearnet-resolver>
{% endif -%}
{% if application_needs_service_provider_country -%}
COUNTRY=<your-country>
{% endif -%}
TLS_MODE=self_signed
SSH_PUBLIC_KEY="<your-ssh-public-key>"
docker run --rm -it \
-v "$PWD/inventories:/etc/infinito.nexus/inventories" \
-e APP="$APP" -e HOST="$HOST" -e DOMAIN="$DOMAIN" {% if application_needs_clearnet_resolver %}-e DNS="$DNS" {% endif %}-e TLS_MODE="$TLS_MODE" -e SSH_PUBLIC_KEY="$SSH_PUBLIC_KEY" \
-e APP="$APP" -e HOST="$HOST" -e DOMAIN="$DOMAIN" {% if application_needs_clearnet_resolver %}-e DNS="$DNS" {% endif %}{% if application_needs_service_provider_country %}-e COUNTRY="$COUNTRY" {% endif %}-e TLS_MODE="$TLS_MODE" -e SSH_PUBLIC_KEY="$SSH_PUBLIC_KEY" \
ghcr.io/infinito-nexus/core/debian bash -c '
INVENTORY=/etc/infinito.nexus/inventories/production
infinito administration inventory provision "$INVENTORY" \
--inventory-file "$INVENTORY/devices.yml" \
--host "$HOST" \
--include "$APP" \
--vars "{\"TLS_MODE\": \"$TLS_MODE\", \"DOMAIN_PRIMARY\": \"$DOMAIN\", {% if application_needs_clearnet_resolver %}\"networks\": {\"internet\": {\"dns\": \"$DNS\"}}, {% endif %}\"users\": {\"administrator\": {\"authorized_keys\": [\"$SSH_PUBLIC_KEY\"]}}}" &&
--vars "{\"TLS_MODE\": \"$TLS_MODE\", \"DOMAIN_PRIMARY\": \"$DOMAIN\", {% if application_needs_clearnet_resolver %}\"networks\": {\"internet\": {\"dns\": \"$DNS\"}}, {% endif %}{% if application_needs_service_provider_country %}\"service_provider\": {\"company\": {\"address\": {\"country\": \"$COUNTRY\"}}}, {% endif %}\"users\": {\"administrator\": {\"authorized_keys\": [\"$SSH_PUBLIC_KEY\"]}}}" &&
infinito administration deploy dedicated "$INVENTORY/devices.yml" \
--password-file "$INVENTORY/.password" \
--diff -vv'
@@ -0,0 +1,73 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from cli.build.docs.readme.generate import generate_readme
from utils.roles.mapping import (
ROLE_FILE_TASKS_MAIN,
ROLE_FILE_TEMPL_COMPOSE,
ROLE_FILE_VARS_MAIN,
)
ROLE = "web-app-sample"
READS_COUNTRY = 'SAMPLE_COUNTRY: "{{ service_provider.company.address.country }}"\n'
COUNTRY_VARS = (
'\\"service_provider\\": {\\"company\\": {\\"address\\": '
'{\\"country\\": \\"$COUNTRY\\"}}}, '
)
def _production_block(files: dict[str, str]) -> str:
with tempfile.TemporaryDirectory() as tmp:
role_dir = Path(tmp) / ROLE
for rel, content in files.items():
path = role_dir / rel
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
text, _ = generate_readme(role_dir, ROLE)
return text.split("### Production", 1)[1].split("## Credits", 1)[0]
class TestServiceProviderCountry(unittest.TestCase):
def test_stack_role_reading_the_country_in_vars_asks_for_it(self) -> None:
block = _production_block(
{ROLE_FILE_TEMPL_COMPOSE: "", ROLE_FILE_VARS_MAIN: READS_COUNTRY}
)
self.assertIn("COUNTRY=<your-country>", block)
self.assertIn('-e COUNTRY="$COUNTRY"', block)
self.assertIn(COUNTRY_VARS, block)
def test_role_reading_the_country_in_tasks_asks_for_it(self) -> None:
block = _production_block(
{ROLE_FILE_TEMPL_COMPOSE: "", ROLE_FILE_TASKS_MAIN: READS_COUNTRY}
)
self.assertIn("COUNTRY=<your-country>", block)
self.assertIn(COUNTRY_VARS, block)
def test_host_role_reading_the_country_asks_for_it(self) -> None:
block = _production_block({ROLE_FILE_VARS_MAIN: READS_COUNTRY})
self.assertIn("--host localhost", block)
self.assertIn("COUNTRY=<your-country>", block)
self.assertIn(COUNTRY_VARS, block)
def test_role_rendering_the_country_only_in_a_template_does_not_ask(
self,
) -> None:
block = _production_block(
{
ROLE_FILE_TEMPL_COMPOSE: "",
"templates/imprint.html.j2": READS_COUNTRY,
}
)
self.assertNotIn("COUNTRY", block)
self.assertNotIn("service_provider", block)
if __name__ == "__main__":
unittest.main()
Whitespace-only changes.
Whitespace-only changes.
@@ -0,0 +1,5 @@
from pathlib import Path
PROJECT_ROOT: Path = Path(__file__).resolve().parents[7]
__all__ = ["PROJECT_ROOT"]
@@ -0,0 +1,303 @@
"""In-memory stand-in for the slice of Frappe the ERPNext apply scripts call.
Records live in dicts keyed by doctype and name. Inserting a Company does what
ERPNext's controller does on a real site: it refuses a missing ``Transit``
warehouse type or a taken abbreviation, builds a minimal chart of accounts and
picks the default cash account from it.
"""
from __future__ import annotations
import copy
import datetime as dt
import io
import os
import runpy
import sys
from contextlib import redirect_stdout
from types import ModuleType, SimpleNamespace
from unittest.mock import patch
TODAY = dt.date(2026, 10, 22)
COUNTRY_CURRENCIES = {"Germany": "EUR", "United States": "USD"}
CHILD_TABLES = {"Fiscal Year": ("companies",), "User": ("roles",)}
NAME_FIELDS = {
"Company": "company_name",
"Fiscal Year": "year",
"Territory": "territory_name",
"UOM": "uom_name",
"User": "email",
}
class LinkValidationError(Exception):
pass
class DuplicateEntryError(Exception):
pass
def get_country_info(country):
"""Mirror ``frappe.geo.country_info.get_country_info``.
Args:
country: country name.
Returns:
The country's data; an unknown country yields only format defaults.
"""
info = {"date_format": "dd-mm-yyyy", "time_format": "HH:mm:ss"}
if country in COUNTRY_CURRENCIES:
info["currency"] = COUNTRY_CURRENCIES[country]
return info
def _matches(record, filters):
for field, expected in filters.items():
value = record.get(field)
if isinstance(expected, list):
operator, operand = expected
if value is None:
return False
if operator == "<=" and not value <= operand:
return False
if operator == ">=" and not value >= operand:
return False
elif value != expected:
return False
return True
class FakeDoc:
def __init__(self, frappe, doctype, values, *, single=False):
object.__setattr__(self, "_frappe", frappe)
object.__setattr__(self, "_single", single)
object.__setattr__(self, "doctype", doctype)
object.__setattr__(self, "flags", SimpleNamespace())
tables = {table: [] for table in CHILD_TABLES.get(doctype, ())}
object.__setattr__(self, "_values", {**tables, **values})
def __getattr__(self, name):
if name.startswith("__"):
raise AttributeError(name)
return self._values.get(name)
def __setattr__(self, name, value):
self._values[name] = value
def get(self, name):
return self._values.get(name)
def append(self, table, row):
self._values[table].append(SimpleNamespace(**row))
def insert(self, ignore_permissions=False):
self._values["name"] = self._frappe.insert(self.doctype, self._values)
return self
def save(self, ignore_permissions=False):
if self._single:
self._frappe.singles[self.doctype] = copy.deepcopy(self._values)
elif self._values.get("name"):
self._frappe.records[self.doctype][self._values["name"]] = copy.deepcopy(
self._values
)
else:
self.insert(ignore_permissions)
return self
class FakeDb:
def __init__(self, frappe):
self._frappe = frappe
self.committed = False
def _table(self, doctype):
return self._frappe.records.get(doctype, {})
def exists(self, doctype, name):
if doctype == "DocType":
return name in self._frappe.doctypes
return name in self._table(doctype)
def get_value(self, doctype, filters, fieldname, order_by=None):
if isinstance(filters, str):
record = self._table(doctype).get(filters)
else:
candidates = [
r for r in self._table(doctype).values() if _matches(r, filters)
]
if order_by:
candidates.sort(key=lambda r: r[order_by])
record = candidates[0] if candidates else None
if record is None:
return None
if isinstance(fieldname, list):
return tuple(record.get(field) for field in fieldname)
return record.get(fieldname)
def set_value(self, doctype, name, fieldname, value):
self._table(doctype)[name][fieldname] = value
def commit(self):
self.committed = True
def set_default(self, key, value):
self._frappe.defaults[key] = value
def get_default(self, key):
return self._frappe.defaults.get(key)
def table_exists(self, name):
return name in self._frappe.tables
class FakeFrappe:
def __init__(self):
self.records = {}
self.singles = {}
self.defaults = {}
self.doctypes = {"System Settings", "Global Defaults"}
self.tables = set()
self.calls = []
self.db = FakeDb(self)
def seed(self, doctype, name, **values):
self.records.setdefault(doctype, {})[name] = {**values, "name": name}
def init(self, site, sites_path):
self.calls.append(("init", site, sites_path))
def connect(self):
self.calls.append(("connect",))
def clear_cache(self):
self.calls.append(("clear_cache",))
def destroy(self):
self.calls.append(("destroy",))
def get_doc(self, doctype, name=None):
if isinstance(doctype, dict):
values = dict(doctype)
return FakeDoc(self, values.pop("doctype"), values)
if name is None:
return self.get_single(doctype)
return FakeDoc(self, doctype, copy.deepcopy(self.records[doctype][name]))
def get_single(self, doctype):
values = copy.deepcopy(self.singles.get(doctype, {}))
return FakeDoc(self, doctype, values, single=True)
def new_doc(self, doctype):
return FakeDoc(self, doctype, {})
def get_all(self, doctype, filters=None, pluck=None):
rows = [
r for r in self.db._table(doctype).values() if _matches(r, filters or {})
]
return [row[pluck] for row in rows] if pluck else rows
def insert(self, doctype, values):
"""Store a new record the way the doctype's naming rule names it.
Args:
doctype: the record's doctype.
values: the record's fields, ``__newname`` for prompt-named doctypes.
Returns:
The record's name.
"""
stored = copy.deepcopy(values)
name = stored.pop("__newname", None) or self._name_of(doctype, stored)
if name in self.db._table(doctype):
raise DuplicateEntryError(f"{doctype} {name} exists")
if doctype == "Company":
self._validate_company(stored)
if doctype == "Account" and not self.db.exists(
"Account", stored["parent_account"]
):
raise LinkValidationError(
f"Could not find Parent Account: {stored['parent_account']}"
)
if doctype == "Fiscal Year":
stored.setdefault("disabled", 0)
for field in ("year_start_date", "year_end_date"):
stored[field] = dt.date.fromisoformat(stored[field])
self.records.setdefault(doctype, {})[name] = {**stored, "name": name}
if doctype == "Company":
self._create_chart(name, stored["abbr"])
return name
def _name_of(self, doctype, values):
if doctype == "Account":
abbr = self.records["Company"][values["company"]]["abbr"]
return f"{values['account_name']} - {abbr}"
return values[NAME_FIELDS[doctype]]
def _validate_company(self, values):
if not self.db.exists("Warehouse Type", "Transit"):
raise LinkValidationError("Could not find Warehouse Type: Transit")
taken = {company["abbr"] for company in self.db._table("Company").values()}
if values["abbr"] in taken:
raise LinkValidationError("Abbreviation already used for another company")
def _create_chart(self, company, abbr):
self.seed(
"Account",
f"Duties and Taxes - {abbr}",
company=company,
account_type="Tax",
is_group=1,
lft=30,
)
self.seed(
"Account",
f"Cash - {abbr}",
company=company,
account_type="Cash",
is_group=0,
lft=20,
)
self.records["Company"][company]["default_cash_account"] = f"Cash - {abbr}"
def modules(self):
"""Return the ``sys.modules`` entries that make this object ``frappe``.
Returns:
Module name to module, for ``patch.dict(sys.modules, ...)``.
"""
country_info = ModuleType("frappe.geo.country_info")
country_info.get_country_info = get_country_info
geo = ModuleType("frappe.geo")
geo.country_info = country_info
utils = ModuleType("frappe.utils")
utils.getdate = lambda: TODAY
return {
"frappe": self,
"frappe.geo": geo,
"frappe.geo.country_info": country_info,
"frappe.utils": utils,
}
def run_script(path, frappe, env):
"""Run an apply script against a fake site.
Args:
path: the script to run as ``__main__``.
frappe: the fake site it talks to.
env: the complete environment the script sees.
Returns:
What the script printed.
"""
out = io.StringIO()
with (
patch.dict(os.environ, env, clear=True),
patch.dict(sys.modules, frappe.modules()),
redirect_stdout(out),
):
runpy.run_path(str(path), run_name="__main__")
return out.getvalue()
@@ -0,0 +1,106 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from . import PROJECT_ROOT
from ._fake_frappe import FakeFrappe, run_script
SCRIPT = PROJECT_ROOT / "roles/web-app-erpnext/files/python/apply/api_bot.py"
BOT_EMAIL = "erpnext-api-bot@infinito.example"
def _env(**overrides):
return {
"SITE_NAME": "next.erp.infinito.example",
"API_BOT_LOGIN": "erpnext-api-bot",
"API_BOT_EMAIL": BOT_EMAIL,
"API_BOT_FIRSTNAME": "ERPNext",
"API_BOT_LASTNAME": "API Bot",
"API_BOT_PASSWORD": "bot-password",
"API_KEY_SECRET": "keykeysecretsecret",
"COUNTRY": "Germany",
"CURRENCY": "EUR",
**overrides,
}
class TestApiBot(unittest.TestCase):
def setUp(self) -> None:
self.frappe = FakeFrappe()
def _run(self, **overrides) -> None:
run_script(SCRIPT, self.frappe, _env(**overrides))
def test_fills_an_empty_system_country_and_currency(self) -> None:
self._run()
settings = self.frappe.singles["System Settings"]
self.assertEqual(settings["country"], "Germany")
self.assertEqual(settings["currency"], "EUR")
def test_keeps_a_system_country_and_currency_already_set(self) -> None:
self.frappe.singles["System Settings"] = {
"country": "United States",
"currency": "USD",
}
self._run()
settings = self.frappe.singles["System Settings"]
self.assertEqual(settings["country"], "United States")
self.assertEqual(settings["currency"], "USD")
def test_marks_the_setup_wizard_complete(self) -> None:
self.frappe.tables.add("Installed Application")
self.frappe.seed("Installed Application", "erpnext", is_setup_complete=0)
self.frappe.defaults["desktop:home_page"] = "setup-wizard"
self._run()
self.assertEqual(self.frappe.defaults["setup_complete"], "1")
self.assertEqual(self.frappe.singles["System Settings"]["setup_complete"], 1)
self.assertEqual(
self.frappe.records["Installed Application"]["erpnext"][
"is_setup_complete"
],
1,
)
self.assertEqual(self.frappe.defaults["desktop:home_page"], "")
def test_creates_the_bot_as_system_manager_with_split_api_credentials(
self,
) -> None:
self._run()
bot = self.frappe.records["User"][BOT_EMAIL]
self.assertEqual(bot["username"], "erpnext-api-bot")
self.assertEqual(bot["api_key"], "keykeysec")
self.assertEqual(bot["api_secret"], "retsecret")
self.assertEqual([row.role for row in bot["roles"]], ["System Manager"])
def test_keeps_a_single_system_manager_role_on_an_existing_bot(self) -> None:
self.frappe.seed(
"User",
BOT_EMAIL,
email=BOT_EMAIL,
roles=[SimpleNamespace(role="System Manager")],
)
self._run()
roles = self.frappe.records["User"][BOT_EMAIL]["roles"]
self.assertEqual([row.role for row in roles], ["System Manager"])
def test_missing_currency_fails_before_touching_the_site(self) -> None:
env = _env()
del env["CURRENCY"]
with self.assertRaises(KeyError):
run_script(SCRIPT, self.frappe, env)
self.assertEqual(self.frappe.calls, [])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,277 @@
from __future__ import annotations
import datetime as dt
import json
import unittest
from types import SimpleNamespace
from . import PROJECT_ROOT
from ._fake_frappe import FakeFrappe, LinkValidationError, run_script
SCRIPT = PROJECT_ROOT / "roles/web-app-erpnext/files/python/apply/company.py"
COMPANY = "Infinito.Nexus"
TAX_ACCOUNTS = ("VAT 19%", "VAT 7%", "VAT 20%", "Reverse Charge", "Export")
def _env(**overrides):
return {
"SITE_NAME": "next.erp.infinito.example",
"COMPANY_NAME": COMPANY,
"COMPANY_ABBR": "IN",
"COMPANY_COUNTRY": "Germany",
**overrides,
}
def _seed_company(frappe, name, abbr, **values):
frappe.seed(
"Company",
name,
company_name=name,
abbr=abbr,
country=values.pop("country", "United States"),
default_currency=values.pop("default_currency", "USD"),
**values,
)
frappe.seed(
"Account",
f"Duties and Taxes - {abbr}",
company=name,
account_type="Tax",
is_group=1,
lft=30,
)
class TestCompany(unittest.TestCase):
def setUp(self) -> None:
self.frappe = FakeFrappe()
def _run(self, **overrides):
printed = run_script(SCRIPT, self.frappe, _env(**overrides))
return json.loads(printed.strip().splitlines()[-1])
def _company(self, name=COMPANY):
return self.frappe.records["Company"][name]
def test_creates_company_with_the_currency_its_country_uses(self) -> None:
self._run()
company = self._company()
self.assertEqual(company["abbr"], "IN")
self.assertEqual(company["country"], "Germany")
self.assertEqual(company["default_currency"], "EUR")
def test_reports_the_company_and_its_currency(self) -> None:
result = self._run()
self.assertEqual(result["company"], COMPANY)
self.assertEqual(result["abbr"], "IN")
self.assertEqual(result["country"], "Germany")
self.assertEqual(result["currency"], "EUR")
self.assertIn(f"Company: {COMPANY}", result["created"])
def test_creates_the_records_the_skipped_setup_wizard_would(self) -> None:
self._run()
records = self.frappe.records
self.assertIn("Transit", records["Warehouse Type"])
self.assertEqual(records["Territory"]["All Territories"]["is_group"], 1)
self.assertIn("Unit", records["UOM"])
self.assertIn("Sales", records["Opportunity Type"])
def test_creates_a_fiscal_year_covering_today(self) -> None:
self._run()
fiscal_year = self.frappe.records["Fiscal Year"]["2026"]
self.assertEqual(fiscal_year["year_start_date"], dt.date(2026, 1, 1))
self.assertEqual(fiscal_year["year_end_date"], dt.date(2026, 12, 31))
def test_reuses_a_fiscal_year_that_already_covers_today(self) -> None:
self.frappe.seed(
"Fiscal Year",
"2026-2027",
year="2026-2027",
year_start_date=dt.date(2026, 4, 1),
year_end_date=dt.date(2027, 3, 31),
disabled=0,
)
self._run()
self.assertEqual(list(self.frappe.records["Fiscal Year"]), ["2026-2027"])
def test_ignores_a_fiscal_year_of_another_company(self) -> None:
self.frappe.seed(
"Fiscal Year",
"2026 Other",
year="2026 Other",
year_start_date=dt.date(2026, 1, 1),
year_end_date=dt.date(2026, 12, 31),
disabled=0,
companies=[SimpleNamespace(company="Other")],
)
self._run()
self.assertIn("2026", self.frappe.records["Fiscal Year"])
def test_creates_the_tax_accounts_under_duties_and_taxes(self) -> None:
self._run()
accounts = self.frappe.records["Account"]
for account_name in TAX_ACCOUNTS:
account = accounts[f"{account_name} - IN"]
self.assertEqual(account["parent_account"], "Duties and Taxes - IN")
self.assertEqual(account["account_type"], "Tax")
self.assertEqual(account["is_group"], 0)
self.assertEqual(account["company"], COMPANY)
def test_makes_the_company_the_global_default_with_its_country_and_currency(
self,
) -> None:
self._run()
defaults = self.frappe.singles["Global Defaults"]
self.assertEqual(defaults["default_company"], COMPANY)
self.assertEqual(defaults["country"], "Germany")
self.assertEqual(defaults["default_currency"], "EUR")
def test_keeps_a_global_default_company_that_is_already_set(self) -> None:
self.frappe.singles["Global Defaults"] = {
"default_company": "Other",
"default_currency": "USD",
}
self._run()
self.assertEqual(
self.frappe.singles["Global Defaults"],
{"default_company": "Other", "default_currency": "USD"},
)
def test_second_run_creates_nothing(self) -> None:
self._run()
before = json.dumps([self.frappe.records, self.frappe.singles], default=str)
result = self._run()
self.assertEqual(result["created"], [])
self.assertEqual(
json.dumps([self.frappe.records, self.frappe.singles], default=str), before
)
def test_existing_company_keeps_its_own_values(self) -> None:
_seed_company(self.frappe, COMPANY, "XY", default_cash_account="Cash - XY")
result = self._run()
company = self._company()
self.assertEqual(company["abbr"], "XY")
self.assertEqual(company["country"], "United States")
self.assertEqual(company["default_currency"], "USD")
self.assertEqual(company["default_cash_account"], "Cash - XY")
self.assertEqual(result["abbr"], "XY")
self.assertNotIn(f"Company: {COMPANY}", result["created"])
def test_existing_company_gets_tax_accounts_under_its_own_abbreviation(
self,
) -> None:
_seed_company(self.frappe, COMPANY, "XY", default_cash_account="Cash - XY")
self._run()
self.assertIn("VAT 19% - XY", self.frappe.records["Account"])
self.assertNotIn("VAT 19% - IN", self.frappe.records["Account"])
def test_fills_an_empty_cash_account_with_the_first_cash_leaf(self) -> None:
_seed_company(self.frappe, COMPANY, "XY")
self.frappe.seed(
"Account",
"Till - XY",
company=COMPANY,
account_type="Cash",
is_group=0,
lft=21,
)
self.frappe.seed(
"Account",
"Cash - XY",
company=COMPANY,
account_type="Cash",
is_group=0,
lft=20,
)
self._run()
self.assertEqual(self._company()["default_cash_account"], "Cash - XY")
def test_falls_back_to_a_bank_account_without_a_cash_leaf(self) -> None:
_seed_company(self.frappe, COMPANY, "XY")
self.frappe.seed(
"Account",
"Bank - XY",
company=COMPANY,
account_type="Bank",
is_group=0,
lft=40,
)
self._run()
self.assertEqual(self._company()["default_cash_account"], "Bank - XY")
def test_company_without_cash_or_bank_account_fails(self) -> None:
_seed_company(self.frappe, COMPANY, "XY")
with self.assertRaisesRegex(SystemExit, "neither a Cash nor a Bank account"):
self._run()
def test_company_without_a_tax_group_fails(self) -> None:
self.frappe.seed(
"Company",
COMPANY,
company_name=COMPANY,
abbr="XY",
country="Germany",
default_currency="EUR",
default_cash_account="Cash - XY",
)
with self.assertRaisesRegex(SystemExit, "Duties and Taxes - XY"):
self._run()
def test_abbreviation_of_another_company_fails(self) -> None:
_seed_company(self.frappe, "Other", "IN", default_cash_account="Cash - IN")
with self.assertRaises(LinkValidationError):
self._run()
def test_unknown_country_names_the_inventory_key_to_set(self) -> None:
with self.assertRaisesRegex(
SystemExit, "Nexusland.*service_provider.company.address.country"
):
self._run(COMPANY_COUNTRY="Nexusland")
self.assertEqual(self.frappe.calls, [])
def test_empty_company_name_names_the_inventory_key_to_set(self) -> None:
with self.assertRaisesRegex(SystemExit, "service_provider.company.titel"):
self._run(COMPANY_NAME=" ")
def test_empty_abbreviation_names_the_inventory_key_to_set(self) -> None:
with self.assertRaisesRegex(
SystemExit, "applications.web-app-erpnext.services.erpnext.company.abbr"
):
self._run(COMPANY_ABBR="")
def test_commits_and_closes_the_site(self) -> None:
self._run()
self.assertTrue(self.frappe.db.committed)
self.assertEqual(self.frappe.calls[-1], ("destroy",))
if __name__ == "__main__":
unittest.main()