diff --git a/.github/workflows/sdk-compliance.yml b/.github/workflows/sdk-compliance.yml index ba4a057..b8b60d1 100644 --- a/.github/workflows/sdk-compliance.yml +++ b/.github/workflows/sdk-compliance.yml @@ -13,9 +13,47 @@ on: jobs: compliance: - name: PostHog SDK compliance tests + name: PostHog PHP compliance (${{ matrix.consumer }}) + strategy: + fail-fast: false + matrix: + include: + - consumer: lib_curl + dockerfile: sdk_compliance_adapter/Dockerfile + - consumer: socket + dockerfile: sdk_compliance_adapter/Dockerfile.socket + - consumer: fork_curl + dockerfile: sdk_compliance_adapter/Dockerfile.fork_curl uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@6d19abb9c81e2262dacbe340e7dddda9c871c178 with: - adapter-dockerfile: "sdk_compliance_adapter/Dockerfile" + adapter-dockerfile: ${{ matrix.dockerfile }} adapter-context: "." - test-harness-version: "0.10.0" + test-harness-version: "1.0.0" + report-name: sdk-compliance-${{ matrix.consumer }} + sdk-type: server + concurrency: 1 + continue-on-error: true + + report-completeness: + name: PHP report completeness (${{ matrix.consumer }}) + needs: compliance + if: always() + runs-on: ubuntu-latest + permissions: + contents: read + actions: read + strategy: + fail-fast: false + matrix: + consumer: [lib_curl, socket, fork_curl] + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - name: Test report checker + run: python3 -m unittest discover -s sdk_compliance_adapter -p 'test_check_report.py' -v + - name: Download profile report + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 + with: + name: sdk-compliance-${{ matrix.consumer }} + path: report + - name: Require the complete test inventory + run: python3 sdk_compliance_adapter/check_report.py "${{ matrix.consumer }}" report/sdk-compliance-report.md diff --git a/sdk_compliance_adapter/Dockerfile b/sdk_compliance_adapter/Dockerfile index e37fa73..6fadc8d 100644 --- a/sdk_compliance_adapter/Dockerfile +++ b/sdk_compliance_adapter/Dockerfile @@ -3,7 +3,7 @@ FROM php:8.3-cli WORKDIR /app RUN apt-get update \ - && apt-get install -y --no-install-recommends git unzip \ + && apt-get install -y --no-install-recommends git unzip curl gzip python3 \ && rm -rf /var/lib/apt/lists/* COPY --from=composer:2 /usr/bin/composer /usr/bin/composer @@ -16,6 +16,12 @@ RUN composer install --no-interaction --prefer-dist --no-dev --no-progress COPY sdk_compliance_adapter/ /app/sdk_compliance_adapter/ +RUN php -r 'exit(extension_loaded("curl") ? 0 : 1);' + +ENV POSTHOG_CONSUMER=lib_curl + +RUN PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s sdk_compliance_adapter -p 'test_*.py' -v + EXPOSE 8080 -CMD ["php", "/app/sdk_compliance_adapter/adapter.php"] +CMD ["python3", "/app/sdk_compliance_adapter/server.py"] diff --git a/sdk_compliance_adapter/Dockerfile.fork_curl b/sdk_compliance_adapter/Dockerfile.fork_curl new file mode 100644 index 0000000..9e46ef1 --- /dev/null +++ b/sdk_compliance_adapter/Dockerfile.fork_curl @@ -0,0 +1,27 @@ +FROM php:8.3-cli + +WORKDIR /app + +RUN apt-get update \ + && apt-get install -y --no-install-recommends git unzip curl gzip python3 \ + && rm -rf /var/lib/apt/lists/* + +COPY --from=composer:2 /usr/bin/composer /usr/bin/composer + +COPY composer.json composer.lock /app/ +COPY lib/ /app/lib/ +COPY bin/ /app/bin/ + +RUN composer install --no-interaction --prefer-dist --no-dev --no-progress + +COPY sdk_compliance_adapter/ /app/sdk_compliance_adapter/ + +RUN php -r 'exit(extension_loaded("curl") ? 0 : 1);' + +ENV POSTHOG_CONSUMER=fork_curl + +RUN PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s sdk_compliance_adapter -p 'test_*.py' -v + +EXPOSE 8080 + +CMD ["python3", "/app/sdk_compliance_adapter/server.py"] diff --git a/sdk_compliance_adapter/Dockerfile.socket b/sdk_compliance_adapter/Dockerfile.socket new file mode 100644 index 0000000..6b9258b --- /dev/null +++ b/sdk_compliance_adapter/Dockerfile.socket @@ -0,0 +1,27 @@ +FROM php:8.3-cli + +WORKDIR /app + +RUN apt-get update \ + && apt-get install -y --no-install-recommends git unzip curl gzip python3 \ + && rm -rf /var/lib/apt/lists/* + +COPY --from=composer:2 /usr/bin/composer /usr/bin/composer + +COPY composer.json composer.lock /app/ +COPY lib/ /app/lib/ +COPY bin/ /app/bin/ + +RUN composer install --no-interaction --prefer-dist --no-dev --no-progress + +COPY sdk_compliance_adapter/ /app/sdk_compliance_adapter/ + +RUN php -r 'exit(extension_loaded("curl") ? 0 : 1);' + +ENV POSTHOG_CONSUMER=socket + +RUN PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s sdk_compliance_adapter -p 'test_*.py' -v + +EXPOSE 8080 + +CMD ["python3", "/app/sdk_compliance_adapter/server.py"] diff --git a/sdk_compliance_adapter/README.md b/sdk_compliance_adapter/README.md index 09d83d6..1c4e30b 100644 --- a/sdk_compliance_adapter/README.md +++ b/sdk_compliance_adapter/README.md @@ -1,12 +1,131 @@ -# PostHog PHP SDK Compliance Adapter +# PostHog PHP SDK compliance adapter -This adapter wraps the PostHog PHP SDK for the PostHog SDK compliance test harness. +The adapter runs the repository's `PostHog\Client` with production capture consumers +and production `HttpClient` feature-flag requests. Harness 1.0.0 selects **30 server +capture V0 tests and 17 flag tests per profile**, including UTC timestamp overrides +and gzip. No tests are filtered; compliance assertions remain advisory in CI. -## Local run +## Profiles + +| `POSTHOG_CONSUMER` | Capture transport | Completion | +| --- | --- | --- | +| `lib_curl` (default) | SDK LibCurl + HttpClient | Verified synchronous HTTP | +| `socket` | SDK Socket | Synchronous socket response handling | +| `fork_curl` | SDK ForkCurl + system curl/gzip | Foreground, using existing `debug=true` | + +CI runs all three profiles with distinct report artifacts. An independent +`report-completeness` job checks each artifact against `expected_inventory.json` +(the harness 1.0.0 server V0 and flags IDs). Missing, empty, incomplete, duplicate, +or unexpected results and inconsistent summary counts fail this gate. Complete +reports with failing assertions still pass the inventory gate; SDK compliance +assertions remain advisory. The checker reads the pinned harness's Markdown result +tables, not diagnostic text or the reusable job's conclusion. + +All profiles use +`debug=true`, no local flag definitions/secret key, and SDK flag-called events. +Compression is enabled only when requested at `/init`. Flags use the SDK's own +uncompressed HTTP client even when capture compression is enabled. + +The adapter maps `flush_at` to `batch_size` and `flush_interval_ms` to +`flush_interval_seconds` (SDK default: 5 seconds, enforced on enqueue, not an idle +timer). `max_retries` maps to the existing backoff-duration option: for three +retries, 801 ms for HttpClient and 800 ms for Socket, reflecting their different +limit checks. ForkCurl does not implement that retry option. Socket's connection +timeout is expressed in seconds; the PHP worker's `default_socket_timeout=1` +bounds reads, including idle retry connections. This is a configured runtime +profile, not certification of default PHP socket timing. + +## Observation and isolation + +`server.py` exposes `/health`, `/init`, `/capture`, `/get_feature_flag`, `/flush`, +`/state`, and `/reset`. It starts one PHP SDK worker per initialization. SDK traffic +passes through a loopback TCP relay to the supplied **HTTP mock host**. The relay +forwards request and response bytes unchanged, once per connection, without +retries or response substitution. It parses copies only for request telemetry; +flag values and side-effect events come from SDK APIs. HTTPS/TLS is not covered. + +The public `before_send` callback returns the SDK-enriched event unchanged and +observes its UUID and capture count, including flag-called events. Capture input +omits UUID so generation happens inside `Client::capture`. Capture and flush +responses preserve the SDK's boolean result, including failures. + +State has explicit observation limits: + +- `pending_events` is `null` while initialized: there is no public queue-length + accessor. Captured counts are events observed by `before_send`, not proof of + enqueue or delivery. +- `requests_made` records HTTP responses observed at the relay, including flags. + Connections without an HTTP response are not represented. Retry indices group + identical request bytes within a single adapter action; separate flag calls + start independent sequences. +- `total_events_sent` counts events in observed HTTP 200 batches. + `events_flushed` counts those events only during that flush action, not over the + client lifetime. Neither counter overrides the SDK's success/failure result. + +Reset kills and waits for only the owned PHP worker, without running its +queue-flushing destructors. It then closes the relay. This provides test isolation, +not SDK shutdown certification; `/flush` always calls the real public SDK method. + +## Known failures and applicability + +LibCurl is expected to pass all 47 selected tests. Socket and foreground ForkCurl +each expose these 11 existing capture failures: + +- `capture.retry_behavior.retries_on_503` +- `capture.retry_behavior.retries_on_500` +- `capture.retry_behavior.retries_on_502` +- `capture.retry_behavior.retries_on_504` +- `capture.retry_behavior.respects_retry_after_header` +- `capture.retry_behavior.implements_backoff` +- `capture.retry_behavior.max_retries_respected` +- `capture.error_handling.retries_on_408` +- `capture.deduplication.preserves_uuid_on_retry` +- `capture.deduplication.preserves_uuid_and_timestamp_on_retry` +- `capture.deduplication.preserves_uuid_and_timestamp_on_batch_retry` + +Socket does not reset its write state on retry, treats 408 as terminal, and ignores +Retry-After. ForkCurl does not retry HTTP rejection and reports a successful curl +process exit as successful delivery even for HTTP errors. The adapter exposes +these behaviors rather than retrying on the SDK's behalf. + +Background ForkCurl and LibCurl's explicit fire-and-forget option do not promise +blocking verified delivery and are not selected. File/noop consumers are not HTTP +sinks; file replay is not covered. Capture V1, dedicated AI capture, and non-gzip +encodings are not supported capabilities of this adapter's SDK entry. + +## Running ```sh cd sdk_compliance_adapter -docker compose up --build --abort-on-container-exit --exit-code-from test-harness +POSTHOG_CONSUMER=socket docker compose up --build --abort-on-container-exit --exit-code-from test-harness +``` + +For native PHP 8.3, Python 3, curl and gzip, from the repository root: + +```sh +composer install --prefer-dist --no-progress +PORT=18270 PROXY_PORT=19271 BIND_HOST=127.0.0.1 POSTHOG_CONSUMER=lib_curl \ + python3 sdk_compliance_adapter/server.py +``` + +Run the harness against that adapter with a separate mock port. The adapter does +not support parallel test execution. `PORT` defaults to 8080 and `PROXY_PORT` to +8082 inside containers; choose unused loopback ports for concurrent local work. + +Adapter regression checks run during each Docker build and can also run locally: + +```sh +TEST_MOCK_PORT=19276 TEST_PROXY_PORT=19277 PYTHONDONTWRITEBYTECODE=1 \ + python3 -m unittest discover -s sdk_compliance_adapter -p 'test_*.py' -v +``` + +The report checker can also run independently without PHP or network access: + +```sh +python3 -m unittest discover -s sdk_compliance_adapter -p 'test_check_report.py' -v +python3 sdk_compliance_adapter/check_report.py socket report/sdk-compliance-report.md ``` -The adapter exposes the standard harness endpoints: `/health`, `/init`, `/capture`, `/flush`, `/state`, and `/reset`. +The adapter checks cover byte-preserving relay behavior, SDK-generated UUIDs, immediate +capture, production retries, compression plus flags, real SDK result booleans, +worker failure, and reset isolation. They use local mock traffic only. diff --git a/sdk_compliance_adapter/adapter.php b/sdk_compliance_adapter/adapter.php index 8d47f04..b19a0dc 100644 --- a/sdk_compliance_adapter/adapter.php +++ b/sdk_compliance_adapter/adapter.php @@ -6,89 +6,19 @@ require __DIR__ . '/../vendor/autoload.php'; use PostHog\Client; -use PostHog\HttpClient; -use PostHog\HttpResponse; -use PostHog\PostHog; -use PostHog\Uuid; - -final class RequestInfo -{ - public function __construct( - public int $timestampMs, - public int $statusCode, - public int $retryAttempt, - public int $eventCount, - public array $uuidList, - ) { - } - - public function toArray(): array - { - return [ - 'timestamp_ms' => $this->timestampMs, - 'status_code' => $this->statusCode, - 'retry_attempt' => $this->retryAttempt, - 'event_count' => $this->eventCount, - 'uuid_list' => $this->uuidList, - ]; - } -} final class AdapterState { public ?Client $client = null; public int $totalEventsCaptured = 0; - public int $totalEventsSent = 0; - public int $totalRetries = 0; + public ?string $lastUuid = null; public ?string $lastError = null; - /** @var list */ - public array $requestsMade = []; - public int $pendingEvents = 0; - public function reset(): void - { - if ($this->client !== null) { - $this->discardQueuedEvents($this->client); - try { - $this->client->shutdown(); - } catch (Throwable $e) { - error_log('[adapter] error shutting down client: ' . $e->getMessage()); - } - } - - $this->client = null; - $this->totalEventsCaptured = 0; - $this->totalEventsSent = 0; - $this->totalRetries = 0; - $this->lastError = null; - $this->requestsMade = []; - $this->pendingEvents = 0; - } - - public function recordCaptured(): void + public function observeCapture(array $message): array { $this->totalEventsCaptured++; - $this->pendingEvents++; - } - - public function recordRequest(int $statusCode, int $retryAttempt, int $eventCount, array $uuidList): void - { - $this->requestsMade[] = new RequestInfo( - (int) floor(microtime(true) * 1000), - $statusCode, - $retryAttempt, - $eventCount, - $uuidList, - ); - - if ($retryAttempt > 0) { - $this->totalRetries++; - } - - if ($statusCode === 200) { - $this->totalEventsSent += $eventCount; - $this->pendingEvents = max(0, $this->pendingEvents - $eventCount); - } + $this->lastUuid = $message['uuid'] ?? null; + return $message; } public function recordError(string $error): void @@ -96,261 +26,17 @@ public function recordError(string $error): void $this->lastError = $error; } - private function discardQueuedEvents(Client $client): void - { - try { - $clientReflection = new ReflectionObject($client); - $consumerProperty = $clientReflection->getProperty('consumer'); - $consumerProperty->setAccessible(true); - $consumer = $consumerProperty->getValue($client); - if (!is_object($consumer)) { - return; - } - - $consumerReflection = new ReflectionObject($consumer); - while (!$consumerReflection->hasProperty('queue')) { - $parent = $consumerReflection->getParentClass(); - if ($parent === false) { - return; - } - $consumerReflection = $parent; - } - - $queueProperty = $consumerReflection->getProperty('queue'); - $queueProperty->setAccessible(true); - $queueProperty->setValue($consumer, []); - } catch (Throwable $e) { - error_log('[adapter] error discarding queued events: ' . $e->getMessage()); - } - } - public function toArray(): array { return [ - 'pending_events' => $this->pendingEvents, + // The public SDK has no queue-length accessor. Do not infer delivery from enqueue. + 'pending_events' => null, 'total_events_captured' => $this->totalEventsCaptured, - 'total_events_sent' => $this->totalEventsSent, - 'total_retries' => $this->totalRetries, 'last_error' => $this->lastError, - 'requests_made' => array_map(static fn (RequestInfo $r): array => $r->toArray(), $this->requestsMade), ]; } } -final class TrackedHttpClient extends HttpClient -{ - public function __construct( - private AdapterState $state, - private string $trackedHost, - private bool $trackedUseSsl = true, - private int $trackedMaximumBackoffDuration = 10000, - private bool $trackedCompressRequests = false, - private bool $trackedDebug = false, - private int $trackedCurlTimeoutMilliseconds = 10000, - ) { - parent::__construct( - $trackedHost, - $trackedUseSsl, - $trackedMaximumBackoffDuration, - $trackedCompressRequests, - $trackedDebug, - null, - $trackedCurlTimeoutMilliseconds, - ); - } - - public function sendRequest(string $path, ?string $payload, array $extraHeaders = [], array $requestOptions = []): HttpResponse - { - $protocol = $this->trackedUseSsl ? 'https://' : 'http://'; - $backoff = 100; - $shouldRetry = $requestOptions['shouldRetry'] ?? true; - $shouldVerify = $requestOptions['shouldVerify'] ?? true; - $includeEtag = $requestOptions['includeEtag'] ?? false; - $timeout = isset($requestOptions['timeout']) - ? (int) $requestOptions['timeout'] - : $this->trackedCurlTimeoutMilliseconds; - $retryAttempt = 0; - $httpResponse = new HttpResponse(false, 0, null, 0); - - do { - $ch = curl_init(); - $responseHeaders = []; - - if ($payload !== null) { - curl_setopt($ch, CURLOPT_POSTFIELDS, $payload); - } - - $headers = ['Content-Type: application/json']; - if ($this->trackedCompressRequests) { - $headers[] = 'Content-Encoding: gzip'; - } - - curl_setopt($ch, CURLOPT_HTTPHEADER, array_merge($headers, $extraHeaders)); - curl_setopt($ch, CURLOPT_URL, $protocol . $this->trackedHost . $path); - curl_setopt($ch, CURLOPT_RETURNTRANSFER, $shouldVerify); - curl_setopt($ch, CURLOPT_TIMEOUT_MS, $shouldVerify ? $timeout : 1); - curl_setopt($ch, CURLOPT_CONNECTTIMEOUT_MS, $timeout); - if (!$shouldVerify) { - curl_setopt($ch, CURLOPT_NOSIGNAL, true); - curl_setopt($ch, CURLOPT_FRESH_CONNECT, true); - } - if ($includeEtag) { - curl_setopt($ch, CURLOPT_HEADER, true); - } else { - curl_setopt($ch, CURLOPT_HEADERFUNCTION, static function ($ch, string $header) use (&$responseHeaders): int { - $responseHeaders[] = trim($header); - return strlen($header); - }); - } - - $response = curl_exec($ch); - $responseCode = (int) curl_getinfo($ch, CURLINFO_RESPONSE_CODE); - $curlErrno = (int) curl_errno($ch); - $etag = null; - - if ($includeEtag && $response !== false) { - $headerSize = (int) curl_getinfo($ch, CURLINFO_HEADER_SIZE); - $rawHeaders = substr((string) $response, 0, $headerSize); - $body = substr((string) $response, $headerSize); - if (preg_match('/^etag:\s*(.+)$/mi', $rawHeaders, $matches)) { - $etag = trim($matches[1]); - } - $response = $body; - } - - curl_close($ch); - $httpResponse = new HttpResponse($response, $responseCode, $etag, $curlErrno); - - if ($path === '/batch/') { - [$eventCount, $uuidList] = $this->extractBatchInfo($payload); - $this->state->recordRequest($responseCode, $retryAttempt, $eventCount, $uuidList); - } - - if ($responseCode === 304) { - break; - } - - if ($shouldVerify && $responseCode !== 200) { - if ($shouldRetry === false) { - break; - } - - if ($this->isRetryableStatus($responseCode)) { - $retryAfterMs = $this->retryAfterMilliseconds($responseHeaders); - usleep(($retryAfterMs ?? $backoff) * 1000); - $backoff *= 2; - $retryAttempt++; - } else { - break; - } - } else { - break; - } - } while ($shouldRetry && $backoff < $this->trackedMaximumBackoffDuration); - - return $httpResponse; - } - - /** @return array{0:int,1:list} */ - private function extractBatchInfo(?string $payload): array - { - if ($payload === null || $payload === '') { - return [0, []]; - } - - $json = $payload; - if ($this->trackedCompressRequests) { - $decoded = gzdecode($payload); - if ($decoded !== false) { - $json = $decoded; - } - } - - $decoded = json_decode($json, true); - if (!is_array($decoded) || !isset($decoded['batch']) || !is_array($decoded['batch'])) { - return [0, []]; - } - - $uuidList = []; - foreach ($decoded['batch'] as $event) { - if (is_array($event) && isset($event['uuid']) && is_string($event['uuid'])) { - $uuidList[] = $event['uuid']; - } - } - - return [count($decoded['batch']), $uuidList]; - } -} - -function isValidUuid(mixed $uuid): bool -{ - return is_string($uuid) - && preg_match( - '/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i', - $uuid - ) === 1; -} - -function jsonResponse($client, int $status, array $payload): void -{ - $body = json_encode($payload, JSON_UNESCAPED_SLASHES); - if ($body === false) { - $status = 500; - $body = '{"error":"failed to encode response"}'; - } - - $reason = [ - 200 => 'OK', - 400 => 'Bad Request', - 404 => 'Not Found', - 405 => 'Method Not Allowed', - 500 => 'Internal Server Error', - ][$status] ?? 'OK'; - - fwrite($client, "HTTP/1.1 {$status} {$reason}\r\n"); - fwrite($client, "Content-Type: application/json\r\n"); - fwrite($client, 'Content-Length: ' . strlen($body) . "\r\n"); - fwrite($client, "Connection: close\r\n\r\n"); - fwrite($client, $body); -} - -function readRequest($client): ?array -{ - $requestLine = fgets($client); - if ($requestLine === false || trim($requestLine) === '') { - return null; - } - - $parts = explode(' ', trim($requestLine), 3); - if (count($parts) < 2) { - return null; - } - - $headers = []; - while (($line = fgets($client)) !== false) { - $line = rtrim($line, "\r\n"); - if ($line === '') { - break; - } - $headerParts = explode(':', $line, 2); - if (count($headerParts) === 2) { - $headers[strtolower(trim($headerParts[0]))] = trim($headerParts[1]); - } - } - - $length = isset($headers['content-length']) ? (int) $headers['content-length'] : 0; - $body = ''; - while (strlen($body) < $length && !feof($client)) { - $body .= fread($client, $length - strlen($body)); - } - - return [ - 'method' => strtoupper($parts[0]), - 'path' => parse_url($parts[1], PHP_URL_PATH) ?: '/', - 'body' => $body, - ]; -} - function requestJson(array $request): array { if ($request['body'] === '') { @@ -376,27 +62,19 @@ function normalizeHost(string $host): array return [$normalized, $useSsl]; } -function maxBackoffDurationForRetries(int $maxRetries): int +function maxBackoffDurationForRetries(int $maxRetries, string $consumer): int { if ($maxRetries <= 0) { return 100; } - return (100 * (2 ** $maxRetries)) + 1; + // Socket checks its limit before sleeping; HttpClient checks after doubling. + return (100 * (2 ** $maxRetries)) + ($consumer === 'socket' ? 0 : 1); } function handleRequest(array $request, AdapterState $state): array { try { - if ($request['method'] === 'GET' && $request['path'] === '/health') { - return [200, [ - 'sdk_name' => 'posthog-php', - 'sdk_version' => PostHog::VERSION, - 'adapter_version' => '1.0.0', - 'capabilities' => ['capture_v0', 'encoding_gzip'], - ]]; - } - if ($request['method'] === 'POST' && $request['path'] === '/init') { $data = requestJson($request); $apiKey = isset($data['api_key']) ? trim((string) $data['api_key']) : ''; @@ -408,36 +86,34 @@ function handleRequest(array $request, AdapterState $state): array return [400, ['error' => 'host is required']]; } - $state->reset(); [$normalizedHost, $useSsl] = normalizeHost($host); $flushAt = max(1, (int) ($data['flush_at'] ?? 100)); $flushIntervalMs = max(0, (int) ($data['flush_interval_ms'] ?? 5000)); $maxRetries = max(0, (int) ($data['max_retries'] ?? 3)); $enableCompression = (bool) ($data['enable_compression'] ?? false); - $maximumBackoffDuration = maxBackoffDurationForRetries($maxRetries); $timeoutMs = max(1000, (int) ($data['timeout_ms'] ?? 10000)); - $httpClient = new TrackedHttpClient( - $state, - $normalizedHost, - $useSsl, - $maximumBackoffDuration, - $enableCompression, - true, - $timeoutMs, - ); + $consumer = getenv('POSTHOG_CONSUMER') ?: 'lib_curl'; + if (!in_array($consumer, ['lib_curl', 'socket', 'fork_curl'], true)) { + throw new InvalidArgumentException('Unsupported consumer: ' . $consumer); + } + $maximumBackoffDuration = maxBackoffDurationForRetries($maxRetries, $consumer); $state->client = new Client($apiKey, [ 'host' => $normalizedHost, 'ssl' => $useSsl, - 'consumer' => 'lib_curl', + 'consumer' => $consumer, 'batch_size' => $flushAt, 'flush_interval_seconds' => $flushIntervalMs / 1000, 'maximum_backoff_duration' => $maximumBackoffDuration, 'compress_request' => $enableCompression ? 'true' : 'false', 'debug' => true, - 'timeout' => $timeoutMs, - ], $httpClient, null, false); + 'timeout' => $consumer === 'socket' ? $timeoutMs / 1000 : $timeoutMs, + 'before_send' => [$state, 'observeCapture'], + 'error_handler' => static function ($code, $message) use ($state): void { + $state->recordError(is_string($message) ? $message : json_encode($message)); + }, + ], null, null, false); return [200, ['success' => true]]; } @@ -466,14 +142,12 @@ function handleRequest(array $request, AdapterState $state): array $message['timestamp'] = $data['timestamp']; } - if (!isset($message['uuid']) || !isValidUuid($message['uuid'])) { - $message['uuid'] = Uuid::v4(); + $state->lastUuid = null; + $success = $state->client->capture($message); + if (!$success) { + $state->recordError('SDK capture returned false'); } - - $state->client->capture($message); - - $state->recordCaptured(); - return [200, ['success' => true, 'uuid' => $message['uuid']]]; + return [200, ['success' => $success, 'uuid' => $state->lastUuid]]; } if ($request['method'] === 'POST' && $request['path'] === '/get_feature_flag') { @@ -532,19 +206,17 @@ function handleRequest(array $request, AdapterState $state): array return [400, ['error' => 'SDK not initialized']]; } - $state->client->flush(); - return [200, ['success' => true, 'events_flushed' => $state->totalEventsSent]]; + $success = $state->client->flush(); + if (!$success) { + $state->recordError('SDK flush returned false'); + } + return [200, ['success' => $success]]; } if ($request['method'] === 'GET' && $request['path'] === '/state') { return [200, $state->toArray()]; } - if ($request['method'] === 'POST' && $request['path'] === '/reset') { - $state->reset(); - return [200, ['success' => true]]; - } - return [404, ['error' => 'not found']]; } catch (Throwable $e) { $state->recordError($e->getMessage()); @@ -553,29 +225,11 @@ function handleRequest(array $request, AdapterState $state): array } } -$server = stream_socket_server('tcp://0.0.0.0:8080', $errno, $errstr); -if ($server === false) { - fwrite(STDERR, "Failed to start server: {$errstr} ({$errno})\n"); - exit(1); -} - +// One SDK instance per process. The controller owns reset and terminates this worker +// without running destructors, so queued events cannot leak into the next test. $state = new AdapterState(); -fwrite(STDERR, "PostHog PHP SDK compliance adapter listening on :8080\n"); - -while (true) { - $client = @stream_socket_accept($server, -1); - if ($client === false) { - usleep(10000); - continue; - } - - $request = readRequest($client); - if ($request === null) { - fclose($client); - continue; - } - - [$status, $payload] = handleRequest($request, $state); - jsonResponse($client, $status, $payload); - fclose($client); +while (($line = fgets(STDIN)) !== false) { + $request = json_decode($line, true, 512, JSON_THROW_ON_ERROR); + echo json_encode(handleRequest($request, $state), JSON_THROW_ON_ERROR) . "\n"; + fflush(STDOUT); } diff --git a/sdk_compliance_adapter/check_report.py b/sdk_compliance_adapter/check_report.py new file mode 100644 index 0000000..a694668 --- /dev/null +++ b/sdk_compliance_adapter/check_report.py @@ -0,0 +1,118 @@ +"""Check inventory completeness in harness 1.0.0 Markdown; assertions stay advisory.""" + +import argparse +import json +from pathlib import Path +import re + + +PROFILES = ("lib_curl", "socket", "fork_curl") +INVENTORY = Path(__file__).with_name("expected_inventory.json") + + +def load_inventory(): + inventory = json.loads(INVENTORY.read_text()) + if set(inventory) != {"capture", "feature_flags"}: + raise ValueError("Expected capture and feature_flags inventories") + for suite, count in (("capture", 30), ("feature_flags", 17)): + ids = inventory[suite] + if len(ids) != count or len(set(ids)) != count: + raise ValueError(f"Expected {count} unique {suite} IDs") + if any(not name.startswith(suite + ".") for name in ids): + raise ValueError(f"Invalid {suite} ID") + return inventory + + +def check_report(report, profile): + if profile not in PROFILES: + raise ValueError(f"Unknown profile: {profile}") + inventory = load_inventory() + lines = iter(line for line in report.splitlines() if line.strip()) + + def take(pattern): + line = next(lines, "") + match = re.fullmatch(pattern, line) + if not match: + raise ValueError(f"Expected {pattern!r}, got {line!r}") + return match + + def counts(prefix=""): + match = take(prefix + r"\*\*(\d+)/(\d+)\*\* tests passed(?:, \*\*(\d+)\*\* failed)?") + passed, total, failed = int(match[1]), int(match[2]), int(match[3] or 0) + if passed + failed != total: + raise ValueError("Inconsistent summary totals") + return passed, total, failed + + take(re.escape(f"# posthog-php-{profile} Compliance Report")) + take(r"\*\*Date\*\*: .+") + take(r"\*\*Duration\*\*: \d+ms") + take(r"## (?:✅ All Tests Passed!|⚠️ Some Tests Failed)") + overall = counts() + take("---") + seen_suites = set() + all_passed = 0 + all_total = 0 + for line in lines: + heading = re.fullmatch(r"## (\w+) Tests", line) + if not heading: + raise ValueError(f"Unexpected report content: {line!r}") + suite = heading[1].lower() + if suite not in inventory or suite in seen_suites: + raise ValueError(f"Unexpected or duplicate suite: {suite}") + seen_suites.add(suite) + summary = counts(r"(?:✅|⚠️) ") + take("
") + take("View Details") + take(re.escape("| Test | Status | Duration |")) + take(re.escape("|------|--------|----------|")) + # The pinned renderer title-cases IDs and replaces underscores with spaces. + expected = {name.split(".", 1)[1].replace("_", " ").title(): name + for name in inventory[suite]} + if len(expected) != len(inventory[suite]): + raise ValueError("Inventory has ambiguous rendered names") + seen = set() + passed = 0 + line = next(lines, "") + while line.startswith("|"): + row = re.fullmatch(r"\| (.+?) \| (✅|❌) \| \d+ms \|", line) + if not row or row[1] not in expected: + raise ValueError(f"Unexpected test row: {line!r}") + if row[1] in seen: + raise ValueError(f"Duplicate test ID: {expected[row[1]]}") + seen.add(row[1]) + passed += row[2] == "✅" + line = next(lines, "") + if seen != set(expected): + missing = [expected[name] for name in sorted(set(expected) - seen)] + raise ValueError(f"Missing IDs: {missing}") + if summary != (passed, len(seen), len(seen) - passed): + raise ValueError(f"{suite} summary differs from test rows") + if line == "### Failures": + # Diagnostics may contain table-like text; only the results table counts. + for line in lines: + if line == "
": + break + if line != "": + raise ValueError("Missing suite closing details tag") + all_passed += passed + all_total += len(seen) + if seen_suites != set(inventory): + raise ValueError("Missing test suites") + if overall != (all_passed, all_total, all_total - all_passed): + raise ValueError("Overall summary differs from test rows") + return f"{profile}: selected={all_total}, passed={all_passed}, failed={all_total - all_passed}" + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("profile", choices=PROFILES) + parser.add_argument("report", type=Path) + args = parser.parse_args() + try: + print(check_report(args.report.read_text(), args.profile)) + except (OSError, ValueError) as error: + parser.exit(1, f"Report completeness failed: {error}\n") + + +if __name__ == "__main__": + main() diff --git a/sdk_compliance_adapter/docker-compose.yml b/sdk_compliance_adapter/docker-compose.yml index 9793a73..f8b396d 100644 --- a/sdk_compliance_adapter/docker-compose.yml +++ b/sdk_compliance_adapter/docker-compose.yml @@ -5,11 +5,13 @@ services: build: context: .. dockerfile: sdk_compliance_adapter/Dockerfile + environment: + POSTHOG_CONSUMER: ${POSTHOG_CONSUMER:-lib_curl} networks: - test-network test-harness: - image: ghcr.io/posthog/sdk-test-harness:0.10.0 + image: ghcr.io/posthog/sdk-test-harness:1.0.0 command: ["run", "--adapter-url", "http://sdk-adapter:8080", "--mock-url", "http://test-harness:8081"] networks: - test-network diff --git a/sdk_compliance_adapter/expected_inventory.json b/sdk_compliance_adapter/expected_inventory.json new file mode 100644 index 0000000..924fef5 --- /dev/null +++ b/sdk_compliance_adapter/expected_inventory.json @@ -0,0 +1,53 @@ +{ + "capture": [ + "capture.format_validation.event_has_required_fields", + "capture.format_validation.event_has_uuid", + "capture.format_validation.event_has_lib_properties", + "capture.format_validation.distinct_id_is_string", + "capture.format_validation.token_is_present", + "capture.format_validation.custom_properties_preserved", + "capture.format_validation.event_has_timestamp", + "capture.format_validation.non_utc_event_timestamp_is_converted_to_utc", + "capture.retry_behavior.retries_on_503", + "capture.retry_behavior.does_not_retry_on_400", + "capture.retry_behavior.does_not_retry_on_401", + "capture.retry_behavior.respects_retry_after_header", + "capture.retry_behavior.implements_backoff", + "capture.retry_behavior.retries_on_500", + "capture.retry_behavior.retries_on_502", + "capture.retry_behavior.retries_on_504", + "capture.retry_behavior.max_retries_respected", + "capture.deduplication.generates_unique_uuids", + "capture.deduplication.preserves_uuid_on_retry", + "capture.deduplication.preserves_uuid_and_timestamp_on_retry", + "capture.deduplication.preserves_uuid_and_timestamp_on_batch_retry", + "capture.deduplication.no_duplicate_events_in_batch", + "capture.deduplication.different_events_have_different_uuids", + "capture.compression.sends_gzip_when_enabled", + "capture.batch_format.uses_proper_batch_structure", + "capture.batch_format.flush_with_no_events_sends_nothing", + "capture.batch_format.multiple_events_batched_together", + "capture.error_handling.does_not_retry_on_403", + "capture.error_handling.does_not_retry_on_413", + "capture.error_handling.retries_on_408" + ], + "feature_flags": [ + "feature_flags.request_lifecycle.mock_response_value_is_returned_to_caller", + "feature_flags.request_lifecycle.no_flags_request_on_init_alone", + "feature_flags.request_lifecycle.no_flags_request_on_normal_capture", + "feature_flags.request_lifecycle.two_flag_calls_produce_two_remote_requests", + "feature_flags.request_payload.disable_geoip_false_propagates_as_geoip_disable_false", + "feature_flags.request_payload.disable_geoip_omitted_defaults_to_false", + "feature_flags.request_payload.flag_keys_to_evaluate_contains_only_requested_key", + "feature_flags.request_payload.flags_request_hits_flags_path_not_decide", + "feature_flags.request_payload.flags_request_omits_authorization_header", + "feature_flags.request_payload.flags_request_uses_v2_query_param", + "feature_flags.request_payload.groups_default_to_empty_object", + "feature_flags.request_payload.groups_round_trip", + "feature_flags.request_payload.request_with_person_properties_device_id", + "feature_flags.request_payload.token_in_flags_body_matches_init", + "feature_flags.retry_behavior.retries_flags_on_502", + "feature_flags.retry_behavior.retries_flags_on_504", + "feature_flags.side_effect_events.get_feature_flag_captures_feature_flag_called_event" + ] +} diff --git a/sdk_compliance_adapter/server.py b/sdk_compliance_adapter/server.py new file mode 100644 index 0000000..0e4ad92 --- /dev/null +++ b/sdk_compliance_adapter/server.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +"""Sequential HTTP controller and observation-only TCP relay for the PHP SDK.""" + +import gzip +import json +import os +from pathlib import Path +import select +import signal +import sys +import socket +import socketserver +import subprocess +import threading +import time +from http.server import BaseHTTPRequestHandler, HTTPServer +from urllib.parse import urlsplit + + +class Relay(socketserver.ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = False + + def __init__(self, address, target): + super().__init__(address, RelayConnection) + self.target = target + self.requests = [] + self.lock = threading.Lock() + self.attempts = {} + self.connections = set() + self.closed = False + + def register(self, connection): + with self.lock: + if self.closed: + return False + self.connections.add(connection) + return True + + def unregister(self, connection): + with self.lock: + self.connections.discard(connection) + + def server_close(self): + with self.lock: + self.closed = True + for connection in self.connections: + try: + connection.shutdown(socket.SHUT_RDWR) + except OSError: + pass # The SDK may already have closed its end. + # Join connection handlers as well as closing the listening socket. + super().server_close() + + def begin_action(self): + # Identical independent flag evaluations are not retries of each other. + with self.lock: + self.attempts = {} + + def observe(self, request, response, timestamp_ms): + header, body = bytes(request).split(b"\r\n\r\n", 1) + headers = {} + for line in header.split(b"\r\n")[1:]: + key, value = line.split(b":", 1) + headers[key.lower()] = value.strip() + path = header.split(b" ")[1] + events = [] + if path == b"/batch/": + if headers.get(b"content-encoding") == b"gzip": + body = gzip.decompress(body) + events = json.loads(body).get("batch", []) + status = int(response.split(b" ", 2)[1]) + with self.lock: + fingerprint = bytes(request) + attempt = self.attempts.get(fingerprint, 0) + self.attempts[fingerprint] = attempt + 1 + self.requests.append({ + "timestamp_ms": timestamp_ms, + "status_code": status, + "retry_attempt": attempt, + "event_count": len(events), + "uuid_list": [event["uuid"] for event in events if "uuid" in event], + }) + + def snapshot(self): + with self.lock: + return list(self.requests) + + +class RelayConnection(socketserver.BaseRequestHandler): + def handle(self): + if not self.server.register(self.request): + return + try: + self.relay() + except OSError as error: + print(f"[relay] connection closed: {error}", file=sys.stderr) + finally: + self.server.unregister(self.request) + + def relay(self): + # Each incoming SDK connection opens exactly one upstream connection. Raw + # bytes flow unchanged in both directions; parsing only populates telemetry. + with socket.create_connection(self.server.target, timeout=10) as upstream: + if not self.server.register(upstream): + return + try: + self.forward(upstream) + finally: + self.server.unregister(upstream) + + def forward(self, upstream): + upstream.settimeout(None) + streams = [self.request, upstream] + request = bytearray() + response = bytearray() + timestamp_ms = None + observed = False + while True: + readable, _, _ = select.select(streams, [], [], 30) + if not readable: + return + for source in readable: + data = source.recv(65536) + if not data: + return + if source is self.request: + if timestamp_ms is None: + timestamp_ms = int(time.time() * 1000) + request.extend(data) + upstream.sendall(data) + else: + response.extend(data) + if not observed and b"\r\n\r\n" in response: + try: + self.server.observe(request, response, timestamp_ms) + except Exception as error: + # Observation must never change SDK traffic or outcomes. + print(f"[observer] {error}", file=sys.stderr) + observed = True + self.request.sendall(data) + + +class Controller: + def __init__(self): + self.worker = None + self.relay = None + self.relay_thread = None + self.version = subprocess.check_output([ + "php", "-r", "require 'vendor/autoload.php'; echo PostHog\\PostHog::VERSION;", + ], cwd=Path(__file__).resolve().parent.parent, text=True) + + def reset(self): + if self.worker is not None: + self.worker.kill() + self.worker.wait() + try: + self.worker.stdin.close() + except BrokenPipeError: + pass # A failed worker may have left a buffered IPC command. + self.worker.stdout.close() + self.worker = None + if self.relay is not None: + self.relay.shutdown() + self.relay.server_close() + self.relay_thread.join() + self.relay = None + + def sdk_call(self, method, path, data): + request = {"method": method, "path": path, "body": json.dumps(data)} + self.worker.stdin.write(json.dumps(request) + "\n") + self.worker.stdin.flush() + line = self.worker.stdout.readline() + if not line: + raise RuntimeError(f"SDK worker exited: {self.worker.poll()}") + return json.loads(line) + + def handle(self, method, path, data): + if method == "GET" and path == "/health": + return 200, { + "sdk_name": f"posthog-php-{os.environ.get('POSTHOG_CONSUMER', 'lib_curl')}", + "sdk_version": self.version, + "adapter_version": "1.0.0", + "capabilities": ["capture_v0", "encoding_gzip"], + } + if method == "POST" and path == "/reset": + self.reset() + return 200, {"success": True} + if method == "POST" and path == "/init": + host = urlsplit(data.get("host", "")) + if host.scheme != "http" or not host.hostname or host.username or host.path not in ("", "/"): + return 400, {"error": "This local wire profile requires an http:// mock host"} + if not data.get("api_key"): + return 400, {"error": "api_key is required"} + self.reset() + self.relay = Relay(("127.0.0.1", int(os.environ.get("PROXY_PORT", "8082"))), + (host.hostname, host.port or 80)) + self.relay_thread = threading.Thread(target=self.relay.serve_forever, daemon=True) + self.relay_thread.start() + self.worker = subprocess.Popen([ + "php", "-d", "default_socket_timeout=1", str(Path(__file__).with_name("adapter.php")), + ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, bufsize=1) + data = dict(data, host=f"http://127.0.0.1:{self.relay.server_address[1]}") + if self.worker is None: + if method == "GET" and path == "/state": + return 200, {"pending_events": 0, "total_events_captured": 0, + "total_events_sent": 0, "total_retries": 0, + "last_error": None, "requests_made": []} + return 400, {"error": "SDK not initialized"} + self.relay.begin_action() + before = self.relay.snapshot() + status, result = self.sdk_call(method, path, data) + requests = self.relay.snapshot() + if path == "/state": + result.update({ + "total_events_sent": sum(r["event_count"] for r in requests if r["status_code"] == 200), + "total_retries": sum(r["retry_attempt"] > 0 for r in requests), + "requests_made": requests, + }) + elif path == "/flush" and status == 200: + result["events_flushed"] = sum(r["event_count"] for r in requests[len(before):] + if r["status_code"] == 200) + return status, result + + +class Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.handle_action() + + def do_POST(self): + self.handle_action() + + def handle_action(self): + try: + body = self.rfile.read(int(self.headers.get("Content-Length", 0))) + data = json.loads(body) if body else {} + status, result = self.server.controller.handle(self.command, urlsplit(self.path).path, data) + except Exception as error: + status, result = 500, {"error": str(error)} + payload = json.dumps(result).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + +if __name__ == "__main__": + consumer = os.environ.get("POSTHOG_CONSUMER", "lib_curl") + if consumer not in ("lib_curl", "socket", "fork_curl"): + raise SystemExit(f"Unsupported POSTHOG_CONSUMER: {consumer}") + server = HTTPServer((os.environ.get("BIND_HOST", "0.0.0.0"), int(os.environ.get("PORT", "8080"))), Handler) + server.controller = Controller() + signal.signal(signal.SIGTERM, lambda *_: sys.exit(0)) + try: + server.serve_forever() + finally: + server.controller.reset() + server.server_close() diff --git a/sdk_compliance_adapter/test_check_report.py b/sdk_compliance_adapter/test_check_report.py new file mode 100644 index 0000000..401289d --- /dev/null +++ b/sdk_compliance_adapter/test_check_report.py @@ -0,0 +1,108 @@ +"""Regressions for the pinned harness Markdown inventory gate (no SDK/network).""" + +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +from check_report import check_report, load_inventory, PROFILES + + +def report_fixture(profile="lib_curl", failed=False): + inventory = load_inventory() + lines = [f"# posthog-php-{profile} Compliance Report", "", "**Date**: 2026-09-07T00:00:00Z", + "**Duration**: 100ms", "", "## ⚠️ Some Tests Failed" if failed else "## ✅ All Tests Passed!", + "", "**46/47** tests passed, **1** failed" if failed else "**47/47** tests passed", "", "---", ""] + for suite, ids in inventory.items(): + suite_failed = failed and suite == "capture" + lines += [f"## {suite.title()} Tests", "", + f"⚠️ **{len(ids) - 1}/{len(ids)}** tests passed, **1** failed" if suite_failed + else f"✅ **{len(ids)}/{len(ids)}** tests passed", "", "
", + "View Details", "", "| Test | Status | Duration |", + "|------|--------|----------|"] + for index, name in enumerate(ids): + status = "❌" if suite_failed and index == 0 else "✅" + display = name.split(".", 1)[1].replace("_", " ").title() + lines.append(f"| {display} | {status} | 1ms |") + if suite_failed: + lines += ["", "### Failures", "", f"**{ids[0].split('.', 1)[1]}**", "```", + "Expected 1 requests, got 0", "```"] + lines += ["", "
", ""] + return "\n".join(lines) + + +class ReportInventoryTest(unittest.TestCase): + def test_complete_passing_and_failing_reports_for_every_profile(self): + for profile in PROFILES: + for failed in (False, True): + with self.subTest(profile=profile, failed=failed): + self.assertEqual(check_report(report_fixture(profile, failed), profile), + f"{profile}: selected=47, passed={46 if failed else 47}, failed={int(failed)}") + + def test_missing_empty_and_incomplete_reports_fail(self): + report = report_fixture() + row = next(line for line in report.splitlines() if line.startswith("| Format")) + variants = ["", "# posthog-php-lib_curl Compliance Report", report.replace(row + "\n", ""), + report.split("## Feature_Flags Tests")[0], report.replace("", "", 1)] + for malformed in variants: + with self.subTest(report=malformed[:80]): + with self.assertRaises(ValueError): + check_report(malformed, "lib_curl") + with tempfile.TemporaryDirectory() as directory: + result = subprocess.run([sys.executable, str(Path(__file__).with_name("check_report.py")), + "lib_curl", str(Path(directory) / "missing.md")], capture_output=True) + self.assertEqual(result.returncode, 1) + self.assertIn(b"Report completeness failed", result.stderr) + + def test_zero_test_report_fails(self): + report = report_fixture() + for suite_ids in load_inventory().values(): + for name in suite_ids: + display = name.split(".", 1)[1].replace("_", " ").title() + report = report.replace(f"| {display} | ✅ | 1ms |\n", "") + for count in (47, 30, 17): + report = report.replace(f"{count}/{count}", "0/0") + with self.assertRaises(ValueError): + check_report(report, "lib_curl") + + def test_duplicate_unexpected_and_malformed_rows_fail(self): + report = report_fixture() + rows = [line for line in report.splitlines() if line.startswith("| Format")] + variants = [report.replace(rows[1], rows[0]), report.replace(rows[0], rows[0] + "\n" + rows[0]), + report.replace(rows[0], "| Unexpected Test | ✅ | 1ms |"), + report.replace(rows[0], rows[0].replace("✅", "SKIP")), + report.replace(rows[0], rows[0].replace("1ms", "unknown"))] + for malformed in variants: + with self.subTest(report=malformed[:80]): + with self.assertRaises(ValueError): + check_report(malformed, "lib_curl") + + def test_duplicate_unexpected_suites_and_wrong_profile_fail(self): + report = report_fixture() + variants = [report + report[report.index("## Feature_Flags Tests"):], + report.replace("## Feature_Flags Tests", "## Unexpected Tests"), + report.replace("posthog-php-lib_curl", "posthog-php-socket")] + for malformed in variants: + with self.assertRaises(ValueError): + check_report(malformed, "lib_curl") + + def test_inconsistent_overall_and_suite_counts_fail(self): + report = report_fixture() + variants = [report.replace("47/47", "46/47"), report.replace("47/47", "47/48"), + report.replace("30/30", "29/30"), report.replace("17/17", "17/18"), + report.replace("47/47", "46/47").replace("**46/47** tests passed", "**46/47** tests passed, **1** failed"), + report_fixture(failed=True).replace("**1** failed", "**2** failed"), + report.replace("| ✅ |", "| ❌ |", 1)] + for malformed in variants: + with self.assertRaises(ValueError): + check_report(malformed, "lib_curl") + + def test_failure_diagnostics_are_not_counted_as_result_rows(self): + report = report_fixture(failed=True).replace("Expected 1 requests, got 0", + "| Diagnostic Text | ❌ | 0ms |") + self.assertIn("selected=47, passed=46, failed=1", check_report(report, "lib_curl")) + + +if __name__ == "__main__": + unittest.main() diff --git a/sdk_compliance_adapter/test_server.py b/sdk_compliance_adapter/test_server.py new file mode 100644 index 0000000..0224c32 --- /dev/null +++ b/sdk_compliance_adapter/test_server.py @@ -0,0 +1,230 @@ +"""Adapter fidelity checks using only loopback HTTP and the installed public SDK.""" + +import gzip +import json +import os +import socket +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import patch + +from server import Controller, Relay + + +MOCK_PORT = int(os.environ.get("TEST_MOCK_PORT", "19276")) +PROXY_PORT = int(os.environ.get("TEST_PROXY_PORT", "19277")) + + +class MockHandler(BaseHTTPRequestHandler): + def do_POST(self): + body = self.rfile.read(int(self.headers["Content-Length"])) + self.server.requests.append((self.path, dict(self.headers), body)) + status = self.server.statuses.pop(0) if self.server.statuses else 200 + response = b'{"featureFlags":{"test-flag":"variant-a"}}' if "/flags/" in self.path else b'{}' + self.send_response(status) + self.send_header("Content-Length", str(len(response))) + self.end_headers() + self.wfile.write(response) + + def log_message(self, *_): + pass + + +class AdapterTest(unittest.TestCase): + def setUp(self): + self.mock = ThreadingHTTPServer(("127.0.0.1", MOCK_PORT), MockHandler) + self.mock.requests = [] + self.mock.statuses = [] + self.thread = threading.Thread(target=self.mock.serve_forever, daemon=True) + self.thread.start() + self.environment = patch.dict(os.environ, {"PROXY_PORT": str(PROXY_PORT)}) + self.environment.start() + self.controller = Controller() + + def tearDown(self): + self.controller.reset() + self.mock.shutdown() + self.mock.server_close() + self.thread.join() + self.environment.stop() + + def call(self, path, data=None): + status, result = self.controller.handle("GET" if data is None else "POST", path, data or {}) + self.assertEqual(status, 200, result) + return result + + def init(self, consumer="lib_curl", **options): + os.environ["POSTHOG_CONSUMER"] = consumer + self.mock.requests.clear() + self.mock.statuses.clear() + self.call("/init", {"host": f"http://127.0.0.1:{MOCK_PORT}", + "api_key": "phc_local_test", **options}) + + def capture(self): + return self.call("/capture", {"event": "test-event", "distinct_id": "test-user"}) + + def test_health_names_consumer_profiles(self): + os.environ.pop("POSTHOG_CONSUMER", None) + self.assertEqual(self.call("/health")["sdk_name"], "posthog-php-lib_curl") + for consumer in ("lib_curl", "socket", "fork_curl"): + with self.subTest(consumer=consumer): + os.environ["POSTHOG_CONSUMER"] = consumer + health = self.call("/health") + self.assertEqual(health["sdk_name"], f"posthog-php-{consumer}") + self.assertEqual(health["sdk_version"], self.controller.version) + self.assertEqual(health["capabilities"], ["capture_v0", "encoding_gzip"]) + + def test_sdk_generated_uuid_and_immediate_flush(self): + for consumer in ("lib_curl", "socket", "fork_curl"): + with self.subTest(consumer=consumer): + self.init(consumer, flush_at=1) + result = self.capture() + self.assertTrue(result["success"]) + event = json.loads(self.mock.requests[0][2])["batch"][0] + self.assertEqual(result["uuid"], event["uuid"]) + self.assertRegex(event["uuid"], r"^[0-9a-f-]{36}$") + state = self.call("/state") + self.assertEqual(state["total_events_captured"], 1) + self.assertEqual(state["total_events_sent"], 1) + self.assertIsNone(state["pending_events"]) + self.assertEqual(self.call("/flush", {})["events_flushed"], 0) + + def test_compressed_capture_does_not_change_flags_headers_or_side_effects(self): + for consumer in ("lib_curl", "socket", "fork_curl"): + with self.subTest(consumer=consumer): + self.init(consumer, enable_compression=True) + data = {"key": "test-flag", "distinct_id": "test-user", "force_remote": True} + for _ in range(2): + self.assertEqual(self.call("/get_feature_flag", data)["value"], "variant-a") + self.assertEqual(self.call("/state")["total_events_captured"], 1) + self.call("/flush", {}) + self.assertEqual(len(self.mock.requests), 3) + for path, headers, body in self.mock.requests[:2]: + self.assertEqual(path, "/flags/?v=2") + self.assertNotIn("Content-Encoding", headers) + self.assertEqual(json.loads(body)["distinct_id"], "test-user") + _, headers, body = self.mock.requests[2] + self.assertEqual(headers["Content-Encoding"], "gzip") + event = json.loads(gzip.decompress(body))["batch"][0] + self.assertEqual(event["event"], "$feature_flag_called") + state = self.call("/state") + self.assertEqual(len(state["requests_made"]), 3) + self.assertEqual(state["total_retries"], 0) + + def test_production_retry_attempts_are_observed(self): + self.init() + self.mock.statuses = [503, 200] + captured = self.capture() + self.assertTrue(self.call("/flush", {})["success"]) + requests = self.call("/state")["requests_made"] + self.assertEqual([r["status_code"] for r in requests], [503, 200]) + self.assertEqual([r["retry_attempt"] for r in requests], [0, 1]) + self.assertTrue(all(r["uuid_list"] == [captured["uuid"]] for r in requests)) + self.assertEqual(self.mock.requests[0][2], self.mock.requests[1][2]) + + def test_terminal_response_exposes_sdk_boolean_not_observer_success(self): + for consumer in ("lib_curl", "socket", "fork_curl"): + with self.subTest(consumer=consumer): + self.init(consumer) + self.mock.statuses = [400] + self.capture() + flushed = self.call("/flush", {}) + self.assertEqual(flushed["success"], consumer == "fork_curl") + self.assertEqual(flushed["events_flushed"], 0) + self.assertEqual(self.call("/state")["requests_made"][0]["status_code"], 400) + + def test_reset_discards_worker_without_sending_queue(self): + self.init() + self.capture() + worker = self.controller.worker + self.call("/reset", {}) + self.assertIsNotNone(worker.poll()) + self.assertEqual(self.mock.requests, []) + self.assertEqual(self.call("/state")["total_events_captured"], 0) + self.init() + self.assertTrue(self.call("/flush", {})["success"]) + self.assertEqual(self.mock.requests, []) + + def test_failed_sdk_worker_is_not_reported_as_success(self): + self.init() + self.controller.worker.kill() + self.controller.worker.wait() + with self.assertRaises((BrokenPipeError, RuntimeError)): + self.call("/flush", {}) + + +class RelayTest(unittest.TestCase): + def test_close_terminates_active_connections_before_returning(self): + accepted = threading.Event() + with socket.socket() as target: + target.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + target.bind(("127.0.0.1", MOCK_PORT)) + target.listen() + + def accept(): + with target.accept()[0] as connection: + accepted.set() + connection.recv(1) + + target_thread = threading.Thread(target=accept) + target_thread.start() + relay = Relay(("127.0.0.1", PROXY_PORT), target.getsockname()) + relay_thread = threading.Thread(target=relay.serve_forever) + relay_thread.start() + with socket.create_connection(relay.server_address) as client: + self.assertTrue(accepted.wait(5)) + relay.shutdown() + relay.server_close() + relay_thread.join() + target_thread.join(timeout=5) + self.assertFalse(target_thread.is_alive()) + self.assertEqual(client.recv(1), b"") + self.assertEqual(relay.connections, set()) + + def test_raw_request_response_unchanged_and_no_proxy_retry(self): + request = (b"POST /batch/ HTTP/1.1\r\nHost: original\r\nX-Custom: a B\r\n" + b"Content-Length: 12\r\n\r\n{\"batch\":[]}") + response = (b"HTTP/1.1 503 Unavailable\r\nRetry-After: 3\r\n" + b"Content-Length: 4\r\nConnection: close\r\n\r\nfail") + received = [] + with socket.socket() as target: + target.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + target.bind(("127.0.0.1", MOCK_PORT)) + target.listen() + + def respond(): + with target.accept()[0] as client: + data = b"" + while len(data) < len(request): + data += client.recv(4096) + received.append(data) + client.sendall(response) + + target_thread = threading.Thread(target=respond) + target_thread.start() + relay = Relay(("127.0.0.1", PROXY_PORT), target.getsockname()) + relay_thread = threading.Thread(target=relay.serve_forever) + relay_thread.start() + try: + with socket.create_connection(relay.server_address) as client: + client.sendall(request) + data = b"" + while True: + chunk = client.recv(4096) + if not chunk: + break + data += chunk + self.assertEqual(data, response) + self.assertEqual(received, [request]) + self.assertEqual(len(relay.snapshot()), 1) + self.assertEqual(relay.snapshot()[0]["status_code"], 503) + finally: + relay.shutdown() + relay.server_close() + relay_thread.join() + target_thread.join() + + +if __name__ == "__main__": + unittest.main()