mirror of
https://github.com/PostHog/posthog.git
synced 2026-10-26 09:51:43 +08:00
1424 lines
50 KiB
Python
1424 lines
50 KiB
Python
import json
|
|
import time
|
|
from collections.abc import Callable
|
|
from datetime import timedelta
|
|
from typing import Any, Optional, cast
|
|
|
|
import pytest
|
|
from unittest.mock import patch
|
|
|
|
from parameterized import parameterized
|
|
|
|
from posthog.hogql.compiler.bytecode import create_bytecode
|
|
from posthog.hogql.parser import parse_expr, parse_program, parse_string_template
|
|
|
|
from common.hogvm.python.execute import execute_bytecode, get_nested_value
|
|
from common.hogvm.python.operation import (
|
|
HOGQL_BYTECODE_IDENTIFIER as _H,
|
|
HOGQL_BYTECODE_VERSION as VERSION,
|
|
Operation as op,
|
|
)
|
|
from common.hogvm.python.stl import _MAX_SEQUENCE_LENGTH, STL, _guard_sequence_length, sleep
|
|
from common.hogvm.python.utils import (
|
|
COST_PER_UNIT,
|
|
MAX_MEMORY,
|
|
MAX_REGEX_PATTERN_LENGTH,
|
|
HogVMException,
|
|
HogVMMemoryExceededException,
|
|
HogVMRuntimeExceededException,
|
|
UncaughtHogVMException,
|
|
)
|
|
|
|
|
|
class TestBytecodeExecute:
|
|
def _run(self, expr: str) -> Any:
|
|
globals = {
|
|
"properties": {"foo": "bar", "nullValue": None},
|
|
}
|
|
return execute_bytecode(create_bytecode(parse_expr(expr)).bytecode, globals).result
|
|
|
|
def _run_program(
|
|
self, code: str, functions: Optional[dict[str, Callable[..., Any]]] = None, globals: Optional[dict] = None
|
|
) -> Any:
|
|
if not globals:
|
|
globals = {
|
|
"properties": {"foo": "bar", "nullValue": None},
|
|
}
|
|
program = parse_program(code)
|
|
bytecode = create_bytecode(program, supported_functions=set(functions.keys()) if functions else None).bytecode
|
|
response = execute_bytecode(bytecode, globals, functions)
|
|
return response.result
|
|
|
|
def test_bytecode_create(self):
|
|
assert self._run("1 + 2") == 3
|
|
assert self._run("1 - 2") == -1
|
|
assert self._run("3 * 2") == 6
|
|
assert self._run("3 / 2") == 1.5
|
|
assert self._run("3 % 2") == 1
|
|
assert self._run("1 and 2") is True
|
|
assert self._run("1 or 0") is True
|
|
assert self._run("1 and 0") is False
|
|
assert self._run("1 or (0 and 1) or 2") is True
|
|
assert self._run("(1 and 0) and 1") is False
|
|
assert self._run("(1 or 2) and (1 or 2)") is True
|
|
assert self._run("true") is True
|
|
assert self._run("not true") is False
|
|
assert self._run("false") is False
|
|
assert self._run("null") is None
|
|
assert self._run("3.14") == 3.14
|
|
assert self._run("1 = 2") is False
|
|
assert self._run("1 == 2") is False
|
|
assert self._run("1 != 2") is True
|
|
assert self._run("1 < 2") is True
|
|
assert self._run("1 <= 2") is True
|
|
assert self._run("1 > 2") is False
|
|
assert self._run("1 >= 2") is False
|
|
assert self._run("'a' like 'b'") is False
|
|
assert self._run("'baa' like '%a%'") is True
|
|
assert self._run("'baa' like '%x%'") is False
|
|
assert self._run("'baa' ilike '%A%'") is True
|
|
assert self._run("'baa' ilike '%C%'") is False
|
|
assert self._run("'a' ilike 'b'") is False
|
|
assert self._run("'a' not like 'b'") is True
|
|
assert self._run("'a' not ilike 'b'") is True
|
|
assert self._run("'a' in 'car'") is True
|
|
assert self._run("'a' in 'foo'") is False
|
|
assert self._run("'a' not in 'car'") is False
|
|
assert self._run("properties.bla") is None
|
|
assert self._run("properties.foo") == "bar"
|
|
assert self._run("ifNull(properties.foo, false)") == "bar"
|
|
assert self._run("ifNull(properties.nullValue, false)") is False
|
|
assert self._run("concat('arg', 'another')") == "arganother"
|
|
assert self._run("concat(1, NULL)") == "1"
|
|
assert self._run("concat(true, false)") == "truefalse"
|
|
assert self._run("match('test', 'e.*')") is True
|
|
assert self._run("match('test', '^e.*')") is False
|
|
assert self._run("match('test', 'x.*')") is False
|
|
assert self._run("match('test', '')") is True
|
|
assert self._run("match('', '')") is True
|
|
assert self._run("'test' =~ 'e.*'") is True
|
|
assert self._run("'test' !~ 'e.*'") is False
|
|
assert self._run("'test' =~ '^e.*'") is False
|
|
assert self._run("'test' !~ '^e.*'") is True
|
|
assert self._run("'test' =~ 'x.*'") is False
|
|
assert self._run("'test' !~ 'x.*'") is True
|
|
assert self._run("'' !~ 'x.*'") is False
|
|
assert self._run("'test' ~* 'EST'") is True
|
|
assert self._run("'test' =~* 'EST'") is True
|
|
assert self._run("'test' !~* 'EST'") is False
|
|
assert self._run("toString(1)") == "1"
|
|
assert self._run("toString(1.5)") == "1.5"
|
|
assert self._run("toString(true)") == "true"
|
|
assert self._run("toString(null)") == "null"
|
|
assert self._run("toString('string')") == "string"
|
|
assert self._run("toInt('1')") == 1
|
|
assert self._run("toInt('bla')") is None
|
|
assert self._run("toInt(null)") is None
|
|
assert self._run("toFloat('1.2')") == 1.2
|
|
assert self._run("toFloat('bla')") is None
|
|
assert self._run("toFloat(null)") is None
|
|
assert self._run("toUUID('asd')") == "asd"
|
|
assert self._run("1 == null") is False
|
|
assert self._run("1 != null") is True
|
|
|
|
def test_ordering_comparison_type_error_raises_hogvm_exception(self):
|
|
with pytest.raises(HogVMException, match="'<=' not supported between instances of 'list' and 'float'"):
|
|
self._run("[1] <= 1.0")
|
|
|
|
@parameterized.expand(
|
|
[
|
|
("function_list_input", "match(['tool_call'], 'tool')", {}, "Function match requires input"),
|
|
("function_invalid_pattern", "match('tool_call', '[')", {}, "Invalid regex pattern"),
|
|
("function_lookbehind_unsupported", "match('ab', '(?<=a)b')", {}, "Invalid regex pattern"),
|
|
("operator_list_input", "['tool_call'] =~ 'tool'", {}, "Function match requires input"),
|
|
(
|
|
"operator_invalid_pattern",
|
|
"'tool_call' =~ properties.pattern",
|
|
{"pattern": "\\u"},
|
|
"Invalid regex pattern: invalid escape sequence: \\u",
|
|
),
|
|
]
|
|
)
|
|
def test_regex_errors_raise_hogvm_exception(self, _name, expr, properties, expected_message):
|
|
globals_dict = {"properties": properties}
|
|
bytecode = create_bytecode(parse_expr(expr)).bytecode
|
|
|
|
with pytest.raises(HogVMException) as exc_info:
|
|
execute_bytecode(bytecode, globals_dict)
|
|
|
|
assert expected_message in str(exc_info.value)
|
|
|
|
@pytest.mark.parametrize(
|
|
"expression, expected",
|
|
[
|
|
("match(input, pattern)", True),
|
|
("extractRegex(input, pattern)", "needle"),
|
|
("like(input, pattern)", True),
|
|
("ilike(input, pattern)", True),
|
|
("notLike(input, pattern)", False),
|
|
("notILike(input, pattern)", False),
|
|
("input like pattern", True),
|
|
("input ilike pattern", True),
|
|
("input not like pattern", False),
|
|
("input not ilike pattern", False),
|
|
("input =~ pattern", True),
|
|
("input !~ pattern", False),
|
|
("input =~* pattern", True),
|
|
("input !~* pattern", False),
|
|
],
|
|
)
|
|
@pytest.mark.parametrize("oversized_pattern", [False, True])
|
|
def test_matching_accepts_large_subjects_but_bounds_patterns(
|
|
self, expression: str, expected: bool | str, oversized_pattern: bool
|
|
) -> None:
|
|
bytecode = create_bytecode(parse_expr(expression)).bytecode
|
|
globals_dict = {
|
|
"input": "z" * (8 * 1024 * 1024) + "needle",
|
|
"pattern": "z" * (MAX_REGEX_PATTERN_LENGTH + 1) if oversized_pattern else "needle",
|
|
}
|
|
|
|
if oversized_pattern:
|
|
with pytest.raises(HogVMException, match=f"exceeds {MAX_REGEX_PATTERN_LENGTH} characters"):
|
|
execute_bytecode(bytecode, globals_dict, timeout=60)
|
|
else:
|
|
assert execute_bytecode(bytecode, globals_dict, timeout=60).result == expected
|
|
|
|
@pytest.mark.parametrize("function_name", ["match", "extractRegex", "like", "ilike"])
|
|
def test_regex_pattern_limit_is_inclusive(self, function_name: str) -> None:
|
|
bytecode = create_bytecode(parse_expr(f"{function_name}(input, pattern)")).bytecode
|
|
pattern = "z" * MAX_REGEX_PATTERN_LENGTH
|
|
result = execute_bytecode(bytecode, {"input": "z", "pattern": pattern}, timeout=60).result
|
|
|
|
assert result == ("" if function_name == "extractRegex" else False)
|
|
|
|
def test_nested_value(self):
|
|
my_dict = {
|
|
"properties": {
|
|
"bla": "hello",
|
|
"list": ["item1", "item2", "item3"],
|
|
"tuple": ("item1", "item2", "item3"),
|
|
}
|
|
}
|
|
chain: list[Any] = ["properties", "bla"]
|
|
assert get_nested_value(my_dict, chain) == "hello"
|
|
|
|
chain = ["properties", "list", 2]
|
|
assert get_nested_value(my_dict, chain) == "item2"
|
|
|
|
chain = ["properties", "tuple", 3]
|
|
assert get_nested_value(my_dict, chain) == "item3"
|
|
|
|
def test_errors(self):
|
|
try:
|
|
execute_bytecode([_H, VERSION, op.TRUE, op.CALL_GLOBAL, "notAFunction", 1], {})
|
|
except Exception as e:
|
|
assert str(e) == "Unsupported function call: notAFunction"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
try:
|
|
execute_bytecode([_H, VERSION, op.CALL_GLOBAL, "replaceOne", 1], {})
|
|
except Exception as e:
|
|
assert str(e) == "Function replaceOne requires at least 3 arguments"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
try:
|
|
execute_bytecode([_H, VERSION, op.STRING, "AB", op.STRING, "extra", op.CALL_GLOBAL, "lower", 2], {})
|
|
except Exception as e:
|
|
assert str(e) == "Function lower requires at most 1 arguments"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
try:
|
|
execute_bytecode([_H, VERSION, op.CALL_GLOBAL, "lower", 1], {})
|
|
except Exception as e:
|
|
assert str(e) == "Stack underflow"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
try:
|
|
execute_bytecode([_H, VERSION, op.TRUE, op.TRUE, op.NOT], {})
|
|
except Exception as e:
|
|
assert str(e) == "Invalid bytecode. More than one value left on stack"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
@pytest.mark.parametrize("indirect", [False, True])
|
|
def test_json_has_without_path(self, indirect: bool) -> None:
|
|
program = "let hasPath := JSONHas; return hasPath('{}');" if indirect else "return JSONHas('{}');"
|
|
assert self._run_program(program) is True
|
|
|
|
def test_every_builtin_tolerates_its_own_min_args(self):
|
|
# A builtin whose fn indexes past its declared minArgs raises a bare IndexError instead of a
|
|
# HogVMException, which callers cannot tell apart from a bug in their own code. Blocking
|
|
# builtins are excluded because calling them would sleep or shell out.
|
|
leaked = []
|
|
for name, stl_fn in STL.items():
|
|
if stl_fn.is_blocking:
|
|
continue
|
|
arg_count = stl_fn.minArgs or 0
|
|
bytecode: list[Any] = [_H, VERSION]
|
|
for _ in range(arg_count):
|
|
bytecode += [op.STRING, "1"]
|
|
bytecode += [op.CALL_GLOBAL, name, arg_count]
|
|
try:
|
|
execute_bytecode(bytecode, {})
|
|
except IndexError:
|
|
leaked.append(name)
|
|
except Exception:
|
|
pass
|
|
|
|
assert leaked == []
|
|
|
|
def test_memory_limits_1(self):
|
|
# let string := 'banana'
|
|
# for (let i := 0; i < 100; i := i + 1) {
|
|
# string := string || string
|
|
# }
|
|
bytecode = [
|
|
"_h",
|
|
32,
|
|
"banana",
|
|
33,
|
|
0,
|
|
33,
|
|
100,
|
|
36,
|
|
1,
|
|
15,
|
|
40,
|
|
18,
|
|
36,
|
|
0,
|
|
36,
|
|
0,
|
|
2,
|
|
"concat",
|
|
2,
|
|
37,
|
|
0,
|
|
33,
|
|
1,
|
|
36,
|
|
1,
|
|
6,
|
|
37,
|
|
1,
|
|
39,
|
|
-25,
|
|
35,
|
|
35,
|
|
]
|
|
try:
|
|
execute_bytecode(bytecode, {})
|
|
except Exception as e:
|
|
assert str(e) == "Memory limit of 67108864 bytes exceeded. Attempted to use 75497504 bytes"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
def test_memory_limits_2(self):
|
|
bytecode = [
|
|
"_h",
|
|
32,
|
|
"key",
|
|
32,
|
|
"value",
|
|
32,
|
|
"key2",
|
|
32,
|
|
"value2",
|
|
42,
|
|
2,
|
|
32,
|
|
"na",
|
|
33,
|
|
0,
|
|
33,
|
|
10000,
|
|
36,
|
|
2,
|
|
15,
|
|
40,
|
|
52,
|
|
33,
|
|
16,
|
|
36,
|
|
2,
|
|
15,
|
|
40,
|
|
9,
|
|
36,
|
|
1,
|
|
36,
|
|
1,
|
|
2,
|
|
"concat",
|
|
2,
|
|
37,
|
|
1,
|
|
36,
|
|
0,
|
|
36,
|
|
2,
|
|
32,
|
|
"key_",
|
|
2,
|
|
"concat",
|
|
2,
|
|
32,
|
|
"wasted",
|
|
32,
|
|
" batman!",
|
|
36,
|
|
1,
|
|
32,
|
|
"memory: ",
|
|
2,
|
|
"concat",
|
|
3,
|
|
32,
|
|
"something",
|
|
36,
|
|
0,
|
|
42,
|
|
2,
|
|
46,
|
|
33,
|
|
1,
|
|
36,
|
|
2,
|
|
6,
|
|
37,
|
|
2,
|
|
39,
|
|
-59,
|
|
35,
|
|
35,
|
|
35,
|
|
]
|
|
try:
|
|
execute_bytecode(bytecode, {})
|
|
except Exception as e:
|
|
assert str(e) == "Memory limit of 67108864 bytes exceeded. Attempted to use 67155164 bytes"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
def test_range_refuses_length_past_memory_ceiling(self):
|
|
# The asserted size proves the guard fired before allocation, not after building the list.
|
|
length = 10**12
|
|
bytecode = [_H, VERSION, op.INTEGER, length, op.CALL_GLOBAL, "range", 1, op.RETURN]
|
|
with pytest.raises(HogVMMemoryExceededException) as exc:
|
|
execute_bytecode(bytecode, {})
|
|
assert exc.value.attempted_memory == (length + 1) * COST_PER_UNIT
|
|
|
|
def test_range_ceiling_matches_stack_accounting(self):
|
|
# The ceiling is the largest length the stack accepts, so a list at the ceiling stays within
|
|
# the limit and one past it is refused. Locks the boundary without allocating either list.
|
|
assert (_MAX_SEQUENCE_LENGTH + 1) * COST_PER_UNIT <= MAX_MEMORY
|
|
_guard_sequence_length(_MAX_SEQUENCE_LENGTH)
|
|
with pytest.raises(HogVMMemoryExceededException):
|
|
_guard_sequence_length(_MAX_SEQUENCE_LENGTH + 1)
|
|
|
|
@parameterized.expand([("range(0, 200)",), ("(range)(0, 200)",)])
|
|
def test_range_checks_remaining_memory_before_allocating(self, expression: str) -> None:
|
|
bytecode = create_bytecode(parse_program("let retained := '" + "x" * 1000 + "'; return " + expression)).bytecode
|
|
with patch("common.hogvm.python.stl.list", side_effect=AssertionError("Allocated range"), create=True):
|
|
with pytest.raises(HogVMMemoryExceededException):
|
|
execute_bytecode(bytecode, memory_limit=2048)
|
|
|
|
@parameterized.expand([("range(7)", 7), ("range(3, 10)", 7), ("range(-1)", 0), ("range(10, 3)", 0)])
|
|
def test_range_within_memory_limit(self, expression: str, expected_length: int) -> None:
|
|
bytecode = create_bytecode(parse_expr(expression)).bytecode
|
|
response = execute_bytecode(bytecode, memory_limit=64)
|
|
assert len(response.result) == expected_length
|
|
|
|
@parameterized.expand([(op.RETURN,), (None,), ()])
|
|
def test_peak_memory_includes_temporary_values(self, *ending: op | None) -> None:
|
|
bytecode = [_H, VERSION, op.INTEGER, 7, op.CALL_GLOBAL, "range", 1, op.CALL_GLOBAL, "length", 1, *ending]
|
|
response = execute_bytecode(bytecode, memory_limit=64)
|
|
assert response.result == 7
|
|
assert response.max_memory_used == 64
|
|
|
|
@parameterized.expand([("length('hello')",), ("(length)('hello')",)])
|
|
def test_stl_call_cannot_return_after_deadline(self, expression: str) -> None:
|
|
bytecode = create_bytecode(parse_expr(expression)).bytecode
|
|
with patch("common.hogvm.python.execute.time.monotonic", side_effect=[0.0, 0.0, 0.0, 2.0]):
|
|
with pytest.raises(HogVMRuntimeExceededException):
|
|
execute_bytecode(bytecode, timeout=timedelta(seconds=1))
|
|
|
|
def test_functions(self):
|
|
def stringify(*args):
|
|
if args[0] == 1:
|
|
return "one"
|
|
elif args[0] == 2:
|
|
return "two"
|
|
return "zero"
|
|
|
|
functions = {"stringify": stringify}
|
|
assert (
|
|
execute_bytecode(
|
|
[_H, VERSION, op.INTEGER, 1, op.CALL_GLOBAL, "stringify", 1, op.RETURN], {}, functions
|
|
).result
|
|
== "one"
|
|
)
|
|
assert (
|
|
execute_bytecode(
|
|
[_H, VERSION, op.INTEGER, 2, op.CALL_GLOBAL, "stringify", 1, op.RETURN], {}, functions
|
|
).result
|
|
== "two"
|
|
)
|
|
assert (
|
|
execute_bytecode(
|
|
[_H, VERSION, op.STRING, "2", op.CALL_GLOBAL, "stringify", 1, op.RETURN], {}, functions
|
|
).result
|
|
== "zero"
|
|
)
|
|
|
|
def test_version_0_and_1(self):
|
|
# version 0 of HogQL bytecode had arguments in a different order
|
|
assert (
|
|
execute_bytecode(["_h", op.STRING, "1", op.STRING, "2", op.CALL_GLOBAL, "concat", 2, op.RETURN]).result
|
|
== "21"
|
|
)
|
|
assert (
|
|
execute_bytecode(["_H", 1, op.STRING, "1", op.STRING, "2", op.CALL_GLOBAL, "concat", 2, op.RETURN]).result
|
|
== "12"
|
|
)
|
|
|
|
def test_bytecode_variable_assignment(self):
|
|
program = parse_program("let a := 1 + 2; return a;")
|
|
bytecode = create_bytecode(program).bytecode
|
|
assert bytecode == ["_H", 1, op.INTEGER, 2, op.INTEGER, 1, op.PLUS, op.GET_LOCAL, 0, op.RETURN, op.POP]
|
|
|
|
assert self._run_program("let a := 1 + 2; return a;") == 3
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let a := 1 + 2;
|
|
let b := a + 4;
|
|
return b;
|
|
"""
|
|
)
|
|
== 7
|
|
)
|
|
|
|
def test_bytecode_if_else(self):
|
|
program = parse_program("if (true) return 1; else return 2;")
|
|
bytecode = create_bytecode(program).bytecode
|
|
assert bytecode == [
|
|
"_H",
|
|
1,
|
|
op.TRUE,
|
|
op.JUMP_IF_FALSE,
|
|
5,
|
|
op.INTEGER,
|
|
1,
|
|
op.RETURN,
|
|
op.JUMP,
|
|
3,
|
|
op.INTEGER,
|
|
2,
|
|
op.RETURN,
|
|
]
|
|
|
|
assert self._run_program("if (true) return 1; else return 2;") == 1
|
|
|
|
assert self._run_program("if (false) return 1; else return 2;") == 2
|
|
|
|
assert self._run_program("if (true) { return 1; } else { return 2; }") == 1
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let a := true;
|
|
if (a) {
|
|
let a := 3;
|
|
return a + 2;
|
|
} else {
|
|
return 2;
|
|
}
|
|
"""
|
|
)
|
|
== 5
|
|
)
|
|
|
|
def test_bytecode_variable_reassignment(self):
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let a := 1;
|
|
a := a + 3;
|
|
a := a * 2;
|
|
return a;
|
|
"""
|
|
)
|
|
== 8
|
|
)
|
|
|
|
@parameterized.expand(
|
|
[
|
|
(
|
|
"call_result_object",
|
|
"let store := {};\nfn ref() { return store; }\nref()['k'] := 'v';\nreturn store.k;",
|
|
"v",
|
|
),
|
|
(
|
|
"call_result_array",
|
|
"let m := [[10, 20], [30, 40]];\nfn row(i) { return m[i]; }\nrow(1)[2] := 99;\nreturn m[1][2];",
|
|
99,
|
|
),
|
|
(
|
|
"call_result_property",
|
|
"let cfg := {'items': [1, 2, 3]};\nfn getCfg() { return cfg; }\ngetCfg().items[2] := 88;\nreturn cfg.items[2];",
|
|
88,
|
|
),
|
|
("object_literal", "{'a': 1}['a'] := 2; return 1;", 1),
|
|
("array_literal", "[1, 2, 3][1] := 99; return 1;", 1),
|
|
]
|
|
)
|
|
def test_assignment_target_with_expression_base(self, _name, program, expected):
|
|
# The compiler visits a subscript/tuple-access base as a plain value, so a
|
|
# call result or literal base is a valid assignment target.
|
|
assert self._run_program(program) == expected
|
|
|
|
def test_assignment_to_non_lvalue_rejected_at_compile_time(self):
|
|
# A non-assignable target is rejected at compile time, after a successful parse.
|
|
with pytest.raises(Exception, match="Can not assign to this type of expression"):
|
|
self._run_program("fn f() { return 1; }\nf() := 1;\nreturn 1;")
|
|
|
|
def test_assignment_to_parenthesized_target(self):
|
|
# A parenthesised target collapses to the underlying place.
|
|
assert self._run_program("let x := 1; (x) := 5; return x;") == 5
|
|
assert self._run_program("let x := 1; ((x)) := 7; return x;") == 7
|
|
assert self._run_program("let o := {}; (o.a) := 3; return o.a;") == 3
|
|
assert self._run_program("let o := {}; (o).a := 4; return o.a;") == 4
|
|
# Adjacent assignments without separators parse as distinct statements.
|
|
assert self._run_program("let a := 0 let b := 0 (a) := 1 (b) := 2 return a + b") == 3
|
|
|
|
def test_bytecode_while(self):
|
|
program = parse_program("while (true) 1 + 1;")
|
|
bytecode = create_bytecode(program).bytecode
|
|
assert bytecode == [
|
|
"_H",
|
|
1,
|
|
op.TRUE,
|
|
op.JUMP_IF_FALSE,
|
|
8,
|
|
op.INTEGER,
|
|
1,
|
|
op.INTEGER,
|
|
1,
|
|
op.PLUS,
|
|
op.POP,
|
|
op.JUMP,
|
|
-11,
|
|
]
|
|
|
|
program = parse_program("while (toString('a')) { 1 + 1; } return 3;")
|
|
bytecode = create_bytecode(program).bytecode
|
|
assert bytecode == [
|
|
"_H",
|
|
1,
|
|
op.STRING,
|
|
"a",
|
|
op.CALL_GLOBAL,
|
|
"toString",
|
|
1,
|
|
op.JUMP_IF_FALSE,
|
|
8,
|
|
op.INTEGER,
|
|
1,
|
|
op.INTEGER,
|
|
1,
|
|
op.PLUS,
|
|
op.POP,
|
|
op.JUMP,
|
|
-15,
|
|
op.INTEGER,
|
|
3,
|
|
op.RETURN,
|
|
]
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let i := -1;
|
|
while (false) {
|
|
1 + 1;
|
|
}
|
|
return i;
|
|
"""
|
|
)
|
|
== -1
|
|
)
|
|
|
|
number_of_times = 0
|
|
|
|
def call_three_times():
|
|
nonlocal number_of_times
|
|
number_of_times += 1
|
|
return number_of_times <= 3
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let i := 0;
|
|
while (call_three_times()) {
|
|
true;
|
|
}
|
|
return i;
|
|
""",
|
|
{"call_three_times": call_three_times, "print": print},
|
|
)
|
|
== 0
|
|
)
|
|
|
|
def test_bytecode_while_var(self):
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let i := 0;
|
|
while (i < 3) {
|
|
i := i + 1;
|
|
}
|
|
return i;
|
|
"""
|
|
)
|
|
== 3
|
|
)
|
|
|
|
def test_bytecode_for(self):
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let j := 0
|
|
for (let i := 0; i < 3; i := i + 1) {
|
|
print(i) -- prints 3 times
|
|
j := j + 2
|
|
}
|
|
// print(i) -- global does not print
|
|
return j
|
|
"""
|
|
)
|
|
== 6
|
|
)
|
|
|
|
def test_bytecode_functions(self):
|
|
program = parse_program(
|
|
"""
|
|
fun add(a, b) {
|
|
return a + b;
|
|
}
|
|
return add(3, 4);
|
|
"""
|
|
)
|
|
bytecode = create_bytecode(program).bytecode
|
|
assert bytecode == [
|
|
"_H",
|
|
VERSION,
|
|
op.CALLABLE,
|
|
"add",
|
|
2,
|
|
0,
|
|
6,
|
|
op.GET_LOCAL,
|
|
1,
|
|
op.GET_LOCAL,
|
|
0,
|
|
op.PLUS,
|
|
op.RETURN,
|
|
op.CLOSURE,
|
|
0,
|
|
op.INTEGER,
|
|
3,
|
|
op.INTEGER,
|
|
4,
|
|
op.GET_LOCAL,
|
|
0,
|
|
op.CALL_LOCAL,
|
|
2,
|
|
op.RETURN,
|
|
op.POP,
|
|
]
|
|
|
|
response = execute_bytecode(bytecode).result
|
|
assert response == 7
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
fun add(a, b) {
|
|
return a + b;
|
|
}
|
|
return add(3, 4) + 100 + add(1, 1);
|
|
"""
|
|
)
|
|
== 109
|
|
)
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
fun add(a, b) {
|
|
return a + b;
|
|
}
|
|
fun divide(a, b) {
|
|
return a / b;
|
|
}
|
|
return divide(add(3, 4) + 100 + add(2, 1), 2);
|
|
"""
|
|
)
|
|
== 55
|
|
)
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
fun add(a, b) {
|
|
let c := a + b;
|
|
return c;
|
|
}
|
|
fun divide(a, b) {
|
|
return a / b;
|
|
}
|
|
return divide(add(3, 4) + 100 + add(2, 1), 10);
|
|
"""
|
|
)
|
|
== 11
|
|
)
|
|
|
|
def test_bytecode_recursion(self):
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
fun fibonacci(number) {
|
|
if (number < 2) {
|
|
return number;
|
|
} else {
|
|
return fibonacci(number - 1) + fibonacci(number - 2);
|
|
}
|
|
}
|
|
return fibonacci(6);
|
|
"""
|
|
)
|
|
== 8
|
|
)
|
|
|
|
def test_bytecode_no_args(self):
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
fun doIt(a) {
|
|
let url := 'basdfasdf';
|
|
let second := 2 + 3;
|
|
return second;
|
|
}
|
|
let nr := doIt(1);
|
|
return nr;
|
|
"""
|
|
)
|
|
== 5
|
|
)
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
fun doIt() {
|
|
let url := 'basdfasdf';
|
|
let second := 2 + 3;
|
|
return second;
|
|
}
|
|
let nr := doIt();
|
|
return nr;
|
|
"""
|
|
)
|
|
== 5
|
|
)
|
|
|
|
def test_bytecode_functions_stl(self):
|
|
assert self._run_program("if (empty('') and notEmpty('234')) return length('123');") == 3
|
|
assert self._run_program("if (lower('Tdd4gh') == 'tdd4gh') return upper('test');") == "TEST"
|
|
assert self._run_program("return reverse('spinner');") == "rennips"
|
|
|
|
def test_bytecode_string_functions_return_null_for_null(self):
|
|
for program in (
|
|
"return length(null);",
|
|
"return upper(null);",
|
|
"return reverse(null);",
|
|
"return replaceOne(null, 'a', 'b');",
|
|
"return replaceAll(null, 'a', 'b');",
|
|
"return trim(null);",
|
|
"return splitByString(' ', null);",
|
|
):
|
|
assert self._run_program(program) is None, program
|
|
|
|
def test_bytecode_ordering_with_a_null_operand_is_false(self):
|
|
# A filter comparing a missing value must not match, and must not fail either.
|
|
for program in (
|
|
"return length(null) > 3;",
|
|
"return length(null) < 3;",
|
|
"return null >= 0;",
|
|
"return 1 > null;",
|
|
"return null > true;",
|
|
"return null < null;",
|
|
):
|
|
assert self._run_program(program) is False, program
|
|
# Equality does not coerce: a zero is not a null.
|
|
assert self._run_program("return 0 == null;") is False
|
|
assert self._run_program("return 0 != null;") is True
|
|
|
|
@parameterized.expand(
|
|
[
|
|
("arg_over_budget_clamped_to_budget", 600.0, 0.5, 0.5),
|
|
("arg_under_budget_left_as_is", 0.1, 5.0, 0.1),
|
|
("negative_arg_clamped_to_zero", -5.0, 5.0, 0.0),
|
|
]
|
|
)
|
|
def test_sleep_bounds_duration_to_remaining_timeout(self, _name, arg, budget, expected):
|
|
with patch("common.hogvm.python.stl.time.sleep") as mock_sleep:
|
|
sleep([arg], None, None, budget)
|
|
mock_sleep.assert_called_once_with(expected)
|
|
|
|
@parameterized.expand(
|
|
[
|
|
("direct_call", "sleep(0)"), # CALL_GLOBAL dispatch
|
|
("expression_call", "(sleep)(0)"), # CALL_LOCAL closure dispatch
|
|
]
|
|
)
|
|
def test_disallowed_functions_rejected_at_dispatch(self, _name, expr):
|
|
bytecode = create_bytecode(parse_expr(expr)).bytecode
|
|
execute_bytecode(bytecode, {}) # allowed by default (sleeps 0s)
|
|
with pytest.raises(HogVMException, match="Function sleep is not allowed here"):
|
|
execute_bytecode(bytecode, {}, disallowed_functions=frozenset({"sleep"}))
|
|
|
|
def test_random_float(self):
|
|
for _ in range(50):
|
|
value = self._run_program("return randomFloat();")
|
|
assert isinstance(value, float)
|
|
assert 0.0 <= value < 1.0
|
|
|
|
def test_bytecode_empty_statements(self):
|
|
assert self._run_program(";") is None
|
|
assert self._run_program(";;") is None
|
|
assert self._run_program(";;return 1;;") == 1
|
|
assert self._run_program("return 1;;") == 1
|
|
assert self._run_program("return 1;") == 1
|
|
assert self._run_program("return 1;return 2;") == 1
|
|
assert self._run_program("return 1;return 2;;") == 1
|
|
assert self._run_program("return 1;return 2;return 3;") == 1
|
|
assert self._run_program("return 1;return 2;return 3;;") == 1
|
|
|
|
def test_bytecode_dicts(self):
|
|
assert self._run_program("return {};") == {}
|
|
assert self._run_program("return {'key': 'value'};") == {"key": "value"}
|
|
assert self._run_program("return {'key': 'value', 'other': 'thing'};") == {"key": "value", "other": "thing"}
|
|
assert self._run_program("return {'key': {'otherKey': 'value'}};") == {"key": {"otherKey": "value"}}
|
|
try:
|
|
self._run_program("return {key: 'value'};")
|
|
except Exception as e:
|
|
assert str(e) == "Global variable not found: key"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
assert self._run_program("let key := 3; return {key: 'value'};") == {3: "value"}
|
|
|
|
assert self._run_program("return {'key': 'value'}.key;") == "value"
|
|
assert self._run_program("return {'key': 'value'}['key'];") == "value"
|
|
assert self._run_program("return {'key': {'otherKey': 'value'}}.key.otherKey;") == "value"
|
|
assert self._run_program("return {'key': {'otherKey': 'value'}}['key'].otherKey;") == "value"
|
|
|
|
def test_bytecode_arrays(self):
|
|
assert self._run_program("return [];") == []
|
|
assert self._run_program("return [1, 2, 3];") == [1, 2, 3]
|
|
assert self._run_program("return [1, '2', 3];") == [1, "2", 3]
|
|
assert self._run_program("return [1, [2, 3], 4];") == [1, [2, 3], 4]
|
|
assert self._run_program("return [1, [2, [3, 4]], 5];") == [1, [2, [3, 4]], 5]
|
|
|
|
assert self._run_program("let a := [1, 2, 3]; return a[2];") == 2
|
|
assert self._run_program("return [1, 2, 3][2];") == 2
|
|
assert self._run_program("return [1, [2, [3, 4]], 5][2][2][2];") == 4
|
|
assert self._run_program("return [1, [2, [3, 4]], 5][2][2][2] + 1;") == 5
|
|
assert self._run_program("return [1, [2, [3, 4]], 5].2.2.2;") == 4
|
|
|
|
try:
|
|
self._run_program("return [1, 2, 3][0]")
|
|
except Exception as e:
|
|
assert str(e) == "Array access starts from 1"
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
with pytest.raises(HogVMException, match="Index 1 out of range for array of length 0"):
|
|
self._run_program("let calls := []; calls[1] := 'tool_call'; return true")
|
|
|
|
def test_bytecode_tuples(self):
|
|
# assert self._run_program("return (,);"), ()
|
|
assert self._run_program("return (1, 2, 3);") == (1, 2, 3)
|
|
assert self._run_program("return (1, '2', 3);") == (1, "2", 3)
|
|
assert self._run_program("return (1, (2, 3), 4);") == (1, (2, 3), 4)
|
|
assert self._run_program("return (1, (2, (3, 4)), 5);") == (1, (2, (3, 4)), 5)
|
|
assert self._run_program("let a := (1, 2, 3); return a[2];") == 2
|
|
assert self._run_program("return (1, (2, (3, 4)), 5)[2][2][2];") == 4
|
|
assert self._run_program("return (1, (2, (3, 4)), 5).2.2.2;") == 4
|
|
assert self._run_program("return (1, (2, (3, 4)), 5)[2][2][2] + 1;") == 5
|
|
|
|
def test_bytecode_nested(self):
|
|
assert self._run_program("let r := [1, 2, {'d': (1, 3, 42, 6)}]; return r.3.d.2;") == 3
|
|
assert self._run_program("let r := [1, 2, {'d': (1, 3, 42, 6)}]; return r[3].d[3];") == 42
|
|
assert self._run_program("let r := [1, 2, {'d': (1, 3, 42, 6)}]; return r.3['d'][4];") == 6
|
|
assert self._run_program("let r := {'d': (1, 3, 42, 6)}; return r.d.2;") == 3
|
|
|
|
def test_bytecode_nested_modify(self):
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let r := [1, 2, {'d': [1, 3, 42, 3]}];
|
|
r.3.d.3 := 3;
|
|
return r.3.d.3;
|
|
"""
|
|
)
|
|
== 3
|
|
)
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let r := [1, 2, {'d': [1, 3, 42, 3]}];
|
|
r[3].d[3] := 3;
|
|
return r[3].d[3];
|
|
"""
|
|
)
|
|
== 3
|
|
)
|
|
|
|
assert self._run_program(
|
|
"""
|
|
let r := [1, 2, {'d': [1, 3, 42, 3]}];
|
|
r[3].c := [666];
|
|
return r[3];
|
|
"""
|
|
) == {"d": [1, 3, 42, 3], "c": [666]}
|
|
|
|
assert self._run_program(
|
|
"""
|
|
let r := [1, 2, {'d': [1, 3, 42, 3]}];
|
|
r[3].d[3] := 3;
|
|
return r[3].d;
|
|
"""
|
|
) == [1, 3, 3, 3]
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let r := [1, 2, {'d': [1, 3, 42, 3]}];
|
|
r.3['d'] := ['a', 'b', 'c', 'd'];
|
|
return r[3].d[3];
|
|
"""
|
|
)
|
|
== "c"
|
|
)
|
|
|
|
assert (
|
|
self._run_program(
|
|
"""
|
|
let r := [1, 2, {'d': [1, 3, 42, 3]}];
|
|
let g := 'd';
|
|
r.3[g] := ['a', 'b', 'c', 'd'];
|
|
return r[3].d[3];
|
|
"""
|
|
)
|
|
== "c"
|
|
)
|
|
|
|
def test_bytecode_nested_modify_dict(self):
|
|
assert self._run_program(
|
|
"""
|
|
let event := {
|
|
'event': '$pageview',
|
|
'properties': {
|
|
'$browser': 'Chrome',
|
|
'$os': 'Windows'
|
|
}
|
|
};
|
|
event['properties']['$browser'] := 'Firefox';
|
|
return event;
|
|
"""
|
|
) == {"event": "$pageview", "properties": {"$browser": "Firefox", "$os": "Windows"}}
|
|
assert self._run_program(
|
|
"""
|
|
let event := {
|
|
'event': '$pageview',
|
|
'properties': {
|
|
'$browser': 'Chrome',
|
|
'$os': 'Windows'
|
|
}
|
|
};
|
|
event.properties.$browser := 'Firefox';
|
|
return event;
|
|
"""
|
|
) == {"event": "$pageview", "properties": {"$browser": "Firefox", "$os": "Windows"}}
|
|
assert self._run_program(
|
|
"""
|
|
let event := {
|
|
'event': '$pageview',
|
|
'properties': {
|
|
'$browser': 'Chrome',
|
|
'$os': 'Windows'
|
|
}
|
|
};
|
|
let config := {};
|
|
return event;
|
|
"""
|
|
) == {"event": "$pageview", "properties": {"$browser": "Chrome", "$os": "Windows"}}
|
|
|
|
def test_bytecode_parse_stringify_json(self):
|
|
assert self._run_program("return jsonStringify({'$browser': 'Chrome', '$os': 'Windows' });") == json.dumps(
|
|
{"$browser": "Chrome", "$os": "Windows"}
|
|
)
|
|
|
|
assert self._run_program(
|
|
"return jsonStringify({'$browser': 'Chrome', '$os': 'Windows' }, 3);" # pretty
|
|
) == json.dumps({"$browser": "Chrome", "$os": "Windows"}, indent=3)
|
|
|
|
assert self._run_program("return jsonParse('[1,2,3]');") == [1, 2, 3]
|
|
|
|
assert self._run_program(
|
|
"""
|
|
let event := {
|
|
'event': '$pageview',
|
|
'properties': {
|
|
'$browser': 'Chrome',
|
|
'$os': 'Windows'
|
|
}
|
|
};
|
|
let json := jsonStringify(event);
|
|
return jsonParse(json);
|
|
"""
|
|
) == {"event": "$pageview", "properties": {"$browser": "Chrome", "$os": "Windows"}}
|
|
|
|
def test_bytecode_modify_globals_after_copying(self):
|
|
globals = {"globalEvent": {"event": "$pageview", "properties": {"$browser": "Chrome"}}}
|
|
assert self._run_program(
|
|
"""
|
|
let event := globalEvent;
|
|
event.event := '$autocapture';
|
|
event.properties.$browser := 'Firefox';
|
|
return event;
|
|
""",
|
|
globals=globals,
|
|
) == {"event": "$autocapture", "properties": {"$browser": "Firefox"}}
|
|
global_event = cast(dict[str, Any], globals["globalEvent"])
|
|
assert global_event["event"] == "$pageview"
|
|
assert cast(dict[str, str], global_event["properties"])["$browser"] == "Chrome"
|
|
|
|
def test_bytecode_if_multiif_ternary(self):
|
|
values = []
|
|
|
|
def noisy_print(str):
|
|
nonlocal values
|
|
values.append(str)
|
|
return str
|
|
|
|
self._run_program(
|
|
"""
|
|
if (true) {
|
|
noisy_print('true')
|
|
} else {
|
|
noisy_print('false')
|
|
}
|
|
""",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
assert values == ["true"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program("return true ? noisy_print('true') : noisy_print('false')", {"noisy_print": noisy_print})
|
|
== "true"
|
|
)
|
|
assert values == ["true"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return true ? true ? noisy_print('true1') : noisy_print('true') : noisy_print('false')",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "true1"
|
|
)
|
|
assert values == ["true1"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return true ? false ? noisy_print('true1') : noisy_print('false1') : noisy_print('false2')",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "false1"
|
|
)
|
|
assert values == ["false1"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return false ? false ? noisy_print('true1') : noisy_print('false1') : noisy_print('false2')",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "false2"
|
|
)
|
|
assert values == ["false2"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program("return false ? noisy_print('true') : noisy_print('false')", {"noisy_print": noisy_print})
|
|
== "false"
|
|
)
|
|
assert values == ["false"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return if(false, noisy_print('true'), noisy_print('false'))", {"noisy_print": noisy_print}
|
|
)
|
|
== "false"
|
|
)
|
|
assert values == ["false"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return multiIf(false, noisy_print('true'), false, noisy_print('true'), noisy_print('false2'))",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "false2"
|
|
)
|
|
assert values == ["false2"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return multiIf(false, noisy_print('true'), true, noisy_print('true'), noisy_print('false2'))",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "true"
|
|
)
|
|
assert values == ["true"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return multiIf(true, noisy_print('true1'), false, noisy_print('true2'), noisy_print('false2'))",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "true1"
|
|
)
|
|
assert values == ["true1"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return multiIf(true, noisy_print('true1'), true, noisy_print('true2'), noisy_print('false2'))",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "true1"
|
|
)
|
|
assert values == ["true1"]
|
|
|
|
def test_bytecode_ifnull(self):
|
|
values = []
|
|
|
|
def noisy_print(str):
|
|
nonlocal values
|
|
values.append(str)
|
|
return str
|
|
|
|
assert (
|
|
self._run_program(
|
|
"return null ?? noisy_print('no'); noisy_print('post')",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "no"
|
|
)
|
|
assert values == ["no"]
|
|
|
|
values = []
|
|
assert (
|
|
self._run_program(
|
|
"return noisy_print('yes') ?? noisy_print('no'); noisy_print('post')",
|
|
{"noisy_print": noisy_print},
|
|
)
|
|
== "yes"
|
|
)
|
|
assert values == ["yes"]
|
|
|
|
def test_bytecode_nullish(self):
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return (((a??{}).b)??{}).c") is None
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return (((a??{}).b)??{}).d") == 2
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a?.b?.c") is None
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a.b.c") is None
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a?.b?.d") == 2
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a.b.d") == 2
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a?.b?.['c']") is None
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a.b['c']") is None
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a?.b?.['d']") == 2
|
|
assert self._run_program("let a := {'b': {'d': 2}}; return a.b['d']") == 2
|
|
assert self._run_program("return properties.foo") == "bar"
|
|
assert self._run_program("return properties.not.here") is None
|
|
|
|
def test_bytecode_uncaught_errors(self):
|
|
try:
|
|
self._run_program("throw Error('Not a good day')")
|
|
except UncaughtHogVMException as e:
|
|
assert str(e) == "Error('Not a good day')"
|
|
assert e.type == "Error"
|
|
assert e.message == "Not a good day"
|
|
assert e.payload is None
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
try:
|
|
self._run_program("throw RetryError('Not a good day', {'key': 'value'})")
|
|
except UncaughtHogVMException as e:
|
|
assert str(e) == "RetryError('Not a good day')"
|
|
assert e.type == "RetryError"
|
|
assert e.message == "Not a good day"
|
|
assert e.payload == {"key": "value"}
|
|
else:
|
|
raise AssertionError("Expected Exception not raised")
|
|
|
|
def test_multiple_bytecodes(self):
|
|
ret = lambda string: {"bytecode": ["_H", 1, op.STRING, string, op.RETURN]}
|
|
call = lambda chunk: {"bytecode": ["_H", 1, op.STRING, chunk, op.CALL_GLOBAL, "import", 1, op.RETURN]}
|
|
res = execute_bytecode(
|
|
{
|
|
"root": call("code2"),
|
|
"code2": ret("banana"),
|
|
}
|
|
)
|
|
assert res.result == "banana"
|
|
|
|
def test_multiple_bytecodes_callback(self):
|
|
ret = lambda string: {"bytecode": ["_H", 1, op.STRING, string, op.RETURN]}
|
|
call = lambda chunk: {"bytecode": ["_H", 1, op.STRING, chunk, op.CALL_GLOBAL, "import", 1, op.RETURN]}
|
|
res = execute_bytecode(
|
|
{
|
|
"root": call("code2"),
|
|
"code2": call("code3"),
|
|
"code3": call("code4"),
|
|
"code4": call("code5"),
|
|
"code5": ret("tomato"),
|
|
}
|
|
)
|
|
assert res.result == "tomato"
|
|
|
|
def test_extract_regex(self):
|
|
# Basic extraction with capture group
|
|
assert self._run("extractRegex('hello world', '(\\\\w+)')") == "hello"
|
|
assert self._run("extractRegex('version 1.2.3', '(\\\\d+\\\\.\\\\d+\\\\.\\\\d+)')") == "1.2.3"
|
|
|
|
# No capture group - returns whole match
|
|
assert self._run("extractRegex('hello world', '\\\\w+')") == "hello"
|
|
|
|
# No match - returns empty string
|
|
assert self._run("extractRegex('hello', '\\\\d+')") == ""
|
|
|
|
# Null handling
|
|
assert self._run("extractRegex(null, '\\\\w+')") == ""
|
|
assert self._run("extractRegex('hello', null)") == ""
|
|
|
|
# A pattern with a group that captured nothing still returns the group, not the whole match
|
|
assert self._run("extractRegex('b', '(a)?b')") == ""
|
|
assert self._run("extractRegex('b', '(?:(a)|b)')") == ""
|
|
|
|
# Complex pattern like ClickHouse sortableSemver uses
|
|
assert self._run("extractRegex('v1.2.3-alpha', '(\\\\d+(\\\\.\\\\d+)+)')") == "1.2.3"
|
|
assert self._run("extractRegex('version 10.20.30', '(\\\\d+(\\\\.\\\\d+)+)')") == "10.20.30"
|
|
|
|
@parameterized.expand(
|
|
[
|
|
("match", False),
|
|
("extractRegex", ""),
|
|
]
|
|
)
|
|
def test_regex_functions_run_in_linear_time(self, fn_name: str, expected: bool | str) -> None:
|
|
# Python's re engine needs exponential time on this pattern, so this pins the engine choice.
|
|
# CPU time rather than wall clock, so a paused runner cannot fail the assertion on its own.
|
|
subject = "a" * 26 + "!"
|
|
start = time.process_time()
|
|
result = STL[fn_name].fn([subject, "(a+)+$"], None, None, 5.0)
|
|
elapsed = time.process_time() - start
|
|
assert result == expected
|
|
assert elapsed < 1.0
|
|
|
|
@parameterized.expand(
|
|
[
|
|
("extractRegex", "café", r"\w+", "caf"),
|
|
("extractRegex", "日本語", r"\w+", ""),
|
|
("match", "Müller", r"^\w+$", False),
|
|
("match", "١٢٣", r"\d+", False),
|
|
]
|
|
)
|
|
def test_regex_character_classes_are_ascii_only(
|
|
self, fn_name: str, subject: str, pattern: str, expected: bool | str
|
|
) -> None:
|
|
# The linear-time test cannot pin this, because another linear-time engine could restore the
|
|
# Unicode classes and still run fast. Python's re engine gives "café", a match, and true here.
|
|
assert STL[fn_name].fn([subject, pattern], None, None, 5.0) == expected
|
|
|
|
def test_sortable_semver(self):
|
|
# Basic semver parsing
|
|
assert self._run("sortableSemver('1.2.3')") == [1, 2, 3]
|
|
assert self._run("sortableSemver('10.20.30')") == [10, 20, 30]
|
|
|
|
# With v prefix
|
|
assert self._run("sortableSemver('v1.2.3')") == [1, 2, 3]
|
|
|
|
# With prerelease suffix
|
|
assert self._run("sortableSemver('1.2.3-alpha')") == [1, 2, 3]
|
|
assert self._run("sortableSemver('v2.0.0-beta.1')") == [2, 0, 0]
|
|
|
|
# Version embedded in string
|
|
assert self._run("sortableSemver('version 1.2.3')") == [1, 2, 3]
|
|
|
|
# More components
|
|
assert self._run("sortableSemver('1.2.3.4')") == [1, 2, 3, 4]
|
|
|
|
# Two components
|
|
assert self._run("sortableSemver('1.2')") == [1, 2]
|
|
|
|
# Single component - regex requires at least X.Y format, so this returns empty
|
|
# This matches ClickHouse behavior
|
|
assert self._run("sortableSemver('1')") == []
|
|
|
|
# Null and empty handling
|
|
assert self._run("sortableSemver(null)") == []
|
|
assert self._run("sortableSemver('')") == []
|
|
assert self._run("sortableSemver('no version here')") == []
|
|
|
|
# Comparison use case (what it's designed for)
|
|
assert self._run("sortableSemver('1.2.3') < sortableSemver('1.2.4')") is True
|
|
assert self._run("sortableSemver('1.2.3') < sortableSemver('1.3.0')") is True
|
|
assert self._run("sortableSemver('1.2.3') < sortableSemver('2.0.0')") is True
|
|
assert self._run("sortableSemver('2.0.0') > sortableSemver('1.9.9')") is True
|
|
assert self._run("sortableSemver('1.2.3') = sortableSemver('1.2.3')") is True
|
|
|
|
def test_boolean_template_preserves_type(self):
|
|
"""Boolean template expressions like {true} or {event.properties.flag} should produce actual booleans, not strings."""
|
|
cases: list[tuple[str, dict[str, Any], bool | None | str]] = [
|
|
("{true}", {}, True),
|
|
("{false}", {}, False),
|
|
("{event.properties.opt_out}", {"event": {"properties": {"opt_out": True}}}, True),
|
|
("{event.properties.opt_out}", {"event": {"properties": {"opt_out": False}}}, False),
|
|
("{event.properties.opt_out}", {"event": {"properties": {}}}, None),
|
|
(
|
|
"{event.properties.opt_out}",
|
|
{"event": {"properties": {"opt_out": "a non boolean value"}}},
|
|
"a non boolean value",
|
|
),
|
|
]
|
|
for template, globals, expected in cases:
|
|
bytecode = create_bytecode(parse_string_template(template)).bytecode
|
|
result = execute_bytecode(bytecode, globals=globals).result
|
|
assert result == expected, (
|
|
f"Template '{template}' should produce {expected!r}, got {result!r} ({type(result).__name__})"
|
|
)
|