logos-messaging/logos-messaging-python-bindings
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-python-bindings.git synced 2026-07-17 22:29:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove the temp test code

Browse Source
This commit is contained in:
Aya Hassan 2026-07-12 18:23:19 +02:00
parent 7fd83e40eb
commit 694833acfb
1 changed files with 0 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
waku/wrapper.py
View File

@ -520,69 +520,3 @@ class NodeWrapper:
return Err(cb_msg.decode("utf-8") if cb_msg else f"channel_close({channel_id}): callback failed (ret={cb_ret})") return Err(cb_msg.decode("utf-8") if cb_msg else f"channel_close({channel_id}): callback failed (ret={cb_ret})")
return Ok(cb_ret) return Ok(cb_ret)
# ---------------------------------------------------------------------------
# TEMPORARY manual smoke test for the channel wrappers.
# Run: python -m waku.wrapper (from the repo root, with lib/liblogosdelivery.so built)
# Remove before merging.
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import base64
import socket
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("", 0))
return s.getsockname()[1]
def _show(label, result):
tag = "OK " if result.is_ok() else "ERR"
value = result.ok_value if result.is_ok() else result.err()
print(f"[{tag}] {label}: {value!r}")
config = {
"logLevel": "INFO",
"listenAddress": "0.0.0.0",
"tcpPort": _free_port(),
"discv5UdpPort": _free_port(),
"restPort": _free_port(),
"restAddress": "0.0.0.0",
"clusterId": "198",
"relay": True,
"store": False,
"filter": False,
"lightpush": False,
"peerExchange": False,
"discv5Discovery": False,
}
CHANNEL_ID = "smoke-channel"
CONTENT_TOPIC = "/test/1/channel/proto"
SENDER_ID = "smoke-sender"
node_result = NodeWrapper.create_and_start(config)
if node_result.is_err():
print(f"[ERR] create_and_start: {node_result.err()!r}")
raise SystemExit(1)
node = node_result.ok_value
print("[OK ] node created and started")
try:
# RC01: create, then duplicate create rejected.
_show("channel_create", node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID))
_show("channel_create (duplicate)", node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID))
# RC02: send on unknown channel.
payload = base64.b64encode(b"hello channel").decode("ascii")
_show("channel_send (unknown id)", node.channel_send("no-such-channel", {"payload": payload, "ephemeral": False}))
# RC04: empty payload rejected; then a valid send returns a handle.
_show("channel_send (empty payload)", node.channel_send(CHANNEL_ID, {"payload": "", "ephemeral": False}))
_show("channel_send", node.channel_send(CHANNEL_ID, {"payload": payload, "ephemeral": False}))
# RC03: close, then close unknown rejected.
_show("channel_close", node.channel_close(CHANNEL_ID))
_show("channel_close (unknown id)", node.channel_close(CHANNEL_ID))
finally:
_show("stop_and_destroy", node.stop_and_destroy())