4532

Committed 23 Sep 2019 - 19:39 coverage decreased (-49.7%) to 17.596%

Build # 4532

Build Type

buildkite

Committed by Richard van der Hoff

Commit Message

update changelog

Pull Request Pull Request #6079: Add submit_url response parameter to msisdn /requestToken

Run Details

359 of 12986 branches covered (2.76%)

Branch coverage included in aggregate %.

0 of 7 new or added lines in 1 file covered. (0.0%)

18869 existing lines in 281 files now uncovered.

8809 of 39116 relevant lines covered (22.52%)

0.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.99

/synapse/types.py

# -*- coding: utf-8 -*-
# Copyright 2014-2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# 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.
import re
import string
from collections import namedtuple

import attr

from synapse.api.errors import SynapseError


class Requester(
    namedtuple(
        "Requester", ["user", "access_token_id", "is_guest", "device_id", "app_service"]
    )
):
    """
    Represents the user making a request

    Attributes:
        user (UserID):  id of the user making the request
        access_token_id (int|None):  *ID* of the access token used for this
            request, or None if it came via the appservice API or similar
        is_guest (bool):  True if the user making this request is a guest user
        device_id (str|None):  device_id which was set at authentication time
        app_service (ApplicationService|None):  the AS requesting on behalf of the user
    """

    def serialize(self):
        """Converts self to a type that can be serialized as JSON, and then
        deserialized by `deserialize`

        Returns:
            dict
        """
        return {
            "user_id": self.user.to_string(),
            "access_token_id": self.access_token_id,
            "is_guest": self.is_guest,
            "device_id": self.device_id,
            "app_server_id": self.app_service.id if self.app_service else None,
        }

    @staticmethod
    def deserialize(store, input):
        """Converts a dict that was produced by `serialize` back into a
        Requester.

        Args:
            store (DataStore): Used to convert AS ID to AS object
            input (dict): A dict produced by `serialize`

        Returns:
            Requester
        """
        appservice = None
        if input["app_server_id"]:
            appservice = store.get_app_service_by_id(input["app_server_id"])

        return Requester(
            user=UserID.from_string(input["user_id"]),
            access_token_id=input["access_token_id"],
            is_guest=input["is_guest"],
            device_id=input["device_id"],
            app_service=appservice,
        )


def create_requester(
    user_id, access_token_id=None, is_guest=False, device_id=None, app_service=None
):
    """
    Create a new ``Requester`` object

    Args:
        user_id (str|UserID):  id of the user making the request
        access_token_id (int|None):  *ID* of the access token used for this
            request, or None if it came via the appservice API or similar
        is_guest (bool):  True if the user making this request is a guest user
        device_id (str|None):  device_id which was set at authentication time
        app_service (ApplicationService|None):  the AS requesting on behalf of the user

    Returns:
        Requester
    """
    if not isinstance(user_id, UserID):
        user_id = UserID.from_string(user_id)
    return Requester(user_id, access_token_id, is_guest, device_id, app_service)


def get_domain_from_id(string):
    idx = string.find(":")
    if idx == -1:
        raise SynapseError(400, "Invalid ID: %r" % (string,))
    return string[idx + 1 :]


def get_localpart_from_id(string):
    idx = string.find(":")
    if idx == -1:
        raise SynapseError(400, "Invalid ID: %r" % (string,))
    return string[1:idx]


