Wait for verified Merkle profiles as well as BGP reachability

tests/stream_netns/node.c

@@ -154,7 +154,18 @@ static void statistics(struct lab *lab, uint64_t id, int loads) {
             printf("%s\"%" PRIu64 "\"", separator++ ? "," : "", node->node_id);
         }
     }
-    printf("],\"streams\":["); separator = 0;
+    printf("],\"profile_status\":{"); separator = 0;
+    if (group) {
+        struct stream_member_state state;
+        printf("\"%" PRIu64 "\":%d", lab->inst->node_id,
+               stream_member_load(lab->inst, lab->group, lab->inst->node_id, &state, NULL)); separator = 1;
+        for (struct ll_entry *e = group->nodes->head; e; e = e->next) {
+            struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e;
+            printf("%s\"%" PRIu64 "\":%d", separator++ ? "," : "", node->node_id,
+                   stream_member_load(lab->inst, lab->group, node->node_id, &state, NULL));
+        }
+    }
+    printf("},\"streams\":["); separator = 0;
     for (unsigned i = 0; i < LAB_STREAMS; i++) {
         struct lab_stream *stream = &lab->streams[i];
         if (!stream->source) continue;

tests/stream_netns/scenario.py

@@ -130,7 +130,9 @@ class Scenario:
             link = owner.rpc("invite", timeout=15)["link"]
             self.agents[node.name].rpc("join", link=link)
         identities = {agent.identity for agent in self.agents.values()}
-        self.wait("join-merkle-bgp", lambda s: all(set(n["online"]) == identities for n in s.values()), 120)
+        self.wait("join-merkle-bgp", lambda s: all(set(n["online"]) == identities and
+                  set(n["profile_status"]) == identities and all(rc == 0 for rc in n["profile_status"].values())
+                  for n in s.values()), 120)
         source = self.agents[self.args.publisher]
         source_id = source.identity
         source.rpc("publish", seed=self.args.seed)