#!/usr/bin/env python3 import pytest import logging import os import pathlib from spindle import pytest_utils from spindle.pytest_utils import ( API_AUTH_BASE, API_BASE, client, client_create_user, client_get_raise, client_login, 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_400_BAD_REQUEST, HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND, HTTP_406_NOT_ACCEPTABLE, ) from accounts.views import ( delete_cached_verify_email_salt, get_cached_verify_email_salt, ) from core.utils import ( abn_formatter, cryptography_load_pem_public_key, cryptography_rsa_oaep_encrypt, csv_file_reader_dict, generate_password, json_dumps, sha3_256_hex4, urlsafe_b64encode, urllib_parse, ) 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_end_to_end(): assert settings.IS_PYTEST_SETTINGS # superuser, superuser_created = pytest_utils.create_superuser("superuser-pytest") pytest_utils.create_superuser("superuser-pytest") # Extra entities created outside the API sl_sys, sl_sys_admin, sl_sys_user = pytest_utils.create_seller("pytest seller sys", Seller.TYPE_SYSTEM_INTEGRATOR) sl_sys_seller_uuid = str(sl_sys.seller_uuid) sl_sys_abn = sl_sys.abn cu_five_device_count = 13 cu_five, cu_five_admin, cu_five_user, cu_devices = pytest_utils.create_customer( "pytest customer five", device_factory=pytest_utils.spindle_device_factory, num_devices=cu_five_device_count, delete_existing_devices=True) cu_five_customer_uuid = str(cu_five.customer_uuid) cu_five_abn = cu_five.abn dt_gamma, dt_gamma_admin, dt_gamma_user = pytest_utils.create_distributor("pytest distributor gamma") dt_gamma_distributor_uuid = str(dt_gamma.distributor_uuid) dt_gamma_abn = dt_gamma.abn # Distributor distributor, distributor_admin, distributor_user = pytest_utils.create_distributor("pytest distributor delta") distributor_abn = distributor.abn # DISTRIBUTOR USER /api_auth/ login/metadata/logout coverage user = distributor_user # ------------------ # TEST LOGIN PROCESS # Look in metadata to verify user status metadata_url = f"{API_AUTH_BASE}/metadata" r = client_get_raise("dist_user", metadata_url) json = r.json() assert not json["user_is_authenticated"], json assert not json["user_authorization_bearer"], json assert not json["user_is_verified"], json assert not json["user_x_verified_jwt"], json # get public key and encode from GET login login_url = f"{API_AUTH_BASE}/login" r = client_get_raise("dist_user", login_url) json = r.json() rsa_public_key = cryptography_load_pem_public_key(json["public_rsa_key"]) enc_password = urlsafe_b64encode(cryptography_rsa_oaep_encrypt(rsa_public_key, user.raw_password)).decode("ASCII") # terms_checkbox set to False, rejected data = { "terms_checkbox": False, "trust": True, "email": user.email, "enc_password": enc_password, } client_post_raise("dist_user", login_url, data, expect_http_status_code=HTTP_406_NOT_ACCEPTABLE) data["terms_checkbox"] = True r = client_post_raise("dist_user", login_url, data) json = r.json() token = json["token"] assert token, (data, json) # Look in metadata to verify user status (using cookie) r = client_get_raise("dist_user", metadata_url) json = r.json() assert json["user_is_authenticated"], json assert json["user_authorization_bearer"], json assert not json["user_is_verified"], json assert not json["user_x_verified_jwt"], json # Verify GET to trigger email verify_url = f"{API_AUTH_BASE}/verify" data = {"email": user.email} client_post_raise("dist_user", verify_url, data, expect_http_status_code=HTTP_403_FORBIDDEN) # verify POST again with the code to get JWT data["code"] = cached_user_verify_code(user) # Grab the code out of django cache instead of email r = client_post_raise("dist_user", verify_url, data) json = r.json() x_verified_jwt = json["x_verified_jwt"] assert x_verified_jwt, json # Look in metadata to verify user status r = client_get_raise("dist_user", metadata_url) json = r.json() assert json["user_is_authenticated"], json assert json["user_is_verified"], json assert client("dist_user").cookies["terms-accepted"] assert client("dist_user").cookies["verified"] assert client("dist_user").cookies["sessionid"] # delete the session id and keep the verified cookie old_sessionid = client("dist_user").cookies["sessionid"] # del client("dist_user").cookies["sessionid"] logout_url = f"{API_AUTH_BASE}/logout" r = client_get_raise("dist_user", logout_url) assert r.status_text == "OK" assert client("dist_user").cookies["sessionid"] != old_sessionid assert client("dist_user").cookies["terms-accepted"] assert client("dist_user").cookies["verified"] # Look in metadata to verify user status r = client_get_raise("dist_user", metadata_url) json = r.json() assert not json["user_is_authenticated"], json assert not json["user_is_verified"], json # Log in again with verified cookie data = { "terms_checkbox": True, "trust": True, "email": user.email, "enc_password": enc_password, } r = client_post_raise("dist_user", login_url, data) json = r.json() token = json["token"] assert token, (data, json) assert json["user_is_verified"], json assert client("dist_user").cookies["sessionid"] assert client("dist_user").cookies["sessionid"] != old_sessionid # Look in metadata to verify user status r = client_get_raise("dist_user", metadata_url) json = r.json() assert json["user_is_authenticated"], json assert json["user_is_verified"], json json_user = json["user"] assert not json_user["distributor_profile"]["is_distributor_admin"], json_user distributor_uuid = json_user["distributor_uuid"] assert distributor_uuid, json_user distributor_url = json_user["distributor_url"] assert distributor_url, json_user scheme, netloc, path, params, query, fragment = urllib_parse(distributor_url) distributor_base = f"{API_BASE}/distributors/{distributor_uuid}" assert f"{distributor_base}/" == path # Check we can do things before password change client_get_raise("dist_user", distributor_url) # do change_password change_password_url = f"{API_BASE}/my-user/{distributor_user.username}/change_password/" old_password = distributor_user.raw_password old_enc_password = urlsafe_b64encode(cryptography_rsa_oaep_encrypt(rsa_public_key, old_password)).decode("ASCII") new_password = generate_password() new_enc_password = urlsafe_b64encode(cryptography_rsa_oaep_encrypt(rsa_public_key, new_password)).decode("ASCII") data = { "old_enc_password": old_enc_password, "new_enc_password": new_enc_password, } r = client_patch_raise("dist_user", change_password_url, data) json = r.json() assert json["old_enc_password"] == "removed", repr(json) assert json["new_enc_password"] == "set", repr(json) # Stuff is forbidden after we changed the password client_get_raise("dist_user", distributor_url, expect_http_status_code=HTTP_403_FORBIDDEN) # Logout logout_url = f"{API_AUTH_BASE}/logout" r = client_get_raise("dist_user", logout_url) assert r.status_text == "OK" # Try to log in with old password, this should fail with pytest.raises(RuntimeError, match="HTTP STATUS: 403"): client_login("dist_user_old_password", distributor_user) # Update and log in with new password distributor_user.raw_password = new_password client_login("dist_user", distributor_user) # ----------------------------------------------- # TEST PRODUCT / ENTITY / SUBSCRIPTIONS LIFECYCLE # Get the distributor r = client_get_raise("dist_user", distributor_url) json = r.json() assert json["distributor_uuid"] == distributor_uuid available_subscriptions_reports_url = f"{distributor_base}/available_subscriptions_reports/" r = client_get_raise("dist_user", available_subscriptions_reports_url) json = r.json() assert json["distributor_uuid"] == distributor_uuid assert len(json["available_reports"]) == 1 assert json["available_reports"][0]["filename"].startswith("incomplete-as-at-") assert json["subscriptions_count"] == 0 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_all = Product.objects_annotated().all() product_for_code = {p.product_code: p for p in products_all} assert len(product_for_code) == 7 # Create subscriptions for distributor create_subscriptions_url = f"{distributor_base}/create_subscriptions/" for product_code, product in product_for_code.items(): data = { "product": {"product_code": product.product_code }, "subscription_count": 90, } r = client_post_raise("dist_user", create_subscriptions_url, data) json = r.json() assert json["product"]["product_code"] == product_code, json assert len(json["subscriptions"]) == 90, json assert not any(s["seller_assigned_timestamp"] for s in json["subscriptions"]) subscription_keys = [s["subscription_key"] for s in json["subscriptions"]] subscriptions = Subscription.objects_annotated().filter(subscription_key__in=subscription_keys).all() for sub in subscriptions: assert not sub.seller_assigned_timestamp assert len(subscriptions) == 90 r = client_get_raise("dist_user", available_subscriptions_reports_url) json = r.json() assert json["distributor_uuid"] == distributor_uuid assert len(json["available_reports"]) == 1 assert json["available_reports"][0]["filename"].startswith("incomplete-as-at-") assert json["subscriptions_count"] == 7 * 90 # Create a seller seller_abn = pytest_utils.next_random_abn() seller_admin_email = "foxtrot.golf@example.com.invalid" data = { "abn": seller_abn, "name": "pytest Seller Foxtrot", "seller_type": Seller.TYPE_RESELLER, "contact_name": "Foxtrot Golf", "contact_email": seller_admin_email, } create_seller_url = f"{distributor_base}/create_seller/" r = client_post_raise("dist_user", create_seller_url, data) json = r.json() seller_uuid = json["seller"]["seller_uuid"] seller = Seller.objects.get(seller_uuid=seller_uuid) assert seller.abn == seller_abn assert list(seller.distributors.all()) == [distributor] # Create subscriptions which are assigned immediately to seller for product_code, product in product_for_code.items(): data = { "product": {"product_code": product.product_code }, "subscription_count": 10, "seller": {"seller_uuid": seller_uuid}, } r = client_post_raise("dist_user", create_subscriptions_url, data) json = r.json() assert json["product"]["product_code"] == product_code, json assert len(json["subscriptions"]) == 10, json subscription_keys = [s["subscription_key"] for s in json["subscriptions"]] subscriptions = Subscription.objects_annotated().filter(subscription_key__in=subscription_keys).all() assert len(subscriptions) == 10 r = client_get_raise("dist_user", available_subscriptions_reports_url) json = r.json() assert json["distributor_uuid"] == distributor_uuid assert len(json["available_reports"]) == 1 assert json["available_reports"][0]["filename"].startswith("incomplete-as-at-") assert json["subscriptions_count"] == 7 * 100 # for product_code, product in product_for_code.items(): # data = { # "product": {"product_code": product.product_code }, # "subscription_count": 90, # } # r = client_post_raise("dist_user", create_subscriptions_url, data) # json = r.json() # assert json["product"]["product_code"] == product_code, json # assert len(json["subscriptions"]) == 90, json # subscription_keys = [s["subscription_key"] for s in json["subscriptions"]] # subscriptions = Subscription.objects_annotated().filter(subscription_key__in=subscription_keys).all() # assert len(subscriptions) == 90 # Assign some more subscriptions from initial create to seller assign_subscriptions_url = f"{distributor_base}/assign_subscriptions/" for product_code, product in product_for_code.items(): data = { "product": {"product_code": product.product_code }, "subscription_count": 40, "seller": {"seller_uuid": seller_uuid}, } r = client_post_raise("dist_user", assign_subscriptions_url, data) json = r.json() assert json["product"]["product_code"] == product_code, json assert len(json["subscriptions"]) == 40, json subscription_keys = [s["subscription_key"] for s in json["subscriptions"]] subscriptions = Subscription.objects_annotated().filter(subscription_key__in=subscription_keys).all() assert len(subscriptions) == 40 sl_reseller_admin = User.objects.get(email=seller_admin_email) sl_reseller_admin.raw_password = raw_password = generate_password() sl_reseller_admin.set_password(raw_password) sl_reseller_admin.save(update_fields=["password"]) # Seller login and check json = client_login("sell_admin", sl_reseller_admin) json_user = json["user"] assert json_user["seller_profile"]["is_seller_admin"], json_user seller_uuid = json_user["seller_uuid"] assert seller_uuid, json_user seller_url = json_user["seller_url"] assert seller_url, json_user scheme, netloc, path, params, query, fragment = urllib_parse(seller_url) seller_base = f"{API_BASE}/sellers/{seller_uuid}" assert f"{seller_base}/" == path available_subscriptions_reports_url = f"{seller_base}/available_subscriptions_reports/" r = client_get_raise("sell_admin", available_subscriptions_reports_url) json = r.json() assert json["seller_uuid"] == seller_uuid assert len(json["available_reports"]) == 1 assert json["available_reports"][0]["filename"].startswith("incomplete-as-at-") # assert json["subscriptions_count"] == 7 * 50 # Create a customer (by seller admin) create_customer_url = f"{seller_base}/create_customer" cu_create_abn = pytest_utils.next_random_abn() cu_create_admin_email = "pytest.customer.create@example.com.invalid" cu_create_admin_name = "PytestCustomerCreate CreateCustomerAdmin" data = { "abn": cu_create_abn, "contact_email": cu_create_admin_email, "contact_name": cu_create_admin_name, "name": "Pytest Customer Create", "timezone": "Australia/Perth", } r = client_post_raise("sell_admin", create_customer_url, data) json = r.json() cu_create_customer_uuid = json["customer"]["customer_uuid"] cu_create = Customer.objects.get(customer_uuid=cu_create_customer_uuid) assert cu_create.abn == cu_create_abn assert list(cu_create.sellers.all()) == [seller] # Customer self signup customer_signup_url = f"{API_AUTH_BASE}/customer_signup" customer_abn = pytest_utils.next_random_abn() customer_admin_email = "pytest.customer@example.com.invalid" delete_cached_verify_email_salt(customer_admin_email) verify_salt = get_cached_verify_email_salt(customer_admin_email) assert not verify_salt, customer_admin_email data = { "abn": customer_abn, "contact_email": customer_admin_email, "contact_name": "Admin CustomerEleven", "name": "Pytest Customer Eleven", "sellers": [{"seller_uuid": seller_uuid}], "timezone": "Pacific/Auckland", } r = client_post_raise("cust_admin", customer_signup_url, data, expect_http_status_code=HTTP_401_UNAUTHORIZED) json = r.json() assert json == {"errors": {"code": ["Enter code received in email"]}} verify_salt = get_cached_verify_email_salt(customer_admin_email) assert verify_salt, customer_admin_email verify_code = sha3_256_hex4(customer_admin_email, verify_salt) data = { "abn": customer_abn, "code": verify_code, "contact_email": customer_admin_email, "contact_name": "Admin CustomerTwelve", "name": "Pytest Customer Twelve", "sellers": [{"seller_uuid": seller_uuid}], "timezone": "Pacific/Auckland", } client_post_raise("cust_admin", customer_signup_url, data, expect_http_status_code=HTTP_201_CREATED) customer = Customer.objects_annotated().filter(abn=customer_abn).get() user = cu_twelve_admin = User.objects.get(email=customer_admin_email) cu_twelve_admin.raw_password = raw_password = generate_password() cu_twelve_admin.set_password(raw_password) cu_twelve_admin.save(update_fields=["password"]) json = client_login("cust_admin", user) json_user = json["user"] assert json_user["customer_profile"]["is_customer_admin"], json_user customer_uuid = json_user["customer_uuid"] assert customer_uuid, json_user customer_url = json_user["customer_url"] assert customer_url, json_user assert list(cu_create.sellers.all()) == [seller] scheme, netloc, path, params, query, fragment = urllib_parse(customer_url) customer_base = f"{API_BASE}/customers/{customer_uuid}" assert f"{customer_base}/" == path available_subscriptions_reports_url = f"{customer_base}/available_subscriptions_reports/" r = client_get_raise("cust_admin", available_subscriptions_reports_url) json = r.json() assert json["customer_uuid"] == customer_uuid assert len(json["available_reports"]) == 1 assert json["available_reports"][0]["filename"].startswith("incomplete-as-at-") # assert json["subscriptions_count"] == 0 # Assign some subscriptions to customer assign_subscriptions_url = f"{seller_base}/assign_subscriptions/" for product_code, product in product_for_code.items(): data = { "product": {"product_code": product.product_code }, "subscription_count": 25, "customer": {"customer_uuid": customer_uuid}, } r = client_post_raise("sell_admin", assign_subscriptions_url, data) json = r.json() assert json["product"]["product_code"] == product_code, json assert len(json["subscriptions"]) == 25, json subscription_keys = [s["subscription_key"] for s in json["subscriptions"]] subscriptions = Subscription.objects_annotated().filter(subscription_key__in=subscription_keys).all() assert len(subscriptions) == 25 # Check customer suubscriptions available_subscriptions_reports_url = f"{customer_base}/available_subscriptions_reports/" r = client_get_raise("cust_admin", available_subscriptions_reports_url) json = r.json() assert len(json["available_reports"]) == 1 assert json["customer_uuid"] == customer_uuid assert json["available_reports"][0]["filename"].startswith("incomplete-as-at-") assert json["subscriptions_count"] == 7 * 25 # Do extra logins of all availabler users client_login("dist_admin", distributor_admin) client_login("dt_gamma_admin", dt_gamma_admin) client_login("dt_gamma_user", dt_gamma_user) dt_gamma_base = f"{API_BASE}/distributors/{dt_gamma_distributor_uuid}" client_login("sl_sys_admin", sl_sys_admin) client_login("sl_sys_user", sl_sys_user) sl_sys_base = f"{API_BASE}/sellers/{sl_sys_seller_uuid}" client_login("cu_five_admin", cu_five_admin) client_login("cu_five_user", cu_five_user) cu_five_base = f"{API_BASE}/customers/{cu_five_customer_uuid}" # ------------------------- # TEST PATCHING OF ENTITIES cu_five_url = f"{cu_five_base}/" data = {"sellers": [{"seller_uuid": sl_sys_seller_uuid}]} r = client_patch_raise("cu_five_admin", cu_five_url, data) json = r.json() print(json) assert len(json["sellers"]) == 1 sl_sys_url = f"{sl_sys_base}/" data = {"distributors": [{"distributor_uuid": dt_gamma_distributor_uuid}, {"distributor_uuid": distributor_uuid}]} r = client_patch_raise("sl_sys_admin", sl_sys_url, data) json = r.json() print(json) assert len(json["distributors"]) == 2 # ------------------ # TEST USER CREATION dist_second_user = client_create_user( view_prefix="distributor-", client_key="dist_admin", user_full_name="UserDeltaDist SecondDistUser", password=generate_password(), entity_type="distributor", entity_uuid=distributor_uuid, is_entity_admin=False) assert distributor.is_view_user(dist_second_user) assert not distributor.is_admin_user(dist_second_user) client_login("dist_second_user", dist_second_user) dist_second_admin = client_create_user( "distributor-", "dist_admin", "AdminDeltaDist SecondDistAdmin", generate_password(), "distributor", distributor_uuid, is_entity_admin=True) assert distributor.is_view_user(dist_second_user) assert distributor.is_admin_user(dist_second_admin) client_login("dist_second_admin", dist_second_admin) sell_user = client_create_user( "seller-", "sell_admin", "UserFoxtrotSeller FoxtrotUser", generate_password(), "seller", seller_uuid, is_entity_admin=False) assert seller.is_view_user(sell_user) assert not seller.is_admin_user(sell_user) client_login("sell_user", sell_user) cust_user = client_create_user( "", "cust_admin", "UserElevenCustomer ElevenUser", generate_password(), "customer", customer_uuid, is_entity_admin=False) assert customer.is_view_user(cust_user) assert not customer.is_admin_user(cust_user) client_login("cust_user", cust_user) # ------------ # TEST DEVICES devices_url = f"{API_BASE}/devices/" cu_five_next_cam = pytest_utils.next_test_cam() data = { "name": "cu five added test device", "username": cu_five_next_cam.username, "password": cu_five_next_cam.password, "lat": settings.GEO_DEFAULT_LAT + 0.01, "lon": settings.GEO_DEFAULT_LON + 0.01, "is_active": True, "sync_device_recordings": True, "customer_uuid": cu_five_customer_uuid, "mac": cu_five_next_cam.mac, "oak": cu_five_next_cam.oak, } client_post_raise("cu_five_admin", devices_url, data, expect_http_status_code=HTTP_201_CREATED) cu_five_device_count += 1 data["customer_uuid"] = customer_uuid client_post_raise("cu_five_admin", devices_url, data, expect_http_status_code=HTTP_404_NOT_FOUND) data["customer_uuid"] = cu_five_customer_uuid data["oak"] = "abcd-ef12-3456" client_post_raise("cu_five_admin", devices_url, data, expect_http_status_code=HTTP_400_BAD_REQUEST) client_post_raise("cu_five_admin", devices_url, {}, expect_http_status_code=HTTP_400_BAD_REQUEST) for client_key, exp_devices_count in ( ("cust_admin", 0), ("cust_user", 0), ("cu_five_admin", cu_five_device_count), ("cu_five_user", cu_five_device_count), ("dist_admin", None), ("dist_user", None), ("dt_gamma_admin", None), ("dt_gamma_user", None), ("sell_admin", None), ("sell_user", None), ("sl_sys_admin", None), ("sl_sys_user", None), ): if exp_devices_count is None: client_get_raise(client_key, devices_url, expect_http_status_code=HTTP_403_FORBIDDEN) continue r = client_get_raise(client_key, devices_url) json = r.json() assert json["count"] == exp_devices_count, repr(json) # ------------------ # TEST SUBSCRIPTIONS subscriptions_url = f"{API_BASE}/subscriptions/" for product_code, product in product_for_code.items(): params = {"product_code": product_code} r = client_get_raise("cust_admin", subscriptions_url, params) json = r.json() assert json["count"] == 25 assert json["page_count"] == 2 results = json["results"] params["page"] = 2 r = client_get_raise("cust_admin", subscriptions_url, params) json = r.json() results.extend(json["results"]) assert len(results) == 25 for sub in results: assert not (sub["ended_by_timestamp"] or sub["end_timestamp"]) end_subscriptions_url = f"{customer_base}/end_subscriptions/" end_subscription_keys = set() for sub in (results[:3] + results[-2:]): end_subscription_keys.add(sub["subscription_key"]) assert len(end_subscription_keys) == 5 data = {"subscriptions": [{"subscription_key": sk} for sk in end_subscription_keys]} assert len(data["subscriptions"]) == 5, json_dumps(data) r = client_post_raise("cust_admin", end_subscriptions_url, data) json = r.json() assert len(json["subscriptions"]) == 5, json_dumps(json) got_subscription_keys = frozenset(sk["subscription_key"] for sk in json["subscriptions"]) assert got_subscription_keys == end_subscription_keys product_subs = customer.subscriptions.filter(product__product_code=product_code) subscriptions = product_subs.filter(subscription_key__in=end_subscription_keys).all() assert len(subscriptions) == 5, subscriptions exp_ended_by_desc = f"{customer.name} (ABN: {customer_abn})" for sub in subscriptions: assert sub.ended_by_entity_description == exp_ended_by_desc subscriptions = product_subs.exclude(subscription_key__in=end_subscription_keys).all() assert len(subscriptions) == 20, subscriptions for sub in subscriptions: assert sub.ended_by_entity_description == "" end_subscriptions_url = f"{seller_base}/end_subscriptions/" end_subscription_keys = set() for sub in (results[3:8]): end_subscription_keys.add(sub["subscription_key"]) data = {"subscriptions": [{"subscription_key": sk} for sk in end_subscription_keys]} r = client_post_raise("sell_admin", end_subscriptions_url, data) json = r.json() got_subscription_keys = frozenset(sk["subscription_key"] for sk in json["subscriptions"]) assert got_subscription_keys == end_subscription_keys subscriptions = product_subs.filter(subscription_key__in=end_subscription_keys).all() assert len(subscriptions) == 5, subscriptions exp_descriptions = {"", exp_ended_by_desc} exp_ended_by_desc = f"{seller.name} (ABN: {seller_abn})" for sub in subscriptions: assert sub.ended_by_entity_description == exp_ended_by_desc subscriptions = product_subs.exclude(subscription_key__in=end_subscription_keys).all() assert len(subscriptions) == 20, subscriptions for sub in subscriptions: assert sub.ended_by_entity_description in exp_descriptions end_subscriptions_url = f"{distributor_base}/end_subscriptions/" end_subscription_keys = set() for sub in (results[-12:-7]): end_subscription_keys.add(sub["subscription_key"]) data = {"subscriptions": [{"subscription_key": sk} for sk in end_subscription_keys]} r = client_post_raise("dist_user", end_subscriptions_url, data) json = r.json() got_subscription_keys = frozenset(sk["subscription_key"] for sk in json["subscriptions"]) assert got_subscription_keys == end_subscription_keys exp_descriptions.add(exp_ended_by_desc) subscriptions = product_subs.exclude(subscription_key__in=end_subscription_keys).all() assert len(subscriptions) == 20, subscriptions for sub in subscriptions: assert sub.ended_by_entity_description in exp_descriptions exp_ended_by_desc = f"{distributor.name} (ABN: {distributor_abn})" subscriptions = product_subs.filter(subscription_key__in=end_subscription_keys).all() assert len(subscriptions) == 5, subscriptions for sub in subscriptions: assert sub.ended_by_entity_description == exp_ended_by_desc subscriptions = product_subs.filter(end_timestamp__isnull=True) assert len(subscriptions) == 10, subscriptions subscriptions = product_subs.filter(end_timestamp__isnull=False) assert len(subscriptions) == 15, subscriptions for client_key, base, exp_sub_csv_len, exp_metadata in ( ("dist_admin", distributor_base, 7 * 25 + 1, {"distributor_uuid": distributor_uuid, "distributor_abn": abn_formatter(distributor_abn)}), ("dist_user", distributor_base, 7 * 25 + 1, {"distributor_uuid": distributor_uuid, "distributor_abn": abn_formatter(distributor_abn)}), ("sell_admin", seller_base, 7 * 25 + 1, {"seller_uuid": seller_uuid, "seller_abn": abn_formatter(seller_abn)}), ("sell_user", seller_base, 7 * 25 + 1, {"seller_uuid": seller_uuid, "seller_abn": abn_formatter(seller_abn)}), ("cust_admin", customer_base, 7 * 25 + 1, {"customer_uuid": customer_uuid, "customer_abn": abn_formatter(customer_abn)}), ("cust_user", customer_base, 7 * 25 + 1, {"customer_uuid": customer_uuid, "customer_abn": abn_formatter(customer_abn)}), ("dt_gamma_admin", dt_gamma_base, 1, {"distributor_uuid": dt_gamma_distributor_uuid, "distributor_abn": abn_formatter(dt_gamma_abn)}), ("dt_gamma_user", dt_gamma_base, 1, {"distributor_uuid": dt_gamma_distributor_uuid, "distributor_abn": abn_formatter(dt_gamma_abn)}), ("cu_five_admin", cu_five_base, 1, {"customer_uuid": cu_five_customer_uuid, "customer_abn": abn_formatter(cu_five_abn)}), ("cu_five_user", cu_five_base, 1, {"customer_uuid": cu_five_customer_uuid, "customer_abn": abn_formatter(cu_five_abn)}), ("sl_sys_admin", sl_sys_base, 1, {"seller_uuid": sl_sys_seller_uuid, "seller_abn": abn_formatter(sl_sys_abn)}), ("sl_sys_user", sl_sys_base, 1, {"seller_uuid": sl_sys_seller_uuid, "seller_abn": abn_formatter(sl_sys_abn)}), ): report_url = f"{base}/get_subscriptions_report/" if client_key.startswith("cu") and client_key.endswith("_user"): client_get_raise(client_key, report_url, expect_http_status_code=HTTP_403_FORBIDDEN) else: r = client_get_raise(client_key, report_url) val = r.getvalue().decode("utf-8") csv_lines = val.splitlines() assert ((r.headers["Content-Type"] == "text/csv") and (len(csv_lines) == exp_sub_csv_len) ), (client_key, report_url, r, csv_lines) for report_method in ( "get_summary_subscriptions_active_for", "get_summary_subscriptions_not_activated", "get_summary_device_storage_for", "get_summary_users_for", ): expect_http_status_code = None report_url = f"{base}/{report_method}/" if not client_key.endswith("_admin") and report_method == "get_summary_users_for": expect_http_status_code = HTTP_403_FORBIDDEN elif client_key.startswith("cu") and report_method == "get_summary_subscriptions_not_activated": expect_http_status_code = HTTP_404_NOT_FOUND elif client_key.startswith("cu") and client_key.endswith("_user"): expect_http_status_code = HTTP_403_FORBIDDEN if expect_http_status_code: client_get_raise(client_key, report_url, expect_http_status_code=expect_http_status_code) continue r = client_get_raise(client_key, report_url) json = r.json() assert json["metadata"] == exp_metadata # -------------------------------------------------- # TEST entity lists (distributors/sellers/customers) for client_key, base, exp_distributors, exp_sellers, exp_customers in ( ("dist_admin", distributor_base, [distributor_uuid], [seller_uuid, sl_sys_seller_uuid], None), ("dist_user", distributor_base, [distributor_uuid], [seller_uuid, sl_sys_seller_uuid], None), ("sell_admin", seller_base, [distributor_uuid, dt_gamma_distributor_uuid], [seller_uuid], [customer_uuid, cu_five_customer_uuid, cu_create_customer_uuid]), ("cust_admin", customer_base, None, [seller_uuid, sl_sys_seller_uuid], [customer_uuid]), ("sl_sys_admin", sl_sys_base, [distributor_uuid, dt_gamma_distributor_uuid], [sl_sys_seller_uuid], [customer_uuid, cu_five_customer_uuid, cu_create_customer_uuid]), ("sl_sys_user", sl_sys_base, [distributor_uuid, dt_gamma_distributor_uuid], [sl_sys_seller_uuid], [customer_uuid, cu_five_customer_uuid, cu_create_customer_uuid]), ("cu_five_admin", cu_five_base, None, [seller_uuid, sl_sys_seller_uuid], [cu_five_customer_uuid]), ("cu_five_user", cu_five_base, None, None, [cu_five_customer_uuid]), ): for (obj_name, exp_uuids) in ( ("distributor", exp_distributors), ("seller", exp_sellers), ("customer", exp_customers), ): list_view_url = f"{API_BASE}/{obj_name}s/" if exp_uuids is None: client_get_raise(client_key, list_view_url, expect_http_status_code=HTTP_403_FORBIDDEN) continue r = client_get_raise(client_key, list_view_url) json = r.json() assert json["count"] == len(exp_uuids), repr(json) assert json["page_count"] == 1 uuid_field = f"{obj_name}_uuid" got_uuids = frozenset(r[uuid_field] for r in json["results"]) assert got_uuids == frozenset(exp_uuids), (client_key, base, obj_name, got_uuids, exp_uuids) # ------------- # TEST PRODUCTS # Load in extra products for testing test_csv_path = SPINDLE_DIR / "data" / "products-2022-08-16.csv" csv_f, csv_r = csv_file_reader_dict(test_csv_path) with csv_f: upload_product_csv(csv_r, upsert=False) products_url = f"{API_BASE}/products/" # Distributor user/admin can see all products # Seller User/Admin can see only products they have subscriptions for # Customer Admin can see only products they have subscriptions for # Customer User cannot see products for client_key, exp_products_count in ( ("cu_five_admin", 0), ("cu_five_user", None), ("cust_admin", 7), ("cust_user", None), ("dist_admin", 14), ("dist_user", 14), ("dt_gamma_admin", 14), ("dt_gamma_user", 14), ("sell_admin", 7), ("sell_user", 7), ("sl_sys_admin", 0), ("sl_sys_user", 0), ): if exp_products_count is None: client_get_raise(client_key, products_url, expect_http_status_code=HTTP_403_FORBIDDEN) continue r = client_get_raise(client_key, products_url) json = r.json() assert json["count"] == exp_products_count, repr(json)