Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/sdk-compliance.yml
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ jobs:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}-${{ matrix.compression }}
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
env:
HARNESS_IMAGE: ghcr.io/posthog/sdk-test-harness:1.13.1
HARNESS_IMAGE: ghcr.io/posthog/sdk-test-harness:1.25.0
steps:
- name: Check out source code
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
Expand Down
4 changes: 2 additions & 2 deletions sdk_compliance_adapter/CONTRIBUTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ This package contains the PostHog Python SDK compliance adapter used with the Po

## Running tests

Tests run automatically in CI via GitHub Actions against harness **1.13.1**. The compliance job runs once for each capture compression (gzip, deflate and zstd). Each run opts into the local-evaluation suite and fails on compliance regressions.
Tests run automatically in CI via GitHub Actions against harness **1.25.0**. The compliance job runs once for each capture compression (gzip, deflate and zstd). Each run opts into the local-evaluation suite and fails on compliance regressions.

Run adapter protocol tests from the repository root in an activated virtual environment:

Expand Down Expand Up @@ -44,7 +44,7 @@ docker run -d --name sdk-adapter --network test-network -p 8080:8080 posthog-pyt
docker run --rm \
--name test-harness \
--network test-network \
ghcr.io/posthog/sdk-test-harness:1.13.1 \
ghcr.io/posthog/sdk-test-harness:1.25.0 \
run --adapter-url http://sdk-adapter:8080 --mock-url http://test-harness:8081

