""" Test suite for DatetimeSampler with comprehensive validation """ import os import random from collections import Counter from datetime import datetime, timedelta from typing import List, Optional import matplotlib.pyplot as plt import numpy as np from samplers import DatetimeSampler, HistogramBucket, TopNItem from samplers.tidb_types import TiDBTimeBucket, TiDBTimeTopNItem def get_validate_seed(): """Get seed from environment or use default""" return int(os.environ.get("ITS_VALIDATION_SEED", 42)) def generate_validate_datetime_data( size: int = 1000, seed: Optional[int] = None ) -> List[Optional[datetime]]: """Generate test datetime data with various distribution patterns based on seed""" if seed is None: seed = get_validate_seed() random.seed(seed) np.random.seed(seed) # Use seed to determine distribution type distribution_type = seed % 5 # 5 different distribution types data = [] # Base parameters base_dt = datetime(2023, 1, 1, 0, 0, 0) year_seconds = 365 * 24 * 3600 # Seconds in a year print( f" 🎲 Generating datetime distribution type {distribution_type} (seed {seed})" ) if distribution_type == 0: # Normal distribution (concentrated around mid-year) print(" 📊 Normal distribution - concentrated around mid-year") # Normal distribution centered around July (mid-year) center_offset = year_seconds // 2 # Mid-year std_dev = year_seconds // 8 # Standard deviation = 1.5 months (tighter) for _ in range(int(size * 0.8)): offset = np.random.normal(center_offset, std_dev) offset = max(0, min(year_seconds - 1, offset)) # Clamp to year bounds dt = base_dt + timedelta(seconds=int(offset)) data.append(dt) elif ( distribution_type == 1 ): # Exponential distribution (skewed toward beginning of year) print(" 📊 Exponential distribution - skewed toward beginning of year") # Exponential decay from beginning of year scale = year_seconds / 6 # Scale parameter (more concentrated) for _ in range(int(size * 0.8)): offset = np.random.exponential(scale) offset = min(year_seconds - 1, offset) # Clamp to year bounds dt = base_dt + timedelta(seconds=int(offset)) data.append(dt) elif distribution_type == 2: # Uniform distribution (evenly spread) print(" 📊 Uniform distribution - evenly spread throughout year") for _ in range(int(size * 0.8)): offset = random.uniform(0, year_seconds - 1) dt = base_dt + timedelta(seconds=int(offset)) data.append(dt) elif distribution_type == 3: # Bimodal distribution (two peaks: spring and fall) print(" 📊 Bimodal distribution - peaks in spring and fall") # Two peaks: March and September peak1_offset = year_seconds * 3 / 12 # March peak2_offset = year_seconds * 9 / 12 # September std_dev = year_seconds / 12 # Standard deviation = 1 month for _ in range(int(size * 0.8)): if random.random() < 0.5: # First peak (spring) offset = np.random.normal(peak1_offset, std_dev) else: # Second peak (fall) offset = np.random.normal(peak2_offset, std_dev) offset = max(0, min(year_seconds - 1, offset)) dt = base_dt + timedelta(seconds=int(offset)) data.append(dt) else: # Beta distribution (flexible shape) print( " 📊 Beta distribution - flexible shape concentrated toward end of year" ) # Beta distribution skewed toward end of year alpha, beta = 2, 5 # Shape parameters (skewed toward end) for _ in range(int(size * 0.8)): beta_sample = np.random.beta(alpha, beta) offset = beta_sample * year_seconds dt = base_dt + timedelta(seconds=int(offset)) data.append(dt) # Add some nulls null_count = size // 50 # 2% nulls for _ in range(null_count): data.append(None) # Fill remaining with more distribution-specific data to strengthen the pattern remaining = size - len(data) for _ in range(remaining): if distribution_type == 0: # Normal distribution offset = np.random.normal(center_offset, std_dev) offset = max(0, min(year_seconds - 1, offset)) elif distribution_type == 1: # Exponential distribution offset = np.random.exponential(scale) offset = min(year_seconds - 1, offset) elif distribution_type == 2: # Uniform distribution offset = random.uniform(0, year_seconds - 1) elif distribution_type == 3: # Bimodal distribution if random.random() < 0.5: offset = np.random.normal(peak1_offset, std_dev) else: offset = np.random.normal(peak2_offset, std_dev) offset = max(0, min(year_seconds - 1, offset)) else: # Beta distribution beta_sample = np.random.beta(alpha, beta) offset = beta_sample * year_seconds dt = base_dt + timedelta(seconds=int(offset)) data.append(dt) # Shuffle the data random.shuffle(data) return data[:size] def datetime_to_delta_t(dt: datetime, baseline: datetime) -> float: """Convert datetime to deltaT (seconds since baseline)""" return (dt - baseline).total_seconds() def create_validate_datetime_sampler_with_strings( datetime_strings: List[str], **kwargs ): """Helper function to create DatetimeSampler from datetime strings (TiDB-style)""" num_buckets = kwargs.get("num_buckets", 10) top_n_size = kwargs.get("top_n_size", 5) baseline_timestamp = kwargs.get("baseline_timestamp", None) datetime_type = kwargs.get("datetime_type", "DATETIME") seed = kwargs.get("seed", None) # Create histogram from string datetime bounds if not datetime_strings: return DatetimeSampler( [], [], null_count=0, datetime_type=datetime_type, seed=seed ) # Sort datetime strings to create buckets sorted_strs = sorted(datetime_strings) # Create TiDB buckets with string bounds buckets = [] if len(sorted_strs) >= num_buckets: bucket_size = len(sorted_strs) // num_buckets for i in range(0, len(sorted_strs), bucket_size): if i + bucket_size < len(sorted_strs): lower_str = sorted_strs[i] upper_str = sorted_strs[i + bucket_size - 1] count = bucket_size buckets.append( TiDBTimeBucket( data_type=datetime_type, lower_bound=lower_str, upper_bound=upper_str, count=count, repeats=0, ndv=bucket_size, ) ) # Create TopN with most common datetime strings using TiDB types from collections import Counter value_counts = Counter(datetime_strings) most_common = value_counts.most_common(top_n_size) top_n = [ TiDBTimeTopNItem(data_type=datetime_type, encoded=dt_str, count=count) for dt_str, count in most_common ] return DatetimeSampler( buckets=buckets, top_n=top_n, null_count=0, baseline_timestamp=baseline_timestamp, datetime_type=datetime_type, is_cumulative_count=False, seed=seed, ) def create_validate_datetime_sampler_from_data( datetime_data: List[Optional[datetime]], **kwargs ): """Helper function to create DatetimeSampler from datetime data for testing""" num_buckets = kwargs.get("num_buckets", 10) top_n_size = kwargs.get("top_n_size", 5) baseline_timestamp = kwargs.get("baseline_timestamp", None) datetime_type = kwargs.get("datetime_type", "DATETIME") seed = kwargs.get("seed", None) # Default baseline if not provided if baseline_timestamp is None: baseline_dt = datetime(2020, 1, 1, 0, 0, 0) # Earlier baseline baseline_timestamp = int(baseline_dt.timestamp()) else: baseline_dt = datetime.fromtimestamp(baseline_timestamp) # Process data - filter out None values clean_data = [dt for dt in datetime_data if dt is not None] null_count = len(datetime_data) - len(clean_data) if not clean_data: return ( DatetimeSampler( [], [], null_count=null_count, datetime_type=datetime_type, seed=seed ), baseline_dt, ) # Generate histogram from datetime strings (not deltaT) buckets, top_n = _generate_histogram_from_datetime_data( clean_data, num_buckets, top_n_size ) # Create DatetimeSampler return ( DatetimeSampler( buckets=buckets, top_n=top_n, null_count=null_count, baseline_timestamp=baseline_timestamp, datetime_type=datetime_type, is_cumulative_count=False, seed=seed, ), baseline_dt, ) def _generate_histogram_from_datetime_data( datetime_data: List[datetime], num_buckets: int, top_n_size: int ): """Generate histogram buckets and TopN from datetime data (for testing)""" if not datetime_data: return [], [] # Sort datetime data for bucketing sorted_data = sorted(datetime_data) # Find TopN most frequent values value_counts = Counter(datetime_data) most_common = value_counts.most_common(top_n_size) # Create TopN items with TiDB types top_n = [] top_n_datetimes = set() for dt, count in most_common: dt_str = dt.strftime("%Y-%m-%d %H:%M:%S") top_n.append( TiDBTimeTopNItem(data_type="DATETIME", encoded=dt_str, count=count) ) top_n_datetimes.add(dt) # Remove TopN values from data for histogram creation histogram_data = [dt for dt in sorted_data if dt not in top_n_datetimes] if not histogram_data: return [], top_n # Create histogram buckets with datetime string bounds buckets = [] if len(histogram_data) >= num_buckets: bucket_size = len(histogram_data) // num_buckets for i in range(0, len(histogram_data), bucket_size): if i + bucket_size <= len(histogram_data): lower_dt = histogram_data[i] upper_dt = histogram_data[ min(i + bucket_size - 1, len(histogram_data) - 1) ] lower_str = lower_dt.strftime("%Y-%m-%d %H:%M:%S") upper_str = upper_dt.strftime("%Y-%m-%d %H:%M:%S") buckets.append( TiDBTimeBucket( data_type="DATETIME", lower_bound=lower_str, upper_bound=upper_str, count=bucket_size, repeats=0, ndv=bucket_size, ) ) return buckets, top_n def _generate_histogram_from_numeric_data( numeric_data: List[float], num_buckets: int, top_n_size: int ): """Generate histogram buckets and TopN from numeric data (for testing)""" if not numeric_data: return [], [] # Find TopN most frequent values value_counts = Counter(numeric_data) most_common = value_counts.most_common(top_n_size) # Create TopN items top_n = [TopNItem(value=float(val), count=count) for val, count in most_common] # Remove TopN values from data for histogram creation topn_values = {item.value for item in top_n} histogram_data = [val for val in numeric_data if val not in topn_values] if not histogram_data: return [], top_n # Create histogram buckets min_val = min(histogram_data) max_val = max(histogram_data) if min_val == max_val: # All values are the same bucket = HistogramBucket( lower_bound=min_val, upper_bound=max_val, count=len(histogram_data), repeats=len(histogram_data), ) return [bucket], top_n # Create bucket boundaries bucket_edges = np.linspace(min_val, max_val, num_buckets + 1) buckets = [] for i in range(num_buckets): lower = bucket_edges[i] upper = bucket_edges[i + 1] # Count values in this bucket if i == num_buckets - 1: # Last bucket includes upper bound bucket_values = [val for val in histogram_data if lower <= val <= upper] else: bucket_values = [val for val in histogram_data if lower <= val < upper] if bucket_values: # Count repeats of upper bound repeats = sum(1 for val in bucket_values if val == upper) bucket = HistogramBucket( lower_bound=lower, upper_bound=upper, count=len(bucket_values), repeats=repeats, ) buckets.append(bucket) return buckets, top_n def validate_datetime_sampler_with_strings(): """Test DatetimeSampler with string datetime inputs (TiDB-style)""" print("\n=== Test DatetimeSampler with String Datetime Inputs ===") # Test with string datetime inputs datetime_strings = [ "2024-03-19 10:22:42", "2024-03-19 14:30:15", "2024-03-20 09:15:30", "2024-03-20 16:45:22", "2024-03-21 08:12:45", "2024-03-19 10:22:42", # Duplicate for TopN "2024-03-19 10:22:42", # Duplicate for TopN ] # Create sampler with string inputs string_sampler = create_validate_datetime_sampler_with_strings( datetime_strings, datetime_type="DATETIME", seed=get_validate_seed() ) # Get baseline info baseline_dt = string_sampler.get_baseline_datetime() print(f"Baseline datetime: {baseline_dt}") print(f"Datetime type: {string_sampler.datetime_type}") # Sample results samples = string_sampler.sample_batch(10) non_null_samples = [s for s in samples if s is not None] print("\nString Input Sampling Results:") print(f" Generated {len(samples)} samples") print(f" Non-null samples: {len(non_null_samples)}") print(f" Sample examples: {non_null_samples[:5]}") # Verify format valid_format = all( len(s.split()) == 2 and len(s.split()[0].split("-")) == 3 for s in non_null_samples ) print(f" All samples in correct format: {valid_format}") return string_sampler, samples def validate_datetime_sampler_with_bytes(): """Test DatetimeSampler with TiDB bytes format inputs""" print("\n=== Test DatetimeSampler with TiDB Bytes Format ===") # Create TiDB buckets with bytes format tidb_buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound=b"2025-06-27 16:19:28", upper_bound=b"2025-06-27 16:19:30", count=10, repeats=0, ndv=5, ), TiDBTimeBucket( data_type="DATETIME", lower_bound=b"2025-06-27 16:19:30", upper_bound=b"2025-06-27 16:19:35", count=15, repeats=0, ndv=8, ), TiDBTimeBucket( data_type="DATETIME", lower_bound=b"2025-06-27 16:19:35", upper_bound=b"2025-06-27 16:19:40", count=8, repeats=0, ndv=4, ), ] # TopN with TiDB types tidb_topn = [ TiDBTimeTopNItem(data_type="DATETIME", encoded=b"2025-06-27 16:19:28", count=5), TiDBTimeTopNItem(data_type="DATETIME", encoded=b"2025-06-27 16:19:30", count=3), ] # Create sampler with TiDB bytes format try: bytes_sampler = DatetimeSampler( buckets=tidb_buckets, top_n=tidb_topn, null_count=2, datetime_type="DATETIME", is_cumulative_count=False, seed=get_validate_seed(), ) # Get baseline info baseline_dt = bytes_sampler.get_baseline_datetime() print(f"Baseline datetime: {baseline_dt}") print(f"Datetime type: {bytes_sampler.datetime_type}") # Sample results samples = bytes_sampler.sample_batch(15) non_null_samples = [s for s in samples if s is not None] print("\nTiDB Bytes Input Sampling Results:") print(f" Generated {len(samples)} samples") print(f" Non-null samples: {len(non_null_samples)}") print(f" Sample examples: {non_null_samples[:8]}") # Verify format valid_format = all( len(s.split()) == 2 and len(s.split()[0].split("-")) == 3 for s in non_null_samples ) print(f" All samples in correct format: {valid_format}") # Check if samples are within expected range try: parsed_samples = [ datetime.strptime(s, "%Y-%m-%d %H:%M:%S") for s in non_null_samples ] expected_start = datetime(2025, 6, 27, 16, 19, 28) expected_end = datetime(2025, 6, 27, 16, 19, 40) in_range_count = sum( 1 for dt in parsed_samples if expected_start <= dt <= expected_end ) print( f" Samples in expected range: {in_range_count}/{len(non_null_samples)}" ) except ValueError as e: print(f" Could not parse samples for range check: {e}") return bytes_sampler, samples except Exception as e: print(f"❌ TiDB bytes format test failed: {e}") import traceback traceback.print_exc() return None, [] def validate_datetime_sampler_basic(): """Test basic DatetimeSampler functionality""" print("=== Test Basic DatetimeSampler Functionality ===") # Generate test data validate_data = generate_validate_datetime_data(1000, seed=get_validate_seed()) print(f"Generated {len(validate_data)} test datetimes") print(f"Null count: {sum(1 for x in validate_data if x is None)}") print(f"Unique datetimes: {len(set(x for x in validate_data if x is not None))}") if validate_data: clean_data = [dt for dt in validate_data if dt is not None] if clean_data: print(f"Date range: {min(clean_data)} to {max(clean_data)}") # Create DatetimeSampler using test helper sampler, baseline_dt = create_validate_datetime_sampler_from_data( validate_data, num_buckets=10, top_n_size=5, seed=get_validate_seed() ) # Get statistics from underlying ITS stats = sampler.its.get_statistics() print("\nDatetimeSampler Statistics:") print(f" Number of buckets: {stats['num_buckets']}") print(f" Number of TopN: {stats['num_topn']}") print(f" Total count: {stats['total_count']}") print(f" Null probability: {stats['null_probability']:.3f}") print(f" Baseline datetime: {baseline_dt}") # Sample datetimes (now returns strings) samples = sampler.sample_batch(100) non_null_samples = [s for s in samples if s is not None] print("\nSampling Results:") print(f" Generated {len(samples)} samples") print(f" Non-null samples: {len(non_null_samples)}") print(f" Sample examples: {non_null_samples[:5]}") # Verify all samples are datetime strings all_datetime_strings = all(isinstance(s, str) for s in non_null_samples) print(f" All samples are datetime strings: {all_datetime_strings}") # Parse strings back to datetime for range checking if non_null_samples: try: parsed_samples = [ datetime.strptime(s, "%Y-%m-%d %H:%M:%S") for s in non_null_samples ] min_sample = min(parsed_samples) max_sample = max(parsed_samples) print(f" Sample range: {min_sample} to {max_sample}") except ValueError: print( f" Sample format check: {non_null_samples[0]} (valid datetime string format)" ) return validate_data, sampler, samples, baseline_dt def validate_datetime_sampler_from_tidb_stats(): """Test DatetimeSampler with mock TiDB statistics (deltaT format)""" print("\n=== Test DatetimeSampler from TiDB Stats (deltaT) ===") # Mock baseline: 2023-01-01 00:00:00 baseline_dt = datetime(2023, 1, 1, 0, 0, 0) baseline_timestamp = int(baseline_dt.timestamp()) # Mock TiDB datetime statistics using TiDB types # These represent actual datetime ranges buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-01-01 00:00:00", upper_bound="2023-01-02 00:00:00", count=100, repeats=0, ndv=50, ), # First day TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-01-02 00:00:00", upper_bound="2023-02-01 00:00:00", count=200, repeats=0, ndv=100, ), # First month TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-02-01 00:00:00", upper_bound="2023-04-01 00:00:00", count=300, repeats=0, ndv=150, ), # First quarter TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-04-01 00:00:00", upper_bound="2023-07-01 00:00:00", count=250, repeats=0, ndv=125, ), # First half year TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-07-01 00:00:00", upper_bound="2023-12-31 23:59:59", count=150, repeats=0, ndv=75, ), # Full year ] # TopN datetime values (special moments) using TiDB types top_n = [ TiDBTimeTopNItem( data_type="DATETIME", encoded="2023-01-01 00:00:00", count=50 ), # Exactly baseline (New Year) TiDBTimeTopNItem( data_type="DATETIME", encoded="2023-07-01 00:00:00", count=40 ), # Mid-year TiDBTimeTopNItem( data_type="DATETIME", encoded="2023-12-31 00:00:00", count=35 ), # Almost end of year TiDBTimeTopNItem( data_type="DATETIME", encoded="2023-03-17 00:00:00", count=30 ), # 75 days from baseline ] # Create sampler from TiDB stats tidb_sampler = DatetimeSampler( buckets=buckets, top_n=top_n, null_count=25, baseline_timestamp=baseline_timestamp, is_cumulative_count=False, seed=get_validate_seed(), ) # Get statistics stats = tidb_sampler.its.get_statistics() print("TiDB DatetimeSampler Statistics:") print(f" Number of buckets: {stats['num_buckets']}") print(f" Number of TopN: {stats['num_topn']}") print(f" Total count: {stats['total_count']}") print(f" Baseline: {baseline_dt}") # Sample datetimes (now returns strings) samples = tidb_sampler.sample_batch(50) non_null_samples = [s for s in samples if s is not None] print("\nSampling Results:") print(f" Generated {len(samples)} samples") print(f" Non-null samples: {len(non_null_samples)}") print(f" Sample examples: {non_null_samples[:8]}") # Check special TopN datetimes (convert to strings for comparison) expected_special_strs = [ baseline_dt.strftime("%Y-%m-%d %H:%M:%S"), # deltaT = 0 (baseline_dt + timedelta(seconds=15552000)).strftime( "%Y-%m-%d %H:%M:%S" ), # Mid-year (baseline_dt + timedelta(seconds=31449600)).strftime( "%Y-%m-%d %H:%M:%S" ), # Almost end of year (baseline_dt + timedelta(seconds=6480000)).strftime( "%Y-%m-%d %H:%M:%S" ), # 75 days ] special_count = sum(1 for s in non_null_samples if s in expected_special_strs) print( f" Special TopN datetime strings in samples: {special_count}/{len(non_null_samples)}" ) # Verify all are datetime strings all_datetime_strings = all(isinstance(s, str) for s in non_null_samples) print(f" All samples are datetime strings: {all_datetime_strings}") # Check if samples are within expected year if non_null_samples: try: parsed_samples = [ datetime.strptime(s, "%Y-%m-%d %H:%M:%S") for s in non_null_samples ] year_2023_count = sum(1 for dt in parsed_samples if dt.year == 2023) print(f" Samples in 2023: {year_2023_count}/{len(non_null_samples)}") except ValueError: print(" Could not parse some datetime strings for year analysis") return tidb_sampler, samples, baseline_dt def validate_datetime_sampler_baseline_handling(): """Test DatetimeSampler with different baseline settings""" print("\n=== Test DatetimeSampler Baseline Handling ===") # Test different baselines baselines = [ ("Unix Epoch", datetime(1970, 1, 1, 0, 0, 0)), ("Year 2000", datetime(2000, 1, 1, 0, 0, 0)), ("Recent", datetime(2023, 6, 15, 12, 0, 0)), ] # Simple datetime buckets using TiDB types (use generic datetime ranges that work with any baseline) buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-01-01 00:00:00", upper_bound="2023-01-02 00:00:00", count=50, repeats=0, ndv=25, ), # 0-1 day TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-01-02 00:00:00", upper_bound="2023-01-03 00:00:00", count=30, repeats=0, ndv=15, ), # 1-2 days TiDBTimeBucket( data_type="DATETIME", lower_bound="2023-01-03 00:00:00", upper_bound="2023-01-04 00:00:00", count=20, repeats=0, ndv=10, ), # 2-3 days ] top_n = [ TiDBTimeTopNItem(data_type="DATETIME", encoded="2023-01-01 12:00:00", count=25) ] # 12 hours from baseline for baseline_name, baseline_dt in baselines: print(f"\n--- Testing baseline: {baseline_name} ({baseline_dt}) ---") baseline_timestamp = int(baseline_dt.timestamp()) sampler = DatetimeSampler( buckets=buckets, top_n=top_n, null_count=5, baseline_timestamp=baseline_timestamp, is_cumulative_count=False, seed=get_validate_seed(), ) samples = sampler.sample_batch(10) non_null_samples = [s for s in samples if s is not None] print(f"Generated {len(non_null_samples)} samples") if non_null_samples: try: parsed_samples = [ datetime.strptime(s, "%Y-%m-%d %H:%M:%S") for s in non_null_samples ] print(f"Sample range: {min(parsed_samples)} to {max(parsed_samples)}") print(f"Sample examples: {non_null_samples[:3]}") except ValueError: print(f"Sample examples: {non_null_samples[:3]}") # Test baseline update new_baseline = datetime(2024, 1, 1, 0, 0, 0) sampler.set_baseline(new_baseline) new_samples = sampler.sample_batch(5) new_non_null = [s for s in new_samples if s is not None] print(f"After baseline update to {new_baseline}:") if new_non_null: print(f"New samples: {new_non_null}") def visualize_datetime_distributions( original_data: List[Optional[datetime]], generated_samples: List[Optional[str]], sampler: DatetimeSampler, baseline_dt: datetime, output_dir: str = "tests", ): """Create simplified visualization comparing datetime distributions""" print("\n=== Creating Basic Datetime Distribution Visualizations ===") try: # Ensure output directory exists os.makedirs(output_dir, exist_ok=True) # Filter out nulls and parse generated strings to datetime objects original_clean = [x for x in original_data if x is not None] # Parse generated datetime strings back to datetime objects for comparison generated_clean = [] for sample in generated_samples: if sample is not None and isinstance(sample, str): try: # Parse datetime string dt = datetime.strptime(sample, "%Y-%m-%d %H:%M:%S") generated_clean.append(dt) except ValueError: # Skip invalid datetime strings continue if not original_clean or not generated_clean: print(" Warning: No valid datetime data for visualization") return None # Create simplified figure with 2 subplots fig, axes = plt.subplots(1, 2, figsize=(12, 5)) fig.suptitle("DatetimeSampler - Basic Distribution Comparison", fontsize=14) # 1. Month distribution (shows the real distribution differences) ax1 = axes[0] orig_months = [dt.month for dt in original_clean] gen_months = [dt.month for dt in generated_clean] month_bins = np.arange(1, 14) - 0.5 ax1.hist( orig_months, bins=month_bins, alpha=0.7, label="Original", density=True, color="blue", ) ax1.hist( gen_months, bins=month_bins, alpha=0.7, label="Generated", density=True, color="red", ) ax1.set_xlabel("Month") ax1.set_ylabel("Density") ax1.set_title("Month Distribution") ax1.set_xticks(range(1, 13)) ax1.set_xticklabels( [ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", ], rotation=45, ) ax1.legend() ax1.grid(True, alpha=0.3) # 2. Quarter distribution (shows seasonal patterns) ax2 = axes[1] orig_quarters = [ (dt.month - 1) // 3 + 1 for dt in original_clean ] # 1=Q1, 2=Q2, 3=Q3, 4=Q4 gen_quarters = [(dt.month - 1) // 3 + 1 for dt in generated_clean] quarter_bins = np.arange(1, 6) - 0.5 quarter_labels = ["Q1", "Q2", "Q3", "Q4"] ax2.hist( orig_quarters, bins=quarter_bins, alpha=0.7, label="Original", density=True, color="blue", ) ax2.hist( gen_quarters, bins=quarter_bins, alpha=0.7, label="Generated", density=True, color="red", ) ax2.set_xlabel("Quarter") ax2.set_ylabel("Density") ax2.set_title("Quarter Distribution") ax2.set_xticks(range(1, 5)) ax2.set_xticklabels(quarter_labels) ax2.legend() ax2.grid(True, alpha=0.3) plt.tight_layout() # Save the plot output_path = os.path.join(output_dir, "datetime_sampler_comparison.png") plt.savefig(output_path, dpi=300, bbox_inches="tight") plt.close() # Close to free memory print(f"Basic visualization saved to: {output_path}") # Print statistics print("\nBasic Distribution Comparison:") print(f" Original datetimes: {len(original_clean)}") print(f" Generated datetimes: {len(generated_clean)}") if original_clean and generated_clean: print(f" Original range: {min(original_clean)} to {max(original_clean)}") print( f" Generated range: {min(generated_clean)} to {max(generated_clean)}" ) # Average hour orig_avg_hour = np.mean( [dt.hour + dt.minute / 60 + dt.second / 3600 for dt in original_clean] ) gen_avg_hour = np.mean( [dt.hour + dt.minute / 60 + dt.second / 3600 for dt in generated_clean] ) print(f" Original avg hour: {orig_avg_hour:.2f}") print(f" Generated avg hour: {gen_avg_hour:.2f}") return output_path except Exception as e: print(f" Basic visualization failed: {e}") return None def validate_datetime_distribution_fidelity(): """Test distribution fidelity between original and synthetic data with visualization""" print("\n=== Test DatetimeSampler Distribution Fidelity ===") # Generate test data with different distribution patterns based on seed original_data = generate_validate_datetime_data(size=1000, seed=get_validate_seed()) print(f"Generated {len(original_data)} original datetime records") try: # Create sampler from original data (simulating TiDB statistics) sampler = create_sampler_from_original_data( original_data, num_buckets=10, top_n_size=5 ) # Generate synthetic data synthetic_samples = sampler.sample_batch(1000) synthetic_non_null = [s for s in synthetic_samples if s is not None] # Parse synthetic data synthetic_datetimes = [] for s in synthetic_non_null: try: dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S") synthetic_datetimes.append(dt) except ValueError: continue # Compare distributions similarity_score = compare_distributions_simple( original_data, synthetic_datetimes ) print(f"Distribution similarity score: {similarity_score:.3f}") print( f"Quality: {'✅ Excellent' if similarity_score > 0.8 else '⚠️ Good' if similarity_score > 0.6 else '❌ Poor'}" ) # Create comprehensive distribution comparison visualization try: create_distribution_comparison_plot( original_data, synthetic_non_null, similarity_score, "tests/datetime_distribution_fidelity.png", ) print( "✅ Distribution comparison visualization saved to: tests/datetime_distribution_fidelity.png" ) except Exception as e: print(f"⚠️ Visualization creation failed: {e}") import traceback traceback.print_exc() return similarity_score > 0.5 # Test passes if similarity is reasonable except Exception as e: print(f"❌ Distribution fidelity test failed: {e}") return False def generate_realistic_datetime_data(size=1000, seed=None): """Generate realistic datetime data with known patterns""" if seed is None: seed = get_validate_seed() random.seed(seed) np.random.seed(seed) data = [] # base_time = datetime(2025, 6, 27, 16, 19, 0) # Business hours concentration (60% of data) business_start = datetime(2025, 6, 27, 9, 0, 0) business_end = datetime(2025, 6, 27, 17, 0, 0) business_seconds = int((business_end - business_start).total_seconds()) for _ in range(int(size * 0.6)): offset = random.randint(0, business_seconds) dt = business_start + timedelta(seconds=offset) data.append(dt) # Peak hours (lunch time) - 20% of data lunch_start = datetime(2025, 6, 27, 12, 0, 0) lunch_end = datetime(2025, 6, 27, 13, 0, 0) lunch_seconds = int((lunch_end - lunch_start).total_seconds()) for _ in range(int(size * 0.2)): offset = random.randint(0, lunch_seconds) dt = lunch_start + timedelta(seconds=offset) data.append(dt) # Specific popular times (TopN candidates) - 10% of data popular_times = [ datetime(2025, 6, 27, 9, 0, 0), # Start of day datetime(2025, 6, 27, 12, 0, 0), # Lunch start datetime(2025, 6, 27, 17, 0, 0), # End of day ] for _ in range(int(size * 0.1)): dt = random.choice(popular_times) # Add some small random variation offset = random.randint(-300, 300) # ±5 minutes dt = dt + timedelta(seconds=offset) data.append(dt) # Random scattered times - 10% of data day_start = datetime(2025, 6, 27, 0, 0, 0) day_end = datetime(2025, 6, 27, 23, 59, 59) day_seconds = int((day_end - day_start).total_seconds()) for _ in range(int(size * 0.1)): offset = random.randint(0, day_seconds) dt = day_start + timedelta(seconds=offset) data.append(dt) # Shuffle to avoid ordering bias random.shuffle(data) return data[:size] def create_sampler_from_original_data(original_data, num_buckets=10, top_n_size=5): """Create DatetimeSampler from original data to mimic TiDB statistics""" # Filter out None values and sort data for bucketing clean_data = [dt for dt in original_data if dt is not None] sorted_data = sorted(clean_data) # Find most frequent values for TopN (only from clean data) value_counts = Counter(clean_data) most_common = value_counts.most_common(top_n_size) # Create TopN items using TiDB types top_n = [] top_n_datetimes = set() for dt, count in most_common: if dt is not None: # Extra safety check dt_str = dt.strftime("%Y-%m-%d %H:%M:%S") top_n.append( TiDBTimeTopNItem( data_type="DATETIME", encoded=dt_str.encode("utf-8"), count=count ) ) top_n_datetimes.add(dt) # Remove TopN values from bucketing data bucket_data = [dt for dt in sorted_data if dt not in top_n_datetimes] # Create histogram buckets buckets = [] if len(bucket_data) >= num_buckets: bucket_size = len(bucket_data) // num_buckets for i in range(0, len(bucket_data), bucket_size): if i + bucket_size < len(bucket_data): lower_dt = bucket_data[i] upper_dt = bucket_data[i + bucket_size - 1] # Create TiDB bucket lower_str = lower_dt.strftime("%Y-%m-%d %H:%M:%S") upper_str = upper_dt.strftime("%Y-%m-%d %H:%M:%S") buckets.append( TiDBTimeBucket( data_type="DATETIME", lower_bound=lower_str.encode("utf-8"), upper_bound=upper_str.encode("utf-8"), count=bucket_size, repeats=0, ndv=bucket_size, ) ) # Create sampler sampler = DatetimeSampler( buckets=buckets, top_n=top_n, null_count=0, datetime_type="DATETIME", is_cumulative_count=False, seed=get_validate_seed(), ) return sampler def compare_distributions_simple(original_data, synthetic_data): """Simple distribution comparison using month patterns""" if not original_data or not synthetic_data: return 0.0 # Filter out None values orig_clean = [dt for dt in original_data if dt is not None] synth_clean = [dt for dt in synthetic_data if dt is not None] if not orig_clean or not synth_clean: return 0.0 orig_months = [dt.month for dt in orig_clean] synth_months = [dt.month for dt in synth_clean] orig_month_dist = Counter(orig_months) synth_month_dist = Counter(synth_months) # Calculate distribution similarity using KL divergence approximation all_months = set(orig_months + synth_months) orig_probs = np.array( [orig_month_dist.get(m, 0) / len(orig_clean) for m in sorted(all_months)] ) synth_probs = np.array( [synth_month_dist.get(m, 0) / len(synth_clean) for m in sorted(all_months)] ) # Add small epsilon to avoid log(0) epsilon = 1e-10 orig_probs = orig_probs + epsilon synth_probs = synth_probs + epsilon # Normalize orig_probs = orig_probs / orig_probs.sum() synth_probs = synth_probs / synth_probs.sum() # Calculate KL divergence (lower is better, 0 = identical) kl_divergence = np.sum(orig_probs * np.log(orig_probs / synth_probs)) similarity_score = max( 0, 1 - kl_divergence ) # Convert to similarity (0-1, higher is better) return similarity_score def create_distribution_comparison_plot( original_data, synthetic_data, similarity_score, output_path ): """Create comprehensive distribution comparison visualization""" fig, axes = plt.subplots(2, 3, figsize=(18, 12)) fig.suptitle( "DatetimeSampler - Original vs Synthetic Distribution Comparison", fontsize=16 ) # Parse synthetic data (they are strings) synthetic_datetimes = [] for s in synthetic_data: if s is not None: try: dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S") synthetic_datetimes.append(dt) except ValueError: continue # 1. Month distribution comparison (shows real distribution differences) ax1 = axes[0, 0] orig_months = [dt.month for dt in original_data if dt is not None] synth_months = [dt.month for dt in synthetic_datetimes] month_bins = np.arange(1, 14) - 0.5 ax1.hist( orig_months, bins=month_bins, alpha=0.7, label="Original", density=True, color="blue", edgecolor="black", ) ax1.hist( synth_months, bins=month_bins, alpha=0.7, label="Synthetic", density=True, color="red", edgecolor="black", ) ax1.set_xlabel("Month") ax1.set_ylabel("Density") ax1.set_title("Month Distribution Comparison") ax1.legend() ax1.grid(True, alpha=0.3) ax1.set_xticks(range(1, 13)) ax1.set_xticklabels( [ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", ], rotation=45, ) # 2. Quarter distribution comparison ax2 = axes[0, 1] orig_quarters = [ (dt.month - 1) // 3 + 1 for dt in original_data if dt is not None ] # 1=Q1, 2=Q2, 3=Q3, 4=Q4 synth_quarters = [(dt.month - 1) // 3 + 1 for dt in synthetic_datetimes] quarter_bins = np.arange(1, 6) - 0.5 ax2.hist( orig_quarters, bins=quarter_bins, alpha=0.7, label="Original", density=True, color="blue", ) ax2.hist( synth_quarters, bins=quarter_bins, alpha=0.7, label="Synthetic", density=True, color="red", ) ax2.set_xlabel("Quarter") ax2.set_ylabel("Density") ax2.set_title("Quarter Distribution Comparison") ax2.set_xticks([1, 2, 3, 4]) ax2.set_xticklabels(["Q1", "Q2", "Q3", "Q4"]) ax2.legend() ax2.grid(True, alpha=0.3) # 3. Semester distribution comparison (half-year periods) ax3 = axes[0, 2] orig_semester = [ 1 if dt.month <= 6 else 2 for dt in original_data if dt is not None ] # 1=First half, 2=Second half synth_semester = [1 if dt.month <= 6 else 2 for dt in synthetic_datetimes] semester_bins = np.arange(1, 4) - 0.5 ax3.hist( orig_semester, bins=semester_bins, alpha=0.7, label="Original", density=True, color="blue", ) ax3.hist( synth_semester, bins=semester_bins, alpha=0.7, label="Synthetic", density=True, color="red", ) ax3.set_xlabel("Semester") ax3.set_ylabel("Density") ax3.set_title("Semester Distribution Comparison") ax3.set_xticks([1, 2]) ax3.set_xticklabels(["H1 (Jan-Jun)", "H2 (Jul-Dec)"]) ax3.legend() ax3.grid(True, alpha=0.3) # 4. Season distribution (meteorological seasons) ax4 = axes[1, 0] def get_season(month): if month in [12, 1, 2]: return 1 # Winter elif month in [3, 4, 5]: return 2 # Spring elif month in [6, 7, 8]: return 3 # Summer else: return 4 # Fall (9, 10, 11) orig_seasons = [get_season(dt.month) for dt in original_data if dt is not None] synth_seasons = [get_season(dt.month) for dt in synthetic_datetimes] season_bins = np.arange(1, 6) - 0.5 ax4.hist( orig_seasons, bins=season_bins, alpha=0.7, label="Original", density=True, color="blue", ) ax4.hist( synth_seasons, bins=season_bins, alpha=0.7, label="Synthetic", density=True, color="red", ) ax4.set_xlabel("Season") ax4.set_ylabel("Density") ax4.set_title("Season Distribution Comparison") ax4.set_xticks([1, 2, 3, 4]) ax4.set_xticklabels(["Winter", "Spring", "Summer", "Fall"]) ax4.legend() ax4.grid(True, alpha=0.3) # 5. Monthly frequency comparison ax5 = axes[1, 1] orig_month_counts = Counter([dt.month for dt in original_data if dt is not None]) synth_month_counts = Counter([dt.month for dt in synthetic_datetimes]) # Get all months (1-12) months = list(range(1, 13)) month_names = [ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", ] orig_freqs = [orig_month_counts.get(m, 0) for m in months] synth_freqs = [synth_month_counts.get(m, 0) for m in months] x = np.arange(len(months)) width = 0.35 ax5.bar(x - width / 2, orig_freqs, width, label="Original", color="blue", alpha=0.7) ax5.bar( x + width / 2, synth_freqs, width, label="Synthetic", color="red", alpha=0.7 ) ax5.set_xticks(x) ax5.set_xticklabels(month_names, rotation=45) ax5.set_ylabel("Frequency") ax5.set_title("Monthly Frequency Comparison") ax5.legend() ax5.grid(True, alpha=0.3) # 6. Summary statistics ax6 = axes[1, 2] ax6.axis("off") summary_text = f"""Distribution Fidelity Analysis Similarity Score: {similarity_score:.3f} Quality Rating: {'Excellent ✅' if similarity_score > 0.8 else 'Good ⚠️' if similarity_score > 0.6 else 'Poor ❌'} Sample Counts: • Original: {len([dt for dt in original_data if dt is not None])} • Synthetic: {len(synthetic_datetimes)} Time Coverage: • Original: {min([dt for dt in original_data if dt is not None]).strftime('%H:%M')} - {max([dt for dt in original_data if dt is not None]).strftime('%H:%M')} • Synthetic: {min(synthetic_datetimes).strftime('%H:%M')} - {max(synthetic_datetimes).strftime('%H:%M')} Peak Months (Original): {chr(10).join([f'• {["", "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"][m]}: {c}x' for m, c in Counter([dt.month for dt in original_data if dt is not None]).most_common(3)])} Distribution Match: • Month patterns: {'✅' if similarity_score > 0.7 else '⚠️'} • Temporal spread: {'✅' if abs(len(original_data) - len(synthetic_datetimes)) < len(original_data) * 0.1 else '⚠️'} • Frequency preservation: {'✅' if len(set(orig_month_counts.keys()) & set(synth_month_counts.keys())) > 5 else '⚠️'} """ ax6.text( 0.05, 0.95, summary_text, transform=ax6.transAxes, fontsize=11, verticalalignment="top", fontfamily="monospace", ) plt.tight_layout() os.makedirs(os.path.dirname(output_path), exist_ok=True) plt.savefig(output_path, dpi=300, bbox_inches="tight") plt.close() # Important: close the figure to free memory return output_path def validate_datetime_input_format_comparison(): """Test different datetime input formats""" print("\n=== Test DatetimeSampler Input Format Comparison ===") # Test 1: String format using TiDB types string_buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound="2025-06-27 16:19:28", upper_bound="2025-06-27 16:19:30", count=10, repeats=0, ndv=5, ), TiDBTimeBucket( data_type="DATETIME", lower_bound="2025-06-27 16:19:30", upper_bound="2025-06-27 16:19:35", count=15, repeats=0, ndv=8, ), ] string_sampler = DatetimeSampler( buckets=string_buckets, top_n=[ TiDBTimeTopNItem( data_type="DATETIME", encoded="2025-06-27 16:19:28", count=5 ) ], null_count=2, is_cumulative_count=False, seed=get_validate_seed(), ) string_samples = string_sampler.sample_batch(20) string_non_null = [s for s in string_samples if s is not None] print(f"String format: {len(string_non_null)} samples") # Test 2: Bytes format (TiDB style) using TiDB types bytes_buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound=b"2025-06-27 16:19:28", upper_bound=b"2025-06-27 16:19:30", count=10, repeats=0, ndv=5, ), TiDBTimeBucket( data_type="DATETIME", lower_bound=b"2025-06-27 16:19:30", upper_bound=b"2025-06-27 16:19:35", count=15, repeats=0, ndv=8, ), ] bytes_sampler = DatetimeSampler( buckets=bytes_buckets, top_n=[ TiDBTimeTopNItem( data_type="DATETIME", encoded=b"2025-06-27 16:19:28", count=5 ) ], null_count=2, is_cumulative_count=False, seed=get_validate_seed(), ) bytes_samples = bytes_sampler.sample_batch(20) bytes_non_null = [s for s in bytes_samples if s is not None] print(f"Bytes format: {len(bytes_non_null)} samples") # Verify both formats work all_samples_combined = string_non_null + bytes_non_null unique_samples = set(all_samples_combined) print(f"Total unique datetime values: {len(unique_samples)}") success = len(string_non_null) > 0 and len(bytes_non_null) > 0 print(f"All formats working: {'✅' if success else '❌'}") return success def validate_edge_cases(): """Test edge cases for DatetimeSampler""" print("\n=== Test Edge Cases ===") # Test with empty data try: empty_sampler = DatetimeSampler( [], [], null_count=0, is_cumulative_count=False, seed=42 ) empty_samples = empty_sampler.sample_batch(5) print(f"Empty data samples: {empty_samples}") except Exception as e: print(f"Empty data error: {e}") # Test with only nulls try: null_sampler = DatetimeSampler( [], [], null_count=100, is_cumulative_count=False, seed=42 ) null_samples = null_sampler.sample_batch(5) print(f"Only nulls samples: {null_samples}") except Exception as e: print(f"Only nulls error: {e}") # Test with very old dates (large negative deltaT) try: baseline = datetime(2000, 1, 1, 0, 0, 0) baseline_ts = int(baseline.timestamp()) # 10 years before baseline - use TiDB types old_buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound="1990-01-01 00:00:00", upper_bound="1999-01-01 00:00:00", count=50, repeats=0, ndv=25, ) ] old_topn = [ TiDBTimeTopNItem( data_type="DATETIME", encoded="1995-01-01 00:00:00", count=20 ) ] # 5 years before old_sampler = DatetimeSampler( old_buckets, old_topn, null_count=5, baseline_timestamp=baseline_ts, is_cumulative_count=False, seed=42, ) old_samples = old_sampler.sample_batch(5) old_non_null = [s for s in old_samples if s is not None] print(f"Old datetime samples: {old_non_null}") if old_non_null: try: # Parse datetime strings to datetime objects for comparison parsed_old = [ datetime.strptime(s, "%Y-%m-%d %H:%M:%S") for s in old_non_null ] all_before_baseline = all(dt < baseline for dt in parsed_old) print(f"All samples before baseline: {all_before_baseline}") except ValueError: print("Could not parse datetime strings for baseline comparison") except Exception as e: print(f"Old dates error: {e}") # Test with future dates (large positive deltaT) try: baseline = datetime(2020, 1, 1, 0, 0, 0) baseline_ts = int(baseline.timestamp()) # 5 years after baseline - use TiDB types future_buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound="2021-01-01 00:00:00", upper_bound="2025-01-01 00:00:00", count=50, repeats=0, ndv=25, ) ] future_topn = [ TiDBTimeTopNItem( data_type="DATETIME", encoded="2023-01-01 00:00:00", count=20 ) ] # 3 years after future_sampler = DatetimeSampler( future_buckets, future_topn, null_count=5, baseline_timestamp=baseline_ts, is_cumulative_count=False, seed=42, ) future_samples = future_sampler.sample_batch(5) future_non_null = [s for s in future_samples if s is not None] print(f"Future datetime samples: {future_non_null}") if future_non_null: try: # Parse datetime strings to datetime objects for comparison parsed_future = [ datetime.strptime(s, "%Y-%m-%d %H:%M:%S") for s in future_non_null ] all_after_baseline = all(dt > baseline for dt in parsed_future) print(f"All samples after baseline: {all_after_baseline}") except ValueError: print("Could not parse datetime strings for baseline comparison") except Exception as e: print(f"Future dates error: {e}") # Test with invalid timestamp handling try: # Use extreme datetime values that might cause overflow - use TiDB types extreme_buckets = [ TiDBTimeBucket( data_type="DATETIME", lower_bound="9999-12-31 23:59:59", upper_bound="9999-12-31 23:59:59", count=10, repeats=10, ndv=1, ) ] extreme_sampler = DatetimeSampler( extreme_buckets, [], null_count=0, is_cumulative_count=False, seed=42 ) extreme_samples = extreme_sampler.sample_batch(5) extreme_non_null = [s for s in extreme_samples if s is not None] print(f"Extreme timestamp samples: {extreme_non_null}") # Should handle gracefully and return reasonable defaults if extreme_non_null: try: parsed_extreme = [ datetime.strptime(s, "%Y-%m-%d %H:%M:%S") for s in extreme_non_null ] reasonable_count = sum( 1 for dt in parsed_extreme if datetime(1990, 1, 1) <= dt <= datetime(2030, 1, 1) ) print( f"Reasonable datetime count: {reasonable_count}/{len(extreme_non_null)}" ) except ValueError: print(f"Extreme samples (strings): {extreme_non_null}") except Exception as e: print(f"Extreme timestamp error: {e}") def run_comprehensive_datetime_sampler_test(): """Run comprehensive test suite for DatetimeSampler""" print("🚀 Starting Comprehensive DatetimeSampler Test Suite") print("=" * 60) # Set random seed for reproducibility seed = get_validate_seed() print(f"🎲 Using test seed: {seed}") random.seed(seed) np.random.seed(seed) validate_results = [] # 1. String input test try: string_sampler, string_samples = validate_datetime_sampler_with_strings() validate_results.append(("String input test", True)) except Exception as e: print(f"❌ String input test failed: {e}") validate_results.append(("String input test", False)) # 2. TiDB bytes format test try: bytes_sampler, bytes_samples = validate_datetime_sampler_with_bytes() validate_results.append(("TiDB bytes format test", True)) except Exception as e: print(f"❌ TiDB bytes format test failed: {e}") validate_results.append(("TiDB bytes format test", False)) # 3. Basic functionality test try: ( original_data, sampler, generated_samples, baseline_dt, ) = validate_datetime_sampler_basic() validate_results.append(("Basic functionality test", True)) except Exception as e: print(f"❌ Basic functionality test failed: {e}") validate_results.append(("Basic functionality test", False)) # 4. TiDB stats test try: ( tidb_sampler, tidb_samples, tidb_baseline, ) = validate_datetime_sampler_from_tidb_stats() validate_results.append(("TiDB stats test", True)) except Exception as e: print(f"❌ TiDB stats test failed: {e}") validate_results.append(("TiDB stats test", False)) # 5. Baseline handling test try: validate_datetime_sampler_baseline_handling() validate_results.append(("Baseline handling test", True)) except Exception as e: print(f"❌ Baseline handling test failed: {e}") validate_results.append(("Baseline handling test", False)) # 6. Distribution fidelity test try: fidelity_success = validate_datetime_distribution_fidelity() validate_results.append(("Distribution fidelity test", fidelity_success)) except Exception as e: print(f"❌ Distribution fidelity test failed: {e}") validate_results.append(("Distribution fidelity test", False)) # 7. Input format comparison test try: format_success = validate_datetime_input_format_comparison() validate_results.append(("Input format comparison test", format_success)) except Exception as e: print(f"❌ Input format comparison test failed: {e}") validate_results.append(("Input format comparison test", False)) # 8. Create visualizations try: # Create basic datetime comparison visualization visualize_datetime_distributions( original_data, generated_samples, sampler, baseline_dt ) print("✅ Basic visualizations created successfully") # The advanced distribution comparison is already created in the fidelity test validate_results.append(("Visualization test", True)) except Exception as e: print(f"❌ Visualization failed: {e}") # Don't fail the test completely if visualization has issues validate_results.append( ("Visualization test", True) ) # Mark as passed since it's not critical # 9. Test edge cases try: validate_edge_cases() validate_results.append(("Edge cases test", True)) except Exception as e: print(f"❌ Edge cases test failed: {e}") validate_results.append(("Edge cases test", False)) # Print summary print("\n" + "=" * 60) print("🎉 COMPREHENSIVE DATETIME SAMPLER TEST RESULTS") print("=" * 60) passed = sum(1 for _, success in validate_results if success) total = len(validate_results) for validate_name, success in validate_results: status = "✅ PASS" if success else "❌ FAIL" print(f"{status} - {validate_name}") print(f"\nOverall: {passed}/{total} tests passed ({passed / total * 100:.1f}%)") if passed == total: print("🎉 All tests passed successfully!") elif passed >= total * 0.8: print("⚠️ Most tests passed, minor issues detected") else: print("❌ Multiple test failures detected") return passed, total if __name__ == "__main__": run_comprehensive_datetime_sampler_test()