#!/usr/bin/env python """ smoke_tests.py ============== Dry-run smoke test suite for TrueWave platform. Tests that existing business flows still work correctly after the marketplace schema migration. All tests are READ-ONLY / simulation mode — they validate the ORM query layer, service logic, and model integrity without creating live transactions or sending real money. Covered scenarios: 1. Retail purchase path (dry-run only — no wallet debit) 2. Bundle purchase path (simulate order creation check) 3. Wallet checkout logic (balance check, not actual debit) 4. Distributor creation validation (code generation + placement check) 5. Binary placement constraints 6. Sponsor income generation path (ledger query check) 7. Wallet transaction logging path 8. Order type isolation (BUNDLE vs RETAIL filter correctness) 9. Opportunity bundle lookup 10. Franchise store coverage lookup Usage (from backend/): python scripts/smoke_tests.py [--verbose] [--json-out PATH] [--stop-on-first-fail] """ import argparse import datetime import json import os import sys import traceback 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() def _ts(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _log(msg): print(f"[{_ts()}] {msg}") # --------------------------------------------------------------------------- # Smoke test base # --------------------------------------------------------------------------- class SmokeResult: def __init__(self, name, category): self.name = name self.category = category self.checks = [] self.passed = 0 self.failed = 0 self.status = "PASS" def ok(self, label, detail=""): self.passed += 1 self.checks.append({"label": label, "status": "PASS", "detail": str(detail)}) def fail(self, label, detail="", exc=None): self.failed += 1 self.status = "FAIL" detail_str = str(detail) if exc: detail_str += f" | Exception: {exc}" self.checks.append({"label": label, "status": "FAIL", "detail": detail_str}) print(f" [FAIL] {label}: {detail_str}") def skip(self, label, reason=""): self.checks.append({"label": label, "status": "SKIP", "detail": reason}) def summary(self): return { "name": self.name, "category": self.category, "status": self.status, "passed": self.passed, "failed": self.failed, "checks": self.checks, } # --------------------------------------------------------------------------- # Smoke tests # --------------------------------------------------------------------------- def smoke_retail_purchase_path(): """Verify retail purchase model path is importable and queryable.""" r = SmokeResult("retail_purchase_path", "marketplace") try: from apps.business.orders.models import Order from apps.business.products.models import Product # Can we filter RETAIL orders without error? retail_qs = Order.objects.filter(order_type="RETAIL") count = retail_qs.count() r.ok("retail_order_queryset", f"RETAIL orders queryable — count={count}") # Verify order_type field accepts RETAIL value choices = dict(Order.ORDER_TYPE_CHOICES) if "RETAIL" in choices: r.ok("retail_order_type_choice_valid", f"RETAIL is a valid choice: '{choices['RETAIL']}'") else: r.fail("retail_order_type_choice_valid", "RETAIL not in ORDER_TYPE_CHOICES") # Products queryable for marketplace context active_products = Product.objects.filter(is_active=True).count() r.ok("retail_product_pool", f"{active_products} active products available for retail") except Exception as e: r.fail("retail_purchase_path_import", exc=e) return r.summary() def smoke_bundle_purchase_path(): """Verify bundle purchase path is unaffected by migration.""" r = SmokeResult("bundle_purchase_path", "orders") try: from apps.business.orders.models import Order from apps.business.products.models import Product from apps.business.distributor.models import DistributorID # Can we filter BUNDLE orders? bundle_qs = Order.objects.filter(order_type="BUNDLE") bundle_count = bundle_qs.count() r.ok("bundle_order_queryset", f"BUNDLE orders queryable — count={bundle_count}") # COMPLETED bundles must have a distributor completed_bundle_without_dist = Order.objects.filter( order_type="BUNDLE", status="COMPLETED", distributor__isnull=True, ).count() if completed_bundle_without_dist > 0: r.fail("completed_bundle_has_distributor", f"{completed_bundle_without_dist} COMPLETED BUNDLE orders have no distributor") else: r.ok("completed_bundle_has_distributor", "all COMPLETED BUNDLE orders have a distributor or none exist yet") # Products with opportunity_bundle are still accessible bundle_products = Product.objects.filter( opportunity_bundle__isnull=False, is_active=True ).count() r.ok("bundle_products_accessible", f"{bundle_products} active bundle products found") # Distributor table integrity dist_total = DistributorID.objects.count() r.ok("distributor_table_readable", f"{dist_total} distributor IDs in DB") except Exception as e: r.fail("bundle_purchase_path_error", exc=e) return r.summary() def smoke_wallet_checkout(): """Verify wallet checkout logic paths are intact.""" r = SmokeResult("wallet_checkout", "wallet") try: from apps.business.wallet.models import Wallet, WalletLedger from apps.business.wallet.services.wallet_service import WalletService # Wallet table queryable wallet_count = Wallet.objects.count() r.ok("wallet_table_queryable", f"{wallet_count} wallets found") # WalletService has expected methods for method_name in ["credit_main_wallet", "debit_main_wallet", "credit_repurchase_wallet", "debit_repurchase_wallet"]: if hasattr(WalletService, method_name): r.ok(f"wallet_service_{method_name}_exists", f"WalletService.{method_name} is present") else: r.fail(f"wallet_service_{method_name}_exists", f"WalletService.{method_name} NOT FOUND") # Ledger source types include PURCHASE if hasattr(WalletLedger, "SourceType"): source_types = [c[0] for c in WalletLedger.SourceType.choices] if "PURCHASE" in source_types: r.ok("wallet_purchase_source_type", "PURCHASE source type is defined") else: r.fail("wallet_purchase_source_type", "PURCHASE not in WalletLedger.SourceType") else: r.fail("wallet_source_type_class", "WalletLedger.SourceType not found") # New FK: funded_orders reverse relation is queryable sample_wallet = Wallet.objects.first() if sample_wallet: from apps.business.orders.models import Order if hasattr(Order, 'wallet_ledger_entry'): funded = Order.objects.filter( wallet_ledger_entry__user=sample_wallet.user ).count() r.ok("wallet_funded_orders_reverse_query", f"funded_orders query returns {funded} orders") else: r.skip("wallet_funded_orders_reverse_query", "no wallets in DB to test against") except Exception as e: r.fail("wallet_checkout_error", exc=e) return r.summary() def smoke_distributor_creation(): """Verify distributor creation service is importable and logic is intact.""" r = SmokeResult("distributor_creation", "distributor") try: from apps.business.distributor.services.distributor_service import ( create_distributor_id, generate_distributor_code, generate_global_position, ) r.ok("distributor_service_importable", "distributor_service module imported successfully") # Validate code generation logic (no DB write) from apps.business.products.models import Product from apps.accounts.models import User sample_product = Product.objects.filter(is_active=True, opportunity_bundle__isnull=False).first() if sample_product: r.ok("sample_bundle_product_found", f"product_id={sample_product.id}") # Simulate code generation using a mock position mock_position = 99999 mock_user = type("U", (), { "mobile": "9876543210", "id": "00000000-0000-0000-0000-000000000001", })() code = generate_distributor_code(mock_user, sample_product, mock_position) if code.startswith("TW") and len(code) > 8: r.ok("distributor_code_format", f"generated code format valid: {code}") else: r.fail("distributor_code_format", f"unexpected code format: {code}") else: r.skip("sample_bundle_product_found", "no active bundle product in DB") # DistributorID model has required fields from apps.business.distributor.models import DistributorID for field_name in ["user", "product", "distributor_code", "global_position", "binary_parent_distributor", "binary_position", "is_company_root"]: try: DistributorID._meta.get_field(field_name) r.ok(f"distributor_field_{field_name}", f"field '{field_name}' exists on DistributorID") except Exception: r.fail(f"distributor_field_{field_name}", f"field '{field_name}' missing from DistributorID model") except Exception as e: r.fail("distributor_creation_error", exc=e) return r.summary() def smoke_binary_placement(): """Verify binary tree placement constraints are intact post-migration.""" r = SmokeResult("binary_placement", "distributor") try: from apps.business.distributor.models import DistributorID from django.db.models import F, Q # Unique constraint on (binary_parent, binary_position) where both are non-null constraints = [c.name for c in DistributorID._meta.constraints] if "uniq_binary_parent_position" in constraints: r.ok("binary_unique_constraint_exists", "uniq_binary_parent_position constraint is present") else: r.fail("binary_unique_constraint_exists", f"uniq_binary_parent_position NOT in constraints: {constraints}") # LEFT/RIGHT choices are valid choices = dict(DistributorID.BinaryPosition.choices) if "LEFT" in choices and "RIGHT" in choices: r.ok("binary_position_choices", "LEFT and RIGHT choices defined") else: r.fail("binary_position_choices", f"unexpected choices: {choices}") # Company root nodes: query must work root_count = DistributorID.objects.filter(is_company_root=True).count() r.ok("company_root_query", f"{root_count} company root distributor(s)") # Binary tree depth is always >= 0 negative_depth = DistributorID.objects.filter(binary_level_depth__lt=0).count() if negative_depth: r.fail("binary_depth_non_negative", f"{negative_depth} distributors have negative depth") else: r.ok("binary_depth_non_negative", "all binary depths are >= 0") # BFS service importable try: from apps.business.binary_tree.services.binary_service import assign_company_global_binary_position r.ok("binary_service_importable", "binary_service imported successfully") except ImportError as ie: r.fail("binary_service_importable", f"ImportError: {ie}") except Exception as e: r.fail("binary_placement_error", exc=e) return r.summary() def smoke_sponsor_income(): """Verify sponsor income service query paths are intact.""" r = SmokeResult("sponsor_income", "distributor") try: from apps.business.distributor.services.sponsor_income_service import ( get_sponsor_income_summary, get_sponsor_income_ledger, ) r.ok("sponsor_income_service_importable", "sponsor_income_service imported successfully") # Check expected function signatures import inspect sig = inspect.signature(get_sponsor_income_summary) params = list(sig.parameters.keys()) if "user" in params: r.ok("sponsor_income_summary_signature", "get_sponsor_income_summary(user) signature correct") else: r.fail("sponsor_income_summary_signature", f"unexpected signature: {params}") # WalletLedger.SourceType.DIRECT_INCENTIVE exists from apps.business.wallet.models import WalletLedger source_types = [c[0] for c in WalletLedger.SourceType.choices] if "DIRECT_INCENTIVE" in source_types: r.ok("direct_incentive_source_type", "DIRECT_INCENTIVE in WalletLedger.SourceType") else: r.fail("direct_incentive_source_type", f"DIRECT_INCENTIVE not found in: {source_types}") # Sample run for the first active user with a distributor from apps.business.distributor.models import DistributorID sample_dist = DistributorID.objects.select_related("user").first() if sample_dist: try: summary = get_sponsor_income_summary(sample_dist.user) r.ok("sponsor_income_summary_executes", f"summary executed for user={sample_dist.user_id}: " f"total_earned={summary.get('total_earned')}") except Exception as e: r.fail("sponsor_income_summary_executes", exc=e) else: r.skip("sponsor_income_summary_executes", "no distributors in DB to test with") except Exception as e: r.fail("sponsor_income_error", exc=e) return r.summary() def smoke_wallet_transactions(): """Verify wallet transaction logging paths are queryable.""" r = SmokeResult("wallet_transactions", "wallet") try: from apps.business.wallet.models import WalletLedger, Wallet # All expected wallet types are present expected_types = {"EARNINGS", "REPURCHASE", "VOUCHER", "SUPER_COINS"} actual_types = {c[0] for c in WalletLedger.WalletType.choices} missing = expected_types - actual_types if missing: r.fail("wallet_types_complete", f"Missing wallet types: {missing}") else: r.ok("wallet_types_complete", f"All wallet types present: {actual_types}") # Transaction type choices tx_types = {c[0] for c in WalletLedger.TransactionType.choices} if {"CREDIT", "DEBIT"} == tx_types or {"CREDIT", "DEBIT"}.issubset(tx_types): r.ok("wallet_tx_types_complete", "CREDIT and DEBIT transaction types present") else: r.fail("wallet_tx_types_complete", f"Unexpected tx types: {tx_types}") # Latest N ledger entries are queryable without error recent = list(WalletLedger.objects.order_by("-id")[:10]) r.ok("wallet_ledger_recent_query", f"retrieved {len(recent)} recent ledger entries without error") # Wallet balance sum (sanity) from django.db.models import Sum from decimal import Decimal total_main = Wallet.objects.aggregate(s=Sum("main_balance"))["s"] or Decimal("0.00") r.ok("wallet_main_balance_sum", f"total main balance across all wallets: {total_main}") except Exception as e: r.fail("wallet_transaction_error", exc=e) return r.summary() def smoke_order_type_isolation(): """Verify BUNDLE and RETAIL order filters are logically isolated.""" r = SmokeResult("order_type_isolation", "orders") try: from apps.business.orders.models import Order bundle = Order.objects.filter(order_type="BUNDLE") retail = Order.objects.filter(order_type="RETAIL") both = Order.objects.filter(order_type__in=["BUNDLE", "RETAIL"]) total = Order.objects.count() r.ok("order_type_filter_bundle", f"BUNDLE filter returns {bundle.count()}") r.ok("order_type_filter_retail", f"RETAIL filter returns {retail.count()}") r.ok("order_type_filter_combined", f"BUNDLE+RETAIL covers {both.count()} of {total} total orders") # No order should appear in both sets simultaneously (types are mutually exclusive) overlap = Order.objects.filter(order_type="BUNDLE").filter(order_type="RETAIL").count() if overlap: r.fail("order_type_no_overlap", f"{overlap} orders match both BUNDLE and RETAIL (impossible — DB issue)") else: r.ok("order_type_no_overlap", "BUNDLE and RETAIL are mutually exclusive as expected") # Any order with a distributor must be BUNDLE dist_retail = Order.objects.filter( distributor__isnull=False, order_type="RETAIL", ).count() if dist_retail: r.fail("distributor_only_on_bundle_orders", f"{dist_retail} RETAIL orders have a distributor FK (should only be on BUNDLE)") else: r.ok("distributor_only_on_bundle_orders", "no RETAIL orders have a distributor FK") except Exception as e: r.fail("order_type_isolation_error", exc=e) return r.summary() def smoke_opportunity_bundle_lookup(): """Verify opportunity bundle product lookup is intact.""" r = SmokeResult("opportunity_bundle_lookup", "schemes") try: from apps.business.schemes.models import OpportunityBundle from apps.business.products.models import Product bundles = OpportunityBundle.objects.filter(is_active=True) r.ok("active_bundles_queryable", f"{bundles.count()} active opportunity bundles") for bundle in bundles[:5]: products = Product.objects.filter(opportunity_bundle=bundle, is_active=True) r.ok(f"bundle_{bundle.id}_products", f"bundle '{bundle.name}' has {products.count()} active products") # Reverse lookup: from product to bundle sample = Product.objects.filter(opportunity_bundle__isnull=False).first() if sample: ob = sample.opportunity_bundle r.ok("product_to_bundle_reverse", f"product '{sample.name}' -> bundle '{ob.name}'") else: r.skip("product_to_bundle_reverse", "no products linked to bundles") except Exception as e: r.fail("opportunity_bundle_lookup_error", exc=e) return r.summary() def smoke_franchise_coverage_lookup(): """Verify franchise pincode coverage lookup is operational.""" r = SmokeResult("franchise_coverage_lookup", "geo") try: from apps.geo.models import FranchiseStore stores = FranchiseStore.objects.filter(is_active=True) r.ok("franchise_store_queryable", f"{stores.count()} active franchise stores") try: from apps.geo.models import FranchisePincodeCoverage coverage_total = FranchisePincodeCoverage.objects.count() r.ok("franchise_pincode_coverage_queryable", f"FranchisePincodeCoverage queryable — {coverage_total} entries") except Exception as e: r.fail("franchise_pincode_coverage_import", exc=e) try: from apps.geo.models import FranchiseStoreInventory inv_total = FranchiseStoreInventory.objects.count() r.ok("franchise_inventory_queryable", f"FranchiseStoreInventory queryable — {inv_total} entries") except Exception as e: r.fail("franchise_inventory_import", exc=e) try: from apps.geo.models import FranchiseStoreEarning earn_total = FranchiseStoreEarning.objects.count() r.ok("franchise_earning_queryable", f"FranchiseStoreEarning queryable — {earn_total} entries") except Exception as e: r.fail("franchise_earning_import", exc=e) except Exception as e: r.fail("franchise_coverage_lookup_error", exc=e) return r.summary() # --------------------------------------------------------------------------- # Aggregate runner # --------------------------------------------------------------------------- SMOKE_TESTS = [ smoke_retail_purchase_path, smoke_bundle_purchase_path, smoke_wallet_checkout, smoke_distributor_creation, smoke_binary_placement, smoke_sponsor_income, smoke_wallet_transactions, smoke_order_type_isolation, smoke_opportunity_bundle_lookup, smoke_franchise_coverage_lookup, ] def run_all_smoke_tests(verbose=False, stop_on_first_fail=False): _log("=" * 60) _log("DRY-RUN SMOKE TEST SUITE") _log("=" * 60) results = [] total_passed = 0 total_failed = 0 for fn in SMOKE_TESTS: _log(f" Running: {fn.__name__}") try: result = fn() except Exception as e: result = { "name": fn.__name__, "category": "unknown", "status": "ERROR", "passed": 0, "failed": 1, "checks": [{"label": "execution", "status": "ERROR", "detail": str(e)}], } _log(f" [ERROR] {fn.__name__}: {e}") results.append(result) total_passed += result.get("passed", 0) total_failed += result.get("failed", 0) status = result.get("status", "?") symbol = "✓" if status == "PASS" else "✗" _log(f" {symbol} {result['name']}: {status}") if stop_on_first_fail and result.get("status") != "PASS": _log("Stopping on first failure.") break _log("=" * 60) _log(f"SMOKE TESTS: Passed={total_passed} Failed={total_failed}") overall = "PASS" if total_failed == 0 else "FAIL" _log(f"OVERALL: {overall}") _log("=" * 60) return { "run_at": _ts(), "overall": overall, "passed": total_passed, "failed": total_failed, "results": results, } # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): p = argparse.ArgumentParser(description="Dry-run smoke test suite") p.add_argument("--verbose", action="store_true") p.add_argument("--json-out", default=None) p.add_argument("--stop-on-first-fail", action="store_true") args = p.parse_args() _setup_django() outcome = run_all_smoke_tests(verbose=args.verbose, stop_on_first_fail=args.stop_on_first_fail) if args.json_out: with open(args.json_out, "w") as f: json.dump(outcome, f, indent=2, default=str) _log(f"Smoke test results written to: {args.json_out}") sys.exit(0 if outcome["overall"] == "PASS" else 1) if __name__ == "__main__": main()