# -*- coding: utf-8 -*- """ Smoke test for: 彻底修复底层写入路径与掩膜联动 验证三件事（不依赖 GUI / 不实例化 Pipeline，避免触发 osgeo/gdal 导入）： 1. Step12KrigingHandler.execute 内部对 output_image_path 的 override 逻辑正确： - 路径不在 visualization_dir 下 → 被强制重定向 - 路径在 visualization_dir 下 → 保留 - 路径为 None/空 → 用 forced 默认值 2. step11_map_panel.update_from_config 含 pipeline.get_step_output_dir('step1') 调用 3. _step_path_resolver._FALLBACK_DIR_TABLE['water_mask'] == '1_water_mask' 注：原 water_quality_inversion_pipeline_GUI.py 已删除， step10_map() 逻辑已迁移至 Step12KrigingHandler（src/core/handlers/step12_kriging.py）。 """ import re import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] HANDLER_FILE = ROOT / "src" / "core" / "handlers" / "step12_kriging.py" PANEL_FILE = ROOT / "src" / "gui" / "panels" / "step11_map_panel.py" RESOLVER_FILE = ROOT / "src" / "gui" / "panels" / "_step_path_resolver.py" MAP_FILE = ROOT / "src" / "postprocessing" / "map.py" def test_step10_map_forced_override(): """纯文本级检查 Step12KrigingHandler.execute 是否含强制重定向逻辑。""" text = HANDLER_FILE.read_text(encoding="utf-8") # 找 def execute( 起点；用下一个 def (4 空格缩进) 作锚点截取函数体 m = re.search( r"def execute\(self, context[^\)]*\)[^\n]*:\n(.*?)(?=\n def |\nclass |\Z)", text, re.DOTALL, ) assert m, "找不到 execute 方法" body = m.group(1) # 关键标记 assert "forced_image_path" in body, "execute 应计算 forced_image_path" assert "context.visualization_dir" in body, "execute 应引用 context.visualization_dir" print("✅ Step12KrigingHandler.execute 含强制 override 逻辑（forced_image_path）") def test_step10_map_accepts_in_visualization_dir(): """模拟：当用户传入的路径已在 visualization_dir 内，应被保留（不被覆盖）。""" # 走 source 的逻辑分支：在 startswith 命中 → 保留原路径 norm_viz = "/work/14_visualization" candidates = [ ("/work/14_visualization/sub/foo.png", True), # 子目录 → 保留 ("/work/14_visualization/foo.png", True), # 自身 → 保留（== 情形走 else 分支） ("/work/11_12_13_predictions/foo.png", False), # 外部 → 强制重定向 ("/work/foo.png", False), ] for path, should_keep in candidates: norm_user = path.replace("\\", "/").rstrip("/") keep = norm_user.startswith(norm_viz + "/") or norm_user == norm_viz assert keep == should_keep, f"路径 {path!r} 判断错误：keep={keep}, expect={should_keep}" print("✅ step10_map 路径归属判断（startswith + ==）正确") def test_step11_panel_calls_pipeline_get_step_output_dir(): """验证 step11 panel 真的调用了 main_window.pipeline.get_step_output_dir('step1')。""" text = PANEL_FILE.read_text(encoding="utf-8") assert "get_step_output_dir('step1')" in text, \ "step11 panel 应调用 get_step_output_dir('step1')" assert "getattr(_win, 'pipeline', None)" in text or "getattr(main_window, 'pipeline'" in text, \ "step11 panel 应安全访问 main_window.pipeline（None 守护）" assert "resolve_subdir(self.work_dir, 'water_mask')" in text, \ "step11 panel 应有 resolve_subdir 兜底" # 不应硬编码具体掩膜文件名（应通过 glob 探测）—— 排除注释行 code_lines = [ ln for ln in text.splitlines() if ln.strip() and not ln.strip().startswith("#") ] code_only = "\n".join(code_lines) assert "water_mask_from_ndwi" not in code_only, \ "step11 panel 不应硬编码 water_mask_from_ndwi（应通过 glob 探测）" assert "water_mask_from_shp" not in code_only, \ "step11 panel 不应硬编码 water_mask_from_shp（应通过 glob 探测）" print("✅ step11 panel 含 pipeline.get_step_output_dir('step1') 调用 + 兜底 + 安全守护") def test_fallback_dir_table_water_mask(): """验证 _FALLBACK_DIR_TABLE['water_mask'] == '1_water_mask'。""" text = RESOLVER_FILE.read_text(encoding="utf-8") m = re.search(r"'water_mask'\s*:\s*'([^']+)'", text) assert m, "_FALLBACK_DIR_TABLE 缺 'water_mask' 键" assert m.group(1) == "1_water_mask", f"应为 1_water_mask，实为 {m.group(1)}" print(f"✅ _FALLBACK_DIR_TABLE['water_mask'] = {m.group(1)!r}") def test_panel_guard_does_not_overwrite_existing(): """验证 step11 panel 的新段不覆盖已有 boundary（feedback_never_overwrite_with_empty）。""" text = PANEL_FILE.read_text(encoding="utf-8") # 4.5 段必须先读 existing_boundary，再判断 m = re.search( r"# 4\.5\..*?(?=\n # 5\.)", text, re.DOTALL ) assert m, "找不到 4.5 段" body = m.group(0) assert "existing_boundary" in body, "4.5 段应读 existing_boundary" assert "if not existing_boundary" in body, "4.5 段应仅在空时填入" print("✅ step11 panel 4.5 段遵守 '非空值不覆盖' 原则") def test_map_supports_raster_mask_formats(): """验证 ContentMapper.read_boundary_shapefile 内部已支持栅格格式分发（.dat/.bsq/.tif/.tiff/.img）。 之前 bug：4.5 段成功把 .dat 填入 boundary_file，但 ContentMapper 内部 gpd.read_file(.dat) 直接报错。修复后 ContentMapper 内部按后缀分发： - .shp → 矢量（保持原行为） - .dat/.bsq/.tif/.tiff/.img → rasterio.features.shapes 矢量化成 GeoDataFrame """ text = MAP_FILE.read_text(encoding="utf-8") # 找 def read_boundary_shapefile m = re.search( r"def read_boundary_shapefile\(self,[^\)]*\)[^\n]*:\n(.*?)(?=\n def |\nclass |\Z)", text, re.DOTALL, ) assert m, "找不到 read_boundary_shapefile 方法" body = m.group(1) # 关键标记：format dispatch assert ".dat" in body, "read_boundary_shapefile 应支持 .dat 栅格" assert ".bsq" in body, "read_boundary_shapefile 应支持 .bsq 栅格" assert ".tif" in body, "read_boundary_shapefile 应支持 .tif 栅格" assert "gpd.read_file(shp_file)" in body, ".shp 分支应保留 gpd.read_file 矢量读取" assert "rasterio.features.shapes" in body or "from rasterio.features import" in text, \ "栅格分支应使用 rasterio.features.shapes 矢量化" # 验证 helper 方法存在 assert "def _raster_to_boundary_gdf" in text, \ "应新增 _raster_to_boundary_gdf helper 方法" # 验证 helper 内有栅格读取 + 矢量化 + GeoDataFrame 返回 helper_m = re.search( r"def _raster_to_boundary_gdf\(self,[^\)]*\)[^\n]*:\n(.*?)(?=\n def |\nclass |\Z)", text, re.DOTALL, ) assert helper_m, "找不到 _raster_to_boundary_gdf 方法" helper = helper_m.group(1) assert "rasterio.open" in helper, "helper 应调用 rasterio.open 读栅格" assert "shapes(" in helper, "helper 应调用 rasterio.features.shapes 矢量化" assert "gpd.GeoDataFrame" in helper, "helper 应返回 gpd.GeoDataFrame" print("✅ ContentMapper.read_boundary_shapefile 支持 .shp/.dat/.bsq/.tif 多格式分发") if __name__ == "__main__": print("=" * 60) print("Smoke test: 彻底修复底层写入路径与掩膜联动") print("=" * 60) test_step10_map_forced_override() test_step10_map_accepts_in_visualization_dir() test_step11_panel_calls_pipeline_get_step_output_dir() test_fallback_dir_table_water_mask() test_panel_guard_does_not_overwrite_existing() test_map_supports_raster_mask_formats() print("=" * 60) print("全部通过 ✅") sys.exit(0)