tegwick
20e7f0f5bd
Automated issue wrap-up including: - Implementation completion verification - Test execution and validation - Cost tracking and note generation - Repository state commit 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
809 lines
30 KiB
Python
809 lines
30 KiB
Python
"""
|
|
Tests for Issue #114 - Cost Allocation Engine Implementation
|
|
|
|
This module contains comprehensive tests for the cost allocation engine
|
|
that distributes operational costs across active issues according to the
|
|
algorithm defined in Issue #88.
|
|
|
|
Tests cover:
|
|
- Core allocation algorithm with equal distribution
|
|
- Edge cases (no active issues, no costs, closed periods)
|
|
- Transaction audit trail creation
|
|
- Loss carried forward handling
|
|
- Integration with existing cost and activity tracking
|
|
- CLI command functionality
|
|
- Allocation reversal capabilities
|
|
"""
|
|
|
|
import pytest
|
|
import sqlite3
|
|
import tempfile
|
|
from datetime import datetime, date, timedelta
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import json
|
|
from contextlib import redirect_stdout
|
|
import io
|
|
|
|
from markitect.finance.allocation_engine import (
|
|
AllocationEngine, TransactionManager, AllocationStatus,
|
|
AllocationResult, IssueAllocation
|
|
)
|
|
from markitect.finance.models import FinanceModels
|
|
from markitect.finance.cost_manager import CostItemManager, CostItem
|
|
from markitect.finance.period_manager import PeriodManager, PeriodStatus
|
|
from markitect.issues.activity_tracker import IssueActivityTracker, ActivityType
|
|
|
|
|
|
def create_test_cost_item(cost_manager, name, category_id, cost_type, amount_eur, starting_from_date):
|
|
"""Helper function to create cost items with proper interface."""
|
|
cost_item = CostItem(
|
|
name=name,
|
|
category_id=category_id,
|
|
cost_type=cost_type,
|
|
amount_eur=amount_eur,
|
|
starting_from_date=starting_from_date
|
|
)
|
|
return cost_manager.create_cost_item(cost_item)
|
|
|
|
|
|
def create_unique_category(cost_manager, base_name, description="Test category"):
|
|
"""Helper function to create categories with unique names."""
|
|
import time
|
|
unique_name = f"{base_name}-{int(time.time()*1000000)}"
|
|
return cost_manager.create_category(unique_name, description)
|
|
|
|
|
|
class TestTransactionManager:
|
|
"""Test suite for TransactionManager class."""
|
|
|
|
@pytest.fixture
|
|
def temp_db(self):
|
|
"""Create temporary database for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
|
|
# Initialize schema
|
|
finance_models = FinanceModels(db_path)
|
|
finance_models.initialize_finance_schema()
|
|
|
|
yield db_path
|
|
|
|
# Cleanup
|
|
Path(db_path).unlink(missing_ok=True)
|
|
|
|
@pytest.fixture
|
|
def transaction_manager(self, temp_db):
|
|
"""Create TransactionManager instance for testing."""
|
|
return TransactionManager(temp_db)
|
|
|
|
@pytest.fixture
|
|
def sample_period(self, temp_db):
|
|
"""Create a sample period for testing."""
|
|
period_manager = PeriodManager(temp_db)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31)
|
|
)
|
|
return period_id
|
|
|
|
def test_transaction_manager_initialization(self, temp_db):
|
|
"""Test that TransactionManager initializes properly."""
|
|
manager = TransactionManager(temp_db)
|
|
assert manager.db_path == temp_db
|
|
assert isinstance(manager.finance_models, FinanceModels)
|
|
|
|
def test_create_allocation_transaction(self, transaction_manager, sample_period):
|
|
"""Test creating allocation transaction records."""
|
|
transaction_id = transaction_manager.create_allocation_transaction(
|
|
period_id=sample_period,
|
|
amount=Decimal('15.50'),
|
|
issue_id=123,
|
|
transaction_date=date(2025, 10, 15),
|
|
description="Test allocation transaction"
|
|
)
|
|
|
|
assert transaction_id is not None
|
|
assert isinstance(transaction_id, int)
|
|
|
|
# Verify transaction was created
|
|
with transaction_manager.finance_models.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
'SELECT * FROM cost_transactions WHERE id = ?',
|
|
(transaction_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
|
|
assert row is not None
|
|
assert row[1] == sample_period # period_id
|
|
assert row[3] == 'cost_allocated' # transaction_type
|
|
assert float(row[4]) == 15.50 # amount_eur
|
|
assert row[5] == 123 # issue_id
|
|
|
|
def test_create_loss_forward_transaction(self, transaction_manager, sample_period):
|
|
"""Test creating loss carried forward transactions."""
|
|
# Create next period
|
|
period_manager = PeriodManager(transaction_manager.db_path)
|
|
next_period_id = period_manager.create_period(
|
|
period_start=date(2025, 11, 1),
|
|
period_end=date(2025, 11, 30)
|
|
)
|
|
|
|
transaction_id = transaction_manager.create_loss_forward_transaction(
|
|
from_period_id=sample_period,
|
|
to_period_id=next_period_id,
|
|
amount=Decimal('25.75'),
|
|
transaction_date=date(2025, 11, 1),
|
|
description="Loss carried forward"
|
|
)
|
|
|
|
assert transaction_id is not None
|
|
|
|
# Verify transaction was created
|
|
with transaction_manager.finance_models.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
'SELECT * FROM cost_transactions WHERE id = ?',
|
|
(transaction_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
|
|
assert row is not None
|
|
assert row[1] == next_period_id # period_id
|
|
assert row[3] == 'loss_forward' # transaction_type
|
|
assert float(row[4]) == 25.75 # amount_eur
|
|
assert row[5] is None # issue_id (null for loss forward)
|
|
|
|
|
|
class TestAllocationEngine:
|
|
"""Test suite for AllocationEngine class."""
|
|
|
|
@pytest.fixture
|
|
def temp_db(self):
|
|
"""Create temporary database for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
|
|
# Initialize schema
|
|
finance_models = FinanceModels(db_path)
|
|
finance_models.initialize_finance_schema()
|
|
|
|
yield db_path
|
|
|
|
# Cleanup
|
|
Path(db_path).unlink(missing_ok=True)
|
|
|
|
@pytest.fixture
|
|
def allocation_engine(self, temp_db):
|
|
"""Create AllocationEngine instance for testing."""
|
|
return AllocationEngine(temp_db)
|
|
|
|
@pytest.fixture
|
|
def sample_costs(self, temp_db):
|
|
"""Create sample cost items for testing."""
|
|
cost_manager = CostItemManager(temp_db)
|
|
|
|
# Create cost category
|
|
category_id = create_unique_category(cost_manager, "Test Services", "Test cost category")
|
|
|
|
# Create cost items
|
|
cost_ids = []
|
|
cost_ids.append(create_test_cost_item(
|
|
cost_manager, "Monthly Service", category_id, "monthly",
|
|
Decimal('20.00'), date(2025, 10, 1)
|
|
))
|
|
|
|
cost_ids.append(create_test_cost_item(
|
|
cost_manager, "One-time Setup", category_id, "one_time",
|
|
Decimal('30.00'), date(2025, 10, 15)
|
|
))
|
|
|
|
return cost_ids
|
|
|
|
@pytest.fixture
|
|
def sample_period_with_costs(self, temp_db, sample_costs):
|
|
"""Create a period with associated costs."""
|
|
period_manager = PeriodManager(temp_db)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31)
|
|
)
|
|
return period_id
|
|
|
|
@pytest.fixture
|
|
def sample_issue_activities(self, temp_db, sample_period_with_costs):
|
|
"""Create sample issue activities for testing."""
|
|
activity_tracker = IssueActivityTracker(temp_db)
|
|
|
|
# Log activities for different issues
|
|
activity_ids = []
|
|
activity_ids.append(activity_tracker.log_activity(
|
|
issue_id=101,
|
|
activity_type=ActivityType.CREATED,
|
|
activity_date=date(2025, 10, 5),
|
|
period_id=sample_period_with_costs
|
|
))
|
|
|
|
activity_ids.append(activity_tracker.log_activity(
|
|
issue_id=102,
|
|
activity_type=ActivityType.MODIFIED,
|
|
activity_date=date(2025, 10, 10),
|
|
period_id=sample_period_with_costs
|
|
))
|
|
|
|
activity_ids.append(activity_tracker.log_activity(
|
|
issue_id=103,
|
|
activity_type=ActivityType.COMMENTED,
|
|
activity_date=date(2025, 10, 20),
|
|
period_id=sample_period_with_costs
|
|
))
|
|
|
|
return activity_ids
|
|
|
|
def test_allocation_engine_initialization(self, temp_db):
|
|
"""Test that AllocationEngine initializes with all required components."""
|
|
engine = AllocationEngine(temp_db)
|
|
|
|
assert engine.db_path == temp_db
|
|
assert isinstance(engine.finance_models, FinanceModels)
|
|
assert isinstance(engine.cost_manager, CostItemManager)
|
|
assert isinstance(engine.period_manager, PeriodManager)
|
|
assert isinstance(engine.activity_tracker, IssueActivityTracker)
|
|
assert isinstance(engine.transaction_manager, TransactionManager)
|
|
|
|
def test_successful_allocation(self, allocation_engine, sample_period_with_costs, sample_issue_activities):
|
|
"""Test successful cost allocation with active issues."""
|
|
result = allocation_engine.allocate_period_costs(sample_period_with_costs)
|
|
|
|
assert result.status == AllocationStatus.SUCCESS
|
|
assert result.period_id == sample_period_with_costs
|
|
assert result.total_costs == Decimal('50.00') # 20.00 + 30.00
|
|
assert len(result.active_issues) == 3 # Issues 101, 102, 103
|
|
assert result.cost_per_issue == Decimal('50.00') / 3
|
|
assert result.allocations_created == 3
|
|
assert result.transactions_created == 3
|
|
assert result.loss_carried_forward == Decimal('0.00')
|
|
|
|
def test_allocation_no_active_issues(self, allocation_engine, sample_period_with_costs):
|
|
"""Test allocation when no active issues exist (should carry forward loss)."""
|
|
result = allocation_engine.allocate_period_costs(sample_period_with_costs)
|
|
|
|
assert result.status == AllocationStatus.NO_ACTIVE_ISSUES
|
|
assert result.period_id == sample_period_with_costs
|
|
assert result.total_costs == Decimal('50.00')
|
|
assert len(result.active_issues) == 0
|
|
assert result.cost_per_issue == Decimal('0.00')
|
|
assert result.allocations_created == 0
|
|
assert result.transactions_created == 0
|
|
assert result.loss_carried_forward == Decimal('50.00')
|
|
|
|
def test_allocation_no_costs(self, allocation_engine):
|
|
"""Test allocation when no costs exist for period."""
|
|
# Create empty period
|
|
period_manager = PeriodManager(allocation_engine.db_path)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 11, 1),
|
|
period_end=date(2025, 11, 30)
|
|
)
|
|
|
|
result = allocation_engine.allocate_period_costs(period_id)
|
|
|
|
assert result.status == AllocationStatus.NO_COSTS_TO_ALLOCATE
|
|
assert result.total_costs == Decimal('0.00')
|
|
|
|
def test_allocation_period_closed(self, allocation_engine, sample_period_with_costs):
|
|
"""Test allocation on already closed period."""
|
|
# First allocation (should succeed)
|
|
result1 = allocation_engine.allocate_period_costs(sample_period_with_costs)
|
|
assert result1.status in [AllocationStatus.SUCCESS, AllocationStatus.NO_ACTIVE_ISSUES]
|
|
|
|
# Second allocation (should fail - period closed)
|
|
result2 = allocation_engine.allocate_period_costs(sample_period_with_costs)
|
|
assert result2.status == AllocationStatus.PERIOD_CLOSED
|
|
|
|
def test_allocation_period_not_found(self, allocation_engine):
|
|
"""Test allocation with non-existent period ID."""
|
|
result = allocation_engine.allocate_period_costs(99999)
|
|
|
|
assert result.status == AllocationStatus.ERROR
|
|
assert "not found" in result.message
|
|
|
|
def test_get_issue_allocations(self, allocation_engine, sample_period_with_costs, sample_issue_activities):
|
|
"""Test retrieving allocations for a specific issue."""
|
|
# Perform allocation first
|
|
allocation_engine.allocate_period_costs(sample_period_with_costs)
|
|
|
|
# Get allocations for issue 101
|
|
allocations = allocation_engine.get_issue_allocations(101)
|
|
|
|
assert len(allocations) == 1
|
|
allocation = allocations[0]
|
|
assert allocation['issue_id'] == 101
|
|
assert allocation['period_id'] == sample_period_with_costs
|
|
assert allocation['allocated_amount'] > 0
|
|
assert allocation['transaction_id'] is not None
|
|
|
|
def test_get_period_allocations(self, allocation_engine, sample_period_with_costs, sample_issue_activities):
|
|
"""Test retrieving all allocations for a specific period."""
|
|
# Perform allocation first
|
|
allocation_engine.allocate_period_costs(sample_period_with_costs)
|
|
|
|
# Get all allocations for the period
|
|
allocations = allocation_engine.get_period_allocations(sample_period_with_costs)
|
|
|
|
assert len(allocations) == 3
|
|
issue_ids = [alloc['issue_id'] for alloc in allocations]
|
|
assert 101 in issue_ids
|
|
assert 102 in issue_ids
|
|
assert 103 in issue_ids
|
|
|
|
def test_reverse_allocation(self, allocation_engine, sample_period_with_costs, sample_issue_activities):
|
|
"""Test reversing a cost allocation."""
|
|
# Perform allocation first
|
|
allocation_engine.allocate_period_costs(sample_period_with_costs)
|
|
|
|
# Get allocation to reverse
|
|
allocations = allocation_engine.get_issue_allocations(101)
|
|
assert len(allocations) == 1
|
|
allocation_id = allocations[0]['id']
|
|
|
|
# Reverse the allocation
|
|
success = allocation_engine.reverse_allocation(allocation_id)
|
|
assert success is True
|
|
|
|
# Verify allocation is removed
|
|
allocations_after = allocation_engine.get_issue_allocations(101)
|
|
assert len(allocations_after) == 0
|
|
|
|
def test_reverse_nonexistent_allocation(self, allocation_engine):
|
|
"""Test reversing non-existent allocation."""
|
|
success = allocation_engine.reverse_allocation(99999)
|
|
assert success is False
|
|
|
|
def test_allocation_with_carried_forward_loss(self, allocation_engine, temp_db):
|
|
"""Test allocation including loss carried forward from previous period."""
|
|
# Create period with carried forward loss
|
|
period_manager = PeriodManager(temp_db)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31),
|
|
loss_carried_forward=Decimal('15.00')
|
|
)
|
|
|
|
# Create some costs
|
|
cost_manager = CostItemManager(temp_db)
|
|
category_id = create_unique_category(cost_manager, "Test", "Test category")
|
|
create_test_cost_item(
|
|
cost_manager, "Test Cost", category_id, "one_time",
|
|
Decimal('10.00'), date(2025, 10, 15)
|
|
)
|
|
|
|
# Create issue activity
|
|
activity_tracker = IssueActivityTracker(temp_db)
|
|
activity_tracker.log_activity(
|
|
issue_id=201,
|
|
activity_type=ActivityType.CREATED,
|
|
activity_date=date(2025, 10, 10),
|
|
period_id=period_id
|
|
)
|
|
|
|
# Perform allocation
|
|
result = allocation_engine.allocate_period_costs(period_id)
|
|
|
|
assert result.status == AllocationStatus.SUCCESS
|
|
assert result.total_costs == Decimal('25.00') # 10.00 + 15.00 carried forward
|
|
assert len(result.active_issues) == 1
|
|
assert result.cost_per_issue == Decimal('25.00')
|
|
|
|
|
|
class TestAllocationIntegration:
|
|
"""Integration tests for allocation engine with other components."""
|
|
|
|
@pytest.fixture
|
|
def temp_db(self):
|
|
"""Create temporary database for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
|
|
# Initialize schema
|
|
finance_models = FinanceModels(db_path)
|
|
finance_models.initialize_finance_schema()
|
|
|
|
yield db_path
|
|
|
|
# Cleanup
|
|
Path(db_path).unlink(missing_ok=True)
|
|
|
|
def test_complete_allocation_workflow(self, temp_db):
|
|
"""Test complete workflow from cost creation to allocation."""
|
|
# Step 1: Create cost categories and items
|
|
cost_manager = CostItemManager(temp_db)
|
|
category_id = create_unique_category(cost_manager, "Infrastructure", "Server and hosting costs")
|
|
|
|
monthly_cost_id = create_test_cost_item(
|
|
cost_manager, "Server Hosting", category_id, "monthly",
|
|
Decimal('25.00'), date(2025, 10, 1)
|
|
)
|
|
|
|
oneoff_cost_id = create_test_cost_item(
|
|
cost_manager, "SSL Certificate", category_id, "one_time",
|
|
Decimal('15.00'), date(2025, 10, 10)
|
|
)
|
|
|
|
# Step 2: Create period
|
|
period_manager = PeriodManager(temp_db)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31)
|
|
)
|
|
|
|
# Step 3: Log issue activities
|
|
activity_tracker = IssueActivityTracker(temp_db)
|
|
activity_tracker.log_activity(
|
|
issue_id=301,
|
|
activity_type=ActivityType.CREATED,
|
|
activity_date=date(2025, 10, 5),
|
|
period_id=period_id
|
|
)
|
|
|
|
activity_tracker.log_activity(
|
|
issue_id=302,
|
|
activity_type=ActivityType.MODIFIED,
|
|
activity_date=date(2025, 10, 12),
|
|
period_id=period_id
|
|
)
|
|
|
|
# Step 4: Perform allocation
|
|
allocation_engine = AllocationEngine(temp_db)
|
|
result = allocation_engine.allocate_period_costs(period_id)
|
|
|
|
# Verify allocation success
|
|
assert result.status == AllocationStatus.SUCCESS
|
|
assert result.total_costs == Decimal('40.00')
|
|
assert len(result.active_issues) == 2
|
|
assert result.cost_per_issue == Decimal('20.00')
|
|
|
|
# Step 5: Verify database state
|
|
# Check allocations exist
|
|
allocations_301 = allocation_engine.get_issue_allocations(301)
|
|
allocations_302 = allocation_engine.get_issue_allocations(302)
|
|
|
|
assert len(allocations_301) == 1
|
|
assert len(allocations_302) == 1
|
|
assert allocations_301[0]['allocated_amount'] == 20.00
|
|
assert allocations_302[0]['allocated_amount'] == 20.00
|
|
|
|
# Check transactions exist
|
|
with allocation_engine.finance_models.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
'SELECT COUNT(*) FROM cost_transactions WHERE transaction_type = "cost_allocated"'
|
|
)
|
|
transaction_count = cursor.fetchone()[0]
|
|
assert transaction_count == 2
|
|
|
|
# Check period is closed
|
|
period_data = period_manager.get_period_by_id(period_id)
|
|
periods = [period_data] if period_data else []
|
|
assert len(periods) == 1
|
|
assert periods[0]['status'] == PeriodStatus.CLOSED.value
|
|
assert float(periods[0]['total_costs']) == 40.00
|
|
assert periods[0]['active_issues_count'] == 2
|
|
assert float(periods[0]['cost_per_issue']) == 20.00
|
|
|
|
def test_multi_period_allocation_workflow(self, temp_db):
|
|
"""Test allocation across multiple periods with loss carry forward."""
|
|
# Create cost items
|
|
cost_manager = CostItemManager(temp_db)
|
|
category_id = create_unique_category(cost_manager, "Services", "Monthly services")
|
|
|
|
create_test_cost_item(
|
|
cost_manager, "Monthly Service", category_id, "monthly",
|
|
Decimal('30.00'), date(2025, 10, 1)
|
|
)
|
|
|
|
# Create periods
|
|
period_manager = PeriodManager(temp_db)
|
|
period1_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31)
|
|
)
|
|
|
|
period2_id = period_manager.create_period(
|
|
period_start=date(2025, 11, 1),
|
|
period_end=date(2025, 11, 30)
|
|
)
|
|
|
|
# Period 1: No activities (should carry forward loss)
|
|
allocation_engine = AllocationEngine(temp_db)
|
|
result1 = allocation_engine.allocate_period_costs(period1_id)
|
|
|
|
assert result1.status == AllocationStatus.NO_ACTIVE_ISSUES
|
|
assert result1.loss_carried_forward == Decimal('30.00')
|
|
|
|
# Period 2: Add activity and allocate
|
|
activity_tracker = IssueActivityTracker(temp_db)
|
|
activity_tracker.log_activity(
|
|
issue_id=401,
|
|
activity_type=ActivityType.CREATED,
|
|
activity_date=date(2025, 11, 15),
|
|
period_id=period2_id
|
|
)
|
|
|
|
# Manually set carried forward for period 2 (simulating automatic carry forward)
|
|
with allocation_engine.finance_models.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
'UPDATE cost_periods SET loss_carried_forward = ? WHERE id = ?',
|
|
(float(Decimal('30.00')), period2_id)
|
|
)
|
|
|
|
result2 = allocation_engine.allocate_period_costs(period2_id)
|
|
|
|
assert result2.status == AllocationStatus.SUCCESS
|
|
assert result2.total_costs == Decimal('60.00') # 30.00 current + 30.00 carried forward
|
|
assert len(result2.active_issues) == 1
|
|
assert result2.cost_per_issue == Decimal('60.00')
|
|
|
|
|
|
class TestAllocationCLI:
|
|
"""Test suite for allocation CLI commands."""
|
|
|
|
@pytest.fixture
|
|
def temp_db(self):
|
|
"""Create temporary database for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
|
|
# Initialize schema
|
|
finance_models = FinanceModels(db_path)
|
|
finance_models.initialize_finance_schema()
|
|
|
|
yield db_path
|
|
|
|
# Cleanup
|
|
Path(db_path).unlink(missing_ok=True)
|
|
|
|
@pytest.fixture
|
|
def setup_test_data(self, temp_db):
|
|
"""Set up test data for CLI testing."""
|
|
# Create costs and period
|
|
cost_manager = CostItemManager(temp_db)
|
|
category_id = create_unique_category(cost_manager, "Test", "Test category")
|
|
create_test_cost_item(
|
|
cost_manager, "Test Cost", category_id, "one_time",
|
|
Decimal('45.00'), date(2025, 10, 15)
|
|
)
|
|
|
|
period_manager = PeriodManager(temp_db)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31)
|
|
)
|
|
|
|
# Create issue activities
|
|
activity_tracker = IssueActivityTracker(temp_db)
|
|
activity_tracker.log_activity(
|
|
issue_id=501,
|
|
activity_type=ActivityType.CREATED,
|
|
activity_date=date(2025, 10, 5),
|
|
period_id=period_id
|
|
)
|
|
|
|
activity_tracker.log_activity(
|
|
issue_id=502,
|
|
activity_type=ActivityType.MODIFIED,
|
|
activity_date=date(2025, 10, 15),
|
|
period_id=period_id
|
|
)
|
|
|
|
activity_tracker.log_activity(
|
|
issue_id=503,
|
|
activity_type=ActivityType.COMMENTED,
|
|
activity_date=date(2025, 10, 25),
|
|
period_id=period_id
|
|
)
|
|
|
|
return period_id
|
|
|
|
def test_cli_allocation_period_command(self, temp_db, setup_test_data):
|
|
"""Test CLI allocation period command."""
|
|
from markitect.finance.cli import allocate_period
|
|
from click.testing import CliRunner
|
|
|
|
runner = CliRunner()
|
|
|
|
# Mock configuration manager
|
|
with patch('markitect.finance.cli.ConfigurationManager') as mock_config_manager:
|
|
mock_config = Mock()
|
|
mock_config.get_current_config.return_value = {'database_path': temp_db}
|
|
mock_config_manager.return_value = mock_config
|
|
|
|
result = runner.invoke(allocate_period, [str(setup_test_data)])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Cost Allocation Complete" in result.output
|
|
assert "Total Costs Allocated: €45.00" in result.output
|
|
assert "Active Issues: 3" in result.output
|
|
assert "Cost Per Issue: €15.00" in result.output
|
|
|
|
def test_cli_show_allocations_command(self, temp_db, setup_test_data):
|
|
"""Test CLI show allocations command."""
|
|
from markitect.finance.cli import show_allocations
|
|
from click.testing import CliRunner
|
|
|
|
# First perform allocation
|
|
allocation_engine = AllocationEngine(temp_db)
|
|
allocation_engine.allocate_period_costs(setup_test_data)
|
|
|
|
runner = CliRunner()
|
|
|
|
# Mock configuration manager
|
|
with patch('markitect.finance.cli.ConfigurationManager') as mock_config_manager:
|
|
mock_config = Mock()
|
|
mock_config.get_current_config.return_value = {'database_path': temp_db}
|
|
mock_config_manager.return_value = mock_config
|
|
|
|
# Test issue allocations
|
|
result = runner.invoke(show_allocations, ['issue:501'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Cost Allocations for Issue #501" in result.output
|
|
assert "€15.00" in result.output
|
|
|
|
# Test period allocations
|
|
result = runner.invoke(show_allocations, [f'period:{setup_test_data}'])
|
|
|
|
assert result.exit_code == 0
|
|
assert f"Cost Allocations for Period {setup_test_data}" in result.output
|
|
assert "Total: 3 allocations" in result.output
|
|
|
|
|
|
class TestAllocationEdgeCases:
|
|
"""Test edge cases and error conditions."""
|
|
|
|
@pytest.fixture
|
|
def temp_db(self):
|
|
"""Create temporary database for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
|
|
# Initialize schema
|
|
finance_models = FinanceModels(db_path)
|
|
finance_models.initialize_finance_schema()
|
|
|
|
yield db_path
|
|
|
|
# Cleanup
|
|
Path(db_path).unlink(missing_ok=True)
|
|
|
|
def test_allocation_with_very_small_amounts(self, temp_db):
|
|
"""Test allocation with very small cost amounts."""
|
|
# Create tiny cost
|
|
cost_manager = CostItemManager(temp_db)
|
|
category_id = cost_manager.create_category("Test", "Test")
|
|
create_test_cost_item(
|
|
cost_manager, "Tiny Cost", category_id, "one_time",
|
|
Decimal('0.01'), date(2025, 10, 15) # 1 cent
|
|
)
|
|
|
|
# Create period and activities
|
|
period_manager = PeriodManager(temp_db)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31)
|
|
)
|
|
|
|
activity_tracker = IssueActivityTracker(temp_db)
|
|
activity_tracker.log_activity(
|
|
issue_id=601,
|
|
activity_type=ActivityType.CREATED,
|
|
period_id=period_id
|
|
)
|
|
|
|
activity_tracker.log_activity(
|
|
issue_id=602,
|
|
activity_type=ActivityType.MODIFIED,
|
|
period_id=period_id
|
|
)
|
|
|
|
activity_tracker.log_activity(
|
|
issue_id=603,
|
|
activity_type=ActivityType.COMMENTED,
|
|
period_id=period_id
|
|
)
|
|
|
|
# Perform allocation
|
|
allocation_engine = AllocationEngine(temp_db)
|
|
result = allocation_engine.allocate_period_costs(period_id)
|
|
|
|
assert result.status == AllocationStatus.SUCCESS
|
|
assert result.total_costs == Decimal('0.01')
|
|
# With 3 issues, each gets 0.01/3 = 0.0033... which should be handled properly
|
|
expected_per_issue = Decimal('0.01') / 3
|
|
assert abs(result.cost_per_issue - expected_per_issue) < Decimal('0.0001')
|
|
|
|
def test_allocation_with_duplicate_activities_same_issue(self, temp_db):
|
|
"""Test that duplicate activities for same issue don't create multiple allocations."""
|
|
# Create cost and period
|
|
cost_manager = CostItemManager(temp_db)
|
|
category_id = cost_manager.create_category("Test", "Test")
|
|
create_test_cost_item(
|
|
cost_manager, "Test Cost", category_id, "one_time",
|
|
Decimal('30.00'), date(2025, 10, 15)
|
|
)
|
|
|
|
period_manager = PeriodManager(temp_db)
|
|
period_id = period_manager.create_period(
|
|
period_start=date(2025, 10, 1),
|
|
period_end=date(2025, 10, 31)
|
|
)
|
|
|
|
# Create multiple activities for same issue
|
|
activity_tracker = IssueActivityTracker(temp_db)
|
|
activity_tracker.log_activity(
|
|
issue_id=701,
|
|
activity_type=ActivityType.CREATED,
|
|
activity_date=date(2025, 10, 5),
|
|
period_id=period_id
|
|
)
|
|
|
|
activity_tracker.log_activity(
|
|
issue_id=701, # Same issue
|
|
activity_type=ActivityType.MODIFIED,
|
|
activity_date=date(2025, 10, 10),
|
|
period_id=period_id
|
|
)
|
|
|
|
activity_tracker.log_activity(
|
|
issue_id=701, # Same issue again
|
|
activity_type=ActivityType.COMMENTED,
|
|
activity_date=date(2025, 10, 15),
|
|
period_id=period_id
|
|
)
|
|
|
|
# Perform allocation
|
|
allocation_engine = AllocationEngine(temp_db)
|
|
result = allocation_engine.allocate_period_costs(period_id)
|
|
|
|
assert result.status == AllocationStatus.SUCCESS
|
|
assert len(result.active_issues) == 1 # Only one unique issue
|
|
assert result.active_issues[0] == 701
|
|
assert result.cost_per_issue == Decimal('30.00') # Full amount to single issue
|
|
assert result.allocations_created == 1 # Only one allocation
|
|
|
|
def test_database_constraint_violations(self, temp_db):
|
|
"""Test handling of database constraint violations."""
|
|
allocation_engine = AllocationEngine(temp_db)
|
|
|
|
# Try to create duplicate allocation manually
|
|
with allocation_engine.finance_models.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# Create period first
|
|
cursor.execute('''
|
|
INSERT INTO cost_periods (period_start, period_end)
|
|
VALUES ('2025-10-01', '2025-10-31')
|
|
''')
|
|
period_id = cursor.lastrowid
|
|
|
|
# Create first allocation
|
|
cursor.execute('''
|
|
INSERT INTO issue_cost_allocations
|
|
(issue_id, period_id, allocated_amount, allocation_date)
|
|
VALUES (801, ?, 10.00, '2025-10-15')
|
|
''', (period_id,))
|
|
|
|
# Try to create duplicate (should fail due to unique constraint)
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
cursor.execute('''
|
|
INSERT INTO issue_cost_allocations
|
|
(issue_id, period_id, allocated_amount, allocation_date)
|
|
VALUES (801, ?, 20.00, '2025-10-16')
|
|
''', (period_id,))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__]) |