clawd/dashboard/tests/test_habits_integration.py

#!/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)