700 lines
32 KiB
Python
700 lines
32 KiB
Python
"""T-103 脱敏器测试：全部证据为合成数据，绝不读取真实 raw 目录。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from hashlib import sha256
|
||
import json
|
||
from pathlib import Path
|
||
import sys
|
||
from tempfile import TemporaryDirectory
|
||
import unittest
|
||
from unittest.mock import patch
|
||
from xml.etree import ElementTree
|
||
|
||
from PIL import Image
|
||
|
||
|
||
CLIENT_ROOT = Path(__file__).resolve().parents[2]
|
||
sys.path.insert(0, str(CLIENT_ROOT / "src"))
|
||
|
||
from cmbuyer_client.device.sku_evidence_sanitizer import (
|
||
EXPECTED_GOODS_ID,
|
||
EXPECTED_SCREENSHOT_HEIGHT,
|
||
EXPECTED_SCREENSHOT_WIDTH,
|
||
EXPECTED_XML_HEIGHT,
|
||
EXPECTED_XML_WIDTH,
|
||
HUMAN_DECLARED_STATES,
|
||
SkuEvidenceSanitizationError,
|
||
sanitize_sku_panel_evidence,
|
||
)
|
||
|
||
|
||
TEST_SERIAL = "synthetic-serial-never-publish"
|
||
TEST_ADDRESS = "SYNTHETIC_ADDRESS_NEVER_PUBLISH"
|
||
FULL_PHONE = "13800138000"
|
||
MASKED_PHONE = "138****0000"
|
||
SAFE_TEXT = "synthetic-safe-lower-content"
|
||
CURRENT_PRICE = "快卖光 ¥12.88"
|
||
ORIGINAL_PRICE = "¥29.00"
|
||
PRICE_CURRENT_BOUNDS = "[396,503][712,570]"
|
||
PRICE_ORIGINAL_BOUNDS = "[730,503][895,570]"
|
||
|
||
|
||
def _hash(path: Path) -> str:
|
||
digest = sha256()
|
||
with path.open("rb") as source:
|
||
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
||
digest.update(chunk)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def _default_xml() -> str:
|
||
return (
|
||
"<hierarchy rotation='0'>"
|
||
f"<node bounds='[0,0][1080,540]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE} {FULL_PHONE}' />"
|
||
f"{_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS)}"
|
||
f"{_price_node(ORIGINAL_PRICE, PRICE_ORIGINAL_BOUNDS)}"
|
||
f"<node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' />"
|
||
"</hierarchy>"
|
||
)
|
||
|
||
|
||
def _price_node(
|
||
text: str,
|
||
bounds: str,
|
||
*,
|
||
package: str = "com.xunmeng.pinduoduo",
|
||
node_class: str = "android.widget.TextView",
|
||
clickable: str = "false",
|
||
enabled: str = "true",
|
||
visible: str = "true",
|
||
extra_attributes: str = "",
|
||
children: str = "",
|
||
) -> str:
|
||
attributes = (
|
||
f"bounds='{bounds}' text='{text}' package='{package}' class='{node_class}' "
|
||
f"clickable='{clickable}' enabled='{enabled}' visible-to-user='{visible}'{extra_attributes}"
|
||
)
|
||
return f"<node {attributes}>{children}</node>"
|
||
|
||
|
||
def _xml_with_prices(
|
||
current: str = CURRENT_PRICE,
|
||
original: str = ORIGINAL_PRICE,
|
||
*,
|
||
current_node: str | None = None,
|
||
original_node: str | None = None,
|
||
include_original: bool = True,
|
||
extra_nodes: str = "",
|
||
) -> str:
|
||
current_markup = current_node if current_node is not None else _price_node(current, PRICE_CURRENT_BOUNDS)
|
||
original_markup = (
|
||
original_node if original_node is not None else _price_node(original, PRICE_ORIGINAL_BOUNDS)
|
||
) if include_original else ""
|
||
return (
|
||
"<hierarchy>"
|
||
f"<node bounds='[0,0][1080,540]' text='{TEST_ADDRESS}' />"
|
||
f"{current_markup}"
|
||
f"{original_markup}"
|
||
f"{extra_nodes}"
|
||
f"<node bounds='[0,570][1080,2376]' text='{SAFE_TEXT}' />"
|
||
"</hierarchy>"
|
||
)
|
||
|
||
|
||
def _write_raw(
|
||
root: Path,
|
||
*,
|
||
state: str = "panel-opened-target-preselected",
|
||
goods_id: str = EXPECTED_GOODS_ID,
|
||
model: str = "PKG110",
|
||
android_version: str = "16",
|
||
pdd_version: str = "8.17.0",
|
||
size: tuple[int, int] = (EXPECTED_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_HEIGHT),
|
||
xml: str | None = None,
|
||
) -> Path:
|
||
raw = root / "raw"
|
||
raw.mkdir(parents=True)
|
||
screenshot = raw / "screenshot.png"
|
||
image = Image.new("RGB", size, color=(0, 180, 0))
|
||
if size == (EXPECTED_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_HEIGHT):
|
||
for y in range(540):
|
||
for x in range(8):
|
||
image.putpixel((x, y), (255, 0, 0))
|
||
image.save(screenshot, format="PNG")
|
||
hierarchy = raw / "hierarchy.xml"
|
||
hierarchy.write_text(_default_xml() if xml is None else xml, encoding="utf-8")
|
||
manifest = {
|
||
"schema_version": 1,
|
||
"product": {
|
||
"goods_id": goods_id,
|
||
"canonical_url": f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}",
|
||
},
|
||
"human_declared_state": state,
|
||
"serial": TEST_SERIAL,
|
||
"channel": "wifi",
|
||
"device": {
|
||
"model": model,
|
||
"android_version": android_version,
|
||
"pdd_package": "com.xunmeng.pinduoduo",
|
||
"pdd_version": pdd_version,
|
||
},
|
||
"artifacts": [
|
||
{"path": "screenshot.png", "sha256": _hash(screenshot)},
|
||
{"path": "hierarchy.xml", "sha256": _hash(hierarchy)},
|
||
],
|
||
}
|
||
(raw / "manifest.json").write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")
|
||
return raw
|
||
|
||
|
||
class SkuEvidenceSanitizerTests(unittest.TestCase):
|
||
def test_all_declared_states_mask_screenshot_and_xml_without_raw_metadata(self) -> None:
|
||
for state in sorted(HUMAN_DECLARED_STATES):
|
||
with self.subTest(state=state), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), state=state)
|
||
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
with Image.open(result.screenshot_path) as image:
|
||
self.assertEqual(image.getpixel((0, 0)), (0, 0, 0, 255))
|
||
self.assertEqual(image.getpixel((100, 600)), (0, 180, 0, 255))
|
||
derived_xml = result.hierarchy_path.read_text(encoding="utf-8")
|
||
manifest = result.manifest_path.read_text(encoding="utf-8")
|
||
|
||
self.assertIn(SAFE_TEXT, derived_xml)
|
||
self.assertNotIn(TEST_ADDRESS, derived_xml)
|
||
self.assertNotIn(FULL_PHONE, derived_xml)
|
||
self.assertNotIn(MASKED_PHONE, derived_xml)
|
||
self.assertIn(f'"human_declared_state": "{state}"', manifest)
|
||
self.assertIn('"privacy_tier": "SANITIZED"', manifest)
|
||
self.assertIn('"sanitizer_version": "t103-privacy-v4"', manifest)
|
||
self.assertIn('"screenshot_space": {', manifest)
|
||
self.assertIn('"xml_coordinate_space": {', manifest)
|
||
self.assertIn('"height": 2376', manifest)
|
||
self.assertIn('"height": 2376', manifest)
|
||
self.assertIn('"privacy_mask_rectangle": [', manifest)
|
||
self.assertIn('"removed_nodes": 1', manifest)
|
||
self.assertIn('"cleared_crossing_nodes": 0', manifest)
|
||
self.assertIn('"preserved_crossing_price_nodes": 2', manifest)
|
||
self.assertIn('"retained_below_nodes": 1', manifest)
|
||
self.assertIn('"max_right": 1080', manifest)
|
||
self.assertIn('"max_bottom": 2376', manifest)
|
||
self.assertNotIn("canonical_url", manifest)
|
||
self.assertNotIn(TEST_SERIAL, manifest)
|
||
self.assertNotIn("serial", manifest)
|
||
self.assertNotIn("channel", manifest)
|
||
self.assertNotIn(TEST_ADDRESS, manifest)
|
||
self.assertNotIn(FULL_PHONE, manifest)
|
||
|
||
def test_crossing_container_keeps_lower_children_but_clears_its_text(self) -> None:
|
||
xml = (
|
||
"<hierarchy>"
|
||
f"<node bounds='[0,0][1080,2376]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE}'>"
|
||
f"<node bounds='[0,0][1080,540]' text='{FULL_PHONE}' />"
|
||
f"{_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS)}"
|
||
f"{_price_node(ORIGINAL_PRICE, PRICE_ORIGINAL_BOUNDS)}"
|
||
f"<node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' />"
|
||
"</node></hierarchy>"
|
||
)
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
root = ElementTree.parse(result.hierarchy_path).getroot()
|
||
crossing = root.find("node")
|
||
|
||
self.assertIsNotNone(crossing)
|
||
assert crossing is not None
|
||
self.assertEqual(crossing.attrib, {"bounds": "[0,0][1080,2376]"})
|
||
self.assertEqual(len(list(crossing)), 3)
|
||
self.assertEqual(list(crossing)[-1].get("text"), SAFE_TEXT)
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
self.assertEqual(
|
||
manifest["privacy_cleanup"],
|
||
{
|
||
"removed_nodes": 1,
|
||
"cleared_crossing_nodes": 1,
|
||
"preserved_crossing_price_nodes": 2,
|
||
"retained_below_nodes": 1,
|
||
"max_right": 1080,
|
||
"max_bottom": 2376,
|
||
},
|
||
)
|
||
|
||
def test_only_strict_crossing_price_leaves_are_projected_with_whitelisted_attributes(self) -> None:
|
||
xml = _xml_with_prices(
|
||
current_node=_price_node(
|
||
CURRENT_PRICE,
|
||
PRICE_CURRENT_BOUNDS,
|
||
extra_attributes=" content-desc='discard' resource-id='discard' focused='true'",
|
||
),
|
||
original_node=_price_node(
|
||
ORIGINAL_PRICE,
|
||
PRICE_ORIGINAL_BOUNDS,
|
||
extra_attributes=" content-desc='discard-too' resource-id='discard-too'",
|
||
),
|
||
)
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
root = ElementTree.parse(result.hierarchy_path).getroot()
|
||
prices = [node for node in root.findall("node") if node.get("text") in {CURRENT_PRICE, ORIGINAL_PRICE}]
|
||
|
||
self.assertEqual(len(prices), 2)
|
||
for node in prices:
|
||
self.assertEqual(
|
||
set(node.attrib),
|
||
{"bounds", "text", "package", "class", "clickable", "enabled", "visible-to-user"},
|
||
)
|
||
self.assertEqual(node.get("package"), "com.xunmeng.pinduoduo")
|
||
self.assertEqual(node.get("class"), "android.widget.TextView")
|
||
self.assertEqual(node.get("clickable"), "false")
|
||
self.assertEqual(node.get("enabled"), "true")
|
||
self.assertEqual(node.get("visible-to-user"), "true")
|
||
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
self.assertEqual(manifest["privacy_cleanup"]["preserved_crossing_price_nodes"], 2)
|
||
self.assertNotIn("discard", result.hierarchy_path.read_text(encoding="utf-8"))
|
||
|
||
def test_crossing_price_window_rejects_text_and_structure_drift(self) -> None:
|
||
bad_texts = (
|
||
f"快卖光 ¥12.88 {TEST_ADDRESS}",
|
||
f"快卖光 ¥12.88 {FULL_PHONE}",
|
||
"快卖光 ¥12.88 使用微信支付",
|
||
"快卖光 ¥12.88 提交订单",
|
||
"快卖光 ¥12.88 优惠-11元",
|
||
"快要抢光 ¥12.88",
|
||
"快卖光 ¥0.00",
|
||
"快卖光 ¥12.8",
|
||
"快卖光 ¥12.880",
|
||
"快卖光 ¥12.88",
|
||
)
|
||
for text in bad_texts:
|
||
with self.subTest(text=text), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=_xml_with_prices(current=text))
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
def test_crossing_price_text_mismatch_reports_only_fixed_slot_and_reason(self) -> None:
|
||
newline_node = _price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS).replace(
|
||
"快卖光 ¥12.88", "快卖光 ¥12.88"
|
||
)
|
||
cases = (
|
||
("newline", _xml_with_prices(current_node=newline_node), "newline", PRICE_CURRENT_BOUNDS),
|
||
(
|
||
"non-ascii-whitespace",
|
||
_xml_with_prices(current="快卖光 ¥12.88"),
|
||
"non_ascii_whitespace",
|
||
PRICE_CURRENT_BOUNDS,
|
||
),
|
||
(
|
||
"known-prefix-missing",
|
||
_xml_with_prices(current="快要抢光 ¥12.88"),
|
||
"known_prefix_missing",
|
||
PRICE_CURRENT_BOUNDS,
|
||
),
|
||
(
|
||
"currency-missing",
|
||
_xml_with_prices(current="快卖光 12.88"),
|
||
"currency_missing",
|
||
PRICE_CURRENT_BOUNDS,
|
||
),
|
||
(
|
||
"amount-shape",
|
||
_xml_with_prices(current="快卖光 ¥12.8"),
|
||
"amount_shape",
|
||
PRICE_CURRENT_BOUNDS,
|
||
),
|
||
(
|
||
"extra-or-order",
|
||
_xml_with_prices(current="提交订单 ¥12.88"),
|
||
"extra_or_order",
|
||
PRICE_CURRENT_BOUNDS,
|
||
),
|
||
(
|
||
"forbidden-characters",
|
||
_xml_with_prices(current="商品 ¥12.88"),
|
||
"forbidden_characters",
|
||
PRICE_CURRENT_BOUNDS,
|
||
),
|
||
(
|
||
"right-slot-amount-shape",
|
||
_xml_with_prices(original="¥29.0"),
|
||
"amount_shape",
|
||
PRICE_ORIGINAL_BOUNDS,
|
||
),
|
||
)
|
||
for name, xml, reason, bounds in cases:
|
||
with self.subTest(name=name), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
with self.assertRaises(SkuEvidenceSanitizationError) as raised:
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
|
||
self.assertEqual(
|
||
str(raised.exception),
|
||
f"跨界价格节点文本不匹配：slot={bounds}；reason={reason}。",
|
||
)
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
def test_crossing_price_text_mismatch_never_echoes_sensitive_or_order_text(self) -> None:
|
||
cases = (
|
||
f"快卖光 ¥12.88 {TEST_ADDRESS}",
|
||
f"快卖光 ¥12.88 {FULL_PHONE}",
|
||
"快卖光 ¥12.88 使用微信支付",
|
||
"快卖光 ¥12.88 提交订单",
|
||
)
|
||
for text in cases:
|
||
with self.subTest(text=text), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=_xml_with_prices(current=text))
|
||
with self.assertRaises(SkuEvidenceSanitizationError) as raised:
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
|
||
message = str(raised.exception)
|
||
self.assertIn("slot=[396,503][712,570]", message)
|
||
self.assertIn("reason=extra_or_order", message)
|
||
for raw_fragment in (TEST_ADDRESS, FULL_PHONE, "使用微信支付", "提交订单", "¥12.88"):
|
||
self.assertNotIn(raw_fragment, message)
|
||
|
||
def test_crossing_price_projection_allows_only_limited_ascii_spaces_and_yen_variants(self) -> None:
|
||
for current in ("快卖光 ¥12.88", " 快卖光 ¥ 12.88 ", "快卖光 ￥12.88"):
|
||
with self.subTest(current=current), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=_xml_with_prices(current=current))
|
||
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
hierarchy = result.hierarchy_path.read_text(encoding="utf-8")
|
||
self.assertIn(current, hierarchy)
|
||
|
||
def test_unique_current_price_without_original_price_is_published(self) -> None:
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=_xml_with_prices(include_original=False))
|
||
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
hierarchy = result.hierarchy_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
|
||
self.assertIn(CURRENT_PRICE, hierarchy)
|
||
self.assertNotIn(ORIGINAL_PRICE, hierarchy)
|
||
self.assertEqual(manifest["privacy_cleanup"]["preserved_crossing_price_nodes"], 1)
|
||
|
||
def test_crossing_price_projection_rejects_newline_and_structure_drift(self) -> None:
|
||
encoded_newline = _price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS).replace(
|
||
"快卖光 ¥12.88", "快卖光 ¥12.88"
|
||
)
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=_xml_with_prices(current_node=encoded_newline))
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
bad_structure = (
|
||
("package", {"package": "com.android.systemui"}),
|
||
("class", {"node_class": "android.view.View"}),
|
||
("clickable", {"clickable": "true"}),
|
||
("disabled", {"enabled": "false"}),
|
||
("hidden", {"visible": "false"}),
|
||
("children", {"children": "<node bounds='[400,510][500,520]' />"}),
|
||
)
|
||
for name, kwargs in bad_structure:
|
||
with self.subTest(structure=name), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(
|
||
Path(temporary),
|
||
xml=_xml_with_prices(current_node=_price_node(CURRENT_PRICE, PRICE_CURRENT_BOUNDS, **kwargs)),
|
||
)
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
def test_crossing_price_candidates_require_unique_current_and_at_most_one_original(self) -> None:
|
||
scenarios = (
|
||
(
|
||
"missing-current",
|
||
_xml_with_prices(current="¥12.88", original=ORIGINAL_PRICE),
|
||
),
|
||
(
|
||
"duplicate-current",
|
||
_xml_with_prices(current=CURRENT_PRICE, original=f"快卖光 {ORIGINAL_PRICE}"),
|
||
),
|
||
(
|
||
"multiple-original",
|
||
_xml_with_prices(
|
||
extra_nodes=_price_node("¥39.88", PRICE_ORIGINAL_BOUNDS),
|
||
),
|
||
),
|
||
)
|
||
for name, xml in scenarios:
|
||
with self.subTest(name=name), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
def test_crossing_price_outside_fixed_windows_is_cleared_and_submit_price_is_not_candidate(self) -> None:
|
||
outside_crossing = _price_node("快卖光 ¥99.99", "[396,498][712,570]")
|
||
submit = (
|
||
"<node bounds='[369,2225][710,2284]' text='提交订单 ¥12.88' "
|
||
"package='com.xunmeng.pinduoduo' class='android.widget.TextView' clickable='false' "
|
||
"enabled='true' visible-to-user='true' resource-id='submit-button' />"
|
||
)
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=_xml_with_prices(extra_nodes=outside_crossing + submit))
|
||
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
hierarchy = result.hierarchy_path.read_text(encoding="utf-8")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
|
||
self.assertNotIn("快卖光 ¥99.99", hierarchy)
|
||
self.assertIn("提交订单 ¥12.88", hierarchy)
|
||
self.assertIn("submit-button", hierarchy)
|
||
self.assertEqual(manifest["privacy_cleanup"]["preserved_crossing_price_nodes"], 2)
|
||
|
||
def test_same_raw_and_config_produce_identical_derived_files(self) -> None:
|
||
with TemporaryDirectory() as temporary:
|
||
root = Path(temporary)
|
||
left_raw = _write_raw(root / "left")
|
||
right_raw = _write_raw(root / "right")
|
||
left = sanitize_sku_panel_evidence(left_raw, left_raw.parent / "derived")
|
||
right = sanitize_sku_panel_evidence(right_raw, right_raw.parent / "derived")
|
||
|
||
for left_path, right_path in (
|
||
(left.screenshot_path, right.screenshot_path),
|
||
(left.hierarchy_path, right.hierarchy_path),
|
||
(left.manifest_path, right.manifest_path),
|
||
):
|
||
self.assertEqual(left_path.read_bytes(), right_path.read_bytes())
|
||
|
||
def test_manifest_records_distinct_screenshot_and_xml_coordinate_spaces(self) -> None:
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary))
|
||
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
||
|
||
self.assertEqual(
|
||
manifest["screenshot_space"],
|
||
{"width": 1080, "height": 2376, "privacy_mask_rectangle": [0, 0, 1080, 540]},
|
||
)
|
||
self.assertEqual(
|
||
manifest["xml_coordinate_space"],
|
||
{
|
||
"width": 1080,
|
||
"height": 2376,
|
||
"privacy_mask_rectangle": [0, 0, 1080, 540],
|
||
"observed_max": {"right": 1080, "bottom": 2376},
|
||
},
|
||
)
|
||
|
||
def test_hash_and_metadata_mismatch_fail_closed_without_leak(self) -> None:
|
||
scenarios = (
|
||
("hash", {}, "screenshot"),
|
||
("model", {"model": "other"}, None),
|
||
("android", {"android_version": "15"}, None),
|
||
("version", {"pdd_version": "8.17.1"}, None),
|
||
("goods", {"goods_id": "123"}, None),
|
||
("state", {"state": "guessed"}, None),
|
||
("old-screenshot-space", {"size": (1080, 2400)}, None),
|
||
("other-screenshot-space", {"size": (100, 100)}, None),
|
||
)
|
||
for name, kwargs, corrupt_file in scenarios:
|
||
with self.subTest(name=name), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), **kwargs)
|
||
if corrupt_file is not None:
|
||
(raw / f"{corrupt_file}.png").write_bytes(b"changed")
|
||
with self.assertRaises(SkuEvidenceSanitizationError) as raised:
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
|
||
message = str(raised.exception)
|
||
self.assertNotIn(TEST_SERIAL, message)
|
||
self.assertNotIn(TEST_ADDRESS, message)
|
||
self.assertNotIn(FULL_PHONE, message)
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
|
||
|
||
def test_old_and_unknown_human_states_are_rejected(self) -> None:
|
||
for state in ("initial", "one-dimension-selected", "all-dimensions-selected", "guessed"):
|
||
with self.subTest(state=state), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), state=state)
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
def test_malformed_inputs_and_bounds_or_phone_residue_fail_closed(self) -> None:
|
||
malformed = (
|
||
("manifest", None),
|
||
("png", None),
|
||
("xml", None),
|
||
("bounds", "<hierarchy><node text='missing bounds' /></hierarchy>"),
|
||
(
|
||
"private-parent-with-lower-child",
|
||
"<hierarchy><node bounds='[0,0][1080,540]'><node bounds='[0,540][1080,2376]' text='x' /></node></hierarchy>",
|
||
),
|
||
("full-phone-below", f"<hierarchy><node bounds='[0,540][1080,2376]' text='{FULL_PHONE}' /></hierarchy>"),
|
||
("masked-phone-below", f"<hierarchy><node bounds='[0,540][1080,2376]' text='{MASKED_PHONE}' /></hierarchy>"),
|
||
)
|
||
for kind, xml in malformed:
|
||
with self.subTest(kind=kind), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
if kind == "manifest":
|
||
(raw / "manifest.json").write_text("{invalid", encoding="utf-8")
|
||
elif kind == "png":
|
||
(raw / "screenshot.png").write_bytes(b"not a png")
|
||
manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8"))
|
||
manifest["artifacts"][0]["sha256"] = _hash(raw / "screenshot.png")
|
||
(raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
||
elif kind == "xml":
|
||
(raw / "hierarchy.xml").write_text("<hierarchy>", encoding="utf-8")
|
||
manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8"))
|
||
manifest["artifacts"][1]["sha256"] = _hash(raw / "hierarchy.xml")
|
||
(raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
|
||
|
||
def test_phone_recheck_rejects_separator_mask_and_cross_node_bypasses(self) -> None:
|
||
variants = (
|
||
"138 0013-8000",
|
||
"138＊＊＊＊0000",
|
||
"138••••0000",
|
||
"138xxxx0000",
|
||
"138XXXX0000",
|
||
)
|
||
for value in variants:
|
||
xml = (
|
||
"<hierarchy>"
|
||
"<node bounds='[0,0][1080,540]' text='private' />"
|
||
f"<node bounds='[0,540][1080,2376]' text='{value}' />"
|
||
"</hierarchy>"
|
||
)
|
||
with self.subTest(value=value), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
split_xml = (
|
||
"<hierarchy>"
|
||
"<node bounds='[0,0][1080,540]' text='private' />"
|
||
"<node bounds='[0,540][1080,1000]' text='138' content-desc='0013' />"
|
||
"<node bounds='[0,1000][1080,2376]' text='8000' />"
|
||
"</hierarchy>"
|
||
)
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=split_xml)
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
def test_privacy_geometry_requires_removed_and_retained_nodes(self) -> None:
|
||
scenarios = (
|
||
("no-private", f"<hierarchy><node bounds='[0,540][1080,2376]' text='{SAFE_TEXT}' /></hierarchy>"),
|
||
("no-below", "<hierarchy><node bounds='[0,0][1080,540]' text='private' /></hierarchy>"),
|
||
)
|
||
for name, xml in scenarios:
|
||
with self.subTest(name=name), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
|
||
|
||
def test_xml_coordinate_space_must_have_exact_configured_maximums(self) -> None:
|
||
scenarios = (
|
||
(
|
||
"short-width",
|
||
"<hierarchy><node bounds='[0,0][1079,540]' text='private' />"
|
||
"<node bounds='[0,540][1079,2376]' text='safe' /></hierarchy>",
|
||
"1079x2376",
|
||
),
|
||
(
|
||
"short-height",
|
||
"<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
|
||
"<node bounds='[0,540][1080,2375]' text='safe' /></hierarchy>",
|
||
"1080x2375",
|
||
),
|
||
(
|
||
"wide-width",
|
||
"<hierarchy><node bounds='[0,0][1081,540]' text='private' />"
|
||
"<node bounds='[0,540][1081,2376]' text='safe' /></hierarchy>",
|
||
"1081x2376",
|
||
),
|
||
(
|
||
"old-v2-xml-height",
|
||
"<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
|
||
"<node bounds='[0,540][1080,2400]' text='safe' /></hierarchy>",
|
||
"1080x2400",
|
||
),
|
||
)
|
||
for name, xml, observed in scenarios:
|
||
with self.subTest(name=name), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary), xml=xml)
|
||
with self.assertRaises(SkuEvidenceSanitizationError) as raised:
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
|
||
self.assertIn(observed, str(raised.exception))
|
||
self.assertNotIn(TEST_SERIAL, str(raised.exception))
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
|
||
|
||
def test_manifest_schema_package_and_artifact_structure_are_required(self) -> None:
|
||
def mutate(manifest: dict[str, object], kind: str) -> None:
|
||
if kind == "schema":
|
||
manifest["schema_version"] = 2
|
||
elif kind == "package":
|
||
manifest["device"]["pdd_package"] = "com.example.other" # type: ignore[index]
|
||
elif kind == "missing":
|
||
manifest.pop("artifacts")
|
||
elif kind == "duplicate":
|
||
manifest["artifacts"].append(manifest["artifacts"][0]) # type: ignore[index]
|
||
elif kind == "bad-hash":
|
||
manifest["artifacts"][0]["sha256"] = "g" * 64 # type: ignore[index]
|
||
|
||
for kind in ("schema", "package", "missing", "duplicate", "bad-hash"):
|
||
with self.subTest(kind=kind), TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary))
|
||
manifest_path = raw / "manifest.json"
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
mutate(manifest, kind)
|
||
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
with self.assertRaises(SkuEvidenceSanitizationError) as raised:
|
||
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
||
|
||
self.assertNotIn(TEST_SERIAL, str(raised.exception))
|
||
self.assertNotIn(TEST_ADDRESS, str(raised.exception))
|
||
self.assertFalse((raw.parent / "derived").exists())
|
||
|
||
def test_target_created_during_publish_is_preserved_and_staging_is_removed(self) -> None:
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary))
|
||
target = raw.parent / "derived"
|
||
|
||
def race_rename(destination: Path) -> None:
|
||
destination.mkdir()
|
||
(destination / "sentinel.txt").write_text("keep", encoding="utf-8")
|
||
raise FileExistsError("simulated publish race")
|
||
|
||
with patch("cmbuyer_client.device.sku_evidence_sanitizer.Path.rename", side_effect=race_rename):
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, target)
|
||
|
||
self.assertEqual((target / "sentinel.txt").read_text(encoding="utf-8"), "keep")
|
||
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
|
||
|
||
def test_existing_derived_is_preserved_without_reading_or_writing_raw(self) -> None:
|
||
with TemporaryDirectory() as temporary:
|
||
raw = _write_raw(Path(temporary))
|
||
target = raw.parent / "derived"
|
||
target.mkdir()
|
||
sentinel = target / "sentinel.txt"
|
||
sentinel.write_text("keep", encoding="utf-8")
|
||
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, target)
|
||
|
||
self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
|
||
self.assertEqual(list(raw.parent.glob(".derived.staging-*")), [])
|
||
|
||
def test_directory_contract_rejects_non_sibling_paths(self) -> None:
|
||
with TemporaryDirectory() as temporary:
|
||
root = Path(temporary)
|
||
raw = _write_raw(root)
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(raw, root / "not-derived")
|
||
with self.assertRaises(SkuEvidenceSanitizationError):
|
||
sanitize_sku_panel_evidence(root / "not-raw", root / "derived")
|