forked from MirrorHub/synapse
Factor out a copy_power_levels_contents
method
I'm going to need another copy (hah!) of this.
This commit is contained in:
parent
6b9e1014cf
commit
ee42a5513e
3 changed files with 90 additions and 15 deletions
|
@ -12,8 +12,9 @@
|
|||
# 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 collections
|
||||
import re
|
||||
from typing import Mapping, Union
|
||||
|
||||
from six import string_types
|
||||
|
||||
|
@ -422,3 +423,37 @@ class EventClientSerializer(object):
|
|||
return yieldable_gather_results(
|
||||
self.serialize_event, events, time_now=time_now, **kwargs
|
||||
)
|
||||
|
||||
|
||||
def copy_power_levels_contents(
|
||||
old_power_levels: Mapping[str, Union[int, Mapping[str, int]]]
|
||||
):
|
||||
"""Copy the content of a power_levels event, unfreezing frozendicts along the way
|
||||
|
||||
Raises:
|
||||
TypeError if the input does not look like a valid power levels event content
|
||||
"""
|
||||
if not isinstance(old_power_levels, collections.Mapping):
|
||||
raise TypeError("Not a valid power-levels content: %r" % (old_power_levels,))
|
||||
|
||||
power_levels = {}
|
||||
for k, v in old_power_levels.items():
|
||||
|
||||
if isinstance(v, int):
|
||||
power_levels[k] = v
|
||||
continue
|
||||
|
||||
if isinstance(v, collections.Mapping):
|
||||
power_levels[k] = h = {}
|
||||
for k1, v1 in v.items():
|
||||
# we should only have one level of nesting
|
||||
if not isinstance(v1, int):
|
||||
raise TypeError(
|
||||
"Invalid power_levels value for %s.%s: %r" % (k, k1, v)
|
||||
)
|
||||
h[k1] = v1
|
||||
continue
|
||||
|
||||
raise TypeError("Invalid power_levels value for %s: %r" % (k, v))
|
||||
|
||||
return power_levels
|
||||
|
|
|
@ -30,6 +30,7 @@ from twisted.internet import defer
|
|||
from synapse.api.constants import EventTypes, JoinRules, RoomCreationPreset
|
||||
from synapse.api.errors import AuthError, Codes, NotFoundError, StoreError, SynapseError
|
||||
from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion
|
||||
from synapse.events.utils import copy_power_levels_contents
|
||||
from synapse.http.endpoint import parse_and_validate_server_name
|
||||
from synapse.storage.state import StateFilter
|
||||
from synapse.types import (
|
||||
|
@ -367,6 +368,15 @@ class RoomCreationHandler(BaseHandler):
|
|||
if old_event:
|
||||
initial_state[k] = old_event.content
|
||||
|
||||
# deep-copy the power-levels event before we start modifying it
|
||||
# note that if frozen_dicts are enabled, `power_levels` will be a frozen
|
||||
# dict so we can't just copy.deepcopy it.
|
||||
initial_state[
|
||||
(EventTypes.PowerLevels, "")
|
||||
] = power_levels = copy_power_levels_contents(
|
||||
initial_state[(EventTypes.PowerLevels, "")]
|
||||
)
|
||||
|
||||
# Resolve the minimum power level required to send any state event
|
||||
# We will give the upgrading user this power level temporarily (if necessary) such that
|
||||
# they are able to copy all of the state events over, then revert them back to their
|
||||
|
@ -375,8 +385,6 @@ class RoomCreationHandler(BaseHandler):
|
|||
# Copy over user power levels now as this will not be possible with >100PL users once
|
||||
# the room has been created
|
||||
|
||||
power_levels = initial_state[(EventTypes.PowerLevels, "")]
|
||||
|
||||
# Calculate the minimum power level needed to clone the room
|
||||
event_power_levels = power_levels.get("events", {})
|
||||
state_default = power_levels.get("state_default", 0)
|
||||
|
@ -386,16 +394,7 @@ class RoomCreationHandler(BaseHandler):
|
|||
# Raise the requester's power level in the new room if necessary
|
||||
current_power_level = power_levels["users"][user_id]
|
||||
if current_power_level < needed_power_level:
|
||||
# make sure we copy the event content rather than overwriting it.
|
||||
# note that if frozen_dicts are enabled, `power_levels` will be a frozen
|
||||
# dict so we can't just copy.deepcopy it.
|
||||
|
||||
new_power_levels = {k: v for k, v in power_levels.items() if k != "users"}
|
||||
new_power_levels["users"] = {
|
||||
k: v for k, v in power_levels.get("users", {}).items() if k != user_id
|
||||
}
|
||||
new_power_levels["users"][user_id] = needed_power_level
|
||||
initial_state[(EventTypes.PowerLevels, "")] = new_power_levels
|
||||
power_levels["users"][user_id] = needed_power_level
|
||||
|
||||
yield self._send_events_for_new_room(
|
||||
requester,
|
||||
|
|
|
@ -15,9 +15,14 @@
|
|||
|
||||
|
||||
from synapse.events import FrozenEvent
|
||||
from synapse.events.utils import prune_event, serialize_event
|
||||
from synapse.events.utils import (
|
||||
copy_power_levels_contents,
|
||||
prune_event,
|
||||
serialize_event,
|
||||
)
|
||||
from synapse.util.frozenutils import freeze
|
||||
|
||||
from .. import unittest
|
||||
from tests import unittest
|
||||
|
||||
|
||||
def MockEvent(**kwargs):
|
||||
|
@ -241,3 +246,39 @@ class SerializeEventTestCase(unittest.TestCase):
|
|||
self.serialize(
|
||||
MockEvent(room_id="!foo:bar", content={"foo": "bar"}), ["room_id", 4]
|
||||
)
|
||||
|
||||
|
||||
class CopyPowerLevelsContentTestCase(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.test_content = {
|
||||
"ban": 50,
|
||||
"events": {"m.room.name": 100, "m.room.power_levels": 100},
|
||||
"events_default": 0,
|
||||
"invite": 50,
|
||||
"kick": 50,
|
||||
"notifications": {"room": 20},
|
||||
"redact": 50,
|
||||
"state_default": 50,
|
||||
"users": {"@example:localhost": 100},
|
||||
"users_default": 0,
|
||||
}
|
||||
|
||||
def _test(self, input):
|
||||
a = copy_power_levels_contents(input)
|
||||
|
||||
self.assertEqual(a["ban"], 50)
|
||||
self.assertEqual(a["events"]["m.room.name"], 100)
|
||||
|
||||
# make sure that changing the copy changes the copy and not the orig
|
||||
a["ban"] = 10
|
||||
a["events"]["m.room.power_levels"] = 20
|
||||
|
||||
self.assertEqual(input["ban"], 50)
|
||||
self.assertEqual(input["events"]["m.room.power_levels"], 100)
|
||||
|
||||
def test_unfrozen(self):
|
||||
self._test(self.test_content)
|
||||
|
||||
def test_frozen(self):
|
||||
input = freeze(self.test_content)
|
||||
self._test(input)
|
||||
|
|
Loading…
Reference in a new issue