codex-bot
0fc6c68a99
Issue persistent signed anonymous IDs and accept cookie/token-bound events. Resolve verified email, phone, OAuth, and namespaced third-party IDs into transactionally merged master identities with historical event aliases. Protect shared-device account switching, tenant isolation, and event retries. Store sparse canonical profiles as PostgreSQL JSONB or local SQLite JSON. Support an optional versioned Cassandra wide-column projection through a transactional outbox, retry worker, and immediate canonical read fallback. Document deployment settings and provide opt-in database integration tests. Validation: 43 local tests passed; 6 external database tests skipped. Ruff checks, formatting, and offline uv lock verification passed.
106 lines
3.6 KiB
Python
106 lines
3.6 KiB
Python
"""Optional real PostgreSQL integration: set ADCLOUD_TEST_POSTGRES_URL."""
|
|
|
|
import os
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from adcloud_server.database import Database
|
|
from adcloud_server.identity_models import EventRequest, ProfilePatch, StitchRequest
|
|
from adcloud_server.identity_service import IdentityService
|
|
from adcloud_server.settings import Settings
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not os.getenv("ADCLOUD_TEST_POSTGRES_URL"),
|
|
reason="No PostgreSQL integration URL configured",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def service():
|
|
settings = Settings(
|
|
database_url=os.environ["ADCLOUD_TEST_POSTGRES_URL"],
|
|
tenant_id="test-" + str(uuid4()),
|
|
identity_secret="postgres-test-" + "x" * 48,
|
|
profile_backend="cassandra",
|
|
)
|
|
db = Database(settings.database_url, settings.tenant_id)
|
|
db.initialize()
|
|
try:
|
|
yield IdentityService(settings, db)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_postgres_jsonb_stitch_and_immediate_history(service):
|
|
known = service.anonymous()
|
|
identity = StitchRequest(
|
|
identifiers=[{"kind": "email", "value": "alice@example.com"}]
|
|
)
|
|
service.stitch(known["anonymous_token"], identity)
|
|
visitor = service.anonymous(rotate=True)
|
|
service.patch_profile(
|
|
visitor["master_id"],
|
|
ProfilePatch(
|
|
attributes={"nested": {"interest": "running"}, "absent": None},
|
|
tags=["runner"],
|
|
),
|
|
)
|
|
event = service.record_event(
|
|
visitor["anonymous_token"], EventRequest(name="product.view")
|
|
)
|
|
stitched = service.stitch(visitor["anonymous_token"], identity)
|
|
assert stitched["master_id"] == known["master_id"]
|
|
assert service.profile(known["master_id"])["attributes"] == {
|
|
"nested": {"interest": "running"}
|
|
}
|
|
assert service.history(visitor["master_id"])["items"][0]["id"] == event["event_id"]
|
|
with service.db.transaction() as conn:
|
|
result = conn.execute(
|
|
text(
|
|
"SELECT pg_typeof(document)::text FROM identity_profiles WHERE tenant_id=:tenant LIMIT 1"
|
|
),
|
|
{"tenant": service.tenant},
|
|
).scalar_one()
|
|
assert result == "jsonb"
|
|
|
|
|
|
def test_postgres_multiple_process_connections_serialize_identity_merges(service):
|
|
second_db = Database(service.settings.database_url, service.tenant)
|
|
other = IdentityService(service.settings, second_db)
|
|
tokens = [
|
|
service.anonymous()["anonymous_token"],
|
|
other.anonymous()["anonymous_token"],
|
|
]
|
|
instances = [service, other]
|
|
request = StitchRequest(identifiers=[{"kind": "phone", "value": "+14155552671"}])
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(
|
|
pool.map(lambda i: instances[i].stitch(tokens[i], request), [0, 1])
|
|
)
|
|
assert results[0]["master_id"] == results[1]["master_id"]
|
|
assert len(service.graph(results[0]["master_id"])["identifiers"]) == 1
|
|
finally:
|
|
second_db.close()
|
|
|
|
|
|
def test_postgres_failed_transaction_rolls_back_graph_and_outbox(service, monkeypatch):
|
|
visitor = service.anonymous()
|
|
|
|
def fail(*args):
|
|
raise OSError("Simulated outbox failure")
|
|
|
|
monkeypatch.setattr(service, "_queue", fail)
|
|
with pytest.raises(OSError):
|
|
service.stitch(
|
|
visitor["anonymous_token"],
|
|
StitchRequest(
|
|
identifiers=[{"kind": "email", "value": "alice@example.com"}]
|
|
),
|
|
)
|
|
assert service.graph(visitor["master_id"])["identifiers"] == []
|
|
assert service.profile(visitor["master_id"])["revision"] == 1
|