Skip to content

Commit

Permalink
Changed to Go style, table-driven structure
Browse files Browse the repository at this point in the history
  • Loading branch information
@eprbell
eprbell committed Jan 14, 2025
1 parent 37a334b commit 244ccff
Showing 1 changed file with 130 additions and 133 deletions.
263 changes: 130 additions & 133 deletions tests/test_accounting_method.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,10 +36,18 @@ class SeekLotResult:


@dataclass(frozen=True, eq=True)
class InTransactionSpec:
class InTransactionDescriptor:
spot_price: int
amount: int

@dataclass(frozen=True, eq=True)
class _Test:
description: str
lot_selection_method: AbstractAccountingMethod
in_transactions: List[InTransactionDescriptor]
amounts_to_match: List[int]
want: List[SeekLotResult]


class TestAccountingMethod(unittest.TestCase):
_configuration: Configuration
Expand All @@ -51,10 +59,10 @@ def setUpClass(cls) -> None:
def setUp(self) -> None:
self.maxDiff = None # pylint: disable=invalid-name

def _initialize_acquired_lots(self, in_transaction_spec_list: List[InTransactionSpec]) -> List[InTransaction]:
def _initialize_acquired_lots(self, in_transaction_descriptors: List[InTransactionDescriptor]) -> List[InTransaction]:
date = datetime.strptime("2021-01-01", "%Y-%m-%d")
in_transactions: List[InTransaction] = []
for i, in_transaction_spec in enumerate(in_transaction_spec_list):
for i, in_transaction_descriptor in enumerate(in_transaction_descriptors):
in_transactions.append(
InTransaction(
self._configuration,
Expand All @@ -63,49 +71,46 @@ def _initialize_acquired_lots(self, in_transaction_spec_list: List[InTransaction
"Coinbase",
"Bob",
"Buy",
RP2Decimal(in_transaction_spec.spot_price),
RP2Decimal(in_transaction_spec.amount),
RP2Decimal(in_transaction_descriptor.spot_price),
RP2Decimal(in_transaction_descriptor.amount),
row=1 + i,
)
)
date += timedelta(days=1)
return in_transactions

# This function adds all acquired lots at first and then does amount pairings.
def _test_fixed_lot_candidates(
self, lot_selection_method: AbstractAccountingMethod, in_transactions: List[InTransaction], amounts_to_match: List[int], want: List[SeekLotResult]
) -> None:
def _run_test_fixed_lot_candidates(self, lot_selection_method: AbstractAccountingMethod, test: _Test) -> None:
print(f"\nDescription: {test.description:}")
in_transactions = self._initialize_acquired_lots(test.in_transactions)
acquired_lot_candidates = lot_selection_method.create_lot_candidates(in_transactions, {})
acquired_lot_candidates.set_to_index(len(in_transactions) - 1)
print(in_transactions)
i = 0
for int_amount in amounts_to_match:
for int_amount in test.amounts_to_match:
amount = RP2Decimal(int_amount)
while True:
result = lot_selection_method.seek_non_exhausted_acquired_lot(acquired_lot_candidates, amount)
if result is None:
break
if result.amount >= amount:
acquired_lot_candidates.set_partial_amount(result.acquired_lot, result.amount - amount)
print(i, want[i], amount, result)
self.assertEqual(result.amount, RP2Decimal(want[i].amount))
self.assertEqual(result.acquired_lot.row, want[i].row)
self.assertEqual(result.amount, RP2Decimal(test.want[i].amount))
self.assertEqual(result.acquired_lot.row, test.want[i].row)
i += 1
break
acquired_lot_candidates.clear_partial_amount(result.acquired_lot)
amount -= result.amount
print(i, want[i], amount, result)
self.assertEqual(result.amount, RP2Decimal(want[i].amount))
self.assertEqual(result.acquired_lot.row, want[i].row)
self.assertEqual(result.amount, RP2Decimal(test.want[i].amount))
self.assertEqual(result.acquired_lot.row, test.want[i].row)
i += 1

# This function grows lot_candidates dynamically: it adds an acquired lot, does an amount pairing and repeats.
def _test_dynamic_lot_candidates(
self, lot_selection_method: AbstractAccountingMethod, in_transactions: List[InTransaction], amounts_to_match: List[int], want: List[SeekLotResult]
) -> None:
def _run_test_dynamic_lot_candidates(self, lot_selection_method: AbstractAccountingMethod, test: _Test) -> None:
print(f"\nDescription: {test.description:}")
in_transactions = self._initialize_acquired_lots(test.in_transactions)
acquired_lot_candidates = lot_selection_method.create_lot_candidates([], {})
i = 0
for int_amount in amounts_to_match:
for int_amount in test.amounts_to_match:
amount = RP2Decimal(int_amount)
while True:
if i < len(in_transactions):
Expand All @@ -116,125 +121,117 @@ def _test_dynamic_lot_candidates(
break
if result.amount >= amount:
acquired_lot_candidates.set_partial_amount(result.acquired_lot, result.amount - amount)
print(i, want[i], amount, result)
self.assertEqual(result.amount, RP2Decimal(want[i].amount))
self.assertEqual(result.acquired_lot.row, want[i].row)
self.assertEqual(result.amount, RP2Decimal(test.want[i].amount))
self.assertEqual(result.acquired_lot.row, test.want[i].row)
i += 1
break
acquired_lot_candidates.clear_partial_amount(result.acquired_lot)
amount -= result.amount
print(i, want[i], amount, result)
self.assertEqual(result.amount, RP2Decimal(want[i].amount))
self.assertEqual(result.acquired_lot.row, want[i].row)
self.assertEqual(result.amount, RP2Decimal(test.want[i].amount))
self.assertEqual(result.acquired_lot.row, test.want[i].row)
i += 1

def test_lot_candidates_with_fifo(self) -> None:
lot_selection_method = AccountingMethodFIFO()

# Simple test.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(11, 20), InTransactionSpec(12, 30)]),
amounts_to_match=[6, 4, 2, 18, 3],
want=[SeekLotResult(10, 1), SeekLotResult(4, 1), SeekLotResult(20, 2), SeekLotResult(18, 2), SeekLotResult(30, 3)],
)

# Test with requested amount greater than acquired lot.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(11, 20), InTransactionSpec(12, 30)]),
amounts_to_match=[15, 10, 5],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(15, 2), SeekLotResult(5, 2)],
)

