diff --git a/tests/push/test_bulk_push_rule_evaluator.py b/tests/push/test_bulk_push_rule_evaluator.py
index 7567756135..46df0102f7 100644
--- a/tests/push/test_bulk_push_rule_evaluator.py
+++ b/tests/push/test_bulk_push_rule_evaluator.py
@@ -33,7 +33,6 @@ from tests.unittest import HomeserverTestCase, override_config
class TestBulkPushRuleEvaluator(HomeserverTestCase):
-
servlets = [
admin.register_servlets_for_client_rest_resource,
room.register_servlets,
@@ -131,7 +130,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
# Create a new message event, and try to evaluate it under the dodgy
# power level event.
- event, context = self.get_success(
+ event, unpersisted_context = self.get_success(
self.event_creation_handler.create_event(
self.requester,
{
@@ -146,6 +145,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
prev_event_ids=[pl_event_id],
)
)
+ context = self.get_success(unpersisted_context.persist(event))
bulk_evaluator = BulkPushRuleEvaluator(self.hs)
# should not raise
@@ -171,7 +171,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
"""Ensure that push rules are not calculated when disabled in the config"""
# Create a new message event which should cause a notification.
- event, context = self.get_success(
+ event, unpersisted_context = self.get_success(
self.event_creation_handler.create_event(
self.requester,
{
@@ -185,6 +185,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
},
)
)
+ context = self.get_success(unpersisted_context.persist(event))
bulk_evaluator = BulkPushRuleEvaluator(self.hs)
# Mock the method which calculates push rules -- we do this instead of
@@ -201,7 +202,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
) -> bool:
"""Returns true iff the `mentions` trigger an event push action."""
# Create a new message event which should cause a notification.
- event, context = self.get_success(
+ event, unpersisted_context = self.get_success(
self.event_creation_handler.create_event(
self.requester,
{
@@ -212,7 +213,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
},
)
)
-
+ context = self.get_success(unpersisted_context.persist(event))
# Execute the push rule machinery.
self.get_success(bulk_evaluator.action_for_events_by_user([(event, context)]))
@@ -377,7 +378,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
bulk_evaluator = BulkPushRuleEvaluator(self.hs)
# Create & persist an event to use as the parent of the relation.
- event, context = self.get_success(
+ event, unpersisted_context = self.get_success(
self.event_creation_handler.create_event(
self.requester,
{
@@ -391,6 +392,7 @@ class TestBulkPushRuleEvaluator(HomeserverTestCase):
},
)
)
+ context = self.get_success(unpersisted_context.persist(event))
self.get_success(
self.event_creation_handler.handle_new_client_event(
self.requester, events_and_context=[(event, context)]
diff --git a/tests/push/test_email.py b/tests/push/test_email.py
index ab8bb417e7..4ea5472eb4 100644
--- a/tests/push/test_email.py
+++ b/tests/push/test_email.py
@@ -23,6 +23,7 @@ from twisted.test.proto_helpers import MemoryReactor
import synapse.rest.admin
from synapse.api.errors import Codes, SynapseError
+from synapse.push.emailpusher import EmailPusher
from synapse.rest.client import login, room
from synapse.server import HomeServer
from synapse.util import Clock
@@ -38,7 +39,6 @@ class _User:
class EmailPusherTests(HomeserverTestCase):
-
servlets = [
synapse.rest.admin.register_servlets_for_client_rest_resource,
room.register_servlets,
@@ -47,7 +47,6 @@ class EmailPusherTests(HomeserverTestCase):
hijack_auth = False
def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
-
config = self.default_config()
config["email"] = {
"enable_notifs": True,
@@ -105,6 +104,7 @@ class EmailPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(self.access_token)
)
+ assert user_tuple is not None
self.token_id = user_tuple.token_id
# We need to add email to account before we can create a pusher.
@@ -114,7 +114,7 @@ class EmailPusherTests(HomeserverTestCase):
)
)
- self.pusher = self.get_success(
+ pusher = self.get_success(
self.hs.get_pusherpool().add_or_update_pusher(
user_id=self.user_id,
access_token=self.token_id,
@@ -127,6 +127,8 @@ class EmailPusherTests(HomeserverTestCase):
data={},
)
)
+ assert isinstance(pusher, EmailPusher)
+ self.pusher = pusher
self.auth_handler = hs.get_auth_handler()
self.store = hs.get_datastores().main
@@ -367,18 +369,19 @@ class EmailPusherTests(HomeserverTestCase):
# disassociate the user's email address
self.get_success(
- self.auth_handler.delete_threepid(
- user_id=self.user_id,
- medium="email",
- address="a@example.com",
+ self.auth_handler.delete_local_threepid(
+ user_id=self.user_id, medium="email", address="a@example.com"
)
)
# check that the pusher for that email address has been deleted
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": self.user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by(
+ {"user_name": self.user_id}
+ )
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 0)
def test_remove_unlinked_pushers_background_job(self) -> None:
@@ -413,10 +416,13 @@ class EmailPusherTests(HomeserverTestCase):
self.wait_for_background_updates()
# Check that all pushers with unlinked addresses were deleted
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": self.user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by(
+ {"user_name": self.user_id}
+ )
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 0)
def _check_for_mail(self) -> Tuple[Sequence, Dict]:
@@ -428,10 +434,13 @@ class EmailPusherTests(HomeserverTestCase):
that notification.
"""
# Get the stream ordering before it gets sent
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": self.user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by(
+ {"user_name": self.user_id}
+ )
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 1)
last_stream_ordering = pushers[0].last_stream_ordering
@@ -439,10 +448,13 @@ class EmailPusherTests(HomeserverTestCase):
self.pump(10)
# It hasn't succeeded yet, so the stream ordering shouldn't have moved
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": self.user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by(
+ {"user_name": self.user_id}
+ )
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertEqual(last_stream_ordering, pushers[0].last_stream_ordering)
@@ -458,10 +470,13 @@ class EmailPusherTests(HomeserverTestCase):
self.assertEqual(len(self.email_attempts), 1)
# The stream ordering has increased
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": self.user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by(
+ {"user_name": self.user_id}
+ )
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertTrue(pushers[0].last_stream_ordering > last_stream_ordering)
diff --git a/tests/push/test_http.py b/tests/push/test_http.py
index 23447cc310..c280ddcdf6 100644
--- a/tests/push/test_http.py
+++ b/tests/push/test_http.py
@@ -11,7 +11,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from typing import List, Optional, Tuple
+from typing import Any, List, Tuple
from unittest.mock import Mock
from twisted.internet.defer import Deferred
@@ -22,7 +22,6 @@ from synapse.logging.context import make_deferred_yieldable
from synapse.push import PusherConfig, PusherConfigException
from synapse.rest.client import login, push_rule, pusher, receipts, room
from synapse.server import HomeServer
-from synapse.storage.databases.main.registration import TokenLookupResult
from synapse.types import JsonDict
from synapse.util import Clock
@@ -67,9 +66,10 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
- def test_data(data: Optional[JsonDict]) -> None:
+ def test_data(data: Any) -> None:
self.get_failure(
self.hs.get_pusherpool().add_or_update_pusher(
user_id=user_id,
@@ -113,6 +113,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
self.get_success(
@@ -140,10 +141,11 @@ class HTTPPusherTests(HomeserverTestCase):
self.helper.send(room, body="There!", tok=other_access_token)
# Get the stream ordering before it gets sent
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 1)
last_stream_ordering = pushers[0].last_stream_ordering
@@ -151,10 +153,11 @@ class HTTPPusherTests(HomeserverTestCase):
self.pump()
# It hasn't succeeded yet, so the stream ordering shouldn't have moved
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertEqual(last_stream_ordering, pushers[0].last_stream_ordering)
@@ -172,10 +175,11 @@ class HTTPPusherTests(HomeserverTestCase):
self.pump()
# The stream ordering has increased
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertTrue(pushers[0].last_stream_ordering > last_stream_ordering)
last_stream_ordering = pushers[0].last_stream_ordering
@@ -194,10 +198,11 @@ class HTTPPusherTests(HomeserverTestCase):
self.pump()
# The stream ordering has increased, again
- pushers = self.get_success(
- self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ pushers = list(
+ self.get_success(
+ self.hs.get_datastores().main.get_pushers_by({"user_name": user_id})
+ )
)
- pushers = list(pushers)
self.assertEqual(len(pushers), 1)
self.assertTrue(pushers[0].last_stream_ordering > last_stream_ordering)
@@ -229,6 +234,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
self.get_success(
@@ -349,6 +355,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
self.get_success(
@@ -435,6 +442,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
self.get_success(
@@ -512,6 +520,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
self.get_success(
@@ -618,6 +627,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
self.get_success(
@@ -753,6 +763,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
self.get_success(
@@ -895,6 +906,7 @@ class HTTPPusherTests(HomeserverTestCase):
user_tuple = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert user_tuple is not None
token_id = user_tuple.token_id
device_id = user_tuple.device_id
@@ -941,9 +953,10 @@ class HTTPPusherTests(HomeserverTestCase):
)
# Look up the user info for the access token so we can compare the device ID.
- lookup_result: TokenLookupResult = self.get_success(
+ lookup_result = self.get_success(
self.hs.get_datastores().main.get_user_by_access_token(access_token)
)
+ assert lookup_result is not None
# Get the user's devices and check it has the correct device ID.
channel = self.make_request("GET", "/pushers", access_token=access_token)
diff --git a/tests/push/test_push_rule_evaluator.py b/tests/push/test_push_rule_evaluator.py
index 7c430c4ecb..52c4aafea6 100644
--- a/tests/push/test_push_rule_evaluator.py
+++ b/tests/push/test_push_rule_evaluator.py
@@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from typing import Any, Dict, List, Optional, Set, Union, cast
+from typing import Any, Dict, List, Optional, Union, cast
import frozendict
@@ -22,7 +22,7 @@ import synapse.rest.admin
from synapse.api.constants import EventTypes, HistoryVisibility, Membership
from synapse.api.room_versions import RoomVersions
from synapse.appservice import ApplicationService
-from synapse.events import FrozenEvent
+from synapse.events import FrozenEvent, make_event_from_dict
from synapse.push.bulk_push_rule_evaluator import _flatten_dict
from synapse.push.httppusher import tweaks_for_actions
from synapse.rest import admin
@@ -32,6 +32,7 @@ from synapse.storage.databases.main.appservice import _make_exclusive_regex
from synapse.synapse_rust.push import PushRuleEvaluator
from synapse.types import JsonDict, JsonMapping, UserID
from synapse.util import Clock
+from synapse.util.frozenutils import freeze
from tests import unittest
from tests.test_utils.event_injection import create_event, inject_member_event
@@ -48,17 +49,93 @@ class FlattenDictTestCase(unittest.TestCase):
input = {"foo": {"bar": "abc"}}
self.assertEqual({"foo.bar": "abc"}, _flatten_dict(input))
+ # If a field has a dot in it, escape it.
+ input = {"m.foo": {"b\\ar": "abc"}}
+ self.assertEqual({"m\\.foo.b\\\\ar": "abc"}, _flatten_dict(input))
+
def test_non_string(self) -> None:
- """Non-string items are dropped."""
+ """String, booleans, ints, nulls and list of those should be kept while other items are dropped."""
input: Dict[str, Any] = {
"woo": "woo",
"foo": True,
"bar": 1,
"baz": None,
- "fuzz": [],
+ "fuzz": ["woo", True, 1, None, [], {}],
"boo": {},
}
- self.assertEqual({"woo": "woo"}, _flatten_dict(input))
+ self.assertEqual(
+ {
+ "woo": "woo",
+ "foo": True,
+ "bar": 1,
+ "baz": None,
+ "fuzz": ["woo", True, 1, None],
+ },
+ _flatten_dict(input),
+ )
+
+ def test_event(self) -> None:
+ """Events can also be flattened."""
+ event = make_event_from_dict(
+ {
+ "room_id": "!test:test",
+ "type": "m.room.message",
+ "sender": "@alice:test",
+ "content": {
+ "msgtype": "m.text",
+ "body": "Hello world!",
+ "format": "org.matrix.custom.html",
+ "formatted_body": "<h1>Hello world!</h1>",
+ },
+ },
+ room_version=RoomVersions.V8,
+ )
+ expected = {
+ "content.msgtype": "m.text",
+ "content.body": "Hello world!",
+ "content.format": "org.matrix.custom.html",
+ "content.formatted_body": "<h1>Hello world!</h1>",
+ "room_id": "!test:test",
+ "sender": "@alice:test",
+ "type": "m.room.message",
+ }
+ self.assertEqual(expected, _flatten_dict(event))
+
+ def test_extensible_events(self) -> None:
+ """Extensible events has compatibility behaviour."""
+ event_dict = {
+ "room_id": "!test:test",
+ "type": "m.room.message",
+ "sender": "@alice:test",
+ "content": {
+ "org.matrix.msc1767.markup": [
+ {"mimetype": "text/plain", "body": "Hello world!"},
+ {"mimetype": "text/html", "body": "<h1>Hello world!</h1>"},
+ ]
+ },
+ }
+
+ # For a current room version, there's no special behavior.
+ event = make_event_from_dict(event_dict, room_version=RoomVersions.V8)
+ expected = {
+ "room_id": "!test:test",
+ "sender": "@alice:test",
+ "type": "m.room.message",
+ "content.org\\.matrix\\.msc1767\\.markup": [],
+ }
+ self.assertEqual(expected, _flatten_dict(event))
+
+ # For a room version with extensible events, they parse out the text/plain
+ # to a content.body property.
+ event = make_event_from_dict(event_dict, room_version=RoomVersions.MSC1767v10)
+ expected = {
+ "content.body": "hello world!",
+ "room_id": "!test:test",
+ "sender": "@alice:test",
+ "type": "m.room.message",
+ "content.org\\.matrix\\.msc1767\\.markup": [],
+ }
+ self.assertEqual(expected, _flatten_dict(event))
class PushRuleEvaluatorTestCase(unittest.TestCase):
@@ -66,9 +143,6 @@ class PushRuleEvaluatorTestCase(unittest.TestCase):
self,
content: JsonMapping,
*,
- has_mentions: bool = False,
- user_mentions: Optional[Set[str]] = None,
- room_mention: bool = False,
related_events: Optional[JsonDict] = None,
) -> PushRuleEvaluator:
event = FrozenEvent(
@@ -87,9 +161,7 @@ class PushRuleEvaluatorTestCase(unittest.TestCase):
power_levels: Dict[str, Union[int, Dict[str, int]]] = {}
return PushRuleEvaluator(
_flatten_dict(event),
- has_mentions,
- user_mentions or set(),
- room_mention,
+ False,
room_member_count,
sender_power_level,
cast(Dict[str, int], power_levels.get("notifications", {})),
@@ -123,53 +195,6 @@ class PushRuleEvaluatorTestCase(unittest.TestCase):
# A display name with spaces should work fine.
self.assertTrue(evaluator.matches(condition, "@user:test", "foo bar"))
- def test_user_mentions(self) -> None:
- """Check for user mentions."""
- condition = {"kind": "org.matrix.msc3952.is_user_mention"}
-
- # No mentions shouldn't match.
- evaluator = self._get_evaluator({}, has_mentions=True)
- self.assertFalse(evaluator.matches(condition, "@user:test", None))
-
- # An empty set shouldn't match
- evaluator = self._get_evaluator({}, has_mentions=True, user_mentions=set())
- self.assertFalse(evaluator.matches(condition, "@user:test", None))
-
- # The Matrix ID appearing anywhere in the mentions list should match
- evaluator = self._get_evaluator(
- {}, has_mentions=True, user_mentions={"@user:test"}
- )
- self.assertTrue(evaluator.matches(condition, "@user:test", None))
-
- evaluator = self._get_evaluator(
- {}, has_mentions=True, user_mentions={"@another:test", "@user:test"}
- )
- self.assertTrue(evaluator.matches(condition, "@user:test", None))
-
- # Note that invalid data is tested at tests.push.test_bulk_push_rule_evaluator.TestBulkPushRuleEvaluator.test_mentions
- # since the BulkPushRuleEvaluator is what handles data sanitisation.
-
- def test_room_mentions(self) -> None:
- """Check for room mentions."""
- condition = {"kind": "org.matrix.msc3952.is_room_mention"}
-
- # No room mention shouldn't match.
- evaluator = self._get_evaluator({}, has_mentions=True)
- self.assertFalse(evaluator.matches(condition, None, None))
-
- # Room mention should match.
- evaluator = self._get_evaluator({}, has_mentions=True, room_mention=True)
- self.assertTrue(evaluator.matches(condition, None, None))
-
- # A room mention and user mention is valid.
- evaluator = self._get_evaluator(
- {}, has_mentions=True, user_mentions={"@another:test"}, room_mention=True
- )
- self.assertTrue(evaluator.matches(condition, None, None))
-
- # Note that invalid data is tested at tests.push.test_bulk_push_rule_evaluator.TestBulkPushRuleEvaluator.test_mentions
- # since the BulkPushRuleEvaluator is what handles data sanitisation.
-
def _assert_matches(
self, condition: JsonDict, content: JsonMapping, msg: Optional[str] = None
) -> None:
@@ -341,6 +366,193 @@ class PushRuleEvaluatorTestCase(unittest.TestCase):
"pattern should not match before a newline",
)
+ def test_event_match_pattern(self) -> None:
+ """Check that event_match conditions do not use a "pattern_type" from user data."""
+
+ # The pattern_type should not be deserialized into anything valid.
+ condition = {
+ "kind": "event_match",
+ "key": "content.value",
+ "pattern_type": "user_id",
+ }
+ self._assert_not_matches(
+ condition,
+ {"value": "@user:test"},
+ "should not be possible to pass a pattern_type in",
+ )
+
+ # This is an internal-only condition which shouldn't get deserialized.
+ condition = {
+ "kind": "event_match_type",
+ "key": "content.value",
+ "pattern_type": "user_id",
+ }
+ self._assert_not_matches(
+ condition,
+ {"value": "@user:test"},
+ "should not be possible to pass a pattern_type in",
+ )
+
+ def test_exact_event_match_string(self) -> None:
+ """Check that exact_event_match conditions work as expected for strings."""
+
+ # Test against a string value.
+ condition = {
+ "kind": "event_property_is",
+ "key": "content.value",
+ "value": "foobaz",
+ }
+ self._assert_matches(
+ condition,
+ {"value": "foobaz"},
+ "exact value should match",
+ )
+ self._assert_not_matches(
+ condition,
+ {"value": "FoobaZ"},
+ "values should match and be case-sensitive",
+ )
+ self._assert_not_matches(
+ condition,
+ {"value": "test foobaz test"},
+ "values must exactly match",
+ )
+ value: Any
+ for value in (True, False, 1, 1.1, None, [], {}):
+ self._assert_not_matches(
+ condition,
+ {"value": value},
+ "incorrect types should not match",
+ )
+
+ # it should work on frozendicts too
+ self._assert_matches(
+ condition,
+ frozendict.frozendict({"value": "foobaz"}),
+ "values should match on frozendicts",
+ )
+
+ def test_exact_event_match_boolean(self) -> None:
+ """Check that exact_event_match conditions work as expected for booleans."""
+
+ # Test against a True boolean value.
+ condition = {"kind": "event_property_is", "key": "content.value", "value": True}
+ self._assert_matches(
+ condition,
+ {"value": True},
+ "exact value should match",
+ )
+ self._assert_not_matches(
+ condition,
+ {"value": False},
+ "incorrect values should not match",
+ )
+ for value in ("foobaz", 1, 1.1, None, [], {}):
+ self._assert_not_matches(
+ condition,
+ {"value": value},
+ "incorrect types should not match",
+ )
+
+ # Test against a False boolean value.
+ condition = {
+ "kind": "event_property_is",
+ "key": "content.value",
+ "value": False,
+ }
+ self._assert_matches(
+ condition,
+ {"value": False},
+ "exact value should match",
+ )
+ self._assert_not_matches(
+ condition,
+ {"value": True},
+ "incorrect values should not match",
+ )
+ # Choose false-y values to ensure there's no type coercion.
+ for value in ("", 0, 1.1, None, [], {}):
+ self._assert_not_matches(
+ condition,
+ {"value": value},
+ "incorrect types should not match",
+ )
+
+ def test_exact_event_match_null(self) -> None:
+ """Check that exact_event_match conditions work as expected for null."""
+
+ condition = {"kind": "event_property_is", "key": "content.value", "value": None}
+ self._assert_matches(
+ condition,
+ {"value": None},
+ "exact value should match",
+ )
+ for value in ("foobaz", True, False, 1, 1.1, [], {}):
+ self._assert_not_matches(
+ condition,
+ {"value": value},
+ "incorrect types should not match",
+ )
+
+ def test_exact_event_match_integer(self) -> None:
+ """Check that exact_event_match conditions work as expected for integers."""
+
+ condition = {"kind": "event_property_is", "key": "content.value", "value": 1}
+ self._assert_matches(
+ condition,
+ {"value": 1},
+ "exact value should match",
+ )
+ value: Any
+ for value in (1.1, -1, 0):
+ self._assert_not_matches(
+ condition,
+ {"value": value},
+ "incorrect values should not match",
+ )
+ for value in ("1", True, False, None, [], {}):
+ self._assert_not_matches(
+ condition,
+ {"value": value},
+ "incorrect types should not match",
+ )
+
+ def test_exact_event_property_contains(self) -> None:
+ """Check that exact_event_property_contains conditions work as expected."""
+
+ condition = {
+ "kind": "event_property_contains",
+ "key": "content.value",
+ "value": "foobaz",
+ }
+ self._assert_matches(
+ condition,
+ {"value": ["foobaz"]},
+ "exact value should match",
+ )
+ self._assert_matches(
+ condition,
+ {"value": ["foobaz", "bugz"]},
+ "extra values should match",
+ )
+ self._assert_not_matches(
+ condition,
+ {"value": ["FoobaZ"]},
+ "values should match and be case-sensitive",
+ )
+ self._assert_not_matches(
+ condition,
+ {"value": "foobaz"},
+ "does not search in a string",
+ )
+
+ # it should work on frozendicts too
+ self._assert_matches(
+ condition,
+ freeze({"value": ["foobaz"]}),
+ "values should match on frozendicts",
+ )
+
def test_no_body(self) -> None:
"""Not having a body shouldn't break the evaluator."""
evaluator = self._get_evaluator({})
|
