Matrix-Mirrors/synapse-product
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse-product.git synced 2024-12-12 10:14:19 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add device management to admin API (#7481)

Browse Source 
- Admin is able to
  - change displaynames
  - delete devices
  - list devices
  - get device informations

Fixes #7330
This commit is contained in:
Dirk Klimpel 2020-06-05 14:07:22 +02:00 committed by GitHub
parent 02f345d053
commit 2970ce8367
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 920 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/7481.feature Normal file
View File

@ -0,0 +1 @@
Add admin APIs to allow server admins to manage users' devices. Contributed by @dklimpel.

209
docs/admin_api/user_admin_api.rst
View File

@ -1,3 +1,5 @@
.. contents::
Create or modify Account
========================
@ -245,3 +247,210 @@ with a body of:
}
including an ``access_token`` of a server admin.
User devices
============
List all devices
----------------
Gets information about all devices for a specific ``user_id``.
**Usage**
A standard request to query the devices of an user:
::
GET /_synapse/admin/v2/users/<user_id>/devices
{}
Response:
.. code:: json
{
"devices": [
{
"device_id": "QBUAZIFURK",
"display_name": "android",
"last_seen_ip": "1.2.3.4",
"last_seen_ts": 1474491775024,
"user_id": "<user_id>"
},
{
"device_id": "AUIECTSRND",
"display_name": "ios",
"last_seen_ip": "1.2.3.5",
"last_seen_ts": 1474491775025,
"user_id": "<user_id>"
}
]
}
**Parameters**
The following query parameters are available:
- ``user_id`` - fully qualified: for example, ``@user:server.com``.
The following fields are possible in the JSON response body:
- ``devices`` - An array of objects, each containing information about a device.
Device objects contain the following fields:
- ``device_id`` - Identifier of device.
- ``display_name`` - Display name set by the user for this device.
Absent if no name has been set.
- ``last_seen_ip`` - The IP address where this device was last seen.
(May be a few minutes out of date, for efficiency reasons).
- ``last_seen_ts`` - The timestamp (in milliseconds since the unix epoch) when this
devices was last seen. (May be a few minutes out of date, for efficiency reasons).
- ``user_id`` - Owner of device.
Delete multiple devices
------------------
Deletes the given devices for a specific ``user_id``, and invalidates
any access token associated with them.
**Usage**
A standard request to delete devices:
::
POST /_synapse/admin/v2/users/<user_id>/delete_devices
{
"devices": [
"QBUAZIFURK",
"AUIECTSRND"
],
}
Response:
.. code:: json
{}
**Parameters**
The following query parameters are available:
- ``user_id`` - fully qualified: for example, ``@user:server.com``.
The following fields are required in the JSON request body:
- ``devices`` - The list of device IDs to delete.
Show a device
---------------
Gets information on a single device, by ``device_id`` for a specific ``user_id``.
**Usage**
A standard request to get a device:
::
GET /_synapse/admin/v2/users/<user_id>/devices/<device_id>
{}
Response:
.. code:: json
{
"device_id": "<device_id>",
"display_name": "android",
"last_seen_ip": "1.2.3.4",
"last_seen_ts": 1474491775024,
"user_id": "<user_id>"
}
**Parameters**
The following query parameters are available:
- ``user_id`` - fully qualified: for example, ``@user:server.com``.
- ``device_id`` - The device to retrieve.
The following fields are possible in the JSON response body:
- ``device_id`` - Identifier of device.
- ``display_name`` - Display name set by the user for this device.
Absent if no name has been set.
- ``last_seen_ip`` - The IP address where this device was last seen.
(May be a few minutes out of date, for efficiency reasons).
- ``last_seen_ts`` - The timestamp (in milliseconds since the unix epoch) when this
devices was last seen. (May be a few minutes out of date, for efficiency reasons).
- ``user_id`` - Owner of device.
Update a device
---------------
Updates the metadata on the given ``device_id`` for a specific ``user_id``.
**Usage**
A standard request to update a device:
::
PUT /_synapse/admin/v2/users/<user_id>/devices/<device_id>
{
"display_name": "My other phone"
}
Response:
.. code:: json
{}
**Parameters**
The following query parameters are available:
- ``user_id`` - fully qualified: for example, ``@user:server.com``.
- ``device_id`` - The device to update.
The following fields are required in the JSON request body:
- ``display_name`` - The new display name for this device. If not given,
the display name is unchanged.
Delete a device
---------------
Deletes the given ``device_id`` for a specific ``user_id``,
and invalidates any access token associated with it.
**Usage**
A standard request for delete a device:
::
DELETE /_synapse/admin/v2/users/<user_id>/devices/<device_id>
{}
Response:
.. code:: json
{}
**Parameters**
The following query parameters are available:
- ``user_id`` - fully qualified: for example, ``@user:server.com``.
- ``device_id`` - The device to delete.

