summaryrefslogtreecommitdiff
path: root/tests/test_push.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_push.py')
-rw-r--r--tests/test_push.py226
1 files changed, 118 insertions, 108 deletions
diff --git a/tests/test_push.py b/tests/test_push.py
index e6d732a..37688ad 100644
--- a/tests/test_push.py
+++ b/tests/test_push.py
@@ -2,32 +2,38 @@ import pytest
from api import *
-def test_account_destroy(account, push_server):
- dev = Device(account, "dev", pcb=push_server.good("09d0114e-3b23-4ba3-8474-efac7433e3ba"))
- account.destroy_account(account.email, "")
- p = push_server.wait()
- assert p[0] == "/09d0114e-3b23-4ba3-8474-efac7433e3ba"
- assert dev.decrypt(p[2]) == {
- 'command': 'fxaccounts:account_destroyed',
- 'data': {'uid': account.props['uid']},
- 'version': 1,
- }
- assert push_server.done()
+# scope each test to not leak devices into the others
-def test_account_verify(client, push_server, mail_server):
- account = client.create_account("test@test", "")
- try:
- (to, body) = mail_server.wait()
- assert to == ["test@test"]
- data = json.loads(base64.urlsafe_b64decode(body.split("#/verify/", maxsplit=1)[1]).decode('utf8'))
- dev = Device(account, "dev", pcb=push_server.good("8accbe08-4040-44c2-8fd9-cf2669b56cb1"))
- account.post_a('/recovery_email/verify_code', { 'uid': data['uid'], 'code': data['code'] })
+class TestAccountDestroy:
+ def test_account_destroy(self, account, push_server):
+ dev = Device(account, "dev", pcb=push_server.good("09d0114e-3b23-4ba3-8474-efac7433e3ba"))
+ account.destroy_account(account.email, "")
p = push_server.wait()
- assert p[0] == "/8accbe08-4040-44c2-8fd9-cf2669b56cb1"
- assert p[2] == b''
+ assert p[0] == "/09d0114e-3b23-4ba3-8474-efac7433e3ba"
+ assert dev.decrypt(p[2]) == {
+ 'command': 'fxaccounts:account_destroyed',
+ 'data': {'uid': account.props['uid']},
+ 'version': 1,
+ }
assert push_server.done()
- finally:
- account.destroy_account(account.email, "")
+
+class TestAccountVerify:
+ def test_account_verify(self, client, push_server, mail_server):
+ account = client.create_account("test@test", "")
+ try:
+ (to, body) = mail_server.wait()
+ assert to == ["test@test"]
+ data = json.loads(base64.urlsafe_b64decode(body.split("#/verify/", maxsplit=1)[1]).decode('utf8'))
+ dev = Device(account, "dev", pcb=push_server.good("8accbe08-4040-44c2-8fd9-cf2669b56cb1"))
+ account.post_a('/recovery_email/verify_code', { 'uid': data['uid'], 'code': data['code'] })
+ p = push_server.wait()
+ assert p[0] == "/8accbe08-4040-44c2-8fd9-cf2669b56cb1"
+ assert p[2] == b''
+ assert push_server.done()
+ finally:
+ account.destroy_account(account.email, "")
+
+# these two destroy their sessions
def test_session_destroy(client, account, push_server):
dev1 = Device(account, "dev1")
@@ -57,91 +63,95 @@ def test_device_connected(client, account, push_server):
}
assert push_server.done()
-def test_device_invoke(account, login, push_server):
- dev = Device(account, "dev1", commands={'a':'a'}, pcb=push_server.good("3610b071-e2ef-4daa-a4e3-eaa74e50f2a0"))
- account.post_a("/account/devices/invoke_command", {
- "target": dev.id,
- "command": "a",
- "payload": {"data": "foo"},
- "ttl": 10,
- })
- p = push_server.wait()
- assert p[0] == "/3610b071-e2ef-4daa-a4e3-eaa74e50f2a0"
- msg = dev.decrypt(p[2])
- # NOTE needed because index is unpredictable due to pg sequence use
- del msg['data']['index']
- del msg['data']['url']
- assert msg == {
- 'command': 'fxaccounts:command_received',
- 'data': {'command': 'a', 'sender': dev.id},
- 'version': 1,
- }
- assert push_server.done()
+class TestDeviceInvoke:
+ def test_device_invoke(self, account, login, push_server):
+ dev = Device(account, "dev1", commands={'a':'a'}, pcb=push_server.good("3610b071-e2ef-4daa-a4e3-eaa74e50f2a0"))
+ account.post_a("/account/devices/invoke_command", {
+ "target": dev.id,
+ "command": "a",
+ "payload": {"data": "foo"},
+ "ttl": 10,
+ })
+ p = push_server.wait()
+ assert p[0] == "/3610b071-e2ef-4daa-a4e3-eaa74e50f2a0"
+ msg = dev.decrypt(p[2])
+ # NOTE needed because index is unpredictable due to pg sequence use
+ del msg['data']['index']
+ del msg['data']['url']
+ assert msg == {
+ 'command': 'fxaccounts:command_received',
+ 'data': {'command': 'a', 'sender': dev.id},
+ 'version': 1,
+ }
+ assert push_server.done()
-def test_expiry(account, login, push_server):
- dev = Device(account, "dev1", commands={'a':'a'}, pcb=push_server.bad("59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"))
- account.post_a("/account/devices/invoke_command", {
- "target": dev.id,
- "command": "a",
- "payload": {"data": "foo"},
- "ttl": 10,
- })
- p = push_server.wait()
- assert p[0] == "/err/59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"
- account.post_a("/account/devices/invoke_command", {
- "target": dev.id,
- "command": "a",
- "payload": {"data": "foo"},
- "ttl": 10,
- })
- with pytest.raises(queue.Empty):
- push_server.wait()
- dev.update_pcb(push_server.good("59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"))
- account.post_a("/account/devices/invoke_command", {
- "target": dev.id,
- "command": "a",
- "payload": {"data": "foo"},
- "ttl": 10,
- })
- p = push_server.wait()
- assert p[0] == "/59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"
- assert push_server.done()
+class TestExpiry:
+ def test_expiry(self, account, login, push_server):
+ dev = Device(account, "dev1", commands={'a':'a'}, pcb=push_server.bad("59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"))
+ account.post_a("/account/devices/invoke_command", {
+ "target": dev.id,
+ "command": "a",
+ "payload": {"data": "foo"},
+ "ttl": 10,
+ })
+ p = push_server.wait()
+ assert p[0] == "/err/59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"
+ account.post_a("/account/devices/invoke_command", {
+ "target": dev.id,
+ "command": "a",
+ "payload": {"data": "foo"},
+ "ttl": 10,
+ })
+ with pytest.raises(queue.Empty):
+ push_server.wait()
+ dev.update_pcb(push_server.good("59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"))
+ account.post_a("/account/devices/invoke_command", {
+ "target": dev.id,
+ "command": "a",
+ "payload": {"data": "foo"},
+ "ttl": 10,
+ })
+ p = push_server.wait()
+ assert p[0] == "/59ba8fcc-f3b0-4b1f-ac27-36d66e022d1e"
+ assert push_server.done()
-def test_device_notify(account, login, push_server):
- dev1 = Device(account, "dev1")
- dev2 = Device(login, "dev2")
- dev1.update_pcb(push_server.good("738ac7e3-96ef-461c-880c-0af20e311354"))
- dev2.update_pcb(push_server.good("85a98191-9486-46e2-877a-1152e3d4af4e"))
- account.post_a("/account/devices/notify", {
- "to": "all",
- "_endpointAction": "accountVerify",
- "excluded": [dev2.id],
- "payload": {'a':1},
- "TTL": 0,
- })
- p = push_server.wait()
- assert p[0] == "/738ac7e3-96ef-461c-880c-0af20e311354"
- assert dev1.decrypt(p[2]) == {'a':1}
- account.post_a("/account/devices/notify", {
- "to": [dev2.id],
- "_endpointAction": "accountVerify",
- "payload": {'a':2},
- "TTL": 0,
- })
- p = push_server.wait()
- assert p[0] == "/85a98191-9486-46e2-877a-1152e3d4af4e"
- assert dev2.decrypt(p[2]) == {'a':2}
- assert push_server.done()
+class TestDeviceNotify:
+ def test_device_notify(self, account, login, push_server):
+ dev1 = Device(account, "dev1")
+ dev2 = Device(login, "dev2")
+ dev1.update_pcb(push_server.good("738ac7e3-96ef-461c-880c-0af20e311354"))
+ dev2.update_pcb(push_server.good("85a98191-9486-46e2-877a-1152e3d4af4e"))
+ account.post_a("/account/devices/notify", {
+ "to": "all",
+ "_endpointAction": "accountVerify",
+ "excluded": [dev2.id],
+ "payload": {'a':1},
+ "TTL": 0,
+ })
+ p = push_server.wait()
+ assert p[0] == "/738ac7e3-96ef-461c-880c-0af20e311354"
+ assert dev1.decrypt(p[2]) == {'a':1}
+ account.post_a("/account/devices/notify", {
+ "to": [dev2.id],
+ "_endpointAction": "accountVerify",
+ "payload": {'a':2},
+ "TTL": 0,
+ })
+ p = push_server.wait()
+ assert p[0] == "/85a98191-9486-46e2-877a-1152e3d4af4e"
+ assert dev2.decrypt(p[2]) == {'a':2}
+ assert push_server.done()
-def test_profile(account, push_server):
- dev = Device(account, "dev", pcb=push_server.good("12608154-8942-4f1c-9de2-a56465d27d6e"))
- profile = account.profile()
- profile.post_a("/display_name", {"displayName": "foo"})
- p = push_server.wait()
- assert p[0] == "/12608154-8942-4f1c-9de2-a56465d27d6e"
- assert dev.decrypt(p[2]) == {'command': 'fxaccounts:profile_updated', 'version': 1}
- profile.post_a("/avatar/upload", "doesn't parse the image", headers={'content-type': 'image/png'})
- p = push_server.wait()
- assert p[0] == "/12608154-8942-4f1c-9de2-a56465d27d6e"
- assert dev.decrypt(p[2]) == {'command': 'fxaccounts:profile_updated', 'version': 1}
- assert push_server.done()
+class TestProfile:
+ def test_profile(self, account, push_server):
+ dev = Device(account, "dev", pcb=push_server.good("12608154-8942-4f1c-9de2-a56465d27d6e"))
+ profile = account.profile()
+ profile.post_a("/display_name", {"displayName": "foo"})
+ p = push_server.wait()
+ assert p[0] == "/12608154-8942-4f1c-9de2-a56465d27d6e"
+ assert dev.decrypt(p[2]) == {'command': 'fxaccounts:profile_updated', 'version': 1}
+ profile.post_a("/avatar/upload", "doesn't parse the image", headers={'content-type': 'image/png'})
+ p = push_server.wait()
+ assert p[0] == "/12608154-8942-4f1c-9de2-a56465d27d6e"
+ assert dev.decrypt(p[2]) == {'command': 'fxaccounts:profile_updated', 'version': 1}
+ assert push_server.done()
generated by cgit v1.2.3 (git 2.25.1) at 2024-05-19 14:02:25 +0000