class DomainSpecificString(namedtuple("DomainSpecificString", ("localpart", "domain"))):
    """Common base class among ID/name strings that have a local part and a
    domain name, prefixed with a sigil.

    Has the fields:

        'localpart' : The local part of the name (without the leading sigil)
        'domain' : The domain part of the name
    """

    # Deny iteration because it will bite you if you try to create a singleton
    # set by:
    #    users = set(user)
    def __iter__(self):
        raise ValueError("Attempted to iterate a %s" % (type(self).__name__,))

    # Because this class is a namedtuple of strings and booleans, it is deeply
    # immutable.
    def __copy__(self):
        return self

    def __deepcopy__(self, memo):
        return self

    @classmethod
    def from_string(cls, s):
        """Parse the string given by 's' into a structure object."""
        if len(s) < 1 or s[0:1] != cls.SIGIL:
            raise SynapseError(
                400, "Expected %s string to start with '%s'" % (cls.__name__, cls.SIGIL)
            )

        parts = s[1:].split(":", 1)
        if len(parts) != 2:
            raise SynapseError(
                400,
                "Expected %s of the form '%slocalname:domain'"
                % (cls.__name__, cls.SIGIL),
            )

        domain = parts[1]

        # This code will need changing if we want to support multiple domain
        # names on one HS
        return cls(localpart=parts[0], domain=domain)

    def to_string(self):
        """Return a string encoding the fields of the structure object."""
        return "%s%s:%s" % (self.SIGIL, self.localpart, self.domain)

    @classmethod
    def is_valid(cls, s):
        try:
            cls.from_string(s)
            return True
        except Exception:
            return False

    __repr__ = to_string


class UserID(DomainSpecificString):
    """Structure representing a user ID."""

    SIGIL = "@"


class RoomAlias(DomainSpecificString):
    """Structure representing a room name."""

    SIGIL = "#"


class RoomID(DomainSpecificString):
    """Structure representing a room id. """

    SIGIL = "!"


class EventID(DomainSpecificString):
    """Structure representing an event id. """

    SIGIL = "$"


class GroupID(DomainSpecificString):
    """Structure representing a group ID."""

    SIGIL = "+"

    @classmethod
    def from_string(cls, s):
        group_id = super(GroupID, cls).from_string(s)
        if not group_id.localpart:
            raise SynapseError(400, "Group ID cannot be empty")

        if contains_invalid_mxid_characters(group_id.localpart):
            raise SynapseError(
                400, "Group ID can only contain characters a-z, 0-9, or '=_-./'"
            )

        return group_id


mxid_localpart_allowed_characters = set(
    "_-./=" + string.ascii_lowercase + string.digits
)


def contains_invalid_mxid_characters(localpart):
    """Check for characters not allowed in an mxid or groupid localpart

    Args:
        localpart (basestring): the localpart to be checked

    Returns:
        bool: True if there are any naughty characters
    """
    return any(c not in mxid_localpart_allowed_characters for c in localpart)


UPPER_CASE_PATTERN = re.compile(b"[A-Z_]")

# the following is a pattern which matches '=', and bytes which are not allowed in a mxid
# localpart.
#
# It works by:
#  * building a string containing the allowed characters (excluding '=')
#  * escaping every special character with a backslash (to stop '-' being interpreted as a
#    range operator)
#  * wrapping it in a '[^...]' regex
#  * converting the whole lot to a 'bytes' sequence, so that we can use it to match
#    bytes rather than strings
#
NON_MXID_CHARACTER_PATTERN = re.compile(
    ("[^%s]" % (re.escape("".join(mxid_localpart_allowed_characters - {"="})),)).encode(
        "ascii"
    )
)


def map_username_to_mxid_localpart(username, case_sensitive=False):
    """Map a username onto a string suitable for a MXID

    This follows the algorithm laid out at
    https://matrix.org/docs/spec/appendices.html#mapping-from-other-character-sets.

    Args:
        username (unicode|bytes): username to be mapped
        case_sensitive (bool): true if TEST and test should be mapped
            onto different mxids

    Returns:
        unicode: string suitable for a mxid localpart
    """
    if not isinstance(username, bytes):
        username = username.encode("utf-8")

    # first we sort out upper-case characters
    if case_sensitive:

        def f1(m):
            return b"_" + m.group().lower()

        username = UPPER_CASE_PATTERN.sub(f1, username)
    else:
        username = username.lower()

    # then we sort out non-ascii characters
    def f2(m):
        g = m.group()[0]
        if isinstance(g, str):
            # on python 2, we need to do a ord(). On python 3, the
            # byte itself will do.
            g = ord(g)
        return b"=%02x" % (g,)

    username = NON_MXID_CHARACTER_PATTERN.sub(f2, username)

    # we also do the =-escaping to mxids starting with an underscore.
    username = re.sub(b"^_", b"=5f", username)

    # we should now only have ascii bytes left, so can decode back to a
    # unicode.
    return username.decode("ascii")


