Docker
70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Test: subscribe WITHOUT ws connection - does QMT push continue?"""
|
|
import sys, io, os, json, time, urllib.request
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src"))
|
|
if sys.stdout is None:
|
|
sys.stdout = io.StringIO()
|
|
|
|
import bridge_util, bridge_http_server, bridge_subscription
|
|
|
|
class FakeCtx:
|
|
def __init__(self):
|
|
self.subs = {}
|
|
self._next = 1
|
|
self.callback = None
|
|
self.push_calls = [] # record every QMT callback invocation
|
|
def subscribe_whole_quote(self, codes, callback=None):
|
|
h = self._next
|
|
self._next += 1
|
|
self.subs[h] = list(codes)
|
|
self.callback = callback
|
|
return h
|
|
def unsubscribe_quote(self, h):
|
|
self.subs.pop(h, None)
|
|
|
|
fake = FakeCtx()
|
|
bridge_util.CTX = fake
|
|
bridge_http_server.PORT = 18692
|
|
bridge_http_server.TOKEN = ""
|
|
bridge_http_server.start_server()
|
|
time.sleep(0.5)
|
|
|
|
# 1. subscribe whole (NO ws connection established)
|
|
req = urllib.request.Request(
|
|
"http://127.0.0.1:18692/data/subscribe",
|
|
data=json.dumps({"type": "whole", "codes": ["SH", "SZ"]}).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as r:
|
|
print("subscribe:", r.read().decode())
|
|
|
|
# 2. simulate QMT pushes (no ws connected)
|
|
print("QMT subs before pushes:", fake.subs)
|
|
print("WS clients count: (none connected)")
|
|
for i in range(3):
|
|
fake.callback({"600000.SH": {"lastPrice": 10.0 + i, "volume": 100}})
|
|
print("QMT push %d fired (callback invoked)" % (i + 1))
|
|
|
|
# 3. verify subscription still active (QMT sub not torn down)
|
|
print("QMT subs after pushes:", fake.subs)
|
|
print("sub active:", bool(fake.subs))
|
|
|
|
# 4. unsubscribe
|
|
req = urllib.request.Request(
|
|
"http://127.0.0.1:18692/data/unsubscribe",
|
|
data=json.dumps({"sub_id": 1}).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as r:
|
|
print("unsubscribe:", r.read().decode())
|
|
except Exception as e:
|
|
# find actual sub_id from state
|
|
import bridge_subscription as bs
|
|
print("unsub err:", e, "active subs:", bs.get_subscriptions())
|
|
|
|
print("QMT subs after unsubscribe:", fake.subs)
|
|
bridge_http_server.stop_server()
|
|
print("=== DONE ===")
|