RustAst/tests/data_server.rs

//! Integration tests for the DataServer, verifying correct parsing of real
//! tick data files from /mnt/tickdata/Pepperstone/.

use std::path::Path;
use std::sync::Arc;

const TICKDATA_PATH: &str = "/mnt/tickdata/Pepperstone";

/// Skip tests when the tick data directory is not available.
fn skip_if_no_data() -> bool {
    !Path::new(TICKDATA_PATH).exists()
}

#[test]
fn test_real_symbol_index() {
    if skip_if_no_data() {
        return;
    }
    let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH));
    let symbols = server.symbols();

    assert!(!symbols.is_empty(), "Should find at least one symbol");
    assert!(
        symbols.iter().any(|s| s.as_ref() == "EURUSD"),
        "EURUSD should be in the symbol list"
    );

    // Verify file count for a known symbol
    let m1_count = server
        .file_count("EURUSD", myc::ast::data_server::records::DataFormat::M1)
        .unwrap();
    assert!(m1_count > 0, "EURUSD should have M1 files, found {m1_count}");
}

#[test]
fn test_real_m1_stream_eurusd() {
    if skip_if_no_data() {
        return;
    }
    let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH));
    let mut iter = server.stream_m1("EURUSD").expect("EURUSD M1 stream");

    // Read the first chunk
    let first_chunk = iter.next_chunk().expect("At least one chunk");
    assert!(!first_chunk.is_empty());

    let rec = &first_chunk[0];
    // Verify the record has sane values
    assert!(rec.time_ms > 0, "Timestamp should be positive");
    assert!(rec.open > 0.0, "EURUSD open should be > 0");
    assert!(rec.high >= rec.low, "High should be >= Low");
    assert!(rec.close > 0.0, "EURUSD close should be > 0");

    // EURUSD is always between ~0.8 and ~1.6
    assert!(
        (0.5..2.0).contains(&rec.open),
        "EURUSD open should be between 0.5 and 2.0, got {}",
        rec.open
    );

    println!(
        "First EURUSD M1 record: time_ms={}, O={:.5} H={:.5} L={:.5} C={:.5} S={:.1} V={}",
        rec.time_ms, rec.open, rec.high, rec.low, rec.close, rec.spread, rec.volume
    );
}

#[test]
fn test_real_tick_stream_eurusd() {
    if skip_if_no_data() {
        return;
    }
    let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH));

    // EURUSD tick data may not exist — check gracefully
    let Some(mut iter) = server.stream_tick("EURUSD") else {
        println!("No EURUSD tick data found, skipping");
        return;
    };

    let first_chunk = iter.next_chunk().expect("At least one tick chunk");
    let rec = &first_chunk[0];

    assert!(rec.time_ms > 0);
    assert!(rec.ask > 0.0);
    assert!(rec.bid > 0.0);
    // Spread should be small for EURUSD
    let spread = (rec.ask - rec.bid).abs();
    assert!(
        spread < 0.01,
        "EURUSD spread should be < 100 pips, got {spread}"
    );

    println!(
        "First EURUSD tick: time_ms={}, Ask={:.5} Bid={:.5} Spread={:.5}",
        rec.time_ms, rec.ask, rec.bid, spread
    );
}

#[test]
fn test_real_m1_full_iteration() {
    if skip_if_no_data() {
        return;
    }
    let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH));

    // Use a symbol with few files for speed (AAPL.US starts 2006)
    let Some(mut iter) = server.stream_m1("AAPL.US") else {
        println!("No AAPL.US data found, skipping");
        return;
    };

    let mut total_records = 0u64;
    let mut chunks = 0u64;
    while let Some(chunk) = iter.next_chunk() {
        chunks += 1;
        total_records += chunk.len() as u64;
    }

    println!("AAPL.US M1: {total_records} records in {chunks} chunks");
    assert!(total_records > 0, "Should have read some records");
}

// FIXME: This test is flaky — thread scheduling can cause uneven read counts.
#[test]
fn test_real_concurrent_access() {
    if skip_if_no_data() {
        return;
    }
    let server = Arc::new(myc::ast::data_server::DataServer::new(TICKDATA_PATH));

    // Spawn 4 threads all reading the same symbol simultaneously
    let handles: Vec<_> = (0..4)
        .map(|_| {
            let srv = Arc::clone(&server);
            std::thread::spawn(move || {
                let mut iter = srv.stream_m1("EURUSD").expect("EURUSD");
                let mut count = 0u64;
                // Read first 10 chunks only (for speed)
                for _ in 0..10 {
                    if let Some(chunk) = iter.next_chunk() {
                        count += chunk.len() as u64;
                    } else {
                        break;
                    }
                }
                count
            })
        })
        .collect();

    let counts: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect();

    // All threads should read the same number of records
    assert!(counts[0] > 0);
    assert!(
        counts.iter().all(|&c| c == counts[0]),
        "All threads should read the same amount: {counts:?}"
    );
    println!("Concurrent access: 4 threads each read {} records", counts[0]);
}