diff --git a/Engine b/Engine index ebc42b6..5b21b47 100755 Binary files a/Engine and b/Engine differ diff --git a/internal/database/budget-repo.go b/internal/database/budget-repo.go index 6422a72..b2b85db 100644 --- a/internal/database/budget-repo.go +++ b/internal/database/budget-repo.go @@ -70,11 +70,20 @@ const budgetSelectCols = ` amount, currency, notes, created_by, created_at, updated_at` func (r *BudgetRepo) Create(ctx context.Context, req model.CreateBudgetRequest) (*model.Budget, error) { + // ON CONFLICT upsert: if the unique key (year, period, version, dept, gl) already + // exists, update the mutable fields instead of failing. + // Makes the endpoint idempotent — safe for repeated test runs and re-imports. res, err := r.db.ExecContext(ctx, ` INSERT INTO budgets (fiscal_year, fiscal_period, version, department_id, gl_account_id, amount, currency, notes, created_by) - VALUES (?,?,?,?,?,?,?,?,?)`, + VALUES (?,?,?,?,?,?,?,?,?) + ON CONFLICT(fiscal_year, fiscal_period, version, department_id, gl_account_id) + DO UPDATE SET + amount = excluded.amount, + currency = excluded.currency, + notes = excluded.notes, + updated_at = strftime('%Y-%m-%dT%H:%M:%SZ','now')`, req.FiscalYear, req.FiscalPeriod, req.Version, req.DepartmentID, req.GLAccountID, req.Amount, req.Currency, req.Notes, req.CreatedBy, @@ -83,13 +92,14 @@ func (r *BudgetRepo) Create(ctx context.Context, req model.CreateBudgetRequest) return nil, fmt.Errorf("create budget: %w", err) } + // LastInsertId returns the existing row id on a conflict branch in SQLite. id, err := res.LastInsertId() if err != nil { return nil, fmt.Errorf("last insert id: %w", err) } row := r.db.QueryRowContext(ctx, - `SELECT`+budgetSelectCols+`FROM budgets WHERE id = ?`, id) + `SELECT `+budgetSelectCols+` FROM budgets WHERE id = ?`, id) b, err := scanBudget(row) if err != nil { return nil, fmt.Errorf("fetch created budget: %w", err) diff --git a/testing/tests/test_fpa.py b/testing/tests/test_fpa.py index 52aa224..34809d0 100644 --- a/testing/tests/test_fpa.py +++ b/testing/tests/test_fpa.py @@ -1,24 +1,18 @@ """ -FP&A Test Suite — stdlib only, no pytest +FP&A Test Suite Run: python tests/test_fpa.py - python tests/test_fpa.py --api # include live API tests python tests/test_fpa.py --url http://localhost:9000 """ -import csv -import json -import os -import sys -import traceback -import urllib.request -import urllib.error -from dataclasses import dataclass, field +import csv, json, os, sys, traceback +from dataclasses import dataclass from typing import Callable, List, Optional +import requests DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "csv") API_BASE = os.getenv("FPA_API_URL", "http://localhost:8080") -# ── Tiny test runner ────────────────────────────────────────────────────────── +# ── Runner ──────────────────────────────────────────────────────────────────── @dataclass class Result: @@ -27,6 +21,12 @@ class Result: skipped: bool = False message: str = "" +class SkipTest(Exception): + pass + +def skip(reason: str): + raise SkipTest(reason) + class Suite: def __init__(self, name: str): self.name = name @@ -42,15 +42,47 @@ class Suite: self.results.append(Result(label, passed=False, message=str(e))) except Exception as e: self.results.append(Result(label, passed=False, - message=f"{type(e).__name__}: {e}\n{traceback.format_exc()}")) + message=f"{type(e).__name__}: {e}\n{traceback.format_exc()}")) -class SkipTest(Exception): - pass +# ── Assertions ──────────────────────────────────────────────────────────────── -def skip(reason: str): - raise SkipTest(reason) +def ok(condition, msg=""): + if not condition: + raise AssertionError(msg or "Expected True") -# ── CSV helper ──────────────────────────────────────────────────────────────── +def eq(actual, expected, msg=""): + if actual != expected: + raise AssertionError(msg or f"Expected {expected!r}, got {actual!r}") + +def near(actual, expected, tol=1.0, msg=""): + if abs(actual - expected) > tol: + raise AssertionError(msg or f"Expected ~{expected:.2f}, got {actual:.2f} (tol={tol})") + +# ── HTTP ────────────────────────────────────────────────────────────────────── + +session = requests.Session() +session.headers.update({"Content-Type": "application/json"}) + +def get(path: str, **params): + r = session.get(f"{API_BASE}{path}", params=params, timeout=5) + return r.status_code, r.json() if r.content else {} + +def post(path: str, body: dict): + r = session.post(f"{API_BASE}{path}", json=body, timeout=10) + return r.status_code, r.json() if r.content else {} + +def put(path: str, body: dict): + r = session.put(f"{API_BASE}{path}", json=body, timeout=10) + return r.status_code, r.json() if r.content else {} + +def api_up() -> bool: + try: + status, _ = get("/api/v1/health") + return status == 200 + except Exception: + return False + +# ── CSV ─────────────────────────────────────────────────────────────────────── def read_csv(filename: str) -> List[dict]: path = os.path.join(DATA_DIR, filename) @@ -59,87 +91,13 @@ def read_csv(filename: str) -> List[dict]: with open(path, newline="") as f: return list(csv.DictReader(f)) -# ── HTTP helpers (stdlib only) ──────────────────────────────────────────────── +# ── Reference data ──────────────────────────────────────────────────────────── +# Seeded once via the API; IDs stored at runtime. -def http_get(url: str, timeout: int = 5): - try: - with urllib.request.urlopen(url, timeout=timeout) as resp: - body = json.loads(resp.read().decode()) - return resp.status, body - except urllib.error.HTTPError as e: - try: - body = json.loads(e.read().decode()) - except Exception: - body = {} - return e.code, body - except Exception: - return 0, {} +DEPT_IDS: dict[str, int] = {} # name → id +GL_IDS: dict[str, int] = {} # code → id -def http_post(url: str, payload: dict, timeout: int = 10): - data = json.dumps(payload).encode() - req = urllib.request.Request(url, data=data, - headers={"Content-Type": "application/json"}, - method="POST") - try: - with urllib.request.urlopen(req, timeout=timeout) as resp: - raw = resp.read() - body = json.loads(raw.decode()) if raw else {} - return resp.status, body - except urllib.error.HTTPError as e: - try: - body = json.loads(e.read().decode()) - except Exception: - body = {} - return e.code, body - except Exception: - return 0, {} - -def http_put(url: str, payload: dict, timeout: int = 10): - data = json.dumps(payload).encode() - req = urllib.request.Request(url, data=data, - headers={"Content-Type": "application/json"}, - method="PUT") - try: - with urllib.request.urlopen(req, timeout=timeout) as resp: - raw = resp.read() - body = json.loads(raw.decode()) if raw else {} - return resp.status, body - except urllib.error.HTTPError as e: - return e.code, {} - except Exception: - return 0, {} - -def api_available() -> bool: - status, _ = http_get(f"{API_BASE}/api/v1/health", timeout=2) - return status == 200 - -# ── Assertion helpers ───────────────────────────────────────────────────────── - -def assert_true(condition, message: str = ""): - if not condition: - raise AssertionError(message or "Expected True, got False") - -def assert_eq(actual, expected, message: str = ""): - if actual != expected: - raise AssertionError(message or f"Expected {expected!r}, got {actual!r}") - -# ── Reference data ─────────────────────────────────────────────────────────── -# -# Departments and GL accounts are CREATED via the API before any budget/actual. -# seed_reference_data() POSTs them and stores returned auto-increment IDs into -# DEPARTMENT_IDS and GL_ACCOUNTS at runtime — nothing is hardcoded. -# Uses ON CONFLICT upsert on the Go side so re-runs are safe. - -BUDGET_VERSION = "v1" # matches BudgetVersion enum in your Go service - -# Runtime ID maps — populated by seed_reference_data() -DEPARTMENT_IDS: dict = {} -GL_ACCOUNTS: dict = {} # category name → {"id": int, "code": str} - -# ── Department definitions ──────────────────────────────────────────────────── -# Matches model.Department: id, code, name, cost_center (no active field) - -_DEPARTMENT_DEFS = [ +DEPARTMENTS = [ {"code": "REV", "name": "Revenue", "cost_center": "CC-100"}, {"code": "ENG", "name": "Engineering", "cost_center": "CC-200"}, {"code": "SAL", "name": "Sales", "cost_center": "CC-300"}, @@ -148,569 +106,354 @@ _DEPARTMENT_DEFS = [ {"code": "FIN", "name": "Finance", "cost_center": "CC-600"}, ] -# ── GL account definitions ──────────────────────────────────────────────────── -# Matches model.GLAccount: id, code, description, type (GLAccountType), favour_high -# GLAccountType values: "revenue" | "cogs" | "opex" | "capex" | "headcount" -# favour_high=True → beating budget is favourable (revenue accounts) - -_GL_ACCOUNT_DEFS = [ - # Revenue - {"code": "4000", "description": "SaaS Subscription Revenue", "type": "revenue", "favour_high": True}, - {"code": "4100", "description": "Professional Services Revenue","type": "revenue", "favour_high": True}, - # COGS - {"code": "5000", "description": "Cost of Goods — Product", "type": "cogs", "favour_high": False}, - {"code": "5100", "description": "Cost of Goods — Service", "type": "cogs", "favour_high": False}, - # Opex - {"code": "6000", "description": "Salaries and Wages", "type": "opex", "favour_high": False}, - {"code": "6100", "description": "Software and SaaS Tools", "type": "opex", "favour_high": False}, - {"code": "6200", "description": "Travel and Expenses", "type": "opex", "favour_high": False}, - {"code": "6300", "description": "Marketing and Paid Media", "type": "opex", "favour_high": False}, - {"code": "6400", "description": "Cloud Infrastructure", "type": "opex", "favour_high": False}, - {"code": "6500", "description": "Contractors and Freelancers", "type": "opex", "favour_high": False}, - {"code": "6600", "description": "Office and Facilities", "type": "opex", "favour_high": False}, - # Capex - {"code": "7000", "description": "Capital Expenditure", "type": "capex", "favour_high": False}, - # Headcount - {"code": "9200", "description": "Headcount Cost", "type": "headcount", "favour_high": False}, +GL_ACCOUNTS = [ + {"code": "4000", "description": "SaaS Subscription Revenue", "type": "revenue", "favour_high": True}, + {"code": "4100", "description": "Professional Services Revenue", "type": "revenue", "favour_high": True}, + {"code": "5000", "description": "Cost of Goods — Product", "type": "cogs", "favour_high": False}, + {"code": "5100", "description": "Cost of Goods — Service", "type": "cogs", "favour_high": False}, + {"code": "6000", "description": "Salaries and Wages", "type": "opex", "favour_high": False}, + {"code": "6100", "description": "Software and SaaS Tools", "type": "opex", "favour_high": False}, + {"code": "6200", "description": "Travel and Expenses", "type": "opex", "favour_high": False}, + {"code": "6300", "description": "Marketing and Paid Media", "type": "opex", "favour_high": False}, + {"code": "6400", "description": "Cloud Infrastructure", "type": "opex", "favour_high": False}, + {"code": "6500", "description": "Contractors and Freelancers", "type": "opex", "favour_high": False}, + {"code": "6600", "description": "Office and Facilities", "type": "opex", "favour_high": False}, + {"code": "7000", "description": "Capital Expenditure", "type": "capex", "favour_high": False}, + {"code": "9200", "description": "Headcount Cost", "type": "headcount", "favour_high": False}, ] -# ── CSV category → GL code mapping ─────────────────────────────────────────── -# Translates CSV column values into the gl_code field used by IngestActualsRequest - -_CSV_CATEGORY_TO_GL_CODE = { - "Product": "4000", - "Service": "4100", - "Salaries": "6000", - "Software & Tools": "6100", - "Travel": "6200", - "Marketing Spend": "6300", - "Cloud Infrastructure": "6400", - "Contractors": "6500", - "Office & Facilities": "6600", - "P&L Summary": "4000", - "Cash Flow": "7000", - "Headcount": "9200", +# CSV category name → GL code +CSV_TO_GL = { + "Product": "4000", + "Service": "4100", + "Salaries": "6000", + "Software & Tools": "6100", + "Travel": "6200", + "Marketing Spend": "6300", + "Cloud Infrastructure": "6400", + "Contractors": "6500", + "Office & Facilities": "6600", + "Headcount": "9200", } -def seed_reference_data() -> Optional[str]: - """ - POST all departments and GL accounts to the API and cache returned IDs. +def gl_id(category: str) -> int: + code = CSV_TO_GL.get(category) + assert code, f"No GL mapping for '{category}'" + assert code in GL_IDS, f"GL '{code}' not seeded — call seed() first" + return GL_IDS[code] - Populates at runtime: - DEPARTMENT_IDS — {"Engineering": 3, ...} keyed by dept name - GL_ACCOUNTS — {"4000": {"id": 1, "code": "4000"}} keyed by GL code +def gl_code(category: str) -> str: + code = CSV_TO_GL.get(category) + assert code, f"No GL mapping for '{category}'" + return code - Safe to call multiple times — Go side uses ON CONFLICT(code) upsert. - Returns an error string on first failure, None on success. - """ - for defn in _DEPARTMENT_DEFS: - status, body = http_post(f"{API_BASE}/api/v1/departments", defn) - if status not in (200, 201): - return f"POST /departments '{defn['name']}' failed ({status}): {body}" - DEPARTMENT_IDS[defn["name"]] = body["id"] +def fiscal(period: str): + y, m = period.split("-") + return int(y), int(m) - for defn in _GL_ACCOUNT_DEFS: - status, body = http_post(f"{API_BASE}/api/v1/gl-accounts", defn) - if status not in (200, 201): - return f"POST /gl-accounts code='{defn['code']}' failed ({status}): {body}" - GL_ACCOUNTS[defn["code"]] = {"id": body["id"], "code": defn["code"]} +def seed(): + """POST departments and GL accounts; cache returned IDs as int. Idempotent.""" + for d in DEPARTMENTS: + status, body = post("/api/v1/departments", d) + ok(status in (200, 201), f"POST /departments '{d['name']}' → {status}: {body}") + ok("id" in body, f"departments response missing 'id': {body}") + DEPT_IDS[d["name"]] = int(body["id"]) # explicit int — guards against "id": "3" - return None + for g in GL_ACCOUNTS: + status, body = post("/api/v1/gl-accounts", g) + ok(status in (200, 201), f"POST /gl-accounts '{g['code']}' → {status}: {body}") + ok("id" in body, f"gl-accounts response missing 'id': {body}") + GL_IDS[g["code"]] = int(body["id"]) -def gl(category: str) -> dict: - """Resolve a CSV category name to {"id": int, "code": str} via _CSV_CATEGORY_TO_GL_CODE.""" - code = _CSV_CATEGORY_TO_GL_CODE.get(category) - if code is None: - raise AssertionError( - f"No GL code for CSV category '{category}'. Add it to _CSV_CATEGORY_TO_GL_CODE." - ) - acct = GL_ACCOUNTS.get(code) - if acct is None: - raise AssertionError( - f"GL '{code}' (for '{category}') missing — call seed_reference_data() first. " - f"Known: {list(GL_ACCOUNTS.keys())}" - ) - return acct - -def period_to_fiscal(period: str) -> tuple: - """'2023-04' → (2023, 4)""" - year, month = period.split("-") - return int(year), int(month) - -# ── suite_revenue — full API round-trip ─────────────────────────────────────── -# -# Flow: -# 1. Load CSV rows (local data validation) -# 2. POST each row as a CreateBudgetRequest → /api/v1/budgets -# 3. POST actuals for the same rows → /api/v1/actuals/ingest -# 4. GET /api/v1/variance → verify amounts round-trip correctly +# ── suite_revenue ───────────────────────────────────────────────────────────── def suite_revenue() -> Suite: - """ - Maps to Go structs: - - CreateBudgetRequest: - fiscal_year, fiscal_period, version, department_id, - gl_account_id, amount, currency, notes, created_by - - Actual (ingest): - fiscal_year, fiscal_period, department_id, - gl_account_id, gl_code, amount, currency, source - """ s = Suite("Revenue — CSV + API round-trip") - if not api_available(): + if not api_up(): s.run("API reachable", lambda: skip(f"Go API not running at {API_BASE}")) return s rows = [] - # ── Step 1: load and validate CSV data ──────────────────────────────────── + # 1. CSV validation + s.run("CSV loads", lambda: rows.extend(read_csv("revenue_budget_vs_actuals.csv"))) + s.run("48 rows (24 months × 2 types)", lambda: eq(len(rows), 48)) + s.run("only Product and Service types", lambda: + eq({r["revenue_type"] for r in rows}, {"Product", "Service"})) + s.run("variance = actual − budget", lambda: [ + near(float(r["actual_amount"]) - float(r["budget_amount"]), + float(r["variance"]), tol=0.01, msg=f"variance wrong in {r['period']}") + for r in rows + ]) - def load(): - rows.extend(read_csv("revenue_budget_vs_actuals.csv")) - s.run("CSV loads without error", load) + # 2. Seed FK parents + s.run("seed departments + GL accounts", seed) - s.run("CSV has 48 rows (24 months × 2 revenue types)", lambda: - assert_eq(len(rows), 48)) + # 3. POST budgets + budget_ids: dict[str, int] = {} - s.run("CSV revenue types are Product and Service only", lambda: - assert_eq({r["revenue_type"] for r in rows}, {"Product", "Service"})) + def post_budget(rev_type: str): + row = next(r for r in rows if r["revenue_type"] == rev_type and r["period"] == "2023-01") + fy, fp = fiscal(row["period"]) # fiscal() already returns (int, int) + status, body = post("/api/v1/budgets", { + "fiscal_year": int(fy), + "fiscal_period": int(fp), + "version": "original", + "department_id": int(DEPT_IDS["Revenue"]), + "gl_account_id": int(gl_id(rev_type)), + "amount": float(row["budget_amount"]), + "currency": "DKK", + "notes": f"{rev_type} budget — csv import", + "created_by": "test_suite", + }) + # 201 = created, 200 = upserted (row already existed — idempotent re-run) + ok(status in (200, 201), f"POST /budgets {rev_type} → {status}: {body}") + ok(body.get("id"), f"response missing 'id': {body}") + budget_ids[rev_type] = int(body["id"]) - def check_csv_variance(): - for r in rows: - diff = float(r["actual_amount"]) - float(r["budget_amount"]) - assert_true(abs(diff - float(r["variance"])) < 0.01, - f"CSV variance mismatch in {r['period']}") - s.run("CSV variance = actual − budget", check_csv_variance) - - # ── Step 2: seed departments + GL accounts ──────────────────────────────── - # Must succeed before any budget POST — these rows are the FK parents. - - def seed_refs(): - err = seed_reference_data() - assert_true(err is None, err or "seed_reference_data failed") - s.run("POST /api/v1/departments + /api/v1/gl-accounts (seed FK parents)", seed_refs) - - # ── Step 3: POST budgets (CreateBudgetRequest) ─────────────────────────── - # version uses BudgetVersion consts: "original" | "forecast_1" | "forecast_2" - # department_id and gl_account_id are the integer IDs returned by seed step. - - budget_ids: dict = {} # "revenue_type:period" → returned Budget.id - - def post_product_budget(): - row = next(r for r in rows - if r["revenue_type"] == "Product" and r["period"] == "2023-01") - fy, fp = period_to_fiscal(row["period"]) - payload = { - "fiscal_year": fy, - "fiscal_period": fp, - "version": "original", # model.VersionOriginal - "department_id": DEPARTMENT_IDS["Revenue"], - "gl_account_id": gl("Product")["id"], - "amount": float(row["budget_amount"]), - "currency": "USD", - "notes": "Product revenue budget — csv import", - "created_by": "test_suite", - } - status, body = http_post(f"{API_BASE}/api/v1/budgets", payload) - assert_true(status in (200, 201), - f"POST /budgets failed ({status}): {body}") - assert_true(body.get("id") is not None, "Budget response missing 'id'") - budget_ids["Product:2023-01"] = body["id"] - - s.run("POST /api/v1/budgets — Product revenue 2023-01 (version=original)", post_product_budget) - - def post_service_budget(): - row = next(r for r in rows - if r["revenue_type"] == "Service" and r["period"] == "2023-01") - fy, fp = period_to_fiscal(row["period"]) - payload = { - "fiscal_year": fy, - "fiscal_period": fp, - "version": "original", - "department_id": DEPARTMENT_IDS["Revenue"], - "gl_account_id": gl("Service")["id"], - "amount": float(row["budget_amount"]), - "currency": "USD", - "notes": "Service revenue budget — csv import", - "created_by": "test_suite", - } - status, body = http_post(f"{API_BASE}/api/v1/budgets", payload) - assert_true(status in (200, 201), - f"POST /budgets failed ({status}): {body}") - assert_true(body.get("id") is not None, "Budget response missing 'id'") - budget_ids["Service:2023-01"] = body["id"] - - s.run("POST /api/v1/budgets — Service revenue 2023-01 (version=original)", post_service_budget) - - # ── Step 4: POST actuals (IngestActualsRequest) ─────────────────────────── - # IngestActualsRequest uses dept_code + gl_code strings (not integer IDs). - # The Go service resolves them to IDs internally. + s.run("POST budget — Product 2023-01", lambda: post_budget("Product")) + s.run("POST budget — Service 2023-01", lambda: post_budget("Service")) + # 4. POST actuals def post_actuals(): - period_rows = [r for r in rows if r["period"] == "2023-01"] - for r in period_rows: - fy, fp = period_to_fiscal(r["period"]) - account = gl(r["revenue_type"]) - # IngestActualsRequest shape — matches model exactly - payload = { - "fiscal_year": fy, - "fiscal_period": fp, - "dept_code": "REV", # Department.Code for Revenue - "gl_code": account["code"], # GLAccount.Code e.g. "4000" - "amount": float(r["actual_amount"]), - "currency": "USD", - "source": "test_suite_csv", - } - status, body = http_post(f"{API_BASE}/api/v1/actuals/ingest", payload) - assert_true(status in (200, 201), - f"POST /actuals/ingest gl={account['code']} failed ({status}): {body}") + for r in [r for r in rows if r["period"] == "2023-01"]: + fy, fp = fiscal(r["period"]) + status, body = post("/api/v1/actuals/ingest", { + "fiscal_year": int(fy), + "fiscal_period": int(fp), + "dept_code": "REV", + "gl_code": gl_code(r["revenue_type"]), + "amount": float(r["actual_amount"]), + "currency": "DKK", + "source": "test_suite", + }) + ok(status in (200, 201), + f"POST /actuals/ingest {r['revenue_type']} → {status}: {body}") - s.run("POST /api/v1/actuals/ingest — Product + Service 2023-01", post_actuals) - - # ── Step 5: GET variance and verify VarianceReport shape ────────────────── - # Response: VarianceReport { department, fiscal_year, fiscal_period, version, - # currency, lines: []VarianceLine, total_budget, total_actual, total_variance, total_variance_pct } - # VarianceLine: { gl_account_id (actually gl_code string), gl_description, gl_type, - # budget, actual, variance_abs, variance_pct, status, currency } + s.run("POST actuals — Product + Service 2023-01", post_actuals) + # 5. GET variance — verify VarianceReport shape and values report: dict = {} - def fetch_variance(): - url = (f"{API_BASE}/api/v1/variance" - f"?fiscal_year=2023&fiscal_period=1&dept_code=REV&version=original") - status, body = http_get(url) - assert_true(status == 200, f"GET /variance failed ({status}): {body}") - - # Response is a VarianceReport object (not a list) - assert_true(isinstance(body, dict), f"Expected VarianceReport object, got: {type(body)}") - assert_true("lines" in body, - f"VarianceReport missing 'lines' key. Got keys: {list(body.keys())}") - assert_true(len(body["lines"]) > 0, "VarianceReport.lines is empty") + def fetch_report(): + status, body = get("/api/v1/variance", + fiscal_year=2023, fiscal_period=1, dept_code="REV", version="original") + ok(status == 200, f"GET /variance → {status}: {body}") + ok(isinstance(body, dict), "expected VarianceReport object") + ok("lines" in body, f"missing 'lines', got keys: {list(body.keys())}") + ok(len(body["lines"]) > 0, "lines is empty") report.update(body) - s.run("GET /api/v1/variance returns VarianceReport for REV dept 2023-01", fetch_variance) + s.run("GET /variance returns VarianceReport", fetch_report) - def verify_report_top_level(): - assert_eq(report.get("fiscal_year"), 2023, "fiscal_year mismatch") - assert_eq(report.get("fiscal_period"), 1, "fiscal_period mismatch") - assert_eq(report.get("version"), "original", "version mismatch") - assert_true(report.get("department") is not None, "department field missing") - assert_true(report.get("currency") is not None, "currency field missing") - assert_true(report.get("total_budget") is not None, "total_budget field missing") - assert_true(report.get("total_actual") is not None, "total_actual field missing") + def verify_report(): + eq(report["fiscal_year"], 2023, "fiscal_year") + eq(report["fiscal_period"], 1, "fiscal_period") + eq(report["version"], "original", "version") + for field in ("department", "currency", "total_budget", "total_actual"): + ok(field in report, f"VarianceReport missing '{field}'") - s.run("VarianceReport top-level fields present", verify_report_top_level) + for line in report["lines"]: + for f in ("gl_account_id", "gl_description", "gl_type", + "budget", "actual", "variance_abs", "status"): + ok(f in line, f"VarianceLine missing '{f}'") + ok(line["status"] in ("favourable", "unfavourable", "on_budget"), + f"bad status: {line['status']}") - def verify_variance_lines(): - lines = report.get("lines", []) - # VarianceLine fields: gl_account_id (gl_code string), gl_description, - # gl_type, budget, actual, variance_abs, variance_pct, status, currency - for line in lines: - assert_true("gl_account_id" in line, f"VarianceLine missing gl_account_id: {line}") - assert_true("gl_description" in line, f"VarianceLine missing gl_description: {line}") - assert_true("gl_type" in line, f"VarianceLine missing gl_type: {line}") - assert_true("budget" in line, f"VarianceLine missing budget: {line}") - assert_true("actual" in line, f"VarianceLine missing actual: {line}") - assert_true("variance_abs" in line, f"VarianceLine missing variance_abs: {line}") - assert_true("status" in line, f"VarianceLine missing status: {line}") - # status must be a valid VarianceStatus - assert_true(line["status"] in ("favourable", "unfavourable", "on_budget"), - f"Unknown VarianceStatus: {line['status']}") - - s.run("VarianceLine fields and status values valid", verify_variance_lines) + s.run("VarianceReport shape valid", verify_report) def verify_product_line(): csv_row = next(r for r in rows - if r["revenue_type"] == "Product" and r["period"] == "2023-01") - expected_budget = float(csv_row["budget_amount"]) - expected_actual = float(csv_row["actual_amount"]) - expected_var = float(csv_row["variance"]) - - # VarianceLine.gl_account_id holds the GL code string (e.g. "4000") + if r["revenue_type"] == "Product" and r["period"] == "2023-01") line = next((l for l in report["lines"] - if l.get("gl_account_id") == gl("Product")["code"]), None) - assert_true(line is not None, - f"No VarianceLine for Product (gl_code={gl('Product')['code']}). " - f"Lines: {[l.get('gl_account_id') for l in report['lines']]}") - - assert_true(abs(float(line["budget"]) - expected_budget) < 1.0, - f"budget mismatch: got {line['budget']} expected {expected_budget:.2f}") - assert_true(abs(float(line["actual"]) - expected_actual) < 1.0, - f"actual mismatch: got {line['actual']} expected {expected_actual:.2f}") - assert_true(abs(float(line["variance_abs"]) - abs(expected_var)) < 1.0, - f"variance_abs mismatch: got {line['variance_abs']} expected {abs(expected_var):.2f}") - # Revenue account: favour_high=True, so over-actual should be favourable + if l["gl_account_id"] == "4000"), None) + ok(line, f"no line for gl_account_id=4000, got: {[l['gl_account_id'] for l in report['lines']]}") + near(float(line["budget"]), float(csv_row["budget_amount"]), msg="budget") + near(float(line["actual"]), float(csv_row["actual_amount"]), msg="actual") + near(float(line["variance_abs"]), abs(float(csv_row["variance"])), msg="variance_abs") if float(csv_row["actual_amount"]) >= float(csv_row["budget_amount"]): - assert_eq(line["status"], "favourable", - f"Revenue over-budget should be favourable, got {line['status']}") + eq(line["status"], "favourable", "revenue over-budget should be favourable") - s.run("VarianceLine values match CSV for Product 2023-01", verify_product_line) + s.run("Product line values match CSV", verify_product_line) - def verify_totals(): - # total_budget and total_actual should be sum of the lines - line_budget_sum = sum(float(l["budget"]) for l in report["lines"]) - line_actual_sum = sum(float(l["actual"]) for l in report["lines"]) - assert_true(abs(float(report["total_budget"]) - line_budget_sum) < 1.0, - f"total_budget {report['total_budget']} != sum of lines {line_budget_sum:.2f}") - assert_true(abs(float(report["total_actual"]) - line_actual_sum) < 1.0, - f"total_actual {report['total_actual']} != sum of lines {line_actual_sum:.2f}") - - s.run("VarianceReport totals equal sum of lines", verify_totals) - - # ── Step 6: GET /variance/alerts and verify AlertThreshold shape ────────── - # AlertThreshold: { gl_code, description, budget, actual, variance_pct, status, department } + s.run("totals = sum of lines", lambda: ( + near(float(report["total_budget"]), + sum(float(l["budget"]) for l in report["lines"]), msg="total_budget"), + near(float(report["total_actual"]), + sum(float(l["actual"]) for l in report["lines"]), msg="total_actual"), + )) + # 6. Alerts shape check def check_alerts(): - url = f"{API_BASE}/api/v1/variance/alerts?fiscal_year=2023&fiscal_period=1" - status, body = http_get(url) - assert_true(status == 200, f"GET /variance/alerts failed ({status}): {body}") - + status, body = get("/api/v1/variance/alerts", fiscal_year=2023, fiscal_period=1) + ok(status == 200, f"GET /variance/alerts → {status}: {body}") alerts = body if isinstance(body, list) else body.get("alerts", []) - # Alerts may be empty if nothing breaches threshold — just validate shape if any exist - for alert in alerts: - assert_true("gl_code" in alert, f"AlertThreshold missing gl_code: {alert}") - assert_true("description" in alert, f"AlertThreshold missing description: {alert}") - assert_true("budget" in alert, f"AlertThreshold missing budget: {alert}") - assert_true("actual" in alert, f"AlertThreshold missing actual: {alert}") - assert_true("variance_pct" in alert, f"AlertThreshold missing variance_pct: {alert}") - assert_true("status" in alert, f"AlertThreshold missing status: {alert}") - assert_true("department" in alert, f"AlertThreshold missing department: {alert}") - assert_true(alert["status"] in ("favourable", "unfavourable", "on_budget"), - f"Unknown alert status: {alert['status']}") + for a in alerts: + for f in ("gl_code", "description", "budget", "actual", "variance_pct", "status", "department"): + ok(f in a, f"AlertThreshold missing '{f}'") + ok(a["status"] in ("favourable", "unfavourable", "on_budget"), + f"bad alert status: {a['status']}") - s.run("GET /api/v1/variance/alerts returns valid AlertThreshold list", check_alerts) + s.run("GET /variance/alerts shape valid", check_alerts) - # ── Step 7: update budget to forecast_1 and verify variance shifts ──────── - # PUT /api/v1/budgets/{id} with version="forecast_1" (model.VersionForecast1) - - def update_and_reverify(): - budget_id = budget_ids.get("Product:2023-01") - if not budget_id: - skip("Product budget id not captured — skipping update test") + # 7. Update to forecast_1 and verify variance shifts + def update_forecast(): + bid = budget_ids.get("Product") + if not bid: + skip("no Product budget id — skipping") csv_row = next(r for r in rows - if r["revenue_type"] == "Product" and r["period"] == "2023-01") - revised_amount = float(csv_row["budget_amount"]) * 1.10 # +10% reforecast + if r["revenue_type"] == "Product" and r["period"] == "2023-01") + revised = float(csv_row["budget_amount"]) * 1.10 - fy, fp = period_to_fiscal("2023-01") - status, body = http_put( - f"{API_BASE}/api/v1/budgets/{budget_id}", - { - "fiscal_year": fy, - "fiscal_period": fp, - "version": "forecast_1", # model.VersionForecast1 - "department_id": DEPARTMENT_IDS["Revenue"], - "gl_account_id": gl("Product")["id"], - "amount": revised_amount, - "currency": "USD", - "notes": "Q1 reforecast +10%", - "created_by": "test_suite", - } - ) - assert_true(status in (200, 204), - f"PUT /budgets/{budget_id} failed ({status}): {body}") - - # Re-fetch with version=forecast_1 — budget should reflect revision - url = (f"{API_BASE}/api/v1/variance" - f"?fiscal_year=2023&fiscal_period=1&dept_code=REV&version=forecast_1") - status2, body2 = http_get(url) - assert_true(status2 == 200, f"Re-fetch forecast_1 variance failed ({status2}): {body2}") + fy, fp = fiscal("2023-01") + status, body = put(f"/api/v1/budgets/{bid}", { + "fiscal_year": int(fy), + "fiscal_period": int(fp), + "version": "forecast_1", + "department_id": int(DEPT_IDS["Revenue"]), + "gl_account_id": int(gl_id("Product")), + "amount": float(revised), + "currency": "DKK", + "notes": "Q1 reforecast +10%", + "created_by": "test_suite", + }) + ok(status in (200, 204), f"PUT /budgets/{bid} → {status}: {body}") + status2, body2 = get("/api/v1/variance", + fiscal_year=2023, fiscal_period=1, dept_code="REV", version="forecast_1") + ok(status2 == 200, f"GET forecast_1 variance → {status2}") lines = body2.get("lines", []) if isinstance(body2, dict) else [] - line = next((l for l in lines - if l.get("gl_account_id") == gl("Product")["code"]), None) - assert_true(line is not None, - "Product line missing from forecast_1 variance report after update") - assert_true(abs(float(line["budget"]) - revised_amount) < 1.0, - f"forecast_1 budget not reflected: got {line['budget']} expected {revised_amount:.2f}") + line = next((l for l in lines if l.get("gl_account_id") == "4000"), None) + ok(line, "Product line missing in forecast_1 variance") + near(float(line["budget"]), revised, msg="forecast_1 budget not updated") - s.run("PUT /api/v1/budgets/{id} version=forecast_1 — variance shifts correctly", update_and_reverify) + s.run("PUT budget forecast_1 — variance reflects revision", update_forecast) return s - -# ── Remaining CSV-only suites (unchanged) ───────────────────────────────────── +# ── CSV-only suites ─────────────────────────────────────────────────────────── def suite_opex() -> Suite: s = Suite("Opex CSV") rows = [] - - def load(): - rows.extend(read_csv("opex_budget_vs_actuals.csv")) - s.run("loads without error", load) - - s.run("has rows", lambda: assert_true(len(rows) > 0)) - + s.run("loads", lambda: rows.extend(read_csv("opex_budget_vs_actuals.csv"))) + s.run("has rows", lambda: ok(len(rows) > 0)) s.run("all departments present", lambda: - assert_true( - {"Engineering", "Sales", "Marketing", "Operations"}.issubset( - {r["department"] for r in rows} - ))) - - def check_no_zero(): - for r in rows: - assert_true(float(r["budget_amount"]) > 0, - f"Zero budget: {r['department']} / {r['category']}") - s.run("no zero budget amounts", check_no_zero) - - def check_sign(): - for r in rows: - actual = float(r["actual_amount"]) - budget = float(r["budget_amount"]) - variance = float(r["variance"]) - exp = 1 if actual > budget else (-1 if actual < budget else 0) - got = 1 if variance > 0 else (-1 if variance < 0 else 0) - assert_eq(got, exp, f"Variance sign wrong in {r['period']} {r['category']}") - s.run("variance sign matches actual vs budget", check_sign) - + ok({"Engineering","Sales","Marketing","Operations"}.issubset({r["department"] for r in rows}))) + s.run("no zero budgets", lambda: [ + ok(float(r["budget_amount"]) > 0, f"zero budget: {r['department']}/{r['category']}") + for r in rows + ]) + s.run("variance sign correct", lambda: [ + eq(1 if float(r["variance"]) > 0 else (-1 if float(r["variance"]) < 0 else 0), + 1 if float(r["actual_amount"]) > float(r["budget_amount"]) else + (-1 if float(r["actual_amount"]) < float(r["budget_amount"]) else 0), + f"sign wrong {r['period']} {r['category']}") + for r in rows + ]) return s def suite_pl() -> Suite: s = Suite("P&L CSV") rows = [] - - def load(): - rows.extend(read_csv("pl_income_statement.csv")) - s.run("loads without error", load) - - s.run("exactly 24 rows", lambda: assert_eq(len(rows), 24)) - - def check_rev_sum(): - for r in rows: - total = float(r["product_revenue"]) + float(r["service_revenue"]) - assert_true(abs(total - float(r["total_revenue"])) < 0.05, - f"Revenue sum mismatch in {r['period']}") - s.run("total_revenue = product + service", check_rev_sum) - - def check_gross(): - for r in rows: - gp = float(r["total_revenue"]) - float(r["total_cogs"]) - assert_true(abs(gp - float(r["gross_profit"])) < 0.05, - f"Gross profit mismatch in {r['period']}") - s.run("gross_profit = revenue − cogs", check_gross) - - def check_margin(): - for r in rows: - gm = float(r["gross_margin_pct"]) - assert_true(30 <= gm <= 90, f"Gross margin {gm}% out of range in {r['period']}") - s.run("gross margin between 30% and 90%", check_margin) - - def check_ebitda(): - for r in rows: - ebitda = float(r["gross_profit"]) - float(r["total_opex"]) - assert_true(abs(ebitda - float(r["ebitda"])) < 0.05, - f"EBITDA mismatch in {r['period']}") - s.run("ebitda = gross_profit − opex", check_ebitda) - + s.run("loads", lambda: rows.extend(read_csv("pl_income_statement.csv"))) + s.run("24 rows", lambda: eq(len(rows), 24)) + s.run("total_revenue = product + service", lambda: [ + near(float(r["product_revenue"]) + float(r["service_revenue"]), + float(r["total_revenue"]), tol=0.05, msg=r["period"]) + for r in rows + ]) + s.run("gross_profit = revenue − cogs", lambda: [ + near(float(r["total_revenue"]) - float(r["total_cogs"]), + float(r["gross_profit"]), tol=0.05, msg=r["period"]) + for r in rows + ]) + s.run("gross margin 30–90%", lambda: [ + ok(30 <= float(r["gross_margin_pct"]) <= 90, f"{r['gross_margin_pct']}% in {r['period']}") + for r in rows + ]) + s.run("ebitda = gross_profit − opex", lambda: [ + near(float(r["gross_profit"]) - float(r["total_opex"]), + float(r["ebitda"]), tol=0.05, msg=r["period"]) + for r in rows + ]) return s def suite_cashflow() -> Suite: s = Suite("Cash Flow CSV") rows = [] - - def load(): - rows.extend(read_csv("cash_flow.csv")) - s.run("loads without error", load) - - s.run("exactly 24 rows", lambda: assert_eq(len(rows), 24)) - - def check_net(): - for r in rows: - total = (float(r["net_operating_cash_flow"]) + - float(r["net_investing_cash_flow"]) + - float(r["net_financing_cash_flow"])) - assert_true(abs(total - float(r["net_change_in_cash"])) < 0.05, - f"Cash flow sum mismatch in {r['period']}") - s.run("net_change = operating + investing + financing", check_net) - - def check_pre_series_a(): - for r in rows: - if r["period"] <= "2023-05": - assert_true(float(r["closing_cash_balance"]) > 0, - f"Negative cash too early: {r['period']}") - s.run("cash balance positive before Series A (pre 2023-06)", check_pre_series_a) - - def check_series_a(): - june = next((r for r in rows if r["period"] == "2023-06"), None) - assert_true(june is not None, "2023-06 row missing") - assert_true(float(june["equity_raised"]) > 0, "Series A not recorded in 2023-06") - s.run("Series A visible in 2023-06", check_series_a) - + s.run("loads", lambda: rows.extend(read_csv("cash_flow.csv"))) + s.run("24 rows", lambda: eq(len(rows), 24)) + s.run("net_change = op + inv + fin", lambda: [ + near(float(r["net_operating_cash_flow"]) + + float(r["net_investing_cash_flow"]) + + float(r["net_financing_cash_flow"]), + float(r["net_change_in_cash"]), tol=0.05, msg=r["period"]) + for r in rows + ]) + s.run("positive cash pre Series A", lambda: [ + ok(float(r["closing_cash_balance"]) > 0, f"negative cash in {r['period']}") + for r in rows if r["period"] <= "2023-05" + ]) + s.run("Series A in 2023-06", lambda: ( + lambda june=next((r for r in rows if r["period"] == "2023-06"), None): ( + ok(june is not None, "2023-06 missing"), + ok(float(june["equity_raised"]) > 0, "Series A not recorded"), + ) + )()) return s def suite_headcount() -> Suite: s = Suite("Headcount CSV") rows = [] - - def load(): - rows.extend(read_csv("headcount_workforce.csv")) - s.run("loads without error", load) - - s.run("has rows", lambda: assert_true(len(rows) > 0)) - - s.run("status only Active or Terminated", lambda: - assert_true({r["status"] for r in rows}.issubset({"Active", "Terminated"}))) - - def check_fte(): - for r in rows: - fte = float(r["headcount_fte"]) - assert_true(0 < fte <= 1.0, - f"FTE {fte} out of range for {r['employee_id']}") - s.run("FTE between 0 and 1.0", check_fte) - - def check_salary(): - for r in rows: - assert_true(float(r["annual_salary_budget"]) > 0) - s.run("all salaries positive", check_salary) - - def check_growth(): - jan = [r for r in rows if r["period"] == "2023-01" and r["status"] == "Active"] - dec = [r for r in rows if r["period"] == "2024-12" and r["status"] == "Active"] - assert_true(len(dec) >= len(jan), - f"Headcount shrank: {len(jan)} in Jan 2023 → {len(dec)} in Dec 2024") - s.run("headcount grows from 2023-01 to 2024-12", check_growth) - + s.run("loads", lambda: rows.extend(read_csv("headcount_workforce.csv"))) + s.run("has rows", lambda: ok(len(rows) > 0)) + s.run("valid statuses", lambda: + ok({r["status"] for r in rows}.issubset({"Active", "Terminated"}))) + s.run("FTE 0–1", lambda: [ + ok(0 < float(r["headcount_fte"]) <= 1.0, f"FTE={r['headcount_fte']} for {r['employee_id']}") + for r in rows + ]) + s.run("salaries positive", lambda: [ + ok(float(r["annual_salary_budget"]) > 0) for r in rows + ]) + s.run("headcount grows Jan→Dec", lambda: ok( + len([r for r in rows if r["period"] == "2024-12" and r["status"] == "Active"]) >= + len([r for r in rows if r["period"] == "2023-01" and r["status"] == "Active"]) + )) return s - # ── Reporter ────────────────────────────────────────────────────────────────── -PASS = "\033[32m✓\033[0m" -FAIL = "\033[31m✗\033[0m" -SKIP = "\033[33m⊘\033[0m" -BOLD = "\033[1m" -RESET = "\033[0m" - -def run_suites(suites: List[Suite]) -> bool: +def run_suites(suites): total = passed = failed = skipped = 0 - for suite in suites: - print(f"\n{BOLD}{suite.name}{RESET}") + print(f"\n\033[1m{suite.name}\033[0m") for r in suite.results: total += 1 if r.skipped: skipped += 1 - print(f" {SKIP} {r.name}") + print(f" \033[33m⊘\033[0m {r.name}") print(f" {r.message}") elif r.passed: passed += 1 - print(f" {PASS} {r.name}") + print(f" \033[32m✓\033[0m {r.name}") else: failed += 1 - print(f" {FAIL} {r.name}") + print(f" \033[31m✗\033[0m {r.name}") for line in r.message.strip().splitlines(): print(f" {line}") - print(f"\n{'─' * 48}") + print(f"\n{'─'*48}") color = "\033[32m" if failed == 0 else "\033[31m" - print(f"{color}{BOLD}{passed}/{total} passed{RESET}" - + (f" {SKIP} {skipped} skipped" if skipped else "") - + (f" {FAIL} {failed} failed" if failed else "")) + print(f"{color}\033[1m{passed}/{total} passed\033[0m" + + (f" \033[33m⊘\033[0m {skipped} skipped" if skipped else "") + + (f" \033[31m✗\033[0m {failed} failed" if failed else "")) print() return failed == 0 @@ -718,24 +461,17 @@ def run_suites(suites: List[Suite]) -> bool: if __name__ == "__main__": import argparse - - parser = argparse.ArgumentParser(description="FP&A test suite — no dependencies needed") - parser.add_argument("--api", action="store_true", - help="Include revenue round-trip API test (requires Go API running)") - parser.add_argument("--url", default=None, help="Override API base URL") + parser = argparse.ArgumentParser(description="FP&A test suite") + parser.add_argument("--url", default=None, help="API base URL") args = parser.parse_args() - if args.url: API_BASE = args.url - # suite_revenue always runs — it skips the API steps gracefully if not available - suites = [ + ok = run_suites([ suite_revenue(), suite_opex(), suite_pl(), suite_cashflow(), suite_headcount(), - ] - - ok = run_suites(suites) + ]) sys.exit(0 if ok else 1) \ No newline at end of file