""" Seed test subscription data for Playwright e2e tests. Creates three subscriptions for the default merchant: 1. ACTIVE subscription (for general detail + pause/resume tests) 2. ACTIVE subscription #2 (will be used by cancel tests — destroyed during run) 3. CANCELLED subscription (for cancelled-subscription tests) Writes seeded IDs to: /var/www/html/hubwallet/hubwallet-ui/e2e/fixtures/seed-data.json Usage (from hubwallet-api/): /var/www/html/hubwallet/.venv/bin/python scripts/seed_test_subscriptions.py """ import json import os import sys from datetime import datetime, timezone, timedelta # Make sure the project root is on the Python path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from src.core.config import settings # Register all ORM models so SQLAlchemy mapper can resolve relationships from src.core.database import register_models register_models() from src.apps.base.utils.functions import generate_secure_id from src.apps.payment_requests.models.payment_request import PaymentRequest from src.apps.payment_requests.models.recurring_payment_request import RecurringPaymentRequests from src.apps.payment_requests.enums import ( PaymentFrequencies, PaymentAuthorizationTypes, PaymentRequestStatusTypes, RecurringPaymentIntervals, RecurringPaymentEndTypes, RecurringPaymentRepeatTypes, PaymentCurrencies, ) from src.apps.subscriptions.models.subscription import Subscription from src.apps.subscriptions.models.subscription_activity import SubscriptionActivity from src.apps.subscriptions.enums import SubscriptionStatus, SubscriptionActivityTypes from src.apps.subscriptions.crud import ( generate_subscription_literal, generate_subscription_id, ) from src.apps.invoices.models.invoice import Invoice from src.apps.merchants.models.merchant import Merchant from src.apps.customers.models.customer import Customer from src.apps.users.models.user import User from src.core.utils.enums import InvoiceStatusTypes, PaymentTypes from src.apps.payment_requests.utils.functions import ( generate_unique_payment_request_literal, ) FIXTURE_PATH = "/var/www/html/hubwallet/hubwallet-ui/e2e/fixtures/seed-data.json" MERCHANT_EMAIL = "default@merchant.com" def get_or_create_customer(db: Session, merchant_id: int) -> Customer: """Return the first non-deleted customer for this merchant, or create one.""" stmt = ( select(Customer) .where(Customer.merchant_id == merchant_id, Customer.deleted_at.is_(None)) .limit(1) ) customer = db.execute(stmt).scalar_one_or_none() if customer: name = f"{customer.first_name or ''} {customer.last_name or ''}".strip() or str(customer.id) print(f" Using existing customer id={customer.id} name={name}") return customer cus_id = generate_secure_id(prepend="cus", length=16) customer = Customer( customer_id=cus_id, first_name="Test", last_name="Customer", email=f"testcustomer+{cus_id}@example.com", merchant_id=merchant_id, ) db.add(customer) db.flush() print(f" Created customer id={customer.id}") return customer def create_payment_request_and_recurring( db: Session, merchant: Merchant, customer: Customer, created_by_id: int, amount: float = 9900.0, # in cents ) -> tuple[PaymentRequest, RecurringPaymentRequests]: """Create a PaymentRequest + RecurringPaymentRequests row.""" pr_id = generate_secure_id(prepend="pr", length=16) pr_literal = generate_unique_payment_request_literal(db) now = datetime.now(timezone.utc) pr = PaymentRequest( payment_request_id=pr_id, payment_request_literal=pr_literal, amount=amount, currency=PaymentCurrencies.USD, payment_frequency=PaymentFrequencies.RECURRING, authorization_type=PaymentAuthorizationTypes.PRE_AUTH, status=PaymentRequestStatusTypes.CREATED, payment_type=PaymentTypes.RECEIVE_PAYMENT, merchant_id=merchant.id, created_by_id=created_by_id, due_date=now, billing_date=now, save_payment_method=False, allow_tip=False, require_billing_address=False, require_sms_authorization=False, require_shipping_address=False, require_signature_authorization=False, require_cvv=False, is_certified=True, configure_adjustment=False, message="", enable_email=False, enable_sms=False, ) db.add(pr) db.flush() rec = RecurringPaymentRequests( payment_request_id=pr.id, prorate_first_payment=False, start_date=now, repeat_type=RecurringPaymentRepeatTypes.INTERVAL, interval=RecurringPaymentIntervals.MONTH, interval_value=1, end_type=RecurringPaymentEndTypes.UNTIL_CANCELLED, is_autopay=False, checkout_interval=RecurringPaymentIntervals.MONTH, checkout_interval_value=1, checkout_repeat_type=RecurringPaymentRepeatTypes.ON_DATE, ) db.add(rec) db.flush() return pr, rec def create_subscription( db: Session, pr: PaymentRequest, merchant: Merchant, customer: Customer, status: SubscriptionStatus, name: str, ) -> Subscription: """Create a Subscription record with the given status.""" sub_id = generate_subscription_id() sub_literal = generate_subscription_literal(db) now = datetime.now(timezone.utc) next_billing = now + timedelta(days=30) sub = Subscription( subscription_id=sub_id, subscription_literal=sub_literal, name=name, status=status, merchant_id=merchant.id, customer_id=customer.id, payment_request_id=pr.id, total_billed=0.0, total_paid=0.0, invoices_generated=0, invoices_paid=0, next_billing_date=next_billing if status == SubscriptionStatus.ACTIVE else None, ) db.add(sub) db.flush() # Add a creation activity activity = SubscriptionActivity( subscription_id=sub.id, activity_type="subscription.created", description=f"Subscription '{name}' created by seeder.", actor_type="system", ) db.add(activity) db.flush() return sub def create_invoice_for_sub( db: Session, pr: PaymentRequest, sub: Subscription, merchant: Merchant, customer: Customer, status: int, sequence: int, ) -> Invoice: """Create an Invoice linked to the subscription.""" inv_id = generate_secure_id(prepend="inv", length=16) # Generate a unique numeric literal now = datetime.now(timezone.utc) inv_literal = f"INV{int(now.timestamp())}{sequence}" paid_amount = pr.amount if status == InvoiceStatusTypes.PAID else 0.0 paid_date = now if status == InvoiceStatusTypes.PAID else None inv = Invoice( invoice_id=inv_id, invoice_literal=inv_literal, amount=pr.amount, due_date=now, billing_date=now, status=status, payment_request_id=pr.id, subscription_id=sub.id, merchant_id=merchant.id, customer_id=customer.id, paid_amount=paid_amount, paid_date=paid_date, sequence_id=sequence, ) db.add(inv) db.flush() return inv def main(): # Build DB URL from settings but always use localhost for local runner # (Docker container name 'fastapi_db' is only resolvable inside Docker network) db_url = settings.DATABASE_URL or ( f"postgresql+psycopg://{settings.DB_USERNAME or settings.DB_USER}:{settings.DB_PASSWORD}" f"@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}" ) # Replace the Docker service hostname with localhost so the script runs # directly on the host machine (Postgres exposed on localhost:5432) db_url = db_url.replace("@fastapi_db:", "@localhost:") print(f"Connecting to database at host localhost...") engine = create_engine(db_url, echo=False) with Session(engine) as db: # Find merchant stmt = select(Merchant).where( Merchant.email == MERCHANT_EMAIL, Merchant.deleted_at.is_(None), ) merchant = db.execute(stmt).scalar_one_or_none() if not merchant: print(f"ERROR: Merchant with email '{MERCHANT_EMAIL}' not found.") sys.exit(1) print(f"Found merchant: id={merchant.id} email={merchant.email}") # Find or create customer customer = get_or_create_customer(db, merchant.id) # Find a user belonging to the merchant for created_by_id user_stmt = ( select(User) .where(User.merchant_id == merchant.id, User.deleted_at.is_(None)) .limit(1) ) merchant_user = db.execute(user_stmt).scalar_one_or_none() if not merchant_user: # Fallback: use any non-superuser user_stmt2 = select(User).where(User.deleted_at.is_(None)).limit(1) merchant_user = db.execute(user_stmt2).scalar_one_or_none() if not merchant_user: print("ERROR: No user found to use as created_by_id.") sys.exit(1) print(f"Using user id={merchant_user.id} email={merchant_user.email} for created_by_id") seed_data = {} # ── 1. ACTIVE subscription (detail + pause/resume tests) ────────────── print("\n[1/3] Creating ACTIVE subscription...") pr_active, rec_active = create_payment_request_and_recurring( db, merchant, customer, created_by_id=merchant_user.id ) sub_active = create_subscription( db, pr_active, merchant, customer, SubscriptionStatus.ACTIVE, "Test Active Subscription", ) # Add a PAID invoice and a PENDING invoice create_invoice_for_sub( db, pr_active, sub_active, merchant, customer, InvoiceStatusTypes.PAID, sequence=1 ) create_invoice_for_sub( db, pr_active, sub_active, merchant, customer, InvoiceStatusTypes.PENDING, sequence=2 ) sub_active.invoices_generated = 2 sub_active.invoices_paid = 1 sub_active.total_billed = pr_active.amount * 2 sub_active.total_paid = pr_active.amount db.flush() print(f" subscription_id={sub_active.subscription_id}") print(f" subscription_literal={sub_active.subscription_literal}") seed_data["active_subscription"] = { "subscription_id": sub_active.subscription_id, "subscription_literal": sub_active.subscription_literal, "status": SubscriptionStatus.ACTIVE, } # ── 2. ACTIVE subscription #2 (cancel tests will destroy this one) ─── print("\n[2/3] Creating second ACTIVE subscription (for cancel tests)...") pr_cancel, rec_cancel = create_payment_request_and_recurring( db, merchant, customer, created_by_id=merchant_user.id ) sub_cancel = create_subscription( db, pr_cancel, merchant, customer, SubscriptionStatus.ACTIVE, "Test Active Subscription (For Cancel)", ) print(f" subscription_id={sub_cancel.subscription_id}") print(f" subscription_literal={sub_cancel.subscription_literal}") seed_data["cancel_subscription"] = { "subscription_id": sub_cancel.subscription_id, "subscription_literal": sub_cancel.subscription_literal, "status": SubscriptionStatus.ACTIVE, } # ── 3. CANCELLED subscription ───────────────────────────────────────── print("\n[3/3] Creating CANCELLED subscription...") pr_cancelled, rec_cancelled = create_payment_request_and_recurring( db, merchant, customer, created_by_id=merchant_user.id ) sub_cancelled = create_subscription( db, pr_cancelled, merchant, customer, SubscriptionStatus.CANCELLED, "Test Cancelled Subscription", ) # Add cancelled activity db.add(SubscriptionActivity( subscription_id=sub_cancelled.id, activity_type="subscription.cancelled", description="Subscription cancelled (seeded).", actor_type="system", )) db.flush() print(f" subscription_id={sub_cancelled.subscription_id}") print(f" subscription_literal={sub_cancelled.subscription_literal}") seed_data["cancelled_subscription"] = { "subscription_id": sub_cancelled.subscription_id, "subscription_literal": sub_cancelled.subscription_literal, "status": SubscriptionStatus.CANCELLED, } db.commit() print("\nAll records committed successfully.") # Write fixture JSON os.makedirs(os.path.dirname(FIXTURE_PATH), exist_ok=True) with open(FIXTURE_PATH, "w") as f: json.dump(seed_data, f, indent=2) print(f"\nSeed data written to: {FIXTURE_PATH}") print(json.dumps(seed_data, indent=2)) if __name__ == "__main__": main()