773 lines
26 KiB
Python
773 lines
26 KiB
Python
"""Hermetic tests for the unbroker skill.
|
|
|
|
Stdlib + pytest only; NO live network, NO browser, NO email. Each test runs against
|
|
an isolated temp PDD_DATA_DIR. Runnable with pytest or directly:
|
|
|
|
python3 -m pytest tests/test_unbroker_skill.py -q
|
|
python3 tests/test_unbroker_skill.py # portable fallback runner
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
# Resolve the skill's scripts dir across layouts: standalone dev repo (tests/) and hermes-agent
|
|
# (tests/skills/ -> optional-skills/security/unbroker/scripts).
|
|
_HERE = Path(__file__).resolve()
|
|
_REL = ("optional-skills", "security", "unbroker", "scripts")
|
|
_CANDIDATES = [
|
|
_HERE.parent.parent / "skill" / "scripts", # standalone dev repo
|
|
_HERE.parent.parent.joinpath(*_REL), # standalone layout
|
|
_HERE.parent.parent.parent.joinpath(*_REL), # hermes-agent (tests/skills/)
|
|
]
|
|
SCRIPTS = next((c for c in _CANDIDATES if (c / "pdd.py").exists()), _CANDIDATES[0])
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
import autopilot # noqa: E402
|
|
import contextlib as _ctx # noqa: E402
|
|
import io as _io # noqa: E402
|
|
import json as _json # noqa: E402
|
|
import smtplib as _smtplib # noqa: E402
|
|
import time as _time # noqa: E402
|
|
|
|
import badbool # noqa: E402
|
|
import brokers # noqa: E402
|
|
import cdp # noqa: E402
|
|
import config # noqa: E402
|
|
import crypto # noqa: E402
|
|
import dossier # noqa: E402
|
|
import email_modes # noqa: E402
|
|
import emailer # noqa: E402
|
|
import pdd # noqa: E402
|
|
import legal # noqa: E402
|
|
import ledger # noqa: E402
|
|
import paths # noqa: E402
|
|
import registry # noqa: E402
|
|
import report # noqa: E402
|
|
import storage # noqa: E402
|
|
import tiers # noqa: E402
|
|
import vectors # noqa: E402
|
|
|
|
_AGE = bool(shutil.which("age") and shutil.which("age-keygen"))
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def temp_env():
|
|
"""Isolate every test in a fresh PDD_DATA_DIR."""
|
|
prev = os.environ.get("PDD_DATA_DIR")
|
|
with tempfile.TemporaryDirectory() as d:
|
|
os.environ["PDD_DATA_DIR"] = str(Path(d) / "pdd")
|
|
try:
|
|
yield Path(os.environ["PDD_DATA_DIR"])
|
|
finally:
|
|
if prev is None:
|
|
os.environ.pop("PDD_DATA_DIR", None)
|
|
else:
|
|
os.environ["PDD_DATA_DIR"] = prev
|
|
|
|
|
|
def _consenting(full_name="Jane Q. Public"):
|
|
return {
|
|
"subject_id": "sub_test01",
|
|
"consent": {"authorized": True, "method": "self"},
|
|
"identity": {
|
|
"full_name": full_name,
|
|
"emails": ["jane@example.com"],
|
|
"phones": ["+1-415-555-0137"],
|
|
"date_of_birth": "1987-04-12",
|
|
"current_address": {"city": "Oakland", "state": "CA", "postal": "94601"},
|
|
},
|
|
"preferences": {"email_mode": "draft_only"},
|
|
}
|
|
|
|
|
|
# --- config -------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
def test_browser_clears_captcha_logic():
|
|
assert config.browser_clears_captcha({"browser_backend": "browserbase"}) is True
|
|
assert config.browser_clears_captcha({"browser_backend": "agent-browser"}) is False
|
|
assert config.browser_clears_captcha({"browser_backend": "auto"}, env={}) is False
|
|
assert config.browser_clears_captcha({"browser_backend": "auto"}, env={"BROWSERBASE_API_KEY": "x"}) is True
|
|
|
|
|
|
# --- storage ------------------------------------------------------------------
|
|
|
|
def test_storage_json_and_jsonl_roundtrip():
|
|
with temp_env() as data:
|
|
p = data / "x.json"
|
|
storage.write_json(p, {"a": 1})
|
|
assert storage.read_json(p) == {"a": 1}
|
|
assert storage.read_json(data / "missing.json", []) == []
|
|
log = data / "audit.jsonl"
|
|
storage.append_jsonl(log, {"e": 1})
|
|
storage.append_jsonl(log, {"e": 2})
|
|
assert [r["e"] for r in storage.read_jsonl(log)] == [1, 2]
|
|
|
|
|
|
# --- at-rest encryption -------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- broker DB ----------------------------------------------------------------
|
|
|
|
def test_seed_broker_db_loads_and_is_well_formed():
|
|
everyone = brokers.load_all()
|
|
assert len(everyone) >= 10
|
|
ids = {b["id"] for b in everyone}
|
|
assert {"spokeo", "whitepages", "mylife"} <= ids
|
|
for b in everyone:
|
|
assert b.get("id") and b.get("name") and b.get("priority") in {"crucial", "high", "standard", "long_tail"}
|
|
assert (b.get("optout") or {}).get("method")
|
|
|
|
|
|
|
|
|
|
def test_blocked_pass_records_and_cluster_coverage():
|
|
# Records added from the blocked-tail pass load, resolve, and dedupe correctly.
|
|
ids = {b["id"] for b in brokers.load_all()}
|
|
assert {"addresses", "socialcatfish"} <= ids
|
|
# addresses.com is a PeopleConnect/Intelius front-end -> covered by the intelius cluster (deduped).
|
|
assert "addresses" in brokers.clusters().get("intelius", [])
|
|
for bid in ("addresses", "socialcatfish"):
|
|
b = brokers.get(bid)
|
|
assert tiers.select_tier(b) in {"T0", "T1", "T2", "T3"}
|
|
assert b["optout"]["method"]
|
|
|
|
|
|
# --- tier selection -----------------------------------------------------------
|
|
|
|
def test_every_broker_resolves_to_valid_tier():
|
|
for b in brokers.load_all():
|
|
assert tiers.select_tier(b) in {"T0", "T1", "T2", "T3"}
|
|
|
|
|
|
|
|
|
|
def test_captcha_tier_shifts_with_browser():
|
|
tps = brokers.get("truepeoplesearch")
|
|
assert tiers.select_tier(tps, "programmatic", browser_clears_captcha=False) == "T2"
|
|
assert tiers.select_tier(tps, "programmatic", browser_clears_captcha=True) == "T1"
|
|
|
|
|
|
|
|
|
|
def test_plan_excludes_disallowed_fields():
|
|
d = _consenting()
|
|
actions = tiers.plan(d, brokers.load_all(), config.DEFAULT_CONFIG)
|
|
for a in actions:
|
|
assert "ssn" not in a["disclosure_fields"]
|
|
assert "profile_url" not in a["disclosure_fields"]
|
|
|
|
|
|
|
|
|
|
def _mini_broker(bid, owns=None, requires=None, notes="", quirks=None):
|
|
return {"id": bid, "name": bid.title(), "priority": "high",
|
|
"search": {"by": ["name"]},
|
|
"optout": {"method": "web_form", "url": f"https://{bid}.example/optout",
|
|
"requires": requires or {}, "inputs": ["full_name"], "owns": owns or [],
|
|
"notes": notes, "quirks": quirks or []},
|
|
"owns": owns or []}
|
|
|
|
|
|
def test_batch_plan_groups_by_ledger_state():
|
|
d = _consenting()
|
|
bl = [_mini_broker("aaa"), _mini_broker("bbb"), _mini_broker("ccc"), _mini_broker("ddd")]
|
|
ledger = {
|
|
"aaa": {"state": "found"},
|
|
"bbb": {"state": "not_found"},
|
|
"ccc": {"state": "blocked"},
|
|
# ddd absent -> unscanned/new
|
|
}
|
|
bp = tiers.batch_plan(d, bl, config.DEFAULT_CONFIG, ledger)
|
|
assert bp["phase"] == "discover" # ddd is unscanned
|
|
assert bp["counts"]["found"] == 1
|
|
assert bp["counts"]["not_found"] == 1
|
|
assert bp["counts"]["blocked"] == 1
|
|
assert bp["counts"]["unscanned"] == 1
|
|
assert any("PHASE 1" in t for t in bp["next_actions"])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- ledger / state machine ---------------------------------------------------
|
|
|
|
def test_ledger_valid_transition_and_audit():
|
|
with temp_env():
|
|
sid = "sub_test01"
|
|
ledger.transition(sid, "spokeo", "searching")
|
|
case = ledger.transition(sid, "spokeo", "found", found=True)
|
|
assert case["state"] == "found" and case["found"] is True
|
|
# found -> submitted must be allowed directly (action_selected is optional)
|
|
case = ledger.transition(sid, "spokeo", "submitted")
|
|
assert case["state"] == "submitted"
|
|
audit = storage.read_jsonl(__import__("paths").audit_path(sid))
|
|
assert any(e["to"] == "found" for e in audit)
|
|
|
|
|
|
|
|
|
|
def test_indirect_exposure_state_and_transitions():
|
|
with temp_env():
|
|
sid = "sub_test01"
|
|
# a scan can land directly on indirect_exposure (PII on a relative's record)
|
|
case = ledger.transition(sid, "thatsthem", "indirect_exposure",
|
|
evidence={"summary": "email on relative record"})
|
|
assert case["state"] == "indirect_exposure"
|
|
# the lever from there is a targeted delete-my-PII request (-> submitted)
|
|
assert ledger.transition(sid, "thatsthem", "submitted")["state"] == "submitted"
|
|
# and a separate broker: not_found -> indirect_exposure is allowed (found on re-read)
|
|
ledger.transition(sid, "radaris", "not_found")
|
|
assert ledger.transition(sid, "radaris", "indirect_exposure")["state"] == "indirect_exposure"
|
|
# re-scan can clear it
|
|
assert ledger.transition(sid, "radaris", "not_found")["state"] == "not_found"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- dossier / consent / least-disclosure ------------------------------------
|
|
|
|
|
|
|
|
def test_least_disclosure_selection():
|
|
d = _consenting()
|
|
got = dossier.select_disclosure(d, ["full_name", "contact_email", "profile_url", "ssn", "date_of_birth"])
|
|
assert set(got) == {"full_name", "contact_email", "date_of_birth"}
|
|
assert "ssn" not in got and "profile_url" not in got
|
|
|
|
|
|
|
|
|
|
# --- alternates / search vectors ---------------------------------------------
|
|
|
|
def test_all_names_and_locations_dedupe():
|
|
d = _consenting()
|
|
d["identity"]["also_known_as"] = ["Jane Public", "Jane Q. Public"] # 2nd dups primary
|
|
d["identity"]["prior_addresses"] = [{"city": "Berkeley", "state": "CA"}, {"city": "Oakland", "state": "CA"}]
|
|
assert dossier.all_names(d) == ["Jane Q. Public", "Jane Public"]
|
|
assert [loc["city"] for loc in dossier.all_locations(d)] == ["Oakland", "Berkeley"] # current first, deduped
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- opaque ids / fan-out / antibot ------------------------------------------
|
|
|
|
def test_subject_id_is_opaque_no_name_leak():
|
|
sid = dossier.new_subject_id("Maiden Married Person")
|
|
assert sid.startswith("sub_")
|
|
assert "maiden" not in sid.lower() and "person" not in sid.lower()
|
|
assert dossier.new_subject_id("Maiden Married Person") != sid # not derived from the name
|
|
|
|
|
|
def test_fanout_batches_large_runs():
|
|
g = tiers.fanout([{"id": f"b{i}"} for i in range(20)], batch_size=8)
|
|
assert g["broker_count"] == 20 and g["should_fanout"] is True
|
|
assert len(g["batches"]) == 3 and g["batches"][0] == [f"b{i}" for i in range(8)]
|
|
small = tiers.fanout([{"id": "x"}, {"id": "y"}], batch_size=8)
|
|
assert small["should_fanout"] is False and small["batches"] == [["x", "y"]]
|
|
|
|
|
|
|
|
|
|
# --- cdp (operator browser over the DevTools protocol) --------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- legal / templates --------------------------------------------------------
|
|
|
|
|
|
|
|
def test_render_optout_email_includes_listing_and_name():
|
|
b = brokers.get("spokeo")
|
|
out = legal.render_optout_email(b, {"full_name": "Jane Q. Public",
|
|
"contact_email": "jane@example.com",
|
|
"listing_urls": ["https://www.spokeo.com/jane"]})
|
|
assert "Jane Q. Public" in out and "https://www.spokeo.com/jane" in out
|
|
|
|
|
|
|
|
|
|
# --- email verification-link extraction --------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
# --- BADBOOL live-pull parser -------------------------------------------------
|
|
|
|
BADBOOL_FIXTURE = """
|
|
## Search Engines
|
|
### Google
|
|
This is not a broker; ignore it.
|
|
|
|
## People Search Sites
|
|
|
|
### \U0001F490 BeenVerified
|
|
Find your information and opt out of [people search](https://www.beenverified.com/app/optout/search).
|
|
|
|
### \U0001F490 \U0001F4DE MyLife
|
|
[Find your information](https://www.mylife.com), and then [opt out](https://www.mylife.com/privacyrequest).
|
|
|
|
### \U0001F3AB PimEyes
|
|
To opt out, [upload an ID](https://pimeyes.com/en/opt-out-request-form).
|
|
|
|
## Special Circumstances
|
|
### Not A Broker
|
|
Ignore this section entirely.
|
|
"""
|
|
|
|
|
|
def test_badbool_parses_people_search_section_only():
|
|
recs = badbool.parse(BADBOOL_FIXTURE)
|
|
ids = {r["id"] for r in recs}
|
|
assert ids == {"beenverified", "mylife", "pimeyes"} # google + notabroker excluded
|
|
bv = next(r for r in recs if r["id"] == "beenverified")
|
|
assert bv["priority"] == "crucial"
|
|
assert "beenverified.com/app/optout" in (bv["optout"]["url"] or "")
|
|
assert bv["source"] == "BADBOOL-auto" and bv["confidence"] == "auto"
|
|
|
|
|
|
|
|
|
|
def test_badbool_merge_keeps_curated_and_adds_new():
|
|
with temp_env():
|
|
badbool.refresh(__import__("paths").brokers_cache_path(), markdown=BADBOOL_FIXTURE)
|
|
merged = {b["id"]: b for b in brokers.load_all()}
|
|
# curated record wins over the live one
|
|
assert merged["beenverified"]["source"] == "BADBOOL"
|
|
# a non-curated live record is added with auto confidence
|
|
assert "pimeyes" in merged and merged["pimeyes"]["confidence"] == "auto"
|
|
|
|
|
|
# --- report -------------------------------------------------------------------
|
|
|
|
|
|
|
|
# --- autonomy: auto-configure ---------------------------------------------------------------
|
|
|
|
|
|
|
|
def test_auto_configure_picks_most_autonomous():
|
|
with temp_env():
|
|
# bare env -> draft_only floor, auto browser (still fully hands-off policy-wise)
|
|
cfg = config.auto_configure(env={})
|
|
assert cfg["autonomy"] == "full"
|
|
assert cfg["email_mode"] == "draft_only"
|
|
assert cfg["browser_backend"] == "auto"
|
|
# SMTP creds -> programmatic email; Browserbase key -> cloud browser
|
|
cfg = config.auto_configure(env={"EMAIL_ADDRESS": "agent@gmail.com",
|
|
"EMAIL_PASSWORD": "app-pass",
|
|
"BROWSERBASE_API_KEY": "bb"})
|
|
assert cfg["email_mode"] == "programmatic"
|
|
assert cfg["browser_backend"] == "browserbase"
|
|
# AgentMail only -> alias mode
|
|
assert config.auto_configure(env={"AGENTMAIL_API_KEY": "am"})["email_mode"] == "alias"
|
|
# encryption auto-on exactly when age is installed (free privacy, zero human cost)
|
|
assert config.auto_configure(env={})["encryption"] == ("age" if _AGE else "none")
|
|
|
|
|
|
# --- emailer: programmatic send + verification polling --------------------------------------
|
|
|
|
|
|
|
|
class _FakeSMTP:
|
|
sent: list = []
|
|
|
|
def __init__(self, host, port, timeout=None):
|
|
self.host, self.port = host, port
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def ehlo(self):
|
|
pass
|
|
|
|
def starttls(self):
|
|
pass
|
|
|
|
def login(self, user, password):
|
|
self.user = user
|
|
|
|
def send_message(self, msg):
|
|
_FakeSMTP.sent.append(msg)
|
|
|
|
|
|
def test_emailer_send_locks_recipient_to_broker():
|
|
env = {"EMAIL_ADDRESS": "agent@gmail.com", "EMAIL_PASSWORD": "p"}
|
|
broker = {"id": "radaris", "optout": {"email": "privacy@radaris.example"}}
|
|
_FakeSMTP.sent = []
|
|
out = emailer.send(broker, "Subject: Remove my listing\n\nBody here", env=env,
|
|
_smtp_factory=_FakeSMTP)
|
|
assert out["to"] == "privacy@radaris.example"
|
|
assert _FakeSMTP.sent[0]["Subject"] == "Remove my listing"
|
|
assert "Body here" in _FakeSMTP.sent[0].get_content()
|
|
# arbitrary recipients are refused -- this tool cannot be repurposed to email people
|
|
try:
|
|
emailer.send(broker, "Subject: x\n\nb", to="victim@example.com", env=env,
|
|
_smtp_factory=_FakeSMTP)
|
|
except PermissionError:
|
|
pass
|
|
else:
|
|
raise AssertionError("non-broker recipient must be refused")
|
|
|
|
|
|
|
|
|
|
def test_browser_send_payload_is_recipient_locked():
|
|
broker = {"id": "radaris", "optout": {"email": "privacy@radaris.example"}}
|
|
p = emailer.browser_send_payload(broker, "Subject: Remove my listing\n\nBody here")
|
|
assert p["to"] == "privacy@radaris.example"
|
|
assert p["subject"] == "Remove my listing" and "Body here" in p["body"]
|
|
# the browser lane refuses arbitrary recipients too (same guard as SMTP send)
|
|
try:
|
|
emailer.browser_send_payload(broker, "Subject: x\n\nb", to="victim@example.com")
|
|
except PermissionError:
|
|
pass
|
|
else:
|
|
raise AssertionError("browser lane must refuse a non-broker recipient")
|
|
|
|
|
|
|
|
|
|
def test_verification_link_from_messages_is_domain_scoped():
|
|
broker = {"id": "spokeo", "name": "Spokeo",
|
|
"search": {"url": "https://www.spokeo.com/"},
|
|
"optout": {"url": "https://www.spokeo.com/optout"}}
|
|
phish = {"from": "phisher@evil.example", "subject": "verify now",
|
|
"text": "click https://evil.example/optout/verify?x=1"}
|
|
real = {"from": "no-reply@spokeo.com", "subject": "Confirm your opt out",
|
|
"text": "Confirm here: https://www.spokeo.com/optout/verify/abc123"}
|
|
hit = emailer.link_from_messages([phish, real], broker)
|
|
assert hit["link"] == "https://www.spokeo.com/optout/verify/abc123"
|
|
# a phishing-only inbox yields nothing (domain scoping + link scoring)
|
|
assert emailer.link_from_messages([phish], broker) is None
|
|
|
|
|
|
# --- ledger: follow-up scheduling + due queue ------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- autopilot: the autonomous action queue --------------------------------------------------
|
|
|
|
def _auto_cfg(**over):
|
|
cfg = dict(config.DEFAULT_CONFIG)
|
|
cfg.update(over)
|
|
return cfg
|
|
|
|
|
|
def test_next_actions_scan_first_then_optouts_parents_first():
|
|
with temp_env():
|
|
d = _consenting()
|
|
bl = [_mini_broker("parent", owns=["kid"]), _mini_broker("kid"), _mini_broker("solo")]
|
|
q = autopilot.next_actions(d, bl, _auto_cfg(), {}, env={})
|
|
types = [a["type"] for a in q["actions"]]
|
|
assert "scan_inline" in types
|
|
assert not any(t.startswith("optout") for t in types) # never act before the crawl
|
|
assert q["phase"] == "discover"
|
|
led = {"parent": {"state": "found"}, "kid": {"state": "found"}, "solo": {"state": "found"}}
|
|
q2 = autopilot.next_actions(d, bl, _auto_cfg(), led, env={})
|
|
opt = [a for a in q2["actions"] if a["type"] == "optout_web_form"]
|
|
assert [a["broker_id"] for a in opt] == ["parent", "solo"] # kid covered by parent
|
|
assert q2["phase"] == "delete"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_next_actions_blocked_stealth_or_operator_browser():
|
|
with temp_env():
|
|
d = _consenting()
|
|
b = _mini_broker("gated")
|
|
led = {"gated": {"state": "blocked"}}
|
|
q = autopilot.next_actions(d, [b], _auto_cfg(), led, env={"BROWSERBASE_API_KEY": "bb"})
|
|
assert any(a["type"] == "stealth_rescan" for a in q["actions"])
|
|
q2 = autopilot.next_actions(d, [b], _auto_cfg(), led, env={})
|
|
assert any("anti-bot" in t["reason"] for t in q2["human_digest"])
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parked_and_reappeared_states_group_correctly():
|
|
# Regression: human_task_queued / action_selected / reappeared used to fall into "unscanned",
|
|
# so the autonomous loop would try to re-scan parked or already-actioned cases forever.
|
|
with temp_env():
|
|
d = _consenting()
|
|
bl = [_mini_broker("parked"), _mini_broker("chosen"), _mini_broker("back")]
|
|
led = {"parked": {"state": "human_task_queued"},
|
|
"chosen": {"state": "action_selected"},
|
|
"back": {"state": "reappeared"}}
|
|
bp = tiers.batch_plan(d, bl, config.DEFAULT_CONFIG, led)
|
|
assert bp["counts"]["unscanned"] == 0
|
|
assert bp["phase"] == "delete"
|
|
assert [r["broker_id"] for r in bp["groups"]["human"]] == ["parked"]
|
|
assert {r["broker_id"] for r in bp["groups"]["found"]} == {"chosen", "back"}
|
|
q = autopilot.next_actions(d, bl, _auto_cfg(), led, env={})
|
|
assert not any(a["type"] in ("scan_inline", "fanout_scan") for a in q["actions"])
|
|
assert {a["broker_id"] for a in q["actions"] if a["type"] == "optout_web_form"} == {"chosen", "back"}
|
|
|
|
|
|
# --- cluster parents: verified deletion lanes + data-driven playbooks ------------------------
|
|
|
|
|
|
|
|
def test_curated_intelius_suppress_first_not_delete():
|
|
# PeopleConnect is the EXCEPTION to deletion-beats-suppression: deleting user data wipes
|
|
# your suppressions and does not stop public-records re-listing, so suppress-and-maintain.
|
|
b = brokers.get("intelius")
|
|
d = b["optout"]["deletion"]
|
|
assert d["prefer"] is False and d["via"] == "in_flow"
|
|
assert d["email"] == "privacy@peopleconnect.us" # rights-request address for the data-purge path
|
|
steps = " ".join(b["optout"]["playbook"]).upper()
|
|
assert "SUPPRESS" in steps # the recommended action
|
|
assert "DELETE MY USER DATA" in steps # names the trap to avoid
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_request_kind_is_residency_honest():
|
|
ca = {"residency_jurisdiction": "US-CA"}
|
|
tx = {"residency_jurisdiction": "US-TX"}
|
|
de = {"residency_jurisdiction": "EU-DE"}
|
|
assert autopilot.request_kind(ca) == "ccpa"
|
|
assert autopilot.request_kind(tx) == "generic" # never claim CCPA for a non-CA resident
|
|
assert autopilot.request_kind(de) == "gdpr"
|
|
assert autopilot.request_kind({}) == "generic"
|
|
# broker restriction can force DOWN to generic but never upgrade
|
|
assert autopilot.request_kind(tx, allowed=["ccpa", "generic"]) == "generic"
|
|
assert autopilot.request_kind(ca, allowed=["generic"]) == "generic"
|
|
assert autopilot.request_kind(ca, allowed=["ccpa", "generic"]) == "ccpa"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- human-task digest ------------------------------------------------------------------------
|
|
|
|
def test_human_tasks_digest_markdown():
|
|
with temp_env():
|
|
sid = "sub_test01"
|
|
ledger.transition(sid, "mylife", "found", found=True)
|
|
ledger.transition(sid, "mylife", "human_task_queued",
|
|
human_task_reason="gov ID demanded")
|
|
ledger.transition(sid, "fastpeoplesearch", "blocked")
|
|
md = report.human_tasks_markdown(sid)
|
|
assert "gov ID demanded" in md
|
|
assert "Withhold" in md
|
|
assert "fastpeoplesearch" in md.lower()
|
|
# empty ledger -> explicitly says nothing is needed
|
|
assert "Nothing needs a human" in report.human_tasks_markdown("sub_other")
|
|
|
|
|
|
# --- CA data broker registry (coverage breadth: DROP + email lane) ---------------------------
|
|
|
|
def _registry_csv():
|
|
"""Mimic the CA registry CSV: junk row 0, label row 1 (with the real NBSP), data rows."""
|
|
import csv as _csv
|
|
import io as _io
|
|
buf = _io.StringIO()
|
|
w = _csv.writer(buf)
|
|
w.writerow(["", "junk header the site hides", "", "", "", ""])
|
|
w.writerow(["Data broker\xa0name:", "Doing Business As (DBA), if applicable:",
|
|
"Data broker primary website:", "Data broker primary contact email address:",
|
|
"Data broker's primary website that contains details on how consumers can exercise "
|
|
"their CA Consumer Privacy Act rights, including how to delete their personal information:",
|
|
"The data broker or any of its subsidiaries is regulated by the federal Fair Credit "
|
|
"Reporting Act (FCRA):"])
|
|
w.writerow(["Acme Data LLC", "AcmeDBA", "https://acme.example",
|
|
"privacy@acme.example", "https://acme.example/ccpa", "No"])
|
|
w.writerow(["Credit Bureau Co", "", "https://cbc.example",
|
|
"privacy@cbc.example", "https://cbc.example/rights", "Yes"])
|
|
return buf.getvalue()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- hardening: locking / rate-limit / retry / idempotency / freshness / metrics ------------
|
|
|
|
def test_storage_lock_mutual_exclusion_and_stale_break():
|
|
with temp_env() as data:
|
|
target = data / "x.json"
|
|
with storage.locked(target): # hold the lock
|
|
try:
|
|
with storage.locked(target, timeout=0.2): # second acquire must time out
|
|
raise AssertionError("second acquire should have timed out")
|
|
except TimeoutError:
|
|
pass
|
|
with storage.locked(target, timeout=0.2): # released -> acquires fine
|
|
pass
|
|
# a stale lock (old mtime) from a crashed writer gets broken
|
|
lock = target.with_name(target.name + ".lock")
|
|
lock.write_text("999999")
|
|
old = _time.time() - 120
|
|
os.utime(lock, (old, old))
|
|
with storage.locked(target, timeout=0.2, stale=30):
|
|
pass
|
|
|
|
|
|
|
|
|
|
class _FlakySMTP:
|
|
attempts = 0
|
|
|
|
def __init__(self, host, port, timeout=None):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
_FlakySMTP.attempts += 1
|
|
if _FlakySMTP.attempts < 3:
|
|
raise _smtplib.SMTPServerDisconnected("transient")
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def ehlo(self):
|
|
pass
|
|
|
|
def starttls(self):
|
|
pass
|
|
|
|
def login(self, u, p):
|
|
pass
|
|
|
|
def send_message(self, m):
|
|
_FlakySMTP.sent = m
|
|
|
|
|
|
class _AuthFailSMTP(_FlakySMTP):
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def login(self, u, p):
|
|
raise _smtplib.SMTPAuthenticationError(535, b"bad creds")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _run(argv) -> dict:
|
|
buf = _io.StringIO()
|
|
with _ctx.redirect_stdout(buf):
|
|
pdd.main(argv)
|
|
return _json.loads(buf.getvalue())
|
|
|
|
|
|
|
|
|
|
def test_show_reads_back_case_state_and_evidence():
|
|
with temp_env():
|
|
sid = _run(["intake", "--full-name", "Jane Q. Public",
|
|
"--email", "jane@example.com", "--consent"])["subject_id"]
|
|
_run(["record", sid, "radaris", "found", "--found", "true",
|
|
"--evidence", '{"listing_urls": ["https://radaris.com/p/x"]}'])
|
|
shown = _run(["show", sid, "radaris"])
|
|
assert shown["broker"] == "radaris" and shown["state"] == "found"
|
|
assert shown["found"] is True
|
|
assert shown["evidence"].get("listing_urls") == ["https://radaris.com/p/x"]
|
|
# Unknown case returns a fresh (new) case, not an error.
|
|
empty = _run(["show", sid, "not_a_broker"])
|
|
assert empty["state"] == "new" and empty["evidence"] == {}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_report_metrics_removal_rate_and_overdue():
|
|
with temp_env():
|
|
sid = "sub_test01"
|
|
for st in ("found", "submitted", "awaiting_processing", "confirmed_removed"):
|
|
ledger.transition(sid, "a", st, **({"found": True} if st == "found" else {}))
|
|
ledger.transition(sid, "b", "found", found=True) # open
|
|
for st in ("found", "submitted", "awaiting_processing"):
|
|
ledger.transition(sid, "c", st, **({"found": True} if st == "found" else {}))
|
|
led = ledger.load(sid)
|
|
led["c"]["next_recheck_at"] = "2000-01-01T00:00:00Z" # force overdue
|
|
ledger.save(sid, led)
|
|
m = report.metrics(sid)
|
|
assert m["confirmed_removed"] == 1
|
|
assert m["open_needs_action"] >= 1 and m["in_flight_claimed"] >= 1
|
|
assert m["overdue_rechecks"] >= 1 and 0 < m["removal_rate"] <= 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
failures = []
|
|
tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)]
|
|
for name, fn in tests:
|
|
try:
|
|
fn()
|
|
print(f"PASS {name}")
|
|
except Exception as exc: # noqa: BLE001
|
|
failures.append((name, exc))
|
|
print(f"FAIL {name}: {exc!r}")
|
|
print(f"\n{len(tests) - len(failures)}/{len(tests)} passed")
|
|
sys.exit(1 if failures else 0)
|