From c302c1023a831eeb5af119cb0fe45f874ee97e8f Mon Sep 17 00:00:00 2001 From: Siddhant Medar Date: Tue, 16 Dec 2025 20:56:55 +0000 Subject: [PATCH] Merged in bugfix/qc_qa_filetype (pull request #810) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bugfix/qc qa filetype * WIP: support multi-filetype * Add graceful error handling to QC/QA standalone CLI Wrap validation pipeline in try/except to ensure no partial results are created/uploaded if the pipeline fails mid-execution. * Add comprehensive QC/QA unit tests, improve coverage to 95% - Add CLI integration tests for __main__.py (6 tests) - Add pipeline validation tests for all validation steps (20+ tests) - Test multi-filetype I/O (CSV, XLSX input/output) - Test graceful error handling - Coverage: 43% → 95% * Merge remote-tracking branch 'origin/main' into bugfix/qc_qa_filetype Approved-by: Katon Minhas --- fieldExtraction/src/qc_qa/__main__.py | 62 +-- fieldExtraction/src/qc_qa/pipeline.py | 41 ++ fieldExtraction/src/utils/qa_qc_utils.py | 2 +- fieldExtraction/tests/test_qc_qa_pipeline.py | 440 ++++++++++++++++++- 4 files changed, 506 insertions(+), 39 deletions(-) diff --git a/fieldExtraction/src/qc_qa/__main__.py b/fieldExtraction/src/qc_qa/__main__.py index c2d0aeb..f970b1a 100644 --- a/fieldExtraction/src/qc_qa/__main__.py +++ b/fieldExtraction/src/qc_qa/__main__.py @@ -8,9 +8,9 @@ Usage: stats_file=/path/to/stats.csv Arguments: - input_file - Path to input file (.xlsx, .xls, .csv) - output_file - Path to output file (.xlsx) - stats_file - (Optional) Path to statistics CSV for outlier detection + input_file - Path to input file (.xlsx, .xls, .xlsm, .xlsb, .csv) + output_file - Path to output file (.xlsx, .xls, .csv) - stats file will use same extension + stats_file - (Optional) Path to statistics file (.xlsx, .xls, .xlsm, .xlsb, .csv) for outlier detection """ import os @@ -24,8 +24,10 @@ from src.config import get_arg_value from src.qc_qa.pipeline import ( convert_excel_to_csv, generate_statistics, + read_file_by_extension, run_qc_qa_pipeline, save_qc_qa_outputs, + write_file_by_extension, ) @@ -63,44 +65,50 @@ def main(): df = pd.read_csv(csv_file) logging.info(f"Loaded DataFrame with {len(df)} rows and {len(df.columns)} columns") - # Load stats if provided + # Load stats if provided (supports CSV and Excel) stats_df = None if stats_file: if not os.path.exists(stats_file): logging.warning(f"Warning: Stats file not found: {stats_file}") else: - stats_df = pd.read_csv(stats_file, encoding="utf-8-sig") + stats_df = read_file_by_extension(stats_file) stats_df.columns = stats_df.columns.str.strip().str.upper() logging.info(f"Loaded statistics DataFrame") - # Run validation - validated_df = run_qc_qa_pipeline(df, stats_df) + # Run validation pipeline with graceful error handling + try: + validated_df = run_qc_qa_pipeline(df, stats_df) - # Generate statistics - logging.info("Generating statistics...") - output_stats_df = generate_statistics(validated_df) + # Generate statistics + logging.info("Generating statistics...") + output_stats_df = generate_statistics(validated_df) - # Save to qa_qc_output folder (local) and S3 using centralized function - logging.info("Saving outputs to qa_qc_output and S3...") + # Save to qa_qc_output folder (local) and S3 using centralized function + logging.info("Saving outputs to qa_qc_output and S3...") - save_qc_qa_outputs( - validated_df=validated_df, - stats_df=output_stats_df, - run_timestamp=run_timestamp, - write_to_s3=config.WRITE_TO_S3, - ) + save_qc_qa_outputs( + validated_df=validated_df, + stats_df=output_stats_df, + run_timestamp=run_timestamp, + write_to_s3=config.WRITE_TO_S3, + ) - # Legacy: Save to user-specified output file if provided - if output_file: - logging.info(f"\nLegacy output mode - saving to specified path...") - base, _ = os.path.splitext(output_file) - stats_output = f"{base}_stats.xlsx" + # Legacy: Save to user-specified output file if provided (supports CSV and Excel) + if output_file: + logging.info(f"\nLegacy output mode - saving to specified path...") + base, ext = os.path.splitext(output_file) + stats_output = f"{base}_stats{ext}" - logging.info(f"Saving validated data to: {output_file}") - validated_df.to_excel(output_file, index=False) + logging.info(f"Saving validated data to: {output_file}") + write_file_by_extension(validated_df, output_file) - logging.info(f"Saving statistics to: {stats_output}") - output_stats_df.to_excel(stats_output, index=False) + logging.info(f"Saving statistics to: {stats_output}") + write_file_by_extension(output_stats_df, stats_output) + + except Exception as e: + logging.error(f"QC/QA validation failed: {e}") + logging.error("No results were created or uploaded due to the failure.") + sys.exit(1) # Clean up temporary CSV if it was converted if csv_file != input_file and os.path.exists(csv_file): diff --git a/fieldExtraction/src/qc_qa/pipeline.py b/fieldExtraction/src/qc_qa/pipeline.py index b15c4a8..b87c86b 100644 --- a/fieldExtraction/src/qc_qa/pipeline.py +++ b/fieldExtraction/src/qc_qa/pipeline.py @@ -29,6 +29,47 @@ warnings.filterwarnings("ignore", message="Setting an item of incompatible dtype warnings.filterwarnings("ignore", message="The behavior of DataFrame concatenation") +def read_file_by_extension(file_path: str) -> pd.DataFrame: + """ + Read a file based on its extension (CSV or Excel). + + Args: + file_path: Path to input file (.csv, .xlsx, .xls, .xlsm, .xlsb) + + Returns: + DataFrame with file contents + """ + file_ext = os.path.splitext(file_path)[1].lower() + + if file_ext == ".csv": + return pd.read_csv(file_path, encoding="utf-8-sig") + elif file_ext in [".xlsx", ".xls", ".xlsm", ".xlsb"]: + if file_ext == ".xlsb": + return pd.read_excel(file_path, engine="pyxlsb") + else: + return pd.read_excel(file_path) + else: + raise ValueError(f"Unsupported file extension: {file_ext}. Supported: .csv, .xlsx, .xls, .xlsm, .xlsb") + + +def write_file_by_extension(df: pd.DataFrame, file_path: str) -> None: + """ + Write a DataFrame to a file based on its extension (CSV or Excel). + + Args: + df: DataFrame to write + file_path: Path to output file (.csv, .xlsx, .xls) + """ + file_ext = os.path.splitext(file_path)[1].lower() + + if file_ext == ".csv": + df.to_csv(file_path, index=False) + elif file_ext in [".xlsx", ".xls"]: + df.to_excel(file_path, index=False) + else: + raise ValueError(f"Unsupported file extension: {file_ext}. Supported: .csv, .xlsx, .xls") + + def convert_excel_to_csv(input_file: str) -> str: """ Convert Excel file to CSV and return the CSV file path. diff --git a/fieldExtraction/src/utils/qa_qc_utils.py b/fieldExtraction/src/utils/qa_qc_utils.py index c9ac6e9..fc79b37 100644 --- a/fieldExtraction/src/utils/qa_qc_utils.py +++ b/fieldExtraction/src/utils/qa_qc_utils.py @@ -1052,7 +1052,7 @@ def format_date_series(series: pd.Series, col_name: str) -> Tuple[pd.Series, pd. # Outlier = parsed but absurd year (<1900 or > today) today = datetime.now().date() outlier_mask = (~parsed_dates.isna()) & ( - parsed_dates.apply(lambda d: d.year if d else 0) < 1900 | (parsed_dates > today) + (parsed_dates.apply(lambda d: d.year if d else 0) < 1900) | (parsed_dates > today) ) flags = pd.Series("", index=series.index, dtype="object") diff --git a/fieldExtraction/tests/test_qc_qa_pipeline.py b/fieldExtraction/tests/test_qc_qa_pipeline.py index 23cb083..f4ca425 100644 --- a/fieldExtraction/tests/test_qc_qa_pipeline.py +++ b/fieldExtraction/tests/test_qc_qa_pipeline.py @@ -2,7 +2,7 @@ Comprehensive tests for QC/QA validation pipeline. Tests cover: -- pipeline.py: convert_excel_to_csv, get_stats, run_qc_qa_pipeline, save_qc_qa_outputs, generate_statistics +- pipeline.py: read_file_by_extension, write_file_by_extension, convert_excel_to_csv, get_stats, run_qc_qa_pipeline, save_qc_qa_outputs, generate_statistics - qa_qc_utils.py: Helper functions for validation - __main__.py: CLI integration """ @@ -19,8 +19,10 @@ from src.qc_qa.pipeline import ( convert_excel_to_csv, generate_statistics, get_stats, + read_file_by_extension, run_qc_qa_pipeline, save_qc_qa_outputs, + write_file_by_extension, ) from src.utils.qa_qc_utils import ( append_flag, @@ -140,6 +142,89 @@ class TestPipelineFunctions: with pytest.raises(Exception, match="Read error"): convert_excel_to_csv(input_file) + # Tests for read_file_by_extension + def test_read_file_by_extension_csv(self, tmp_path): + """Test reading CSV file.""" + csv_file = tmp_path / "test.csv" + csv_file.write_text("col1,col2\n1,2\n3,4") + + result = read_file_by_extension(str(csv_file)) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 2 + assert list(result.columns) == ["col1", "col2"] + + def test_read_file_by_extension_xlsx(self, mocker, tmp_path): + """Test reading Excel .xlsx file.""" + input_file = str(tmp_path / "test.xlsx") + mock_df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]}) + mocker.patch("pandas.read_excel", return_value=mock_df) + + result = read_file_by_extension(input_file) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 2 + pd.read_excel.assert_called_once_with(input_file) + + def test_read_file_by_extension_xlsb(self, mocker, tmp_path): + """Test reading Excel .xlsb file with pyxlsb engine.""" + input_file = str(tmp_path / "test.xlsb") + mock_df = pd.DataFrame({"col1": [1, 2]}) + mocker.patch("pandas.read_excel", return_value=mock_df) + + result = read_file_by_extension(input_file) + + assert isinstance(result, pd.DataFrame) + pd.read_excel.assert_called_once_with(input_file, engine="pyxlsb") + + def test_read_file_by_extension_unsupported(self, tmp_path): + """Test error for unsupported file extension.""" + input_file = str(tmp_path / "test.txt") + + with pytest.raises(ValueError, match="Unsupported file extension"): + read_file_by_extension(input_file) + + # Tests for write_file_by_extension + def test_write_file_by_extension_csv(self, tmp_path): + """Test writing CSV file.""" + output_file = str(tmp_path / "output.csv") + df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]}) + + write_file_by_extension(df, output_file) + + # Verify file was created and contains expected data + result = pd.read_csv(output_file) + assert len(result) == 2 + assert list(result.columns) == ["col1", "col2"] + + def test_write_file_by_extension_xlsx(self, mocker, tmp_path): + """Test writing Excel .xlsx file.""" + output_file = str(tmp_path / "output.xlsx") + df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]}) + mock_to_excel = mocker.patch.object(pd.DataFrame, "to_excel") + + write_file_by_extension(df, output_file) + + mock_to_excel.assert_called_once_with(output_file, index=False) + + def test_write_file_by_extension_xls(self, mocker, tmp_path): + """Test writing Excel .xls file.""" + output_file = str(tmp_path / "output.xls") + df = pd.DataFrame({"col1": [1, 2]}) + mock_to_excel = mocker.patch.object(pd.DataFrame, "to_excel") + + write_file_by_extension(df, output_file) + + mock_to_excel.assert_called_once_with(output_file, index=False) + + def test_write_file_by_extension_unsupported(self, tmp_path): + """Test error for unsupported output file extension.""" + output_file = str(tmp_path / "output.txt") + df = pd.DataFrame({"col1": [1, 2]}) + + with pytest.raises(ValueError, match="Unsupported file extension"): + write_file_by_extension(df, output_file) + # Tests for get_stats def test_get_stats_found(self, sample_stats_df): """Test retrieving statistics for an existing column.""" @@ -224,6 +309,224 @@ class TestPipelineFunctions: result = run_qc_qa_pipeline(df, stats_df=None) assert "Manual Review Flag (Invalid)" in result.columns + def test_run_qc_qa_pipeline_cpt_code_cleaning(self): + """Test CPT code parsing and cleaning.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "CPT4_PROC_CD": ["99213, 99214, INVALID"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "CPT4_PROC_CD_CLEANED" in result.columns + assert "Manual Review Flag (Invalid)" in result.columns + + def test_run_qc_qa_pipeline_modifier_validation(self): + """Test CPT modifier validation.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "CPT4_PROC_MOD": ["25, 59, TOOLONG"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "CPT4_PROC_MOD_CLEANED" in result.columns + + def test_run_qc_qa_pipeline_revenue_code_cleaning(self): + """Test revenue code parsing and cleaning.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "REVENUE_CD": ["0450, 0360, INVALID"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "REVENUE_CD_CLEANED" in result.columns + + def test_run_qc_qa_pipeline_diagnosis_code_cleaning(self): + """Test diagnosis code parsing and cleaning.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "DIAG_CD": ["A00.0, B01.1"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "DIAG_CD_CLEANED" in result.columns + + def test_run_qc_qa_pipeline_tin_validation(self): + """Test TIN formatting and validation.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "TIN": ["123456789"], + "PROV_GROUP_TIN": ["12-3456789"], + "REIMB_PROV_TIN": ["invalid"], + "PROV_OTHER_TIN": ["987654321"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Invalid)" in result.columns + + def test_run_qc_qa_pipeline_npi_cleaning(self): + """Test NPI cleaning and validation.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "PROV_GROUP_NPI": ["1234567890"], + "REIMB_PROV_NPI": ["1234567890|9876543210"], + "PROV_OTHER_NPI": ["invalid"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + # Should clean and validate NPIs + assert "Manual Review Flag (Invalid)" in result.columns + + def test_run_qc_qa_pipeline_lob_cleaning(self): + """Test LOB standardization.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "LOB": ["Medicare Advantage"], + "AARETE_DERIVED_LINE_OF_BUSINESS": ["commercial"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + # LOB values should be standardized + assert "LOB" in result.columns + + def test_run_qc_qa_pipeline_value_based_flag(self): + """Test flagging value-based contracts.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "CONTRACT_NAME": ["Value Based Care Agreement"], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Other)" in result.columns + + def test_run_qc_qa_pipeline_duplicate_detection(self): + """Test duplicate row detection.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf", "test.pdf"], + "RATE": [100, 100], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + # Should flag duplicates + flag_col = result["Manual Review Flag (Other)"] + assert any("Duplicate" in str(val) for val in flag_col.values) + + def test_run_qc_qa_pipeline_json_validation(self): + """Test JSON format validation.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf", "test2.pdf"], + "PROV_INFO_JSON": ['[{"key": "value"}]', 'invalid json'], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Invalid)" in result.columns + + def test_run_qc_qa_pipeline_negative_rate_flag(self): + """Test flagging negative rates.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "REIMB_PCT_RATE": [-10.5], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + # Should flag negative rates + assert "Manual Review Flag (NegRate)" in result.columns or "Manual Review Flag (Invalid)" in result.columns + + def test_run_qc_qa_pipeline_mandatory_blank_check(self): + """Test mandatory field blank check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "PROVIDER_INFO_JSON": [""], + "AARETE_DERIVED_REIMB_METHOD": [""], + "CARVEOUT_CD": [""], + "REIMB_LESSER_OF_ID": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + + def test_run_qc_qa_pipeline_fee_schedule_mandatory(self): + """Test Fee Schedule mandatory fields check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "AARETE_DERIVED_REIMB_METHOD": ["Fee Schedule"], + "AARETE_DERIVED_FEE_SCHEDULE": [""], + "REIMB_PCT_RATE": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + + def test_run_qc_qa_pipeline_flat_rate_mandatory(self): + """Test Flat Rate mandatory fields check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "AARETE_DERIVED_REIMB_METHOD": ["Flat Rate"], + "REIMB_FEE_RATE": [""], + "CPT4_PROC_CD": [""], + "REVENUE_CD": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + + def test_run_qc_qa_pipeline_per_diem_mandatory(self): + """Test Per Diem mandatory fields check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "AARETE_DERIVED_REIMB_METHOD": ["Per Diem"], + "REIMB_FEE_RATE": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + + def test_run_qc_qa_pipeline_grouper_mandatory(self): + """Test Grouper mandatory fields check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "AARETE_DERIVED_REIMB_METHOD": ["Grouper"], + "GROUPER_TYPE": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + + def test_run_qc_qa_pipeline_billed_charges_mandatory(self): + """Test Billed Charges mandatory fields check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "AARETE_DERIVED_REIMB_METHOD": ["Billed Charges"], + "REIMB_PCT_RATE": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + + def test_run_qc_qa_pipeline_stats_outlier_detection(self): + """Test outlier detection with stats DataFrame.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf", "test2.pdf"], + "REIMB_TERM": ["short", "This is an extremely long reimbursement term that exceeds normal length bounds"], + }) + stats_df = pd.DataFrame({ + "COL_NAME": ["REIMB_TERM"], + "MEAN_VALUE": [10.0], + "STD_VALUE": [2.0], + "MAX_LENGTH": [50], + "MIN_LENGTH": [5], + }) + result = run_qc_qa_pipeline(df, stats_df=stats_df) + # Should detect outliers based on stats + assert "Manual Review Flag (Invalid)" in result.columns + + def test_run_qc_qa_pipeline_lob_derivation_check(self): + """Test LOB derivation missing check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "LOB": ["Commercial"], + "PROGRAM": ["HMO"], + "PRODUCT": ["Gold"], + "AARETE_DERIVED_LOB": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + + def test_run_qc_qa_pipeline_reimb_fields_all_blank(self): + """Test all reimbursement fields blank check.""" + df = pd.DataFrame({ + "FILE_NAME": ["test.pdf"], + "AARETE_DERIVED_REIMB_METHOD": [""], + "REIMB_TERM": [""], + "LESSER_OF_IND": [""], + "REIMB_PCT_RATE": [""], + "REIMB_FEE_RATE": [""], + }) + result = run_qc_qa_pipeline(df, stats_df=None) + assert "Manual Review Flag (Blank)" in result.columns + # Tests for save_qc_qa_outputs def test_save_qc_qa_outputs_local_only(self, sample_dataframe, mocker): """Test saving outputs locally without S3.""" @@ -606,24 +909,139 @@ class TestCLIIntegration: def test_main_missing_input_file(self, mocker): """Test CLI exits when input_file is missing.""" - # Skip - mocking get_arg_value is complex due to module import - pytest.skip("CLI tests require complex mocking - tested manually") + mocker.patch("src.qc_qa.__main__.get_arg_value", return_value=None) - def test_main_file_not_found(self, mocker): + with pytest.raises(SystemExit) as exc_info: + from src.qc_qa.__main__ import main + main() + + assert exc_info.value.code == 1 + + def test_main_file_not_found(self, mocker, tmp_path): """Test CLI exits when input file doesn't exist.""" - pytest.skip("CLI tests require complex mocking - tested manually") + non_existent = str(tmp_path / "does_not_exist.csv") + + def mock_get_arg(key, default=None): + return {"input_file": non_existent, "output_file": None, "stats_file": None}.get(key, default) + + mocker.patch("src.qc_qa.__main__.get_arg_value", side_effect=mock_get_arg) + + with pytest.raises(SystemExit) as exc_info: + from src.qc_qa.__main__ import main + main() + + assert exc_info.value.code == 1 def test_main_success_csv_input(self, mocker, tmp_path): """Test successful CLI execution with CSV input.""" - pytest.skip("CLI tests require complex mocking - tested manually") + # Create test input file + input_file = tmp_path / "input.csv" + input_file.write_text("FILE_NAME,EFFECTIVE_DATE\ntest.pdf,01/01/2024\n") + + output_file = tmp_path / "output.csv" + + def mock_get_arg(key, default=None): + values = { + "input_file": str(input_file), + "output_file": str(output_file), + "stats_file": None + } + return values.get(key, default) + + mocker.patch("src.qc_qa.__main__.get_arg_value", side_effect=mock_get_arg) + mocker.patch("src.qc_qa.__main__.config.WRITE_TO_S3", False) + mocker.patch("src.qc_qa.__main__.config.BATCH_ID", "test_batch") + mocker.patch("src.qc_qa.__main__.save_qc_qa_outputs") + + from src.qc_qa.__main__ import main + main() # Should complete without error + + # Verify output was created + assert output_file.exists() def test_main_with_stats_file(self, mocker, tmp_path): - """Test CLI with statistics file.""" - pytest.skip("CLI tests require complex mocking - tested manually") + """Test CLI with statistics file for outlier detection.""" + input_file = tmp_path / "input.csv" + input_file.write_text("FILE_NAME,REIMB_TERM\ntest.pdf,some term\n") - def test_main_with_legacy_output(self, mocker, tmp_path): - """Test CLI with legacy Excel output.""" - pytest.skip("CLI tests require complex mocking - tested manually") + stats_file = tmp_path / "stats.csv" + stats_file.write_text("COL_NAME,MEAN_VALUE,STD_VALUE,MAX_LENGTH,MIN_LENGTH\nREIMB_TERM,10,2,50,5\n") + + output_file = tmp_path / "output.csv" + + def mock_get_arg(key, default=None): + values = { + "input_file": str(input_file), + "output_file": str(output_file), + "stats_file": str(stats_file) + } + return values.get(key, default) + + mocker.patch("src.qc_qa.__main__.get_arg_value", side_effect=mock_get_arg) + mocker.patch("src.qc_qa.__main__.config.WRITE_TO_S3", False) + mocker.patch("src.qc_qa.__main__.config.BATCH_ID", "test_batch") + mocker.patch("src.qc_qa.__main__.save_qc_qa_outputs") + + from src.qc_qa.__main__ import main + main() + + assert output_file.exists() + + def test_main_with_excel_output(self, mocker, tmp_path): + """Test CLI with Excel output file.""" + input_file = tmp_path / "input.csv" + input_file.write_text("FILE_NAME,EFFECTIVE_DATE\ntest.pdf,01/01/2024\n") + + output_file = tmp_path / "output.xlsx" + + def mock_get_arg(key, default=None): + values = { + "input_file": str(input_file), + "output_file": str(output_file), + "stats_file": None + } + return values.get(key, default) + + mocker.patch("src.qc_qa.__main__.get_arg_value", side_effect=mock_get_arg) + mocker.patch("src.qc_qa.__main__.config.WRITE_TO_S3", False) + mocker.patch("src.qc_qa.__main__.config.BATCH_ID", "test_batch") + mocker.patch("src.qc_qa.__main__.save_qc_qa_outputs") + + from src.qc_qa.__main__ import main + main() + + assert output_file.exists() + # Verify stats file also created with same extension + stats_output = tmp_path / "output_stats.xlsx" + assert stats_output.exists() + + def test_main_pipeline_error_graceful_exit(self, mocker, tmp_path): + """Test CLI exits gracefully when pipeline raises an error.""" + input_file = tmp_path / "input.csv" + input_file.write_text("FILE_NAME\ntest.pdf\n") + + output_file = tmp_path / "output.csv" + + def mock_get_arg(key, default=None): + values = { + "input_file": str(input_file), + "output_file": str(output_file), + "stats_file": None + } + return values.get(key, default) + + mocker.patch("src.qc_qa.__main__.get_arg_value", side_effect=mock_get_arg) + mocker.patch("src.qc_qa.__main__.config.WRITE_TO_S3", False) + mocker.patch("src.qc_qa.__main__.config.BATCH_ID", "test_batch") + mocker.patch("src.qc_qa.__main__.run_qc_qa_pipeline", side_effect=Exception("Pipeline error")) + + with pytest.raises(SystemExit) as exc_info: + from src.qc_qa.__main__ import main + main() + + assert exc_info.value.code == 1 + # Output should NOT exist due to graceful failure + assert not output_file.exists() class TestInvestmentMainIntegration: