forked from retoor/devplacepy
retoorandClaude Opus 5
7e37122f9f
An instance kept production-identical for extended manual testing is otherwise taxed forever by its own safety controls: five consents, a versioned terms gate on every mutating request, and every account predating the trust and safety commit reading terms_version NULL because init_db deliberately never backfills it. AcceptanceService grants each agreement to each account that has not declined it, so the instance stays production byte for byte while nobody clicks the same dialog again. It is opt-in, dry run by default, and one switch per agreement type. The application is not allowed to know it exists. One registration line in main.py is the only import anywhere, there is no route, schema, template, Devii tool or environment flag, and a unit test greps the tree and fails the suite if a second importer appears. The decline register needs no storage: the ledger is append-only in effect, the service only ever grants, so any withdrawn row was written by a human and that pair is never touched again. No provenance column, nothing to observe. Satisfaction is the gate's own expression, never a proxy, which is why the ordering is created_at then id exactly as consent_state selects, and why the live-account clauses are built with has_column: init_db ensures terms_version and deletion_requested_at but not is_active, so a hardcoded reference raises no such column on an instance where nobody was ever suspended. Every write is one conditional statement decided on the real rowcount, proven with sixteen processes racing one account to exactly one ledger row and one audit row. The two existing audit keys carry it, with actor kind service, because a service that silently mutated consent state would be the worst possible exception to the append-only rule. lensfl.md is the source brief accept.md records the design against. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
167 lines
5.1 KiB
Python
167 lines
5.1 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
import pytest
|
|
|
|
from devplacepy.database import CONSENTS_TABLE, get_setting, get_table, set_setting
|
|
from devplacepy.services.acceptance.agreements import agreement_for
|
|
from devplacepy.services.acceptance.pending import (
|
|
LIVE_ACCOUNT_CLAUSES,
|
|
current_version,
|
|
live_account_clauses,
|
|
pending,
|
|
)
|
|
from devplacepy.utils import generate_uid
|
|
|
|
TERMS = agreement_for("terms")
|
|
PRIVACY = agreement_for("privacy")
|
|
AI = agreement_for("ai_third_party")
|
|
|
|
|
|
@pytest.fixture
|
|
def account(local_db):
|
|
uid = generate_uid()
|
|
get_table("users").insert(
|
|
{
|
|
"uid": uid,
|
|
"username": f"pendingprobe_{uid[:8]}",
|
|
"email": f"pendingprobe_{uid[:8]}@example.com",
|
|
"password_hash": "x",
|
|
"role": "Member",
|
|
"terms_version": "",
|
|
"terms_accepted_at": "",
|
|
"deletion_requested_at": "",
|
|
"created_at": "2020-01-01T00:00:00",
|
|
}
|
|
)
|
|
yield uid
|
|
get_table("users").delete(uid=uid)
|
|
get_table(CONSENTS_TABLE).delete(owner_id=uid)
|
|
|
|
|
|
@pytest.fixture
|
|
def versions():
|
|
before = {
|
|
"terms_version": get_setting("terms_version", "1"),
|
|
"privacy_version": get_setting("privacy_version", "1"),
|
|
}
|
|
yield
|
|
for key, value in before.items():
|
|
set_setting(key, value)
|
|
|
|
|
|
def write_consent(uid, kind, state, version):
|
|
get_table(CONSENTS_TABLE).insert(
|
|
{
|
|
"uid": generate_uid(),
|
|
"owner_kind": "user",
|
|
"owner_id": uid,
|
|
"kind": kind,
|
|
"version": version,
|
|
"state": state,
|
|
"granted_at": "2021-01-01T00:00:00" if state == "granted" else "",
|
|
"withdrawn_at": "" if state == "granted" else "2021-01-01T00:00:00",
|
|
"created_at": "2021-01-01T00:00:00",
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
|
|
|
|
def is_pending(agreement, uid):
|
|
return uid in {row["uid"] for row in pending(agreement, 5000)}
|
|
|
|
|
|
def test_an_account_that_never_accepted_is_pending(account):
|
|
assert is_pending(TERMS, account)
|
|
assert is_pending(AI, account)
|
|
|
|
|
|
def test_a_null_gate_column_is_pending_not_invisible(account, versions):
|
|
set_setting("terms_version", "1")
|
|
get_table("users").update({"uid": account, "terms_version": None}, ["uid"])
|
|
assert is_pending(TERMS, account)
|
|
|
|
|
|
def test_a_satisfied_gate_column_is_not_pending(account, versions):
|
|
set_setting("terms_version", "7")
|
|
get_table("users").update({"uid": account, "terms_version": "7"}, ["uid"])
|
|
assert not is_pending(TERMS, account)
|
|
|
|
|
|
def test_a_stale_gate_column_is_pending_again(account, versions):
|
|
set_setting("terms_version", "7")
|
|
get_table("users").update({"uid": account, "terms_version": "7"}, ["uid"])
|
|
set_setting("terms_version", "8")
|
|
assert is_pending(TERMS, account)
|
|
|
|
|
|
def test_a_withdrawal_removes_the_account_permanently(account, versions):
|
|
set_setting("terms_version", "1")
|
|
write_consent(account, "terms", "withdrawn", "1")
|
|
assert not is_pending(TERMS, account)
|
|
set_setting("terms_version", "2")
|
|
assert not is_pending(TERMS, account)
|
|
get_table("users").update({"uid": account, "terms_version": None}, ["uid"])
|
|
assert not is_pending(TERMS, account)
|
|
|
|
|
|
def test_an_unversioned_agreement_is_satisfied_by_a_granted_row(account):
|
|
write_consent(account, "ai_third_party", "granted", "1")
|
|
assert not is_pending(AI, account)
|
|
|
|
|
|
def test_a_versioned_ledger_agreement_tracks_the_policy_version(account, versions):
|
|
set_setting("privacy_version", "3")
|
|
write_consent(account, "privacy", "granted", "3")
|
|
assert not is_pending(PRIVACY, account)
|
|
set_setting("privacy_version", "4")
|
|
assert is_pending(PRIVACY, account)
|
|
|
|
|
|
def test_a_legacy_consent_row_without_a_version_is_stale(account, versions):
|
|
set_setting("privacy_version", "1")
|
|
write_consent(account, "privacy", "granted", None)
|
|
assert is_pending(PRIVACY, account)
|
|
|
|
|
|
def test_a_deleted_account_is_never_pending(account):
|
|
get_table("users").update(
|
|
{"uid": account, "deletion_requested_at": "2024-01-01T00:00:00"}, ["uid"]
|
|
)
|
|
assert not is_pending(TERMS, account)
|
|
|
|
|
|
def test_a_deactivated_account_is_never_pending(account):
|
|
get_table("users").update({"uid": account, "is_active": False}, ["uid"])
|
|
assert not is_pending(TERMS, account)
|
|
|
|
|
|
def test_the_batch_size_bounds_the_sweep_and_the_order_is_stable(account):
|
|
first = [row["uid"] for row in pending(TERMS, 1)]
|
|
second = [row["uid"] for row in pending(TERMS, 1)]
|
|
assert len(first) <= 1
|
|
assert first == second
|
|
|
|
|
|
def test_a_zero_batch_size_still_returns_at_most_one_row(account):
|
|
assert len(pending(TERMS, 0)) <= 1
|
|
|
|
|
|
def test_the_live_account_clauses_only_name_existing_columns(local_db):
|
|
users = get_table("users")
|
|
expected = [
|
|
clause
|
|
for column, clause in LIVE_ACCOUNT_CLAUSES
|
|
if users.has_column(column)
|
|
]
|
|
assert live_account_clauses() == expected
|
|
|
|
|
|
def test_an_empty_version_setting_falls_back_to_one(versions):
|
|
set_setting("terms_version", "")
|
|
assert current_version(TERMS) == "1"
|
|
|
|
|
|
def test_an_unversioned_agreement_reports_version_one():
|
|
assert current_version(AI) == "1"
|