#!/usr/bin/env python3 import pytest import logging import math import os import pathlib import time from spindle import pytest_utils from spindle.pytest_utils import ( API_AUTH_BASE, API_BASE, client, client_get_raise, random_timedelta, client_login, hash_int, client_patch_raise, client_post_raise, ) pytest_utils.env() # noinspection PyUnresolvedReferences from django.conf import settings from rest_framework.status import ( HTTP_201_CREATED, HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND, HTTP_406_NOT_ACCEPTABLE, ) from core.utils import ( abn_formatter, cryptography_load_pem_public_key, cryptography_rsa_oaep_encrypt, date_as_first_day_of_month, csv_file_reader_dict, generate_password, json_dumps, sha3_256_hex4, urlsafe_b64encode, urllib_parse, tz_datetime, billing_datetime, billing_tz, ) from accounts.utils import cached_user_verify_code from accounts.models import ( User, Seller, Customer, Product, Subscription, upload_product_csv, ) # https://www.django-rest-framework.org/api-guide/testing/ SCRIPT_DIR, SCRIPT_FILE = os.path.split(os.path.abspath(__file__)) SPINDLE_DIR = pathlib.Path(SCRIPT_DIR, "..", "..").resolve() logger = pytest_utils.logger = logging.getLogger(__name__) @pytest.mark.django_db(transaction=True) def test_api_subscriptions(): """Test suite for subscription handling via API.""" assert settings.IS_PYTEST_SETTINGS superuser, superuser_created = pytest_utils.create_superuser("test.subs superuser") # Distributor dt_bet, dt_bet_admin, dt_bet_u_ser = pytest_utils.create_distributor("test.subs.distributor bet") dt_gam, dt_gam_admin, dt_gam_u_ser = pytest_utils.create_distributor("test.subs.distributor beta") sl_foo, sl_foo_admin, sl_foo_u_ser = pytest_utils.create_seller("test.subs.seller foo", Seller.TYPE_SYSTEM_INTEGRATOR) sl_bar, sl_bar_admin, sl_bar_u_ser = pytest_utils.create_seller("test.subs.seller bar", Seller.TYPE_RESELLER) cu_one, cu_one_admin, cu_one_u_ser, cu_devices = pytest_utils.create_customer( "test.subs.customer one", device_factory=pytest_utils.spindle_device_factory, num_devices=12) cu_two, cu_two_admin, cu_two_u_ser, cu_devices = pytest_utils.create_customer( "test.subs.customer two", device_factory=pytest_utils.spindle_device_factory, num_devices=12) cu_six, cu_six_admin, cu_six_u_ser, cu_devices = pytest_utils.create_customer( "test.subs.customer six", device_factory=pytest_utils.spindle_device_factory, num_devices=12) test_csv_path = SPINDLE_DIR / "data" / "products-test.csv" csv_f, csv_r = csv_file_reader_dict(test_csv_path) with csv_f: upload_product_csv(csv_r, upsert=False) products = {p.product_code: p for p in Product.objects_active()} # Create a lot of subscriptions in advance: product_count = 2 dt = billing_datetime() for product_code, product in products.items(): for distributor, seller, customer in ( (dt_bet, sl_foo, cu_one), (dt_gam, sl_bar, cu_one), (dt_gam, sl_foo, cu_two), (dt_bet, sl_foo, cu_six), ): for assigned_timestamp in ( dt + random_timedelta(-367, seed=hash_int(customer.name, product_code)), dt + random_timedelta(-36, seed=hash_int(customer.name, product_code)), dt + random_timedelta(-8, seed=hash_int(customer.name, product_code)), date_as_first_day_of_month(dt) + random_timedelta(0, seed=hash_int(customer.name, product_code)), ): if assigned_timestamp > dt: assigned_timestamp = dt pytest_utils.add_subscriptions(superuser, distributor, sl_foo, customer, product, product_count, assigned_timestamp=assigned_timestamp) for product_code, product in products.items(): time.sleep(2) now_dt = billing_datetime() for customer, customer_admin in ( (cu_one, cu_one_admin), (cu_two, cu_two_admin), ): admin_customer_profile = customer_admin.customer_profile for subscription in customer.subscriptions.filter(product=product): # End a random selection of subscriptions start_dt = subscription.customer_assigned_datetime delta_now = now_dt - start_dt delta_days = min(abs(delta_now.days), 1) ended_by_timestamp = (now_dt + random_timedelta(delta_days * 3, seed=hash_int(product_code, subscription.subscription_key)) - random_timedelta(delta_days * 1, seed=hash_int(customer.customer_uuid, subscription.subscription_key))) if ended_by_timestamp < now_dt: subscription.end_subscription_by(admin_customer_profile, ended_by_timestamp=ended_by_timestamp) sellers_url = f"{API_BASE}/sellers/" distributors_url = f"{API_BASE}/distributors/" customers_url = f"{API_BASE}/customers/" # Log in as all the users and run some tests user_metadata = {} for client_key, user, entity, uuid_exp, uuid_key in ( ("dt_bet_u_ser", dt_bet_u_ser, dt_bet, dt_bet.distributor_uuid, "distributor_uuid"), ("dt_bet_admin", dt_bet_admin, dt_bet, dt_bet.distributor_uuid, "distributor_uuid"), ("dt_gam_u_ser", dt_gam_u_ser, dt_gam, dt_gam.distributor_uuid, "distributor_uuid"), ("dt_gam_admin", dt_gam_admin, dt_gam, dt_gam.distributor_uuid, "distributor_uuid"), ("sl_foo_u_ser", sl_foo_u_ser, sl_foo, sl_foo.seller_uuid, "seller_uuid"), ("sl_foo_admin", sl_foo_admin, sl_foo, sl_foo.seller_uuid, "seller_uuid"), ("sl_bar_u_ser", sl_bar_u_ser, sl_bar, sl_bar.seller_uuid, "seller_uuid"), ("sl_bar_admin", sl_bar_admin, sl_bar, sl_bar.seller_uuid, "seller_uuid"), ("cu_one_admin", cu_one_admin, cu_one, cu_one.customer_uuid, "customer_uuid"), ("cu_one_u_ser", cu_one_u_ser, cu_one, cu_one.customer_uuid, "customer_uuid"), ("cu_two_admin", cu_two_admin, cu_two, cu_two.customer_uuid, "customer_uuid"), ("cu_two_u_ser", cu_two_u_ser, cu_two, cu_two.customer_uuid, "customer_uuid"), ): user_metadata[client_key] = json = client_login(client_key, user) assert json["user"][uuid_key] == str(uuid_exp), repr(json) if client_key.startswith("cu_") and client_key.endswith("_u_ser"): # Customer users can't do these queries continue if uuid_key == "seller_uuid": exp_distributor_uuids = frozenset(str(s.distributor_uuid) for s in entity.all_distributors_qs()) data = {uuid_key: str(uuid_exp)} r = client_get_raise(client_key, distributors_url, data) json = r.json() assert json["page_count"] == 1, repr(json) results = json["results"] got_distributor_uuids = frozenset(s["distributor_uuid"] for s in results) assert exp_distributor_uuids == got_distributor_uuids, repr((data, exp_distributor_uuids, json)) # print(repr(("XXX1", data, exp_distributor_uuids, results))) exp_customer_uuids = frozenset(str(s.customer_uuid) for s in entity.all_customers_qs()) data = {uuid_key: str(uuid_exp)} r = client_get_raise(client_key, customers_url, data) json = r.json() assert json["page_count"] == 1, repr(json) results = json["results"] got_customer_uuids = frozenset(s["customer_uuid"] for s in results) assert exp_customer_uuids == got_customer_uuids, repr((data, exp_customer_uuids, json)) # print(repr(("XXX2", data, exp_customer_uuids, results))) elif uuid_key in ("distributor_uuid", "customer_uuid"): exp_seller_uuids = frozenset(str(s.seller_uuid) for s in entity.all_sellers_qs()) data = {uuid_key: str(uuid_exp)} r = client_get_raise(client_key, sellers_url, data) json = r.json() assert json["page_count"] == 1, repr(json) results = json["results"] got_seller_uuids = frozenset(s["seller_uuid"] for s in results) assert exp_seller_uuids == got_seller_uuids, repr((data, exp_seller_uuids, json)) # print(repr(("XXX3", data, exp_seller_uuids, results))) else: assert False products_url = f"{API_BASE}/products/" r = client_get_raise("dt_gam_u_ser", products_url) json = r.json() # print(json)