hive/tests/test_gossip.py at d1f9115aa105dcc92f682d103795540d73ebb3c5 · DJLougen/hive · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
"""Tests for cross-node gossip."""

from __future__ import annotations

from hive.gossip import GossipProtocol
from hive.rust_brain import RustBrain


def test_gossip_publish_and_receive():
    brain = RustBrain()
    gossip = GossipProtocol(brain, peers=[])
    gossip.publish({"key": "k1", "value": "v1", "trust": 0.9})
    # No crash on publish


def test_gossip_receive_applies_events():
    brain = RustBrain()
    gossip = GossipProtocol(brain, peers=[])
    events = [
        {"key": "k1", "value": "v1", "trust": 0.9, "tags": ["a"]},
        {"key": "k2", "value": "v2", "trust": 0.8},
    ]
    applied = gossip.receive(events)
    assert applied == 2
    assert brain.recall("k1") == "v1"
    assert brain.recall("k2") == "v2"


def test_gossip_start_stop():
    brain = RustBrain()
    gossip = GossipProtocol(brain, peers=[], interval=0.1)
    gossip.start()
    gossip.stop()
    # No crash


def test_gossip_rejects_stale_hlc_overwrite():
    brain = RustBrain()
    gossip = GossipProtocol(brain, peers=[])
    brain.remember("shared", "FRESH", hlc=(2000, 0, "local"))

    applied = gossip.receive(
        [{"key": "shared", "value": "STALE", "hlc": [1000, 0, "remote"]}]
    )
    assert applied == 0
    assert brain.recall("shared") == "FRESH"


def test_gossip_applies_newer_hlc():
    brain = RustBrain()
    gossip = GossipProtocol(brain, peers=[])
    brain.remember("shared", "OLD", hlc=(1000, 0, "local"))

    applied = gossip.receive(
        [{"key": "shared", "value": "NEW", "hlc": [2000, 0, "remote"]}]
    )
    assert applied == 1
    assert brain.recall("shared") == "NEW"


def test_gossip_skips_missing_hlc_on_existing_key():
    brain = RustBrain()
    gossip = GossipProtocol(brain, peers=[])
    brain.remember("k", "local")

    applied = gossip.receive([{"key": "k", "value": "remote"}])
    assert applied == 0
    assert brain.recall("k") == "local"