# Test with dynamic lot candidates
self._test_dynamic_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(11, 20), InTransactionSpec(12, 30)]),
amounts_to_match=[6, 4, 2, 18, 3],
want=[SeekLotResult(10, 1), SeekLotResult(4, 1), SeekLotResult(20, 2), SeekLotResult(18, 2), SeekLotResult(30, 3)],
)

def test_lot_candidates_with_lifo(self) -> None:
lot_selection_method = AccountingMethodLIFO()

# Simple test.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(11, 20), InTransactionSpec(12, 30)]),
amounts_to_match=[7, 23, 19, 1, 9],
want=[SeekLotResult(30, 3), SeekLotResult(23, 3), SeekLotResult(20, 2), SeekLotResult(1, 2), SeekLotResult(10, 1)],
)

# Test with requested amount greater than acquired lot.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(11, 20), InTransactionSpec(12, 30)]),
amounts_to_match=[55, 5],
want=[SeekLotResult(30, 3), SeekLotResult(20, 2), SeekLotResult(10, 1), SeekLotResult(5, 1)],
)

# Test with dynamic lot candidates
self._test_dynamic_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(11, 20), InTransactionSpec(12, 30)]),
amounts_to_match=[4, 15, 27, 14],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(30, 3), SeekLotResult(3, 3), SeekLotResult(5, 2), SeekLotResult(6, 1)],
)

def test_fixed_lot_candidates_with_hifo(self) -> None:
lot_selection_method = AccountingMethodHIFO()

# Simple test.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(12, 20), InTransactionSpec(11, 30)]),
amounts_to_match=[15, 5, 20, 10, 7],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 3), SeekLotResult(10, 1)],
)

# Test with requested amount greater than acquired lot.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(12, 20), InTransactionSpec(11, 30)]),
amounts_to_match=[15, 5, 35, 5],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 1), SeekLotResult(5, 1)],
)

# Test with dynamic lot candidates
self._test_dynamic_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(10, 10), InTransactionSpec(12, 20), InTransactionSpec(11, 30)]),
amounts_to_match=[4, 16, 40],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(4, 2), SeekLotResult(30, 3), SeekLotResult(6, 1)],
)

def test_fixed_lot_candidates_with_lofo(self) -> None:
lot_selection_method = AccountingMethodLOFO()

# Simple test.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(12, 10), InTransactionSpec(10, 20), InTransactionSpec(11, 30)]),
amounts_to_match=[15, 5, 20, 10, 7],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 3), SeekLotResult(10, 1)],
)

# Test with requested amount greater than acquired lot.
self._test_fixed_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(12, 10), InTransactionSpec(10, 20), InTransactionSpec(11, 30)]),
amounts_to_match=[15, 5, 35, 5],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 1), SeekLotResult(5, 1)],
)

