clawd/dashboard/test_habits_integration.py

#!/usr/bin/env python3
"""
Integration test for complete habit lifecycle.
Tests: create, check multiple days, view streak, delete
"""

import json
import os
import sys
import time
from datetime import datetime, timedelta
from http.server import HTTPServer
import threading
import urllib.request
import urllib.error

# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from api import TaskBoardHandler

# Test configuration
PORT = 8765
BASE_URL = f"http://localhost:{PORT}"
HABITS_FILE = os.path.join(os.path.dirname(__file__), 'habits.json')

# Global server instance
server = None
server_thread = None


def setup_server():
    """Start test server in background thread"""
    global server, server_thread
    server = HTTPServer(('localhost', PORT), TaskBoardHandler)
    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
    server_thread.start()
    time.sleep(0.5)  # Give server time to start


def teardown_server():
    """Stop test server"""
    global server
    if server:
        server.shutdown()
        server.server_close()


def reset_habits_file():
    """Reset habits.json to empty state"""
    data = {
        "lastUpdated": datetime.utcnow().isoformat() + 'Z',
        "habits": []
    }
    with open(HABITS_FILE, 'w') as f:
        json.dump(data, f, indent=2)


def make_request(method, path, body=None):
    """Make HTTP request to test server"""
    url = BASE_URL + path
    headers = {'Content-Type': 'application/json'} if body else {}
    data = json.dumps(body).encode('utf-8') if body else None
    
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    
    try:
        with urllib.request.urlopen(req) as response:
            response_data = response.read().decode('utf-8')
            return response.status, json.loads(response_data) if response_data else None
    except urllib.error.HTTPError as e:
        response_data = e.read().decode('utf-8')
        return e.code, json.loads(response_data) if response_data else None


def get_today():
    """Get today's date in YYYY-MM-DD format"""
    return datetime.utcnow().strftime('%Y-%m-%d')


def get_yesterday():
    """Get yesterday's date in YYYY-MM-DD format"""
    return (datetime.utcnow() - timedelta(days=1)).strftime('%Y-%m-%d')


def test_complete_habit_lifecycle():
    """
    Integration test: Complete habit flow from creation to deletion
    """
    print("\n=== Integration Test: Complete Habit Lifecycle ===\n")
    
    # Step 1: Create daily habit 'Bazin'
    print("Step 1: Creating daily habit 'Bazin'...")
    status, response = make_request('POST', '/api/habits', {
        'name': 'Bazin',
        'frequency': 'daily'
    })
    
    assert status == 201, f"Expected 201, got {status}"
    assert response['name'] == 'Bazin', "Habit name mismatch"
    assert response['frequency'] == 'daily', "Frequency mismatch"
    assert 'id' in response, "Missing habit ID"
    habit_id = response['id']
    print(f"✓ Created habit: {habit_id}")
    
    # Step 2: Check it today (streak should be 1)
    print("\nStep 2: Checking habit today (expecting streak = 1)...")
    status, response = make_request('POST', f'/api/habits/{habit_id}/check', {})
    
    assert status == 200, f"Expected 200, got {status}"
    assert response['streak'] == 1, f"Expected streak=1, got {response['streak']}"
    assert get_today() in response['completions'], "Today's date not in completions"
    print(f"✓ Checked today, streak = {response['streak']}")
    
    # Step 3: Simulate checking yesterday (manually add to completions)
    print("\nStep 3: Simulating yesterday's check (expecting streak = 2)...")
    # Read current habits.json
    with open(HABITS_FILE, 'r') as f:
        data = json.load(f)
    
    # Find the habit and add yesterday's date
    for habit in data['habits']:
        if habit['id'] == habit_id:
            habit['completions'].append(get_yesterday())
            habit['completions'].sort()  # Keep chronological order
            break
    
    # Write back to file
    data['lastUpdated'] = datetime.utcnow().isoformat() + 'Z'
    with open(HABITS_FILE, 'w') as f:
        json.dump(data, f, indent=2)
    
    # Verify streak calculation by fetching habits
    status, response = make_request('GET', '/api/habits', None)
    assert status == 200, f"Expected 200, got {status}"
    
    habit = next((h for h in response['habits'] if h['id'] == habit_id), None)
    assert habit is not None, "Habit not found in response"
    assert habit['streak'] == 2, f"Expected streak=2 after adding yesterday, got {habit['streak']}"
    print(f"✓ Added yesterday's completion, streak = {habit['streak']}")
    
    # Step 4: Verify streak calculation is correct
    print("\nStep 4: Verifying streak calculation...")
    assert len(habit['completions']) == 2, f"Expected 2 completions, got {len(habit['completions'])}"
    assert get_yesterday() in habit['completions'], "Yesterday not in completions"
    assert get_today() in habit['completions'], "Today not in completions"
    assert habit['checkedToday'] == True, "checkedToday should be True"
    print("✓ Streak calculation verified: 2 consecutive days")
    
    # Step 5: Delete habit successfully
    print("\nStep 5: Deleting habit...")
    status, response = make_request('DELETE', f'/api/habits/{habit_id}', None)
    
    assert status == 200, f"Expected 200, got {status}"
    assert 'message' in response, "Missing success message"
    print(f"✓ Deleted habit: {response['message']}")
    
    # Verify habit is gone
    status, response = make_request('GET', '/api/habits', None)
    assert status == 200, f"Expected 200, got {status}"
    
    habit = next((h for h in response['habits'] if h['id'] == habit_id), None)
    assert habit is None, "Habit still exists after deletion"
    print("✓ Verified habit no longer exists")
    
    print("\n=== All Integration Tests Passed ✓ ===\n")