# Cleanup
Expand Down
2 changes: 1 addition & 1 deletion sdk_compliance_adapter/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ The adapter implements the standard SDK adapter interface defined in the [test h
### Local feature flag evaluation

The adapter advertises `feature_flags_local_evaluation_v1` for harness
**1.13.1**. The capability versions the adapter protocol and tests both legacy and
**1.25.0**. The capability versions the adapter protocol and tests both legacy and
explicit property matching; it does not change the SDK's default matching mode.

- `/init` maps optional `personal_api_key` to the SDK's `secret_key`. Ordinary
Expand Down
98 changes: 95 additions & 3 deletions sdk_compliance_adapter/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
It wraps the posthog-python SDK and exposes a REST API for the test harness to exercise.
"""

import json
import logging
import os
import threading
Expand Down Expand Up @@ -83,6 +84,8 @@ def __init__(self):
self.client: Optional[Client] = None
self.remote_client: Client | None = None
self.reload_thread: threading.Thread | None = None
self.reported_drops: List[Dict[str, Any]] = []
self.log_lines: List[str] = []

def reset(self):
"""Reset all state"""
Expand All @@ -101,7 +104,10 @@ def reset(self):
except Exception as e:
logger.warning(f"Error shutting down client: {e}")

# Cleared after shutdown, which can still report drops and log lines for the old client.
with self.lock:
self.reported_drops = []
self.log_lines = []
self.pending_events = 0
self.total_events_captured = 0
self.total_events_sent = 0
Expand Down Expand Up @@ -157,13 +163,77 @@ def get_state(self) -> Dict[str, Any]:
"total_retries": self.total_retries,
"last_error": self.last_error,
"requests_made": [r.to_dict() for r in self.requests_made],
"reported_drops": list(self.reported_drops),
"log_lines": list(self.log_lines),
}

def record_drops(self, error: Exception, batch: List[Dict[str, Any]]) -> None:
"""on_error hook: record each event capture dropped, by uuid."""
# A duplicate uuid has several names in batch order; capture keeps the first copy.
names: Dict[Optional[str], List[Optional[str]]] = {}
for event in batch or []:
names.setdefault(event.get("uuid"), []).append(event.get("event"))
duplicates_seen: Dict[str, int] = {}
with self.lock:
for uuid, details in getattr(error, "drops", []):
copies = names.get(uuid, [])
index = 0
if details == "duplicate_event_uuid":
index = duplicates_seen.get(uuid, 0) + 1
duplicates_seen[uuid] = index
name = copies[min(index, len(copies) - 1)] if copies else None
self.reported_drops.append(
{"uuid": uuid, "event": name, "details": details}
)

def record_log(self, line: str) -> None:
with self.lock:
self.log_lines.append(line)


# Global state
state = SDKState()


class _StateLogHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
state.record_log(self.format(record))


# The SDK logs under "posthog"; /state returns those lines (log_capture).
_sdk_logger = logging.getLogger("posthog")
_sdk_logger.addHandler(_StateLogHandler())
_sdk_logger.setLevel(logging.DEBUG)

_PADDING_KEY = "$test_padding"


def _padded_properties(
properties: Optional[Dict[str, Any]], size: int, marker: str
) -> Dict[str, Any]:
"""Add one ASCII padding property so the properties' compact JSON is exactly ``size`` bytes."""
padded = {**(properties or {}), _PADDING_KEY: ""}
fill = size - len(
json.dumps(padded, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
)
# The marker's bytes as written in JSON, escapes included, without its quotes.
marker_bytes = len(json.dumps(marker, ensure_ascii=False).encode("utf-8")) - 2
if fill < marker_bytes:
raise ValueError(
f"pad_properties_to_bytes {size} is smaller than the properties"
)
padded[_PADDING_KEY] = marker + "x" * (fill - marker_bytes)
return padded


def _before_send(set_properties: Dict[str, Any]):
def hook(event: Dict[str, Any]) -> Dict[str, Any]:
event.setdefault("properties", {}).update(set_properties)
return event

return hook


def patched_post_v1(
api_key: str,
host: Optional[str],
Expand Down Expand Up @@ -237,6 +307,15 @@ def health():
"capture_v1",
"capture_ai_v1",
"event_options",
"event_uuid_override",
"bounded_retries",
"disable_geoip",
"historical_migration",
"reported_drops",
"log_capture",
"generated_payloads",
"property_layers",
"fail_open",
"encoding_" + compression_name(),
"feature_flags_local_evaluation_v1",
]
Expand Down Expand Up @@ -272,7 +351,8 @@ def init():
disable_geoip = data.get("disable_geoip", False)
historical_migration = data.get("historical_migration", False)

if not api_key:
# An empty or personal key must reach the SDK, which disables itself (fail_open).
if api_key is None:
return jsonify({"error": "api_key is required"}), 400
if not host:
return jsonify({"error": "host is required"}), 400
Expand All @@ -299,7 +379,14 @@ def init():
"disable_geoip": disable_geoip,
"historical_migration": historical_migration,
"enable_local_evaluation": False,
"on_error": state.record_drops,
"super_properties": data.get("super_properties"),
}
before_send = data.get("before_send")
if before_send:
client_options["before_send"] = _before_send(
before_send.get("set_properties", {})
)
personal_api_key = data.get("personal_api_key")
client = Client(**client_options, secret_key=personal_api_key)
state.client = client
Expand All @@ -310,7 +397,7 @@ def init():
state.remote_client = Client(**client_options)

logger.info(
f"Initialized SDK with api_key={api_key[:10]}..., host={host}, "
f"Initialized SDK with api_key={api_key[:4]}..., host={host}, "
f"flush_at={flush_at}, flush_interval={flush_interval}, "
f"max_retries={max_retries}, compression={enable_compression}, "
f"disable_geoip={disable_geoip}, "
Expand Down Expand Up @@ -344,9 +431,14 @@ def _capture_with(capture_method_name: str):
if not event:
return jsonify({"error": "event is required"}), 400

properties = data.get("properties")
if data.get("pad_properties_to_bytes") is not None:
properties = _padded_properties(
properties, data["pad_properties_to_bytes"], data.get("marker") or ""
)
kwargs = {
"distinct_id": distinct_id,
"properties": data.get("properties"),
"properties": properties,
"options": data.get("options"),
}
if timestamp:
Expand Down
2 changes: 1 addition & 1 deletion sdk_compliance_adapter/docker-compose.yml
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ services:

# Test harness
test-harness:
image: ghcr.io/posthog/sdk-test-harness:1.13.1
image: ghcr.io/posthog/sdk-test-harness:1.25.0
command: ["run", "--adapter-url", "http://sdk-adapter:8080", "--mock-url", "http://test-harness:8081"]
networks:
- test-network
Expand Down
10 changes: 9 additions & 1 deletion sdk_compliance_adapter/test_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,15 @@ def test_health_and_init_use_one_codec(adapter, monkeypatch, env_value, codec):
monkeypatch.setenv("COMPRESSION", env_value)
capabilities = adapter.app.test_client().get("/health").json["capabilities"]
assert "feature_flags_local_evaluation_v1" in capabilities
assert {"capture_v1", "capture_ai_v1", "event_options"} <= set(capabilities)
assert {
"capture_v1",
"capture_ai_v1",
"event_options",
"event_uuid_override",
"bounded_retries",
"disable_geoip",
"historical_migration",
} <= set(capabilities)
assert "capture_v0" not in capabilities
assert "capture_ai_v0" not in capabilities
assert [c for c in capabilities if c.startswith("encoding_")] == [
Expand Down
Loading