btest/broker: Add test using Python bindings and zeek -r

This came up in community.zeek.org as a use-case and there
were a few gaps and holes, so add a test showing minimally
the current behavior.

testing/btest/broker/python-bindings.zeek

@@ -0,0 +1,137 @@
+# @TEST-GROUP: broker
+#
+# @TEST-PORT: BROKER_PORT
+#
+# @TEST-DOC: biswa on community.zeek.org has been doing cool stuff with zeek -r, Python broker bindings and suspend_processing(). He ran into a number of issues around suspend_processing(), time management, etc, try to cover some here.
+#
+# @TEST-REQUIRES: python3 -V
+# @TEST-REQUIRES: TOPIC=/btest/connections python3 recv.py check
+#
+# @TEST-EXEC: TOPIC=/btest/connections btest-bg-run recv "python3 -u ../recv.py"
+# @TEST-EXEC: TOPIC=/btest/connections btest-bg-run send "zeek -f 'port 80' -b ../send.zeek -r $TRACES/wikipedia.trace"
+#
+# @TEST-EXEC: btest-bg-wait 10
+# @TEST-EXEC: TEST_DIFF_CANONIFIER= btest-diff recv/.stdout
+# @TEST-EXEC: TEST_DIFF_CANONIFIER= btest-diff send/.stdout
+
+@TEST-START-FILE send.zeek
+
+global new_conn_added: event(c: connection) &is_used;
+global conn_removed: event(c: connection) &is_used;
+
+global my_topic = getenv("TOPIC");
+
+global conn_events = 0;
+
+event zeek_init()
+	{
+	Broker::peer("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	Broker::subscribe(my_topic);
+	suspend_processing();
+	}
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	print network_time(), "peer lost";
+	terminate();
+	}
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	print network_time(), "peer added";
+	continue_processing();
+	}
+
+event new_connection(c: connection)
+	{
+	++conn_events;
+	print network_time(), "new_connection", conn_events, c$uid;
+	Broker::publish(my_topic, new_conn_added, c);
+	}
+
+event connection_state_remove(c: connection)
+	{
+	++conn_events;
+	print network_time(), "connection_state_remove", conn_events, c$uid;
+	Broker::publish(my_topic, conn_removed, c);
+	}
+
+event send_pcap_file_done(path: string)
+	{
+	print network_time(), "send_pcap_file_done";
+	Broker::publish(my_topic, Pcap::file_done, path);
+	}
+
+event Pcap::file_done(path: string)
+	{
+	# Done reading pcap, forward network_time() by
+	# 24 hours to expire timers. Send Pcap::file_done
+	# to Python in a schedule timer to initiate
+	# termination.
+	print network_time(), "Pcap::file_done";
+
+	schedule double_to_interval(24 * 3600 - 1) { send_pcap_file_done(path) };
+
+	set_network_time(network_time() + double_to_interval(24 * 3600));
+	}
+
+global events_from_python = 0;
+event echo_from_python(what: string, c: connection) &is_used
+	{
+	++events_from_python;
+	print network_time(), "from_python", events_from_python, what, c$uid, c$id;
+	}
+@TEST-END-FILE
+
+
+@TEST-START-FILE recv.py
+"""
+Python script subscribing to TOPIC
+"""
+import os
+import sys
+
+# Prep the PYTHONPATH for the build directory.
+broker_path = os.path.join(os.environ["BUILD"], "auxil", "broker", "python")
+sys.path.insert(0, broker_path)
+
+import broker
+
+# 1024/tcp
+broker_port = int(os.environ["BROKER_PORT"].split("/")[0])
+broker_topic = os.environ["TOPIC"]
+
+# We were able to import broker and parse the broker_port, should be good.
+if len(sys.argv) > 1 and sys.argv[1] == "check":
+    sys.exit(0)
+
+# Setup endpoint and connect to Zeek.
+with ( broker.Endpoint() as ep,
+     ep.make_subscriber(broker_topic) as sub,
+     ep.make_status_subscriber(True) as ss):
+
+    ep.listen("127.0.0.1", broker_port)
+
+    while True:
+        statuses = ss.poll()
+        for s in statuses:
+            if s.code() in (broker.SC.PeerLost, broker.SC.EndpointUnreachable):
+                print("peer lost, done")
+                exit(0)
+
+        # Busy poll for a message or later status
+        msg = sub.get(0.5)
+        if msg is None:
+            continue
+        (t, d) = msg
+        my_event = broker.zeek.Event(d)
+        conn = my_event.args()[0]
+        print("Received", t, my_event.name(), my_event.args()[0][0])
+
+        if my_event.name() == "Pcap::file_done":
+            print("Received Pcap::file_done")
+            break
+
+        other_event = broker.zeek.Event("echo_from_python", my_event.name(), conn)
+        ep.publish(broker_topic, other_event)
+@TEST-END-FILE