def test_broken_streak():
    """
    Additional test: Verify broken streak returns 0 (with gap)
    """
    print("\n=== Additional Test: Broken Streak (with gap) ===\n")
    
    # Create habit
    status, response = make_request('POST', '/api/habits', {
        'name': 'Sală',
        'frequency': 'daily'
    })
    assert status == 201
    habit_id = response['id']
    print(f"✓ Created habit: {habit_id}")
    
    # Add check from 3 days ago (creating a gap)
    three_days_ago = (datetime.utcnow() - timedelta(days=3)).strftime('%Y-%m-%d')
    with open(HABITS_FILE, 'r') as f:
        data = json.load(f)
    
    for habit in data['habits']:
        if habit['id'] == habit_id:
            habit['completions'].append(three_days_ago)
            break
    
    data['lastUpdated'] = datetime.utcnow().isoformat() + 'Z'
    with open(HABITS_FILE, 'w') as f:
        json.dump(data, f, indent=2)
    
    # Verify streak is 0 (>1 day gap means broken streak)
    status, response = make_request('GET', '/api/habits', None)
    habit = next((h for h in response['habits'] if h['id'] == habit_id), None)
    assert habit['streak'] == 0, f"Expected streak=0 for broken streak, got {habit['streak']}"
    print(f"✓ Broken streak (>1 day gap) correctly returns 0")
    
    # Cleanup
    make_request('DELETE', f'/api/habits/{habit_id}', None)
    print("✓ Cleanup complete")


def test_weekly_habit_streak():
    """
    Additional test: Weekly habit streak calculation
    """
    print("\n=== Additional Test: Weekly Habit Streak ===\n")
    
    # Create weekly habit
    status, response = make_request('POST', '/api/habits', {
        'name': 'Yoga',
        'frequency': 'weekly'
    })
    assert status == 201
    habit_id = response['id']
    print(f"✓ Created weekly habit: {habit_id}")
    
    # Check today (streak = 1 week)
    status, response = make_request('POST', f'/api/habits/{habit_id}/check', {})
    assert status == 200
    assert response['streak'] == 1, f"Expected streak=1 week, got {response['streak']}"
    print(f"✓ Checked today, weekly streak = {response['streak']}")
    
    # Add check from 8 days ago (last week)
    eight_days_ago = (datetime.utcnow() - timedelta(days=8)).strftime('%Y-%m-%d')
    with open(HABITS_FILE, 'r') as f:
        data = json.load(f)
    
    for habit in data['habits']:
        if habit['id'] == habit_id:
            habit['completions'].append(eight_days_ago)
            habit['completions'].sort()
            break
    
    data['lastUpdated'] = datetime.utcnow().isoformat() + 'Z'
    with open(HABITS_FILE, 'w') as f:
        json.dump(data, f, indent=2)
    
    # Verify streak is 2 weeks
    status, response = make_request('GET', '/api/habits', None)
    habit = next((h for h in response['habits'] if h['id'] == habit_id), None)
    assert habit['streak'] == 2, f"Expected streak=2 weeks, got {habit['streak']}"
    print(f"✓ Weekly streak calculation correct: {habit['streak']} weeks")
    
    # Cleanup
    make_request('DELETE', f'/api/habits/{habit_id}', None)
    print("✓ Cleanup complete")


if __name__ == '__main__':
    try:
        # Setup
        reset_habits_file()
        setup_server()
        
        # Run tests
        test_complete_habit_lifecycle()
        test_broken_streak()
        test_weekly_habit_streak()
        
        print("\n🎉 All Integration Tests Passed! 🎉\n")
        
    except AssertionError as e:
        print(f"\n❌ Test Failed: {e}\n")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ Error: {e}\n")
        import traceback
        traceback.print_exc()
        sys.exit(1)
    finally:
        # Cleanup
        teardown_server()
        reset_habits_file()