feat: implement daily worktime tracking and cost distribution system (issue #122)
- 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>
This commit is contained in:
@@ -6382,6 +6382,10 @@ cli.add_command(issues_group)
|
||||
from markitect.issues.activity_commands import activity as activity_group
|
||||
cli.add_command(activity_group)
|
||||
|
||||
# Register worktime tracking commands
|
||||
from markitect.finance.worktime_commands import worktime as worktime_group
|
||||
cli.add_command(worktime_group)
|
||||
|
||||
|
||||
# Query Paradigm Commands - Issue #62
|
||||
@click.group()
|
||||
|
||||
505
markitect/finance/worktime_commands.py
Normal file
505
markitect/finance/worktime_commands.py
Normal file
@@ -0,0 +1,505 @@
|
||||
"""
|
||||
CLI commands for worktime tracking and cost distribution.
|
||||
|
||||
This module provides command-line interface for logging, viewing, and managing
|
||||
worktime entries, estimating daily effort, and distributing costs based on
|
||||
time allocation across issues.
|
||||
"""
|
||||
|
||||
import click
|
||||
from datetime import datetime, date, timedelta
|
||||
from typing import List, Optional
|
||||
from tabulate import tabulate
|
||||
import json
|
||||
|
||||
from .worktime_tracker import WorktimeTracker, WorktimeEntry
|
||||
|
||||
|
||||
@click.group()
|
||||
def worktime():
|
||||
"""Worktime tracking and cost distribution commands."""
|
||||
pass
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.argument('issue_id', type=int)
|
||||
@click.argument('duration', type=str)
|
||||
@click.option('--date', '-d', type=click.DateTime(formats=['%Y-%m-%d']),
|
||||
help='Work date (defaults to today)')
|
||||
@click.option('--start', '-s', type=click.DateTime(formats=['%H:%M', '%Y-%m-%d %H:%M']),
|
||||
help='Start time (e.g., 09:30 or 2025-10-04 09:30)')
|
||||
@click.option('--end', '-e', type=click.DateTime(formats=['%H:%M', '%Y-%m-%d %H:%M']),
|
||||
help='End time (e.g., 11:30 or 2025-10-04 11:30)')
|
||||
@click.option('--description', '-m', help='Work description')
|
||||
@click.option('--type', 'entry_type',
|
||||
type=click.Choice(['manual', 'estimated', 'tracked']),
|
||||
default='manual', help='Entry type')
|
||||
def log(issue_id: int, duration: str, date: Optional[datetime],
|
||||
start: Optional[datetime], end: Optional[datetime],
|
||||
description: Optional[str], entry_type: str):
|
||||
"""Log worktime for an issue.
|
||||
|
||||
DURATION can be:
|
||||
- Minutes: 90, 120
|
||||
- Hours and minutes: 1h30m, 2h15m, 1.5h
|
||||
- Hours only: 1h, 2.5h
|
||||
"""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
# Parse duration
|
||||
try:
|
||||
duration_minutes = _parse_duration(duration)
|
||||
except ValueError as e:
|
||||
click.echo(f"❌ Invalid duration format: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
work_date = date.date() if date else None
|
||||
|
||||
try:
|
||||
entry_id = tracker.log_worktime(
|
||||
issue_id=issue_id,
|
||||
duration_minutes=duration_minutes,
|
||||
work_date=work_date,
|
||||
start_time=start,
|
||||
end_time=end,
|
||||
description=description,
|
||||
entry_type=entry_type
|
||||
)
|
||||
|
||||
click.echo(f"✅ Logged {duration_minutes}min worktime for issue #{issue_id} (ID: {entry_id})")
|
||||
|
||||
if description:
|
||||
click.echo(f" Description: {description}")
|
||||
|
||||
# Show total time for the day
|
||||
summary = tracker.get_daily_summary(work_date or date.today())
|
||||
if summary:
|
||||
hours = summary.total_minutes // 60
|
||||
minutes = summary.total_minutes % 60
|
||||
click.echo(f"📊 Daily total: {hours}h {minutes}m across {summary.issue_count} issues")
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error logging worktime: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.option('--issue', type=int, help='Filter by specific issue ID')
|
||||
@click.option('--date', type=click.DateTime(formats=['%Y-%m-%d']),
|
||||
help='Filter by specific date')
|
||||
@click.option('--start-date', type=click.DateTime(formats=['%Y-%m-%d']),
|
||||
help='Start date for range filter')
|
||||
@click.option('--end-date', type=click.DateTime(formats=['%Y-%m-%d']),
|
||||
help='End date for range filter')
|
||||
@click.option('--limit', '-l', type=int, default=20,
|
||||
help='Maximum number of entries to show')
|
||||
@click.option('--format', 'output_format', type=click.Choice(['table', 'json']),
|
||||
default='table', help='Output format')
|
||||
def list(issue: Optional[int], date: Optional[datetime],
|
||||
start_date: Optional[datetime], end_date: Optional[datetime],
|
||||
limit: int, output_format: str):
|
||||
"""List worktime entries with optional filtering."""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
try:
|
||||
entries = tracker.get_worktime_entries(
|
||||
issue_id=issue,
|
||||
work_date=date.date() if date else None,
|
||||
start_date=start_date.date() if start_date else None,
|
||||
end_date=end_date.date() if end_date else None,
|
||||
limit=limit
|
||||
)
|
||||
|
||||
if not entries:
|
||||
click.echo("📝 No worktime entries found for the specified criteria")
|
||||
return
|
||||
|
||||
if output_format == 'json':
|
||||
entry_data = []
|
||||
for entry in entries:
|
||||
data = {
|
||||
'id': entry.id,
|
||||
'issue_id': entry.issue_id,
|
||||
'work_date': entry.work_date.isoformat() if entry.work_date else None,
|
||||
'start_time': entry.start_time.isoformat() if entry.start_time else None,
|
||||
'end_time': entry.end_time.isoformat() if entry.end_time else None,
|
||||
'duration_minutes': entry.duration_minutes,
|
||||
'duration_formatted': _format_duration(entry.duration_minutes),
|
||||
'description': entry.description,
|
||||
'entry_type': entry.entry_type,
|
||||
'created_at': entry.created_at.isoformat() if entry.created_at else None
|
||||
}
|
||||
entry_data.append(data)
|
||||
click.echo(json.dumps(entry_data, indent=2))
|
||||
|
||||
else:
|
||||
# Table format
|
||||
click.echo(f"\n⏰ Worktime Entries ({len(entries)} found)\n")
|
||||
|
||||
headers = ['ID', 'Issue', 'Date', 'Duration', 'Start', 'End', 'Type', 'Description']
|
||||
rows = []
|
||||
|
||||
for entry in entries:
|
||||
rows.append([
|
||||
entry.id,
|
||||
f"#{entry.issue_id}",
|
||||
entry.work_date.strftime('%Y-%m-%d') if entry.work_date else 'N/A',
|
||||
_format_duration(entry.duration_minutes),
|
||||
entry.start_time.strftime('%H:%M') if entry.start_time else 'N/A',
|
||||
entry.end_time.strftime('%H:%M') if entry.end_time else 'N/A',
|
||||
entry.entry_type.title(),
|
||||
(entry.description[:30] + '...') if entry.description and len(entry.description) > 30 else (entry.description or '')
|
||||
])
|
||||
|
||||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||||
|
||||
if len(entries) == limit:
|
||||
click.echo(f"\n💡 Showing {limit} most recent entries. Use --limit to see more.")
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error retrieving entries: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.argument('date', type=click.DateTime(formats=['%Y-%m-%d']))
|
||||
@click.option('--format', 'output_format', type=click.Choice(['table', 'json']),
|
||||
default='table', help='Output format')
|
||||
def daily(date: datetime, output_format: str):
|
||||
"""Show daily worktime summary for a specific date."""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
try:
|
||||
summary = tracker.get_daily_summary(date.date())
|
||||
|
||||
if not summary:
|
||||
click.echo(f"📅 No worktime entries found for {date.date()}")
|
||||
return
|
||||
|
||||
if output_format == 'json':
|
||||
data = {
|
||||
'work_date': summary.work_date.isoformat(),
|
||||
'total_minutes': summary.total_minutes,
|
||||
'total_formatted': _format_duration(summary.total_minutes),
|
||||
'issue_count': summary.issue_count,
|
||||
'cost_per_minute': float(summary.cost_per_minute) if summary.cost_per_minute else None,
|
||||
'total_cost_allocated': float(summary.total_cost_allocated) if summary.total_cost_allocated else None,
|
||||
'entries': [
|
||||
{
|
||||
'issue_id': entry.issue_id,
|
||||
'duration_minutes': entry.duration_minutes,
|
||||
'duration_formatted': _format_duration(entry.duration_minutes),
|
||||
'description': entry.description,
|
||||
'entry_type': entry.entry_type
|
||||
}
|
||||
for entry in summary.entries
|
||||
]
|
||||
}
|
||||
click.echo(json.dumps(data, indent=2))
|
||||
|
||||
else:
|
||||
# Table format
|
||||
click.echo(f"\n📅 Daily Summary for {summary.work_date}\n")
|
||||
|
||||
hours = summary.total_minutes // 60
|
||||
minutes = summary.total_minutes % 60
|
||||
click.echo(f"Total Time: {hours}h {minutes}m ({summary.total_minutes} minutes)")
|
||||
click.echo(f"Issues Worked: {summary.issue_count}")
|
||||
|
||||
if summary.cost_per_minute:
|
||||
click.echo(f"Cost per Minute: €{summary.cost_per_minute:.4f}")
|
||||
if summary.total_cost_allocated:
|
||||
click.echo(f"Total Cost Allocated: €{summary.total_cost_allocated:.2f}")
|
||||
|
||||
click.echo("\nBreakdown by Issue:")
|
||||
|
||||
# Group by issue
|
||||
issue_breakdown = {}
|
||||
for entry in summary.entries:
|
||||
if entry.issue_id not in issue_breakdown:
|
||||
issue_breakdown[entry.issue_id] = 0
|
||||
issue_breakdown[entry.issue_id] += entry.duration_minutes
|
||||
|
||||
headers = ['Issue', 'Time', 'Percentage']
|
||||
rows = []
|
||||
|
||||
for issue_id, minutes in sorted(issue_breakdown.items()):
|
||||
percentage = (minutes / summary.total_minutes) * 100
|
||||
rows.append([
|
||||
f"#{issue_id}",
|
||||
_format_duration(minutes),
|
||||
f"{percentage:.1f}%"
|
||||
])
|
||||
|
||||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error generating summary: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.argument('date', type=click.DateTime(formats=['%Y-%m-%d']))
|
||||
@click.argument('hours', type=float)
|
||||
@click.option('--issues', '-i', multiple=True, type=int,
|
||||
help='Issues that were worked on (can specify multiple)')
|
||||
@click.option('--method', type=click.Choice(['equal', 'activity_based']),
|
||||
default='equal', help='Distribution method')
|
||||
def estimate(date: datetime, hours: float, issues: List[int], method: str):
|
||||
"""Estimate worktime distribution for a day."""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
try:
|
||||
result = tracker.estimate_daily_worktime(
|
||||
work_date=date.date(),
|
||||
total_hours=hours,
|
||||
issues=list(issues) if issues else None,
|
||||
distribution_method=method
|
||||
)
|
||||
|
||||
click.echo(f"\n📊 Worktime Estimation for {result['work_date']}")
|
||||
click.echo(f"Total Hours: {hours}h ({result['total_minutes']} minutes)")
|
||||
click.echo(f"Distribution Method: {method}")
|
||||
click.echo(f"Issues: {result['issues_count']}")
|
||||
|
||||
click.echo("\nEstimated Time per Issue:")
|
||||
headers = ['Issue', 'Time', 'Percentage']
|
||||
rows = []
|
||||
|
||||
for issue_id, minutes in result['issue_estimates'].items():
|
||||
percentage = (minutes / result['total_minutes']) * 100
|
||||
rows.append([
|
||||
f"#{issue_id}",
|
||||
_format_duration(minutes),
|
||||
f"{percentage:.1f}%"
|
||||
])
|
||||
|
||||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||||
click.echo(f"\n✅ Created {len(result['issue_estimates'])} estimated worktime entries")
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error creating estimation: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.argument('date', type=click.DateTime(formats=['%Y-%m-%d']))
|
||||
@click.argument('total_cost', type=float)
|
||||
@click.option('--period-id', type=int, help='Cost period ID for tracking')
|
||||
def distribute(date: datetime, total_cost: float, period_id: Optional[int]):
|
||||
"""Distribute daily costs based on time allocation."""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
try:
|
||||
from decimal import Decimal
|
||||
result = tracker.distribute_daily_costs(
|
||||
work_date=date.date(),
|
||||
total_daily_cost=Decimal(str(total_cost)),
|
||||
period_id=period_id
|
||||
)
|
||||
|
||||
if 'message' in result:
|
||||
click.echo(f"⚠️ {result['message']}")
|
||||
return
|
||||
|
||||
click.echo(f"\n💰 Cost Distribution for {result['work_date']}")
|
||||
click.echo(f"Total Cost: €{result['total_cost']:.2f}")
|
||||
click.echo(f"Total Time: {_format_duration(result['total_minutes'])}")
|
||||
click.echo(f"Cost per Minute: €{result['cost_per_minute']:.4f}")
|
||||
|
||||
click.echo("\nCost Allocation by Issue:")
|
||||
headers = ['Issue', 'Time', 'Percentage', 'Cost Allocated']
|
||||
rows = []
|
||||
|
||||
for issue_id, distribution in result['distributions'].items():
|
||||
rows.append([
|
||||
f"#{issue_id}",
|
||||
_format_duration(distribution['minutes']),
|
||||
f"{distribution['percentage']:.1f}%",
|
||||
f"€{distribution['cost_allocated']:.2f}"
|
||||
])
|
||||
|
||||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||||
|
||||
if period_id:
|
||||
click.echo(f"\n✅ Cost distribution logged to period #{period_id}")
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error distributing costs: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.option('--start-date', type=click.DateTime(formats=['%Y-%m-%d']),
|
||||
help='Report start date (defaults to 30 days ago)')
|
||||
@click.option('--end-date', type=click.DateTime(formats=['%Y-%m-%d']),
|
||||
help='Report end date (defaults to today)')
|
||||
@click.option('--issue', type=int, help='Filter by specific issue ID')
|
||||
@click.option('--format', 'output_format', type=click.Choice(['table', 'json']),
|
||||
default='table', help='Output format')
|
||||
def report(start_date: Optional[datetime], end_date: Optional[datetime],
|
||||
issue: Optional[int], output_format: str):
|
||||
"""Generate comprehensive worktime report."""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
try:
|
||||
result = tracker.get_worktime_report(
|
||||
start_date=start_date.date() if start_date else None,
|
||||
end_date=end_date.date() if end_date else None,
|
||||
issue_id=issue
|
||||
)
|
||||
|
||||
if output_format == 'json':
|
||||
click.echo(json.dumps(result, indent=2))
|
||||
return
|
||||
|
||||
# Table format
|
||||
click.echo(f"\n📈 Worktime Report")
|
||||
click.echo(f"Period: {result['period']}")
|
||||
click.echo(f"Total Entries: {result['total_entries']}")
|
||||
click.echo(f"Total Time: {result['total_time']['hours']}h {result['total_time']['minutes']}m")
|
||||
click.echo(f"Unique Issues: {result['unique_issues']}")
|
||||
click.echo(f"Unique Dates: {result['unique_dates']}")
|
||||
|
||||
if result['unique_dates'] > 0:
|
||||
avg_minutes = result['average_minutes_per_day']
|
||||
avg_hours = int(avg_minutes // 60)
|
||||
avg_mins = int(avg_minutes % 60)
|
||||
click.echo(f"Average per Day: {avg_hours}h {avg_mins}m")
|
||||
|
||||
if result.get('issue_breakdown'):
|
||||
click.echo("\nTime by Issue:")
|
||||
headers = ['Issue', 'Total Time', 'Entries', 'Days', 'Avg/Day']
|
||||
rows = []
|
||||
|
||||
for issue_id, data in sorted(result['issue_breakdown'].items()):
|
||||
total_time = _format_duration(data['total_minutes'])
|
||||
avg_per_day = data['total_minutes'] / data['unique_dates']
|
||||
avg_formatted = _format_duration(int(avg_per_day))
|
||||
|
||||
rows.append([
|
||||
f"#{issue_id}",
|
||||
total_time,
|
||||
data['entry_count'],
|
||||
data['unique_dates'],
|
||||
avg_formatted
|
||||
])
|
||||
|
||||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error generating report: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.argument('entry_id', type=int)
|
||||
@click.confirmation_option(prompt='Are you sure you want to delete this worktime entry?')
|
||||
def delete(entry_id: int):
|
||||
"""Delete a worktime entry."""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
try:
|
||||
if tracker.delete_worktime_entry(entry_id):
|
||||
click.echo(f"✅ Deleted worktime entry #{entry_id}")
|
||||
else:
|
||||
click.echo(f"❌ Worktime entry #{entry_id} not found")
|
||||
raise click.Abort()
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error deleting entry: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@worktime.command()
|
||||
@click.argument('entry_id', type=int)
|
||||
@click.option('--duration', type=str, help='New duration (e.g., 90, 1h30m, 2.5h)')
|
||||
@click.option('--description', help='New description')
|
||||
@click.option('--start', type=click.DateTime(formats=['%H:%M', '%Y-%m-%d %H:%M']),
|
||||
help='New start time')
|
||||
@click.option('--end', type=click.DateTime(formats=['%H:%M', '%Y-%m-%d %H:%M']),
|
||||
help='New end time')
|
||||
def update(entry_id: int, duration: Optional[str], description: Optional[str],
|
||||
start: Optional[datetime], end: Optional[datetime]):
|
||||
"""Update a worktime entry."""
|
||||
tracker = WorktimeTracker()
|
||||
|
||||
try:
|
||||
duration_minutes = None
|
||||
if duration:
|
||||
duration_minutes = _parse_duration(duration)
|
||||
|
||||
success = tracker.update_worktime_entry(
|
||||
entry_id=entry_id,
|
||||
duration_minutes=duration_minutes,
|
||||
description=description,
|
||||
start_time=start,
|
||||
end_time=end
|
||||
)
|
||||
|
||||
if success:
|
||||
click.echo(f"✅ Updated worktime entry #{entry_id}")
|
||||
else:
|
||||
click.echo(f"❌ Worktime entry #{entry_id} not found or no changes made")
|
||||
raise click.Abort()
|
||||
|
||||
except ValueError as e:
|
||||
click.echo(f"❌ Invalid duration format: {e}", err=True)
|
||||
raise click.Abort()
|
||||
except Exception as e:
|
||||
click.echo(f"❌ Error updating entry: {e}", err=True)
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
def _parse_duration(duration_str: str) -> int:
|
||||
"""Parse duration string into minutes."""
|
||||
duration_str = duration_str.lower().strip()
|
||||
|
||||
# Handle pure numbers (assume minutes)
|
||||
if duration_str.isdigit():
|
||||
return int(duration_str)
|
||||
|
||||
# Handle decimal hours (e.g., 1.5h, 2.25h)
|
||||
if duration_str.endswith('h') and '.' in duration_str:
|
||||
try:
|
||||
hours = float(duration_str[:-1])
|
||||
return int(hours * 60)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Handle hours and minutes (e.g., 1h30m, 2h15m)
|
||||
if 'h' in duration_str:
|
||||
parts = duration_str.split('h')
|
||||
hours = int(parts[0]) if parts[0] else 0
|
||||
minutes = 0
|
||||
|
||||
if len(parts) > 1 and parts[1]:
|
||||
minute_part = parts[1].replace('m', '').strip()
|
||||
if minute_part:
|
||||
minutes = int(minute_part)
|
||||
|
||||
return hours * 60 + minutes
|
||||
|
||||
# Handle minutes only (e.g., 90m)
|
||||
if duration_str.endswith('m'):
|
||||
return int(duration_str[:-1])
|
||||
|
||||
raise ValueError(f"Unable to parse duration: {duration_str}. Use formats like: 90, 1h30m, 2.5h")
|
||||
|
||||
|
||||
def _format_duration(minutes: int) -> str:
|
||||
"""Format minutes into human-readable duration."""
|
||||
if minutes < 60:
|
||||
return f"{minutes}m"
|
||||
|
||||
hours = minutes // 60
|
||||
remaining_minutes = minutes % 60
|
||||
|
||||
if remaining_minutes == 0:
|
||||
return f"{hours}h"
|
||||
else:
|
||||
return f"{hours}h{remaining_minutes}m"
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
worktime()
|
||||
677
markitect/finance/worktime_tracker.py
Normal file
677
markitect/finance/worktime_tracker.py
Normal file
@@ -0,0 +1,677 @@
|
||||
"""
|
||||
Daily Worktime Tracking and Cost Distribution System
|
||||
|
||||
This module provides comprehensive worktime tracking functionality that estimates
|
||||
daily work time spent on issues and distributes costs proportionally based on
|
||||
time allocation rather than equal distribution.
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
import json
|
||||
from datetime import datetime, date, timedelta
|
||||
from typing import List, Dict, Any, Optional, Tuple
|
||||
from dataclasses import dataclass, asdict
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from .models import FinanceModels
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorktimeEntry:
|
||||
"""Data class representing a worktime tracking entry."""
|
||||
id: Optional[int] = None
|
||||
issue_id: int = None
|
||||
work_date: date = None
|
||||
start_time: Optional[datetime] = None
|
||||
end_time: Optional[datetime] = None
|
||||
duration_minutes: int = None
|
||||
description: Optional[str] = None
|
||||
entry_type: str = "manual" # manual, estimated, tracked
|
||||
created_at: Optional[datetime] = None
|
||||
updated_at: Optional[datetime] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class DailySummary:
|
||||
"""Data class for daily worktime summary."""
|
||||
work_date: date
|
||||
total_minutes: int
|
||||
issue_count: int
|
||||
entries: List[WorktimeEntry]
|
||||
cost_per_minute: Optional[Decimal] = None
|
||||
total_cost_allocated: Optional[Decimal] = None
|
||||
|
||||
|
||||
class WorktimeTracker:
|
||||
"""
|
||||
Service for tracking worktime and distributing costs based on time allocation.
|
||||
|
||||
Provides functionality to log work time, estimate daily effort, and calculate
|
||||
proportional cost distribution across issues based on time spent.
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: str = "markitect.db"):
|
||||
"""
|
||||
Initialize the worktime tracker.
|
||||
|
||||
Args:
|
||||
db_path: Path to the SQLite database file
|
||||
"""
|
||||
self.db_path = db_path
|
||||
self.finance_models = FinanceModels(db_path)
|
||||
self._ensure_schema()
|
||||
|
||||
def _ensure_schema(self):
|
||||
"""Ensure the database schema is properly initialized."""
|
||||
self.finance_models.initialize_finance_schema()
|
||||
self._create_worktime_tables()
|
||||
|
||||
def _create_worktime_tables(self):
|
||||
"""Create worktime-specific database tables."""
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Enable foreign key constraints
|
||||
cursor.execute('PRAGMA foreign_keys = ON')
|
||||
|
||||
# Create worktime entries table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS worktime_entries (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
issue_id INTEGER NOT NULL,
|
||||
work_date DATE NOT NULL,
|
||||
start_time TIMESTAMP,
|
||||
end_time TIMESTAMP,
|
||||
duration_minutes INTEGER NOT NULL CHECK (duration_minutes > 0),
|
||||
description TEXT,
|
||||
entry_type TEXT CHECK (entry_type IN ('manual', 'estimated', 'tracked')) DEFAULT 'manual',
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
CONSTRAINT unique_issue_date_time UNIQUE (issue_id, work_date, start_time)
|
||||
)
|
||||
''')
|
||||
|
||||
# Create daily worktime summaries table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS daily_worktime_summaries (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
work_date DATE NOT NULL UNIQUE,
|
||||
total_minutes INTEGER NOT NULL DEFAULT 0,
|
||||
issue_count INTEGER NOT NULL DEFAULT 0,
|
||||
cost_per_minute DECIMAL(10,6),
|
||||
total_cost_allocated DECIMAL(10,2),
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
''')
|
||||
|
||||
# Create cost distribution log table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS worktime_cost_distributions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
work_date DATE NOT NULL,
|
||||
issue_id INTEGER NOT NULL,
|
||||
time_minutes INTEGER NOT NULL,
|
||||
cost_allocated DECIMAL(10,2) NOT NULL,
|
||||
cost_per_minute DECIMAL(10,6) NOT NULL,
|
||||
period_id INTEGER REFERENCES cost_periods(id),
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
CONSTRAINT unique_date_issue_distribution UNIQUE (work_date, issue_id)
|
||||
)
|
||||
''')
|
||||
|
||||
def log_worktime(self,
|
||||
issue_id: int,
|
||||
duration_minutes: int,
|
||||
work_date: Optional[date] = None,
|
||||
start_time: Optional[datetime] = None,
|
||||
end_time: Optional[datetime] = None,
|
||||
description: Optional[str] = None,
|
||||
entry_type: str = "manual") -> int:
|
||||
"""
|
||||
Log worktime for an issue.
|
||||
|
||||
Args:
|
||||
issue_id: ID of the issue
|
||||
duration_minutes: Duration of work in minutes
|
||||
work_date: Date when work was performed (defaults to today)
|
||||
start_time: Optional start timestamp
|
||||
end_time: Optional end timestamp
|
||||
description: Description of work performed
|
||||
entry_type: Type of entry (manual, estimated, tracked)
|
||||
|
||||
Returns:
|
||||
ID of the created worktime entry
|
||||
|
||||
Raises:
|
||||
ValueError: If duration is invalid
|
||||
sqlite3.Error: If database operation fails
|
||||
"""
|
||||
if duration_minutes <= 0:
|
||||
raise ValueError("Duration must be positive")
|
||||
|
||||
if work_date is None:
|
||||
work_date = date.today()
|
||||
|
||||
# Validate time consistency
|
||||
if start_time and end_time:
|
||||
calculated_duration = int((end_time - start_time).total_seconds() / 60)
|
||||
if abs(calculated_duration - duration_minutes) > 5: # Allow 5-minute tolerance
|
||||
print(f"⚠️ Warning: Calculated duration ({calculated_duration}min) differs from specified ({duration_minutes}min)")
|
||||
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
INSERT INTO worktime_entries
|
||||
(issue_id, work_date, start_time, end_time, duration_minutes, description, entry_type)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
''', (issue_id, work_date, start_time, end_time, duration_minutes, description, entry_type))
|
||||
|
||||
entry_id = cursor.lastrowid
|
||||
|
||||
# Update daily summary
|
||||
self._update_daily_summary(work_date, conn)
|
||||
|
||||
return entry_id
|
||||
|
||||
def _update_daily_summary(self, work_date: date, conn: sqlite3.Connection):
|
||||
"""Update the daily worktime summary for a given date."""
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Calculate daily totals
|
||||
cursor.execute('''
|
||||
SELECT
|
||||
COUNT(DISTINCT issue_id) as issue_count,
|
||||
SUM(duration_minutes) as total_minutes
|
||||
FROM worktime_entries
|
||||
WHERE work_date = ?
|
||||
''', (work_date,))
|
||||
|
||||
result = cursor.fetchone()
|
||||
issue_count = result[0] or 0
|
||||
total_minutes = result[1] or 0
|
||||
|
||||
# Insert or update summary
|
||||
cursor.execute('''
|
||||
INSERT OR REPLACE INTO daily_worktime_summaries
|
||||
(work_date, total_minutes, issue_count)
|
||||
VALUES (?, ?, ?)
|
||||
''', (work_date, total_minutes, issue_count))
|
||||
|
||||
def get_worktime_entries(self,
|
||||
issue_id: Optional[int] = None,
|
||||
work_date: Optional[date] = None,
|
||||
start_date: Optional[date] = None,
|
||||
end_date: Optional[date] = None,
|
||||
limit: Optional[int] = None) -> List[WorktimeEntry]:
|
||||
"""
|
||||
Get worktime entries with optional filtering.
|
||||
|
||||
Args:
|
||||
issue_id: Filter by specific issue
|
||||
work_date: Filter by specific date
|
||||
start_date: Filter by date range start
|
||||
end_date: Filter by date range end
|
||||
limit: Maximum number of entries to return
|
||||
|
||||
Returns:
|
||||
List of worktime entries
|
||||
"""
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
query = '''
|
||||
SELECT id, issue_id, work_date, start_time, end_time,
|
||||
duration_minutes, description, entry_type, created_at, updated_at
|
||||
FROM worktime_entries
|
||||
WHERE 1=1
|
||||
'''
|
||||
params = []
|
||||
|
||||
if issue_id:
|
||||
query += ' AND issue_id = ?'
|
||||
params.append(issue_id)
|
||||
|
||||
if work_date:
|
||||
query += ' AND work_date = ?'
|
||||
params.append(work_date)
|
||||
elif start_date and end_date:
|
||||
query += ' AND work_date BETWEEN ? AND ?'
|
||||
params.extend([start_date, end_date])
|
||||
elif start_date:
|
||||
query += ' AND work_date >= ?'
|
||||
params.append(start_date)
|
||||
elif end_date:
|
||||
query += ' AND work_date <= ?'
|
||||
params.append(end_date)
|
||||
|
||||
query += ' ORDER BY work_date DESC, start_time DESC'
|
||||
|
||||
if limit:
|
||||
query += ' LIMIT ?'
|
||||
params.append(limit)
|
||||
|
||||
cursor.execute(query, params)
|
||||
rows = cursor.fetchall()
|
||||
|
||||
entries = []
|
||||
for row in rows:
|
||||
entry = WorktimeEntry(
|
||||
id=row[0],
|
||||
issue_id=row[1],
|
||||
work_date=datetime.strptime(row[2], '%Y-%m-%d').date() if row[2] else None,
|
||||
start_time=datetime.fromisoformat(row[3]) if row[3] else None,
|
||||
end_time=datetime.fromisoformat(row[4]) if row[4] else None,
|
||||
duration_minutes=row[5],
|
||||
description=row[6],
|
||||
entry_type=row[7],
|
||||
created_at=datetime.fromisoformat(row[8]) if row[8] else None,
|
||||
updated_at=datetime.fromisoformat(row[9]) if row[9] else None
|
||||
)
|
||||
entries.append(entry)
|
||||
|
||||
return entries
|
||||
|
||||
def get_daily_summary(self, work_date: date) -> Optional[DailySummary]:
|
||||
"""
|
||||
Get daily worktime summary for a specific date.
|
||||
|
||||
Args:
|
||||
work_date: Date to get summary for
|
||||
|
||||
Returns:
|
||||
Daily summary or None if no data
|
||||
"""
|
||||
entries = self.get_worktime_entries(work_date=work_date)
|
||||
if not entries:
|
||||
return None
|
||||
|
||||
total_minutes = sum(entry.duration_minutes for entry in entries)
|
||||
issue_count = len(set(entry.issue_id for entry in entries))
|
||||
|
||||
# Get cost allocation if available
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
SELECT cost_per_minute, total_cost_allocated
|
||||
FROM daily_worktime_summaries
|
||||
WHERE work_date = ?
|
||||
''', (work_date,))
|
||||
result = cursor.fetchone()
|
||||
|
||||
cost_per_minute = Decimal(str(result[0])) if result and result[0] else None
|
||||
total_cost_allocated = Decimal(str(result[1])) if result and result[1] else None
|
||||
|
||||
return DailySummary(
|
||||
work_date=work_date,
|
||||
total_minutes=total_minutes,
|
||||
issue_count=issue_count,
|
||||
entries=entries,
|
||||
cost_per_minute=cost_per_minute,
|
||||
total_cost_allocated=total_cost_allocated
|
||||
)
|
||||
|
||||
def estimate_daily_worktime(self,
|
||||
work_date: date,
|
||||
total_hours: float = 8.0,
|
||||
issues: Optional[List[int]] = None,
|
||||
distribution_method: str = "equal") -> Dict[str, Any]:
|
||||
"""
|
||||
Estimate worktime distribution for a day when specific times aren't tracked.
|
||||
|
||||
Args:
|
||||
work_date: Date to estimate for
|
||||
total_hours: Total work hours for the day
|
||||
issues: List of issue IDs that were worked on
|
||||
distribution_method: How to distribute time (equal, activity_based, manual)
|
||||
|
||||
Returns:
|
||||
Dictionary with estimation results
|
||||
"""
|
||||
total_minutes = int(total_hours * 60)
|
||||
|
||||
if not issues:
|
||||
# Try to identify issues from activity log
|
||||
issues = self._get_active_issues_for_date(work_date)
|
||||
|
||||
if not issues:
|
||||
raise ValueError("No issues specified and none found in activity log")
|
||||
|
||||
estimates = {}
|
||||
|
||||
if distribution_method == "equal":
|
||||
# Equal distribution across all issues
|
||||
minutes_per_issue = total_minutes // len(issues)
|
||||
remainder = total_minutes % len(issues)
|
||||
|
||||
for i, issue_id in enumerate(issues):
|
||||
minutes = minutes_per_issue + (1 if i < remainder else 0)
|
||||
estimates[issue_id] = minutes
|
||||
|
||||
elif distribution_method == "activity_based":
|
||||
# Distribute based on activity frequency
|
||||
activity_weights = self._get_activity_weights_for_date(work_date, issues)
|
||||
total_weight = sum(activity_weights.values())
|
||||
|
||||
if total_weight == 0:
|
||||
# Fallback to equal distribution
|
||||
minutes_per_issue = total_minutes // len(issues)
|
||||
for issue_id in issues:
|
||||
estimates[issue_id] = minutes_per_issue
|
||||
else:
|
||||
for issue_id in issues:
|
||||
weight = activity_weights.get(issue_id, 0)
|
||||
minutes = int((weight / total_weight) * total_minutes)
|
||||
estimates[issue_id] = minutes
|
||||
|
||||
# Log estimated entries
|
||||
for issue_id, minutes in estimates.items():
|
||||
if minutes > 0:
|
||||
self.log_worktime(
|
||||
issue_id=issue_id,
|
||||
duration_minutes=minutes,
|
||||
work_date=work_date,
|
||||
description=f"Estimated worktime ({distribution_method} distribution)",
|
||||
entry_type="estimated"
|
||||
)
|
||||
|
||||
return {
|
||||
"work_date": work_date,
|
||||
"total_minutes": total_minutes,
|
||||
"distribution_method": distribution_method,
|
||||
"issue_estimates": estimates,
|
||||
"issues_count": len(issues)
|
||||
}
|
||||
|
||||
def _get_active_issues_for_date(self, work_date: date) -> List[int]:
|
||||
"""Get issues that had activity on a specific date."""
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
SELECT DISTINCT issue_id
|
||||
FROM issue_activity_log
|
||||
WHERE activity_date = ?
|
||||
''', (work_date,))
|
||||
return [row[0] for row in cursor.fetchall()]
|
||||
|
||||
def _get_activity_weights_for_date(self, work_date: date, issues: List[int]) -> Dict[int, int]:
|
||||
"""Get activity weights for issues on a specific date."""
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
placeholders = ','.join(['?' for _ in issues])
|
||||
cursor.execute(f'''
|
||||
SELECT issue_id, COUNT(*) as activity_count
|
||||
FROM issue_activity_log
|
||||
WHERE activity_date = ? AND issue_id IN ({placeholders})
|
||||
GROUP BY issue_id
|
||||
''', [work_date] + issues)
|
||||
return dict(cursor.fetchall())
|
||||
|
||||
def distribute_daily_costs(self,
|
||||
work_date: date,
|
||||
total_daily_cost: Decimal,
|
||||
period_id: Optional[int] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Distribute daily costs proportionally based on time spent on each issue.
|
||||
|
||||
Args:
|
||||
work_date: Date to distribute costs for
|
||||
total_daily_cost: Total cost to distribute
|
||||
period_id: Cost period ID for tracking
|
||||
|
||||
Returns:
|
||||
Dictionary with distribution results
|
||||
"""
|
||||
# Get worktime entries for the date
|
||||
entries = self.get_worktime_entries(work_date=work_date)
|
||||
if not entries:
|
||||
return {
|
||||
"work_date": work_date,
|
||||
"total_cost": float(total_daily_cost),
|
||||
"distributions": {},
|
||||
"message": "No worktime entries found for this date"
|
||||
}
|
||||
|
||||
# Calculate time per issue
|
||||
issue_minutes = {}
|
||||
for entry in entries:
|
||||
issue_minutes[entry.issue_id] = issue_minutes.get(entry.issue_id, 0) + entry.duration_minutes
|
||||
|
||||
total_minutes = sum(issue_minutes.values())
|
||||
if total_minutes == 0:
|
||||
return {
|
||||
"work_date": work_date,
|
||||
"total_cost": float(total_daily_cost),
|
||||
"distributions": {},
|
||||
"message": "No time tracked for this date"
|
||||
}
|
||||
|
||||
cost_per_minute = total_daily_cost / total_minutes
|
||||
|
||||
# Calculate cost distribution
|
||||
distributions = {}
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
for issue_id, minutes in issue_minutes.items():
|
||||
cost_allocated = cost_per_minute * minutes
|
||||
distributions[issue_id] = {
|
||||
"minutes": minutes,
|
||||
"percentage": (minutes / total_minutes) * 100,
|
||||
"cost_allocated": float(cost_allocated),
|
||||
"cost_per_minute": float(cost_per_minute)
|
||||
}
|
||||
|
||||
# Log the distribution
|
||||
cursor.execute('''
|
||||
INSERT OR REPLACE INTO worktime_cost_distributions
|
||||
(work_date, issue_id, time_minutes, cost_allocated, cost_per_minute, period_id)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
''', (work_date, issue_id, minutes, float(cost_allocated), float(cost_per_minute), period_id))
|
||||
|
||||
# Update daily summary with cost information
|
||||
cursor.execute('''
|
||||
UPDATE daily_worktime_summaries
|
||||
SET cost_per_minute = ?, total_cost_allocated = ?, updated_at = CURRENT_TIMESTAMP
|
||||
WHERE work_date = ?
|
||||
''', (float(cost_per_minute), float(total_daily_cost), work_date))
|
||||
|
||||
return {
|
||||
"work_date": work_date,
|
||||
"total_cost": float(total_daily_cost),
|
||||
"total_minutes": total_minutes,
|
||||
"cost_per_minute": float(cost_per_minute),
|
||||
"distributions": distributions,
|
||||
"issues_count": len(distributions)
|
||||
}
|
||||
|
||||
def get_worktime_report(self,
|
||||
start_date: Optional[date] = None,
|
||||
end_date: Optional[date] = None,
|
||||
issue_id: Optional[int] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Generate comprehensive worktime report.
|
||||
|
||||
Args:
|
||||
start_date: Report start date
|
||||
end_date: Report end date
|
||||
issue_id: Filter by specific issue
|
||||
|
||||
Returns:
|
||||
Comprehensive worktime report
|
||||
"""
|
||||
if not start_date:
|
||||
start_date = date.today() - timedelta(days=30) # Last 30 days
|
||||
if not end_date:
|
||||
end_date = date.today()
|
||||
|
||||
entries = self.get_worktime_entries(
|
||||
issue_id=issue_id,
|
||||
start_date=start_date,
|
||||
end_date=end_date
|
||||
)
|
||||
|
||||
if not entries:
|
||||
return {
|
||||
"period": f"{start_date} to {end_date}",
|
||||
"total_entries": 0,
|
||||
"total_time": {"hours": 0, "minutes": 0},
|
||||
"summary": "No worktime entries found"
|
||||
}
|
||||
|
||||
# Calculate summary statistics
|
||||
total_minutes = sum(entry.duration_minutes for entry in entries)
|
||||
issue_breakdown = {}
|
||||
daily_breakdown = {}
|
||||
|
||||
for entry in entries:
|
||||
# Issue breakdown
|
||||
if entry.issue_id not in issue_breakdown:
|
||||
issue_breakdown[entry.issue_id] = {
|
||||
"total_minutes": 0,
|
||||
"entry_count": 0,
|
||||
"dates": set()
|
||||
}
|
||||
issue_breakdown[entry.issue_id]["total_minutes"] += entry.duration_minutes
|
||||
issue_breakdown[entry.issue_id]["entry_count"] += 1
|
||||
issue_breakdown[entry.issue_id]["dates"].add(entry.work_date)
|
||||
|
||||
# Daily breakdown
|
||||
date_str = entry.work_date.isoformat()
|
||||
if date_str not in daily_breakdown:
|
||||
daily_breakdown[date_str] = {
|
||||
"total_minutes": 0,
|
||||
"issues": set(),
|
||||
"entries": 0
|
||||
}
|
||||
daily_breakdown[date_str]["total_minutes"] += entry.duration_minutes
|
||||
daily_breakdown[date_str]["issues"].add(entry.issue_id)
|
||||
daily_breakdown[date_str]["entries"] += 1
|
||||
|
||||
# Convert sets to counts for JSON serialization
|
||||
for issue_id in issue_breakdown:
|
||||
issue_breakdown[issue_id]["unique_dates"] = len(issue_breakdown[issue_id]["dates"])
|
||||
del issue_breakdown[issue_id]["dates"]
|
||||
|
||||
for date_str in daily_breakdown:
|
||||
daily_breakdown[date_str]["unique_issues"] = len(daily_breakdown[date_str]["issues"])
|
||||
del daily_breakdown[date_str]["issues"]
|
||||
|
||||
return {
|
||||
"period": f"{start_date} to {end_date}",
|
||||
"total_entries": len(entries),
|
||||
"total_time": {
|
||||
"hours": total_minutes // 60,
|
||||
"minutes": total_minutes % 60,
|
||||
"total_minutes": total_minutes
|
||||
},
|
||||
"unique_issues": len(issue_breakdown),
|
||||
"unique_dates": len(daily_breakdown),
|
||||
"average_minutes_per_day": total_minutes / len(daily_breakdown) if daily_breakdown else 0,
|
||||
"issue_breakdown": issue_breakdown,
|
||||
"daily_breakdown": daily_breakdown
|
||||
}
|
||||
|
||||
def delete_worktime_entry(self, entry_id: int) -> bool:
|
||||
"""
|
||||
Delete a worktime entry and update summaries.
|
||||
|
||||
Args:
|
||||
entry_id: ID of the entry to delete
|
||||
|
||||
Returns:
|
||||
True if deleted successfully, False otherwise
|
||||
"""
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Get the entry details before deletion for summary update
|
||||
cursor.execute('''
|
||||
SELECT work_date FROM worktime_entries WHERE id = ?
|
||||
''', (entry_id,))
|
||||
result = cursor.fetchone()
|
||||
|
||||
if not result:
|
||||
return False
|
||||
|
||||
work_date = datetime.strptime(result[0], '%Y-%m-%d').date()
|
||||
|
||||
# Delete the entry
|
||||
cursor.execute('DELETE FROM worktime_entries WHERE id = ?', (entry_id,))
|
||||
|
||||
if cursor.rowcount > 0:
|
||||
# Update daily summary
|
||||
self._update_daily_summary(work_date, conn)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def update_worktime_entry(self,
|
||||
entry_id: int,
|
||||
duration_minutes: Optional[int] = None,
|
||||
description: Optional[str] = None,
|
||||
start_time: Optional[datetime] = None,
|
||||
end_time: Optional[datetime] = None) -> bool:
|
||||
"""
|
||||
Update an existing worktime entry.
|
||||
|
||||
Args:
|
||||
entry_id: ID of the entry to update
|
||||
duration_minutes: New duration in minutes
|
||||
description: New description
|
||||
start_time: New start time
|
||||
end_time: New end time
|
||||
|
||||
Returns:
|
||||
True if updated successfully, False otherwise
|
||||
"""
|
||||
updates = {}
|
||||
params = []
|
||||
|
||||
if duration_minutes is not None:
|
||||
updates["duration_minutes"] = "?"
|
||||
params.append(duration_minutes)
|
||||
if description is not None:
|
||||
updates["description"] = "?"
|
||||
params.append(description)
|
||||
if start_time is not None:
|
||||
updates["start_time"] = "?"
|
||||
params.append(start_time)
|
||||
if end_time is not None:
|
||||
updates["end_time"] = "?"
|
||||
params.append(end_time)
|
||||
|
||||
if not updates:
|
||||
return False
|
||||
|
||||
updates["updated_at"] = "CURRENT_TIMESTAMP"
|
||||
|
||||
with self.finance_models.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Get work date for summary update
|
||||
cursor.execute('SELECT work_date FROM worktime_entries WHERE id = ?', (entry_id,))
|
||||
result = cursor.fetchone()
|
||||
if not result:
|
||||
return False
|
||||
|
||||
work_date = datetime.strptime(result[0], '%Y-%m-%d').date()
|
||||
|
||||
# Build update query
|
||||
set_clause = ", ".join(f"{col} = {val}" for col, val in updates.items())
|
||||
params.append(entry_id)
|
||||
|
||||
cursor.execute(f'''
|
||||
UPDATE worktime_entries
|
||||
SET {set_clause}
|
||||
WHERE id = ?
|
||||
''', params)
|
||||
|
||||
if cursor.rowcount > 0:
|
||||
# Update daily summary
|
||||
self._update_daily_summary(work_date, conn)
|
||||
return True
|
||||
|
||||
return False
|
||||
794
tests/test_issue_122_worktime_tracking.py
Normal file
794
tests/test_issue_122_worktime_tracking.py
Normal file
@@ -0,0 +1,794 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user