test_admin_analytics.py

"""Tests for the comprehensive admin analytics layer (informal-market science +
platform aggregation). Pure math is tested directly; _collect_platform_stats is
tested against a fake Firestore. Run: python test_admin_analytics.py"""
import os, sys, math
os.environ.setdefault("FIREBASE", "{}")
from unittest import mock

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import analytics

PASS = FAIL = 0
def check(desc, got, want):
    global PASS, FAIL
    ok = got == want
    PASS += ok; FAIL += (not ok)
    print(f"PASS  {desc}" if ok else f"FAIL  {desc}\n      got={got!r} want={want!r}")
def approx(desc, got, want, tol=1e-6):
    global PASS, FAIL
    ok = got is not None and abs(got - want) <= tol
    PASS += ok; FAIL += (not ok)
    print(f"PASS  {desc}" if ok else f"FAIL  {desc}\n      got={got!r} want≈{want!r}")
def check_true(desc, got):  check(desc, bool(got), True)

print("== Gini coefficient ==")
check(" empty → None", analytics.gini_coefficient([]), None)
check(" all zero → None", analytics.gini_coefficient([0, 0, 0]), None)
check(" perfect equality → 0", analytics.gini_coefficient([5, 5, 5, 5]), 0.0)
# One trader has everything → approaches (n-1)/n
approx(" total concentration", analytics.gini_coefficient([0, 0, 0, 100]), 0.75, tol=0.001)
g = analytics.gini_coefficient([1, 2, 3, 4, 5])
check_true(" mild inequality in (0,1)", 0 < g < 1)

print("== HHI concentration ==")
check(" monopoly", analytics.hhi_concentration({"a": 100}), 1.0)
approx(" 4 equal → 0.25", analytics.hhi_concentration({"a": 1, "b": 1, "c": 1, "d": 1}), 0.25, tol=1e-9)
check(" empty → None", analytics.hhi_concentration({}), None)

print("== top_share / median ==")
check(" top-10% of 10 equal earners = 10%", analytics.top_share([10]*10, 0.10), 10.0)
approx(" top earner dominates", analytics.top_share([1, 1, 1, 97], 0.25), 97.0, tol=0.5)
check(" median odd", analytics.median([3, 1, 2]), 2.0)
check(" median even", analytics.median([1, 2, 3, 4]), 2.5)

print("== price dispersion (law of one price) ==")
disp = analytics.price_dispersion({
    "tomato": [1.00, 1.00, 1.00],       # no dispersion
    "dovi":   [2.0, 4.0, 6.0],          # high dispersion
    "rare":   [5.0],                    # only 1 trader → excluded
})
check(" items compared (>=2 traders)", disp["itemsCompared"], 2)
check(" most dispersed first", disp["mostDispersed"][0]["item"], "dovi")
check(" tomato zero CV", [d for d in disp["mostDispersed"] if d["item"] == "tomato"][0]["cv"], 0.0)
check(" dovi spread%", disp["mostDispersed"][0]["spreadPct"], 100.0)

print("== scan_transactions ==")
txns = [
    {"transaction_type": "sale", "created_at": "2026-07-01T10:00:00Z",
     "details": {"customer_credit": 5.0}},
    {"transaction_type": "sale", "created_at": "2026-07-02T10:00:00Z", "details": {}},
    {"transaction_type": "stock_in", "created_at": "2026-07-02T09:00:00Z", "details": {}},
    {"transaction_type": "expense", "created_at": "2026-07-03T09:00:00Z", "details": {}},
]
scan = analytics.scan_transactions(txns)
check(" type mix sale", scan["byType"]["sale"], 2)
check(" type mix stock_in", scan["byType"]["stock_in"], 1)
check(" total", scan["total"], 4)
check(" sale count", scan["saleCount"], 2)
check(" credit sale count", scan["creditSaleCount"], 1)
check(" credit value", scan["creditSalesValue"], 5.0)
check(" last activity", scan["lastActivity"][:10], "2026-07-03")
check(" daily buckets", scan["daily"]["2026-07-02"], 2)

