personal memory agent
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

pairing: add integration round-trip test + fix any remaining issues

+124
+124
tests/test_pairing_integration.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + from cryptography.hazmat.primitives import serialization 7 + from cryptography.hazmat.primitives.asymmetric import ed25519 8 + 9 + from convey import create_app 10 + from think.pairing.devices import load_devices 11 + 12 + 13 + def _owner_login(client) -> None: 14 + with client.session_transaction() as session: 15 + session["logged_in"] = True 16 + session.permanent = True 17 + 18 + 19 + def _public_key() -> str: 20 + return ( 21 + ed25519.Ed25519PrivateKey.generate() 22 + .public_key() 23 + .public_bytes( 24 + encoding=serialization.Encoding.OpenSSH, 25 + format=serialization.PublicFormat.OpenSSH, 26 + ) 27 + .decode("utf-8") 28 + ) 29 + 30 + 31 + def test_pairing_round_trip(journal_copy): 32 + app = create_app(str(journal_copy)) 33 + app.config["TESTING"] = True 34 + 35 + owner_client = app.test_client() 36 + _owner_login(owner_client) 37 + bearer_client = app.test_client() 38 + public_key = _public_key() 39 + 40 + create_response = owner_client.post("/api/pairing/create", json={}) 41 + create_body = create_response.get_json() 42 + token = create_body["token"] 43 + 44 + confirm_payload = { 45 + "token": token, 46 + "public_key": public_key, 47 + "device_name": "Round Trip Phone", 48 + "platform": "ios", 49 + "bundle_id": "org.solpbc.solstone-swift", 50 + "app_version": "0.1.0", 51 + } 52 + confirm_response = owner_client.post("/api/pairing/confirm", json=confirm_payload) 53 + confirm_body = confirm_response.get_json() 54 + session_key = confirm_body["session_key"] 55 + device_id = confirm_body["device_id"] 56 + 57 + assert create_response.status_code == 200 58 + assert confirm_response.status_code == 200 59 + assert session_key.startswith("dsk_") 60 + assert device_id.startswith("dev_") 61 + 62 + second_confirm = owner_client.post("/api/pairing/confirm", json=confirm_payload) 63 + assert second_confirm.status_code == 410 64 + assert second_confirm.get_json()["reason"] == "token_consumed" 65 + 66 + wrong_bearer = bearer_client.get( 67 + "/api/pairing/devices", 68 + headers={ 69 + "Authorization": "Bearer dsk_wrong", 70 + "X-Forwarded-For": "1.2.3.4", 71 + }, 72 + ) 73 + assert wrong_bearer.status_code == 401 74 + assert wrong_bearer.get_json()["reason"] == "auth_required" 75 + 76 + bearer_list = bearer_client.get( 77 + "/api/pairing/devices", 78 + headers={ 79 + "Authorization": f"Bearer {session_key}", 80 + "X-Forwarded-For": "1.2.3.4", 81 + }, 82 + ) 83 + assert bearer_list.status_code == 200 84 + assert bearer_list.get_json()["devices"] == [ 85 + { 86 + "id": device_id, 87 + "name": "Round Trip Phone", 88 + "platform": "ios", 89 + "paired_at": load_devices()[0]["paired_at"], 90 + "last_seen_at": None, 91 + } 92 + ] 93 + 94 + heartbeat = bearer_client.post( 95 + "/api/pairing/heartbeat", 96 + headers={ 97 + "Authorization": f"Bearer {session_key}", 98 + "X-Forwarded-For": "1.2.3.4", 99 + }, 100 + json={}, 101 + ) 102 + assert heartbeat.status_code == 200 103 + assert heartbeat.get_json() == {"ok": True} 104 + 105 + owner_list = owner_client.get("/api/pairing/devices") 106 + devices = owner_list.get_json()["devices"] 107 + assert owner_list.status_code == 200 108 + assert devices[0]["id"] == device_id 109 + assert devices[0]["last_seen_at"] is not None 110 + 111 + unpair = owner_client.delete(f"/api/pairing/devices/{device_id}") 112 + assert unpair.status_code == 200 113 + assert unpair.get_json() == {"unpaired": True} 114 + assert load_devices() == [] 115 + 116 + stale_bearer = bearer_client.get( 117 + "/api/pairing/devices", 118 + headers={ 119 + "Authorization": f"Bearer {session_key}", 120 + "X-Forwarded-For": "1.2.3.4", 121 + }, 122 + ) 123 + assert stale_bearer.status_code == 401 124 + assert stale_bearer.get_json()["reason"] == "auth_required"