204 lines
7.6 KiB
Python
204 lines
7.6 KiB
Python
|
|
import json
|
||
|
|
import os
|
||
|
|
import tempfile
|
||
|
|
from datetime import datetime
|
||
|
|
import pytest
|
||
|
|
from unittest.mock import patch, MagicMock
|
||
|
|
|
||
|
|
from time_tracker.core.session_manager import SessionManager
|
||
|
|
|
||
|
|
|
||
|
|
def test_create_new_session():
|
||
|
|
"""Test creating a new session"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
# Create tasks directory
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
|
||
|
|
# Create a new session
|
||
|
|
session = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
|
||
|
|
# Check session has required fields
|
||
|
|
assert 'session_id' in session
|
||
|
|
assert session['task_id'] == 'PRJ-123'
|
||
|
|
assert session['project_id'] == 'PRJ'
|
||
|
|
assert 'started_at' in session
|
||
|
|
assert session['duration_seconds'] == 0
|
||
|
|
assert session['window_details'] == {}
|
||
|
|
assert session['distractions'] == []
|
||
|
|
assert session['synced_to_youtrack'] == False
|
||
|
|
|
||
|
|
|
||
|
|
def test_end_current_session():
|
||
|
|
"""Test ending the current session"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
|
||
|
|
# Start a session
|
||
|
|
session = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
session_manager.current_session = session
|
||
|
|
|
||
|
|
# Add some window time
|
||
|
|
session_manager.record_window_time('PyCharm', 300) # 5 minutes
|
||
|
|
session_manager.record_distraction('YouTube', 120) # 2 minutes
|
||
|
|
|
||
|
|
# End the session
|
||
|
|
ended_session = session_manager.end_current_session()
|
||
|
|
|
||
|
|
# Check that session was properly ended
|
||
|
|
assert 'ended_at' in ended_session
|
||
|
|
assert ended_session['duration_seconds'] >= 420 # At least 7 minutes
|
||
|
|
assert ended_session['window_details']['PyCharm'] == 300
|
||
|
|
assert len(ended_session['distractions']) == 1
|
||
|
|
assert ended_session['distractions'][0]['window'] == 'YouTube'
|
||
|
|
assert ended_session['distractions'][0]['duration_seconds'] == 120
|
||
|
|
|
||
|
|
# Check that file was saved
|
||
|
|
session_file_path = os.path.join(tasks_dir, 'PRJ-123', 'sessions',
|
||
|
|
f"{ended_session['session_id']}_PRJ-123.json")
|
||
|
|
assert os.path.exists(session_file_path)
|
||
|
|
|
||
|
|
# Verify content of saved file
|
||
|
|
with open(session_file_path, 'r') as f:
|
||
|
|
saved_session = json.load(f)
|
||
|
|
assert saved_session == ended_session
|
||
|
|
|
||
|
|
|
||
|
|
def test_record_window_time():
|
||
|
|
"""Test recording time for specific windows"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
session = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
session_manager.current_session = session
|
||
|
|
|
||
|
|
# Record time for multiple windows
|
||
|
|
session_manager.record_window_time('PyCharm', 300) # 5 minutes
|
||
|
|
session_manager.record_window_time('Chrome', 120) # 2 minutes
|
||
|
|
session_manager.record_window_time('PyCharm', 60) # Additional minute
|
||
|
|
|
||
|
|
# Check accumulated times
|
||
|
|
assert session['window_details']['PyCharm'] == 360 # 6 minutes
|
||
|
|
assert session['window_details']['Chrome'] == 120 # 2 minutes
|
||
|
|
|
||
|
|
|
||
|
|
def test_record_distraction():
|
||
|
|
"""Test recording distractions"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
session = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
session_manager.current_session = session
|
||
|
|
|
||
|
|
# Record a distraction
|
||
|
|
session_manager.record_distraction('YouTube', 120)
|
||
|
|
|
||
|
|
# Check distraction was recorded
|
||
|
|
assert len(session['distractions']) == 1
|
||
|
|
distraction = session['distractions'][0]
|
||
|
|
assert distraction['window'] == 'YouTube'
|
||
|
|
assert distraction['duration_seconds'] == 120
|
||
|
|
assert 'start' in distraction
|
||
|
|
|
||
|
|
|
||
|
|
def test_manual_adjustment():
|
||
|
|
"""Test adding manual time adjustments"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
session = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
session_manager.current_session = session
|
||
|
|
|
||
|
|
# Add manual adjustment
|
||
|
|
session_manager.add_manual_adjustment(900, 'Zoom meeting')
|
||
|
|
|
||
|
|
# Check adjustment was recorded
|
||
|
|
assert len(session['manual_adjustments']) == 1
|
||
|
|
adjustment = session['manual_adjustments'][0]
|
||
|
|
assert adjustment['operation'] == 'add'
|
||
|
|
assert adjustment['seconds'] == 900
|
||
|
|
assert adjustment['reason'] == 'Zoom meeting'
|
||
|
|
assert 'applied_at' in adjustment
|
||
|
|
|
||
|
|
# Test subtraction
|
||
|
|
session_manager.add_manual_adjustment(-300, 'Break')
|
||
|
|
assert len(session['manual_adjustments']) == 2
|
||
|
|
sub_adjustment = session['manual_adjustments'][1]
|
||
|
|
assert sub_adjustment['operation'] == 'subtract'
|
||
|
|
assert sub_adjustment['seconds'] == 300
|
||
|
|
assert sub_adjustment['reason'] == 'Break'
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_current_session_duration():
|
||
|
|
"""Test getting current session duration with adjustments"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
session = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
session_manager.current_session = session
|
||
|
|
|
||
|
|
# Add some base duration
|
||
|
|
session['duration_seconds'] = 1800 # 30 minutes
|
||
|
|
|
||
|
|
# Add manual adjustments
|
||
|
|
session_manager.add_manual_adjustment(600, 'Extra work') # +10 min
|
||
|
|
session_manager.add_manual_adjustment(-300, 'Break') # -5 min
|
||
|
|
|
||
|
|
# Calculate final duration
|
||
|
|
expected_duration = 1800 + 600 - 300 # 2100 seconds = 35 minutes
|
||
|
|
assert session_manager.get_current_session_duration() == expected_duration
|
||
|
|
|
||
|
|
|
||
|
|
def test_load_sessions_for_task():
|
||
|
|
"""Test loading all sessions for a specific task"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
|
||
|
|
# Create multiple sessions for the same task
|
||
|
|
session1 = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
session1['duration_seconds'] = 1800
|
||
|
|
session1['ended_at'] = '2026-03-10T10:00:00'
|
||
|
|
session_manager.end_session(session1)
|
||
|
|
|
||
|
|
session2 = session_manager.create_session('PRJ-123', 'PRJ')
|
||
|
|
session2['duration_seconds'] = 2400
|
||
|
|
session2['ended_at'] = '2026-03-10T11:00:00'
|
||
|
|
session_manager.end_session(session2)
|
||
|
|
|
||
|
|
# Load sessions for the task
|
||
|
|
sessions = session_manager.load_sessions_for_task('PRJ-123')
|
||
|
|
|
||
|
|
assert len(sessions) == 2
|
||
|
|
# Sessions should be sorted by end time (newest first)
|
||
|
|
assert sessions[0]['duration_seconds'] == 2400
|
||
|
|
assert sessions[1]['duration_seconds'] == 1800
|
||
|
|
|
||
|
|
|
||
|
|
def test_no_sessions_for_task():
|
||
|
|
"""Test that loading sessions for non-existent task returns empty list"""
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
tasks_dir = os.path.join(temp_dir, 'tasks')
|
||
|
|
os.makedirs(tasks_dir, exist_ok=True)
|
||
|
|
|
||
|
|
session_manager = SessionManager(tasks_dir)
|
||
|
|
|
||
|
|
# Load sessions for non-existent task
|
||
|
|
sessions = session_manager.load_sessions_for_task('NONEXISTENT-123')
|
||
|
|
|
||
|
|
assert sessions == []
|