Removed a transaction.commit() that was unnecessary.
This commit is contained in:
7
examples/misc/script.py
Normal file
7
examples/misc/script.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
from packetserver.server import Server
|
||||||
|
from packetserver.common import PacketServerConnection
|
||||||
|
import logging
|
||||||
|
logging.basicConfig(level=logging.DEBUG)
|
||||||
|
s = Server('localhost', 8000, 'KQ4PEC')
|
||||||
|
s.start()
|
||||||
|
cm = s.app._engine._active_handler._handlers[1]._connection_map
|
||||||
51
examples/misc/test.py
Normal file
51
examples/misc/test.py
Normal file
@@ -0,0 +1,51 @@
|
|||||||
|
from packetserver.common import DummyPacketServerConnection, Request, Response, Message
|
||||||
|
from packetserver.server import TestServer
|
||||||
|
from packetserver.server.objects import Object
|
||||||
|
from packetserver.server.messages import Message as Mail
|
||||||
|
from packetserver.server.messages import Attachment
|
||||||
|
import time
|
||||||
|
import logging
|
||||||
|
import json
|
||||||
|
|
||||||
|
logging.basicConfig(level=logging.DEBUG)
|
||||||
|
|
||||||
|
server_callsign = "KQ4PEC"
|
||||||
|
client_callsign = 'KQ4PEC-7'
|
||||||
|
#client_callsign = "TEST1"
|
||||||
|
|
||||||
|
ts = TestServer(server_callsign, zeo=True)
|
||||||
|
ts.start()
|
||||||
|
|
||||||
|
time.sleep(1)
|
||||||
|
print("creating connection")
|
||||||
|
|
||||||
|
conn = DummyPacketServerConnection(client_callsign, server_callsign, incoming=True)
|
||||||
|
print(conn.remote_callsign)
|
||||||
|
print(conn.call_to)
|
||||||
|
print(conn.call_from)
|
||||||
|
conn.connected()
|
||||||
|
|
||||||
|
req = Request.blank()
|
||||||
|
|
||||||
|
req.set_var('fetch_attachments', 1)
|
||||||
|
req.path = "message"
|
||||||
|
|
||||||
|
#req.method=Request.Method.POST
|
||||||
|
#attach = [Attachment("test.txt", "Hello sir, I hope that this message finds you well. The other day..")]
|
||||||
|
#req.payload = Mail("Hi there from a test user!", "KQ4PEC", attachments=attach).to_dict()
|
||||||
|
|
||||||
|
|
||||||
|
#req.payload = Object(name="test.txt", data="hello there").to_dict()
|
||||||
|
|
||||||
|
print("sending request")
|
||||||
|
conn.data_received(0, bytearray(req.pack()))
|
||||||
|
#ts.send_test_data(conn, bytearray(req.pack()))
|
||||||
|
print("Waiting on response.")
|
||||||
|
time.sleep(.5)
|
||||||
|
ts.stop()
|
||||||
|
msg = conn.sent_data.unpack()
|
||||||
|
#print(f"msg: {msg}")
|
||||||
|
response = Response(Message.partial_unpack(msg))
|
||||||
|
#print(type(response.payload))
|
||||||
|
#print(f"Response: {response}: {response.payload}")
|
||||||
|
print(json.dumps(response.payload, indent=4))
|
||||||
51
examples/misc/testdb.py
Normal file
51
examples/misc/testdb.py
Normal file
@@ -0,0 +1,51 @@
|
|||||||
|
from packetserver.common import DummyPacketServerConnection, Request, Response, Message
|
||||||
|
from packetserver.server import TestServer
|
||||||
|
from packetserver.server.objects import Object
|
||||||
|
from packetserver.server.messages import Message as Mail
|
||||||
|
from packetserver.server.messages import Attachment
|
||||||
|
import time
|
||||||
|
import logging
|
||||||
|
import json
|
||||||
|
|
||||||
|
logging.basicConfig(level=logging.DEBUG)
|
||||||
|
|
||||||
|
server_callsign = "KQ4PEC"
|
||||||
|
client_callsign = 'KQ4PEC-7'
|
||||||
|
#client_callsign = "TEST1"
|
||||||
|
|
||||||
|
ts = TestServer(server_callsign, zeo=True)
|
||||||
|
ts.start()
|
||||||
|
|
||||||
|
time.sleep(1)
|
||||||
|
print("creating connection")
|
||||||
|
|
||||||
|
conn = DummyPacketServerConnection(client_callsign, server_callsign, incoming=True)
|
||||||
|
print(conn.remote_callsign)
|
||||||
|
print(conn.call_to)
|
||||||
|
print(conn.call_from)
|
||||||
|
conn.connected()
|
||||||
|
|
||||||
|
req = Request.blank()
|
||||||
|
|
||||||
|
req.set_var('fetch_attachments', 1)
|
||||||
|
req.path = "message"
|
||||||
|
|
||||||
|
#req.method=Request.Method.POST
|
||||||
|
#attach = [Attachment("test.txt", "Hello sir, I hope that this message finds you well. The other day..")]
|
||||||
|
#req.payload = Mail("Hi there from a test user!", "KQ4PEC", attachments=attach).to_dict()
|
||||||
|
|
||||||
|
|
||||||
|
#req.payload = Object(name="test.txt", data="hello there").to_dict()
|
||||||
|
|
||||||
|
print("sending request")
|
||||||
|
conn.data_received(0, bytearray(req.pack()))
|
||||||
|
#ts.send_test_data(conn, bytearray(req.pack()))
|
||||||
|
print("Waiting on response.")
|
||||||
|
time.sleep(.5)
|
||||||
|
ts.stop()
|
||||||
|
msg = conn.sent_data.unpack()
|
||||||
|
#print(f"msg: {msg}")
|
||||||
|
response = Response(Message.partial_unpack(msg))
|
||||||
|
#print(type(response.payload))
|
||||||
|
#print(f"Response: {response}: {response.payload}")
|
||||||
|
print(json.dumps(response.payload, indent=4))
|
||||||
Reference in New Issue
Block a user