""" FP&A Test Suite Run: python tests/test_fpa.py python tests/test_fpa.py --url http://localhost:9000 """ 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") # ── Runner ──────────────────────────────────────────────────────────────────── @dataclass class Result: name: str passed: bool 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 self.results: List[Result] = [] def run(self, label: str, fn: Callable): try: fn() self.results.append(Result(label, passed=True)) except SkipTest as e: self.results.append(Result(label, passed=False, skipped=True, message=str(e))) except AssertionError as e: 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()}")) # ── Assertions ──────────────────────────────────────────────────────────────── def ok(condition, msg=""): if not condition: raise AssertionError(msg or "Expected True") 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) if not os.path.exists(path): skip(f"{filename} not found — run generators/generate_data.py first") with open(path, newline="") as f: return list(csv.DictReader(f)) # ── Reference data ──────────────────────────────────────────────────────────── # Seeded once via the API; IDs stored at runtime. DEPT_IDS: dict[str, int] = {} # name → id GL_IDS: dict[str, int] = {} # code → id 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"}, {"code": "MKT", "name": "Marketing", "cost_center": "CC-400"}, {"code": "OPS", "name": "Operations", "cost_center": "CC-500"}, {"code": "FIN", "name": "Finance", "cost_center": "CC-600"}, ] 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 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 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] def gl_code(category: str) -> str: code = CSV_TO_GL.get(category) assert code, f"No GL mapping for '{category}'" return code def fiscal(period: str): y, m = period.split("-") return int(y), int(m) 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" 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"]) # ── suite_revenue ───────────────────────────────────────────────────────────── def suite_revenue() -> Suite: s = Suite("Revenue — CSV + API round-trip") if not api_up(): s.run("API reachable", lambda: skip(f"Go API not running at {API_BASE}")) return s rows = [] # 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 ]) # 2. Seed FK parents s.run("seed departments + GL accounts", seed) # 3. POST budgets budget_ids: dict[str, int] = {} 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"]) 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(): 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 actuals — Product + Service 2023-01", post_actuals) # 5. GET variance — verify VarianceReport shape and values report: dict = {} 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 /variance returns VarianceReport", fetch_report) 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}'") 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']}") 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") line = next((l for l in report["lines"] 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"]): eq(line["status"], "favourable", "revenue over-budget should be favourable") s.run("Product line values match CSV", verify_product_line) 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(): 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", []) 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 /variance/alerts shape valid", check_alerts) # 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 = float(csv_row["budget_amount"]) * 1.10 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") == "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 budget forecast_1 — variance reflects revision", update_forecast) return s # ── CSV-only suites ─────────────────────────────────────────────────────────── def suite_opex() -> Suite: s = Suite("Opex CSV") rows = [] 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: 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 = [] 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 = [] 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 = [] 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 ────────────────────────────────────────────────────────────────── def run_suites(suites): total = passed = failed = skipped = 0 for suite in suites: print(f"\n\033[1m{suite.name}\033[0m") for r in suite.results: total += 1 if r.skipped: skipped += 1 print(f" \033[33m⊘\033[0m {r.name}") print(f" {r.message}") elif r.passed: passed += 1 print(f" \033[32m✓\033[0m {r.name}") else: failed += 1 print(f" \033[31m✗\033[0m {r.name}") for line in r.message.strip().splitlines(): print(f" {line}") print(f"\n{'─'*48}") color = "\033[32m" if failed == 0 else "\033[31m" 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 # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": import argparse 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 ok = run_suites([ suite_revenue(), suite_opex(), suite_pl(), suite_cashflow(), suite_headcount(), ]) sys.exit(0 if ok else 1)