class StreamToken(
    namedtuple(
        "Token",
        (
            "room_key",
            "presence_key",
            "typing_key",
            "receipt_key",
            "account_data_key",
            "push_rules_key",
            "to_device_key",
            "device_list_key",
            "groups_key",
        ),
    )
):
    _SEPARATOR = "_"

    @classmethod
    def from_string(cls, string):
        try:
            keys = string.split(cls._SEPARATOR)
            while len(keys) < len(cls._fields):
                # i.e. old token from before receipt_key
                keys.append("0")
            return cls(*keys)
        except Exception:
            raise SynapseError(400, "Invalid Token")

    def to_string(self):
        return self._SEPARATOR.join([str(k) for k in self])

    @property
    def room_stream_id(self):
        # TODO(markjh): Awful hack to work around hacks in the presence tests
        # which assume that the keys are integers.
        if type(self.room_key) is int:
            return self.room_key
        else:
            return int(self.room_key[1:].split("-")[-1])

    def is_after(self, other):
        """Does this token contain events that the other doesn't?"""
        return (
            (other.room_stream_id < self.room_stream_id)
            or (int(other.presence_key) < int(self.presence_key))
            or (int(other.typing_key) < int(self.typing_key))
            or (int(other.receipt_key) < int(self.receipt_key))
            or (int(other.account_data_key) < int(self.account_data_key))
            or (int(other.push_rules_key) < int(self.push_rules_key))
            or (int(other.to_device_key) < int(self.to_device_key))
            or (int(other.device_list_key) < int(self.device_list_key))
            or (int(other.groups_key) < int(self.groups_key))
        )

    def copy_and_advance(self, key, new_value):
        """Advance the given key in the token to a new value if and only if the
        new value is after the old value.
        """
        new_token = self.copy_and_replace(key, new_value)
        if key == "room_key":
            new_id = new_token.room_stream_id
            old_id = self.room_stream_id
        else:
            new_id = int(getattr(new_token, key))
            old_id = int(getattr(self, key))
        if old_id < new_id:
            return new_token
        else:
            return self

    def copy_and_replace(self, key, new_value):
        return self._replace(**{key: new_value})


StreamToken.START = StreamToken(*(["s0"] + ["0"] * (len(StreamToken._fields) - 1)))


class RoomStreamToken(namedtuple("_StreamToken", "topological stream")):
    """Tokens are positions between events. The token "s1" comes after event 1.

            s0    s1
            |     |
        [0] V [1] V [2]

    Tokens can either be a point in the live event stream or a cursor going
    through historic events.

    When traversing the live event stream events are ordered by when they
    arrived at the homeserver.

    When traversing historic events the events are ordered by their depth in
    the event graph "topological_ordering" and then by when they arrived at the
    homeserver "stream_ordering".

    Live tokens start with an "s" followed by the "stream_ordering" id of the
    event it comes after. Historic tokens start with a "t" followed by the
    "topological_ordering" id of the event it comes after, followed by "-",
    followed by the "stream_ordering" id of the event it comes after.
    """

    __slots__ = []

    @classmethod
    def parse(cls, string):
        try:
            if string[0] == "s":
                return cls(topological=None, stream=int(string[1:]))
            if string[0] == "t":
                parts = string[1:].split("-", 1)
                return cls(topological=int(parts[0]), stream=int(parts[1]))
        except Exception:
            pass
        raise SynapseError(400, "Invalid token %r" % (string,))

    @classmethod
    def parse_stream_token(cls, string):
        try:
            if string[0] == "s":
                return cls(topological=None, stream=int(string[1:]))
        except Exception:
            pass
        raise SynapseError(400, "Invalid token %r" % (string,))

    def __str__(self):
        if self.topological is not None:
            return "t%d-%d" % (self.topological, self.stream)
        else:
            return "s%d" % (self.stream,)


