import socket
import asyncio
from test_framework.electrumutil import ElectrumTestFramework
from test_framework.electrumconnection import ElectrumConnection
from test_framework.portseed import electrum_rpc_port
class FeatureInvalidJson(ElectrumTestFramework):
async def run_test(self):
async def wait_for_rostrum():
cli = ElectrumConnection()
await cli.connect()
cli.disconnect()
await wait_for_rostrum()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect(("127.0.0.1", await electrum_rpc_port(n=0)))
s.sendall("invalid json\n".encode())
response = s.recv(1024)
print(response.decode())
assert response.decode().startswith(
'{"error":{"code":-32600,"message":"invalid JSON format"},"id":-1,"jsonrpc":"2.0"}'
)
if __name__ == "__main__":
asyncio.run(FeatureInvalidJson().main())