# Test with dynamic lot candidates
self._test_dynamic_lot_candidates(
lot_selection_method=lot_selection_method,
in_transactions=self._initialize_acquired_lots([InTransactionSpec(12, 10), InTransactionSpec(10, 20), InTransactionSpec(11, 30)]),
amounts_to_match=[4, 16, 40],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(4, 2), SeekLotResult(30, 3), SeekLotResult(6, 1)],
)
def test_with_fixed_lot_candidates(self) -> None:
# Go-style, table-based tests. The want field contains the expected results.
tests: List[_Test] = [
_Test(
description="Simple test (FIFO)",
lot_selection_method=AccountingMethodFIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(11, 20), InTransactionDescriptor(12, 30)],
amounts_to_match=[6, 4, 2, 18, 3],
want=[SeekLotResult(10, 1), SeekLotResult(4, 1), SeekLotResult(20, 2), SeekLotResult(18, 2), SeekLotResult(30, 3)],
),
_Test(
description="Requested amount greater than acquired lot (FIFO)",
lot_selection_method=AccountingMethodFIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(11, 20), InTransactionDescriptor(12, 30)],
amounts_to_match=[15, 10, 5],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(15, 2), SeekLotResult(5, 2)],
),
_Test(
description="Simple test (LIFO)",
lot_selection_method=AccountingMethodLIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(11, 20), InTransactionDescriptor(12, 30)],
amounts_to_match=[7, 23, 19, 1, 9],
want=[SeekLotResult(30, 3), SeekLotResult(23, 3), SeekLotResult(20, 2), SeekLotResult(1, 2), SeekLotResult(10, 1)],
),
_Test(
description="Requested amount greater than acquired lot (LIFO)",
lot_selection_method=AccountingMethodLIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(11, 20), InTransactionDescriptor(12, 30)],
amounts_to_match=[55, 5],
want=[SeekLotResult(30, 3), SeekLotResult(20, 2), SeekLotResult(10, 1), SeekLotResult(5, 1)],
),
_Test(
description="Simple test (HIFO)",
lot_selection_method=AccountingMethodHIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(12, 20), InTransactionDescriptor(11, 30)],
amounts_to_match=[15, 5, 20, 10, 7],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 3), SeekLotResult(10, 1)],
),
_Test(
description="Requested amount greater than acquired lot (HIFO)",
lot_selection_method=AccountingMethodHIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(12, 20), InTransactionDescriptor(11, 30)],
amounts_to_match=[15, 5, 35, 5],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 1), SeekLotResult(5, 1)],
),
_Test(
description="Simple test (LOFO)",
lot_selection_method=AccountingMethodLOFO(),
in_transactions=[InTransactionDescriptor(12, 10), InTransactionDescriptor(10, 20), InTransactionDescriptor(11, 30)],
amounts_to_match=[15, 5, 20, 10, 7],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 3), SeekLotResult(10, 1)],
),
_Test(
description="Requested amount greater than acquired lot (LOFO)",
lot_selection_method=AccountingMethodLOFO(),
in_transactions=[InTransactionDescriptor(12, 10), InTransactionDescriptor(10, 20), InTransactionDescriptor(11, 30)],
amounts_to_match=[15, 5, 35, 5],
want=[SeekLotResult(20, 2), SeekLotResult(5, 2), SeekLotResult(30, 3), SeekLotResult(10, 1), SeekLotResult(5, 1)],
)

]
for test in tests:
with self.subTest(name=f"{test.description}"):
self._run_test_fixed_lot_candidates(lot_selection_method=test.lot_selection_method, test=test)


def test_with_dynamic_lot_candidates(self) -> None:
# Go-style, table-based tests. The want field contains the expected results.
tests: List[_Test] = [
_Test(
description="Dynamic test (FIFO)",
lot_selection_method=AccountingMethodFIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(11, 20), InTransactionDescriptor(12, 30)],
amounts_to_match=[6, 4, 2, 18, 3],
want=[SeekLotResult(10, 1), SeekLotResult(4, 1), SeekLotResult(20, 2), SeekLotResult(18, 2), SeekLotResult(30, 3)],
),
_Test(
description="Dynamic test (LIFO)",
lot_selection_method=AccountingMethodLIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(11, 20), InTransactionDescriptor(12, 30)],
amounts_to_match=[4, 15, 27, 14],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(30, 3), SeekLotResult(3, 3), SeekLotResult(5, 2), SeekLotResult(6, 1)],
),
_Test(
description="Dynamic test (HIFO)",
lot_selection_method=AccountingMethodHIFO(),
in_transactions=[InTransactionDescriptor(10, 10), InTransactionDescriptor(12, 20), InTransactionDescriptor(11, 30)],
amounts_to_match=[4, 16, 40],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(4, 2), SeekLotResult(30, 3), SeekLotResult(6, 1)],
),
_Test(
description="Dynamic test (LOFO)",
lot_selection_method=AccountingMethodLOFO(),
in_transactions=[InTransactionDescriptor(12, 10), InTransactionDescriptor(10, 20), InTransactionDescriptor(11, 30)],
amounts_to_match=[4, 16, 40],
want=[SeekLotResult(10, 1), SeekLotResult(20, 2), SeekLotResult(4, 2), SeekLotResult(30, 3), SeekLotResult(6, 1)],
)
]
for test in tests:
with self.subTest(name=f"{test.description}"):
self._run_test_dynamic_lot_candidates(lot_selection_method=test.lot_selection_method, test=test)


if __name__ == "__main__":
Expand Down

0 comments on commit 244ccff

Please sign in to comment.