""".#test.py main module."""
|
import time
|
import httpx
|
from whereismysock import all
|
|
import book
|
import protocol
|
|
test = True
|
|
HOST: str = 'sytev070'
|
if (test):
|
PORT: str = '9005'
|
else:
|
PORT: str = '9000'
|
|
USER: str = 'test'
|
PASS: str = 'none'
|
|
|
def logAllIncomingMessages():
|
"""Test function to print websocket."""
|
messages = all(f"ws://{HOST}:{PORT}/information")
|
while True:
|
for message in next(messages):
|
print(message)
|
time.sleep(0.1)
|
|
|
def printRecoveryLog():
|
"""Test function to print recovery log."""
|
so_far_today = httpx.get(f"http://{HOST}:{PORT}/recover").json()
|
for message in so_far_today:
|
print(message)
|
|
|
def recoverBook(bs: {str, book.Book}):
|
"""Recover book from completely lost state."""
|
so_far_today = httpx.get(f"http://{HOST}:{PORT}/recover").json()
|
for message in so_far_today:
|
protocol.handleMessage(bs, message)
|
|
def testRecoverBook(bs: {str, book.Book}):
|
f = open("data.test", "r");
|
so_far_today = f.read();
|
for message in so_far_today:
|
protocol.handleMessage(bs, message)
|
|
|
message = {"type": "ADD", "product": "F_THS_APP0119T2230", "price": 99.0,
|
"side": "BUY", "volume": 1}
|
|
# print(json.dumps(message))
|
print(protocol.send(message))
|
# printRecoveryLog()
|
# logAllIncomingMessages()
|
# print(getInformation())
|
# wstest()
|
|
# bs = {}
|
# recoverBook(bs)
|
# testRecoverBook(bs)
|
# print(list(bs))
|
# for i in bs:
|
# print(bs[i].product)
|
# bs[i].printBook()
|
# print()
|
# if (len(bs[i].askSide.levels) > 0):
|
# print(bs[i].askSide.levels[0])
|
# if (len(bs[i].bidSide.levels)):
|
# print(bs[i].bidSide.levels[0])
|
# print()
|
# book.testBook()
|
