| """Unit & integration test suite for locked research dataset ingestion and preprocessing.""" |
|
|
| import unittest |
| import os |
| import pandas as pd |
| import numpy as np |
| from pathlib import Path |
|
|
| from ml_rl.preprocessing.dataset_loader import load_raw_dataset, create_test_fixture |
| from ml_rl.preprocessing.schema import ( |
| TRIAGE_MODEL_INPUT_FEATURES, |
| METADATA_IDENTIFIERS, |
| ALL_TRIAGE_SAFE_FIELDS, |
| TARGET_COLUMN, |
| LLM_INPUT_SAFE_FIELDS, |
| ) |
| from ml_rl.preprocessing.feature_availability import get_triage_model_inputs, get_llm_input_features, audit_dataframe_leakage |
| from ml_rl.preprocessing.cleaning import clean_text, build_combined_text |
| from ml_rl.preprocessing.normalization import normalize_priority_value, normalize_target_column |
| from ml_rl.preprocessing.validation import handle_missing_values, validate_schema, check_data_quality |
| from ml_rl.preprocessing.splitting import select_reproducible_subset, split_group_aware_by_project, verify_zero_project_leakage |
| from ml_rl.preprocessing.manifest import generate_dataset_manifest, MANIFEST_PATH |
| from ml_rl.preprocessing.reporting import generate_quality_report, REPORT_PATH |
| from ml_rl.preprocessing.runner import run_preprocessing_pipeline |
|
|
|
|
| class TestPreprocessingPipeline(unittest.TestCase): |
| """Comprehensive test suite verifying official ground truth, schema, splitting, and leakage guards.""" |
|
|
| def setUp(self): |
| self.df_raw = create_test_fixture(num_rows=100) |
|
|
| def test_1_target_comes_exclusively_from_priority_score(self): |
| """1. Verifies research target comes strictly from priority_score in {1,2,3,4,5}.""" |
| self.assertEqual(normalize_priority_value(1), 0) |
| self.assertEqual(normalize_priority_value(2), 1) |
| self.assertEqual(normalize_priority_value(3), 2) |
| self.assertEqual(normalize_priority_value(4), 3) |
| self.assertEqual(normalize_priority_value(5), 4) |
|
|
| df_norm = normalize_target_column(self.df_raw) |
| self.assertTrue((df_norm["priority_normalized"].isin([0, 1, 2, 3, 4])).all()) |
|
|
| def test_2_heuristic_aliases_cannot_become_research_labels(self): |
| """2. Verifies heuristic priority aliases (Major, Minor, Blocker, etc.) are ignored if priority_score is missing.""" |
| df_alias = self.df_raw.copy() |
| df_alias["priority_score"] = np.nan |
| df_alias["Priority"] = "Major" |
| df_alias["priority_name"] = "Blocker" |
|
|
| df_norm = normalize_target_column(df_alias) |
| self.assertEqual(len(df_norm), 0, "Heuristic aliases must not become research labels!") |
|
|
| def test_3_all_research_labels_in_0_to_4(self): |
| """3. Verifies all valid normalized labels fall strictly in range 0..4.""" |
| for val in [1, 2, 3, 4, 5]: |
| lbl = normalize_priority_value(val) |
| self.assertIn(lbl, [0, 1, 2, 3, 4]) |
| for invalid_val in [0, 6, -1, "Major", "Blocker", None, np.nan]: |
| self.assertIsNone(normalize_priority_value(invalid_val)) |
|
|
| def test_4_identifiers_excluded_from_model_inputs(self): |
| """4. Verifies metadata identifiers (issue_key, project_id, etc.) are excluded from model input features.""" |
| df_inputs = get_triage_model_inputs(self.df_raw) |
| for ident in METADATA_IDENTIFIERS: |
| self.assertNotIn(ident, df_inputs.columns, f"Identifier {ident} must be excluded from model inputs!") |
| for input_feat in TRIAGE_MODEL_INPUT_FEATURES: |
| self.assertIn(input_feat, df_inputs.columns) |
|
|
| def test_5_project_splits_are_completely_disjoint(self): |
| """5. Verifies project splits are 100% disjoint (train_projects ∩ val_projects ∩ test_projects = ∅).""" |
| norm_df = normalize_target_column(self.df_raw) |
| t, v, te = split_group_aware_by_project(norm_df, seed=42) |
| is_clean, info = verify_zero_project_leakage(t, v, te) |
|
|
| self.assertTrue(is_clean) |
| self.assertEqual(len(info["train_val_intersection"]), 0) |
| self.assertEqual(len(info["train_test_intersection"]), 0) |
| self.assertEqual(len(info["val_test_intersection"]), 0) |
|
|
| def test_6_research_subset_comes_only_from_priority_score_1_to_5(self): |
| """6. Verifies subset population originates exclusively from priority_score in {1..5}.""" |
| norm_df = normalize_target_column(self.df_raw) |
| subset_df = select_reproducible_subset(norm_df, target_issue_count=50, seed=42) |
| self.assertTrue((subset_df["priority_raw"].isin([1, 2, 3, 4, 5])).all()) |
|
|
| def test_7_dataset_manifest_records_authoritative_target(self): |
| """7. Verifies dataset manifest declares priority_score as authoritative target.""" |
| manifest = generate_dataset_manifest( |
| dataset_name="TEST FIXTURE", |
| is_real_dataset=False, |
| total_rows=100, |
| usable_rows=100, |
| missing_target_rows=0, |
| num_projects=3, |
| selected_issue_count=100, |
| train_projects_count=1, |
| val_projects_count=1, |
| test_projects_count=1, |
| train_issue_count=34, |
| val_issue_count=33, |
| test_issue_count=33, |
| zero_leakage_verified=True, |
| ) |
| self.assertEqual(manifest["authoritative_target_column"], "priority_score") |
| self.assertEqual(manifest["feature_summary"]["candidate_model_input_features_count"], 20) |
| self.assertEqual(manifest["feature_summary"]["metadata_identifiers_count"], 5) |
|
|
| def test_8_schema_and_llm_input_isolation(self): |
| """8. Verifies LLM prompt input features strictly exclude targets and post-triage metrics.""" |
| df_llm = get_llm_input_features(self.df_raw) |
| self.assertNotIn("priority_score", df_llm.columns) |
| self.assertNotIn("priority_name", df_llm.columns) |
| self.assertNotIn("Cycle Time (hours)", df_llm.columns) |
| self.assertIn("issue_summary", df_llm.columns) |
|
|
| def test_9_end_to_end_pipeline_runner(self): |
| """9. Verifies full pipeline execution produces valid splits and manifests.""" |
| train_df, val_df, test_df, summary = run_preprocessing_pipeline() |
| self.assertTrue(summary["zero_leakage_verified"]) |
| self.assertTrue(MANIFEST_PATH.exists()) |
| self.assertTrue(REPORT_PATH.exists()) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
