Testing Strategy for LocalPulse Signals Implementation

Overview

This document outlines the comprehensive testing approach for the new signals functionality, including unit tests, contract tests, and integration tests.

Test Structure

apps/action-engine-api/tests/
├── unit/
│   └── test_index_v0_logic.py          # Core logic tests
├── contract/
│   └── test_signals_and_actions_contract.py  # API contract tests
└── fixtures/
    ├── traffic_sample.csv              # Test traffic data
    └── weather_sample.csv              # Test weather data

Unit Tests (test_index_v0_logic.py)

Test Categories

1. CSV Ingestion Tests

def test_load_traffic_csv_parsing():
    """Test that traffic CSV is parsed correctly"""
    
def test_load_traffic_csv_normalization():
    """Test column normalization and timestamp conversion"""
    
def test_load_weather_csv_parsing():
    """Test that weather CSV is parsed correctly"""
    
def test_weather_hourly_aggregation():
    """Test weather data aggregation to hourly means"""
    
def test_latest_common_hour():
    """Test finding common hour between traffic and weather data"""

2. Data Aggregation Tests

def test_aggregate_hour_weighted_sum():
    """Test traffic aggregation with proper weights"""
    # weights = {PEOPLE:1.0, MOTORCYCLE:0.6, BIKE:0.5, AUTOCAR:0.7, BUS:0.8, TRUCK:0.3}
    
def test_robust_minmax_normal_case():
    """Test 5th/95th percentile calculation"""
    
def test_robust_minmax_edge_cases():
    """Test fallback to min/max when 95th <= 5th percentile"""

3. Scoring Function Tests

def test_score_linear():
    """Test linear scoring function"""
    # score_linear(50, 0, 100) should return 50
    
def test_score_peak():
    """Test triangular scoring function"""
    # score_peak(24, 10, 24, 35) should return 100
    
def test_compute_flow_score():
    """Test flow score calculation with history"""
    # With history [10, 20, 30, 40, 50], current=50 should be high
    
def test_compute_env_score_good_conditions():
    """Test environmental score with good weather"""
    # 24°C, 55%RH, 2m/s, pm25=10 should be >=80
    
def test_compute_env_score_poor_air():
    """Test environmental score with poor air quality"""
    # pm25=120 should lower score significantly
    
def test_compute_capacity_index():
    """Test capacity index calculation"""
    # 0.65*flow + 0.35*env
    
def test_band_classification():
    """Test band classification logic"""
    # 0-39: low, 40-59: mid, 60-79: high, 80-100: peak

4. Action Generation Tests

def test_generate_actions_low_capacity():
    """Test actions for low capacity (<40)"""
    # Should include online/delivery suggestions
    
def test_generate_actions_peak_capacity():
    """Test actions for peak capacity (>=80)"""
    # Should include staffing and flow control
    
def test_generate_actions_pm25_override():
    """Test PM2.5 override action injection"""
    # pm25>=55 should insert air quality action
    
def test_generate_actions_wind_override():
    """Test wind override action injection"""
    # wind>=10 should insert outdoor safety action
    
def test_action_object_structure():
    """Test action object has required fields"""
    # id, title, reason, category, confidence

Contract Tests (test_signals_and_actions_contract.py)

Test Setup

import pytest
import tempfile
import os
from fastapi.testclient import TestClient
from app.main import app

@pytest.fixture
def temp_traffic_csv():
    """Create temporary traffic CSV for testing"""
    
@pytest.fixture
def temp_weather_csv():
    """Create temporary weather CSV for testing"""
    
@pytest.fixture
def client():
    """Test client for API requests"""
    return TestClient(app)

API Contract Tests

1. GET /signals/current Tests

