From b85a11860db12fc7fefb7927ab1e15ba01d8a5e2 Mon Sep 17 00:00:00 2001 From: Ben White Date: Mon, 5 Aug 2024 11:55:49 +0200 Subject: [PATCH] Added failing tests --- posthog/cdp/test/test_validation.py | 148 ++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 posthog/cdp/test/test_validation.py diff --git a/posthog/cdp/test/test_validation.py b/posthog/cdp/test/test_validation.py new file mode 100644 index 0000000000000..48b29b4574a74 --- /dev/null +++ b/posthog/cdp/test/test_validation.py @@ -0,0 +1,148 @@ +import json +from typing import Any, Optional +from unittest.mock import ANY, patch + +from inline_snapshot import snapshot +import pytest +from rest_framework import status + +from posthog.cdp.validation import validate_inputs, validate_inputs_schema +from posthog.constants import AvailableFeature +from posthog.models.action.action import Action +from posthog.models.hog_functions.hog_function import HogFunction +from posthog.test.base import APIBaseTest, ClickhouseTestMixin, QueryMatchingTest +from posthog.cdp.templates.webhook.template_webhook import template as template_webhook +from posthog.cdp.templates.slack.template_slack import template as template_slack + + +def create_example_inputs_schema(): + return [ + {"key": "url", "type": "string", "label": "Webhook URL", "required": True}, + {"key": "payload", "type": "json", "label": "JSON Payload", "required": True}, + { + "key": "method", + "type": "choice", + "label": "HTTP Method", + "choices": [ + {"label": "POST", "value": "POST"}, + {"label": "PUT", "value": "PUT"}, + {"label": "PATCH", "value": "PATCH"}, + {"label": "GET", "value": "GET"}, + ], + "required": True, + }, + {"key": "headers", "type": "dictionary", "label": "Headers", "required": False}, + ] + + +def create_example_inputs(): + return { + "url": { + "value": "http://localhost:2080/0e02d917-563f-4050-9725-aad881b69937", + }, + "method": {"value": "POST"}, + "headers": { + "value": {"version": "v={event.properties.$lib_version}"}, + }, + "payload": { + "value": { + "event": "{event}", + "groups": "{groups}", + "nested": {"foo": "{event.url}"}, + "person": "{person}", + "event_url": "{f'{event.url}-test'}", + }, + }, + } + + +class TestHogFunctionValidation(ClickhouseTestMixin, APIBaseTest, QueryMatchingTest): + def test_validate_inputs_schema(self): + inputs_schema = create_example_inputs_schema() + assert validate_inputs_schema(inputs_schema) == snapshot( + [ + {"type": "string", "key": "url", "label": "Webhook URL", "required": True, "secret": False}, + {"type": "json", "key": "payload", "label": "JSON Payload", "required": True, "secret": False}, + { + "type": "choice", + "key": "method", + "label": "HTTP Method", + "choices": [ + {"label": "POST", "value": "POST"}, + {"label": "PUT", "value": "PUT"}, + {"label": "PATCH", "value": "PATCH"}, + {"label": "GET", "value": "GET"}, + ], + "required": True, + "secret": False, + }, + {"type": "dictionary", "key": "headers", "label": "Headers", "required": False, "secret": False}, + ] + ) + + def test_validate_inputs(self): + inputs_schema = create_example_inputs_schema() + inputs = create_example_inputs() + assert json.loads(json.dumps(validate_inputs(inputs_schema, inputs))) == snapshot( + { + "url": { + "value": "http://localhost:2080/0e02d917-563f-4050-9725-aad881b69937", + "bytecode": ["_h", 32, "http://localhost:2080/0e02d917-563f-4050-9725-aad881b69937"], + }, + "payload": { + "value": { + "event": "{event}", + "groups": "{groups}", + "nested": {"foo": "{event.url}"}, + "person": "{person}", + "event_url": "{f'{event.url}-test'}", + }, + "bytecode": { + "event": ["_h", 32, "event", 1, 1], + "groups": ["_h", 32, "groups", 1, 1], + "nested": {"foo": ["_h", 32, "url", 32, "event", 1, 2]}, + "person": ["_h", 32, "person", 1, 1], + "event_url": ["_h", 32, "-test", 32, "url", 32, "event", 1, 2, 2, "concat", 2], + }, + }, + "method": {"value": "POST"}, + "headers": { + "value": {"version": "v={event.properties.$lib_version}"}, + "bytecode": { + "version": [ + "_h", + 32, + "$lib_version", + 32, + "properties", + 32, + "event", + 1, + 3, + 32, + "v=", + 2, + "concat", + 2, + ] + }, + }, + } + ) + + def test_validate_inputs_creates_bytecode_for_html(self): + # NOTE: This currently fails due to the css blocks being interpolated as hog + html_with_css = '\n\n\n\n\n\n

Hi {person.properties.email}

\n\n' + + assert json.loads( + json.dumps( + validate_inputs( + [ + {"key": "html", "type": "string", "label": "HTML", "required": True}, + ], + { + "html": {"value": html_with_css}, + }, + ) + ) + ) == snapshot("")