8
synapse/rest/admin/__init__.py
View File

@ -26,6 +26,11 @@ from synapse.rest.admin._base import (
assert_requester_is_admin,
historical_admin_path_patterns,
)
from synapse.rest.admin.devices import (
DeleteDevicesRestServlet,
DeviceRestServlet,
DevicesRestServlet,
)
from synapse.rest.admin.groups import DeleteGroupAdminRestServlet
from synapse.rest.admin.media import ListMediaInRoom, register_servlets_for_media_repo
from synapse.rest.admin.purge_room_servlet import PurgeRoomServlet
@ -202,6 +207,9 @@ def register_servlets(hs, http_server):
UserAdminServlet(hs).register(http_server)
UserRestServletV2(hs).register(http_server)
UsersRestServletV2(hs).register(http_server)
DeviceRestServlet(hs).register(http_server)
DevicesRestServlet(hs).register(http_server)
DeleteDevicesRestServlet(hs).register(http_server)
def register_servlets_for_client_rest_resource(hs, http_server):

161
synapse/rest/admin/devices.py Normal file
View File

@ -0,0 +1,161 @@
# -*- coding: utf-8 -*-
# Copyright 2020 Dirk Klimpel
#
# 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 logging
import re
from synapse.api.errors import NotFoundError, SynapseError
from synapse.http.servlet import (
RestServlet,
assert_params_in_dict,
parse_json_object_from_request,
)
from synapse.rest.admin._base import assert_requester_is_admin
from synapse.types import UserID
logger = logging.getLogger(__name__)
class DeviceRestServlet(RestServlet):
"""
Get, update or delete the given user's device
"""
PATTERNS = (
re.compile(
"^/_synapse/admin/v2/users/(?P<user_id>[^/]*)/devices/(?P<device_id>[^/]*)$"
),
)
def __init__(self, hs):
super(DeviceRestServlet, self).__init__()
self.hs = hs
self.auth = hs.get_auth()
self.device_handler = hs.get_device_handler()
self.store = hs.get_datastore()
async def on_GET(self, request, user_id, device_id):
await assert_requester_is_admin(self.auth, request)
target_user = UserID.from_string(user_id)
if not self.hs.is_mine(target_user):
raise SynapseError(400, "Can only lookup local users")
u = await self.store.get_user_by_id(target_user.to_string())
if u is None:
raise NotFoundError("Unknown user")
device = await self.device_handler.get_device(
target_user.to_string(), device_id
)
return 200, device
async def on_DELETE(self, request, user_id, device_id):
await assert_requester_is_admin(self.auth, request)
target_user = UserID.from_string(user_id)
if not self.hs.is_mine(target_user):
raise SynapseError(400, "Can only lookup local users")
u = await self.store.get_user_by_id(target_user.to_string())
if u is None:
raise NotFoundError("Unknown user")
await self.device_handler.delete_device(target_user.to_string(), device_id)
return 200, {}
async def on_PUT(self, request, user_id, device_id):
await assert_requester_is_admin(self.auth, request)
target_user = UserID.from_string(user_id)
if not self.hs.is_mine(target_user):
raise SynapseError(400, "Can only lookup local users")
u = await self.store.get_user_by_id(target_user.to_string())
if u is None:
raise NotFoundError("Unknown user")
body = parse_json_object_from_request(request, allow_empty_body=True)
await self.device_handler.update_device(
target_user.to_string(), device_id, body
)
return 200, {}
class DevicesRestServlet(RestServlet):
"""
Retrieve the given user's devices
"""
PATTERNS = (re.compile("^/_synapse/admin/v2/users/(?P<user_id>[^/]*)/devices$"),)
def __init__(self, hs):
"""
Args:
hs (synapse.server.HomeServer): server
"""
self.hs = hs
self.auth = hs.get_auth()
self.device_handler = hs.get_device_handler()
self.store = hs.get_datastore()
async def on_GET(self, request, user_id):
await assert_requester_is_admin(self.auth, request)
target_user = UserID.from_string(user_id)
if not self.hs.is_mine(target_user):
raise SynapseError(400, "Can only lookup local users")
u = await self.store.get_user_by_id(target_user.to_string())
if u is None:
raise NotFoundError("Unknown user")
devices = await self.device_handler.get_devices_by_user(target_user.to_string())
return 200, {"devices": devices}
class DeleteDevicesRestServlet(RestServlet):
"""
API for bulk deletion of devices. Accepts a JSON object with a devices
key which lists the device_ids to delete.
"""
PATTERNS = (
re.compile("^/_synapse/admin/v2/users/(?P<user_id>[^/]*)/delete_devices$"),
)
def __init__(self, hs):
self.hs = hs
self.auth = hs.get_auth()
self.device_handler = hs.get_device_handler()
self.store = hs.get_datastore()
async def on_POST(self, request, user_id):
await assert_requester_is_admin(self.auth, request)
target_user = UserID.from_string(user_id)
if not self.hs.is_mine(target_user):
raise SynapseError(400, "Can only lookup local users")
u = await self.store.get_user_by_id(target_user.to_string())
if u is None:
raise NotFoundError("Unknown user")
body = parse_json_object_from_request(request, allow_empty_body=False)
assert_params_in_dict(body, ["devices"])
await self.device_handler.delete_devices(
target_user.to_string(), body["devices"]
)
return 200, {}

