from django.test import TransactionTestCase from core.clients.platform import LocalPlatformClient from core.lib.money import money from core.models import LinkGroup, OperatorWallets, Platform, Requisition, Wallet from core.tests.helpers import AggregatorPlatformMixin class TestMoney(AggregatorPlatformMixin, TransactionTestCase): def setUp(self): super().setUp() self.plat_client = LocalPlatformClient() self.plat_client.instance = self.platform self.linkgroup = LinkGroup.objects.create( user=self.user, name="Test", platform_owner_cut_percentage=30, requisition_owner_cut_percentage=40, operator_cut_percentage=30, ) self.aggregator.link_group = self.linkgroup self.aggregator.save() self.platform.link_group = self.linkgroup self.platform.save() self.req = Requisition.objects.create( user=self.user, aggregator=self.aggregator, requisition_id="3ba3e65d-f44c-4c4e-9e28-08cc080830f6", payment_details="CUSTOM PAYMENT", ) def create_wallets(self): self.wallet_1 = Wallet.objects.create( user=self.user, name="Platform wallet 1", address="alpha", ) self.wallet_2 = Wallet.objects.create( user=self.user, name="Requisition wallet 1", address="beta", ) self.wallet_3 = Wallet.objects.create( user=self.user, name="Operator wallet 1", address="gamma", ) self.platform.payees.set([self.wallet_1]) self.req.payees.set([self.wallet_2]) op, _ = OperatorWallets.objects.get_or_create(user=self.user) op.payees.set([self.wallet_3]) self.platform.save() self.req.save() op.save() # Platform: 30 # Requisition: 40 # Operator: 30 def test_get_payment_list_full(self): self.create_wallets() self.platform.throughput = 1000 self.req.throughput = 1000 profit = 100 pay_list = money.get_pay_list( self.linkgroup, [self.req], [self.platform], self.user, profit, ) expected_list = { self.wallet_3: [(30.0, "Operator cut for 1 of 1 operators, total 30.0")], self.wallet_1: [(30.0, "Platform Test cut for 1 of 1 payees, total 30.0")], self.wallet_2: [ ( 40.0, ( f"Requisition Aggregator: {self.aggregator.name} ID: " f"{self.req.requisition_id} cut for 1 of 1 payees, total 40.0" ), ) ], } self.assertDictEqual(pay_list, expected_list) def test_get_payment_list_full_duplicates(self): self.create_wallets() self.req.payees.set([self.wallet_1]) self.platform.throughput = 1000 self.req.throughput = 1000 profit = 100 pay_list = money.get_pay_list( self.linkgroup, [self.req], [self.platform], self.user, profit, ) expected_list = { self.wallet_3: [(30.0, "Operator cut for 1 of 1 operators, total 30.0")], self.wallet_1: [ (30.0, "Platform Test cut for 1 of 1 payees, total 30.0"), ( 40.0, ( f"Requisition Aggregator: {self.aggregator.name} ID: " f"{self.req.requisition_id} cut for 1 of 1 payees, total 40.0" ), ), ], } self.assertDictEqual(pay_list, expected_list) def create_wallets_for_operator(self, num): wallets = [] for i in range(num): wallet = Wallet.objects.create( user=self.user, name=f"Operator wallet {i}", address=f"operator{i}", ) wallets.append(wallet) op, _ = OperatorWallets.objects.get_or_create(user=self.user) op.payees.set(wallets) return wallets def create_platforms_and_wallets(self, num): platforms = [] wallets = [] for i in range(num): platform = Platform.objects.create( user=self.user, name=f"Platform {i}", service="agora", token="a", password="a", otp_token="a", username="myuser", link_group=self.linkgroup, ) wallet = Wallet.objects.create( user=self.user, name=f"Platform wallet {i}", address=f"platform{i}", ) platform.payees.set([wallet]) platforms.append(platform) wallets.append(wallet) return platforms, wallets def create_requisitions_and_wallets(self, num): requisitions = [] wallets = [] for i in range(num): req = Requisition.objects.create( user=self.user, aggregator=self.aggregator, requisition_id=f"3ba3e65d-f44c-4c4e-9e28-08cc080830f6{i}", ) wallet = Wallet.objects.create( user=self.user, name=f"Requisition wallet {i}", address=f"requisition{i}", ) req.payees.set([wallet]) requisitions.append(req) wallets.append(wallet) return requisitions, wallets def test_get_payment_list_multi(self): wallet_num_operator = 3 wallet_num_platform = 3 wallet_num_requisition = 3 wallets_operator = self.create_wallets_for_operator(wallet_num_operator) platforms, wallets_platforms = self.create_platforms_and_wallets( wallet_num_platform ) requisitions, wallets_requisition = self.create_requisitions_and_wallets( wallet_num_requisition ) profit = 100 throughput_platform = [400, 300, 300] throughput_requisition = [300, 400, 300] expected_operator = [10, 10, 10] expected_platform = [12, 9, 9] expected_requisition = [12, 16, 12] for index, platform in enumerate(platforms): platform.throughput = throughput_platform[index] for index, req in enumerate(requisitions): req.throughput = throughput_requisition[index] pay_list = money.get_pay_list( self.linkgroup, requisitions, platforms, self.user, profit, ) for wallet, payments in pay_list.items(): if wallet in wallets_operator: for amount, detail in payments: self.assertEqual( amount, expected_operator[wallets_operator.index(wallet)] ) self.assertIn("Operator cut", detail) elif wallet in wallets_platforms: for amount, detail in payments: self.assertEqual( amount, expected_platform[wallets_platforms.index(wallet)] ) self.assertIn("Platform", detail) elif wallet in wallets_requisition: for amount, detail in payments: self.assertEqual( amount, expected_requisition[wallets_requisition.index(wallet)] ) self.assertIn("Requisition", detail)