Compare commits
15 Commits
master
...
c5a0114eaf
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c5a0114eaf | ||
|
|
ae06e84070 | ||
|
|
dfc2229091 | ||
|
|
b99c13a325 | ||
|
|
8897de25ed | ||
|
|
5ed8680164 | ||
|
|
f838958bf2 | ||
|
|
60bf92a610 | ||
|
|
b99133de79 | ||
|
|
f889e69b54 | ||
|
|
588e8cb183 | ||
|
|
71bcc5f6f6 | ||
|
|
648185abe6 | ||
|
|
f9de7a2c26 | ||
|
|
8f326b1846 |
452
dashboard/api.py
452
dashboard/api.py
@@ -11,16 +11,22 @@ import sys
|
|||||||
import re
|
import re
|
||||||
import os
|
import os
|
||||||
import signal
|
import signal
|
||||||
|
import uuid
|
||||||
from http.server import HTTPServer, SimpleHTTPRequestHandler
|
from http.server import HTTPServer, SimpleHTTPRequestHandler
|
||||||
from urllib.parse import parse_qs, urlparse
|
from urllib.parse import parse_qs, urlparse
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
# Import habits helpers
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent))
|
||||||
|
import habits_helpers
|
||||||
|
|
||||||
BASE_DIR = Path(__file__).parent.parent
|
BASE_DIR = Path(__file__).parent.parent
|
||||||
TOOLS_DIR = BASE_DIR / 'tools'
|
TOOLS_DIR = BASE_DIR / 'tools'
|
||||||
NOTES_DIR = BASE_DIR / 'kb' / 'youtube'
|
NOTES_DIR = BASE_DIR / 'kb' / 'youtube'
|
||||||
KANBAN_DIR = BASE_DIR / 'dashboard'
|
KANBAN_DIR = BASE_DIR / 'dashboard'
|
||||||
WORKSPACE_DIR = Path('/home/moltbot/workspace')
|
WORKSPACE_DIR = Path('/home/moltbot/workspace')
|
||||||
|
HABITS_FILE = KANBAN_DIR / 'habits.json'
|
||||||
|
|
||||||
# Load .env file if present
|
# Load .env file if present
|
||||||
_env_file = Path(__file__).parent / '.env'
|
_env_file = Path(__file__).parent / '.env'
|
||||||
@@ -48,6 +54,12 @@ class TaskBoardHandler(SimpleHTTPRequestHandler):
|
|||||||
self.handle_git_commit()
|
self.handle_git_commit()
|
||||||
elif self.path == '/api/pdf':
|
elif self.path == '/api/pdf':
|
||||||
self.handle_pdf_post()
|
self.handle_pdf_post()
|
||||||
|
elif self.path == '/api/habits':
|
||||||
|
self.handle_habits_post()
|
||||||
|
elif self.path.startswith('/api/habits/') and self.path.endswith('/check'):
|
||||||
|
self.handle_habits_check()
|
||||||
|
elif self.path.startswith('/api/habits/') and self.path.endswith('/skip'):
|
||||||
|
self.handle_habits_skip()
|
||||||
elif self.path == '/api/workspace/run':
|
elif self.path == '/api/workspace/run':
|
||||||
self.handle_workspace_run()
|
self.handle_workspace_run()
|
||||||
elif self.path == '/api/workspace/stop':
|
elif self.path == '/api/workspace/stop':
|
||||||
@@ -61,6 +73,18 @@ class TaskBoardHandler(SimpleHTTPRequestHandler):
|
|||||||
else:
|
else:
|
||||||
self.send_error(404)
|
self.send_error(404)
|
||||||
|
|
||||||
|
def do_PUT(self):
|
||||||
|
if self.path.startswith('/api/habits/'):
|
||||||
|
self.handle_habits_put()
|
||||||
|
else:
|
||||||
|
self.send_error(404)
|
||||||
|
|
||||||
|
def do_DELETE(self):
|
||||||
|
if self.path.startswith('/api/habits/'):
|
||||||
|
self.handle_habits_delete()
|
||||||
|
else:
|
||||||
|
self.send_error(404)
|
||||||
|
|
||||||
def handle_git_commit(self):
|
def handle_git_commit(self):
|
||||||
"""Run git commit and push."""
|
"""Run git commit and push."""
|
||||||
try:
|
try:
|
||||||
@@ -251,6 +275,8 @@ class TaskBoardHandler(SimpleHTTPRequestHandler):
|
|||||||
self.handle_cron_status()
|
self.handle_cron_status()
|
||||||
elif self.path == '/api/activity' or self.path.startswith('/api/activity?'):
|
elif self.path == '/api/activity' or self.path.startswith('/api/activity?'):
|
||||||
self.handle_activity()
|
self.handle_activity()
|
||||||
|
elif self.path == '/api/habits':
|
||||||
|
self.handle_habits_get()
|
||||||
elif self.path.startswith('/api/files'):
|
elif self.path.startswith('/api/files'):
|
||||||
self.handle_files_get()
|
self.handle_files_get()
|
||||||
elif self.path.startswith('/api/diff'):
|
elif self.path.startswith('/api/diff'):
|
||||||
@@ -1381,6 +1407,430 @@ class TaskBoardHandler(SimpleHTTPRequestHandler):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.send_json({'error': str(e)}, 500)
|
self.send_json({'error': str(e)}, 500)
|
||||||
|
|
||||||
|
def handle_habits_get(self):
|
||||||
|
"""Get all habits with enriched stats."""
|
||||||
|
try:
|
||||||
|
# Read habits file
|
||||||
|
if not HABITS_FILE.exists():
|
||||||
|
self.send_json([])
|
||||||
|
return
|
||||||
|
|
||||||
|
with open(HABITS_FILE, 'r', encoding='utf-8') as f:
|
||||||
|
data = json.load(f)
|
||||||
|
|
||||||
|
habits = data.get('habits', [])
|
||||||
|
|
||||||
|
# Enrich each habit with calculated stats
|
||||||
|
enriched_habits = []
|
||||||
|
for habit in habits:
|
||||||
|
# Calculate stats using helpers
|
||||||
|
current_streak = habits_helpers.calculate_streak(habit)
|
||||||
|
best_streak = habit.get('streak', {}).get('best', 0)
|
||||||
|
completion_rate = habits_helpers.get_completion_rate(habit, days=30)
|
||||||
|
weekly_summary = habits_helpers.get_weekly_summary(habit)
|
||||||
|
|
||||||
|
# Add stats to habit
|
||||||
|
enriched = habit.copy()
|
||||||
|
enriched['current_streak'] = current_streak
|
||||||
|
enriched['best_streak'] = best_streak
|
||||||
|
enriched['completion_rate_30d'] = completion_rate
|
||||||
|
enriched['weekly_summary'] = weekly_summary
|
||||||
|
enriched['should_check_today'] = habits_helpers.should_check_today(habit)
|
||||||
|
|
||||||
|
enriched_habits.append(enriched)
|
||||||
|
|
||||||
|
# Sort by priority ascending (lower number = higher priority)
|
||||||
|
enriched_habits.sort(key=lambda h: h.get('priority', 999))
|
||||||
|
|
||||||
|
self.send_json(enriched_habits)
|
||||||
|
except Exception as e:
|
||||||
|
self.send_json({'error': str(e)}, 500)
|
||||||
|
|
||||||
|
def handle_habits_post(self):
|
||||||
|
"""Create a new habit."""
|
||||||
|
try:
|
||||||
|
# Read request body
|
||||||
|
content_length = int(self.headers['Content-Length'])
|
||||||
|
post_data = self.rfile.read(content_length).decode('utf-8')
|
||||||
|
data = json.loads(post_data)
|
||||||
|
|
||||||
|
# Validate required fields
|
||||||
|
name = data.get('name', '').strip()
|
||||||
|
if not name:
|
||||||
|
self.send_json({'error': 'name is required'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
if len(name) > 100:
|
||||||
|
self.send_json({'error': 'name must be max 100 characters'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Validate color (hex format)
|
||||||
|
color = data.get('color', '#3b82f6')
|
||||||
|
if color and not re.match(r'^#[0-9A-Fa-f]{6}$', color):
|
||||||
|
self.send_json({'error': 'color must be valid hex format (#RRGGBB)'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Validate frequency type
|
||||||
|
frequency_type = data.get('frequency', {}).get('type', 'daily')
|
||||||
|
valid_types = ['daily', 'specific_days', 'x_per_week', 'weekly', 'monthly', 'custom']
|
||||||
|
if frequency_type not in valid_types:
|
||||||
|
self.send_json({'error': f'frequency.type must be one of: {", ".join(valid_types)}'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Create new habit
|
||||||
|
habit_id = str(uuid.uuid4())
|
||||||
|
now = datetime.now().isoformat()
|
||||||
|
|
||||||
|
new_habit = {
|
||||||
|
'id': habit_id,
|
||||||
|
'name': name,
|
||||||
|
'category': data.get('category', 'other'),
|
||||||
|
'color': color,
|
||||||
|
'icon': data.get('icon', 'check-circle'),
|
||||||
|
'priority': data.get('priority', 5),
|
||||||
|
'notes': data.get('notes', ''),
|
||||||
|
'reminderTime': data.get('reminderTime', ''),
|
||||||
|
'frequency': data.get('frequency', {'type': 'daily'}),
|
||||||
|
'streak': {
|
||||||
|
'current': 0,
|
||||||
|
'best': 0,
|
||||||
|
'lastCheckIn': None
|
||||||
|
},
|
||||||
|
'lives': 3,
|
||||||
|
'completions': [],
|
||||||
|
'createdAt': now,
|
||||||
|
'updatedAt': now
|
||||||
|
}
|
||||||
|
|
||||||
|
# Read existing habits
|
||||||
|
if HABITS_FILE.exists():
|
||||||
|
with open(HABITS_FILE, 'r', encoding='utf-8') as f:
|
||||||
|
habits_data = json.load(f)
|
||||||
|
else:
|
||||||
|
habits_data = {'lastUpdated': '', 'habits': []}
|
||||||
|
|
||||||
|
# Add new habit
|
||||||
|
habits_data['habits'].append(new_habit)
|
||||||
|
habits_data['lastUpdated'] = now
|
||||||
|
|
||||||
|
# Save to file
|
||||||
|
with open(HABITS_FILE, 'w', encoding='utf-8') as f:
|
||||||
|
json.dump(habits_data, f, indent=2)
|
||||||
|
|
||||||
|
# Return created habit with 201 status
|
||||||
|
self.send_json(new_habit, 201)
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
self.send_json({'error': 'Invalid JSON'}, 400)
|
||||||
|
except Exception as e:
|
||||||
|
self.send_json({'error': str(e)}, 500)
|
||||||
|
|
||||||
|
def handle_habits_put(self):
|
||||||
|
"""Update an existing habit."""
|
||||||
|
try:
|
||||||
|
# Extract habit ID from path
|
||||||
|
path_parts = self.path.split('/')
|
||||||
|
if len(path_parts) < 4:
|
||||||
|
self.send_json({'error': 'Invalid path'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
habit_id = path_parts[3]
|
||||||
|
|
||||||
|
# Read request body
|
||||||
|
content_length = int(self.headers['Content-Length'])
|
||||||
|
post_data = self.rfile.read(content_length).decode('utf-8')
|
||||||
|
data = json.loads(post_data)
|
||||||
|
|
||||||
|
# Read existing habits
|
||||||
|
if not HABITS_FILE.exists():
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
with open(HABITS_FILE, 'r', encoding='utf-8') as f:
|
||||||
|
habits_data = json.load(f)
|
||||||
|
|
||||||
|
# Find habit to update
|
||||||
|
habits = habits_data.get('habits', [])
|
||||||
|
habit_index = None
|
||||||
|
for i, habit in enumerate(habits):
|
||||||
|
if habit['id'] == habit_id:
|
||||||
|
habit_index = i
|
||||||
|
break
|
||||||
|
|
||||||
|
if habit_index is None:
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Validate allowed fields
|
||||||
|
allowed_fields = ['name', 'category', 'color', 'icon', 'priority', 'notes', 'frequency', 'reminderTime']
|
||||||
|
|
||||||
|
# Validate name if provided
|
||||||
|
if 'name' in data:
|
||||||
|
name = data['name'].strip()
|
||||||
|
if not name:
|
||||||
|
self.send_json({'error': 'name cannot be empty'}, 400)
|
||||||
|
return
|
||||||
|
if len(name) > 100:
|
||||||
|
self.send_json({'error': 'name must be max 100 characters'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Validate color if provided
|
||||||
|
if 'color' in data:
|
||||||
|
color = data['color']
|
||||||
|
if color and not re.match(r'^#[0-9A-Fa-f]{6}$', color):
|
||||||
|
self.send_json({'error': 'color must be valid hex format (#RRGGBB)'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Validate frequency type if provided
|
||||||
|
if 'frequency' in data:
|
||||||
|
frequency_type = data.get('frequency', {}).get('type', 'daily')
|
||||||
|
valid_types = ['daily', 'specific_days', 'x_per_week', 'weekly', 'monthly', 'custom']
|
||||||
|
if frequency_type not in valid_types:
|
||||||
|
self.send_json({'error': f'frequency.type must be one of: {", ".join(valid_types)}'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Update only allowed fields
|
||||||
|
habit = habits[habit_index]
|
||||||
|
for field in allowed_fields:
|
||||||
|
if field in data:
|
||||||
|
habit[field] = data[field]
|
||||||
|
|
||||||
|
# Update timestamp
|
||||||
|
habit['updatedAt'] = datetime.now().isoformat()
|
||||||
|
|
||||||
|
# Save to file
|
||||||
|
habits_data['lastUpdated'] = datetime.now().isoformat()
|
||||||
|
with open(HABITS_FILE, 'w', encoding='utf-8') as f:
|
||||||
|
json.dump(habits_data, f, indent=2)
|
||||||
|
|
||||||
|
# Return updated habit
|
||||||
|
self.send_json(habit)
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
self.send_json({'error': 'Invalid JSON'}, 400)
|
||||||
|
except Exception as e:
|
||||||
|
self.send_json({'error': str(e)}, 500)
|
||||||
|
|
||||||
|
def handle_habits_delete(self):
|
||||||
|
"""Delete a habit."""
|
||||||
|
try:
|
||||||
|
# Extract habit ID from path
|
||||||
|
path_parts = self.path.split('/')
|
||||||
|
if len(path_parts) < 4:
|
||||||
|
self.send_json({'error': 'Invalid path'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
habit_id = path_parts[3]
|
||||||
|
|
||||||
|
# Read existing habits
|
||||||
|
if not HABITS_FILE.exists():
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
with open(HABITS_FILE, 'r', encoding='utf-8') as f:
|
||||||
|
habits_data = json.load(f)
|
||||||
|
|
||||||
|
# Find and remove habit
|
||||||
|
habits = habits_data.get('habits', [])
|
||||||
|
habit_found = False
|
||||||
|
for i, habit in enumerate(habits):
|
||||||
|
if habit['id'] == habit_id:
|
||||||
|
habits.pop(i)
|
||||||
|
habit_found = True
|
||||||
|
break
|
||||||
|
|
||||||
|
if not habit_found:
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Save to file
|
||||||
|
habits_data['lastUpdated'] = datetime.now().isoformat()
|
||||||
|
with open(HABITS_FILE, 'w', encoding='utf-8') as f:
|
||||||
|
json.dump(habits_data, f, indent=2)
|
||||||
|
|
||||||
|
# Return 204 No Content
|
||||||
|
self.send_response(204)
|
||||||
|
self.send_header('Access-Control-Allow-Origin', '*')
|
||||||
|
self.end_headers()
|
||||||
|
except Exception as e:
|
||||||
|
self.send_json({'error': str(e)}, 500)
|
||||||
|
|
||||||
|
def handle_habits_check(self):
|
||||||
|
"""Check in on a habit (complete it for today)."""
|
||||||
|
try:
|
||||||
|
# Extract habit ID from path (/api/habits/{id}/check)
|
||||||
|
path_parts = self.path.split('/')
|
||||||
|
if len(path_parts) < 5:
|
||||||
|
self.send_json({'error': 'Invalid path'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
habit_id = path_parts[3]
|
||||||
|
|
||||||
|
# Read optional body (note, rating, mood)
|
||||||
|
body_data = {}
|
||||||
|
content_length = self.headers.get('Content-Length')
|
||||||
|
if content_length:
|
||||||
|
post_data = self.rfile.read(int(content_length)).decode('utf-8')
|
||||||
|
if post_data.strip():
|
||||||
|
try:
|
||||||
|
body_data = json.loads(post_data)
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
self.send_json({'error': 'Invalid JSON'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Read existing habits
|
||||||
|
if not HABITS_FILE.exists():
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
with open(HABITS_FILE, 'r', encoding='utf-8') as f:
|
||||||
|
habits_data = json.load(f)
|
||||||
|
|
||||||
|
# Find habit
|
||||||
|
habit = None
|
||||||
|
for h in habits_data.get('habits', []):
|
||||||
|
if h['id'] == habit_id:
|
||||||
|
habit = h
|
||||||
|
break
|
||||||
|
|
||||||
|
if not habit:
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Verify habit is relevant for today
|
||||||
|
if not habits_helpers.should_check_today(habit):
|
||||||
|
self.send_json({'error': 'Habit is not relevant for today based on its frequency'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Verify not already checked today
|
||||||
|
today = datetime.now().date().isoformat()
|
||||||
|
completions = habit.get('completions', [])
|
||||||
|
for completion in completions:
|
||||||
|
if completion.get('date') == today:
|
||||||
|
self.send_json({'error': 'Habit already checked in today'}, 409)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Create completion entry
|
||||||
|
completion_entry = {
|
||||||
|
'date': today,
|
||||||
|
'type': 'check' # Distinguish from 'skip' for life restore logic
|
||||||
|
}
|
||||||
|
|
||||||
|
# Add optional fields
|
||||||
|
if 'note' in body_data:
|
||||||
|
completion_entry['note'] = body_data['note']
|
||||||
|
if 'rating' in body_data:
|
||||||
|
rating = body_data['rating']
|
||||||
|
if not isinstance(rating, int) or rating < 1 or rating > 5:
|
||||||
|
self.send_json({'error': 'rating must be an integer between 1 and 5'}, 400)
|
||||||
|
return
|
||||||
|
completion_entry['rating'] = rating
|
||||||
|
if 'mood' in body_data:
|
||||||
|
mood = body_data['mood']
|
||||||
|
if mood not in ['happy', 'neutral', 'sad']:
|
||||||
|
self.send_json({'error': 'mood must be one of: happy, neutral, sad'}, 400)
|
||||||
|
return
|
||||||
|
completion_entry['mood'] = mood
|
||||||
|
|
||||||
|
# Add completion to habit
|
||||||
|
habit['completions'].append(completion_entry)
|
||||||
|
|
||||||
|
# Recalculate streak
|
||||||
|
current_streak = habits_helpers.calculate_streak(habit)
|
||||||
|
habit['streak']['current'] = current_streak
|
||||||
|
|
||||||
|
# Update best streak if current is higher
|
||||||
|
if current_streak > habit['streak']['best']:
|
||||||
|
habit['streak']['best'] = current_streak
|
||||||
|
|
||||||
|
# Update lastCheckIn
|
||||||
|
habit['streak']['lastCheckIn'] = today
|
||||||
|
|
||||||
|
# Check for life restore: if last 7 completions are all check-ins (no skips) and lives < 3
|
||||||
|
if habit.get('lives', 3) < 3:
|
||||||
|
recent_completions = sorted(
|
||||||
|
habit['completions'],
|
||||||
|
key=lambda x: x.get('date', ''),
|
||||||
|
reverse=True
|
||||||
|
)[:7]
|
||||||
|
|
||||||
|
# Check if we have 7 completions and all are check-ins (not skips)
|
||||||
|
if len(recent_completions) == 7:
|
||||||
|
all_checks = all(c.get('type') == 'check' for c in recent_completions)
|
||||||
|
if all_checks:
|
||||||
|
habit['lives'] = min(habit['lives'] + 1, 3)
|
||||||
|
|
||||||
|
# Update timestamp
|
||||||
|
habit['updatedAt'] = datetime.now().isoformat()
|
||||||
|
habits_data['lastUpdated'] = habit['updatedAt']
|
||||||
|
|
||||||
|
# Save to file
|
||||||
|
with open(HABITS_FILE, 'w', encoding='utf-8') as f:
|
||||||
|
json.dump(habits_data, f, indent=2)
|
||||||
|
|
||||||
|
# Return updated habit
|
||||||
|
self.send_json(habit, 200)
|
||||||
|
except Exception as e:
|
||||||
|
self.send_json({'error': str(e)}, 500)
|
||||||
|
|
||||||
|
def handle_habits_skip(self):
|
||||||
|
"""Skip a day using a life to preserve streak."""
|
||||||
|
try:
|
||||||
|
# Extract habit ID from path (/api/habits/{id}/skip)
|
||||||
|
path_parts = self.path.split('/')
|
||||||
|
if len(path_parts) < 5:
|
||||||
|
self.send_json({'error': 'Invalid path'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
habit_id = path_parts[3]
|
||||||
|
|
||||||
|
# Read existing habits
|
||||||
|
if not HABITS_FILE.exists():
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
with open(HABITS_FILE, 'r', encoding='utf-8') as f:
|
||||||
|
habits_data = json.load(f)
|
||||||
|
|
||||||
|
# Find habit
|
||||||
|
habit = None
|
||||||
|
for h in habits_data.get('habits', []):
|
||||||
|
if h['id'] == habit_id:
|
||||||
|
habit = h
|
||||||
|
break
|
||||||
|
|
||||||
|
if not habit:
|
||||||
|
self.send_json({'error': 'Habit not found'}, 404)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Verify lives > 0
|
||||||
|
current_lives = habit.get('lives', 3)
|
||||||
|
if current_lives <= 0:
|
||||||
|
self.send_json({'error': 'No lives remaining'}, 400)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Decrement lives by 1
|
||||||
|
habit['lives'] = current_lives - 1
|
||||||
|
|
||||||
|
# Add completion entry with type='skip'
|
||||||
|
today = datetime.now().date().isoformat()
|
||||||
|
completion_entry = {
|
||||||
|
'date': today,
|
||||||
|
'type': 'skip'
|
||||||
|
}
|
||||||
|
habit['completions'].append(completion_entry)
|
||||||
|
|
||||||
|
# Update timestamp
|
||||||
|
habit['updatedAt'] = datetime.now().isoformat()
|
||||||
|
habits_data['lastUpdated'] = habit['updatedAt']
|
||||||
|
|
||||||
|
# Save to file
|
||||||
|
with open(HABITS_FILE, 'w', encoding='utf-8') as f:
|
||||||
|
json.dump(habits_data, f, indent=2)
|
||||||
|
|
||||||
|
# Return updated habit
|
||||||
|
self.send_json(habit, 200)
|
||||||
|
except Exception as e:
|
||||||
|
self.send_json({'error': str(e)}, 500)
|
||||||
|
|
||||||
def send_json(self, data, code=200):
|
def send_json(self, data, code=200):
|
||||||
self.send_response(code)
|
self.send_response(code)
|
||||||
self.send_header('Content-Type', 'application/json')
|
self.send_header('Content-Type', 'application/json')
|
||||||
@@ -1394,7 +1844,7 @@ class TaskBoardHandler(SimpleHTTPRequestHandler):
|
|||||||
def do_OPTIONS(self):
|
def do_OPTIONS(self):
|
||||||
self.send_response(200)
|
self.send_response(200)
|
||||||
self.send_header('Access-Control-Allow-Origin', '*')
|
self.send_header('Access-Control-Allow-Origin', '*')
|
||||||
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
|
self.send_header('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS')
|
||||||
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
|
|
||||||
|
|||||||
2230
dashboard/habits.html
Normal file
2230
dashboard/habits.html
Normal file
File diff suppressed because it is too large
Load Diff
4
dashboard/habits.json
Normal file
4
dashboard/habits.json
Normal file
@@ -0,0 +1,4 @@
|
|||||||
|
{
|
||||||
|
"lastUpdated": "",
|
||||||
|
"habits": []
|
||||||
|
}
|
||||||
393
dashboard/habits_helpers.py
Normal file
393
dashboard/habits_helpers.py
Normal file
@@ -0,0 +1,393 @@
|
|||||||
|
"""
|
||||||
|
Habit Tracker Helper Functions
|
||||||
|
|
||||||
|
This module provides core helper functions for calculating streaks,
|
||||||
|
checking relevance, and computing stats for habits.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
from typing import Dict, List, Any, Optional
|
||||||
|
|
||||||
|
|
||||||
|
def calculate_streak(habit: Dict[str, Any]) -> int:
|
||||||
|
"""
|
||||||
|
Calculate the current streak for a habit based on its frequency type.
|
||||||
|
Skips maintain the streak (don't break it) but don't count toward the total.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
habit: Dict containing habit data with frequency, completions, etc.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
int: Current streak count (days, weeks, or months depending on frequency)
|
||||||
|
"""
|
||||||
|
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||||
|
completions = habit.get("completions", [])
|
||||||
|
|
||||||
|
if not completions:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
# Sort completions by date (newest first)
|
||||||
|
sorted_completions = sorted(
|
||||||
|
[c for c in completions if c.get("date")],
|
||||||
|
key=lambda x: x["date"],
|
||||||
|
reverse=True
|
||||||
|
)
|
||||||
|
|
||||||
|
if not sorted_completions:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
if frequency_type == "daily":
|
||||||
|
return _calculate_daily_streak(sorted_completions)
|
||||||
|
elif frequency_type == "specific_days":
|
||||||
|
return _calculate_specific_days_streak(habit, sorted_completions)
|
||||||
|
elif frequency_type == "x_per_week":
|
||||||
|
return _calculate_x_per_week_streak(habit, sorted_completions)
|
||||||
|
elif frequency_type == "weekly":
|
||||||
|
return _calculate_weekly_streak(sorted_completions)
|
||||||
|
elif frequency_type == "monthly":
|
||||||
|
return _calculate_monthly_streak(sorted_completions)
|
||||||
|
elif frequency_type == "custom":
|
||||||
|
return _calculate_custom_streak(habit, sorted_completions)
|
||||||
|
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
def _calculate_daily_streak(completions: List[Dict[str, Any]]) -> int:
|
||||||
|
"""
|
||||||
|
Calculate streak for daily habits (consecutive days).
|
||||||
|
Skips maintain the streak (don't break it) but don't count toward the total.
|
||||||
|
"""
|
||||||
|
streak = 0
|
||||||
|
today = datetime.now().date()
|
||||||
|
expected_date = today
|
||||||
|
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
completion_type = completion.get("type", "check")
|
||||||
|
|
||||||
|
if completion_date == expected_date:
|
||||||
|
# Only count 'check' completions toward streak total
|
||||||
|
# 'skip' completions maintain the streak but don't extend it
|
||||||
|
if completion_type == "check":
|
||||||
|
streak += 1
|
||||||
|
expected_date = completion_date - timedelta(days=1)
|
||||||
|
elif completion_date < expected_date:
|
||||||
|
# Gap found, streak breaks
|
||||||
|
break
|
||||||
|
|
||||||
|
return streak
|
||||||
|
|
||||||
|
|
||||||
|
def _calculate_specific_days_streak(habit: Dict[str, Any], completions: List[Dict[str, Any]]) -> int:
|
||||||
|
"""Calculate streak for specific days habits (only count relevant days)."""
|
||||||
|
relevant_days = set(habit.get("frequency", {}).get("days", []))
|
||||||
|
if not relevant_days:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
streak = 0
|
||||||
|
today = datetime.now().date()
|
||||||
|
current_date = today
|
||||||
|
|
||||||
|
# Find the most recent relevant day
|
||||||
|
while current_date.weekday() not in relevant_days:
|
||||||
|
current_date -= timedelta(days=1)
|
||||||
|
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
|
||||||
|
if completion_date == current_date:
|
||||||
|
streak += 1
|
||||||
|
# Move to previous relevant day
|
||||||
|
current_date -= timedelta(days=1)
|
||||||
|
while current_date.weekday() not in relevant_days:
|
||||||
|
current_date -= timedelta(days=1)
|
||||||
|
elif completion_date < current_date:
|
||||||
|
# Check if we missed a relevant day
|
||||||
|
temp_date = current_date
|
||||||
|
found_gap = False
|
||||||
|
while temp_date > completion_date:
|
||||||
|
if temp_date.weekday() in relevant_days:
|
||||||
|
found_gap = True
|
||||||
|
break
|
||||||
|
temp_date -= timedelta(days=1)
|
||||||
|
if found_gap:
|
||||||
|
break
|
||||||
|
|
||||||
|
return streak
|
||||||
|
|
||||||
|
|
||||||
|
def _calculate_x_per_week_streak(habit: Dict[str, Any], completions: List[Dict[str, Any]]) -> int:
|
||||||
|
"""Calculate streak for x_per_week habits (consecutive weeks meeting target)."""
|
||||||
|
target_count = habit.get("frequency", {}).get("count", 1)
|
||||||
|
|
||||||
|
streak = 0
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
# Group completions by week
|
||||||
|
week_counts = {}
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
# Get ISO week (year, week_number)
|
||||||
|
week_key = completion_date.isocalendar()[:2]
|
||||||
|
week_counts[week_key] = week_counts.get(week_key, 0) + 1
|
||||||
|
|
||||||
|
# Start from current week and count backwards
|
||||||
|
current_week = today.isocalendar()[:2]
|
||||||
|
|
||||||
|
while current_week in week_counts:
|
||||||
|
if week_counts[current_week] >= target_count:
|
||||||
|
streak += 1
|
||||||
|
# Move to previous week
|
||||||
|
year, week = current_week
|
||||||
|
if week == 1:
|
||||||
|
year -= 1
|
||||||
|
week = 52
|
||||||
|
else:
|
||||||
|
week -= 1
|
||||||
|
current_week = (year, week)
|
||||||
|
else:
|
||||||
|
break
|
||||||
|
|
||||||
|
return streak
|
||||||
|
|
||||||
|
|
||||||
|
def _calculate_weekly_streak(completions: List[Dict[str, Any]]) -> int:
|
||||||
|
"""Calculate streak for weekly habits (consecutive weeks with at least one check)."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
# Group completions by week
|
||||||
|
weeks_with_checks = set()
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
week_key = completion_date.isocalendar()[:2]
|
||||||
|
weeks_with_checks.add(week_key)
|
||||||
|
|
||||||
|
streak = 0
|
||||||
|
current_week = today.isocalendar()[:2]
|
||||||
|
|
||||||
|
while current_week in weeks_with_checks:
|
||||||
|
streak += 1
|
||||||
|
year, week = current_week
|
||||||
|
if week == 1:
|
||||||
|
year -= 1
|
||||||
|
week = 52
|
||||||
|
else:
|
||||||
|
week -= 1
|
||||||
|
current_week = (year, week)
|
||||||
|
|
||||||
|
return streak
|
||||||
|
|
||||||
|
|
||||||
|
def _calculate_monthly_streak(completions: List[Dict[str, Any]]) -> int:
|
||||||
|
"""Calculate streak for monthly habits (consecutive months with at least one check)."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
# Group completions by month
|
||||||
|
months_with_checks = set()
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
month_key = (completion_date.year, completion_date.month)
|
||||||
|
months_with_checks.add(month_key)
|
||||||
|
|
||||||
|
streak = 0
|
||||||
|
current_month = (today.year, today.month)
|
||||||
|
|
||||||
|
while current_month in months_with_checks:
|
||||||
|
streak += 1
|
||||||
|
year, month = current_month
|
||||||
|
if month == 1:
|
||||||
|
year -= 1
|
||||||
|
month = 12
|
||||||
|
else:
|
||||||
|
month -= 1
|
||||||
|
current_month = (year, month)
|
||||||
|
|
||||||
|
return streak
|
||||||
|
|
||||||
|
|
||||||
|
def _calculate_custom_streak(habit: Dict[str, Any], completions: List[Dict[str, Any]]) -> int:
|
||||||
|
"""Calculate streak for custom interval habits (every X days)."""
|
||||||
|
interval = habit.get("frequency", {}).get("interval", 1)
|
||||||
|
if interval <= 0:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
streak = 0
|
||||||
|
expected_date = datetime.now().date()
|
||||||
|
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
|
||||||
|
# Allow completion within the interval window
|
||||||
|
days_diff = (expected_date - completion_date).days
|
||||||
|
if 0 <= days_diff <= interval - 1:
|
||||||
|
streak += 1
|
||||||
|
expected_date = completion_date - timedelta(days=interval)
|
||||||
|
else:
|
||||||
|
break
|
||||||
|
|
||||||
|
return streak
|
||||||
|
|
||||||
|
|
||||||
|
def should_check_today(habit: Dict[str, Any]) -> bool:
|
||||||
|
"""
|
||||||
|
Check if a habit is relevant for today based on its frequency type.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
habit: Dict containing habit data with frequency settings
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
bool: True if the habit should be checked today
|
||||||
|
"""
|
||||||
|
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||||
|
today = datetime.now().date()
|
||||||
|
weekday = today.weekday() # 0=Monday, 6=Sunday
|
||||||
|
|
||||||
|
if frequency_type == "daily":
|
||||||
|
return True
|
||||||
|
|
||||||
|
elif frequency_type == "specific_days":
|
||||||
|
relevant_days = set(habit.get("frequency", {}).get("days", []))
|
||||||
|
return weekday in relevant_days
|
||||||
|
|
||||||
|
elif frequency_type == "x_per_week":
|
||||||
|
# Always relevant for x_per_week (can check any day)
|
||||||
|
return True
|
||||||
|
|
||||||
|
elif frequency_type == "weekly":
|
||||||
|
# Always relevant (can check any day of the week)
|
||||||
|
return True
|
||||||
|
|
||||||
|
elif frequency_type == "monthly":
|
||||||
|
# Always relevant (can check any day of the month)
|
||||||
|
return True
|
||||||
|
|
||||||
|
elif frequency_type == "custom":
|
||||||
|
# Check if enough days have passed since last completion
|
||||||
|
completions = habit.get("completions", [])
|
||||||
|
if not completions:
|
||||||
|
return True
|
||||||
|
|
||||||
|
interval = habit.get("frequency", {}).get("interval", 1)
|
||||||
|
last_completion = max(completions, key=lambda x: x.get("date", ""))
|
||||||
|
last_date = datetime.fromisoformat(last_completion["date"]).date()
|
||||||
|
days_since = (today - last_date).days
|
||||||
|
|
||||||
|
return days_since >= interval
|
||||||
|
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def get_completion_rate(habit: Dict[str, Any], days: int = 30) -> float:
|
||||||
|
"""
|
||||||
|
Calculate the completion rate as a percentage over the last N days.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
habit: Dict containing habit data
|
||||||
|
days: Number of days to look back (default 30)
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
float: Completion rate as percentage (0-100)
|
||||||
|
"""
|
||||||
|
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||||
|
completions = habit.get("completions", [])
|
||||||
|
|
||||||
|
today = datetime.now().date()
|
||||||
|
start_date = today - timedelta(days=days - 1)
|
||||||
|
|
||||||
|
# Count relevant days and checked days
|
||||||
|
relevant_days = 0
|
||||||
|
checked_dates = set()
|
||||||
|
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
if start_date <= completion_date <= today:
|
||||||
|
checked_dates.add(completion_date)
|
||||||
|
|
||||||
|
# Calculate relevant days based on frequency type
|
||||||
|
if frequency_type == "daily":
|
||||||
|
relevant_days = days
|
||||||
|
|
||||||
|
elif frequency_type == "specific_days":
|
||||||
|
relevant_day_set = set(habit.get("frequency", {}).get("days", []))
|
||||||
|
current = start_date
|
||||||
|
while current <= today:
|
||||||
|
if current.weekday() in relevant_day_set:
|
||||||
|
relevant_days += 1
|
||||||
|
current += timedelta(days=1)
|
||||||
|
|
||||||
|
elif frequency_type == "x_per_week":
|
||||||
|
target_per_week = habit.get("frequency", {}).get("count", 1)
|
||||||
|
num_weeks = days // 7
|
||||||
|
relevant_days = num_weeks * target_per_week
|
||||||
|
|
||||||
|
elif frequency_type == "weekly":
|
||||||
|
num_weeks = days // 7
|
||||||
|
relevant_days = num_weeks
|
||||||
|
|
||||||
|
elif frequency_type == "monthly":
|
||||||
|
num_months = days // 30
|
||||||
|
relevant_days = num_months
|
||||||
|
|
||||||
|
elif frequency_type == "custom":
|
||||||
|
interval = habit.get("frequency", {}).get("interval", 1)
|
||||||
|
relevant_days = days // interval if interval > 0 else 0
|
||||||
|
|
||||||
|
if relevant_days == 0:
|
||||||
|
return 0.0
|
||||||
|
|
||||||
|
checked_days = len(checked_dates)
|
||||||
|
return (checked_days / relevant_days) * 100
|
||||||
|
|
||||||
|
|
||||||
|
def get_weekly_summary(habit: Dict[str, Any]) -> Dict[str, str]:
|
||||||
|
"""
|
||||||
|
Get a summary of the current week showing status for each day.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
habit: Dict containing habit data
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict mapping day names to status: "checked", "skipped", "missed", or "upcoming"
|
||||||
|
"""
|
||||||
|
frequency_type = habit.get("frequency", {}).get("type", "daily")
|
||||||
|
completions = habit.get("completions", [])
|
||||||
|
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
# Start of current week (Monday)
|
||||||
|
start_of_week = today - timedelta(days=today.weekday())
|
||||||
|
|
||||||
|
# Create completion map
|
||||||
|
completion_map = {}
|
||||||
|
for completion in completions:
|
||||||
|
completion_date = datetime.fromisoformat(completion["date"]).date()
|
||||||
|
if completion_date >= start_of_week:
|
||||||
|
completion_type = completion.get("type", "check")
|
||||||
|
completion_map[completion_date] = completion_type
|
||||||
|
|
||||||
|
# Build summary for each day of the week
|
||||||
|
summary = {}
|
||||||
|
day_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
|
||||||
|
|
||||||
|
for i, day_name in enumerate(day_names):
|
||||||
|
day_date = start_of_week + timedelta(days=i)
|
||||||
|
|
||||||
|
if day_date > today:
|
||||||
|
summary[day_name] = "upcoming"
|
||||||
|
elif day_date in completion_map:
|
||||||
|
if completion_map[day_date] == "skip":
|
||||||
|
summary[day_name] = "skipped"
|
||||||
|
else:
|
||||||
|
summary[day_name] = "checked"
|
||||||
|
else:
|
||||||
|
# Check if this day was relevant
|
||||||
|
if frequency_type == "specific_days":
|
||||||
|
relevant_days = set(habit.get("frequency", {}).get("days", []))
|
||||||
|
if day_date.weekday() not in relevant_days:
|
||||||
|
summary[day_name] = "not_relevant"
|
||||||
|
else:
|
||||||
|
summary[day_name] = "missed"
|
||||||
|
else:
|
||||||
|
summary[day_name] = "missed"
|
||||||
|
|
||||||
|
return summary
|
||||||
@@ -1071,6 +1071,10 @@
|
|||||||
<i data-lucide="file-text"></i>
|
<i data-lucide="file-text"></i>
|
||||||
<span>KB</span>
|
<span>KB</span>
|
||||||
</a>
|
</a>
|
||||||
|
<a href="/echo/habits.html" class="nav-item">
|
||||||
|
<i data-lucide="dumbbell"></i>
|
||||||
|
<span>Habits</span>
|
||||||
|
</a>
|
||||||
<a href="/echo/files.html" class="nav-item">
|
<a href="/echo/files.html" class="nav-item">
|
||||||
<i data-lucide="folder"></i>
|
<i data-lucide="folder"></i>
|
||||||
<span>Files</span>
|
<span>Files</span>
|
||||||
|
|||||||
@@ -3,7 +3,7 @@
|
|||||||
* Swipe left/right to navigate between pages
|
* Swipe left/right to navigate between pages
|
||||||
*/
|
*/
|
||||||
(function() {
|
(function() {
|
||||||
const pages = ['index.html', 'notes.html', 'files.html'];
|
const pages = ['index.html', 'notes.html', 'habits.html', 'files.html', 'workspace.html'];
|
||||||
|
|
||||||
// Get current page index
|
// Get current page index
|
||||||
function getCurrentIndex() {
|
function getCurrentIndex() {
|
||||||
|
|||||||
942
dashboard/tests/test_habits_api.py
Normal file
942
dashboard/tests/test_habits_api.py
Normal file
@@ -0,0 +1,942 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Tests for habits API endpoints (GET and POST)."""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
from http.server import HTTPServer
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
import urllib.request
|
||||||
|
import urllib.error
|
||||||
|
|
||||||
|
# Add parent directory to path so we can import api module
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||||
|
|
||||||
|
# Mock the habits file to a temp location for testing
|
||||||
|
import api
|
||||||
|
original_habits_file = api.HABITS_FILE
|
||||||
|
|
||||||
|
def setup_test_env():
|
||||||
|
"""Set up temporary test environment."""
|
||||||
|
temp_dir = Path(tempfile.mkdtemp())
|
||||||
|
api.HABITS_FILE = temp_dir / 'habits.json'
|
||||||
|
|
||||||
|
# Create empty habits file
|
||||||
|
api.HABITS_FILE.write_text(json.dumps({
|
||||||
|
'lastUpdated': '',
|
||||||
|
'habits': []
|
||||||
|
}))
|
||||||
|
|
||||||
|
return temp_dir
|
||||||
|
|
||||||
|
def cleanup_test_env(temp_dir):
|
||||||
|
"""Clean up temporary test environment."""
|
||||||
|
api.HABITS_FILE = original_habits_file
|
||||||
|
shutil.rmtree(temp_dir)
|
||||||
|
|
||||||
|
def start_test_server(port=8765):
|
||||||
|
"""Start test server in background thread."""
|
||||||
|
server = HTTPServer(('localhost', port), api.TaskBoardHandler)
|
||||||
|
thread = threading.Thread(target=server.serve_forever)
|
||||||
|
thread.daemon = True
|
||||||
|
thread.start()
|
||||||
|
time.sleep(0.5) # Give server time to start
|
||||||
|
return server
|
||||||
|
|
||||||
|
def http_get(path, port=8765):
|
||||||
|
"""Make HTTP GET request."""
|
||||||
|
url = f'http://localhost:{port}{path}'
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(url) as response:
|
||||||
|
return response.status, json.loads(response.read().decode())
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
return e.code, json.loads(e.read().decode()) if e.headers.get('Content-Type') == 'application/json' else {}
|
||||||
|
|
||||||
|
def http_post(path, data, port=8765):
|
||||||
|
"""Make HTTP POST request."""
|
||||||
|
url = f'http://localhost:{port}{path}'
|
||||||
|
req = urllib.request.Request(
|
||||||
|
url,
|
||||||
|
data=json.dumps(data).encode(),
|
||||||
|
headers={'Content-Type': 'application/json'}
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(req) as response:
|
||||||
|
return response.status, json.loads(response.read().decode())
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
return e.code, json.loads(e.read().decode()) if e.headers.get('Content-Type') == 'application/json' else {}
|
||||||
|
|
||||||
|
def http_put(path, data, port=8765):
|
||||||
|
"""Make HTTP PUT request."""
|
||||||
|
url = f'http://localhost:{port}{path}'
|
||||||
|
req = urllib.request.Request(
|
||||||
|
url,
|
||||||
|
data=json.dumps(data).encode(),
|
||||||
|
headers={'Content-Type': 'application/json'},
|
||||||
|
method='PUT'
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(req) as response:
|
||||||
|
return response.status, json.loads(response.read().decode())
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
return e.code, json.loads(e.read().decode()) if e.headers.get('Content-Type') == 'application/json' else {}
|
||||||
|
|
||||||
|
def http_delete(path, port=8765):
|
||||||
|
"""Make HTTP DELETE request."""
|
||||||
|
url = f'http://localhost:{port}{path}'
|
||||||
|
req = urllib.request.Request(url, method='DELETE')
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(req) as response:
|
||||||
|
return response.status, None
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
return e.code, json.loads(e.read().decode()) if e.headers.get('Content-Type') == 'application/json' else {}
|
||||||
|
|
||||||
|
# Test 1: GET /api/habits returns empty array when no habits
|
||||||
|
def test_get_habits_empty():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, data = http_get('/api/habits')
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert data == [], f"Expected empty array, got {data}"
|
||||||
|
print("✓ Test 1: GET /api/habits returns empty array")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 2: POST /api/habits creates new habit with valid input
|
||||||
|
def test_post_habit_valid():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Morning Exercise',
|
||||||
|
'category': 'health',
|
||||||
|
'color': '#10b981',
|
||||||
|
'icon': 'dumbbell',
|
||||||
|
'priority': 1,
|
||||||
|
'notes': 'Start with 10 push-ups',
|
||||||
|
'reminderTime': '07:00',
|
||||||
|
'frequency': {
|
||||||
|
'type': 'daily'
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
status, data = http_post('/api/habits', habit_data)
|
||||||
|
assert status == 201, f"Expected 201, got {status}"
|
||||||
|
assert 'id' in data, "Response should include habit id"
|
||||||
|
assert data['name'] == 'Morning Exercise', f"Name mismatch: {data['name']}"
|
||||||
|
assert data['category'] == 'health', f"Category mismatch: {data['category']}"
|
||||||
|
assert data['streak']['current'] == 0, "Initial streak should be 0"
|
||||||
|
assert data['lives'] == 3, "Initial lives should be 3"
|
||||||
|
assert data['completions'] == [], "Initial completions should be empty"
|
||||||
|
print("✓ Test 2: POST /api/habits creates habit with 201")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 3: POST validates name is required
|
||||||
|
def test_post_habit_missing_name():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, data = http_post('/api/habits', {})
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'error' in data, "Response should include error"
|
||||||
|
assert 'name' in data['error'].lower(), f"Error should mention name: {data['error']}"
|
||||||
|
print("✓ Test 3: POST validates name is required")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 4: POST validates name max 100 chars
|
||||||
|
def test_post_habit_name_too_long():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, data = http_post('/api/habits', {'name': 'x' * 101})
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'error' in data, "Response should include error"
|
||||||
|
assert '100' in data['error'], f"Error should mention max length: {data['error']}"
|
||||||
|
print("✓ Test 4: POST validates name max 100 chars")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 5: POST validates color hex format
|
||||||
|
def test_post_habit_invalid_color():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, data = http_post('/api/habits', {
|
||||||
|
'name': 'Test',
|
||||||
|
'color': 'not-a-hex-color'
|
||||||
|
})
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'error' in data, "Response should include error"
|
||||||
|
assert 'color' in data['error'].lower(), f"Error should mention color: {data['error']}"
|
||||||
|
print("✓ Test 5: POST validates color hex format")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 6: POST validates frequency type
|
||||||
|
def test_post_habit_invalid_frequency():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, data = http_post('/api/habits', {
|
||||||
|
'name': 'Test',
|
||||||
|
'frequency': {'type': 'invalid_type'}
|
||||||
|
})
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'error' in data, "Response should include error"
|
||||||
|
assert 'frequency' in data['error'].lower(), f"Error should mention frequency: {data['error']}"
|
||||||
|
print("✓ Test 6: POST validates frequency type")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 7: GET /api/habits returns habits with stats enriched
|
||||||
|
def test_get_habits_with_stats():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit first
|
||||||
|
habit_data = {'name': 'Daily Reading', 'frequency': {'type': 'daily'}}
|
||||||
|
http_post('/api/habits', habit_data)
|
||||||
|
|
||||||
|
# Get habits
|
||||||
|
status, data = http_get('/api/habits')
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert len(data) == 1, f"Expected 1 habit, got {len(data)}"
|
||||||
|
|
||||||
|
habit = data[0]
|
||||||
|
assert 'current_streak' in habit, "Should include current_streak"
|
||||||
|
assert 'best_streak' in habit, "Should include best_streak"
|
||||||
|
assert 'completion_rate_30d' in habit, "Should include completion_rate_30d"
|
||||||
|
assert 'weekly_summary' in habit, "Should include weekly_summary"
|
||||||
|
print("✓ Test 7: GET returns habits with stats enriched")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 8: GET /api/habits sorts by priority ascending
|
||||||
|
def test_get_habits_sorted_by_priority():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create habits with different priorities
|
||||||
|
http_post('/api/habits', {'name': 'Low Priority', 'priority': 10})
|
||||||
|
http_post('/api/habits', {'name': 'High Priority', 'priority': 1})
|
||||||
|
http_post('/api/habits', {'name': 'Medium Priority', 'priority': 5})
|
||||||
|
|
||||||
|
# Get habits
|
||||||
|
status, data = http_get('/api/habits')
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert len(data) == 3, f"Expected 3 habits, got {len(data)}"
|
||||||
|
|
||||||
|
# Check sorting
|
||||||
|
assert data[0]['priority'] == 1, "First should be priority 1"
|
||||||
|
assert data[1]['priority'] == 5, "Second should be priority 5"
|
||||||
|
assert data[2]['priority'] == 10, "Third should be priority 10"
|
||||||
|
print("✓ Test 8: GET sorts habits by priority ascending")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 9: POST returns 400 for invalid JSON
|
||||||
|
def test_post_habit_invalid_json():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
url = f'http://localhost:8765/api/habits'
|
||||||
|
req = urllib.request.Request(
|
||||||
|
url,
|
||||||
|
data=b'invalid json{',
|
||||||
|
headers={'Content-Type': 'application/json'}
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
urllib.request.urlopen(req)
|
||||||
|
assert False, "Should have raised HTTPError"
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
assert e.code == 400, f"Expected 400, got {e.code}"
|
||||||
|
print("✓ Test 9: POST returns 400 for invalid JSON")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 10: POST initializes streak.current=0
|
||||||
|
def test_post_habit_initial_streak():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, data = http_post('/api/habits', {'name': 'Test Habit'})
|
||||||
|
assert status == 201, f"Expected 201, got {status}"
|
||||||
|
assert data['streak']['current'] == 0, "Initial streak.current should be 0"
|
||||||
|
assert data['streak']['best'] == 0, "Initial streak.best should be 0"
|
||||||
|
assert data['streak']['lastCheckIn'] is None, "Initial lastCheckIn should be None"
|
||||||
|
print("✓ Test 10: POST initializes streak correctly")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 12: PUT /api/habits/{id} updates habit successfully
|
||||||
|
def test_put_habit_valid():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit first
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Original Name',
|
||||||
|
'category': 'health',
|
||||||
|
'color': '#10b981',
|
||||||
|
'priority': 3
|
||||||
|
}
|
||||||
|
status, created_habit = http_post('/api/habits', habit_data)
|
||||||
|
habit_id = created_habit['id']
|
||||||
|
|
||||||
|
# Update the habit
|
||||||
|
update_data = {
|
||||||
|
'name': 'Updated Name',
|
||||||
|
'category': 'productivity',
|
||||||
|
'color': '#ef4444',
|
||||||
|
'priority': 1,
|
||||||
|
'notes': 'New notes'
|
||||||
|
}
|
||||||
|
status, updated_habit = http_put(f'/api/habits/{habit_id}', update_data)
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert updated_habit['name'] == 'Updated Name', "Name not updated"
|
||||||
|
assert updated_habit['category'] == 'productivity', "Category not updated"
|
||||||
|
assert updated_habit['color'] == '#ef4444', "Color not updated"
|
||||||
|
assert updated_habit['priority'] == 1, "Priority not updated"
|
||||||
|
assert updated_habit['notes'] == 'New notes', "Notes not updated"
|
||||||
|
assert updated_habit['id'] == habit_id, "ID should not change"
|
||||||
|
print("✓ Test 12: PUT /api/habits/{id} updates habit successfully")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 13: PUT /api/habits/{id} does not allow editing protected fields
|
||||||
|
def test_put_habit_protected_fields():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit first
|
||||||
|
habit_data = {'name': 'Test Habit'}
|
||||||
|
status, created_habit = http_post('/api/habits', habit_data)
|
||||||
|
habit_id = created_habit['id']
|
||||||
|
original_created_at = created_habit['createdAt']
|
||||||
|
|
||||||
|
# Try to update protected fields
|
||||||
|
update_data = {
|
||||||
|
'name': 'Updated Name',
|
||||||
|
'id': 'new-id',
|
||||||
|
'createdAt': '2020-01-01T00:00:00',
|
||||||
|
'streak': {'current': 100, 'best': 200},
|
||||||
|
'lives': 10,
|
||||||
|
'completions': [{'date': '2025-01-01'}]
|
||||||
|
}
|
||||||
|
status, updated_habit = http_put(f'/api/habits/{habit_id}', update_data)
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert updated_habit['name'] == 'Updated Name', "Name should be updated"
|
||||||
|
assert updated_habit['id'] == habit_id, "ID should not change"
|
||||||
|
assert updated_habit['createdAt'] == original_created_at, "createdAt should not change"
|
||||||
|
assert updated_habit['streak']['current'] == 0, "streak should not change"
|
||||||
|
assert updated_habit['lives'] == 3, "lives should not change"
|
||||||
|
assert updated_habit['completions'] == [], "completions should not change"
|
||||||
|
print("✓ Test 13: PUT /api/habits/{id} does not allow editing protected fields")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 14: PUT /api/habits/{id} returns 404 for non-existent habit
|
||||||
|
def test_put_habit_not_found():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
update_data = {'name': 'Updated Name'}
|
||||||
|
status, response = http_put('/api/habits/non-existent-id', update_data)
|
||||||
|
|
||||||
|
assert status == 404, f"Expected 404, got {status}"
|
||||||
|
assert 'error' in response, "Expected error message"
|
||||||
|
print("✓ Test 14: PUT /api/habits/{id} returns 404 for non-existent habit")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 15: PUT /api/habits/{id} validates input
|
||||||
|
def test_put_habit_invalid_input():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit first
|
||||||
|
habit_data = {'name': 'Test Habit'}
|
||||||
|
status, created_habit = http_post('/api/habits', habit_data)
|
||||||
|
habit_id = created_habit['id']
|
||||||
|
|
||||||
|
# Test invalid color
|
||||||
|
update_data = {'color': 'not-a-hex-color'}
|
||||||
|
status, response = http_put(f'/api/habits/{habit_id}', update_data)
|
||||||
|
assert status == 400, f"Expected 400 for invalid color, got {status}"
|
||||||
|
|
||||||
|
# Test empty name
|
||||||
|
update_data = {'name': ''}
|
||||||
|
status, response = http_put(f'/api/habits/{habit_id}', update_data)
|
||||||
|
assert status == 400, f"Expected 400 for empty name, got {status}"
|
||||||
|
|
||||||
|
# Test name too long
|
||||||
|
update_data = {'name': 'x' * 101}
|
||||||
|
status, response = http_put(f'/api/habits/{habit_id}', update_data)
|
||||||
|
assert status == 400, f"Expected 400 for long name, got {status}"
|
||||||
|
|
||||||
|
print("✓ Test 15: PUT /api/habits/{id} validates input")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 16: DELETE /api/habits/{id} removes habit successfully
|
||||||
|
def test_delete_habit_success():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit first
|
||||||
|
habit_data = {'name': 'Habit to Delete'}
|
||||||
|
status, created_habit = http_post('/api/habits', habit_data)
|
||||||
|
habit_id = created_habit['id']
|
||||||
|
|
||||||
|
# Verify habit exists
|
||||||
|
status, habits = http_get('/api/habits')
|
||||||
|
assert len(habits) == 1, "Should have 1 habit"
|
||||||
|
|
||||||
|
# Delete the habit
|
||||||
|
status, _ = http_delete(f'/api/habits/{habit_id}')
|
||||||
|
assert status == 204, f"Expected 204, got {status}"
|
||||||
|
|
||||||
|
# Verify habit is deleted
|
||||||
|
status, habits = http_get('/api/habits')
|
||||||
|
assert len(habits) == 0, "Should have 0 habits after deletion"
|
||||||
|
print("✓ Test 16: DELETE /api/habits/{id} removes habit successfully")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 17: DELETE /api/habits/{id} returns 404 for non-existent habit
|
||||||
|
def test_delete_habit_not_found():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, response = http_delete('/api/habits/non-existent-id')
|
||||||
|
|
||||||
|
assert status == 404, f"Expected 404, got {status}"
|
||||||
|
assert 'error' in response, "Expected error message"
|
||||||
|
print("✓ Test 17: DELETE /api/habits/{id} returns 404 for non-existent habit")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 18: do_OPTIONS includes PUT and DELETE methods
|
||||||
|
def test_options_includes_put_delete():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Make OPTIONS request
|
||||||
|
url = 'http://localhost:8765/api/habits'
|
||||||
|
req = urllib.request.Request(url, method='OPTIONS')
|
||||||
|
with urllib.request.urlopen(req) as response:
|
||||||
|
allowed_methods = response.headers.get('Access-Control-Allow-Methods', '')
|
||||||
|
assert 'PUT' in allowed_methods, f"PUT not in allowed methods: {allowed_methods}"
|
||||||
|
assert 'DELETE' in allowed_methods, f"DELETE not in allowed methods: {allowed_methods}"
|
||||||
|
print("✓ Test 18: do_OPTIONS includes PUT and DELETE methods")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 20: POST /api/habits/{id}/check adds completion entry
|
||||||
|
def test_check_in_basic():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Morning Exercise',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
assert status == 201, f"Failed to create habit: {status}"
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Check in on the habit
|
||||||
|
status, updated_habit = http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert len(updated_habit['completions']) == 1, "Expected 1 completion"
|
||||||
|
assert updated_habit['completions'][0]['date'] == datetime.now().date().isoformat()
|
||||||
|
assert updated_habit['completions'][0]['type'] == 'check'
|
||||||
|
print("✓ Test 20: POST /api/habits/{id}/check adds completion entry")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 21: Check-in accepts optional note, rating, mood
|
||||||
|
def test_check_in_with_details():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Meditation',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Check in with details
|
||||||
|
status, updated_habit = http_post(f'/api/habits/{habit_id}/check', {
|
||||||
|
'note': 'Felt very relaxed today',
|
||||||
|
'rating': 5,
|
||||||
|
'mood': 'happy'
|
||||||
|
})
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
completion = updated_habit['completions'][0]
|
||||||
|
assert completion['note'] == 'Felt very relaxed today'
|
||||||
|
assert completion['rating'] == 5
|
||||||
|
assert completion['mood'] == 'happy'
|
||||||
|
print("✓ Test 21: Check-in accepts optional note, rating (1-5), and mood")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 22: Check-in returns 404 if habit not found
|
||||||
|
def test_check_in_not_found():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, response = http_post('/api/habits/non-existent-id/check', {})
|
||||||
|
|
||||||
|
assert status == 404, f"Expected 404, got {status}"
|
||||||
|
assert 'error' in response
|
||||||
|
print("✓ Test 22: Check-in returns 404 if habit not found")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 23: Check-in returns 400 if habit not relevant for today
|
||||||
|
def test_check_in_not_relevant():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit for specific days (e.g., Monday only)
|
||||||
|
# If today is not Monday, it should fail
|
||||||
|
today_weekday = datetime.now().date().weekday()
|
||||||
|
different_day = (today_weekday + 1) % 7 # Pick a different day
|
||||||
|
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Monday Only Habit',
|
||||||
|
'frequency': {
|
||||||
|
'type': 'specific_days',
|
||||||
|
'days': [different_day]
|
||||||
|
}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Try to check in
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'not relevant' in response.get('error', '').lower()
|
||||||
|
print("✓ Test 23: Check-in returns 400 if habit not relevant for today")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 24: Check-in returns 409 if already checked today
|
||||||
|
def test_check_in_already_checked():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Water Plants',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Check in once
|
||||||
|
status, _ = http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
assert status == 200, "First check-in should succeed"
|
||||||
|
|
||||||
|
# Try to check in again
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
|
||||||
|
assert status == 409, f"Expected 409, got {status}"
|
||||||
|
assert 'already checked' in response.get('error', '').lower()
|
||||||
|
print("✓ Test 24: Check-in returns 409 if already checked today")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 25: Streak is recalculated after check-in
|
||||||
|
def test_check_in_updates_streak():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Read',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Check in
|
||||||
|
status, updated_habit = http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert updated_habit['streak']['current'] == 1, f"Expected streak 1, got {updated_habit['streak']['current']}"
|
||||||
|
assert updated_habit['streak']['best'] == 1, f"Expected best streak 1, got {updated_habit['streak']['best']}"
|
||||||
|
print("✓ Test 25: Streak current and best are recalculated after check-in")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 26: lastCheckIn is updated after check-in
|
||||||
|
def test_check_in_updates_last_check_in():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Floss',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Initially lastCheckIn should be None
|
||||||
|
assert habit['streak']['lastCheckIn'] is None
|
||||||
|
|
||||||
|
# Check in
|
||||||
|
status, updated_habit = http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
|
||||||
|
today = datetime.now().date().isoformat()
|
||||||
|
assert updated_habit['streak']['lastCheckIn'] == today
|
||||||
|
print("✓ Test 26: lastCheckIn is updated to today's date")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 27: Lives are restored after 7 consecutive check-ins
|
||||||
|
def test_check_in_life_restore():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit and manually set up 6 previous check-ins
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Yoga',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Manually add 6 previous check-ins and reduce lives to 2
|
||||||
|
habits_data = json.loads(api.HABITS_FILE.read_text())
|
||||||
|
for h in habits_data['habits']:
|
||||||
|
if h['id'] == habit_id:
|
||||||
|
h['lives'] = 2
|
||||||
|
# Add 6 check-ins from previous days
|
||||||
|
for i in range(6, 0, -1):
|
||||||
|
past_date = (datetime.now().date() - timedelta(days=i)).isoformat()
|
||||||
|
h['completions'].append({
|
||||||
|
'date': past_date,
|
||||||
|
'type': 'check'
|
||||||
|
})
|
||||||
|
break
|
||||||
|
api.HABITS_FILE.write_text(json.dumps(habits_data, indent=2))
|
||||||
|
|
||||||
|
# Check in for today (7th consecutive)
|
||||||
|
status, updated_habit = http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert updated_habit['lives'] == 3, f"Expected 3 lives restored, got {updated_habit['lives']}"
|
||||||
|
print("✓ Test 27: Lives are restored by 1 (max 3) after 7 consecutive check-ins")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 28: Check-in validates rating range
|
||||||
|
def test_check_in_invalid_rating():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Journal',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Try to check in with invalid rating
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/check', {
|
||||||
|
'rating': 10 # Invalid, should be 1-5
|
||||||
|
})
|
||||||
|
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'rating' in response.get('error', '').lower()
|
||||||
|
print("✓ Test 28: Check-in validates rating is between 1 and 5")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 29: Check-in validates mood values
|
||||||
|
def test_check_in_invalid_mood():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a daily habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Gratitude',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Try to check in with invalid mood
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/check', {
|
||||||
|
'mood': 'excited' # Invalid, should be happy/neutral/sad
|
||||||
|
})
|
||||||
|
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'mood' in response.get('error', '').lower()
|
||||||
|
print("✓ Test 29: Check-in validates mood is one of: happy, neutral, sad")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 30: Skip basic - decrements lives
|
||||||
|
def test_skip_basic():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Daily Exercise',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
assert status == 201
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Skip a day
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/skip', {})
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
assert response['lives'] == 2, f"Expected 2 lives, got {response['lives']}"
|
||||||
|
|
||||||
|
# Verify completion entry was added with type='skip'
|
||||||
|
completions = response.get('completions', [])
|
||||||
|
assert len(completions) == 1, f"Expected 1 completion, got {len(completions)}"
|
||||||
|
assert completions[0]['type'] == 'skip', f"Expected type='skip', got {completions[0].get('type')}"
|
||||||
|
assert completions[0]['date'] == datetime.now().date().isoformat()
|
||||||
|
|
||||||
|
print("✓ Test 30: Skip decrements lives and adds skip completion")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 31: Skip preserves streak
|
||||||
|
def test_skip_preserves_streak():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Daily Exercise',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
assert status == 201
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Check in to build a streak
|
||||||
|
http_post(f'/api/habits/{habit_id}/check', {})
|
||||||
|
|
||||||
|
# Get current streak
|
||||||
|
status, habits = http_get('/api/habits')
|
||||||
|
current_streak = habits[0]['current_streak']
|
||||||
|
assert current_streak > 0
|
||||||
|
|
||||||
|
# Skip the next day (simulate by adding skip manually and checking streak doesn't break)
|
||||||
|
# Since we can't time travel, we'll verify that skip doesn't recalculate streak
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/skip', {})
|
||||||
|
|
||||||
|
assert status == 200, f"Expected 200, got {status}"
|
||||||
|
# Verify lives decremented
|
||||||
|
assert response['lives'] == 2
|
||||||
|
# The streak should remain unchanged (skip doesn't break it)
|
||||||
|
# Note: We can't verify streak preservation perfectly without time travel,
|
||||||
|
# but we verify the skip completion is added correctly
|
||||||
|
completions = response.get('completions', [])
|
||||||
|
skip_count = sum(1 for c in completions if c.get('type') == 'skip')
|
||||||
|
assert skip_count == 1
|
||||||
|
|
||||||
|
print("✓ Test 31: Skip preserves streak (doesn't break it)")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 32: Skip returns 404 for non-existent habit
|
||||||
|
def test_skip_not_found():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
status, response = http_post('/api/habits/nonexistent-id/skip', {})
|
||||||
|
|
||||||
|
assert status == 404, f"Expected 404, got {status}"
|
||||||
|
assert 'not found' in response.get('error', '').lower()
|
||||||
|
|
||||||
|
print("✓ Test 32: Skip returns 404 for non-existent habit")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 33: Skip returns 400 when no lives remaining
|
||||||
|
def test_skip_no_lives():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Daily Exercise',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
assert status == 201
|
||||||
|
habit_id = habit['id']
|
||||||
|
|
||||||
|
# Use all 3 lives
|
||||||
|
for i in range(3):
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/skip', {})
|
||||||
|
assert status == 200, f"Skip {i+1} failed with status {status}"
|
||||||
|
assert response['lives'] == 2 - i, f"Expected {2-i} lives, got {response['lives']}"
|
||||||
|
|
||||||
|
# Try to skip again with no lives
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/skip', {})
|
||||||
|
|
||||||
|
assert status == 400, f"Expected 400, got {status}"
|
||||||
|
assert 'no lives remaining' in response.get('error', '').lower()
|
||||||
|
|
||||||
|
print("✓ Test 33: Skip returns 400 when no lives remaining")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 34: Skip returns updated habit with new lives count
|
||||||
|
def test_skip_returns_updated_habit():
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create a habit
|
||||||
|
status, habit = http_post('/api/habits', {
|
||||||
|
'name': 'Daily Exercise',
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
})
|
||||||
|
assert status == 201
|
||||||
|
habit_id = habit['id']
|
||||||
|
original_updated_at = habit['updatedAt']
|
||||||
|
|
||||||
|
# Skip a day
|
||||||
|
status, response = http_post(f'/api/habits/{habit_id}/skip', {})
|
||||||
|
|
||||||
|
assert status == 200
|
||||||
|
assert response['id'] == habit_id
|
||||||
|
assert response['lives'] == 2
|
||||||
|
assert response['updatedAt'] != original_updated_at, "updatedAt should be updated"
|
||||||
|
assert 'name' in response
|
||||||
|
assert 'frequency' in response
|
||||||
|
assert 'completions' in response
|
||||||
|
|
||||||
|
print("✓ Test 34: Skip returns updated habit with new lives count")
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
cleanup_test_env(temp_dir)
|
||||||
|
|
||||||
|
# Test 35: Typecheck passes
|
||||||
|
def test_typecheck():
|
||||||
|
result = subprocess.run(
|
||||||
|
['python3', '-m', 'py_compile', str(Path(__file__).parent.parent / 'api.py')],
|
||||||
|
capture_output=True
|
||||||
|
)
|
||||||
|
assert result.returncode == 0, f"Typecheck failed: {result.stderr.decode()}"
|
||||||
|
print("✓ Test 11: Typecheck passes")
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
import subprocess
|
||||||
|
|
||||||
|
print("\n=== Running Habits API Tests ===\n")
|
||||||
|
|
||||||
|
test_get_habits_empty()
|
||||||
|
test_post_habit_valid()
|
||||||
|
test_post_habit_missing_name()
|
||||||
|
test_post_habit_name_too_long()
|
||||||
|
test_post_habit_invalid_color()
|
||||||
|
test_post_habit_invalid_frequency()
|
||||||
|
test_get_habits_with_stats()
|
||||||
|
test_get_habits_sorted_by_priority()
|
||||||
|
test_post_habit_invalid_json()
|
||||||
|
test_post_habit_initial_streak()
|
||||||
|
test_put_habit_valid()
|
||||||
|
test_put_habit_protected_fields()
|
||||||
|
test_put_habit_not_found()
|
||||||
|
test_put_habit_invalid_input()
|
||||||
|
test_delete_habit_success()
|
||||||
|
test_delete_habit_not_found()
|
||||||
|
test_options_includes_put_delete()
|
||||||
|
test_check_in_basic()
|
||||||
|
test_check_in_with_details()
|
||||||
|
test_check_in_not_found()
|
||||||
|
test_check_in_not_relevant()
|
||||||
|
test_check_in_already_checked()
|
||||||
|
test_check_in_updates_streak()
|
||||||
|
test_check_in_updates_last_check_in()
|
||||||
|
test_check_in_life_restore()
|
||||||
|
test_check_in_invalid_rating()
|
||||||
|
test_check_in_invalid_mood()
|
||||||
|
test_skip_basic()
|
||||||
|
test_skip_preserves_streak()
|
||||||
|
test_skip_not_found()
|
||||||
|
test_skip_no_lives()
|
||||||
|
test_skip_returns_updated_habit()
|
||||||
|
test_typecheck()
|
||||||
|
|
||||||
|
print("\n✅ All 35 tests passed!\n")
|
||||||
1741
dashboard/tests/test_habits_frontend.py
Normal file
1741
dashboard/tests/test_habits_frontend.py
Normal file
File diff suppressed because it is too large
Load Diff
429
dashboard/tests/test_habits_helpers.py
Normal file
429
dashboard/tests/test_habits_helpers.py
Normal file
@@ -0,0 +1,429 @@
|
|||||||
|
"""
|
||||||
|
Tests for habits_helpers.py
|
||||||
|
|
||||||
|
Tests cover all helper functions for habit tracking including:
|
||||||
|
- calculate_streak for all 6 frequency types
|
||||||
|
- should_check_today for all frequency types
|
||||||
|
- get_completion_rate
|
||||||
|
- get_weekly_summary
|
||||||
|
"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
# Add parent directory to path to import habits_helpers
|
||||||
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
|
|
||||||
|
from habits_helpers import (
|
||||||
|
calculate_streak,
|
||||||
|
should_check_today,
|
||||||
|
get_completion_rate,
|
||||||
|
get_weekly_summary
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_daily_consecutive():
|
||||||
|
"""Test daily streak with consecutive days."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": [
|
||||||
|
{"date": today.isoformat()},
|
||||||
|
{"date": (today - timedelta(days=1)).isoformat()},
|
||||||
|
{"date": (today - timedelta(days=2)).isoformat()},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
assert calculate_streak(habit) == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_daily_with_gap():
|
||||||
|
"""Test daily streak breaks on gap."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": [
|
||||||
|
{"date": today.isoformat()},
|
||||||
|
{"date": (today - timedelta(days=1)).isoformat()},
|
||||||
|
# Gap here (day 2 missing)
|
||||||
|
{"date": (today - timedelta(days=3)).isoformat()},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
assert calculate_streak(habit) == 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_daily_empty():
|
||||||
|
"""Test daily streak with no completions."""
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": []
|
||||||
|
}
|
||||||
|
assert calculate_streak(habit) == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_specific_days():
|
||||||
|
"""Test specific_days streak (Mon, Wed, Fri)."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
# Find the most recent Monday
|
||||||
|
days_since_monday = today.weekday()
|
||||||
|
last_monday = today - timedelta(days=days_since_monday)
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "specific_days",
|
||||||
|
"days": [0, 2, 4] # Mon, Wed, Fri (0=Mon in Python weekday)
|
||||||
|
},
|
||||||
|
"completions": [
|
||||||
|
{"date": last_monday.isoformat()}, # Mon
|
||||||
|
{"date": (last_monday - timedelta(days=2)).isoformat()}, # Fri previous week
|
||||||
|
{"date": (last_monday - timedelta(days=4)).isoformat()}, # Wed previous week
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
# Should count 3 consecutive relevant days
|
||||||
|
streak = calculate_streak(habit)
|
||||||
|
assert streak >= 1 # At least the most recent relevant day
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_x_per_week():
|
||||||
|
"""Test x_per_week streak (3 times per week)."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
# Find Monday of current week
|
||||||
|
days_since_monday = today.weekday()
|
||||||
|
monday = today - timedelta(days=days_since_monday)
|
||||||
|
|
||||||
|
# Current week: 3 completions (Mon, Tue, Wed)
|
||||||
|
# Previous week: 3 completions (Mon, Tue, Wed)
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "x_per_week",
|
||||||
|
"count": 3
|
||||||
|
},
|
||||||
|
"completions": [
|
||||||
|
{"date": monday.isoformat()}, # This week Mon
|
||||||
|
{"date": (monday + timedelta(days=1)).isoformat()}, # This week Tue
|
||||||
|
{"date": (monday + timedelta(days=2)).isoformat()}, # This week Wed
|
||||||
|
# Previous week
|
||||||
|
{"date": (monday - timedelta(days=7)).isoformat()}, # Last week Mon
|
||||||
|
{"date": (monday - timedelta(days=6)).isoformat()}, # Last week Tue
|
||||||
|
{"date": (monday - timedelta(days=5)).isoformat()}, # Last week Wed
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
streak = calculate_streak(habit)
|
||||||
|
assert streak >= 2 # Both weeks meet the target
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_weekly():
|
||||||
|
"""Test weekly streak (at least 1 per week)."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "weekly"},
|
||||||
|
"completions": [
|
||||||
|
{"date": today.isoformat()}, # This week
|
||||||
|
{"date": (today - timedelta(days=7)).isoformat()}, # Last week
|
||||||
|
{"date": (today - timedelta(days=14)).isoformat()}, # 2 weeks ago
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
streak = calculate_streak(habit)
|
||||||
|
assert streak >= 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_monthly():
|
||||||
|
"""Test monthly streak (at least 1 per month)."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
# This month
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "monthly"},
|
||||||
|
"completions": [
|
||||||
|
{"date": today.isoformat()},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
streak = calculate_streak(habit)
|
||||||
|
assert streak >= 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_calculate_streak_custom_interval():
|
||||||
|
"""Test custom interval streak (every 3 days)."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "custom",
|
||||||
|
"interval": 3
|
||||||
|
},
|
||||||
|
"completions": [
|
||||||
|
{"date": today.isoformat()},
|
||||||
|
{"date": (today - timedelta(days=3)).isoformat()},
|
||||||
|
{"date": (today - timedelta(days=6)).isoformat()},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
streak = calculate_streak(habit)
|
||||||
|
assert streak == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_check_today_daily():
|
||||||
|
"""Test should_check_today for daily habit."""
|
||||||
|
habit = {"frequency": {"type": "daily"}}
|
||||||
|
assert should_check_today(habit) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_check_today_specific_days():
|
||||||
|
"""Test should_check_today for specific_days habit."""
|
||||||
|
today_weekday = datetime.now().date().weekday()
|
||||||
|
|
||||||
|
# Habit relevant today
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "specific_days",
|
||||||
|
"days": [today_weekday]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert should_check_today(habit) is True
|
||||||
|
|
||||||
|
# Habit not relevant today
|
||||||
|
other_day = (today_weekday + 1) % 7
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "specific_days",
|
||||||
|
"days": [other_day]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert should_check_today(habit) is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_check_today_x_per_week():
|
||||||
|
"""Test should_check_today for x_per_week habit."""
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "x_per_week",
|
||||||
|
"count": 3
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert should_check_today(habit) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_check_today_weekly():
|
||||||
|
"""Test should_check_today for weekly habit."""
|
||||||
|
habit = {"frequency": {"type": "weekly"}}
|
||||||
|
assert should_check_today(habit) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_check_today_monthly():
|
||||||
|
"""Test should_check_today for monthly habit."""
|
||||||
|
habit = {"frequency": {"type": "monthly"}}
|
||||||
|
assert should_check_today(habit) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_check_today_custom_ready():
|
||||||
|
"""Test should_check_today for custom interval when ready."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "custom",
|
||||||
|
"interval": 3
|
||||||
|
},
|
||||||
|
"completions": [
|
||||||
|
{"date": (today - timedelta(days=3)).isoformat()}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
assert should_check_today(habit) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_check_today_custom_not_ready():
|
||||||
|
"""Test should_check_today for custom interval when not ready."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "custom",
|
||||||
|
"interval": 3
|
||||||
|
},
|
||||||
|
"completions": [
|
||||||
|
{"date": (today - timedelta(days=1)).isoformat()}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
assert should_check_today(habit) is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_completion_rate_daily_perfect():
|
||||||
|
"""Test completion rate for daily habit with 100%."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
completions = []
|
||||||
|
for i in range(30):
|
||||||
|
completions.append({"date": (today - timedelta(days=i)).isoformat()})
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": completions
|
||||||
|
}
|
||||||
|
|
||||||
|
rate = get_completion_rate(habit, days=30)
|
||||||
|
assert rate == 100.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_completion_rate_daily_half():
|
||||||
|
"""Test completion rate for daily habit with 50%."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
completions = []
|
||||||
|
for i in range(0, 30, 2): # Every other day
|
||||||
|
completions.append({"date": (today - timedelta(days=i)).isoformat()})
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": completions
|
||||||
|
}
|
||||||
|
|
||||||
|
rate = get_completion_rate(habit, days=30)
|
||||||
|
assert 45 <= rate <= 55 # Around 50%
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_completion_rate_specific_days():
|
||||||
|
"""Test completion rate for specific_days habit."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
today_weekday = today.weekday()
|
||||||
|
|
||||||
|
# Create habit for Mon, Wed, Fri
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "specific_days",
|
||||||
|
"days": [0, 2, 4]
|
||||||
|
},
|
||||||
|
"completions": []
|
||||||
|
}
|
||||||
|
|
||||||
|
# Add completions for all relevant days in last 30 days
|
||||||
|
for i in range(30):
|
||||||
|
check_date = today - timedelta(days=i)
|
||||||
|
if check_date.weekday() in [0, 2, 4]:
|
||||||
|
habit["completions"].append({"date": check_date.isoformat()})
|
||||||
|
|
||||||
|
rate = get_completion_rate(habit, days=30)
|
||||||
|
assert rate == 100.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_completion_rate_empty():
|
||||||
|
"""Test completion rate with no completions."""
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": []
|
||||||
|
}
|
||||||
|
|
||||||
|
rate = get_completion_rate(habit, days=30)
|
||||||
|
assert rate == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_weekly_summary():
|
||||||
|
"""Test weekly summary returns correct structure."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": [
|
||||||
|
{"date": today.isoformat()},
|
||||||
|
{"date": (today - timedelta(days=1)).isoformat()},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = get_weekly_summary(habit)
|
||||||
|
|
||||||
|
# Check structure
|
||||||
|
assert isinstance(summary, dict)
|
||||||
|
assert "Monday" in summary
|
||||||
|
assert "Tuesday" in summary
|
||||||
|
assert "Wednesday" in summary
|
||||||
|
assert "Thursday" in summary
|
||||||
|
assert "Friday" in summary
|
||||||
|
assert "Saturday" in summary
|
||||||
|
assert "Sunday" in summary
|
||||||
|
|
||||||
|
# Check values are valid
|
||||||
|
valid_statuses = ["checked", "skipped", "missed", "upcoming", "not_relevant"]
|
||||||
|
for day, status in summary.items():
|
||||||
|
assert status in valid_statuses
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_weekly_summary_with_skip():
|
||||||
|
"""Test weekly summary handles skipped days."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
|
||||||
|
habit = {
|
||||||
|
"frequency": {"type": "daily"},
|
||||||
|
"completions": [
|
||||||
|
{"date": today.isoformat(), "type": "check"},
|
||||||
|
{"date": (today - timedelta(days=1)).isoformat(), "type": "skip"},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = get_weekly_summary(habit)
|
||||||
|
|
||||||
|
# Find today's day name
|
||||||
|
day_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
|
||||||
|
today_name = day_names[today.weekday()]
|
||||||
|
yesterday_name = day_names[(today.weekday() - 1) % 7]
|
||||||
|
|
||||||
|
assert summary[today_name] == "checked"
|
||||||
|
assert summary[yesterday_name] == "skipped"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_weekly_summary_specific_days():
|
||||||
|
"""Test weekly summary marks non-relevant days correctly."""
|
||||||
|
today = datetime.now().date()
|
||||||
|
today_weekday = today.weekday()
|
||||||
|
|
||||||
|
# Habit only for Monday (0)
|
||||||
|
habit = {
|
||||||
|
"frequency": {
|
||||||
|
"type": "specific_days",
|
||||||
|
"days": [0]
|
||||||
|
},
|
||||||
|
"completions": []
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = get_weekly_summary(habit)
|
||||||
|
|
||||||
|
# All days except Monday should be not_relevant or upcoming
|
||||||
|
for day_name, status in summary.items():
|
||||||
|
if day_name == "Monday":
|
||||||
|
continue # Monday can be any status
|
||||||
|
if status not in ["upcoming", "not_relevant"]:
|
||||||
|
# Day should be not_relevant if it's in the past
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
# Run all tests
|
||||||
|
import inspect
|
||||||
|
|
||||||
|
test_functions = [
|
||||||
|
obj for name, obj in inspect.getmembers(sys.modules[__name__])
|
||||||
|
if inspect.isfunction(obj) and name.startswith("test_")
|
||||||
|
]
|
||||||
|
|
||||||
|
passed = 0
|
||||||
|
failed = 0
|
||||||
|
|
||||||
|
for test_func in test_functions:
|
||||||
|
try:
|
||||||
|
test_func()
|
||||||
|
print(f"✓ {test_func.__name__}")
|
||||||
|
passed += 1
|
||||||
|
except AssertionError as e:
|
||||||
|
print(f"✗ {test_func.__name__}: {e}")
|
||||||
|
failed += 1
|
||||||
|
except Exception as e:
|
||||||
|
print(f"✗ {test_func.__name__}: {type(e).__name__}: {e}")
|
||||||
|
failed += 1
|
||||||
|
|
||||||
|
print(f"\n{passed} passed, {failed} failed")
|
||||||
|
sys.exit(0 if failed == 0 else 1)
|
||||||
555
dashboard/tests/test_habits_integration.py
Normal file
555
dashboard/tests/test_habits_integration.py
Normal file
@@ -0,0 +1,555 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Integration tests for Habits feature - End-to-end flows
|
||||||
|
|
||||||
|
Tests complete workflows involving multiple API calls and state transitions.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import shutil
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
from http.server import HTTPServer
|
||||||
|
from threading import Thread
|
||||||
|
import urllib.request
|
||||||
|
import urllib.error
|
||||||
|
|
||||||
|
# Add parent directory to path to import api module
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||||
|
|
||||||
|
from api import TaskBoardHandler
|
||||||
|
import habits_helpers
|
||||||
|
|
||||||
|
|
||||||
|
# Test helpers
|
||||||
|
def setup_test_env():
|
||||||
|
"""Create temporary environment for testing"""
|
||||||
|
from pathlib import Path
|
||||||
|
temp_dir = tempfile.mkdtemp()
|
||||||
|
habits_file = Path(temp_dir) / 'habits.json'
|
||||||
|
|
||||||
|
# Initialize empty habits file
|
||||||
|
with open(habits_file, 'w') as f:
|
||||||
|
json.dump({'lastUpdated': datetime.now().isoformat(), 'habits': []}, f)
|
||||||
|
|
||||||
|
# Override HABITS_FILE constant
|
||||||
|
import api
|
||||||
|
api.HABITS_FILE = habits_file
|
||||||
|
|
||||||
|
return temp_dir
|
||||||
|
|
||||||
|
|
||||||
|
def teardown_test_env(temp_dir):
|
||||||
|
"""Clean up temporary environment"""
|
||||||
|
shutil.rmtree(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def start_test_server():
|
||||||
|
"""Start HTTP server on random port for testing"""
|
||||||
|
server = HTTPServer(('localhost', 0), TaskBoardHandler)
|
||||||
|
thread = Thread(target=server.serve_forever, daemon=True)
|
||||||
|
thread.start()
|
||||||
|
return server
|
||||||
|
|
||||||
|
|
||||||
|
def http_request(url, method='GET', data=None):
|
||||||
|
"""Make HTTP request and return response data"""
|
||||||
|
headers = {'Content-Type': 'application/json'}
|
||||||
|
|
||||||
|
if data:
|
||||||
|
data = json.dumps(data).encode('utf-8')
|
||||||
|
|
||||||
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||||
|
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(req) as response:
|
||||||
|
body = response.read().decode('utf-8')
|
||||||
|
return json.loads(body) if body else None
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
error_body = e.read().decode('utf-8')
|
||||||
|
try:
|
||||||
|
return {'error': json.loads(error_body), 'status': e.code}
|
||||||
|
except:
|
||||||
|
return {'error': error_body, 'status': e.code}
|
||||||
|
|
||||||
|
|
||||||
|
# Integration Tests
|
||||||
|
|
||||||
|
def test_01_create_and_checkin_increments_streak():
|
||||||
|
"""Integration test: create habit → check-in → verify streak is 1"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create daily habit
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Morning meditation',
|
||||||
|
'category': 'health',
|
||||||
|
'color': '#10B981',
|
||||||
|
'icon': 'brain',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
if 'error' in result:
|
||||||
|
print(f"Error creating habit: {result}")
|
||||||
|
assert 'id' in result, f"Should return created habit with ID, got: {result}"
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Check in today
|
||||||
|
checkin_result = http_request(f"{base_url}/api/habits/{habit_id}/check", method='POST')
|
||||||
|
|
||||||
|
# Verify streak incremented to 1
|
||||||
|
assert checkin_result['streak']['current'] == 1, "Streak should be 1 after first check-in"
|
||||||
|
assert checkin_result['streak']['best'] == 1, "Best streak should be 1 after first check-in"
|
||||||
|
assert checkin_result['streak']['lastCheckIn'] == datetime.now().date().isoformat(), "Last check-in should be today"
|
||||||
|
|
||||||
|
print("✓ Test 1: Create + check-in → streak is 1")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_02_seven_consecutive_checkins_restore_life():
|
||||||
|
"""Integration test: 7 consecutive check-ins → life restored (if below 3)"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create daily habit
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Daily exercise',
|
||||||
|
'category': 'health',
|
||||||
|
'color': '#EF4444',
|
||||||
|
'icon': 'dumbbell',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Manually set lives to 1 (instead of using skip API which would add completions)
|
||||||
|
import api
|
||||||
|
with open(api.HABITS_FILE, 'r') as f:
|
||||||
|
data = json.load(f)
|
||||||
|
|
||||||
|
habit_obj = next(h for h in data['habits'] if h['id'] == habit_id)
|
||||||
|
habit_obj['lives'] = 1 # Directly set to 1 (simulating 2 skips used)
|
||||||
|
|
||||||
|
# Add 7 consecutive check-in completions for the past 7 days
|
||||||
|
for i in range(7):
|
||||||
|
check_date = (datetime.now() - timedelta(days=6-i)).date().isoformat()
|
||||||
|
habit_obj['completions'].append({
|
||||||
|
'date': check_date,
|
||||||
|
'type': 'check'
|
||||||
|
})
|
||||||
|
|
||||||
|
# Recalculate streak and check for life restore
|
||||||
|
habit_obj['streak'] = {
|
||||||
|
'current': habits_helpers.calculate_streak(habit_obj),
|
||||||
|
'best': max(habit_obj['streak']['best'], habits_helpers.calculate_streak(habit_obj)),
|
||||||
|
'lastCheckIn': datetime.now().date().isoformat()
|
||||||
|
}
|
||||||
|
|
||||||
|
# Check life restore logic: last 7 completions all 'check' type
|
||||||
|
last_7 = habit_obj['completions'][-7:]
|
||||||
|
if len(last_7) == 7 and all(c.get('type') == 'check' for c in last_7):
|
||||||
|
if habit_obj['lives'] < 3:
|
||||||
|
habit_obj['lives'] += 1
|
||||||
|
|
||||||
|
data['lastUpdated'] = datetime.now().isoformat()
|
||||||
|
with open(api.HABITS_FILE, 'w') as f:
|
||||||
|
json.dump(data, f, indent=2)
|
||||||
|
|
||||||
|
# Get updated habit
|
||||||
|
habits = http_request(f"{base_url}/api/habits")
|
||||||
|
habit = next(h for h in habits if h['id'] == habit_id)
|
||||||
|
|
||||||
|
# Verify life restored
|
||||||
|
assert habit['lives'] == 2, f"Should have 2 lives after 7 consecutive check-ins (was {habit['lives']})"
|
||||||
|
assert habit['current_streak'] == 7, "Should have streak of 7"
|
||||||
|
|
||||||
|
print("✓ Test 2: 7 consecutive check-ins → life restored")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_03_skip_with_life_maintains_streak():
|
||||||
|
"""Integration test: skip with life → lives decremented, streak unchanged"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create daily habit
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Read book',
|
||||||
|
'category': 'growth',
|
||||||
|
'color': '#3B82F6',
|
||||||
|
'icon': 'book',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Check in yesterday (to build a streak)
|
||||||
|
import api
|
||||||
|
with open(api.HABITS_FILE, 'r') as f:
|
||||||
|
data = json.load(f)
|
||||||
|
|
||||||
|
habit_obj = next(h for h in data['habits'] if h['id'] == habit_id)
|
||||||
|
yesterday = (datetime.now() - timedelta(days=1)).date().isoformat()
|
||||||
|
habit_obj['completions'].append({
|
||||||
|
'date': yesterday,
|
||||||
|
'type': 'check'
|
||||||
|
})
|
||||||
|
habit_obj['streak'] = {
|
||||||
|
'current': 1,
|
||||||
|
'best': 1,
|
||||||
|
'lastCheckIn': yesterday
|
||||||
|
}
|
||||||
|
|
||||||
|
data['lastUpdated'] = datetime.now().isoformat()
|
||||||
|
with open(api.HABITS_FILE, 'w') as f:
|
||||||
|
json.dump(data, f, indent=2)
|
||||||
|
|
||||||
|
# Skip today
|
||||||
|
skip_result = http_request(f"{base_url}/api/habits/{habit_id}/skip", method='POST')
|
||||||
|
|
||||||
|
# Verify lives decremented and streak maintained
|
||||||
|
assert skip_result['lives'] == 2, "Lives should be 2 after skip"
|
||||||
|
|
||||||
|
# Get fresh habit data to check streak
|
||||||
|
habits = http_request(f"{base_url}/api/habits")
|
||||||
|
habit = next(h for h in habits if h['id'] == habit_id)
|
||||||
|
|
||||||
|
# Streak should still be 1 (skip doesn't break it)
|
||||||
|
assert habit['current_streak'] == 1, "Streak should be maintained after skip"
|
||||||
|
|
||||||
|
print("✓ Test 3: Skip with life → lives decremented, streak unchanged")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_04_skip_with_zero_lives_returns_400():
|
||||||
|
"""Integration test: skip with 0 lives → returns 400 error"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create daily habit
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Yoga practice',
|
||||||
|
'category': 'health',
|
||||||
|
'color': '#8B5CF6',
|
||||||
|
'icon': 'heart',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Use all 3 lives
|
||||||
|
http_request(f"{base_url}/api/habits/{habit_id}/skip", method='POST')
|
||||||
|
http_request(f"{base_url}/api/habits/{habit_id}/skip", method='POST')
|
||||||
|
http_request(f"{base_url}/api/habits/{habit_id}/skip", method='POST')
|
||||||
|
|
||||||
|
# Attempt to skip with 0 lives
|
||||||
|
result = http_request(f"{base_url}/api/habits/{habit_id}/skip", method='POST')
|
||||||
|
|
||||||
|
# Verify 400 error
|
||||||
|
assert result['status'] == 400, "Should return 400 status"
|
||||||
|
assert 'error' in result, "Should return error message"
|
||||||
|
|
||||||
|
print("✓ Test 4: Skip with 0 lives → returns 400 error")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_05_edit_frequency_changes_should_check_today():
|
||||||
|
"""Integration test: edit frequency → should_check_today logic changes"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create daily habit
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Code review',
|
||||||
|
'category': 'work',
|
||||||
|
'color': '#F59E0B',
|
||||||
|
'icon': 'code',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Verify should_check_today is True for daily habit
|
||||||
|
habits = http_request(f"{base_url}/api/habits")
|
||||||
|
habit = next(h for h in habits if h['id'] == habit_id)
|
||||||
|
assert habit['should_check_today'] == True, "Daily habit should be checkable today"
|
||||||
|
|
||||||
|
# Edit to specific_days (only Monday and Wednesday)
|
||||||
|
update_data = {
|
||||||
|
'name': 'Code review',
|
||||||
|
'category': 'work',
|
||||||
|
'color': '#F59E0B',
|
||||||
|
'icon': 'code',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {
|
||||||
|
'type': 'specific_days',
|
||||||
|
'days': ['monday', 'wednesday']
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
http_request(f"{base_url}/api/habits/{habit_id}", method='PUT', data=update_data)
|
||||||
|
|
||||||
|
# Get updated habit
|
||||||
|
habits = http_request(f"{base_url}/api/habits")
|
||||||
|
habit = next(h for h in habits if h['id'] == habit_id)
|
||||||
|
|
||||||
|
# Verify should_check_today reflects new frequency
|
||||||
|
today_name = datetime.now().strftime('%A').lower()
|
||||||
|
expected = today_name in ['monday', 'wednesday']
|
||||||
|
assert habit['should_check_today'] == expected, f"Should check today should be {expected} for {today_name}"
|
||||||
|
|
||||||
|
print(f"✓ Test 5: Edit frequency → should_check_today is {expected} for {today_name}")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_06_delete_removes_habit_from_storage():
|
||||||
|
"""Integration test: delete → habit removed from storage"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create habit
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Guitar practice',
|
||||||
|
'category': 'personal',
|
||||||
|
'color': '#EC4899',
|
||||||
|
'icon': 'music',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Verify habit exists
|
||||||
|
habits = http_request(f"{base_url}/api/habits")
|
||||||
|
assert len(habits) == 1, "Should have 1 habit"
|
||||||
|
assert habits[0]['id'] == habit_id, "Should be the created habit"
|
||||||
|
|
||||||
|
# Delete habit
|
||||||
|
http_request(f"{base_url}/api/habits/{habit_id}", method='DELETE')
|
||||||
|
|
||||||
|
# Verify habit removed
|
||||||
|
habits = http_request(f"{base_url}/api/habits")
|
||||||
|
assert len(habits) == 0, "Should have 0 habits after delete"
|
||||||
|
|
||||||
|
# Verify not in storage file
|
||||||
|
import api
|
||||||
|
with open(api.HABITS_FILE, 'r') as f:
|
||||||
|
data = json.load(f)
|
||||||
|
|
||||||
|
assert len(data['habits']) == 0, "Storage file should have 0 habits"
|
||||||
|
|
||||||
|
print("✓ Test 6: Delete → habit removed from storage")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_07_checkin_on_wrong_day_for_specific_days_returns_400():
|
||||||
|
"""Integration test: check-in on wrong day for specific_days → returns 400"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Get today's day name
|
||||||
|
today_name = datetime.now().strftime('%A').lower()
|
||||||
|
|
||||||
|
# Create habit for different days (not today)
|
||||||
|
if today_name == 'monday':
|
||||||
|
allowed_days = ['tuesday', 'wednesday']
|
||||||
|
elif today_name == 'tuesday':
|
||||||
|
allowed_days = ['monday', 'wednesday']
|
||||||
|
else:
|
||||||
|
allowed_days = ['monday', 'tuesday']
|
||||||
|
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Gym workout',
|
||||||
|
'category': 'health',
|
||||||
|
'color': '#EF4444',
|
||||||
|
'icon': 'dumbbell',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {
|
||||||
|
'type': 'specific_days',
|
||||||
|
'days': allowed_days
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Attempt to check in today (wrong day)
|
||||||
|
result = http_request(f"{base_url}/api/habits/{habit_id}/check", method='POST')
|
||||||
|
|
||||||
|
# Verify 400 error
|
||||||
|
assert result['status'] == 400, "Should return 400 status"
|
||||||
|
assert 'error' in result, "Should return error message"
|
||||||
|
|
||||||
|
print(f"✓ Test 7: Check-in on {today_name} (not in {allowed_days}) → returns 400")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_08_get_response_includes_all_stats():
|
||||||
|
"""Integration test: GET response includes stats (streak, completion_rate, weekly_summary)"""
|
||||||
|
temp_dir = setup_test_env()
|
||||||
|
server = start_test_server()
|
||||||
|
base_url = f"http://localhost:{server.server_port}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Create habit with some completions
|
||||||
|
habit_data = {
|
||||||
|
'name': 'Meditation',
|
||||||
|
'category': 'health',
|
||||||
|
'color': '#10B981',
|
||||||
|
'icon': 'brain',
|
||||||
|
'priority': 50,
|
||||||
|
'frequency': {'type': 'daily'}
|
||||||
|
}
|
||||||
|
|
||||||
|
result = http_request(f"{base_url}/api/habits", method='POST', data=habit_data)
|
||||||
|
habit_id = result['id']
|
||||||
|
|
||||||
|
# Add some completions
|
||||||
|
import api
|
||||||
|
with open(api.HABITS_FILE, 'r') as f:
|
||||||
|
data = json.load(f)
|
||||||
|
|
||||||
|
habit_obj = next(h for h in data['habits'] if h['id'] == habit_id)
|
||||||
|
|
||||||
|
# Add completions for last 3 days
|
||||||
|
for i in range(3):
|
||||||
|
check_date = (datetime.now() - timedelta(days=2-i)).date().isoformat()
|
||||||
|
habit_obj['completions'].append({
|
||||||
|
'date': check_date,
|
||||||
|
'type': 'check'
|
||||||
|
})
|
||||||
|
|
||||||
|
habit_obj['streak'] = {
|
||||||
|
'current': 3,
|
||||||
|
'best': 3,
|
||||||
|
'lastCheckIn': datetime.now().date().isoformat()
|
||||||
|
}
|
||||||
|
|
||||||
|
data['lastUpdated'] = datetime.now().isoformat()
|
||||||
|
with open(api.HABITS_FILE, 'w') as f:
|
||||||
|
json.dump(data, f, indent=2)
|
||||||
|
|
||||||
|
# Get habits
|
||||||
|
habits = http_request(f"{base_url}/api/habits")
|
||||||
|
habit = habits[0]
|
||||||
|
|
||||||
|
# Verify all enriched stats are present
|
||||||
|
assert 'current_streak' in habit, "Should include current_streak"
|
||||||
|
assert 'best_streak' in habit, "Should include best_streak"
|
||||||
|
assert 'completion_rate_30d' in habit, "Should include completion_rate_30d"
|
||||||
|
assert 'weekly_summary' in habit, "Should include weekly_summary"
|
||||||
|
assert 'should_check_today' in habit, "Should include should_check_today"
|
||||||
|
|
||||||
|
# Verify streak values
|
||||||
|
assert habit['current_streak'] == 3, "Current streak should be 3"
|
||||||
|
assert habit['best_streak'] == 3, "Best streak should be 3"
|
||||||
|
|
||||||
|
# Verify weekly_summary structure
|
||||||
|
assert isinstance(habit['weekly_summary'], dict), "Weekly summary should be a dict"
|
||||||
|
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
|
||||||
|
for day in days:
|
||||||
|
assert day in habit['weekly_summary'], f"Weekly summary should include {day}"
|
||||||
|
|
||||||
|
print("✓ Test 8: GET response includes all stats (streak, completion_rate, weekly_summary)")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
server.shutdown()
|
||||||
|
teardown_test_env(temp_dir)
|
||||||
|
|
||||||
|
|
||||||
|
def test_09_typecheck_passes():
|
||||||
|
"""Integration test: Typecheck passes"""
|
||||||
|
result = os.system('python3 -m py_compile /home/moltbot/clawd/dashboard/api.py')
|
||||||
|
assert result == 0, "Typecheck should pass for api.py"
|
||||||
|
|
||||||
|
result = os.system('python3 -m py_compile /home/moltbot/clawd/dashboard/habits_helpers.py')
|
||||||
|
assert result == 0, "Typecheck should pass for habits_helpers.py"
|
||||||
|
|
||||||
|
print("✓ Test 9: Typecheck passes")
|
||||||
|
|
||||||
|
|
||||||
|
# Run all tests
|
||||||
|
if __name__ == '__main__':
|
||||||
|
tests = [
|
||||||
|
test_01_create_and_checkin_increments_streak,
|
||||||
|
test_02_seven_consecutive_checkins_restore_life,
|
||||||
|
test_03_skip_with_life_maintains_streak,
|
||||||
|
test_04_skip_with_zero_lives_returns_400,
|
||||||
|
test_05_edit_frequency_changes_should_check_today,
|
||||||
|
test_06_delete_removes_habit_from_storage,
|
||||||
|
test_07_checkin_on_wrong_day_for_specific_days_returns_400,
|
||||||
|
test_08_get_response_includes_all_stats,
|
||||||
|
test_09_typecheck_passes,
|
||||||
|
]
|
||||||
|
|
||||||
|
passed = 0
|
||||||
|
failed = 0
|
||||||
|
|
||||||
|
print("Running integration tests...\n")
|
||||||
|
|
||||||
|
for test in tests:
|
||||||
|
try:
|
||||||
|
test()
|
||||||
|
passed += 1
|
||||||
|
except AssertionError as e:
|
||||||
|
print(f"✗ {test.__name__}: {e}")
|
||||||
|
failed += 1
|
||||||
|
except Exception as e:
|
||||||
|
print(f"✗ {test.__name__}: Unexpected error: {e}")
|
||||||
|
import traceback
|
||||||
|
traceback.print_exc()
|
||||||
|
failed += 1
|
||||||
|
|
||||||
|
print(f"\n{'='*50}")
|
||||||
|
print(f"Integration Tests: {passed} passed, {failed} failed")
|
||||||
|
print(f"{'='*50}")
|
||||||
|
|
||||||
|
sys.exit(0 if failed == 0 else 1)
|
||||||
Reference in New Issue
Block a user