print("== model_health accuracy-by-modality ==")
ex = [
    {"modality": "audio", "verdict": "confirmed", "task": "intent_parse"},
    {"modality": "audio", "verdict": "rejected", "task": "intent_parse"},
    {"modality": "text",  "verdict": "confirmed", "task": "intent_parse"},
    {"modality": "text",  "verdict": "confirmed", "task": "intent_parse"},
]
mh = analytics.model_health(ex)
check(" audio 50%", mh["accuracyByModality"]["audio"], 50.0)
check(" text 100%", mh["accuracyByModality"]["text"], 100.0)
check(" overall confirmRate", mh["confirmRate"], 75.0)

print("== report prompt + fallback ==")
prompt = analytics.build_admin_report_prompt({"platform": {"totalUsers": 3}}, "market", "2026-07-04T00:00:00Z")
check_true(" market focus present", "law of one price" in prompt)
check_true(" stats embedded", "totalUsers" in prompt)
fb = analytics.fallback_admin_report("full", "2026-07-04T00:00:00Z")
check(" fallback flags aiError", fb["aiError"], True)
check(" fallback keeps type", fb["reportType"], "full")


# ── _collect_platform_stats against a fake Firestore ─────────────────────────
print("== _collect_platform_stats (fake Firestore) ==")

class _Doc:
    def __init__(self, data, _id="x"): self._d = data; self.id = _id
    @property
    def exists(self): return self._d is not None
    def to_dict(self): return dict(self._d) if self._d else None

class _Col:
    def __init__(self, docs=None, kv=None): self._docs = docs or []; self._kv = kv or {}
    def stream(self):
        if self._kv:
            return [_Doc(v, k) for k, v in self._kv.items()]
        return [_Doc(d, d.get("_id", "d")) for d in self._docs]
    def where(self, *a, **k): return self
    def limit(self, n): return self
    def order_by(self, *a, **k): return self

class _UserDocRef:
    def __init__(self, subs): self.subs = subs
    def collection(self, name): return _Col(kv={} if name not in self.subs else None,
                                            docs=self.subs.get(name, []))

class _DB:
    def __init__(self, users, ledgers, examples):
        self.users = users; self.ledgers = ledgers; self.examples = examples
    def collection(self, name):
        if name == "users":
            outer = self
            class _UsersCol(_Col):
                def __init__(s): super().__init__(docs=outer.users)
                def document(s, pid):
                    return _UserDocRef(outer.ledgers.get(pid, {}))
            return _UsersCol()
        if name == "organizations":
            return _Col(docs=[{"_id": "o1"}])
        if name == "distillation_examples":
            return _Col(docs=self.examples)
        if name == "admin_reports":
            return _Col(docs=[])
        return _Col()

