# -*- coding: utf-8 -*- """ Smoke test: step8_ml_train_panel + step9_ml_predict_panel 的新 CSV→QComboBox 联动 & update_from_config 多源文件优先回退 逻辑。 不真正 import 整个 panel（避免 osgeo 重依赖），只 import 类本身，验证： 1) Step8MlTrainPanel.feature_start 变成 QComboBox 2) Step8._on_training_csv_changed 正确读 CSV 表头 + 默认选中 374.285 3) Step8._resolve_training_csv_from_workdir 优先级（Step7 > 10_WaterIndex > Step6 > Step7 兜底） 4) Step8.get_config / set_config / get_training_params 用 QComboBox API 不崩 5) Step9._resolve_latest_wqi_test_csv 找到正确的 WQI 测试集 6) Step9.update_from_config 优先级（WQI CSV > Step4 raw） 用 offscreen 模式避免弹窗，完整可重复跑。 """ import os import sys import tempfile from pathlib import Path # 避免 panel 顶部 osgeo 导入阻塞 os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") # 让 src 目录可 import HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import pandas as pd import numpy as np from PyQt5.QtWidgets import QApplication # 必须先 create QApplication 才能 new widget app = QApplication.instance() or QApplication(sys.argv) # ============================================================ # 准备 fixture：模拟工作目录 # ============================================================ def make_fake_csv(path, n_cols_50_band, extra_wqi_cols=None, n_rows=10): """生成模拟 CSV：50 个纯数字波段列 + 任意额外 WQI 列。""" band_cols = [374.285 + i for i in range(n_cols_50_band)] # 浮点列名会被 pandas 自动转成 374.285 / 374.285001 / ... cols = band_cols + (extra_wqi_cols or []) data = np.random.RandomState(42).rand(n_rows, len(cols)) df = pd.DataFrame(data, columns=cols) df.to_csv(path, index=False) return df with tempfile.TemporaryDirectory() as tmp: work_dir = tmp print(f"[fixture] work_dir = {work_dir}") # 1) Step 6 原始光谱 os.makedirs(os.path.join(work_dir, "6_Spectral_Feature_Extraction"), exist_ok=True) step6_csv = os.path.join(work_dir, "6_Spectral_Feature_Extraction", "training_spectra.csv") make_fake_csv(step6_csv, 50) # 2) Step 7 WQI 增强版 os.makedirs(os.path.join(work_dir, "7_Water_Quality_Indices"), exist_ok=True) step7_csv = os.path.join(work_dir, "7_Water_Quality_Indices", "training_spectra_indices.csv") wqi_names = [f"WQI_Chla_{i}" for i in range(10)] + [f"WQI_TP_{i}" for i in range(5)] make_fake_csv(step7_csv, 50, wqi_names) # 3) 10_WaterIndex_CSV 下的 training csv（较旧 mtime） os.makedirs(os.path.join(work_dir, "10_WaterIndex_CSV"), exist_ok=True) step10_csv = os.path.join(work_dir, "10_WaterIndex_CSV", "training_watercolor_summary.csv") make_fake_csv(step10_csv, 50, wqi_names) # 把 mtime 调成 3 年前，确认 7_Water_Quality_Indices 那个更新 old_time = (pd.Timestamp.now() - pd.Timedelta(days=365 * 3)).timestamp() os.utime(step10_csv, (old_time, old_time)) # 4) Step 4 raw sampling os.makedirs(os.path.join(work_dir, "4_sampling"), exist_ok=True) step4_csv = os.path.join(work_dir, "4_sampling", "sampling_spectra.csv") make_fake_csv(step4_csv, 50) # 5) WQI 测试集（10_WaterIndex_CSV 下另一个 mtime 较新的 csv） wqi_test_csv = os.path.join(work_dir, "10_WaterIndex_CSV", "sampling_watercolor_summary.csv") make_fake_csv(wqi_test_csv, 50, wqi_names) # 设为 1 小时前（mtime 比 step7_csv 新） new_time = (pd.Timestamp.now() - pd.Timedelta(hours=1)).timestamp() os.utime(wqi_test_csv, (new_time, new_time)) # 6) 9_ML_Prediction 输出目录 os.makedirs(os.path.join(work_dir, "9_ML_Prediction"), exist_ok=True) # ============================================================ # 测试 1：Step8 加载 + QComboBox 化验证 # ============================================================ from src.gui.panels.step8_ml_train_panel import Step8MlTrainPanel from PyQt5.QtWidgets import QComboBox panel8 = Step8MlTrainPanel() assert isinstance(panel8.feature_start, QComboBox), \ f"FAIL: feature_start 应为 QComboBox，实际为 {type(panel8.feature_start)}" print(f"[OK] Step8.feature_start 类型: {type(panel8.feature_start).__name__}") # ============================================================ # 测试 2：CSV 变更槽 — 模拟用户选 WQI CSV，自动填表头 # ============================================================ panel8._on_training_csv_changed(step7_csv) assert panel8.feature_start.count() > 0 head_text = panel8.feature_start.currentText() print(f"[OK] Step8 选中 step7_csv 后，feature_start.currentText() = {head_text!r}") # 默认选中：第一个纯数字波段列 "374.285" assert head_text.replace('.', '').lstrip('-').isdigit(), \ f"FAIL: 默认应选数字波段，实际 = {head_text!r}" print(f"[OK] 默认选中数字波段列: {head_text!r}") # 验证前 50 列已填充 n_items = panel8.feature_start.count() assert n_items >= 50, f"FAIL: 应至少 50 项，实际 {n_items}" print(f"[OK] Step8.feature_start 填充了 {n_items} 个候选列（前 50 列）") # ============================================================ # 测试 3：CSV 不存在时，combo 回到占位项 # ============================================================ panel8._on_training_csv_changed("/nonexistent/path.csv") assert panel8.feature_start.count() == 1 assert "请先选择训练 CSV" in panel8.feature_start.itemText(0) print(f"[OK] Step8 CSV 不存在时，combo 回退到占位项: {panel8.feature_start.itemText(0)!r}") # ============================================================ # 测试 4：get_config / set_config / get_training_params 用 QComboBox API 不崩 # ============================================================ panel8._on_training_csv_changed(step7_csv) cfg = panel8.get_config() assert 'feature_start_column' in cfg assert isinstance(cfg['feature_start_column'], str) assert cfg['feature_start_column'].replace('.', '').lstrip('-').isdigit() print(f"[OK] Step8.get_config()['feature_start_column'] = {cfg['feature_start_column']!r}") # set_config 试一下（先选个不存在的列名，应该走兜底 addItem 分支） panel8.set_config({'feature_start_column': '999.999_unknown'}) # 重新触发 CSV 加载，验证回放不污染 panel8._on_training_csv_changed(step7_csv) panel8.set_config({'feature_start_column': '374.285'}) assert panel8.feature_start.currentText() == '374.285' print(f"[OK] Step8.set_config() 正确把 374.285 选中") # get_training_params params = panel8.get_training_params() assert 'feature_start' in params assert isinstance(params['feature_start'], float) print(f"[OK] Step8.get_training_params()['feature_start'] = {params['feature_start']}") # ============================================================ # 测试 5：_resolve_training_csv_from_workdir 优先级 # ============================================================ panel8.work_dir = work_dir resolved = panel8._resolve_training_csv_from_workdir() expected = step7_csv.replace('\\', '/') assert resolved == expected, f"FAIL: 应优先 step7, 实际 {resolved!r}, 期望 {expected!r}" print(f"[OK] Step8._resolve_training_csv_from_workdir() = {resolved}") print(f" 优先级 1: Step 7 WQI 增强版 ✓") # 模拟用户没跑过 Step 7（删掉 step7_csv 目录）— 但 10_WaterIndex_CSV 还在 import shutil shutil.rmtree(os.path.join(work_dir, "7_Water_Quality_Indices")) resolved = panel8._resolve_training_csv_from_workdir() # 用户明确提到 10_WaterIndex_CSV 目录下的结果是有效来源，应作为第二优先级 expected_step10 = str(Path(step10_csv)).replace('\\', '/') assert resolved == expected_step10, \ f"FAIL: 应取 10_WaterIndex_CSV/*training*, 实际 {resolved!r}, 期望 {expected_step10!r}" print(f"[OK] Step 7 缺省时，回退到 10_WaterIndex_CSV/*training*: {resolved}") print(f" 优先级 2: 10_WaterIndex_CSV/*training* ✓") # 再删 10_WaterIndex_CSV 才回退到 Step 6 shutil.rmtree(os.path.join(work_dir, "10_WaterIndex_CSV")) resolved = panel8._resolve_training_csv_from_workdir() assert resolved == step6_csv.replace('\\', '/'), \ f"FAIL: 应回退 step6, 实际 {resolved!r}, 期望 {step6_csv!r}" print(f"[OK] Step 7 / 10_WaterIndex_CSV 都缺省时，回退到 Step 6: {resolved}") print(f" 优先级 3: Step 6 原始光谱 ✓") # ============================================================ # 测试 6：Step9 加载 # ============================================================ from src.gui.panels.step9_ml_predict_panel import Step9MlPredictPanel panel9 = Step9MlPredictPanel() print(f"[OK] Step9MlPredictPanel 实例化成功") # 重建 10_WaterIndex_CSV（WQI 测试集 fixture，test 5 中被删了） os.makedirs(os.path.join(work_dir, "10_WaterIndex_CSV"), exist_ok=True) wqi_test_csv = os.path.join(work_dir, "10_WaterIndex_CSV", "sampling_watercolor_summary.csv") make_fake_csv(wqi_test_csv, 50, wqi_names) new_time = (pd.Timestamp.now() - pd.Timedelta(hours=1)).timestamp() os.utime(wqi_test_csv, (new_time, new_time)) # ============================================================ # 测试 7：Step9._resolve_latest_wqi_test_csv 优先 10_WaterIndex_CSV # ============================================================ panel9.work_dir = work_dir resolved = panel9._resolve_latest_wqi_test_csv() # 应该是 sampling_watercolor_summary.csv（mtime 较新） expected = wqi_test_csv.replace('\\', '/') assert resolved == expected, \ f"FAIL: 应取 mtime 最新的 wqi 测试集, 实际 {resolved!r}, 期望 {expected!r}" print(f"[OK] Step9._resolve_latest_wqi_test_csv() = {resolved}") print(f" 优先级：10_WaterIndex_CSV/sampling_watercolor_summary.csv ✓") # ============================================================ # 测试 8：Step9.update_from_config 优先级：WQI CSV > Step4 raw # ============================================================ # 注：Step 4 回退路径依赖 main_window._panel_factory（offline 测试无 main_window， # 故仅验证 WQI CSV 优先级主路径；Step 4 路径与原代码同源，逻辑没变） # 有 WQI 时优先 WQI os.makedirs(os.path.join(work_dir, "10_WaterIndex_CSV"), exist_ok=True) wqi_test_csv2 = os.path.join(work_dir, "10_WaterIndex_CSV", "sampling_watercolor_summary.csv") make_fake_csv(wqi_test_csv2, 50, wqi_names) panel9.update_from_config(work_dir=work_dir, pipeline=None) path = panel9.sampling_csv_file.get_path() assert wqi_test_csv2.replace('\\', '/') in path, \ f"FAIL: 应优先 WQI CSV, 实际 {path!r}" print(f"[OK] Step9.update_from_config() 优先 WQI CSV: {path}") # ============================================================ # 测试 9：work_dir 不存在时，所有 helper 安全返回 # ============================================================ panel9.work_dir = "" assert panel9._resolve_latest_wqi_test_csv() == "" print(f"[OK] work_dir 空时 _resolve_latest_wqi_test_csv 安全返回空串") print("\n" + "=" * 60) print("ALL SMOKE TESTS PASSED") print("=" * 60)