160 lines
7.6 KiB
Python
160 lines
7.6 KiB
Python
# -*- 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)
|