435 lines
14 KiB
Python
435 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import gyxx_flow.adapters.acceptance_policy as acceptance_policy_module
|
|
from gyxx_flow.adapters import (
|
|
ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID,
|
|
WORKFLOW_ACCEPTANCE_RECIPIENT_OPEN_ID,
|
|
RuntimeIntegrationBinding,
|
|
WorkflowAcceptancePolicy,
|
|
WorkflowAcceptancePolicyError,
|
|
)
|
|
|
|
|
|
def _write_acceptance_evidence(path: str, worker: int, count: int) -> None:
|
|
environment = {
|
|
"GYXX_WORKFLOW_ACCEPTANCE": "1",
|
|
"GYXX_ACCEPTANCE_EVIDENCE_FILE": path,
|
|
"GYXX_WORKFLOW_ID": "content.metrics.daily",
|
|
"GYXX_RUN_ID": f"worker-{worker}",
|
|
}
|
|
os.environ.update(environment)
|
|
policy = WorkflowAcceptancePolicy.from_environment(environment)
|
|
for index in range(count):
|
|
policy.record(
|
|
"feishu_write_skipped",
|
|
operation="concurrency-test",
|
|
details={"worker": worker, "index": index},
|
|
)
|
|
|
|
|
|
def _binding(tmp_path: Path) -> RuntimeIntegrationBinding:
|
|
return RuntimeIntegrationBinding(
|
|
script_id="content_marketing:collector.py",
|
|
module="content_marketing",
|
|
entry="collector.py",
|
|
cdp_port=22998,
|
|
cdp_url="http://127.0.0.1:22998",
|
|
profile_dir=tmp_path / "profile",
|
|
cookie_file=tmp_path / "cookies.json",
|
|
storage_state_file=tmp_path / "storage_state.json",
|
|
)
|
|
|
|
|
|
def _environment(tmp_path: Path) -> dict[str, str]:
|
|
return {
|
|
"GYXX_WORKFLOW_ACCEPTANCE": "1",
|
|
"GYXX_DATA_ROOT": str(tmp_path),
|
|
"GYXX_WORKFLOW_ID": "content.metrics.daily",
|
|
"GYXX_RUN_ID": "acceptance-run-1",
|
|
}
|
|
|
|
|
|
def test_disabled_policy_does_not_change_recipients_or_writes(tmp_path: Path) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment({})
|
|
|
|
assert policy.notification_recipients(("ou_a", "ou_a", "ou_b")) == (
|
|
"ou_a",
|
|
"ou_b",
|
|
)
|
|
assert policy.skip_feishu_write("base.upsert") is False
|
|
assert policy.evidence_file is None
|
|
|
|
|
|
def test_acceptance_policy_is_fail_closed_and_forces_wang_yunlong(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
|
|
assert policy.notification_recipients(("ou_other", "oc_group")) == (
|
|
WORKFLOW_ACCEPTANCE_RECIPIENT_OPEN_ID,
|
|
)
|
|
assert policy.environment()["GYXX_FEISHU_TABLE_WRITE_DISABLED"] == "1"
|
|
assert policy.environment()["GYXX_COOKIE_INVALID_SKIP"] == "1"
|
|
|
|
with pytest.raises(WorkflowAcceptancePolicyError, match="Wang Yunlong"):
|
|
WorkflowAcceptancePolicy.from_environment(
|
|
{
|
|
**_environment(tmp_path),
|
|
"GYXX_NOTIFICATION_RECIPIENT_OPEN_ID": "ou_other",
|
|
}
|
|
)
|
|
with pytest.raises(WorkflowAcceptancePolicyError, match="writes"):
|
|
WorkflowAcceptancePolicy.from_environment(
|
|
{
|
|
**_environment(tmp_path),
|
|
"GYXX_FEISHU_TABLE_WRITE_DISABLED": "0",
|
|
}
|
|
)
|
|
|
|
|
|
def test_acceptance_policy_recognizes_snapshot_declared_analyzer_app(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(
|
|
{
|
|
**_environment(tmp_path),
|
|
"GYXX_NOTIFICATION_APP_PROFILE": "second-analyzer-app",
|
|
"GYXX_NOTIFICATION_HERMES_PROFILE": "data-analyzer",
|
|
}
|
|
)
|
|
|
|
assert policy.notification_recipients(
|
|
("ou_other",),
|
|
app_profile="second-analyzer-app",
|
|
) == (ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID,)
|
|
|
|
|
|
def test_feishu_write_skip_appends_run_scoped_evidence(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
environment = _environment(tmp_path)
|
|
policy = WorkflowAcceptancePolicy.from_environment(environment)
|
|
monkeypatch.setenv("GYXX_WORKFLOW_ID", environment["GYXX_WORKFLOW_ID"])
|
|
monkeypatch.setenv("GYXX_RUN_ID", environment["GYXX_RUN_ID"])
|
|
|
|
assert policy.skip_feishu_write(
|
|
"bitable.record-upsert",
|
|
details={"table": "tbl1", "api_token": "must-not-leak"},
|
|
)
|
|
|
|
assert policy.evidence_file is not None
|
|
payload = json.loads(policy.evidence_file.read_text(encoding="utf-8"))
|
|
assert payload["event"] == "feishu_write_skipped"
|
|
assert payload["workflow_id"] == "content.metrics.daily"
|
|
assert payload["run_id"] == "acceptance-run-1"
|
|
assert payload["details"]["api_token"] == "<redacted>"
|
|
|
|
|
|
def test_acceptance_evidence_recursively_redacts_nested_secrets(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
environment = _environment(tmp_path)
|
|
policy = WorkflowAcceptancePolicy.from_environment(environment)
|
|
monkeypatch.setenv("GYXX_WORKFLOW_ID", environment["GYXX_WORKFLOW_ID"])
|
|
monkeypatch.setenv("GYXX_RUN_ID", environment["GYXX_RUN_ID"])
|
|
credential_values = {"value": "not-a-real-" + "nested-secret-value"}
|
|
details = {
|
|
"request": {
|
|
"headers": {"access_" + "token": credential_values["value"]},
|
|
"items": [
|
|
{"pass" + "word": credential_values["value"]},
|
|
("safe", {"secret_" + "key": credential_values["value"]}),
|
|
],
|
|
},
|
|
"safe_value": "visible",
|
|
}
|
|
|
|
safe_details = acceptance_policy_module._safe_details(details)
|
|
assert isinstance(safe_details["request"]["items"][1], tuple)
|
|
|
|
policy.record(
|
|
"nested-redaction-test",
|
|
details=details,
|
|
)
|
|
|
|
assert policy.evidence_file is not None
|
|
text = policy.evidence_file.read_text(encoding="utf-8")
|
|
assert credential_values["value"] not in text
|
|
payload = json.loads(text)
|
|
assert payload["details"] == {
|
|
"request": {
|
|
"headers": {"access_token": "<redacted>"},
|
|
"items": [
|
|
{"password": "<redacted>"},
|
|
["safe", {"secret_key": "<redacted>"}],
|
|
],
|
|
},
|
|
"safe_value": "visible",
|
|
}
|
|
|
|
|
|
def test_acceptance_evidence_is_valid_under_multiprocess_writers(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
evidence_file = tmp_path / "concurrent-evidence.jsonl"
|
|
context = multiprocessing.get_context("spawn")
|
|
workers = [
|
|
context.Process(
|
|
target=_write_acceptance_evidence,
|
|
args=(str(evidence_file), worker, 25),
|
|
)
|
|
for worker in range(4)
|
|
]
|
|
|
|
for process in workers:
|
|
process.start()
|
|
for process in workers:
|
|
process.join(timeout=30)
|
|
|
|
assert [process.exitcode for process in workers] == [0, 0, 0, 0]
|
|
lines = evidence_file.read_text(encoding="utf-8").splitlines()
|
|
payloads = [json.loads(line) for line in lines]
|
|
assert len(payloads) == 100
|
|
assert {
|
|
(payload["details"]["worker"], payload["details"]["index"])
|
|
for payload in payloads
|
|
} == {(worker, index) for worker in range(4) for index in range(25)}
|
|
|
|
|
|
@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires POSIX fork")
|
|
def test_acceptance_evidence_resets_inherited_thread_lock_after_fork(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
evidence_file = tmp_path / "fork-evidence.jsonl"
|
|
locked = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def hold_parent_lock() -> None:
|
|
with acceptance_policy_module._EVIDENCE_THREAD_LOCK:
|
|
locked.set()
|
|
release.wait(timeout=10)
|
|
|
|
thread = threading.Thread(target=hold_parent_lock)
|
|
thread.start()
|
|
assert locked.wait(timeout=2)
|
|
context = multiprocessing.get_context("fork")
|
|
process = context.Process(
|
|
target=_write_acceptance_evidence,
|
|
args=(str(evidence_file), 1, 1),
|
|
)
|
|
try:
|
|
process.start()
|
|
process.join(timeout=5)
|
|
finally:
|
|
release.set()
|
|
thread.join(timeout=2)
|
|
if process.is_alive():
|
|
process.terminate()
|
|
process.join(timeout=2)
|
|
|
|
assert process.exitcode == 0
|
|
payload = json.loads(evidence_file.read_text(encoding="utf-8"))
|
|
assert payload["run_id"] == "worker-1"
|
|
|
|
|
|
def test_acceptance_evidence_thread_lock_has_a_timeout(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
monkeypatch.setattr(
|
|
acceptance_policy_module,
|
|
"_EVIDENCE_LOCK_TIMEOUT_SECONDS",
|
|
0.05,
|
|
)
|
|
lock = acceptance_policy_module._EVIDENCE_THREAD_LOCK
|
|
assert lock.acquire(timeout=1)
|
|
try:
|
|
with pytest.raises(WorkflowAcceptancePolicyError, match="this process"):
|
|
policy.record("lock-timeout-test")
|
|
finally:
|
|
lock.release()
|
|
|
|
|
|
def test_cookie_preflight_skips_missing_invalid_and_expired_state(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
binding = _binding(tmp_path)
|
|
|
|
missing = policy.preflight_cookie(binding, now_epoch=100.0)
|
|
assert missing.status == "SKIPPED_COOKIE"
|
|
assert missing.reason == "browser state is missing"
|
|
|
|
binding.cookie_file.write_text("not-json", encoding="utf-8")
|
|
invalid = policy.preflight_cookie(binding, now_epoch=100.0)
|
|
assert invalid.status == "SKIPPED_COOKIE"
|
|
assert invalid.reason == "cookie file is unreadable"
|
|
|
|
binding.cookie_file.write_text(
|
|
json.dumps([{"name": "session", "value": "x", "expires": 99.0}]),
|
|
encoding="utf-8",
|
|
)
|
|
expired = policy.preflight_cookie(binding, now_epoch=100.0)
|
|
assert expired.status == "SKIPPED_COOKIE"
|
|
assert expired.reason == "all cookies are expired"
|
|
|
|
|
|
def test_cookie_preflight_accepts_unexpired_or_profile_cookie_state(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
binding = _binding(tmp_path)
|
|
binding.cookie_file.write_text(
|
|
json.dumps([{"name": "session", "value": "x", "expires": 101.0}]),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert policy.preflight_cookie(binding, now_epoch=100.0).status == "READY"
|
|
|
|
binding.cookie_file.unlink()
|
|
binding.profile_dir.mkdir()
|
|
(binding.profile_dir / "Local State").write_text("{}", encoding="utf-8")
|
|
assert policy.preflight_cookie(binding, now_epoch=100.0).status == "SKIPPED_COOKIE"
|
|
|
|
cookie_db = binding.profile_dir / "Default" / "Network" / "Cookies"
|
|
cookie_db.parent.mkdir(parents=True)
|
|
with sqlite3.connect(cookie_db) as connection:
|
|
connection.execute("CREATE TABLE cookies (expires_utc INTEGER NOT NULL)")
|
|
connection.execute("INSERT INTO cookies VALUES (?)", (11_644_473_701_000_000,))
|
|
assert policy.preflight_cookie(binding, now_epoch=100.0).status == "READY"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
[{"name": "session", "value": "x", "domain": ".example.test"}],
|
|
{
|
|
"cookies": [
|
|
{"name": "session", "value": "x", "domain": ".example.test"}
|
|
]
|
|
},
|
|
{
|
|
"cookies": [
|
|
{"name": "session", "value": "x", "domain": ".example.test"}
|
|
],
|
|
"origins": [{"origin": "https://example.test", "localStorage": []}],
|
|
},
|
|
],
|
|
)
|
|
def test_cookie_preflight_supports_cookie_bundles_and_storage_state(
|
|
tmp_path: Path,
|
|
payload: object,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
binding = replace(
|
|
_binding(tmp_path),
|
|
required_cookie_domains=("example.test",),
|
|
required_cookie_names=("session",),
|
|
)
|
|
binding.cookie_file.write_text(json.dumps(payload), encoding="utf-8")
|
|
|
|
assert policy.preflight_cookie(binding, now_epoch=100.0).status == "READY"
|
|
|
|
|
|
def test_cookie_preflight_rejects_an_unrelated_domain_or_cookie_name(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
binding = replace(
|
|
_binding(tmp_path),
|
|
required_cookie_domains=("target.example",),
|
|
required_cookie_names=("target_session",),
|
|
)
|
|
binding.cookie_file.write_text(
|
|
json.dumps(
|
|
[
|
|
{
|
|
"name": "unrelated_session",
|
|
"value": "x",
|
|
"domain": ".other.example",
|
|
}
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
wrong_domain = policy.preflight_cookie(binding, now_epoch=100.0)
|
|
assert wrong_domain.status == "SKIPPED_COOKIE"
|
|
assert "platform domain" in wrong_domain.reason
|
|
|
|
binding.cookie_file.write_text(
|
|
json.dumps(
|
|
[
|
|
{
|
|
"name": "unrelated_session",
|
|
"value": "x",
|
|
"domain": ".target.example",
|
|
}
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
wrong_name = policy.preflight_cookie(binding, now_epoch=100.0)
|
|
assert wrong_name.status == "SKIPPED_COOKIE"
|
|
assert "login cookie" in wrong_name.reason
|
|
|
|
|
|
@pytest.mark.parametrize("login_mode", ["B", "C"])
|
|
def test_credential_capable_modes_do_not_require_cookie_when_credentials_exist(
|
|
tmp_path: Path,
|
|
login_mode: str,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
account_name = "TARGET_" + "ACCOUNT"
|
|
password_name = "TARGET_" + "PASSWORD"
|
|
configured_value = "config" + "ured"
|
|
binding = replace(
|
|
_binding(tmp_path),
|
|
login_mode=login_mode,
|
|
required_cookie_domains=("target.example",),
|
|
credential_env_names=(account_name, password_name),
|
|
)
|
|
|
|
missing = policy.preflight_cookie(binding, now_epoch=100.0, environment={})
|
|
assert missing.status == "SKIPPED_COOKIE"
|
|
|
|
available = policy.preflight_cookie(
|
|
binding,
|
|
now_epoch=100.0,
|
|
environment={
|
|
account_name: configured_value,
|
|
password_name: configured_value,
|
|
},
|
|
)
|
|
assert available.status == "READY"
|
|
assert available.reason == "credential login fallback is available"
|
|
|
|
|
|
def test_interactive_login_mode_is_always_skipped_during_acceptance(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
policy = WorkflowAcceptancePolicy.from_environment(_environment(tmp_path))
|
|
binding = replace(_binding(tmp_path), login_mode="D")
|
|
binding.cookie_file.write_text(
|
|
json.dumps([{"name": "session", "value": "x"}]),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = policy.preflight_cookie(binding, now_epoch=100.0)
|
|
|
|
assert result.status == "SKIPPED_COOKIE"
|
|
assert "interactive login" in result.reason
|