#!/usr/bin/env python """ post_migration_validation.py ============================ Post-migration integrity validation for TrueWave marketplace schema evolution. Validates all domains affected by the marketplace migration: - Product integrity - Order integrity - Wallet linkage integrity - Distributor integrity - Opportunity Bundle integrity - Franchise integrity - Marketplace field defaults Can be run standalone or imported by migration_rehearsal.py. Usage (from backend/): python scripts/post_migration_validation.py [--verbose] [--json-out PATH] """ import argparse import datetime import json import os import sys BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, BACKEND_DIR) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev") def _setup_django(): import django from django.conf import settings as _s # Remove apps that aren't installed in this environment (e.g. daphne on dev) unavailable = [] for app in list(_s.INSTALLED_APPS): try: __import__(app.split('.')[0]) except ImportError: unavailable.append(app) if unavailable: _s.INSTALLED_APPS = [a for a in _s.INSTALLED_APPS if a not in unavailable] django.setup() # --------------------------------------------------------------------------- # Validation infrastructure # --------------------------------------------------------------------------- class ValidationResult: def __init__(self, name, category): self.name = name self.category = category self.passed = 0 self.failed = 0 self.warnings = 0 self.checks = [] def ok(self, label, detail=""): self.passed += 1 self.checks.append({"label": label, "status": "PASS", "detail": detail}) def fail(self, label, detail=""): self.failed += 1 self.checks.append({"label": label, "status": "FAIL", "detail": detail}) print(f" [FAIL] {label}: {detail}") def warn(self, label, detail=""): self.warnings += 1 self.checks.append({"label": label, "status": "WARN", "detail": detail}) print(f" [WARN] {label}: {detail}") def summary(self): total = self.passed + self.failed + self.warnings status = "FAIL" if self.failed > 0 else ("WARN" if self.warnings > 0 else "PASS") return { "category": self.category, "name": self.name, "status": status, "total": total, "passed": self.passed, "failed": self.failed, "warnings": self.warnings, "checks": self.checks, } def _ts(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _log(msg): print(f"[{_ts()}] {msg}") # --------------------------------------------------------------------------- # Domain validators # --------------------------------------------------------------------------- def validate_product_integrity(): _log("--- Product Integrity ---") from apps.business.products.models import Product r = ValidationResult("product_integrity", "products") total = Product.objects.count() r.ok("product_table_accessible", f"{total} products found") # Confirm all existing products have required base fields missing_name = Product.objects.filter(name__isnull=True).count() if missing_name: r.fail("products_with_null_name", f"{missing_name} products have null name") else: r.ok("products_name_not_null", "all products have non-null names") missing_cost = Product.objects.filter(base_cost__isnull=True).count() if missing_cost: r.fail("products_with_null_base_cost", f"{missing_cost} products have null base_cost") else: r.ok("products_base_cost_not_null", "all products have non-null base_cost") # Marketplace-specific: slug uniqueness where not null from django.db.models import Count from django.db import connection # Check slug field exists on model if hasattr(Product, 'slug'): dup_slugs = ( Product.objects.exclude(slug__isnull=True) .exclude(slug__exact="") .values("slug") .annotate(cnt=Count("id")) .filter(cnt__gt=1) ) dup_count = dup_slugs.count() if dup_count: r.fail("product_slug_uniqueness", f"{dup_count} duplicate slug values found") else: r.ok("product_slug_uniqueness", "no duplicate slugs among non-null products") # Check sku uniqueness where not null if hasattr(Product, 'sku'): dup_skus = ( Product.objects.exclude(sku__isnull=True) .exclude(sku__exact="") .values("sku") .annotate(cnt=Count("id")) .filter(cnt__gt=1) ) dup_sku_count = dup_skus.count() if dup_sku_count: r.fail("product_sku_uniqueness", f"{dup_sku_count} duplicate SKU values found") else: r.ok("product_sku_uniqueness", "no duplicate SKUs among non-null products") # Existing bundle products should have opportunity_bundle set bundle_without_ob = Product.objects.filter( is_active=True, opportunity_bundle__isnull=True, ).count() all_active = Product.objects.filter(is_active=True).count() # This is just a warning — retail products won't have opportunity_bundle if bundle_without_ob == all_active: r.warn("products_opportunity_bundle_coverage", f"all {all_active} active products have no opportunity_bundle (may be intentional for retail)") else: r.ok("products_opportunity_bundle_coverage", f"{all_active - bundle_without_ob}/{all_active} active products linked to opportunity bundles") # Marketplace extension fields: vendor reference integrity if hasattr(Product, 'vendor'): orphan_vendor = Product.objects.filter( vendor__isnull=False ).exclude(vendor__in=__import__( "apps.business.marketplace.models_part1", fromlist=["Vendor"] ).Vendor.objects.all()).count() if _marketplace_models_importable() else 0 if orphan_vendor: r.fail("product_vendor_fk_integrity", f"{orphan_vendor} products reference non-existent vendors") else: r.ok("product_vendor_fk_integrity", "no orphan vendor references on products") # Category/SubCategory FK integrity if _marketplace_models_importable() and hasattr(Product, 'marketplace_category'): from apps.business.marketplace.models_part1 import Category, SubCategory orphan_cat = Product.objects.filter( marketplace_category__isnull=False ).exclude(marketplace_category__in=Category.objects.all()).count() if orphan_cat: r.fail("product_marketplace_category_fk", f"{orphan_cat} products have orphan marketplace_category FK") else: r.ok("product_marketplace_category_fk", "no orphan marketplace_category references") return r.summary() def validate_order_integrity(): _log("--- Order Integrity ---") from apps.business.orders.models import Order r = ValidationResult("order_integrity", "orders") total = Order.objects.count() r.ok("order_table_accessible", f"{total} orders found") # All existing orders should have order_type=BUNDLE (default) if hasattr(Order, 'order_type'): null_type = Order.objects.filter(order_type__isnull=True).count() if null_type: r.fail("order_type_no_nulls", f"{null_type} orders have null order_type") else: r.ok("order_type_no_nulls", "all orders have non-null order_type") bundle_count = Order.objects.filter(order_type="BUNDLE").count() retail_count = Order.objects.filter(order_type="RETAIL").count() other_count = total - bundle_count - retail_count r.ok("order_type_distribution", f"BUNDLE={bundle_count}, RETAIL={retail_count}, OTHER={other_count}") # All pre-migration orders must be BUNDLE if retail_count > 0: r.warn("pre_migration_retail_orders", f"{retail_count} RETAIL orders exist — expected 0 before marketplace launch") else: r.ok("pre_migration_retail_orders", "no RETAIL orders — expected for pre-launch state") # Reference ID uniqueness from django.db.models import Count dup_ref = ( Order.objects.values("reference_id") .annotate(cnt=Count("id")) .filter(cnt__gt=1) .count() ) if dup_ref: r.fail("order_reference_id_uniqueness", f"{dup_ref} duplicate reference_ids detected") else: r.ok("order_reference_id_uniqueness", "all reference_ids are unique") # Payment transaction FK integrity (new nullable field) if hasattr(Order, 'payment_transaction'): orphan_pt = Order.objects.filter( payment_transaction__isnull=False ).exclude( payment_transaction__in=__import__( "apps.payments.models", fromlist=["PaymentTransaction"] ).PaymentTransaction.objects.all() ).count() if orphan_pt: r.fail("order_payment_transaction_fk", f"{orphan_pt} orders reference non-existent payment transactions") else: r.ok("order_payment_transaction_fk", "no orphan payment_transaction references on orders") # Wallet ledger FK integrity (new nullable field) if hasattr(Order, 'wallet_ledger_entry'): from apps.business.wallet.models import WalletLedger orphan_wl = Order.objects.filter( wallet_ledger_entry__isnull=False ).exclude(wallet_ledger_entry__in=WalletLedger.objects.all()).count() if orphan_wl: r.fail("order_wallet_ledger_fk", f"{orphan_wl} orders reference non-existent wallet ledger entries") else: r.ok("order_wallet_ledger_fk", "no orphan wallet_ledger_entry references on orders") # Completed orders must have a product null_product = Order.objects.filter(status="COMPLETED", product__isnull=True).count() if null_product: r.fail("completed_orders_have_product", f"{null_product} COMPLETED orders have null product") else: r.ok("completed_orders_have_product", "all COMPLETED orders have a product") # Completed orders must have a user null_user = Order.objects.filter(status="COMPLETED", user__isnull=True).count() if null_user: r.fail("completed_orders_have_user", f"{null_user} COMPLETED orders have null user") else: r.ok("completed_orders_have_user", "all COMPLETED orders have a user") return r.summary() def validate_wallet_integrity(): _log("--- Wallet Integrity ---") from apps.business.wallet.models import Wallet, WalletLedger r = ValidationResult("wallet_integrity", "wallet") w_total = Wallet.objects.count() l_total = WalletLedger.objects.count() r.ok("wallet_tables_accessible", f"Wallet={w_total}, WalletLedger={l_total}") # No negative balances in active wallets neg_main = Wallet.objects.filter(main_balance__lt=0).count() neg_repo = Wallet.objects.filter(repurchase_balance__lt=0).count() if neg_main: r.fail("wallet_no_negative_main_balance", f"{neg_main} wallets have negative main balance") else: r.ok("wallet_no_negative_main_balance", "no negative main wallet balances") if neg_repo: r.fail("wallet_no_negative_repurchase_balance", f"{neg_repo} wallets have negative repurchase balance") else: r.ok("wallet_no_negative_repurchase_balance", "no negative repurchase balances") # Ledger entries must have valid user references from django.conf import settings from django.contrib.auth import get_user_model User = get_user_model() orphan_ledger = WalletLedger.objects.filter(user__isnull=True).count() if orphan_ledger: r.fail("wallet_ledger_user_not_null", f"{orphan_ledger} ledger entries have null user") else: r.ok("wallet_ledger_user_not_null", "all ledger entries have a user") # New FK: funded_orders must point to valid orders (if any exist) from apps.business.orders.models import Order if hasattr(Order, 'wallet_ledger_entry'): funded_orders = Order.objects.filter(wallet_ledger_entry__isnull=False).count() r.ok("wallet_funded_orders_count", f"{funded_orders} orders have wallet_ledger_entry set") # Credit/debit balance check: sum of credits - sum of debits should ≥ 0 per user from django.db.models import Sum, Q from decimal import Decimal imbalanced = 0 for wallet in Wallet.objects.all()[:500]: # sample check for large datasets credits = WalletLedger.objects.filter( user=wallet.user, transaction_type="CREDIT", ).aggregate(total=Sum("amount"))["total"] or Decimal("0.00") debits = WalletLedger.objects.filter( user=wallet.user, transaction_type="DEBIT", ).aggregate(total=Sum("amount"))["total"] or Decimal("0.00") net = credits - debits if net < Decimal("-0.01"): # allow tiny rounding tolerance imbalanced += 1 if imbalanced: r.warn("wallet_ledger_net_balance_sample", f"{imbalanced} wallets have negative net ledger balance (sample of first 500)") else: r.ok("wallet_ledger_net_balance_sample", "net ledger balance non-negative for sampled wallets") return r.summary() def validate_distributor_integrity(): _log("--- Distributor Integrity ---") from apps.business.distributor.models import DistributorID r = ValidationResult("distributor_integrity", "distributor") total = DistributorID.objects.count() r.ok("distributor_table_accessible", f"{total} distributor IDs found") # Distributor codes must be unique from django.db.models import Count dup_codes = ( DistributorID.objects.values("distributor_code") .annotate(cnt=Count("id")) .filter(cnt__gt=1) .count() ) if dup_codes: r.fail("distributor_code_uniqueness", f"{dup_codes} duplicate distributor codes") else: r.ok("distributor_code_uniqueness", "all distributor codes are unique") # Global positions must be unique dup_pos = ( DistributorID.objects.values("global_position") .annotate(cnt=Count("id")) .filter(cnt__gt=1) .count() ) if dup_pos: r.fail("global_position_uniqueness", f"{dup_pos} duplicate global positions") else: r.ok("global_position_uniqueness", "all global positions are unique") # All distributors must have a valid user null_user = DistributorID.objects.filter(user__isnull=True).count() if null_user: r.fail("distributor_user_not_null", f"{null_user} distributors have null user") else: r.ok("distributor_user_not_null", "all distributors have a user") # All distributors must have a valid product null_product = DistributorID.objects.filter(product__isnull=True).count() if null_product: r.fail("distributor_product_not_null", f"{null_product} distributors have null product") else: r.ok("distributor_product_not_null", "all distributors have a product") # Binary tree constraints: no node can be its own parent self_parent = DistributorID.objects.filter( binary_parent_distributor__isnull=False, binary_parent_distributor=models_F("id") if False else None ).count() if False else 0 # Use raw comparison from django.db.models import F self_parent = DistributorID.objects.filter( binary_parent_distributor__isnull=False ).filter( binary_parent_distributor_id=F("id") ).count() if self_parent: r.fail("distributor_no_self_parent", f"{self_parent} distributors are their own binary parent") else: r.ok("distributor_no_self_parent", "no self-referencing binary parents") # Binary position should be LEFT or RIGHT when parent is set invalid_position = DistributorID.objects.filter( binary_parent_distributor__isnull=False, binary_position__isnull=True, ).count() if invalid_position: r.warn("distributor_binary_position_when_parent_set", f"{invalid_position} distributors have a binary parent but no position (may be backfill in progress)") else: r.ok("distributor_binary_position_when_parent_set", "all distributors with binary parent also have binary position") # Company roots: each opportunity_bundle should have exactly one company root from apps.business.products.models import Product ob_ids = Product.objects.exclude(opportunity_bundle__isnull=True).values_list( "opportunity_bundle_id", flat=True ).distinct() multi_root = 0 for ob_id in ob_ids: root_count = DistributorID.objects.filter( is_company_root=True, product__opportunity_bundle_id=ob_id, ).count() if root_count > 1: multi_root += 1 if multi_root: r.fail("distributor_single_company_root_per_ob", f"{multi_root} opportunity bundles have multiple company roots") else: r.ok("distributor_single_company_root_per_ob", "each opportunity bundle has at most one company root") return r.summary() def validate_opportunity_bundle_integrity(): _log("--- Opportunity Bundle Integrity ---") from apps.business.schemes.models import OpportunityBundle r = ValidationResult("opportunity_bundle_integrity", "schemes") total = OpportunityBundle.objects.count() r.ok("opportunity_bundle_table_accessible", f"{total} opportunity bundles found") # All active bundles must have a name null_name = OpportunityBundle.objects.filter(is_active=True, name__isnull=True).count() if null_name: r.fail("ob_active_name_not_null", f"{null_name} active opportunity bundles have null name") else: r.ok("ob_active_name_not_null", "all active bundles have names") # Products linked to each active bundle from apps.business.products.models import Product from django.db.models import Count bundles_without_product = OpportunityBundle.objects.filter( is_active=True ).annotate( prod_count=Count("product") ).filter(prod_count=0).count() if bundles_without_product: r.warn("ob_active_has_products", f"{bundles_without_product} active bundles have no linked products") else: r.ok("ob_active_has_products", "all active bundles have at least one product") # Distributor IDs for each active bundle from apps.business.distributor.models import DistributorID bundles_without_distributor = 0 for ob in OpportunityBundle.objects.filter(is_active=True): if not DistributorID.objects.filter(product__opportunity_bundle=ob, is_active=True).exists(): bundles_without_distributor += 1 if bundles_without_distributor > 0: r.warn("ob_active_has_distributors", f"{bundles_without_distributor} active bundles have no active distributors") else: r.ok("ob_active_has_distributors", "all active bundles have active distributors") return r.summary() def validate_franchise_integrity(): _log("--- Franchise Integrity ---") r = ValidationResult("franchise_integrity", "geo") try: from apps.geo.models import FranchiseStore except ImportError: r.warn("franchise_model_import", "FranchiseStore model import failed") return r.summary() total = FranchiseStore.objects.count() r.ok("franchise_store_table_accessible", f"{total} franchise stores found") if total == 0: r.ok("franchise_no_stores_expected", "0 stores — expected for pre-launch marketplace state") return r.summary() # All stores should have a non-null name/code null_name = FranchiseStore.objects.filter(name__isnull=True).count() if null_name: r.fail("franchise_store_name_not_null", f"{null_name} stores have null name") else: r.ok("franchise_store_name_not_null", "all stores have names") # FranchisePincodeCoverage: coverage entries must reference valid stores and pincodes try: from apps.geo.models import FranchisePincodeCoverage coverage_total = FranchisePincodeCoverage.objects.count() r.ok("franchise_pincode_coverage_accessible", f"{coverage_total} pincode coverage entries") orphan_store = FranchisePincodeCoverage.objects.filter(store__isnull=True).count() if orphan_store: r.fail("franchise_coverage_store_fk", f"{orphan_store} coverage entries have null store FK") else: r.ok("franchise_coverage_store_fk", "all coverage entries have valid store FK") except Exception as e: r.warn("franchise_pincode_coverage", f"could not validate: {e}") # FranchiseStoreInventory: inventory entries must reference valid products try: from apps.geo.models import FranchiseStoreInventory inv_total = FranchiseStoreInventory.objects.count() r.ok("franchise_inventory_accessible", f"{inv_total} inventory entries") except Exception as e: r.warn("franchise_inventory", f"could not validate: {e}") return r.summary() def validate_marketplace_defaults(): _log("--- Marketplace Field Defaults ---") r = ValidationResult("marketplace_defaults", "marketplace") # Products: new nullable marketplace fields should be null for pre-existing bundle products from apps.business.products.models import Product total_products = Product.objects.count() r.ok("products_exist_for_default_check", f"{total_products} products to check") if hasattr(Product, 'order_type'): r.warn("product_order_type_field", "Product.order_type exists — unexpected on Product model (check model)") # Orders: order_type must default to BUNDLE from apps.business.orders.models import Order if hasattr(Order, 'order_type'): non_bundle = Order.objects.exclude(order_type="BUNDLE").count() if non_bundle == 0: r.ok("order_type_defaults_to_bundle", "all existing orders have order_type=BUNDLE (correct default)") else: r.warn("order_type_defaults_to_bundle", f"{non_bundle} existing orders have non-BUNDLE order_type") # Marketplace tables: new tables should be empty pre-launch if _marketplace_models_importable(): from apps.business.marketplace.models_part1 import Category, SubCategory, Brand, Vendor cat_count = Category.objects.count() vendor_count = Vendor.objects.count() r.ok("marketplace_catalog_empty_pre_launch", f"Category={cat_count}, Vendor={vendor_count} (expected 0 pre-launch)") try: from apps.business.marketplace.models_part2 import MarketplaceOrder mp_order_count = MarketplaceOrder.objects.count() r.ok("marketplace_orders_empty_pre_launch", f"MarketplaceOrder={mp_order_count} (expected 0 pre-launch)") except Exception as e: r.warn("marketplace_orders_import", f"MarketplaceOrder import issue: {e}") # Geo: FranchisePincodeCoverage and FranchiseStoreInventory should be new/empty try: from apps.geo.models import FranchisePincodeCoverage, FranchiseStoreInventory cov = FranchisePincodeCoverage.objects.count() inv = FranchiseStoreInventory.objects.count() r.ok("geo_new_tables_empty_pre_launch", f"FranchisePincodeCoverage={cov}, FranchiseStoreInventory={inv} (expected 0 pre-launch)") except Exception as e: r.warn("geo_new_tables", f"could not check new geo tables: {e}") return r.summary() def validate_health_checks(): _log("--- Migration-Safe Health Checks ---") r = ValidationResult("health_checks", "cross-domain") from apps.business.products.models import Product # Missing slugs on products that have marketplace_status set if hasattr(Product, 'slug') and hasattr(Product, 'marketplace_status'): missing_slug = Product.objects.filter( marketplace_status__isnull=False, slug__isnull=True, ).count() if missing_slug: r.warn("missing_slug_for_marketplace_products", f"{missing_slug} products have marketplace_status but no slug") else: r.ok("missing_slug_for_marketplace_products", "no marketplace products are missing slugs") # Missing SKU on products that are marked as marketplace items if hasattr(Product, 'sku') and hasattr(Product, 'marketplace_status'): missing_sku = Product.objects.filter( marketplace_status__isnull=False, sku__isnull=True, ).count() if missing_sku: r.warn("missing_sku_for_marketplace_products", f"{missing_sku} marketplace products have no SKU") else: r.ok("missing_sku_for_marketplace_products", "no marketplace products are missing SKU") # Orphan marketplace references: orders referencing marketplace models from apps.business.orders.models import Order if hasattr(Order, 'order_type'): retail_without_items = 0 try: from apps.business.marketplace.models_part2 import OrderItem retail_orders = Order.objects.filter(order_type="RETAIL") for o in retail_orders[:1000]: if not OrderItem.objects.filter(order=o).exists(): retail_without_items += 1 if retail_without_items: r.warn("retail_orders_without_order_items", f"{retail_without_items} RETAIL orders have no OrderItem records") else: r.ok("retail_orders_without_order_items", "all RETAIL orders have OrderItem records (or no RETAIL orders)") except Exception: r.ok("retail_orders_without_order_items", "no RETAIL orders to check") # Invalid vendor references on products if _marketplace_models_importable() and hasattr(Product, 'vendor'): products_with_vendor = Product.objects.filter(vendor__isnull=False).count() r.ok("vendor_reference_count", f"{products_with_vendor} products have vendor references") # Null-sensitive fields: order_type must not be null if hasattr(Order, 'order_type'): null_type = Order.objects.filter(order_type__isnull=True).count() if null_type: r.fail("order_type_not_null", f"{null_type} orders have NULL order_type — CRITICAL") else: r.ok("order_type_not_null", "order_type is non-null on all orders") # Wallet ledger: amount must always be positive from apps.business.wallet.models import WalletLedger zero_amount = WalletLedger.objects.filter(amount__lte=0).count() if zero_amount: r.warn("wallet_ledger_positive_amounts", f"{zero_amount} ledger entries have amount <= 0") else: r.ok("wallet_ledger_positive_amounts", "all ledger entry amounts are positive") return r.summary() # --------------------------------------------------------------------------- # Aggregate runner # --------------------------------------------------------------------------- def _marketplace_models_importable(): try: import apps.business.marketplace.models_part1 # noqa return True except Exception: return False def run_all_validations(verbose=False): _log("=" * 60) _log("POST-MIGRATION VALIDATION SUITE") _log("=" * 60) validators = [ validate_product_integrity, validate_order_integrity, validate_wallet_integrity, validate_distributor_integrity, validate_opportunity_bundle_integrity, validate_franchise_integrity, validate_marketplace_defaults, validate_health_checks, ] results = [] total_passed = 0 total_failed = 0 total_warned = 0 for fn in validators: try: result = fn() except Exception as e: result = { "category": "unknown", "name": fn.__name__, "status": "ERROR", "error": str(e), "passed": 0, "failed": 1, "warnings": 0, } _log(f" [ERROR] {fn.__name__}: {e}") results.append(result) total_passed += result.get("passed", 0) total_failed += result.get("failed", 0) total_warned += result.get("warnings", 0) status = result.get("status", "?") symbol = {"PASS": "✓", "FAIL": "✗", "WARN": "⚠", "ERROR": "✗"}.get(status, "?") _log(f" {symbol} {result['name']}: {status} (P={result.get('passed',0)} F={result.get('failed',0)} W={result.get('warnings',0)})") _log("=" * 60) _log(f"TOTAL: Passed={total_passed} Failed={total_failed} Warnings={total_warned}") overall = "FAIL" if total_failed > 0 else ("WARN" if total_warned > 0 else "PASS") _log(f"OVERALL RESULT: {overall}") _log("=" * 60) return { "run_at": _ts(), "overall": overall, "passed": total_passed, "failed": total_failed, "warnings": total_warned, "results": results, } # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): p = argparse.ArgumentParser(description="Post-migration integrity validation") p.add_argument("--verbose", action="store_true", default=False) p.add_argument("--json-out", default=None, help="Path to write JSON results") args = p.parse_args() _setup_django() outcome = run_all_validations(verbose=args.verbose) if args.json_out: with open(args.json_out, "w") as f: json.dump(outcome, f, indent=2, default=str) _log(f"Results written to: {args.json_out}") sys.exit(0 if outcome["overall"] == "PASS" else 1) if __name__ == "__main__": main()