| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081 |
- """安全闸门与三层状态判定守卫（任务书 P1-B 安全 / P1-G 结果分层）。"""
- from __future__ import annotations
- import pytest
- from clients import aidop_probe
- ALLOWED = ["127.0.0.1", "localhost"]
- @pytest.mark.parametrize("url", [
- "http://127.0.0.1:5005",
- "http://localhost:5005",
- ])
- def test_whitelisted_hosts_pass(url):
- aidop_probe.assert_host_allowed(url, ALLOWED)
- @pytest.mark.parametrize("url", [
- "http://10.20.30.40:5005",
- "https://aidop.customer.com",
- "http://192.168.1.99:5005",
- "http://prod-aidop.internal",
- ])
- def test_non_whitelisted_hosts_blocked(url):
- with pytest.raises(aidop_probe.HostNotAllowedError):
- aidop_probe.assert_host_allowed(url, ALLOWED)
- @pytest.mark.parametrize("status,body,expected", [
- (0, {"message": "SOURCE_UNREACHABLE: refused"}, "SOURCE_UNREACHABLE"),
- (200, {"code": 0, "data": {"fields": []}}, "READY"),
- (404, {"message": "entity not registered"}, "CONFIG_NOT_REGISTERED"),
- (403, {"message": "inbound not enabled for entity"}, "CONFIG_NOT_ENABLED"),
- (403, {"message": "grant missing for accessKey"}, "GRANT_MISSING"),
- (401, {"message": "AccessKey 未绑定开放身份"}, "GRANT_MISSING"),
- (401, {"message": "signature mismatch"}, "CONFIG_NOT_ENABLED"),
- (500, {"message": "boom"}, "RUN_FAILED"),
- ])
- def test_precheck_state_classification(status, body, expected):
- assert aidop_probe.classify_inbound_precheck(status, body) == expected
- def test_unreachable_target_reports_source_unreachable():
- result = aidop_probe.check_reachable("http://127.0.0.1:1", ALLOWED, timeout=2)
- assert result["reachable"] is False
- def test_probe_rejects_non_whitelisted_target():
- with pytest.raises(aidop_probe.HostNotAllowedError):
- aidop_probe.check_reachable("http://10.0.0.9:5005", ALLOWED, timeout=2)
- def test_run_store_redacts_before_persist():
- """运行记录写入前必须脱敏（不依赖已启动服务，直接测 RunStore）。"""
- import app as sim_app
- store = sim_app.RunStore(limit=5)
- entry = store.add(
- run_id="20260930120000-abc123", object_code="ITEM", channel="API_INBOUND",
- request_summary={"operation": "push"},
- response={"headers": {"X-Signature": "SIG==", "Authorization": "Bearer abc.def"},
- "note": "password=hunter2"},
- layers={"code": "SUPPORTED", "config": "READY", "data": "ACCEPTED"},
- followup_sql=["SELECT 1;"])
- dumped = str(entry)
- assert "SIG==" not in dumped
- assert "abc.def" not in dumped
- assert "hunter2" not in dumped
- assert entry["layers"]["data"] == "ACCEPTED"
- assert store.recent()[0]["runId"] == "20260930120000-abc123"
- def test_run_store_respects_limit():
- import app as sim_app
- store = sim_app.RunStore(limit=3)
- for i in range(6):
- store.add(str(i), "ITEM", "DB_SYNC", {}, {}, {})
- assert len(store.runs) == 3
- assert [r["runId"] for r in store.recent()] == ["5", "4", "3"]
|
