- Add comprehensive WorktimeTracker service with worktime estimation and cost distribution - Implement full CLI interface with log, list, daily, estimate, distribute, report, delete, update commands - Support flexible duration parsing (90, 1h30m, 2.5h) and time tracking with start/end times - Add worktime estimation with equal and activity-based distribution methods - Implement proportional cost distribution based on actual time spent on issues - Create worktime database schema with entries, summaries, and cost distribution logging - Add 24 comprehensive test cases covering all functionality with integration tests - Support multiple output formats (table/JSON) and comprehensive reporting features - Enable precise cost allocation per minute with audit trail for financial tracking 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
794 lines
30 KiB
Python
794 lines
30 KiB
Python
"""
|
|
Tests for Issue #122 - Daily worktime estimation and distribution of associated cost
|
|
|
|
This module contains comprehensive tests for the worktime tracking system
|
|
that estimates daily work time and distributes costs proportionally based
|
|
on time allocation across issues.
|
|
"""
|
|
|
|
import pytest
|
|
import sqlite3
|
|
from datetime import datetime, date, timedelta
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from pathlib import Path
|
|
import tempfile
|
|
import json
|
|
from decimal import Decimal
|
|
|
|
from markitect.finance.worktime_tracker import WorktimeTracker, WorktimeEntry, DailySummary
|
|
from markitect.finance.worktime_commands import worktime, _parse_duration, _format_duration
|
|
|
|
|
|
class TestWorktimeEntry:
|
|
"""Test suite for WorktimeEntry dataclass."""
|
|
|
|
def test_worktime_entry_creation(self):
|
|
"""Test that WorktimeEntry objects can be created properly."""
|
|
entry = WorktimeEntry(
|
|
id=1,
|
|
issue_id=122,
|
|
work_date=date.today(),
|
|
duration_minutes=90,
|
|
description="Working on worktime tracking"
|
|
)
|
|
|
|
assert entry.id == 1
|
|
assert entry.issue_id == 122
|
|
assert entry.work_date == date.today()
|
|
assert entry.duration_minutes == 90
|
|
assert entry.description == "Working on worktime tracking"
|
|
|
|
def test_worktime_entry_defaults(self):
|
|
"""Test that WorktimeEntry has proper default values."""
|
|
entry = WorktimeEntry()
|
|
|
|
assert entry.id is None
|
|
assert entry.issue_id is None
|
|
assert entry.work_date is None
|
|
assert entry.start_time is None
|
|
assert entry.end_time is None
|
|
assert entry.duration_minutes is None
|
|
assert entry.description is None
|
|
assert entry.entry_type == "manual"
|
|
assert entry.created_at is None
|
|
assert entry.updated_at is None
|
|
|
|
|
|
class TestDailySummary:
|
|
"""Test suite for DailySummary dataclass."""
|
|
|
|
def test_daily_summary_creation(self):
|
|
"""Test that DailySummary objects can be created properly."""
|
|
entries = [
|
|
WorktimeEntry(id=1, issue_id=122, duration_minutes=90),
|
|
WorktimeEntry(id=2, issue_id=123, duration_minutes=60)
|
|
]
|
|
|
|
summary = DailySummary(
|
|
work_date=date.today(),
|
|
total_minutes=150,
|
|
issue_count=2,
|
|
entries=entries,
|
|
cost_per_minute=Decimal('0.1'),
|
|
total_cost_allocated=Decimal('15.0')
|
|
)
|
|
|
|
assert summary.work_date == date.today()
|
|
assert summary.total_minutes == 150
|
|
assert summary.issue_count == 2
|
|
assert len(summary.entries) == 2
|
|
assert summary.cost_per_minute == Decimal('0.1')
|
|
assert summary.total_cost_allocated == Decimal('15.0')
|
|
|
|
|
|
class TestWorktimeTracker:
|
|
"""Test suite for WorktimeTracker service."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test fixtures with temporary database."""
|
|
self.temp_db = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
|
self.temp_db.close()
|
|
self.db_path = self.temp_db.name
|
|
self.tracker = WorktimeTracker(self.db_path)
|
|
|
|
def teardown_method(self):
|
|
"""Clean up test fixtures."""
|
|
Path(self.db_path).unlink(missing_ok=True)
|
|
|
|
def test_tracker_initialization(self):
|
|
"""Test that tracker initializes properly with database."""
|
|
assert self.tracker.db_path == self.db_path
|
|
assert self.tracker.finance_models is not None
|
|
|
|
# Verify worktime tables were created
|
|
with self.tracker.finance_models.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
|
|
tables = [row[0] for row in cursor.fetchall()]
|
|
|
|
expected_tables = ['worktime_entries', 'daily_worktime_summaries', 'worktime_cost_distributions']
|
|
for table in expected_tables:
|
|
assert table in tables
|
|
|
|
def test_log_worktime_basic(self):
|
|
"""Test logging basic worktime entry."""
|
|
entry_id = self.tracker.log_worktime(
|
|
issue_id=122,
|
|
duration_minutes=90,
|
|
description="Implementing worktime tracking"
|
|
)
|
|
|
|
assert entry_id is not None
|
|
|
|
# Verify entry was stored
|
|
entries = self.tracker.get_worktime_entries(issue_id=122)
|
|
assert len(entries) == 1
|
|
assert entries[0].issue_id == 122
|
|
assert entries[0].duration_minutes == 90
|
|
assert entries[0].description == "Implementing worktime tracking"
|
|
|
|
def test_log_worktime_with_timestamps(self):
|
|
"""Test logging worktime with start and end times."""
|
|
now = datetime.now()
|
|
start_time = now.replace(hour=9, minute=0, second=0, microsecond=0)
|
|
end_time = now.replace(hour=10, minute=30, second=0, microsecond=0)
|
|
|
|
entry_id = self.tracker.log_worktime(
|
|
issue_id=122,
|
|
duration_minutes=90,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
description="Morning work session"
|
|
)
|
|
|
|
entries = self.tracker.get_worktime_entries(issue_id=122)
|
|
assert len(entries) == 1
|
|
assert entries[0].start_time.hour == 9
|
|
assert entries[0].end_time.hour == 10
|
|
assert entries[0].end_time.minute == 30
|
|
|
|
def test_log_worktime_validation(self):
|
|
"""Test worktime logging validation."""
|
|
# Test negative duration
|
|
with pytest.raises(ValueError, match="Duration must be positive"):
|
|
self.tracker.log_worktime(issue_id=122, duration_minutes=-30)
|
|
|
|
# Test zero duration
|
|
with pytest.raises(ValueError, match="Duration must be positive"):
|
|
self.tracker.log_worktime(issue_id=122, duration_minutes=0)
|
|
|
|
def test_get_worktime_entries_filtering(self):
|
|
"""Test worktime entry retrieval with various filters."""
|
|
today = date.today()
|
|
yesterday = today - timedelta(days=1)
|
|
|
|
# Create test entries
|
|
self.tracker.log_worktime(122, 60, work_date=today, description="Today's work")
|
|
self.tracker.log_worktime(123, 90, work_date=today, description="Today's other work")
|
|
self.tracker.log_worktime(122, 45, work_date=yesterday, description="Yesterday's work")
|
|
|
|
# Test filtering by issue
|
|
issue_122_entries = self.tracker.get_worktime_entries(issue_id=122)
|
|
assert len(issue_122_entries) == 2
|
|
assert all(e.issue_id == 122 for e in issue_122_entries)
|
|
|
|
# Test filtering by date
|
|
today_entries = self.tracker.get_worktime_entries(work_date=today)
|
|
assert len(today_entries) == 2
|
|
assert all(e.work_date == today for e in today_entries)
|
|
|
|
# Test date range filtering
|
|
range_entries = self.tracker.get_worktime_entries(start_date=yesterday, end_date=today)
|
|
assert len(range_entries) == 3
|
|
|
|
def test_get_daily_summary(self):
|
|
"""Test daily worktime summary generation."""
|
|
today = date.today()
|
|
|
|
# Log multiple entries for today
|
|
self.tracker.log_worktime(122, 90, work_date=today)
|
|
self.tracker.log_worktime(123, 60, work_date=today)
|
|
self.tracker.log_worktime(122, 30, work_date=today) # Second entry for same issue
|
|
|
|
summary = self.tracker.get_daily_summary(today)
|
|
|
|
assert summary is not None
|
|
assert summary.work_date == today
|
|
assert summary.total_minutes == 180 # 90 + 60 + 30
|
|
assert summary.issue_count == 2 # Issues 122 and 123
|
|
assert len(summary.entries) == 3
|
|
|
|
def test_estimate_daily_worktime_equal_distribution(self):
|
|
"""Test daily worktime estimation with equal distribution."""
|
|
today = date.today()
|
|
issues = [122, 123, 124]
|
|
|
|
result = self.tracker.estimate_daily_worktime(
|
|
work_date=today,
|
|
total_hours=6.0,
|
|
issues=issues,
|
|
distribution_method="equal"
|
|
)
|
|
|
|
assert result['work_date'] == today
|
|
assert result['total_minutes'] == 360 # 6 hours
|
|
assert result['distribution_method'] == "equal"
|
|
assert result['issues_count'] == 3
|
|
|
|
# Each issue should get 120 minutes (2 hours)
|
|
for issue_id in issues:
|
|
assert result['issue_estimates'][issue_id] == 120
|
|
|
|
# Verify entries were created
|
|
entries = self.tracker.get_worktime_entries(work_date=today)
|
|
assert len(entries) == 3
|
|
assert all(e.entry_type == "estimated" for e in entries)
|
|
|
|
def test_estimate_daily_worktime_activity_based(self):
|
|
"""Test daily worktime estimation with activity-based distribution."""
|
|
today = date.today()
|
|
|
|
# Mock activity data - issue 122 has more activities
|
|
with patch.object(self.tracker, '_get_activity_weights_for_date') as mock_weights:
|
|
mock_weights.return_value = {122: 5, 123: 2, 124: 1} # Different activity levels
|
|
|
|
result = self.tracker.estimate_daily_worktime(
|
|
work_date=today,
|
|
total_hours=8.0,
|
|
issues=[122, 123, 124],
|
|
distribution_method="activity_based"
|
|
)
|
|
|
|
# Verify distribution is proportional to activities
|
|
total_weight = 5 + 2 + 1 # 8
|
|
expected_122 = int((5/8) * 480) # 300 minutes
|
|
expected_123 = int((2/8) * 480) # 120 minutes
|
|
expected_124 = int((1/8) * 480) # 60 minutes
|
|
|
|
assert result['issue_estimates'][122] == expected_122
|
|
assert result['issue_estimates'][123] == expected_123
|
|
assert result['issue_estimates'][124] == expected_124
|
|
|
|
def test_distribute_daily_costs(self):
|
|
"""Test daily cost distribution based on time allocation."""
|
|
today = date.today()
|
|
|
|
# Log different amounts of time for different issues
|
|
self.tracker.log_worktime(122, 120, work_date=today) # 2 hours
|
|
self.tracker.log_worktime(123, 60, work_date=today) # 1 hour
|
|
self.tracker.log_worktime(124, 120, work_date=today) # 2 hours
|
|
# Total: 5 hours (300 minutes)
|
|
|
|
total_cost = Decimal('150.00') # €150 for the day
|
|
result = self.tracker.distribute_daily_costs(
|
|
work_date=today,
|
|
total_daily_cost=total_cost
|
|
)
|
|
|
|
assert result['work_date'] == today
|
|
assert result['total_cost'] == 150.0
|
|
assert result['total_minutes'] == 300
|
|
assert result['cost_per_minute'] == 0.5 # €150 / 300 minutes
|
|
|
|
# Check cost distribution
|
|
assert result['distributions'][122]['cost_allocated'] == 60.0 # 120 min * €0.5
|
|
assert result['distributions'][123]['cost_allocated'] == 30.0 # 60 min * €0.5
|
|
assert result['distributions'][124]['cost_allocated'] == 60.0 # 120 min * €0.5
|
|
|
|
# Check percentages
|
|
assert result['distributions'][122]['percentage'] == 40.0 # 120/300 * 100
|
|
assert result['distributions'][123]['percentage'] == 20.0 # 60/300 * 100
|
|
assert result['distributions'][124]['percentage'] == 40.0 # 120/300 * 100
|
|
|
|
def test_distribute_daily_costs_no_worktime(self):
|
|
"""Test cost distribution when no worktime is logged."""
|
|
today = date.today()
|
|
total_cost = Decimal('100.00')
|
|
|
|
result = self.tracker.distribute_daily_costs(
|
|
work_date=today,
|
|
total_daily_cost=total_cost
|
|
)
|
|
|
|
assert 'message' in result
|
|
assert "No worktime entries found" in result['message']
|
|
|
|
def test_get_worktime_report(self):
|
|
"""Test comprehensive worktime reporting."""
|
|
today = date.today()
|
|
yesterday = today - timedelta(days=1)
|
|
|
|
# Create test data across multiple days and issues
|
|
self.tracker.log_worktime(122, 90, work_date=yesterday)
|
|
self.tracker.log_worktime(123, 60, work_date=yesterday)
|
|
self.tracker.log_worktime(122, 120, work_date=today)
|
|
self.tracker.log_worktime(124, 45, work_date=today)
|
|
|
|
report = self.tracker.get_worktime_report(
|
|
start_date=yesterday,
|
|
end_date=today
|
|
)
|
|
|
|
assert report['total_entries'] == 4
|
|
assert report['total_time']['total_minutes'] == 315 # 90+60+120+45
|
|
assert report['total_time']['hours'] == 5
|
|
assert report['total_time']['minutes'] == 15
|
|
assert report['unique_issues'] == 3 # Issues 122, 123, 124
|
|
assert report['unique_dates'] == 2
|
|
|
|
# Check issue breakdown
|
|
assert 122 in report['issue_breakdown']
|
|
assert report['issue_breakdown'][122]['total_minutes'] == 210 # 90+120
|
|
assert report['issue_breakdown'][122]['entry_count'] == 2
|
|
assert report['issue_breakdown'][122]['unique_dates'] == 2
|
|
|
|
def test_delete_worktime_entry(self):
|
|
"""Test deleting worktime entries."""
|
|
entry_id = self.tracker.log_worktime(122, 90, description="Test entry")
|
|
|
|
# Verify entry exists
|
|
entries = self.tracker.get_worktime_entries(issue_id=122)
|
|
assert len(entries) == 1
|
|
|
|
# Delete entry
|
|
success = self.tracker.delete_worktime_entry(entry_id)
|
|
assert success is True
|
|
|
|
# Verify entry is gone
|
|
entries = self.tracker.get_worktime_entries(issue_id=122)
|
|
assert len(entries) == 0
|
|
|
|
# Try to delete non-existent entry
|
|
success = self.tracker.delete_worktime_entry(99999)
|
|
assert success is False
|
|
|
|
def test_update_worktime_entry(self):
|
|
"""Test updating worktime entries."""
|
|
entry_id = self.tracker.log_worktime(122, 90, description="Original description")
|
|
|
|
# Update duration and description
|
|
success = self.tracker.update_worktime_entry(
|
|
entry_id=entry_id,
|
|
duration_minutes=120,
|
|
description="Updated description"
|
|
)
|
|
assert success is True
|
|
|
|
# Verify updates
|
|
entries = self.tracker.get_worktime_entries(issue_id=122)
|
|
assert len(entries) == 1
|
|
assert entries[0].duration_minutes == 120
|
|
assert entries[0].description == "Updated description"
|
|
|
|
# Try to update non-existent entry
|
|
success = self.tracker.update_worktime_entry(
|
|
entry_id=99999,
|
|
duration_minutes=60
|
|
)
|
|
assert success is False
|
|
|
|
|
|
class TestWorktimeCommands:
|
|
"""Test suite for worktime CLI commands."""
|
|
|
|
def test_parse_duration_minutes(self):
|
|
"""Test parsing duration strings - minutes format."""
|
|
assert _parse_duration("90") == 90
|
|
assert _parse_duration("120") == 120
|
|
assert _parse_duration("45m") == 45
|
|
|
|
def test_parse_duration_hours(self):
|
|
"""Test parsing duration strings - hours format."""
|
|
assert _parse_duration("1h") == 60
|
|
assert _parse_duration("2h") == 120
|
|
assert _parse_duration("1.5h") == 90
|
|
assert _parse_duration("2.25h") == 135
|
|
|
|
def test_parse_duration_hours_minutes(self):
|
|
"""Test parsing duration strings - hours and minutes format."""
|
|
assert _parse_duration("1h30m") == 90
|
|
assert _parse_duration("2h15m") == 135
|
|
assert _parse_duration("0h45m") == 45
|
|
assert _parse_duration("3h0m") == 180
|
|
|
|
def test_parse_duration_invalid(self):
|
|
"""Test parsing invalid duration strings."""
|
|
with pytest.raises(ValueError):
|
|
_parse_duration("invalid")
|
|
|
|
with pytest.raises(ValueError):
|
|
_parse_duration("1x30m")
|
|
|
|
with pytest.raises(ValueError):
|
|
_parse_duration("")
|
|
|
|
def test_format_duration_minutes_only(self):
|
|
"""Test formatting duration - minutes only."""
|
|
assert _format_duration(30) == "30m"
|
|
assert _format_duration(45) == "45m"
|
|
assert _format_duration(59) == "59m"
|
|
|
|
def test_format_duration_hours_only(self):
|
|
"""Test formatting duration - hours only."""
|
|
assert _format_duration(60) == "1h"
|
|
assert _format_duration(120) == "2h"
|
|
assert _format_duration(180) == "3h"
|
|
|
|
def test_format_duration_hours_and_minutes(self):
|
|
"""Test formatting duration - hours and minutes."""
|
|
assert _format_duration(90) == "1h30m"
|
|
assert _format_duration(135) == "2h15m"
|
|
assert _format_duration(195) == "3h15m"
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_log_command_basic(self, mock_tracker_class):
|
|
"""Test the log command with basic parameters."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
mock_tracker.log_worktime.return_value = 1
|
|
mock_tracker.get_daily_summary.return_value = DailySummary(
|
|
work_date=date.today(),
|
|
total_minutes=90,
|
|
issue_count=1,
|
|
entries=[]
|
|
)
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(worktime, ['log', '122', '1h30m'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "✅ Logged 90min worktime for issue #122" in result.output
|
|
mock_tracker.log_worktime.assert_called_once()
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_log_command_with_description(self, mock_tracker_class):
|
|
"""Test the log command with description."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
mock_tracker.log_worktime.return_value = 1
|
|
mock_tracker.get_daily_summary.return_value = None
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(worktime, ['log', '122', '90', '--description', 'Testing worktime'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Testing worktime" in result.output
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_list_command_table_format(self, mock_tracker_class):
|
|
"""Test the list command with table output format."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
|
|
mock_entries = [
|
|
WorktimeEntry(
|
|
id=1,
|
|
issue_id=122,
|
|
work_date=date.today(),
|
|
duration_minutes=90,
|
|
description="Test worktime",
|
|
entry_type="manual"
|
|
)
|
|
]
|
|
mock_tracker.get_worktime_entries.return_value = mock_entries
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(worktime, ['list'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "⏰ Worktime Entries" in result.output
|
|
assert "#122" in result.output
|
|
assert "1h30m" in result.output
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_list_command_json_format(self, mock_tracker_class):
|
|
"""Test the list command with JSON output format."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
|
|
mock_entries = [
|
|
WorktimeEntry(
|
|
id=1,
|
|
issue_id=122,
|
|
work_date=date.today(),
|
|
duration_minutes=90,
|
|
description="Test worktime",
|
|
entry_type="manual"
|
|
)
|
|
]
|
|
mock_tracker.get_worktime_entries.return_value = mock_entries
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(worktime, ['list', '--format', 'json'])
|
|
|
|
assert result.exit_code == 0
|
|
# Should be valid JSON
|
|
output_data = json.loads(result.output.strip())
|
|
assert len(output_data) == 1
|
|
assert output_data[0]['issue_id'] == 122
|
|
assert output_data[0]['duration_minutes'] == 90
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_daily_command(self, mock_tracker_class):
|
|
"""Test the daily summary command."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
|
|
mock_entries = [
|
|
WorktimeEntry(id=1, issue_id=122, duration_minutes=90, entry_type="manual"),
|
|
WorktimeEntry(id=2, issue_id=123, duration_minutes=60, entry_type="manual")
|
|
]
|
|
|
|
mock_summary = DailySummary(
|
|
work_date=date.today(),
|
|
total_minutes=150,
|
|
issue_count=2,
|
|
entries=mock_entries,
|
|
cost_per_minute=Decimal('0.5'),
|
|
total_cost_allocated=Decimal('75.0')
|
|
)
|
|
mock_tracker.get_daily_summary.return_value = mock_summary
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
today = date.today().strftime('%Y-%m-%d')
|
|
result = runner.invoke(worktime, ['daily', today])
|
|
|
|
assert result.exit_code == 0
|
|
assert f"📅 Daily Summary for {date.today()}" in result.output
|
|
assert "Total Time: 2h30m" in result.output
|
|
assert "Issues Worked: 2" in result.output
|
|
assert "Cost per Minute: €0.5000" in result.output
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_estimate_command(self, mock_tracker_class):
|
|
"""Test the estimate worktime command."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
|
|
mock_result = {
|
|
'work_date': date.today(),
|
|
'total_minutes': 480, # 8 hours
|
|
'distribution_method': 'equal',
|
|
'issue_estimates': {122: 240, 123: 240},
|
|
'issues_count': 2
|
|
}
|
|
mock_tracker.estimate_daily_worktime.return_value = mock_result
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
today = date.today().strftime('%Y-%m-%d')
|
|
result = runner.invoke(worktime, ['estimate', today, '8', '-i', '122', '-i', '123'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "📊 Worktime Estimation" in result.output
|
|
assert "Total Hours: 8.0h" in result.output
|
|
assert "✅ Created 2 estimated worktime entries" in result.output
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_distribute_command(self, mock_tracker_class):
|
|
"""Test the cost distribution command."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
|
|
mock_result = {
|
|
'work_date': date.today(),
|
|
'total_cost': 100.0,
|
|
'total_minutes': 200,
|
|
'cost_per_minute': 0.5,
|
|
'distributions': {
|
|
122: {'minutes': 120, 'percentage': 60.0, 'cost_allocated': 60.0},
|
|
123: {'minutes': 80, 'percentage': 40.0, 'cost_allocated': 40.0}
|
|
},
|
|
'issues_count': 2
|
|
}
|
|
mock_tracker.distribute_daily_costs.return_value = mock_result
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
today = date.today().strftime('%Y-%m-%d')
|
|
result = runner.invoke(worktime, ['distribute', today, '100'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "💰 Cost Distribution" in result.output
|
|
assert "Total Cost: €100.00" in result.output
|
|
assert "Cost per Minute: €0.5000" in result.output
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_delete_command(self, mock_tracker_class):
|
|
"""Test the delete command."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
mock_tracker.delete_worktime_entry.return_value = True
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
# Auto-confirm the deletion
|
|
result = runner.invoke(worktime, ['delete', '1'], input='y\n')
|
|
|
|
assert result.exit_code == 0
|
|
assert "✅ Deleted worktime entry #1" in result.output
|
|
mock_tracker.delete_worktime_entry.assert_called_once_with(1)
|
|
|
|
@patch('markitect.finance.worktime_commands.WorktimeTracker')
|
|
def test_update_command(self, mock_tracker_class):
|
|
"""Test the update command."""
|
|
mock_tracker = Mock()
|
|
mock_tracker_class.return_value = mock_tracker
|
|
mock_tracker.update_worktime_entry.return_value = True
|
|
|
|
from click.testing import CliRunner
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(worktime, ['update', '1', '--duration', '2h', '--description', 'Updated'])
|
|
|
|
assert result.exit_code == 0
|
|
assert "✅ Updated worktime entry #1" in result.output
|
|
|
|
|
|
class TestWorktimeIntegration:
|
|
"""Integration tests for the complete worktime tracking system."""
|
|
|
|
def setup_method(self):
|
|
"""Set up integration test fixtures."""
|
|
self.temp_db = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
|
self.temp_db.close()
|
|
self.db_path = self.temp_db.name
|
|
|
|
def teardown_method(self):
|
|
"""Clean up integration test fixtures."""
|
|
Path(self.db_path).unlink(missing_ok=True)
|
|
|
|
def test_full_worktime_lifecycle(self):
|
|
"""Test the complete lifecycle of worktime tracking."""
|
|
tracker = WorktimeTracker(self.db_path)
|
|
|
|
# 1. Log worktime for multiple issues across multiple days
|
|
today = date.today()
|
|
yesterday = today - timedelta(days=1)
|
|
|
|
tracker.log_worktime(122, 120, work_date=yesterday, description="Initial development")
|
|
tracker.log_worktime(123, 90, work_date=yesterday, description="Code review")
|
|
|
|
tracker.log_worktime(122, 90, work_date=today, description="Bug fixes")
|
|
tracker.log_worktime(124, 60, work_date=today, description="Documentation")
|
|
|
|
# 2. Verify daily summaries
|
|
yesterday_summary = tracker.get_daily_summary(yesterday)
|
|
assert yesterday_summary.total_minutes == 210 # 120 + 90
|
|
assert yesterday_summary.issue_count == 2
|
|
|
|
today_summary = tracker.get_daily_summary(today)
|
|
assert today_summary.total_minutes == 150 # 90 + 60
|
|
assert today_summary.issue_count == 2
|
|
|
|
# 3. Distribute costs for a day
|
|
distribution = tracker.distribute_daily_costs(
|
|
work_date=today,
|
|
total_daily_cost=Decimal('75.00') # €75 for today's work
|
|
)
|
|
|
|
assert distribution['total_cost'] == 75.0
|
|
assert distribution['total_minutes'] == 150
|
|
assert distribution['cost_per_minute'] == 0.5
|
|
|
|
# Issue 122: 90 minutes = €45
|
|
# Issue 124: 60 minutes = €30
|
|
assert distribution['distributions'][122]['cost_allocated'] == 45.0
|
|
assert distribution['distributions'][124]['cost_allocated'] == 30.0
|
|
|
|
# 4. Generate comprehensive report
|
|
report = tracker.get_worktime_report(
|
|
start_date=yesterday,
|
|
end_date=today
|
|
)
|
|
|
|
assert report['total_entries'] == 4
|
|
assert report['total_time']['total_minutes'] == 360 # 210 + 150
|
|
assert report['unique_issues'] == 3 # Issues 122, 123, 124
|
|
assert report['unique_dates'] == 2
|
|
|
|
# 5. Test estimation functionality
|
|
tomorrow = today + timedelta(days=1)
|
|
estimation = tracker.estimate_daily_worktime(
|
|
work_date=tomorrow,
|
|
total_hours=6.0,
|
|
issues=[122, 125, 126],
|
|
distribution_method="equal"
|
|
)
|
|
|
|
assert estimation['total_minutes'] == 360
|
|
assert len(estimation['issue_estimates']) == 3
|
|
# Each issue should get 120 minutes (equal distribution)
|
|
for minutes in estimation['issue_estimates'].values():
|
|
assert minutes == 120
|
|
|
|
# 6. Verify estimated entries were created
|
|
tomorrow_entries = tracker.get_worktime_entries(work_date=tomorrow)
|
|
assert len(tomorrow_entries) == 3
|
|
assert all(e.entry_type == "estimated" for e in tomorrow_entries)
|
|
|
|
def test_cost_distribution_accuracy(self):
|
|
"""Test accurate cost distribution calculations."""
|
|
tracker = WorktimeTracker(self.db_path)
|
|
work_date = date.today()
|
|
|
|
# Log precise worktime amounts
|
|
tracker.log_worktime(122, 100, work_date=work_date) # 100 minutes
|
|
tracker.log_worktime(123, 50, work_date=work_date) # 50 minutes
|
|
tracker.log_worktime(124, 150, work_date=work_date) # 150 minutes
|
|
# Total: 300 minutes
|
|
|
|
# Distribute exactly €300
|
|
distribution = tracker.distribute_daily_costs(
|
|
work_date=work_date,
|
|
total_daily_cost=Decimal('300.00')
|
|
)
|
|
|
|
# Should be exactly €1 per minute
|
|
assert distribution['cost_per_minute'] == 1.0
|
|
|
|
# Verify exact cost allocation
|
|
assert distribution['distributions'][122]['cost_allocated'] == 100.0
|
|
assert distribution['distributions'][123]['cost_allocated'] == 50.0
|
|
assert distribution['distributions'][124]['cost_allocated'] == 150.0
|
|
|
|
# Verify percentages sum to 100%
|
|
total_percentage = sum(
|
|
dist['percentage'] for dist in distribution['distributions'].values()
|
|
)
|
|
assert abs(total_percentage - 100.0) < 0.01 # Allow for rounding
|
|
|
|
# Verify cost allocation was logged to database
|
|
with tracker.finance_models.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT issue_id, cost_allocated
|
|
FROM worktime_cost_distributions
|
|
WHERE work_date = ?
|
|
ORDER BY issue_id
|
|
''', (work_date,))
|
|
results = cursor.fetchall()
|
|
|
|
assert len(results) == 3
|
|
assert results[0] == (122, 100.0)
|
|
assert results[1] == (123, 50.0)
|
|
assert results[2] == (124, 150.0)
|
|
|
|
def test_worktime_modification_and_summary_updates(self):
|
|
"""Test that modifying worktime entries correctly updates summaries."""
|
|
tracker = WorktimeTracker(self.db_path)
|
|
work_date = date.today()
|
|
|
|
# Log initial worktime
|
|
entry_id = tracker.log_worktime(122, 60, work_date=work_date)
|
|
|
|
# Check initial summary
|
|
summary = tracker.get_daily_summary(work_date)
|
|
assert summary.total_minutes == 60
|
|
|
|
# Update the entry
|
|
tracker.update_worktime_entry(entry_id, duration_minutes=120)
|
|
|
|
# Check updated summary
|
|
summary = tracker.get_daily_summary(work_date)
|
|
assert summary.total_minutes == 120
|
|
|
|
# Delete the entry
|
|
tracker.delete_worktime_entry(entry_id)
|
|
|
|
# Check final summary
|
|
summary = tracker.get_daily_summary(work_date)
|
|
assert summary is None or summary.total_minutes == 0 |