def test_signals_current_200_ok(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test successful response with valid data"""
    # Set environment variables to temp files
    # Verify response structure
    # Verify all required fields present
    # Verify scores within [0,100]
    # Verify band in valid set

def test_signals_current_rate_limit_headers(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test rate limit headers are present"""
    # Verify X-RateLimit-* headers exist

def test_signals_current_missing_csv(client, monkeypatch):
    """Test 404 when CSV files don't exist"""
    # Set paths to non-existent files
    # Expect 404 with csv_file_not_found

def test_signals_current_invalid_csv(client, monkeypatch):
    """Test 503 when CSV files are invalid"""
    # Create invalid CSV files
    # Expect 503 with csv_file_invalid

2. GET /actions Tests

def test_actions_default_limit(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test actions with default limit (3)"""
    # Verify exactly 3 actions returned

def test_actions_custom_limit(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test actions with custom limit"""
    # Test limit=1, limit=5
    # Verify correct number of actions

def test_actions_invalid_limit(client):
    """Test 422 with invalid limit parameter"""
    # Test limit=0, limit=6, limit=-1

def test_actions_pm25_override(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test PM2.5 override in actions"""
    # Create weather CSV with PM2.5 >= 55
    # Verify first action mentions air quality

def test_actions_structure(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test action object structure"""
    # Verify each action has id, title, reason, category, confidence
    # Verify confidence in [0,1]

3. Caching Tests

def test_signals_caching(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test that signals are cached"""
    # First request should miss cache
    # Second request should hit cache
    # Verify cache headers if implemented

def test_actions_caching(client, temp_traffic_csv, temp_weather_csv, monkeypatch):
    """Test that actions are cached with different limits"""
    # Verify different limits have separate cache entries

Test Data

Traffic CSV Fixture (traffic_sample.csv)

群組名稱,統計時間_時,類別,數量
西門町,2025-10-19 08:00:00,PEOPLE,50
西門町,2025-10-19 08:00:00,MOTORCYCLE,20
西門町,2025-10-19 08:00:00,BIKE,5
西門町,2025-10-19 08:00:00,AUTOCAR,15
西門町,2025-10-19 08:00:00,BUS,5
西門町,2025-10-19 08:00:00,TRUCK,2
西門町,2025-10-19 09:00:00,PEOPLE,80
西門町,2025-10-19 09:00:00,MOTORCYCLE,30
西門町,2025-10-19 09:00:00,BIKE,10
西門町,2025-10-19 09:00:00,AUTOCAR,25
西門町,2025-10-19 09:00:00,BUS,8
西門町,2025-10-19 09:00:00,TRUCK,3
西門町,2025-10-19 10:00:00,PEOPLE,120
西門町,2025-10-19 10:00:00,MOTORCYCLE,45
西門町,2025-10-19 10:00:00,BIKE,15
西門町,2025-10-19 10:00:00,AUTOCAR,50
西門町,2025-10-19 10:00:00,BUS,10
西門町,2025-10-19 10:00:00,TRUCK,5

Weather CSV Fixture (weather_sample.csv)

裝置,回報時間,溫度,濕度,風速,PM2.5,PM10,噪音
台北站,2025-10-19 08:15:00,22.5,70,1.5,15,20,60.0
台北站,2025-10-19 08:45:00,23.0,68,1.8,16,22,61.5
台北站,2025-10-19 09:15:00,24.0,65,2.0,17,23,62.0
台北站,2025-10-19 09:45:00,24.5,65,2.3,18,25,62.3
台北站,2025-10-19 10:15:00,25.0,63,2.5,19,26,63.0

Coverage Requirements

Minimum Coverage: 70%

Unit tests should cover all core logic functions
Contract tests should cover all API endpoints
Edge cases and error conditions must be tested

Coverage Targets by Module

ingest_csv.py: 90% (data parsing is critical)
index_v0.py: 85% (scoring logic is complex)
routes_signals.py: 80% (API endpoints)

Test Execution

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=app --cov-report=html

# Run only signals tests
pytest tests/unit/test_index_v0_logic.py tests/contract/test_signals_and_actions_contract.py

# Run with verbose output
pytest -v

CI/CD Integration

Tests will run in CI/CD pipeline with:

Python 3.11
All dependencies from requirements.txt
Coverage reporting
Failure on coverage < 70%

Test Data Management

Temporary Files

Use tempfile.NamedTemporaryFile for CSV fixtures
Clean up automatically with context managers
Ensure proper encoding (UTF-8)

Environment Variables

Use monkeypatch.setenv to override paths in tests
Restore original values after tests
Test with both default and custom paths

Performance Considerations

Test Performance

Unit tests should run in < 1 second
Contract tests should run in < 2 seconds
Use efficient test data (small CSV files)

Caching Tests

Test cache hit/miss behavior
Verify TTL functionality
Test cache invalidation

FilesExpand file tree

testing_strategy.md

Latest commit

History

testing_strategy.md

File metadata and controls

Testing Strategy for LocalPulse Signals Implementation

Overview

Test Structure

Unit Tests (test_index_v0_logic.py)

Test Categories

1. CSV Ingestion Tests

2. Data Aggregation Tests

3. Scoring Function Tests

4. Action Generation Tests

Contract Tests (test_signals_and_actions_contract.py)

Test Setup

API Contract Tests

1. GET /signals/current Tests

2. GET /actions Tests

3. Caching Tests

Test Data

Traffic CSV Fixture (traffic_sample.csv)

Weather CSV Fixture (weather_sample.csv)

Coverage Requirements

Minimum Coverage: 70%

Coverage Targets by Module

Test Execution

Running Tests

CI/CD Integration

Test Data Management

Temporary Files

Environment Variables

Performance Considerations

Test Performance

Caching Tests