users = [
    {"_id": "u1", "email": "a@x.com", "displayName": "Rutendo", "phoneStatus": "approved",
     "phone": "+263771000001", "defaultCurrency": "USD", "createdAt": "2026-06-01T00:00:00Z"},
    {"_id": "u2", "email": "b@x.com", "displayName": "Chipo", "phoneStatus": "approved",
     "phone": "+263771000002", "defaultCurrency": "USD", "createdAt": "2026-06-02T00:00:00Z"},
    {"_id": "u3", "email": "c@x.com", "displayName": "Admin", "isAdmin": True,
     "phoneStatus": "pending", "createdAt": "2026-06-03T00:00:00Z"},
]
from datetime import datetime, timezone, timedelta
recent = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
ledgers = {
    "263771000001": {
        "transactions": [
            {"transaction_type": "sale", "created_at": recent,
             "details": {"currency": "USD", "amount": 100.0,
                         "items": [{"item": "dovi", "quantity": 25, "price_per_unit": 4.0}]}},
            {"transaction_type": "sale", "created_at": recent,
             "details": {"currency": "USD", "amount": 20.0, "customer_credit": 20.0,
                         "items": [{"item": "tomato", "quantity": 20, "price_per_unit": 1.0}]}},
        ],
        "stock_batches": [{"name": "dovi", "quantity_remaining": 100, "cost_each": 2.0,
                           "price_each": 4.0, "stocked_at": "2026-06-01"}],
        "customers": [{"name": "Tariro", "receivable": 20.0}],
        "item_prices": [{"name": "dovi", "price": 4.0, "on_sale": False},
                        {"name": "tomato", "price": 1.0, "on_sale": False}],
        "services": [{"_id": "s1", "name": "braiding"}],
    },
    "263771000002": {
        "transactions": [
            {"transaction_type": "sale", "created_at": recent,
             "details": {"currency": "USD", "amount": 10.0,
                         "items": [{"item": "dovi", "quantity": 5, "price_per_unit": 2.0}]}},
        ],
        "stock_batches": [{"name": "dovi", "quantity_remaining": 10, "cost_each": 1.5,
                           "price_each": 2.0, "stocked_at": "2026-06-05"}],
        "customers": [],
        "item_prices": [{"name": "dovi", "price": 2.0, "on_sale": False}],
        "services": [],
    },
}
examples = [{"modality": "audio", "verdict": "confirmed", "task": "asr"},
            {"modality": "text", "verdict": "confirmed", "task": "intent_parse"}]

fake_db = _DB(users, ledgers, examples)
with mock.patch.object(analytics, "money_opt", analytics.money_opt):  # no-op, keep ref
    import importlib
    with mock.patch.dict(os.environ, {"FIREBASE": "{}"}):
        # import main with firebase/genai mocked
        with mock.patch("firebase_admin.credentials.Certificate", return_value=mock.MagicMock()), \
             mock.patch("firebase_admin.initialize_app", return_value=mock.MagicMock()), \
             mock.patch("firebase_admin.firestore.client", return_value=fake_db):
            import main
            main.db = fake_db  # ensure the module global points at the fake

stats = main._collect_platform_stats()
print("  platform:", stats["platform"])
print("  economy.byCurrency:", stats["economy"]["byCurrency"])
print("  creditEconomy:", stats["informalMarketScience"]["creditEconomy"])
print("  inequality:", stats["informalMarketScience"]["traderInequality"])
print("  priceDiscovery:", stats["informalMarketScience"]["priceDiscovery"])
check(" approved traders", stats["platform"]["approvedTraders"], 2)
check(" pending", stats["platform"]["pendingApproval"], 1)
check(" admins", stats["platform"]["admins"], 1)
check(" active 7d (recent txns)", stats["platform"]["activeTraders7d"], 2)
check(" USD sales summed", stats["economy"]["byCurrency"]["USD"]["sales"], 130.0)
check(" txn mix sale count", stats["economy"]["transactionMix"]["sale"], 3)
check(" receivables", stats["informalMarketScience"]["creditEconomy"]["receivables"], 20.0)
check_true(" credit-sale share computed",
           stats["informalMarketScience"]["creditEconomy"]["creditSaleShareOfSalesTxns"] is not None)
check_true(" gini computed", stats["informalMarketScience"]["traderInequality"]["giniRevenue"] is not None)
# dovi priced 4.0 (u1) vs 2.0 (u2) → dispersion tracked
disp_items = [d["item"] for d in stats["informalMarketScience"]["priceDiscovery"]["mostDispersed"]]
check(" dovi price dispersion tracked", "dovi" in disp_items, True)
check(" services counted", stats["informalMarketScience"]["serviceEconomy"]["registeredServices"], 1)
check(" model health wired", stats["modelHealth"]["totalCaptured"], 2)
check(" stock retail > cost", stats["economy"]["stockValueRetail"] > stats["economy"]["stockValueCost"], True)

print(f"\n{'='*40}\nTOTAL: {PASS} passed, {FAIL} failed")
sys.exit(1 if FAIL else 0)