From 2e68b1338ea04125ce4360e6d2ddf03f54df86ac Mon Sep 17 00:00:00 2001
From: 罗广辉 <guanghui.luo@foxmail.com>
Date: Wed, 19 Aug 2026 14:11:30 +0800
Subject: [PATCH] feat: init异常检测
---
tests/test_serve_workbench_console.py | 71 +++++++++++++++++++++++++++++++++++
1 files changed, 71 insertions(+), 0 deletions(-)
diff --git a/tests/test_serve_workbench_console.py b/tests/test_serve_workbench_console.py
index a487e87..bb14f49 100644
--- a/tests/test_serve_workbench_console.py
+++ b/tests/test_serve_workbench_console.py
@@ -5,6 +5,8 @@
import sys
import tempfile
import unittest
+import json
+from urllib.parse import quote
from pathlib import Path
@@ -44,9 +46,18 @@
Path(handler.translate_path("/shared/data/raw/00-change-detection/sample.jpg")),
ROOT / "shared" / "data" / "raw" / "00-change-detection" / "sample.jpg",
)
+ self.assertEqual(
+ Path(handler.translate_path("/shared/data/raw/09-anomaly-detection/sample.tif")),
+ ROOT / "shared" / "data" / "raw" / "09-anomaly-detection" / "sample.tif",
+ )
+ self.assertEqual(
+ Path(handler.translate_path("/shared/data/processed/09-anomaly-detection/sample.tif")),
+ ROOT / ".console-forbidden",
+ )
def test_upload_name_is_sanitized_and_extension_is_allowlisted(self) -> None:
self.assertEqual(MODULE.safe_file_name("../../unsafe name.JPG", {".jpg"}), "unsafe_name.jpg")
+ self.assertEqual(MODULE.safe_file_name("../正常参考图.JPG", {".jpg"}), "正常参考图.jpg")
with self.assertRaises(MODULE.ApiError):
MODULE.safe_file_name("image.exe", {".jpg", ".png"})
@@ -73,6 +84,66 @@
self.assertEqual(staged.read_bytes(), b"raw-tif-bytes")
self.assertEqual(result["name"], "1.tif")
+ def test_binary_anomaly_upload_preserves_bytes_and_checksum(self) -> None:
+ with tempfile.TemporaryDirectory() as temp_dir:
+ handler = object.__new__(MODULE.WorkbenchConsoleHandler)
+ handler.directory = temp_dir
+ handler.path = "/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef?role=reference"
+ handler.headers = {"Content-Length": "15", "X-Upload-Name": "../normal.JPG"}
+ handler.rfile = io.BytesIO(b"reference-bytes")
+ result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef")
+ staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "reference.jpg"
+ self.assertEqual(staged.read_bytes(), b"reference-bytes")
+ self.assertEqual(result["sha256"], MODULE.file_sha256(staged))
+
+ def test_binary_anomaly_upload_decodes_chinese_file_name(self) -> None:
+ with tempfile.TemporaryDirectory() as temp_dir:
+ handler = object.__new__(MODULE.WorkbenchConsoleHandler)
+ handler.directory = temp_dir
+ handler.path = "/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef?role=input"
+ handler.headers = {"Content-Length": "11", "X-Upload-Name": quote("异常测试图.JPG", safe="")}
+ handler.rfile = io.BytesIO(b"image-bytes")
+ result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef")
+ staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "input.jpg"
+ self.assertEqual(result["name"], "异常测试图.jpg")
+ self.assertEqual(staged.read_bytes(), b"image-bytes")
+
+ def test_interrupted_anomaly_upload_removes_partial_file(self) -> None:
+ with tempfile.TemporaryDirectory() as temp_dir:
+ handler = object.__new__(MODULE.WorkbenchConsoleHandler)
+ handler.directory = temp_dir
+ handler.path = "/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef?role=input"
+ handler.headers = {"Content-Length": "20", "X-Upload-Name": "target.tif"}
+ handler.rfile = io.BytesIO(b"short")
+ with self.assertRaisesRegex(MODULE.ApiError, "ended before"):
+ handler.receive_anomaly_upload("/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef")
+ staging = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / "2123456789abcdef0123456789abcdef"
+ self.assertFalse(any(staging.glob("*.part")))
+
+ def test_anomaly_parameters_are_bounded(self) -> None:
+ self.assertEqual(MODULE.validate_anomaly_parameters({}), (256, 128, 0.995, 42))
+ with self.assertRaisesRegex(MODULE.ApiError, "Stride"):
+ MODULE.validate_anomaly_parameters({"tileSize": 128, "stride": 256})
+ with self.assertRaisesRegex(MODULE.ApiError, "quantile"):
+ MODULE.validate_anomaly_parameters({"thresholdQuantile": 1.0})
+
+ def test_anomaly_run_discovery_requires_exposed_raw_sources(self) -> None:
+ with tempfile.TemporaryDirectory() as temp_dir:
+ root = Path(temp_dir)
+ artifact = root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / "anomaly-test"
+ raw_input = root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / "anomaly-test" / "input"
+ raw_reference = raw_input.parent / "reference"
+ artifact.mkdir(parents=True)
+ raw_input.mkdir(parents=True)
+ raw_reference.mkdir(parents=True)
+ (artifact / "target.comparison.overlay.png").write_bytes(b"png")
+ metadata = {"capability": "09-anomaly-detection", "created_at": "2026-08-18", "display_name": "五参考图边界案例", "case_note": "保留真实漏检结果。", "raw_input_dir": raw_input.relative_to(root).as_posix(), "raw_reference_dir": raw_reference.relative_to(root).as_posix(), "images": [{"overlay_file": "target.comparison.overlay.png"}]}
+ (artifact / "run_metadata.json").write_text(json.dumps(metadata), encoding="utf-8")
+ runs = MODULE.anomaly_runs(root)
+ self.assertEqual(runs[0]["id"], "anomaly-test")
+ self.assertEqual(runs[0]["label"], "五参考图边界案例")
+ self.assertEqual(runs[0]["note"], "保留真实漏检结果。")
+
def test_script_failure_becomes_a_useful_api_error(self) -> None:
handler = self.make_handler()
with self.assertRaisesRegex(MODULE.ApiError, "Processing failed"):
--
Gitblit v1.9.3