diff --git a/tests/stream_netns/node.c b/tests/stream_netns/node.c index 40fcd1f1..cc437bed 100644 --- a/tests/stream_netns/node.c +++ b/tests/stream_netns/node.c @@ -208,7 +208,7 @@ static void statistics(struct lab *lab, uint64_t id, int loads) { if (stream_buffer_read(local.buffer, last, sub, &wire, &length, &actual) == STREAM_OK) printf(",\"source_timestamp_ms\":%u", (unsigned)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2)); struct stream_metric_snapshot metric; - if (stream_local_metrics(lab->inst, &info.descriptor, sub, now_ns() / UINT64_C(1000000), &metric) == STREAM_OK) + if (stream_local_metrics(lab->inst, &info.descriptor, sub, get_time_tb() / 10, &metric) == STREAM_OK) printf(",\"lag_ms\":%" PRIu64 ",\"source_rate\":%" PRIu64 ",\"receive_rate\":%" PRIu64 ",\"late_blocks\":%" PRIu64 ",\"stops\":%" PRIu64 ",\"stopped\":%d", metric.lag_ms, metric.source_rate_bytes_sec, metric.receive_rate_bytes_sec, diff --git a/tests/stream_netns/scenario.py b/tests/stream_netns/scenario.py index 66e6a97f..bca2f4a8 100644 --- a/tests/stream_netns/scenario.py +++ b/tests/stream_netns/scenario.py @@ -158,11 +158,13 @@ class Scenario: other = link.b if node.country == link.a else link.a self.network.netem(node, other, rtt=300, loss=5) self.duration("wan-degraded", self.args.fault_duration) + before_recovery = {name: self.stream(self.latest[name], source_id)["received"] for name in names} self.network.bandwidth(link, link.mbps) for node in NODES: if node.country in {link.a, link.b}: self.network.netem(node, link.b if node.country == link.a else link.a) - self.wait("wan-recovery", lambda s: self.ready(s, source_id, names), 90) + self.wait("wan-recovery", lambda s: self.ready(s, source_id, names) and + all(self.stream(s[name], source_id)["received"] > before_recovery[name] for name in names), 90) self.duration("recovered", 10) # Узлы без подписки должны прекратить обработку этого экземпляра. excluded = [name for name in ("n02", "n13") if name != self.args.publisher]