90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
"""Optional supplied Cassandra environment; never install/start a server here."""
|
|
|
|
import os
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from adcloud_server.database import Database
|
|
from adcloud_server.identity_models import ProfilePatch, StitchRequest
|
|
from adcloud_server.identity_service import IdentityService
|
|
from adcloud_server.profile_store import CassandraProfileStore
|
|
from adcloud_server.projection import ProfileProjector
|
|
from adcloud_server.settings import Settings
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not os.getenv("ADCLOUD_TEST_CASSANDRA_HOSTS"),
|
|
reason="No Cassandra integration cluster configured",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def components(tmp_path):
|
|
settings = Settings(
|
|
database_url=os.getenv(
|
|
"ADCLOUD_TEST_POSTGRES_URL", "sqlite:///" + str(tmp_path / "graph.db")
|
|
),
|
|
profile_backend="cassandra",
|
|
tenant_id="test-" + str(uuid4()),
|
|
identity_secret="test-cassandra-" + "x" * 48,
|
|
cassandra_hosts=tuple(os.environ["ADCLOUD_TEST_CASSANDRA_HOSTS"].split(",")),
|
|
cassandra_port=int(os.getenv("ADCLOUD_TEST_CASSANDRA_PORT", "9042")),
|
|
cassandra_keyspace=os.environ["ADCLOUD_TEST_CASSANDRA_KEYSPACE"],
|
|
cassandra_username=os.getenv("ADCLOUD_TEST_CASSANDRA_USERNAME", ""),
|
|
cassandra_password=os.getenv("ADCLOUD_TEST_CASSANDRA_PASSWORD", ""),
|
|
cassandra_tls=os.getenv("ADCLOUD_TEST_CASSANDRA_TLS", "true").lower() == "true",
|
|
cassandra_ca_file=os.getenv("ADCLOUD_TEST_CASSANDRA_CA_FILE") or None,
|
|
)
|
|
db = Database(settings.database_url, settings.tenant_id)
|
|
db.initialize()
|
|
store = CassandraProfileStore(settings)
|
|
try:
|
|
yield IdentityService(settings, db, store), ProfileProjector(db, store), store
|
|
finally:
|
|
store.close()
|
|
db.close()
|
|
|
|
|
|
def test_cassandra_sparse_snapshot_roundtrip(components):
|
|
service, _worker, store = components
|
|
master = str(uuid4())
|
|
document = {
|
|
"attributes": {"nested": {"interest": "running"}, "enabled": True, "score": 0},
|
|
"tags": ["runner"],
|
|
}
|
|
store.put(service.tenant, master, 1, document)
|
|
assert store.get(service.tenant, master, 1) == document
|
|
assert store.get(service.tenant, master, 2) is None
|
|
|
|
|
|
def test_cassandra_immutable_revision_prevents_stale_attribute_resurrection(components):
|
|
service, _worker, store = components
|
|
master = str(uuid4())
|
|
old = {"attributes": {"remove": "old"}, "tags": []}
|
|
new = {"attributes": {}, "tags": []}
|
|
store.put(service.tenant, master, 2, new)
|
|
store.put(service.tenant, master, 1, old)
|
|
assert store.get(service.tenant, master, 2) == new
|
|
|
|
|
|
def test_cassandra_worker_publishes_stitched_profile(components):
|
|
service, worker, _store = components
|
|
anon = service.anonymous()
|
|
stitched = service.stitch(
|
|
anon["anonymous_token"],
|
|
StitchRequest(
|
|
identifiers=[{"kind": "email", "value": "alice@example.com"}],
|
|
attributes={"interest": "running"},
|
|
),
|
|
)
|
|
assert service.profile(stitched["master_id"])["projection_status"] == "pending"
|
|
worker.flush()
|
|
current = service.profile(stitched["master_id"])
|
|
assert current["projection_status"] == "current"
|
|
assert current["attributes"] == {"interest": "running"}
|
|
service.patch_profile(
|
|
stitched["master_id"], ProfilePatch(attributes={"interest": None})
|
|
)
|
|
worker.flush()
|
|
assert service.profile(stitched["master_id"])["attributes"] == {}
|