test_smoke.py raw
1 """Smoke tests — relay serves static files, app loads, basic WebSocket."""
2
3 import json
4 import time
5 import urllib.request
6
7 import pytest
8
9 from conftest import RELAY_URL, WS_URL
10
11
12 class TestRelayHTTP:
13 """Verify the relay serves static files and NIP-11."""
14
15 def test_index_html(self, relay):
16 resp = urllib.request.urlopen(f"{relay['url']}/")
17 html = resp.read().decode()
18 assert "<!DOCTYPE html>" in html
19 assert 'id="app-root"' in html
20
21 def test_entry_mjs(self, relay):
22 resp = urllib.request.urlopen(f"{relay['url']}/$entry.mjs")
23 assert resp.status == 200
24 body = resp.read().decode()
25 assert "$start" in body or "import" in body
26
27 def test_style_css(self, relay):
28 resp = urllib.request.urlopen(f"{relay['url']}/style.css")
29 assert resp.status == 200
30 body = resp.read().decode()
31 assert "--bg" in body
32
33 def test_sw_register(self, relay):
34 resp = urllib.request.urlopen(f"{relay['url']}/sw-register.js")
35 assert resp.status == 200
36
37 def test_sw_entry(self, relay):
38 resp = urllib.request.urlopen(f"{relay['url']}/$sw/$entry.mjs")
39 assert resp.status == 200
40
41 def test_nip11(self, relay):
42 req = urllib.request.Request(
43 relay["url"],
44 headers={"Accept": "application/nostr+json"},
45 )
46 resp = urllib.request.urlopen(req)
47 data = json.loads(resp.read())
48 assert data["name"] == "smesh"
49 assert 1 in data["supported_nips"]
50
51 def test_spa_fallback(self, relay):
52 """Unknown paths serve index.html (SPA fallback) instead of 404."""
53 resp = urllib.request.urlopen(f"{relay['url']}/nonexistent")
54 body = resp.read().decode()
55 assert "app-root" in body, "SPA fallback should serve index.html"
56
57
58 class TestRelayWebSocket:
59 """Basic WebSocket connectivity (REQ → EOSE)."""
60
61 def test_req_eose(self, relay):
62 """Connect via WebSocket, send REQ, expect EOSE back."""
63 import socket
64 import hashlib
65 import base64
66 import os
67
68 host = relay["host"]
69 port = relay["port"]
70
71 sock = socket.create_connection((host, port), timeout=5)
72
73 # WebSocket handshake.
74 ws_key = base64.b64encode(os.urandom(16)).decode()
75 handshake = (
76 f"GET / HTTP/1.1\r\n"
77 f"Host: {host}:{port}\r\n"
78 f"Upgrade: websocket\r\n"
79 f"Connection: Upgrade\r\n"
80 f"Sec-WebSocket-Key: {ws_key}\r\n"
81 f"Sec-WebSocket-Version: 13\r\n"
82 f"\r\n"
83 )
84 sock.sendall(handshake.encode())
85
86 # Read upgrade response.
87 resp = b""
88 while b"\r\n\r\n" not in resp:
89 resp += sock.recv(4096)
90 assert b"101" in resp
91
92 # Send REQ frame.
93 req = json.dumps(["REQ", "smoke-test", {"limit": 1}]).encode()
94 _ws_send_text(sock, req)
95
96 # Read EOSE.
97 data = _ws_read_frame(sock)
98 msg = json.loads(data)
99 # Could be EVENT or EOSE — we want at least EOSE eventually.
100 while msg[0] != "EOSE":
101 data = _ws_read_frame(sock)
102 msg = json.loads(data)
103 assert msg[0] == "EOSE"
104 assert msg[1] == "smoke-test"
105
106 # Send CLOSE subscription before disconnecting.
107 close_msg = json.dumps(["CLOSE", "smoke-test"]).encode()
108 _ws_send_text(sock, close_msg)
109 time.sleep(0.1)
110 # Drop connection (no WS close frame — moxie netpoller bug with close frames).
111 sock.close()
112
113
114 def _ws_send_close(sock):
115 """Send a WebSocket close frame (client-masked)."""
116 mask = os.urandom(4)
117 # Close frame: FIN + opcode 0x8, masked, 2-byte status code 1000.
118 import struct
119 payload = struct.pack(">H", 1000)
120 header = bytearray([0x88, 0x80 | len(payload)])
121 header.extend(mask)
122 masked = bytearray(payload[i] ^ mask[i % 4] for i in range(len(payload)))
123 try:
124 sock.sendall(bytes(header) + bytes(masked))
125 except Exception:
126 pass
127
128
129 def _ws_send_text(sock, payload):
130 """Send a WebSocket text frame (client-masked)."""
131 import struct
132 mask = os.urandom(4)
133 header = bytearray()
134 header.append(0x81) # FIN + text
135 length = len(payload)
136 if length < 126:
137 header.append(0x80 | length)
138 elif length < 65536:
139 header.append(0x80 | 126)
140 header.extend(struct.pack(">H", length))
141 else:
142 header.append(0x80 | 127)
143 header.extend(struct.pack(">Q", length))
144 header.extend(mask)
145 masked = bytearray(payload[i] ^ mask[i % 4] for i in range(len(payload)))
146 sock.sendall(bytes(header) + bytes(masked))
147
148
149 def _ws_read_frame(sock):
150 """Read a single WebSocket text frame (unmasked from server)."""
151 import struct
152 hdr = _recv_exact(sock, 2)
153 length = hdr[1] & 0x7F
154 if length == 126:
155 length = struct.unpack(">H", _recv_exact(sock, 2))[0]
156 elif length == 127:
157 length = struct.unpack(">Q", _recv_exact(sock, 8))[0]
158 return _recv_exact(sock, length)
159
160
161 def _recv_exact(sock, n):
162 buf = bytearray()
163 while len(buf) < n:
164 chunk = sock.recv(n - len(buf))
165 if not chunk:
166 raise ConnectionError("socket closed")
167 buf.extend(chunk)
168 return bytes(buf)
169
170
171 import os
172