diff --git a/.github/workflows/sdk-compliance.yml b/.github/workflows/sdk-compliance.yml index 97d8b4b..2903137 100644 --- a/.github/workflows/sdk-compliance.yml +++ b/.github/workflows/sdk-compliance.yml @@ -13,20 +13,46 @@ on: jobs: compliance: - name: PostHog SDK compliance tests (capture v0) - # Keep the harness image version pinned separately so v0/v1 runs are reproducible. + name: PostHog SDK compliance (${{ matrix.profile }}) + strategy: + fail-fast: false + matrix: + include: + - profile: v0-gzip + dockerfile: Dockerfile + - profile: v1-gzip + dockerfile: Dockerfile.v1 + - profile: v1-deflate + dockerfile: Dockerfile.v1.deflate + - profile: v1-br + dockerfile: Dockerfile.v1.br + - profile: v1-zstd + dockerfile: Dockerfile.v1.zstd uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@03d972e49be84402c491324320b0a0f38c2ddc53 with: - adapter-dockerfile: "sdk_compliance_adapter/Dockerfile" + adapter-dockerfile: sdk_compliance_adapter/${{ matrix.dockerfile }} adapter-context: "." - test-harness-version: "0.10.0" - report-name: "sdk-compliance-report-v0" + test-harness-version: "1.0.0" + sdk-type: server + concurrency: 1 + continue-on-error: true + report-name: sdk-compliance-report-${{ matrix.profile }} - compliance-v1: - name: PostHog SDK compliance tests (capture v1) - uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@03d972e49be84402c491324320b0a0f38c2ddc53 - with: - adapter-dockerfile: "sdk_compliance_adapter/Dockerfile.v1" - adapter-context: "." - test-harness-version: "0.10.0" - report-name: "sdk-compliance-report-v1" + report-inventory: + name: Verify compliance report inventory + needs: compliance + if: always() + runs-on: ubuntu-latest + permissions: + contents: read + actions: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/download-artifact@634f93cb2916e3fdff6788551b99b062d0335ce0 # v5.0.0 + with: + pattern: sdk-compliance-report-* + path: reports + # Assertion failures remain advisory; absent/truncated/wrong-profile reports do not. + - run: | + python3 -m unittest discover -s sdk_compliance_adapter -p 'test_*.py' + python3 sdk_compliance_adapter/check_reports.py reports diff --git a/sdk_compliance_adapter/CONTRIBUTING.md b/sdk_compliance_adapter/CONTRIBUTING.md index 6f4ed0d..8fda72f 100644 --- a/sdk_compliance_adapter/CONTRIBUTING.md +++ b/sdk_compliance_adapter/CONTRIBUTING.md @@ -1,54 +1,105 @@ # Contributing -This package contains the PostHog Go SDK compliance adapter used with the PostHog SDK Test Harness. +This adapter builds against the checked-out Go SDK, not a separately published SDK dependency. -## Running tests +## Local checks -Tests run automatically in CI via GitHub Actions. +From the repository root: -CI runs two jobs: `compliance` (capture v0, `Dockerfile`) and `compliance-v1` -(capture v1, `Dockerfile.v1`). The only difference between the images is the -`CAPTURE_MODE=v1` env var, which flips the adapter's `/health` capabilities and -selects `posthog.CaptureModeAnalyticsV1` at init. Both jobs pin the reusable -workflow to the 0.10.0 release commit and run the `0.10.0` harness image. +```sh +go test -race ./sdk_compliance_adapter +python3 -m unittest discover -s sdk_compliance_adapter -p 'test_*.py' +``` -### Locally with Docker Compose +Adapter tests use an available loopback port; set `COMPLIANCE_TEST_PORT` to reserve +a specific port. Run `make test` for the complete Go test suite. -Run the full compliance suite from the `sdk_compliance_adapter` directory: +## Compliance profiles -```bash -docker-compose up --build --abort-on-container-exit -``` +CI pins the reusable workflow at `03d972e49be84402c491324320b0a0f38c2ddc53` +and the harness image at `1.0.0` (contract 1.2). Each profile gets a separate report artifact: -This will: +| Profile | Dockerfile | Capture tests | Flag tests | +| --- | --- | ---: | ---: | +| v0-gzip | `Dockerfile` | 30 | 17 | +| v1-gzip | `Dockerfile.v1` | 95 | 17 | +| v1-deflate | `Dockerfile.v1.deflate` | 94 | 17 | +| v1-br | `Dockerfile.v1.br` | 94 | 17 | +| v1-zstd | `Dockerfile.v1.zstd` | 94 | 17 | -1. Build the Go SDK adapters (v0 on `:8080`, v1 on `:8082`) -2. Pull the test harness image -3. Run the capture v0 compliance tests against the v0 adapter +Both protocol suites include the non-UTC timestamp override test. `CAPTURE_MODE=v1` +selects `posthog.CaptureModeAnalyticsV1`; `COMPRESSION` selects the codec used when +`enable_compression:true`. Omitted compression retains the SDK default (none), and +explicit false always selects none. Each process advertises only its selected codec. +The extra V1 profiles replace the two gzip-only tests with one codec header test. +The harness does not decode deflate, Brotli or zstd bodies and returns an empty V1 +results map for those requests. The SDK treats UUIDs absent from results as accepted +without firing callbacks, so the adapter cannot observe their completion. The three +codec header tests currently time out at `/flush` before reaching their assertions, +although the SDK emits the requested encodings. These are harness decoding and +adapter completion gaps, not failing codecs or passing header assertions. The +adapter's own codec tests decode actual SDK requests and verify delivery callbacks +and retries using UUID-keyed results. -> **Note:** `docker-compose` currently targets the v0 adapter only. The v1 -> adapter image is built to verify it compiles, but v1 compliance tests run in -> CI via the separate `compliance-v1` workflow job. To run v1 locally, use the -> manual Docker instructions below with `Dockerfile.v1`. +### Public SDK mapping -### Manually with Docker +- Capture uses `NewWithConfig` and `Enqueue(Capture)`. Timestamp input is parsed as + `time.Time`; UTC normalization, UUID generation, batching and retries remain SDK-owned. +- Flags use `EvaluateFlags` with singleton `FlagKeys`, then snapshot `GetFlag`. + SDK transport, response parsing, retries and deduplicated `$feature_flag_called` + events are exercised. Each action waits for SDK exposure delivery callbacks before + returning, so a subsequent mock reset cannot receive the previous action's events. + No personal API key or local evaluator is configured, so each action evaluates + remotely regardless of `force_remote`. +- `BeforeSend` observes the SDK-generated UUID without changing the event. + Public `Callback` notifications track successful and terminally failed events; + the transport passively records actual attempts, including encoded bodies. +- The Go client has no non-closing immediate flush. `/flush` waits for the configured + SDK interval and terminal callbacks, bounded to 30 seconds (or request cancellation). + It returns HTTP 504 rather than claiming completion on timeout, including when V1 + responses omit event UUIDs and the SDK supplies no callback. `events_flushed` + counts successful callbacks during that wait. It does not close/recreate the client. + Default adapter batching is one event / 20 ms, with explicit init options forwarded. +- Reset and reinit close the old client before clearing its observations. Actions are + serialized; parallel test isolation is not supported. -```bash -# Create network -docker network create test-network +### Known contract differences + +Expected reports are 45/47 for V0 gzip, 110/112 for V1 gzip, and 108/111 for each +alternate V1 codec (the completion timeout above plus two flag failures). + +All 17 flag tests remain selected. Two are expected to fail through the native SDK: -# Build and run adapter (use Dockerfile.v1 to exercise capture v1) -docker build -f sdk_compliance_adapter/Dockerfile -t posthog-go-adapter . -docker run -d --name sdk-adapter --network test-network -p 8080:8080 posthog-go-adapter +- `feature_flags.request_payload.disable_geoip_false_propagates_as_geoip_disable_false`: + the SDK omits the false field on the wire. +- `feature_flags.request_payload.disable_geoip_omitted_defaults_to_false`: + the documented server SDK default is true. -# Run test harness -docker run --rm \ - --name test-harness \ - --network test-network \ - ghcr.io/posthog/sdk-test-harness:0.10.0 \ - run --adapter-url http://sdk-adapter:8080 --mock-url http://test-harness:8081 +Dedicated AI capture is not supported. Compliance assertions remain advisory, but +`report-inventory` fails when a profile report is absent, truncated, or lacks the +expected suite counts and UTC/codec/flag cases. Inspect each profile's artifact for +actual failures; an advisory job conclusion or the shared workflow's PR comment +is not a complete multi-profile result. -# Cleanup +## Run with Docker + +From the repository root (choose any Dockerfile from the table): + +```sh +docker network create test-network +docker build -f sdk_compliance_adapter/Dockerfile.v1.deflate -t posthog-go-adapter . +docker run -d --name sdk-adapter --network test-network posthog-go-adapter +docker run --rm --name test-harness --network test-network \ + ghcr.io/posthog/sdk-test-harness:1.0.0 \ + run --adapter-url http://sdk-adapter:8080 --mock-url http://test-harness:8081 \ + --sdk-type server --concurrency 1 docker stop sdk-adapter && docker rm sdk-adapter docker network rm test-network ``` + +`docker compose up --build --abort-on-container-exit` from this directory runs V0 +only; V1 has separate CI profiles. For native development, build with +`go build -o /tmp/posthog-go-adapter ./sdk_compliance_adapter`, then set `PORT`, +`CAPTURE_MODE` and `COMPRESSION` when launching. `/init` accepts HTTP mock URLs on +explicit ports at `127.0.0.1`, `localhost`, `::1`, or Docker's `test-harness` host. +Confirm both adapter and mock HTTP readiness before invoking the pinned suites. diff --git a/sdk_compliance_adapter/Dockerfile.v1.br b/sdk_compliance_adapter/Dockerfile.v1.br new file mode 100644 index 0000000..b815d27 --- /dev/null +++ b/sdk_compliance_adapter/Dockerfile.v1.br @@ -0,0 +1,29 @@ +FROM golang:1.21-alpine AS builder + +WORKDIR /app + +# Copy the SDK source +COPY go.mod go.sum ./ +COPY *.go ./ + +# Copy adapter +COPY sdk_compliance_adapter/main.go /app/adapter/ + +# Download dependencies +RUN go mod download + +# Build adapter +RUN cd /app/adapter && go build -o /app/adapter-server main.go + +FROM alpine:latest + +WORKDIR /app + +COPY --from=builder /app/adapter-server /app/adapter-server + +# Select the capture-v1 protocol; same binary, different runtime mode. +ENV CAPTURE_MODE=v1 COMPRESSION=br + +EXPOSE 8080 + +CMD ["/app/adapter-server"] diff --git a/sdk_compliance_adapter/Dockerfile.v1.deflate b/sdk_compliance_adapter/Dockerfile.v1.deflate new file mode 100644 index 0000000..cd56297 --- /dev/null +++ b/sdk_compliance_adapter/Dockerfile.v1.deflate @@ -0,0 +1,29 @@ +FROM golang:1.21-alpine AS builder + +WORKDIR /app + +# Copy the SDK source +COPY go.mod go.sum ./ +COPY *.go ./ + +# Copy adapter +COPY sdk_compliance_adapter/main.go /app/adapter/ + +# Download dependencies +RUN go mod download + +# Build adapter +RUN cd /app/adapter && go build -o /app/adapter-server main.go + +FROM alpine:latest + +WORKDIR /app + +COPY --from=builder /app/adapter-server /app/adapter-server + +# Select the capture-v1 protocol; same binary, different runtime mode. +ENV CAPTURE_MODE=v1 COMPRESSION=deflate + +EXPOSE 8080 + +CMD ["/app/adapter-server"] diff --git a/sdk_compliance_adapter/Dockerfile.v1.zstd b/sdk_compliance_adapter/Dockerfile.v1.zstd new file mode 100644 index 0000000..9e58194 --- /dev/null +++ b/sdk_compliance_adapter/Dockerfile.v1.zstd @@ -0,0 +1,29 @@ +FROM golang:1.21-alpine AS builder + +WORKDIR /app + +# Copy the SDK source +COPY go.mod go.sum ./ +COPY *.go ./ + +# Copy adapter +COPY sdk_compliance_adapter/main.go /app/adapter/ + +# Download dependencies +RUN go mod download + +# Build adapter +RUN cd /app/adapter && go build -o /app/adapter-server main.go + +FROM alpine:latest + +WORKDIR /app + +COPY --from=builder /app/adapter-server /app/adapter-server + +# Select the capture-v1 protocol; same binary, different runtime mode. +ENV CAPTURE_MODE=v1 COMPRESSION=zstd + +EXPOSE 8080 + +CMD ["/app/adapter-server"] diff --git a/sdk_compliance_adapter/check_reports.py b/sdk_compliance_adapter/check_reports.py new file mode 100644 index 0000000..3335e7c --- /dev/null +++ b/sdk_compliance_adapter/check_reports.py @@ -0,0 +1,54 @@ +"""Verify the pinned harness's Markdown inventory, not advisory job conclusions.""" + +import re +import sys +from pathlib import Path + +PROFILES = {"v0-gzip": 30, "v1-gzip": 95, "v1-deflate": 94, "v1-br": 94, "v1-zstd": 94} + + +def check_report(profile: str, report: str) -> str: + capture_count = PROFILES[profile] + suite = "capture" if profile.startswith("v0-") else "capture_v1" + expected = {suite: capture_count, "feature_flags": 17} + sections = re.split(r"^## (Capture|Capture_V1|Feature_Flags) Tests\n", report, flags=re.M) + observed = {} + for name, section in zip(sections[1::2], sections[2::2]): + rows = re.findall(r"^\| (.+?) \| [✅❌] \| \d+ms \|$", section, re.M) + observed[name.lower()] = len(rows) + if len(rows) != len(set(rows)): + raise ValueError(f"{profile}: duplicate test rows") + if observed != expected: + raise ValueError(f"{profile}: expected {expected}, got {observed}") + total = capture_count + 17 + summary = re.search(r"\*\*(\d+)/(\d+)\*\* tests passed", report) + if not summary or int(summary[2]) != total: + raise ValueError(f"{profile}: missing or unexpected summary total (expected {total})") + passed_rows = len(re.findall(r"^\| .+? \| ✅ \| \d+ms \|$", report, re.M)) + if passed_rows != int(summary[1]): + raise ValueError(f"{profile}: summary pass count differs from test rows") + required = [ + "non_utc_event_timestamp_is_converted_to_utc", + "retries_flags_on_502", + "retries_flags_on_504", + "disable_geoip_false_propagates_as_geoip_disable_false", + "disable_geoip_omitted_defaults_to_false", + ] + if suite == "capture_v1": + required.append(f"sends_{profile.split('-')[1]}_content_encoding") + for name in required: + if name.replace("_", " ").title() + " |" not in report: + raise ValueError(f"{profile}: missing test {name}") + passed = int(summary[1]) + return f"{profile}: selected={total}, passed={passed}, failed={total-passed}" + + +def main() -> None: + root = Path(sys.argv[1]) + for profile in PROFILES: + path = root / f"sdk-compliance-report-{profile}" / "sdk-compliance-report.md" + print(check_report(profile, path.read_text())) + + +if __name__ == "__main__": + main() diff --git a/sdk_compliance_adapter/docker-compose.yml b/sdk_compliance_adapter/docker-compose.yml index 5c9def4..7834d2a 100644 --- a/sdk_compliance_adapter/docker-compose.yml +++ b/sdk_compliance_adapter/docker-compose.yml @@ -21,7 +21,7 @@ services: # Test harness 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/main.go b/sdk_compliance_adapter/main.go index d16e516..accd9aa 100644 --- a/sdk_compliance_adapter/main.go +++ b/sdk_compliance_adapter/main.go @@ -2,133 +2,185 @@ package main import ( "bytes" + "compress/gzip" + "compress/zlib" + "context" "encoding/json" + "fmt" "io" "log" "net/http" + "net/url" "os" "strconv" "strings" "sync" "time" + "github.com/andybalholm/brotli" + "github.com/klauspost/compress/zstd" "github.com/posthog/posthog-go" ) const VERSION = "1.0.0" -// captureMode selects which capture protocol this adapter process speaks. It is -// baked at build time via the CAPTURE_MODE env var ("v1" => capture-v1, anything -// else => legacy v0), mirroring the v0/v1 Dockerfile split. One process speaks -// one mode and advertises it via /health capabilities. +// Each process selects one protocol and compression codec at runtime. var captureMode = os.Getenv("CAPTURE_MODE") +var compression = os.Getenv("COMPRESSION") func isV1() bool { return captureMode == "v1" } -// TrackedTransport wraps http.RoundTripper to track requests +func selectedCompression() (posthog.CompressionMode, error) { + switch compression { + case "", "gzip": + return posthog.CompressionGzip, nil + case "deflate": + if isV1() { + return posthog.CompressionDeflate, nil + } + case "br": + if isV1() { + return posthog.CompressionBrotli, nil + } + case "zstd": + if isV1() { + return posthog.CompressionZstd, nil + } + } + return posthog.CompressionNone, fmt.Errorf("unsupported compression profile %q for capture mode %q", compression, captureMode) +} + +// TrackedTransport observes SDK requests without implementing delivery policy. type TrackedTransport struct { base http.RoundTripper state *AdapterState } -func (t *TrackedTransport) RoundTrip(req *http.Request) (*http.Response, error) { - // Read and restore request body to extract UUIDs - var bodyBytes []byte - if req.Body != nil { - bodyBytes, _ = io.ReadAll(req.Body) - req.Body = io.NopCloser(bytes.NewBuffer(bodyBytes)) +func decodeBody(body []byte, encoding string) ([]byte, error) { + var reader io.ReadCloser + var err error + switch encoding { + case "": + return body, nil + case "gzip": + reader, err = gzip.NewReader(bytes.NewReader(body)) + case "deflate": + reader, err = zlib.NewReader(bytes.NewReader(body)) + case "br": + return io.ReadAll(brotli.NewReader(bytes.NewReader(body))) + case "zstd": + decoder, err := zstd.NewReader(nil) + if err != nil { + return nil, err + } + defer decoder.Close() + return decoder.DecodeAll(body, nil) + default: + return nil, fmt.Errorf("unknown content encoding %q", encoding) } - - // Make the request - resp, err := t.base.RoundTrip(req) - if resp == nil { - return resp, err + if err != nil { + return nil, err } + defer reader.Close() + return io.ReadAll(reader) +} - // Parse batch to get UUIDs. The request body shape is the same for v0 and - // v1 ({"batch":[{"uuid":...}]}), so extraction is unchanged. +func (t *TrackedTransport) RoundTrip(req *http.Request) (*http.Response, error) { + var body []byte + if req.Body != nil { + var err error + body, err = io.ReadAll(req.Body) + req.Body.Close() + req.Body = io.NopCloser(bytes.NewReader(body)) + if err != nil { + return nil, err + } + } var batch struct { Batch []struct { UUID string `json:"uuid"` } `json:"batch"` } - uuids := []string{} - if len(bodyBytes) > 0 { - json.Unmarshal(bodyBytes, &batch) - for _, event := range batch.Batch { - if event.UUID != "" { - uuids = append(uuids, event.UUID) - } - } - } - - // PostHog-Attempt is 1-based and only set on the v1 path. attempt-1 is the - // retry index; attempt > 1 means this request is a retry. - attempt := 1 - if a := req.Header.Get("PostHog-Attempt"); a != "" { - if n, e := strconv.Atoi(a); e == nil && n > 0 { - attempt = n - } + decoded, decodeErr := decodeBody(body, req.Header.Get("Content-Encoding")) + if decodeErr == nil { + decodeErr = json.Unmarshal(decoded, &batch) } - - // For v1, a 200 no longer means "all sent": read+restore the body and count - // only terminal results (anything other than "retry") as sent. - terminal := len(batch.Batch) - if isV1() { - var respBytes []byte - if resp.Body != nil { - respBytes, _ = io.ReadAll(resp.Body) - resp.Body = io.NopCloser(bytes.NewBuffer(respBytes)) - } - if resp.StatusCode == 200 { - var parsed struct { - Results map[string]struct { - Result string `json:"result"` - } `json:"results"` - } - terminal = 0 - if err := json.Unmarshal(respBytes, &parsed); err == nil { - for _, r := range parsed.Results { - if r.Result != "retry" { - terminal++ - } - } - } else { - log.Printf("[adapter] v1 response body unmarshal failed: %v (terminal=0, pending events may appear stuck)", err) - } - } + uuids := []string{} + for _, event := range batch.Batch { + uuids = append(uuids, event.UUID) } t.state.mu.Lock() + attempt := 0 + if strings.TrimRight(req.URL.Path, "/") == "/flags" { + attempt = t.state.flagsAttempt + t.state.flagsAttempt++ + } else if n, err := strconv.Atoi(req.Header.Get("PostHog-Attempt")); err == nil && n > 0 { + attempt = n - 1 + } else if len(uuids) > 0 { + key := strings.Join(uuids, ",") + attempt = t.state.captureAttempts[key] + t.state.captureAttempts[key]++ + } + index := len(t.state.requestsMade) t.state.requestsMade = append(t.state.requestsMade, RequestInfo{ - TimestampMs: time.Now().UnixMilli(), - StatusCode: resp.StatusCode, - RetryAttempt: attempt - 1, - EventCount: len(batch.Batch), - UUIDList: uuids, + TimestampMs: time.Now().UnixMilli(), RetryAttempt: attempt, + EventCount: len(batch.Batch), UUIDList: uuids, }) - if attempt > 1 { + if attempt > 0 { t.state.totalRetries++ } - if resp.StatusCode == 200 { - t.state.totalEventsSent += terminal - t.state.pendingEvents -= terminal - if t.state.pendingEvents < 0 { - t.state.pendingEvents = 0 - } + if decodeErr != nil { + t.state.lastError = decodeErr.Error() } t.state.mu.Unlock() + resp, err := t.base.RoundTrip(req) + t.state.mu.Lock() + if resp != nil { + t.state.requestsMade[index].StatusCode = resp.StatusCode + } + if err != nil { + t.state.lastError = err.Error() + } + t.state.mu.Unlock() return resp, err } +// Public SDK hooks own identity and completion, including flag-called events, +// compression, terminal failures and partial V1 batches. +func (s *AdapterState) beforeSend(message posthog.Message) posthog.Message { + if capture, ok := message.(posthog.Capture); ok { + s.mu.Lock() + s.lastUUID = capture.Uuid + s.totalEventsCaptured++ + s.pendingEvents++ + s.mu.Unlock() + } + return message +} + +func (s *AdapterState) Success(_ posthog.APIMessage) { + s.mu.Lock() + defer s.mu.Unlock() + s.totalEventsSent++ + s.pendingEvents-- +} + +func (s *AdapterState) Failure(_ posthog.APIMessage, err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.pendingEvents-- + s.lastError = err.Error() +} + // AdapterState tracks SDK state for test assertions type AdapterState struct { mu sync.Mutex client posthog.Client - config *posthog.Config - apiKey string - host string + lastUUID string + flagsAttempt int + captureAttempts map[string]int totalEventsCaptured int totalEventsSent int totalRetries int @@ -147,7 +199,8 @@ type RequestInfo struct { } var state = &AdapterState{ - requestsMade: []RequestInfo{}, + requestsMade: []RequestInfo{}, + captureAttempts: map[string]int{}, } // HealthResponse represents /health endpoint response @@ -213,22 +266,30 @@ func jsonResponse(w http.ResponseWriter, data interface{}) { // harness mock targets. The adapter is only intended to call the harness mock // server, so keep the outbound network target on a small allowlist. func validateHarnessHost(raw string) (string, bool) { - switch strings.TrimRight(raw, "/") { - case "http://test-harness:8081": - return "http://test-harness:8081", true - case "http://localhost:8081": - return "http://localhost:8081", true - case "http://127.0.0.1:8081": - return "http://127.0.0.1:8081", true + u, err := url.Parse(strings.TrimRight(raw, "/")) + if err != nil || u.Scheme != "http" || u.User != nil || u.Path != "" || u.RawQuery != "" || u.Fragment != "" { + return "", false + } + switch u.Hostname() { + case "test-harness", "localhost", "127.0.0.1", "::1": default: return "", false } + port, err := strconv.Atoi(u.Port()) + if err != nil || port < 1 || port > 65535 { + return "", false + } + return u.String(), true } func healthHandler(w http.ResponseWriter, r *http.Request) { capabilities := []string{"capture_v0", "encoding_gzip"} if isV1() { - capabilities = []string{"capture_v1", "encoding_gzip"} + codec := compression + if codec == "" { + codec = "gzip" + } + capabilities = []string{"capture_v1", "encoding_" + codec} } response := HealthResponse{ SDKName: "posthog-go", @@ -252,29 +313,19 @@ func initHandler(w http.ResponseWriter, r *http.Request) { return } - state.mu.Lock() - oldClient := state.client - state.client = nil - - // Reset state - state.totalEventsCaptured = 0 - state.totalEventsSent = 0 - state.totalRetries = 0 - state.lastError = "" - state.requestsMade = []RequestInfo{} - state.pendingEvents = 0 - state.mu.Unlock() - - // Close the previous client outside state.mu. Close can wait for in-flight - // sends whose tracked transport also records state under the same mutex. - if oldClient != nil { - oldClient.Close() + codec, err := selectedCompression() + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return } + closeAndReset() // Create new client with tracked transport config := posthog.Config{ - Endpoint: validatedHost, - Transport: &TrackedTransport{base: http.DefaultTransport, state: state}, + Endpoint: validatedHost, + Transport: &TrackedTransport{base: http.DefaultTransport, state: state}, + BeforeSend: state.beforeSend, + Callback: state, // Set test-friendly defaults BatchSize: 1, // Flush after each event by default Interval: 20 * time.Millisecond, // Short interval for tests @@ -296,7 +347,7 @@ func initHandler(w http.ResponseWriter, r *http.Request) { } if req.EnableCompression != nil { if *req.EnableCompression { - config.Compression = posthog.CompressionGzip + config.Compression = codec } else { config.Compression = posthog.CompressionNone } @@ -316,9 +367,6 @@ func initHandler(w http.ResponseWriter, r *http.Request) { state.mu.Lock() state.client = client - state.config = &config - state.apiKey = req.APIKey - state.host = validatedHost state.mu.Unlock() jsonResponse(w, map[string]bool{"success": true}) @@ -371,68 +419,74 @@ func captureHandler(w http.ResponseWriter, r *http.Request) { if req.Timestamp != nil { // Parse timestamp if provided t, err := time.Parse(time.RFC3339, *req.Timestamp) - if err == nil { - capture.Timestamp = t + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return } + capture.Timestamp = t } - // Enqueue event + // BeforeSend runs synchronously inside Enqueue, after SDK UUID generation. + state.mu.Lock() + state.lastUUID = "" + state.mu.Unlock() if err := state.client.Enqueue(capture); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + // Queue rejection has no delivery callback, but enrichment may have run. + if err == posthog.ErrQueueFull || err == posthog.ErrClosed { + state.mu.Lock() + if state.lastUUID != "" { + state.pendingEvents-- + } + state.lastError = err.Error() + state.mu.Unlock() + } + jsonError(w, http.StatusInternalServerError, err.Error()) return } - state.mu.Lock() - state.totalEventsCaptured++ - state.pendingEvents++ + uuid := state.lastUUID state.mu.Unlock() - - // TODO: Get actual UUID from SDK - jsonResponse(w, map[string]interface{}{ - "success": true, - "uuid": "generated-uuid", - }) + jsonResponse(w, map[string]interface{}{"success": true, "uuid": uuid}) } func flushHandler(w http.ResponseWriter, r *http.Request) { state.mu.Lock() - if state.client == nil { - state.mu.Unlock() - http.Error(w, "SDK not initialized", http.StatusBadRequest) + initialized := state.client != nil + state.mu.Unlock() + if !initialized { + jsonError(w, http.StatusBadRequest, "SDK not initialized") return } - state.mu.Unlock() - - eventsFlushed := waitForPendingEvents() - jsonResponse(w, map[string]interface{}{ - "success": true, - "events_flushed": eventsFlushed, - }) + eventsFlushed, err := waitForPendingEvents(r.Context()) + if err != nil { + jsonError(w, http.StatusGatewayTimeout, err.Error()) + return + } + jsonResponse(w, map[string]interface{}{"success": true, "events_flushed": eventsFlushed}) } -func waitForPendingEvents() int { +// There is no non-closing public Flush. Wait for the configured SDK interval +// and terminal delivery callbacks, without closing/recreating the client. +func waitForPendingEvents(ctx context.Context) (int, error) { + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() state.mu.Lock() - interval := state.config.Interval - if interval == 0 { - interval = 5 * time.Second // Default - } + before := state.totalEventsSent state.mu.Unlock() - - // Wait only until the current queue drains, with a small cap. Most harness - // tests use BatchSize=1, while batch-format tests rely on the short interval - // above to flush partial batches. Avoid fixed sleeps per test: the compliance - // suite has many flushes and long retry waits of its own. - deadline := time.Now().Add(interval + (100 * time.Millisecond)) + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() for { state.mu.Lock() - pendingEvents := state.pendingEvents - eventsFlushed := state.totalEventsSent + pending, sent := state.pendingEvents, state.totalEventsSent state.mu.Unlock() - - if pendingEvents == 0 || time.Now().After(deadline) { - return eventsFlushed + if pending == 0 { + return sent - before, nil + } + select { + case <-ctx.Done(): + return sent - before, ctx.Err() + case <-ticker.C: } - time.Sleep(10 * time.Millisecond) } } @@ -484,124 +538,36 @@ func featureFlagHandler(w http.ResponseWriter, r *http.Request) { return } + groupProperties := make(map[string]posthog.Properties, len(req.GroupProperties)) + for key, properties := range req.GroupProperties { + groupProperties[key] = properties + } state.mu.Lock() - apiKey := state.apiKey - host := state.host + state.flagsAttempt = 0 state.mu.Unlock() - - personProperties := map[string]interface{}{"distinct_id": req.DistinctID} - for k, v := range req.PersonProperties { - personProperties[k] = v - } - groups := map[string]interface{}{} - for k, v := range req.Groups { - groups[k] = v - } - groupProperties := map[string]interface{}{} - for k, v := range req.GroupProperties { - groupProperties[k] = v - } - geoipDisable := false - if req.DisableGeoIP != nil { - geoipDisable = *req.DisableGeoIP - } - - payload := map[string]interface{}{ - "api_key": apiKey, - "distinct_id": req.DistinctID, - "person_properties": personProperties, - "groups": groups, - "group_properties": groupProperties, - "geoip_disable": geoipDisable, - "flag_keys_to_evaluate": []string{req.Key}, - } - body, _ := json.Marshal(payload) - flagsURL := strings.TrimRight(host, "/") + "/flags/?v=2" - resp, err := postFlagsWithRetry(flagsURL, body) + // No personal API key/local evaluator is configured. Each EvaluateFlags + // action makes a remote request even when force_remote is false or omitted. + snapshot, err := client.EvaluateFlags(posthog.EvaluateFlagsPayload{ + DistinctId: req.DistinctID, + PersonProperties: req.PersonProperties, + Groups: req.Groups, + GroupProperties: groupProperties, + DisableGeoIP: req.DisableGeoIP, + FlagKeys: []string{req.Key}, + }) if err != nil { - log.Printf("Error evaluating feature flag: %s", sanitizeForLog(err.Error())) jsonError(w, http.StatusInternalServerError, err.Error()) return } - defer resp.Body.Close() - var decoded map[string]interface{} - if err := json.NewDecoder(resp.Body).Decode(&decoded); err != nil { - jsonError(w, http.StatusInternalServerError, err.Error()) - return - } - value := interface{}(false) - if flags, ok := decoded["featureFlags"].(map[string]interface{}); ok { - if flagValue, ok := flags[req.Key]; ok { - value = flagValue - } - } - properties := posthog.Properties{ - "$feature_flag": req.Key, - "$feature_flag_response": value, - "$feature/" + req.Key: value, - } - // Only send $feature_flag_has_experiment when the server explicitly - // reported has_experiment on the flag's metadata; omit it when unknown. - if flags, ok := decoded["flags"].(map[string]interface{}); ok { - if flagDetail, ok := flags[req.Key].(map[string]interface{}); ok { - if metadata, ok := flagDetail["metadata"].(map[string]interface{}); ok { - if v, ok := metadata["has_experiment"].(bool); ok { - properties["$feature_flag_has_experiment"] = v - } - } - } - } - - if err := client.Enqueue(posthog.Capture{ - DistinctId: req.DistinctID, - Event: "$feature_flag_called", - Properties: properties, - }); err != nil { - jsonError(w, http.StatusInternalServerError, err.Error()) + value := snapshot.GetFlag(req.Key) + // Complete SDK-owned exposure delivery before the harness resets its mock + // for the next test. Closing a still-pending client at reset would send the + // previous test's event into that new recording window. + if _, err := waitForPendingEvents(r.Context()); err != nil { + jsonError(w, http.StatusGatewayTimeout, err.Error()) return } - state.mu.Lock() - state.totalEventsCaptured++ - state.pendingEvents++ - state.mu.Unlock() - waitForPendingEvents() - - // Avoid logging user-controlled fields (req.Key, req.DistinctID, value) to prevent log injection. - log.Printf("Evaluated feature flag") - - jsonResponse(w, map[string]interface{}{ - "success": true, - "value": value, - }) -} - -func postFlagsWithRetry(flagsURL string, body []byte) (*http.Response, error) { - var lastStatus int - for attempt := 0; attempt < 2; attempt++ { - resp, err := http.Post(flagsURL, "application/json", bytes.NewReader(body)) - if err != nil { - return nil, err - } - if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices { - return resp, nil - } - - lastStatus = resp.StatusCode - resp.Body.Close() - if resp.StatusCode != http.StatusBadGateway && resp.StatusCode != http.StatusGatewayTimeout { - break - } - } - - return nil, &flagsStatusError{statusCode: lastStatus} -} - -type flagsStatusError struct { - statusCode int -} - -func (e *flagsStatusError) Error() string { - return "flags request failed with status " + strconv.Itoa(e.statusCode) + jsonResponse(w, map[string]interface{}{"success": true, "value": value}) } // sanitizeForLog strips CR/LF characters from a string before logging, so @@ -617,41 +583,61 @@ func jsonError(w http.ResponseWriter, status int, msg string) { json.NewEncoder(w).Encode(map[string]string{"error": msg}) } -func resetHandler(w http.ResponseWriter, r *http.Request) { +func closeAndReset() { state.mu.Lock() oldClient := state.client state.client = nil - state.apiKey = "" - state.host = "" + state.mu.Unlock() + // Close before clearing observations: SDK shutdown may deliver a final batch. + if oldClient != nil { + oldClient.Close() + } + state.mu.Lock() + defer state.mu.Unlock() + state.lastUUID = "" state.totalEventsCaptured = 0 state.totalEventsSent = 0 state.totalRetries = 0 state.lastError = "" state.requestsMade = []RequestInfo{} state.pendingEvents = 0 - state.mu.Unlock() - - // Close outside state.mu for the same reason as initHandler. - if oldClient != nil { - oldClient.Close() - } + state.flagsAttempt = 0 + state.captureAttempts = map[string]int{} +} +func resetHandler(w http.ResponseWriter, r *http.Request) { + closeAndReset() jsonResponse(w, map[string]bool{"success": true}) } +// The adapter has one SDK client, so lifecycle/actions are serialized. State +// remains readable while flush waits; parallel test isolation is not advertised. +var actionMu sync.Mutex + +func serial(handler http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + actionMu.Lock() + defer actionMu.Unlock() + handler(w, r) + } +} + func main() { + if _, err := selectedCompression(); err != nil { + log.Fatal(err) + } port := os.Getenv("PORT") if port == "" { port = "8080" } http.HandleFunc("/health", healthHandler) - http.HandleFunc("/init", initHandler) - http.HandleFunc("/capture", captureHandler) - http.HandleFunc("/flush", flushHandler) + http.HandleFunc("/init", serial(initHandler)) + http.HandleFunc("/capture", serial(captureHandler)) + http.HandleFunc("/flush", serial(flushHandler)) http.HandleFunc("/state", stateHandler) - http.HandleFunc("/reset", resetHandler) - http.HandleFunc("/get_feature_flag", featureFlagHandler) + http.HandleFunc("/reset", serial(resetHandler)) + http.HandleFunc("/get_feature_flag", serial(featureFlagHandler)) log.Printf("Starting PostHog Go SDK adapter on port %s", port) log.Fatal(http.ListenAndServe(":"+port, nil)) diff --git a/sdk_compliance_adapter/main_test.go b/sdk_compliance_adapter/main_test.go index 01bfb72..de346b8 100644 --- a/sdk_compliance_adapter/main_test.go +++ b/sdk_compliance_adapter/main_test.go @@ -1,85 +1,364 @@ package main import ( + "bytes" + "context" + "encoding/json" "io" + "net" "net/http" "net/http/httptest" + "os" + "reflect" + "sync" "testing" + "time" + + "github.com/google/uuid" ) -func TestPostFlagsWithRetryRetriesRetryableStatusThenSucceeds(t *testing.T) { - requests := 0 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requests++ - if requests == 1 { - w.WriteHeader(http.StatusBadGateway) +func action(t *testing.T, handler http.HandlerFunc, body string) map[string]interface{} { + t.Helper() + recorder := httptest.NewRecorder() + handler(recorder, httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))) + if recorder.Code != http.StatusOK { + t.Fatalf("action status %d: %s", recorder.Code, recorder.Body.String()) + } + var result map[string]interface{} + if err := json.Unmarshal(recorder.Body.Bytes(), &result); err != nil { + t.Fatal(err) + } + return result +} + +func mockServer(t *testing.T, handler http.HandlerFunc) *httptest.Server { + t.Helper() + port := os.Getenv("COMPLIANCE_TEST_PORT") + if port == "" { + port = "0" + } + listener, err := net.Listen("tcp", "127.0.0.1:"+port) + if err != nil { + t.Fatal(err) + } + server := &httptest.Server{Listener: listener, Config: &http.Server{Handler: handler}} + server.Start() + t.Cleanup(server.Close) + t.Cleanup(closeAndReset) + return server +} + +func profile(t *testing.T, mode, codec string) { + t.Helper() + oldMode, oldCodec := captureMode, compression + captureMode, compression = mode, codec + t.Cleanup(func() { captureMode, compression = oldMode, oldCodec }) +} + +func TestFeatureFlagHandlerUsesSDKEvaluation(t *testing.T) { + for _, status := range []int{502, 504, 400} { + t.Run(http.StatusText(status), func(t *testing.T) { + profile(t, "v0", "gzip") + var mu sync.Mutex + var flags []map[string]interface{} + var events []map[string]interface{} + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { + var body map[string]interface{} + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + } + mu.Lock() + defer mu.Unlock() + if r.URL.Path == "/flags/" { + if r.URL.RawQuery != "v=2" || r.Header.Get("Authorization") != "" { + t.Errorf("unexpected flags request: %v", r) + } + flags = append(flags, body) + if len(flags) == 1 { + w.WriteHeader(status) + return + } + io.WriteString(w, `{"featureFlags":{"example":"variant-a"}}`) + return + } + for _, event := range body["batch"].([]interface{}) { + events = append(events, event.(map[string]interface{})) + } + io.WriteString(w, `{}`) + }) + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`"}`) + payload := `{"key":"example","distinct_id":"user","person_properties":{"$device_id":"device"},"groups":{"company":"acme"},"group_properties":{"company":{"plan":"enterprise"}},"disable_geoip":false}` + if status == 400 { + recorder := httptest.NewRecorder() + featureFlagHandler(recorder, httptest.NewRequest("POST", "/", bytes.NewBufferString(payload))) + if recorder.Code != 500 { + t.Fatalf("status = %d", recorder.Code) + } + } else { + result := action(t, featureFlagHandler, payload) + if result["value"] != "variant-a" { + t.Fatalf("SDK value = %v", result) + } + // A fresh snapshot makes another remote request, but SDK exposure dedup remains intact. + action(t, featureFlagHandler, payload) + } + action(t, flushHandler, `{}`) + mu.Lock() + defer mu.Unlock() + expected := 3 + if status == 400 { + expected = 1 + } + if len(flags) != expected { + t.Fatalf("flags requests = %d, want %d", len(flags), expected) + } + first := flags[0] + if !reflect.DeepEqual(first["flag_keys_to_evaluate"], []interface{}{"example"}) || first["api_key"] != "test-key" || first["distinct_id"] != "user" { + t.Fatalf("SDK payload = %v", first) + } + if first["person_properties"].(map[string]interface{})["$device_id"] != "device" || first["groups"].(map[string]interface{})["company"] != "acme" || first["group_properties"].(map[string]interface{})["company"].(map[string]interface{})["plan"] != "enterprise" { + t.Fatalf("SDK properties = %v", first) + } + if _, exists := first["geoip_disable"]; exists { + t.Fatalf("SDK serializes false GeoIP by omission: %v", first) + } + if status != 400 { + if len(events) != 1 || events[0]["event"] != "$feature_flag_called" { + t.Fatalf("SDK exposure events = %v", events) + } + props := events[0]["properties"].(map[string]interface{}) + if props["$feature_flag"] != "example" || props["$feature_flag_response"] != "variant-a" { + t.Fatalf("exposure properties = %v", props) + } + } else if len(events) != 0 { + t.Fatalf("unexpected events = %v", events) + } + }) + } +} + +func TestFeatureFlagHandlerCompletesExposureBeforeReset(t *testing.T) { + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/flags/" { + io.WriteString(w, `{"featureFlags":{"example":true}}`) return } + io.WriteString(w, `{}`) + }) + // Keep the SDK exposure queued long enough to detect an early return from + // the flag action; delivery still runs on the SDK's configured interval. + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`","flush_at":100,"flush_interval_ms":100}`) + result := action(t, featureFlagHandler, `{"key":"example","distinct_id":"user"}`) + if result["value"] != true { + t.Fatalf("flag value = %v", result) + } + state.mu.Lock() + pending, sent := state.pendingEvents, state.totalEventsSent + state.mu.Unlock() + if pending != 0 || sent != 1 { + t.Fatalf("flag action returned before exposure completion: pending=%d sent=%d", pending, sent) + } + action(t, resetHandler, `{}`) +} - if r.Method != http.MethodPost { - t.Fatalf("method = %s, want %s", r.Method, http.MethodPost) +func TestFeatureFlagHandlerPreservesDefaultGeoIP(t *testing.T) { + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { + var body map[string]interface{} + json.NewDecoder(r.Body).Decode(&body) + if r.URL.Path == "/flags/" && body["geoip_disable"] != true { + t.Errorf("SDK GeoIP default = %v", body["geoip_disable"]) } - body, err := io.ReadAll(r.Body) - if err != nil { - t.Fatal(err) + io.WriteString(w, `{"featureFlags":{}}`) + }) + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`"}`) + result := action(t, featureFlagHandler, `{"key":"unknown","distinct_id":"user"}`) + if result["value"] != nil { + t.Fatalf("unknown snapshot flag = %v", result) + } +} + +func TestCaptureFlushTracksSDKCompletionAndCodecs(t *testing.T) { + for _, mode := range []string{"v0", "v1"} { + codecs := []string{"gzip"} + if mode == "v1" { + codecs = append(codecs, "deflate", "br", "zstd") } - if string(body) != `{"flag_keys_to_evaluate":["example"]}` { - t.Fatalf("body = %s", string(body)) + for _, codec := range codecs { + t.Run(mode+"/"+codec, func(t *testing.T) { + profile(t, mode, codec) + var mu sync.Mutex + var ids []string + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Content-Encoding"); got != codec { + t.Errorf("encoding = %q, want %q", got, codec) + } + raw, _ := io.ReadAll(r.Body) + decoded, err := decodeBody(raw, codec) + if err != nil { + t.Error(err) + return + } + var body struct { + Batch []struct { + UUID string `json:"uuid"` + } `json:"batch"` + } + if err := json.Unmarshal(decoded, &body); err != nil { + t.Error(err) + return + } + id := body.Batch[0].UUID + mu.Lock() + ids = append(ids, id) + attempt := len(ids) + mu.Unlock() + if attempt == 1 { + w.Header().Set("Retry-After", "1") + w.WriteHeader(503) + return + } + if mode == "v1" { + io.WriteString(w, `{"results":{"`+id+`":{"result":"ok"}}}`) + } else { + io.WriteString(w, `{}`) + } + }) + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`","enable_compression":true}`) + result := action(t, captureHandler, `{"event":"test","distinct_id":"user"}`) + if err := uuid.Validate(result["uuid"].(string)); err != nil { + t.Fatal(err) + } + start := time.Now() + action(t, flushHandler, `{}`) + if time.Since(start) < 900*time.Millisecond { + t.Fatal("flush returned before SDK retry completion") + } + mu.Lock() + if !reflect.DeepEqual(ids, []string{result["uuid"].(string), result["uuid"].(string)}) { + t.Errorf("wire UUIDs = %v, response = %v", ids, result) + } + mu.Unlock() + state.mu.Lock() + defer state.mu.Unlock() + if state.pendingEvents != 0 || state.totalEventsSent != 1 || state.totalEventsCaptured != 1 || state.totalRetries != 1 { + t.Fatalf("pending=%d sent=%d captured=%d retries=%d", state.pendingEvents, state.totalEventsSent, state.totalEventsCaptured, state.totalRetries) + } + for i, request := range state.requestsMade { + if request.EventCount != 1 || request.RetryAttempt != i || !reflect.DeepEqual(request.UUIDList, []string{result["uuid"].(string)}) { + t.Errorf("tracked request = %+v", request) + } + } + }) } - w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte(`{"featureFlags":{"example":true}}`)) - })) - defer server.Close() - - resp, err := postFlagsWithRetry(server.URL, []byte(`{"flag_keys_to_evaluate":["example"]}`)) - if err != nil { - t.Fatal(err) } - defer resp.Body.Close() +} - if requests != 2 { - t.Fatalf("requests = %d, want 2", requests) +func TestResetClosesPendingClientBeforeClearingState(t *testing.T) { + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, `{}`) }) + for _, handler := range []http.HandlerFunc{resetHandler, initHandler} { + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`","flush_at":100,"flush_interval_ms":10000}`) + action(t, captureHandler, `{"event":"test","distinct_id":"user"}`) + action(t, handler, `{"api_key":"test-key","host":"`+server.URL+`"}`) + state.mu.Lock() + if state.pendingEvents != 0 || state.totalEventsSent != 0 || state.totalEventsCaptured != 0 || len(state.requestsMade) != 0 { + t.Error("old-client delivery contaminated reset state") + } + state.mu.Unlock() } } -func TestPostFlagsWithRetryStopsAfterRetryableFailures(t *testing.T) { - requests := 0 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requests++ - w.WriteHeader(http.StatusGatewayTimeout) - })) - defer server.Close() +func TestFlushCompletesOnTerminalFailure(t *testing.T) { + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(400) }) + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`"}`) + action(t, captureHandler, `{"event":"test","distinct_id":"user"}`) + action(t, flushHandler, `{}`) + state.mu.Lock() + defer state.mu.Unlock() + if state.pendingEvents != 0 || state.totalEventsSent != 0 || state.lastError == "" { + t.Fatalf("terminal state: pending=%d sent=%d error=%q", state.pendingEvents, state.totalEventsSent, state.lastError) + } +} - resp, err := postFlagsWithRetry(server.URL, []byte(`{}`)) - if err == nil { - if resp != nil { - resp.Body.Close() +func TestV1PartialCompletionCountsOnlySuccessfulEvents(t *testing.T) { + profile(t, "v1", "gzip") + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { + var body struct { + Batch []struct { + UUID string `json:"uuid"` + } `json:"batch"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + return + } + if len(body.Batch) != 2 { + t.Errorf("batch size = %d", len(body.Batch)) + return } - t.Fatal("expected error") + io.WriteString(w, `{"results":{"`+body.Batch[0].UUID+`":{"result":"ok"},"`+body.Batch[1].UUID+`":{"result":"drop"}}}`) + }) + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`","flush_at":2,"flush_interval_ms":10000}`) + action(t, captureHandler, `{"event":"first","distinct_id":"user"}`) + action(t, captureHandler, `{"event":"second","distinct_id":"user"}`) + action(t, flushHandler, `{}`) + state.mu.Lock() + defer state.mu.Unlock() + if state.pendingEvents != 0 || state.totalEventsSent != 1 || state.totalEventsCaptured != 2 || state.lastError == "" { + t.Fatalf("partial completion: pending=%d sent=%d captured=%d error=%q", state.pendingEvents, state.totalEventsSent, state.totalEventsCaptured, state.lastError) } +} - if requests != 2 { - t.Fatalf("requests = %d, want 2", requests) +func TestFlushDoesNotClaimCompletionWhenCanceled(t *testing.T) { + server := mockServer(t, func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, `{}`) }) + action(t, initHandler, `{"api_key":"test-key","host":"`+server.URL+`","flush_at":100,"flush_interval_ms":10000}`) + action(t, captureHandler, `{"event":"test","distinct_id":"user"}`) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + recorder := httptest.NewRecorder() + flushHandler(recorder, httptest.NewRequest("POST", "/flush", nil).WithContext(ctx)) + if recorder.Code != http.StatusGatewayTimeout { + t.Fatalf("flush status = %d", recorder.Code) + } + state.mu.Lock() + defer state.mu.Unlock() + if state.pendingEvents != 1 || state.totalEventsSent != 0 { + t.Fatal("flush changed pending SDK delivery") } } -func TestPostFlagsWithRetryDoesNotRetryNonRetryableStatus(t *testing.T) { - requests := 0 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requests++ - w.WriteHeader(http.StatusBadRequest) - })) - defer server.Close() - - resp, err := postFlagsWithRetry(server.URL, []byte(`{}`)) - if err == nil { - if resp != nil { - resp.Body.Close() +func TestValidateHarnessHost(t *testing.T) { + for _, host := range []string{"http://127.0.0.1:19230", "http://localhost:8081/", "http://test-harness:8081", "http://[::1]:19231"} { + if _, ok := validateHarnessHost(host); !ok { + t.Errorf("rejected %s", host) + } + } + for _, host := range []string{"https://127.0.0.1:19230", "http://posthog.com:8081", "http://localhost:0", "http://localhost:65536", "http://localhost", "http://user@localhost:8081", "http://localhost:8081/path", "http://localhost:8081?x=1", "http://localhost:8081#fragment"} { + if _, ok := validateHarnessHost(host); ok { + t.Errorf("accepted %s", host) } - t.Fatal("expected error") } +} - if requests != 1 { - t.Fatalf("requests = %d, want 1", requests) +func TestCompressionProfiles(t *testing.T) { + for _, mode := range []string{"v0", "v1"} { + for _, codec := range []string{"gzip", "deflate", "br", "zstd"} { + t.Run(mode+"/"+codec, func(t *testing.T) { + profile(t, mode, codec) + _, err := selectedCompression() + if mode == "v0" && codec != "gzip" { + if err == nil { + t.Fatal("accepted V1 codec for V0") + } + return + } + if err != nil { + t.Fatal(err) + } + health := action(t, healthHandler, `{}`) + if !reflect.DeepEqual(health["capabilities"], []interface{}{"capture_" + mode, "encoding_" + codec}) { + t.Fatalf("capabilities = %v", health) + } + }) + } } } diff --git a/sdk_compliance_adapter/test_check_reports.py b/sdk_compliance_adapter/test_check_reports.py new file mode 100644 index 0000000..f6ece7d --- /dev/null +++ b/sdk_compliance_adapter/test_check_reports.py @@ -0,0 +1,43 @@ +import unittest + +from check_reports import PROFILES, check_report + + +def report_for(profile): + count = PROFILES[profile] + suite = "Capture" if profile.startswith("v0-") else "Capture_V1" + codec = profile.split("-")[1] + capture = ["Non Utc Event Timestamp Is Converted To Utc", f"Sends {codec.title()} Content Encoding"] + capture += [f"Capture {index}" for index in range(count - len(capture))] + flags = [ + "Retries Flags On 502", + "Retries Flags On 504", + "Disable Geoip False Propagates As Geoip Disable False", + "Disable Geoip Omitted Defaults To False", + ] + flags += [f"Flag {index}" for index in range(17 - len(flags))] + report = f"**{count + 15}/{count + 17}** tests passed, **2** failed\n" + for name, rows in [(suite, capture), ("Feature_Flags", flags)]: + report += f"## {name} Tests\n" + for row in rows: + status = "❌" if row.startswith("Disable Geoip") else "✅" + report += f"| {row} | {status} | 1ms |\n" + return report + + +class CheckReportsTests(unittest.TestCase): + def test_all_profile_inventories_accept_advisory_failures(self): + for profile, count in PROFILES.items(): + self.assertIn(f"selected={count+17}", check_report(profile, report_for(profile))) + + def test_missing_empty_truncated_or_wrong_inventory_fails(self): + report = report_for("v1-deflate") + for invalid in ["", report[:100], report.replace("Capture 0", "Capture 1"), + report.replace("111", "0"), report.replace("Sends Deflate", "Sends Gzip"), + report.replace("Non Utc Event Timestamp Is Converted To Utc", "Different Test")]: + with self.assertRaises(ValueError): + check_report("v1-deflate", invalid) + + +if __name__ == "__main__": + unittest.main()