diff --git a/.github/workflows/sdk-compliance.yml b/.github/workflows/sdk-compliance.yml index b77681ca..bb2a8593 100644 --- a/.github/workflows/sdk-compliance.yml +++ b/.github/workflows/sdk-compliance.yml @@ -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 diff --git a/sdk_compliance_adapter/CONTRIBUTING.md b/sdk_compliance_adapter/CONTRIBUTING.md index ffddda4b..e9fcc841 100644 --- a/sdk_compliance_adapter/CONTRIBUTING.md +++ b/sdk_compliance_adapter/CONTRIBUTING.md @@ -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: @@ -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 diff --git a/sdk_compliance_adapter/README.md b/sdk_compliance_adapter/README.md index 9efbee7c..795ed9e5 100644 --- a/sdk_compliance_adapter/README.md +++ b/sdk_compliance_adapter/README.md @@ -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 diff --git a/sdk_compliance_adapter/adapter.py b/sdk_compliance_adapter/adapter.py index 7a9ec437..31f5a6a3 100644 --- a/sdk_compliance_adapter/adapter.py +++ b/sdk_compliance_adapter/adapter.py @@ -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 @@ -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""" @@ -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 @@ -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], @@ -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", ] @@ -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 @@ -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 @@ -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}, " @@ -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: diff --git a/sdk_compliance_adapter/docker-compose.yml b/sdk_compliance_adapter/docker-compose.yml index 9281fa8e..0f0c9ac2 100644 --- a/sdk_compliance_adapter/docker-compose.yml +++ b/sdk_compliance_adapter/docker-compose.yml @@ -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 diff --git a/sdk_compliance_adapter/test_adapter.py b/sdk_compliance_adapter/test_adapter.py index 02eeac26..ad2e550b 100644 --- a/sdk_compliance_adapter/test_adapter.py +++ b/sdk_compliance_adapter/test_adapter.py @@ -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_")] == [