541
tests/rest/admin/test_device.py Normal file
View File

@ -0,0 +1,541 @@
# -*- coding: utf-8 -*-
# Copyright 2020 Dirk Klimpel
#
# 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 json
import urllib.parse
import synapse.rest.admin
from synapse.api.errors import Codes
from synapse.rest.client.v1 import login
from tests import unittest
class DeviceRestTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets,
login.register_servlets,
]
def prepare(self, reactor, clock, hs):
self.handler = hs.get_device_handler()
self.admin_user = self.register_user("admin", "pass", admin=True)
self.admin_user_tok = self.login("admin", "pass")
self.other_user = self.register_user("user", "pass")
self.other_user_token = self.login("user", "pass")
res = self.get_success(self.handler.get_devices_by_user(self.other_user))
self.other_user_device_id = res[0]["device_id"]
self.url = "/_synapse/admin/v2/users/%s/devices/%s" % (
urllib.parse.quote(self.other_user),
self.other_user_device_id,
)
def test_no_auth(self):
"""
Try to get a device of an user without authentication.
"""
request, channel = self.make_request("GET", self.url, b"{}")
self.render(request)
self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
request, channel = self.make_request("PUT", self.url, b"{}")
self.render(request)
self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
request, channel = self.make_request("DELETE", self.url, b"{}")
self.render(request)
self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
def test_requester_is_no_admin(self):
"""
If the user is not a server admin, an error is returned.
"""
request, channel = self.make_request(
"GET", self.url, access_token=self.other_user_token,
)
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
request, channel = self.make_request(
"PUT", self.url, access_token=self.other_user_token,
)
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
request, channel = self.make_request(
"DELETE", self.url, access_token=self.other_user_token,
)
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
def test_user_does_not_exist(self):
"""
Tests that a lookup for a user that does not exist returns a 404
"""
url = (
"/_synapse/admin/v2/users/@unknown_person:test/devices/%s"
% self.other_user_device_id
)
request, channel = self.make_request(
"GET", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(404, channel.code, msg=channel.json_body)
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
request, channel = self.make_request(
"PUT", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(404, channel.code, msg=channel.json_body)
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
request, channel = self.make_request(
"DELETE", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(404, channel.code, msg=channel.json_body)
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
def test_user_is_not_local(self):
"""
Tests that a lookup for a user that is not a local returns a 400
"""
url = (
"/_synapse/admin/v2/users/@unknown_person:unknown_domain/devices/%s"
% self.other_user_device_id
)
request, channel = self.make_request(
"GET", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(400, channel.code, msg=channel.json_body)
self.assertEqual("Can only lookup local users", channel.json_body["error"])
request, channel = self.make_request(
"PUT", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(400, channel.code, msg=channel.json_body)
self.assertEqual("Can only lookup local users", channel.json_body["error"])
request, channel = self.make_request(
"DELETE", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(400, channel.code, msg=channel.json_body)
self.assertEqual("Can only lookup local users", channel.json_body["error"])
def test_unknown_device(self):
"""
Tests that a lookup for a device that does not exist returns either 404 or 200.
"""
url = "/_synapse/admin/v2/users/%s/devices/unknown_device" % urllib.parse.quote(
self.other_user
)
request, channel = self.make_request(
"GET", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(404, channel.code, msg=channel.json_body)
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
request, channel = self.make_request(
"PUT", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
request, channel = self.make_request(
"DELETE", url, access_token=self.admin_user_tok,
)
self.render(request)
# Delete unknown device returns status 200
self.assertEqual(200, channel.code, msg=channel.json_body)
def test_update_device_too_long_display_name(self):
"""
Update a device with a display name that is invalid (too long).
"""
# Set iniital display name.
update = {"display_name": "new display"}
self.get_success(
self.handler.update_device(
self.other_user, self.other_user_device_id, update
)
)
# Request to update a device display name with a new value that is longer than allowed.
update = {
"display_name": "a"
* (synapse.handlers.device.MAX_DEVICE_DISPLAY_NAME_LEN + 1)
}
body = json.dumps(update)
request, channel = self.make_request(
"PUT",
self.url,
access_token=self.admin_user_tok,
content=body.encode(encoding="utf_8"),
)
self.render(request)
self.assertEqual(400, channel.code, msg=channel.json_body)
self.assertEqual(Codes.UNKNOWN, channel.json_body["errcode"])
# Ensure the display name was not updated.
request, channel = self.make_request(
"GET", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
self.assertEqual("new display", channel.json_body["display_name"])
def test_update_no_display_name(self):
"""
Tests that a update for a device without JSON returns a 200
"""
# Set iniital display name.
update = {"display_name": "new display"}
self.get_success(
self.handler.update_device(
self.other_user, self.other_user_device_id, update
)
)
request, channel = self.make_request(
"PUT", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
# Ensure the display name was not updated.
request, channel = self.make_request(
"GET", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
self.assertEqual("new display", channel.json_body["display_name"])
def test_update_display_name(self):
"""
Tests a normal successful update of display name
"""
# Set new display_name
body = json.dumps({"display_name": "new displayname"})
request, channel = self.make_request(
"PUT",
self.url,
access_token=self.admin_user_tok,
content=body.encode(encoding="utf_8"),
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
# Check new display_name
request, channel = self.make_request(
"GET", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
self.assertEqual("new displayname", channel.json_body["display_name"])
def test_get_device(self):
"""
Tests that a normal lookup for a device is successfully
"""
request, channel = self.make_request(
"GET", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
self.assertEqual(self.other_user, channel.json_body["user_id"])
# Check that all fields are available
self.assertIn("user_id", channel.json_body)
self.assertIn("device_id", channel.json_body)
self.assertIn("display_name", channel.json_body)
self.assertIn("last_seen_ip", channel.json_body)
self.assertIn("last_seen_ts", channel.json_body)
def test_delete_device(self):
"""
Tests that a remove of a device is successfully
"""
# Count number of devies of an user.
res = self.get_success(self.handler.get_devices_by_user(self.other_user))
number_devices = len(res)
self.assertEqual(1, number_devices)
# Delete device
request, channel = self.make_request(
"DELETE", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
# Ensure that the number of devices is decreased
res = self.get_success(self.handler.get_devices_by_user(self.other_user))
self.assertEqual(number_devices - 1, len(res))
class DevicesRestTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets,
login.register_servlets,
]
def prepare(self, reactor, clock, hs):
self.admin_user = self.register_user("admin", "pass", admin=True)
self.admin_user_tok = self.login("admin", "pass")
self.other_user = self.register_user("user", "pass")
self.url = "/_synapse/admin/v2/users/%s/devices" % urllib.parse.quote(
self.other_user
)
def test_no_auth(self):
"""
Try to list devices of an user without authentication.
"""
request, channel = self.make_request("GET", self.url, b"{}")
self.render(request)
self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
def test_requester_is_no_admin(self):
"""
If the user is not a server admin, an error is returned.
"""
other_user_token = self.login("user", "pass")
request, channel = self.make_request(
"GET", self.url, access_token=other_user_token,
)
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
def test_user_does_not_exist(self):
"""
Tests that a lookup for a user that does not exist returns a 404
"""
url = "/_synapse/admin/v2/users/@unknown_person:test/devices"
request, channel = self.make_request(
"GET", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(404, channel.code, msg=channel.json_body)
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
def test_user_is_not_local(self):
"""
Tests that a lookup for a user that is not a local returns a 400
"""
url = "/_synapse/admin/v2/users/@unknown_person:unknown_domain/devices"
request, channel = self.make_request(
"GET", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(400, channel.code, msg=channel.json_body)
self.assertEqual("Can only lookup local users", channel.json_body["error"])
def test_get_devices(self):
"""
Tests that a normal lookup for devices is successfully
"""
# Create devices
number_devices = 5
for n in range(number_devices):
self.login("user", "pass")
# Get devices
request, channel = self.make_request(
"GET", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
self.assertEqual(number_devices, len(channel.json_body["devices"]))
self.assertEqual(self.other_user, channel.json_body["devices"][0]["user_id"])
# Check that all fields are available
for d in channel.json_body["devices"]:
self.assertIn("user_id", d)
self.assertIn("device_id", d)
self.assertIn("display_name", d)
self.assertIn("last_seen_ip", d)
self.assertIn("last_seen_ts", d)
class DeleteDevicesRestTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets,
login.register_servlets,
]
def prepare(self, reactor, clock, hs):
self.handler = hs.get_device_handler()
self.admin_user = self.register_user("admin", "pass", admin=True)
self.admin_user_tok = self.login("admin", "pass")
self.other_user = self.register_user("user", "pass")
self.url = "/_synapse/admin/v2/users/%s/delete_devices" % urllib.parse.quote(
self.other_user
)
def test_no_auth(self):
"""
Try to delete devices of an user without authentication.
"""
request, channel = self.make_request("POST", self.url, b"{}")
self.render(request)
self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
def test_requester_is_no_admin(self):
"""
If the user is not a server admin, an error is returned.
"""
other_user_token = self.login("user", "pass")
request, channel = self.make_request(
"POST", self.url, access_token=other_user_token,
)
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
def test_user_does_not_exist(self):
"""
Tests that a lookup for a user that does not exist returns a 404
"""
url = "/_synapse/admin/v2/users/@unknown_person:test/delete_devices"
request, channel = self.make_request(
"POST", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(404, channel.code, msg=channel.json_body)
self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
def test_user_is_not_local(self):
"""
Tests that a lookup for a user that is not a local returns a 400
"""
url = "/_synapse/admin/v2/users/@unknown_person:unknown_domain/delete_devices"
request, channel = self.make_request(
"POST", url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(400, channel.code, msg=channel.json_body)
self.assertEqual("Can only lookup local users", channel.json_body["error"])
def test_unknown_devices(self):
"""
Tests that a remove of a device that does not exist returns 200.
"""
body = json.dumps({"devices": ["unknown_device1", "unknown_device2"]})
request, channel = self.make_request(
"POST",
self.url,
access_token=self.admin_user_tok,
content=body.encode(encoding="utf_8"),
)
self.render(request)
# Delete unknown devices returns status 200
self.assertEqual(200, channel.code, msg=channel.json_body)
def test_delete_devices(self):
"""
Tests that a remove of devices is successfully
"""
# Create devices
number_devices = 5
for n in range(number_devices):
self.login("user", "pass")
# Get devices
res = self.get_success(self.handler.get_devices_by_user(self.other_user))
self.assertEqual(number_devices, len(res))
# Create list of device IDs
device_ids = []
for d in res:
device_ids.append(str(d["device_id"]))
# Delete devices
body = json.dumps({"devices": device_ids})
request, channel = self.make_request(
"POST",
self.url,
access_token=self.admin_user_tok,
content=body.encode(encoding="utf_8"),
)
self.render(request)
self.assertEqual(200, channel.code, msg=channel.json_body)
res = self.get_success(self.handler.get_devices_by_user(self.other_user))
self.assertEqual(0, len(res))