class ThirdPartyInstanceID(
    namedtuple("ThirdPartyInstanceID", ("appservice_id", "network_id"))
):
    # Deny iteration because it will bite you if you try to create a singleton
    # set by:
    #    users = set(user)
    def __iter__(self):
        raise ValueError("Attempted to iterate a %s" % (type(self).__name__,))

    # Because this class is a namedtuple of strings, it is deeply immutable.
    def __copy__(self):
        return self

    def __deepcopy__(self, memo):
        return self

    @classmethod
    def from_string(cls, s):
        bits = s.split("|", 2)
        if len(bits) != 2:
            raise SynapseError(400, "Invalid ID %r" % (s,))

        return cls(appservice_id=bits[0], network_id=bits[1])

    def to_string(self):
        return "%s|%s" % (self.appservice_id, self.network_id)

    __str__ = to_string

    @classmethod
    def create(cls, appservice_id, network_id):
        return cls(appservice_id=appservice_id, network_id=network_id)


@attr.s(slots=True)
class ReadReceipt(object):
    """Information about a read-receipt"""

    room_id = attr.ib()
    receipt_type = attr.ib()
    user_id = attr.ib()
    event_ids = attr.ib()
    data = attr.ib()

1	# -- coding: utf-8 --
2	# Copyright 2014-2016 OpenMarket Ltd
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15	import re	1×
16	import string	1×
17	from collections import namedtuple	1×
18
19	import attr	1×
20
21	from synapse.api.errors import SynapseError	1×
22
23
24	class Requester(	1×
25	namedtuple(
26	"Requester", ["user", "access_token_id", "is_guest", "device_id", "app_service"]
27	)
28	):
29	"""
30	Represents the user making a request
31
32	Attributes:
33	user (UserID): id of the user making the request
34	access_token_id (int\|None): ID of the access token used for this
35	request, or None if it came via the appservice API or similar
36	is_guest (bool): True if the user making this request is a guest user
37	device_id (str\|None): device_id which was set at authentication time
38	app_service (ApplicationService\|None): the AS requesting on behalf of the user
39	"""
40
41	def serialize(self):	1×
42	"""Converts self to a type that can be serialized as JSON, and then
43	deserialized by `deserialize`
44
45	Returns:
46	dict
47	"""
48	return {	!
49	"user_id": self.user.to_string(),
50	"access_token_id": self.access_token_id,
51	"is_guest": self.is_guest,
52	"device_id": self.device_id,
53	"app_server_id": self.app_service.id if self.app_service else None,
54	}
55
56	@staticmethod	1×
57	def deserialize(store, input):
58	"""Converts a dict that was produced by `serialize` back into a
59	Requester.
60
61	Args:
62	store (DataStore): Used to convert AS ID to AS object
63	input (dict): A dict produced by `serialize`
64
65	Returns:
66	Requester
67	"""
68	appservice = None	!
69	if input["app_server_id"]:	Branches [[0, 70], [0, 72]] missed. !
70	appservice = store.get_app_service_by_id(input["app_server_id"])	!
71
72	return Requester(	!
73	user=UserID.from_string(input["user_id"]),
74	access_token_id=input["access_token_id"],
75	is_guest=input["is_guest"],
76	device_id=input["device_id"],
77	app_service=appservice,
78	)
79
80
81	def create_requester(	1×
82	user_id, access_token_id=None, is_guest=False, device_id=None, app_service=None
83	):
84	"""
85	Create a new ``Requester`` object
86
87	Args:
88	user_id (str\|UserID): id of the user making the request
89	access_token_id (int\|None): ID of the access token used for this
90	request, or None if it came via the appservice API or similar
91	is_guest (bool): True if the user making this request is a guest user
92	device_id (str\|None): device_id which was set at authentication time
93	app_service (ApplicationService\|None): the AS requesting on behalf of the user
94
95	Returns:
96	Requester
97	"""
UNCOV 98	if not isinstance(user_id, UserID):	Branches [[0, 99], [0, 100]] missed. !
UNCOV 99	user_id = UserID.from_string(user_id)	!
UNCOV 100	return Requester(user_id, access_token_id, is_guest, device_id, app_service)	!
101
102
103	def get_domain_from_id(string):	1×
UNCOV 104	idx = string.find(":")	!
UNCOV 105	if idx == -1:	Branches [[0, 106], [0, 107]] missed. !
106	raise SynapseError(400, "Invalid ID: %r" % (string,))	!
UNCOV 107	return string[idx + 1 :]	!
108
109
110	def get_localpart_from_id(string):	1×
UNCOV 111	idx = string.find(":")	!
UNCOV 112	if idx == -1:	Branches [[0, 113], [0, 114]] missed. !
113	raise SynapseError(400, "Invalid ID: %r" % (string,))	!
UNCOV 114	return string[1:idx]	!
115
116
117	class DomainSpecificString(namedtuple("DomainSpecificString", ("localpart", "domain"))):	1×
118	"""Common base class among ID/name strings that have a local part and a
119	domain name, prefixed with a sigil.
120
121	Has the fields:
122
123	'localpart' : The local part of the name (without the leading sigil)
124	'domain' : The domain part of the name
125	"""
126
127	# Deny iteration because it will bite you if you try to create a singleton
128	# set by:
129	# users = set(user)
130	def __iter__(self):	1×
131	raise ValueError("Attempted to iterate a %s" % (type(self).__name__,))	!
132
133	# Because this class is a namedtuple of strings and booleans, it is deeply
134	# immutable.
135	def __copy__(self):	1×
136	return self	!
137
138	def __deepcopy__(self, memo):	1×
139	return self	!
140
141	@classmethod	1×
142	def from_string(cls, s):
143	"""Parse the string given by 's' into a structure object."""
UNCOV 144	if len(s) < 1 or s[0:1] != cls.SIGIL:	Branches [[0, 145], [0, 149]] missed. !
UNCOV 145	raise SynapseError(	!
146	400, "Expected %s string to start with '%s'" % (cls.__name__, cls.SIGIL)
147	)
148
UNCOV 149	parts = s[1:].split(":", 1)	!
UNCOV 150	if len(parts) != 2:	Branches [[0, 151], [0, 157]] missed. !
151	raise SynapseError(	!
152	400,
153	"Expected %s of the form '%slocalname:domain'"
154	% (cls.__name__, cls.SIGIL),
155	)
156
UNCOV 157	domain = parts[1]	!
158
159	# This code will need changing if we want to support multiple domain
160	# names on one HS
UNCOV 161	return cls(localpart=parts[0], domain=domain)	!
162
163	def to_string(self):	1×
164	"""Return a string encoding the fields of the structure object."""
UNCOV 165	return "%s%s:%s" % (self.SIGIL, self.localpart, self.domain)	!
166
167	@classmethod	1×
168	def is_valid(cls, s):
UNCOV 169	try:	!
UNCOV 170	cls.from_string(s)	!
UNCOV 171	return True	!
UNCOV 172	except Exception:	!
UNCOV 173	return False	!
174
175	__repr__ = to_string	1×
176
177
178	class UserID(DomainSpecificString):	1×
179	"""Structure representing a user ID."""
180
181	SIGIL = "@"	1×
182
183
184	class RoomAlias(DomainSpecificString):	1×
185	"""Structure representing a room name."""
186
187	SIGIL = "#"	1×
188
189
190	class RoomID(DomainSpecificString):	1×
191	"""Structure representing a room id. """
192
193	SIGIL = "!"	1×
194
195
196	class EventID(DomainSpecificString):	1×
197	"""Structure representing an event id. """
198
199	SIGIL = "$"	1×
200
201
202	class GroupID(DomainSpecificString):	1×
203	"""Structure representing a group ID."""
204
205	SIGIL = "+"	1×
206
207	@classmethod	1×
208	def from_string(cls, s):
UNCOV 209	group_id = super(GroupID, cls).from_string(s)	!
UNCOV 210	if not group_id.localpart:	Branches [[0, 211], [0, 213]] missed. !
211	raise SynapseError(400, "Group ID cannot be empty")	!
212
UNCOV 213	if contains_invalid_mxid_characters(group_id.localpart):	Branches [[0, 214], [0, 218]] missed. !
214	raise SynapseError(	!
215	400, "Group ID can only contain characters a-z, 0-9, or '=_-./'"
216	)
217
UNCOV 218	return group_id	!
219
220
221	mxid_localpart_allowed_characters = set(	1×
222	"_-./=" + string.ascii_lowercase + string.digits
223	)
224
225
226	def contains_invalid_mxid_characters(localpart):	1×
227	"""Check for characters not allowed in an mxid or groupid localpart
228
229	Args:
230	localpart (basestring): the localpart to be checked
231
232	Returns:
233	bool: True if there are any naughty characters
234	"""
UNCOV 235	return any(c not in mxid_localpart_allowed_characters for c in localpart)	Branches [[0, 235], [0, 226]] missed. !
236
237
238	UPPER_CASE_PATTERN = re.compile(b"[A-Z_]")	1×
239
240	# the following is a pattern which matches '=', and bytes which are not allowed in a mxid
241	# localpart.
242	#
243	# It works by:
244	# * building a string containing the allowed characters (excluding '=')
245	# * escaping every special character with a backslash (to stop '-' being interpreted as a
246	# range operator)
247	# * wrapping it in a '[^...]' regex
248	# * converting the whole lot to a 'bytes' sequence, so that we can use it to match
249	# bytes rather than strings
250	#
251	NON_MXID_CHARACTER_PATTERN = re.compile(	1×
252	("[^%s]" % (re.escape("".join(mxid_localpart_allowed_characters - {"="})),)).encode(
253	"ascii"
254	)
255	)
256
257
258	def map_username_to_mxid_localpart(username, case_sensitive=False):	1×
259	"""Map a username onto a string suitable for a MXID
260
261	This follows the algorithm laid out at
262	https://matrix.org/docs/spec/appendices.html#mapping-from-other-character-sets.
263
264	Args:
265	username (unicode\|bytes): username to be mapped
266	case_sensitive (bool): true if TEST and test should be mapped
267	onto different mxids
268
269	Returns:
270	unicode: string suitable for a mxid localpart
271	"""
UNCOV 272	if not isinstance(username, bytes):	Branches [[0, 273], [0, 276]] missed. !
UNCOV 273	username = username.encode("utf-8")	!
274
275	# first we sort out upper-case characters
UNCOV 276	if case_sensitive:	Branches [[0, 278], [0, 283]] missed. !
277
278	def f1(m):	!
279	return b"_" + m.group().lower()	!
280
281	username = UPPER_CASE_PATTERN.sub(f1, username)	!
282	else:
UNCOV 283	username = username.lower()	!
284
285	# then we sort out non-ascii characters
UNCOV 286	def f2(m):	!
UNCOV 287	g = m.group()[0]	!
UNCOV 288	if isinstance(g, str):	Branches [[0, 291], [0, 292]] missed. !
289	# on python 2, we need to do a ord(). On python 3, the
290	# byte itself will do.
291	g = ord(g)	!
UNCOV 292	return b"=%02x" % (g,)	!
293
UNCOV 294	username = NON_MXID_CHARACTER_PATTERN.sub(f2, username)	!
295
296	# we also do the =-escaping to mxids starting with an underscore.
UNCOV 297	username = re.sub(b"^_", b"=5f", username)	!
298
299	# we should now only have ascii bytes left, so can decode back to a
300	# unicode.
UNCOV 301	return username.decode("ascii")	!
302
303
304	class StreamToken(	1×
305	namedtuple(
306	"Token",
307	(
308	"room_key",
309	"presence_key",
310	"typing_key",
311	"receipt_key",
312	"account_data_key",
313	"push_rules_key",
314	"to_device_key",
315	"device_list_key",
316	"groups_key",
317	),
318	)
319	):
320	_SEPARATOR = "_"	1×
321
322	@classmethod	1×
323	def from_string(cls, string):
UNCOV 324	try:	!
UNCOV 325	keys = string.split(cls._SEPARATOR)	!
UNCOV 326	while len(keys) < len(cls._fields):	Branches [[0, 328], [0, 329]] missed. !
327	# i.e. old token from before receipt_key
328	keys.append("0")	!
UNCOV 329	return cls(*keys)	!
330	except Exception:	!
331	raise SynapseError(400, "Invalid Token")	!
332
333	def to_string(self):	1×
UNCOV 334	return self._SEPARATOR.join([str(k) for k in self])	Branches [[0, 334], [0, 333]] missed. !
335
336	@property	1×
337	def room_stream_id(self):
338	# TODO(markjh): Awful hack to work around hacks in the presence tests
339	# which assume that the keys are integers.
UNCOV 340	if type(self.room_key) is int:	Branches [[0, 341], [0, 343]] missed. !
UNCOV 341	return self.room_key	!
342	else:
UNCOV 343	return int(self.room_key[1:].split("-")[-1])	!
344
345	def is_after(self, other):	1×
346	"""Does this token contain events that the other doesn't?"""
UNCOV 347	return (	!
348	(other.room_stream_id < self.room_stream_id)
349	or (int(other.presence_key) < int(self.presence_key))
350	or (int(other.typing_key) < int(self.typing_key))
351	or (int(other.receipt_key) < int(self.receipt_key))
352	or (int(other.account_data_key) < int(self.account_data_key))
353	or (int(other.push_rules_key) < int(self.push_rules_key))
354	or (int(other.to_device_key) < int(self.to_device_key))
355	or (int(other.device_list_key) < int(self.device_list_key))
356	or (int(other.groups_key) < int(self.groups_key))
357	)
358
359	def copy_and_advance(self, key, new_value):	1×
360	"""Advance the given key in the token to a new value if and only if the
361	new value is after the old value.
362	"""
UNCOV 363	new_token = self.copy_and_replace(key, new_value)	!
UNCOV 364	if key == "room_key":	Branches [[0, 365], [0, 368]] missed. !
UNCOV 365	new_id = new_token.room_stream_id	!
UNCOV 366	old_id = self.room_stream_id	!
367	else:
UNCOV 368	new_id = int(getattr(new_token, key))	!
UNCOV 369	old_id = int(getattr(self, key))	!
UNCOV 370	if old_id < new_id:	Branches [[0, 371], [0, 373]] missed. !
UNCOV 371	return new_token	!
372	else:
UNCOV 373	return self	!
374
375	def copy_and_replace(self, key, new_value):	1×
UNCOV 376	return self._replace(**{key: new_value})	!
377
378
379	StreamToken.START = StreamToken((["s0"] + ["0"] (len(StreamToken._fields) - 1)))	1×
380
381
382	class RoomStreamToken(namedtuple("_StreamToken", "topological stream")):	1×
383	"""Tokens are positions between events. The token "s1" comes after event 1.
384
385	s0 s1
386	\| \|
387	[0] V [1] V [2]
388
389	Tokens can either be a point in the live event stream or a cursor going
390	through historic events.
391
392	When traversing the live event stream events are ordered by when they
393	arrived at the homeserver.
394
395	When traversing historic events the events are ordered by their depth in
396	the event graph "topological_ordering" and then by when they arrived at the
397	homeserver "stream_ordering".
398
399	Live tokens start with an "s" followed by the "stream_ordering" id of the
400	event it comes after. Historic tokens start with a "t" followed by the
401	"topological_ordering" id of the event it comes after, followed by "-",
402	followed by the "stream_ordering" id of the event it comes after.
403	"""
404
405	__slots__ = []	1×
406
407	@classmethod	1×
408	def parse(cls, string):
UNCOV 409	try:	!
UNCOV 410	if string[0] == "s":	Branches [[0, 411], [0, 412]] missed. !
UNCOV 411	return cls(topological=None, stream=int(string[1:]))	!
UNCOV 412	if string[0] == "t":	Branches [[0, 413], [0, 417]] missed. !
UNCOV 413	parts = string[1:].split("-", 1)	!
UNCOV 414	return cls(topological=int(parts[0]), stream=int(parts[1]))	!
415	except Exception:	!
416	pass	!
417	raise SynapseError(400, "Invalid token %r" % (string,))	!
418
419	@classmethod	1×
420	def parse_stream_token(cls, string):
UNCOV 421	try:	!
UNCOV 422	if string[0] == "s":	Branches [[0, 423], [0, 426]] missed. !
UNCOV 423	return cls(topological=None, stream=int(string[1:]))	!
424	except Exception:	!
425	pass	!
426	raise SynapseError(400, "Invalid token %r" % (string,))	!
427
428	def __str__(self):	1×
UNCOV 429	if self.topological is not None:	Branches [[0, 430], [0, 432]] missed. !
UNCOV 430	return "t%d-%d" % (self.topological, self.stream)	!
431	else:
UNCOV 432	return "s%d" % (self.stream,)	!
433
434
435	class ThirdPartyInstanceID(	1×
436	namedtuple("ThirdPartyInstanceID", ("appservice_id", "network_id"))
437	):
438	# Deny iteration because it will bite you if you try to create a singleton
439	# set by:
440	# users = set(user)
441	def __iter__(self):	1×
442	raise ValueError("Attempted to iterate a %s" % (type(self).__name__,))	!
443
444	# Because this class is a namedtuple of strings, it is deeply immutable.
445	def __copy__(self):	1×
446	return self	!
447
448	def __deepcopy__(self, memo):	1×
449	return self	!
450
451	@classmethod	1×
452	def from_string(cls, s):
UNCOV 453	bits = s.split("\|", 2)	!
UNCOV 454	if len(bits) != 2:	Branches [[0, 455], [0, 457]] missed. !
455	raise SynapseError(400, "Invalid ID %r" % (s,))	!
456
UNCOV 457	return cls(appservice_id=bits[0], network_id=bits[1])	!
458
459	def to_string(self):	1×
460	return "%s\|%s" % (self.appservice_id, self.network_id)	!
461
462	__str__ = to_string	1×
463
464	@classmethod	1×
465	def create(cls, appservice_id, network_id):
466	return cls(appservice_id=appservice_id, network_id=network_id)	!
467
468
469	@attr.s(slots=True)	1×
470	class ReadReceipt(object):	1×
471	"""Information about a read-receipt"""
472
473	room_id = attr.ib()	1×
474	receipt_type = attr.ib()	1×
475	user_id = attr.ib()	1×
476	event_ids = attr.ib()	1×
477	data = attr.ib()	1×

matrix-org / synapse / 4532

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous