test_room_summary.py

            )
        self._assert_hierarchy(result, expected)

    def test_fed_caching(self) -> None:
        """
        Federation `/hierarchy` responses should be cached.
        """
        fed_hostname = self.hs.hostname + "2"
        fed_subspace = "#space:" + fed_hostname
        fed_room = "#room:" + fed_hostname

        # Add a room to the space which is on another server.
        self._add_child(self.space, fed_subspace, self.token, via=[fed_hostname])

        federation_requests = 0

        async def get_room_hierarchy(
            _self: TransportLayerClient,
            destination: str,
            room_id: str,
            suggested_only: bool,
        ) -> JsonDict:
            nonlocal federation_requests
            federation_requests += 1

            return {
                "room": {
                    "room_id": fed_subspace,
                    "world_readable": True,
                    "room_type": RoomTypes.SPACE,
                    "children_state": [
                        {
                            "type": EventTypes.SpaceChild,
                            "room_id": fed_subspace,
                            "state_key": fed_room,
                            "content": {"via": [fed_hostname]},
                        },
                    ],
                },
                "children": [
                    {
                        "room_id": fed_room,
                        "world_readable": True,
                    },
                ],
                "inaccessible_children": [],
            }

        expected = [
            (self.space, [self.room, fed_subspace]),
            (self.room, ()),
            (fed_subspace, [fed_room]),
            (fed_room, ()),
        ]

        with mock.patch(
            "synapse.federation.transport.client.TransportLayerClient.get_room_hierarchy",
            new=get_room_hierarchy,
        ):
            result = self.get_success(
                self.handler.get_room_hierarchy(create_requester(self.user), self.space)
            )
            self.assertEqual(federation_requests, 1)
            self._assert_hierarchy(result, expected)

            # The previous federation response should be reused.
            result = self.get_success(
                self.handler.get_room_hierarchy(create_requester(self.user), self.space)
            )
            self.assertEqual(federation_requests, 1)
            self._assert_hierarchy(result, expected)

            # Expire the response cache
            self.reactor.advance(5 * 60 + 1)

            # A new federation request should be made.
            result = self.get_success(
                self.handler.get_room_hierarchy(create_requester(self.user), self.space)
            )
            self.assertEqual(federation_requests, 2)
            self._assert_hierarchy(result, expected)


class RoomSummaryTestCase(unittest.HomeserverTestCase):
    servlets = [
        admin.register_servlets_for_client_rest_resource,
        room.register_servlets,
        login.register_servlets,
    ]

    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
        self.hs = hs
        self.handler = self.hs.get_room_summary_handler()

        # Create a user.
        self.user = self.register_user("user", "pass")
        self.token = self.login("user", "pass")

        # Create a simple room.
        self.room = self.helper.create_room_as(self.user, tok=self.token)
        self.helper.send_state(
            self.room,
            event_type=EventTypes.JoinRules,
            body={"join_rule": JoinRules.INVITE},
            tok=self.token,
        )

    def test_own_room(self) -> None:
        """Test a simple room created by the requester."""
        result = self.get_success(self.handler.get_room_summary(self.user, self.room))
        self.assertEqual(result.get("room_id"), self.room)

    def test_visibility(self) -> None:
        """A user not in a private room cannot get its summary."""
        user2 = self.register_user("user2", "pass")
        token2 = self.login("user2", "pass")

        # The user cannot see the room.
        self.get_failure(self.handler.get_room_summary(user2, self.room), NotFoundError)

        # If the room is made world-readable it should return a result.
        self.helper.send_state(
            self.room,
            event_type=EventTypes.RoomHistoryVisibility,
            body={"history_visibility": HistoryVisibility.WORLD_READABLE},
            tok=self.token,
        )
        result = self.get_success(self.handler.get_room_summary(user2, self.room))
        self.assertEqual(result.get("room_id"), self.room)

        # Make it not world-readable again and confirm it results in an error.
        self.helper.send_state(
            self.room,
            event_type=EventTypes.RoomHistoryVisibility,
            body={"history_visibility": HistoryVisibility.JOINED},
            tok=self.token,
        )
        self.get_failure(self.handler.get_room_summary(user2, self.room), NotFoundError)

        # If the room is made public it should return a result.
        self.helper.send_state(
            self.room,
            event_type=EventTypes.JoinRules,
            body={"join_rule": JoinRules.PUBLIC},
            tok=self.token,
        )
        result = self.get_success(self.handler.get_room_summary(user2, self.room))
        self.assertEqual(result.get("room_id"), self.room)

        # Join the space, make it invite-only again and results should be returned.
        self.helper.join(self.room, user2, tok=token2)
        self.helper.send_state(
            self.room,
            event_type=EventTypes.JoinRules,
            body={"join_rule": JoinRules.INVITE},
            tok=self.token,
        )
        result = self.get_success(self.handler.get_room_summary(user2, self.room))
        self.assertEqual(result.get("room_id"), self.room)

    def test_fed(self) -> None:
        """
        Return data over federation and ensure that it is handled properly.
        """
        fed_hostname = self.hs.hostname + "2"
        fed_room = "#fed_room:" + fed_hostname

        requested_room_entry = _RoomEntry(
            fed_room,
            {"room_id": fed_room, "world_readable": True},
        )

        async def summarize_remote_room_hierarchy(
            _self: Any, room: Any, suggested_only: bool
        ) -> Tuple[Optional[_RoomEntry], Dict[str, JsonDict], Set[str]]:
            return requested_room_entry, {}, set()

        with mock.patch(
            "synapse.handlers.room_summary.RoomSummaryHandler._summarize_remote_room_hierarchy",
            new=summarize_remote_room_hierarchy,
        ):
            result = self.get_success(
                self.handler.get_room_summary(
                    self.user, fed_room, remote_room_hosts=[fed_hostname]
                )
            )
        self.assertEqual(result.get("room_id"), fed_room)