distributed/test_raft.py

配套源码，运行方法见同目录 README。返回实验总览。

import unittest
from raft_model import Cluster, Node, Entry


class RaftTests(unittest.TestCase):
    def test_election_requires_fixed_majority(self):
        c=Cluster(); c.partition([0,1],[2,3,4])
        self.assertFalse(c.elect(0)); self.assertTrue(c.elect(2))

    def test_vote_once_and_reject_stale_log(self):
        n=Node(0,log=[Entry(0,None),Entry(2,None)])
        self.assertFalse(n.vote(3,1,9,1)[1])
        self.assertTrue(n.vote(3,1,1,2)[1])
        self.assertFalse(n.vote(3,2,1,2)[1])

    def test_committed_prefix_survives_partition(self):
        c=Cluster();c.elect(0)
        self.assertEqual(c.submit(0,('a',1,'put','x',1)),(True,1))
        c.partition([0,1],[2,3,4])
        self.assertFalse(c.submit(0,('a',2,'put','x',99))[0])
        self.assertEqual(c.nodes[0].state['x'],1)
        self.assertTrue(c.elect(2))
        self.assertEqual(c.submit(2,('b',1,'put','x',2)),(True,2))
        c.heal();c.replicate(2)
        self.assertTrue(all(n.state=={'x':2} for n in c.nodes))
        self.assertEqual(c.nodes[0].role,'follower')
        self.assertTrue(all(n.log[1].command==('a',1,'put','x',1) for n in c.nodes))

    def test_retry_is_applied_once_across_leader_change(self):
        c=Cluster();c.elect(0)
        request=('c',1,'add','counter',3)
        c.submit(0,request)  # imagine reply was lost
        self.assertTrue(c.elect(1))
        self.assertEqual(c.submit(1,request),(True,3))
        self.assertTrue(all(n.state['counter']==3 for n in c.nodes))

    def test_restart_preserves_vote_log_and_rebuilds_state(self):
        c=Cluster();c.elect(0);c.submit(0,('c',1,'put','k',8))
        before=c.nodes[4];c.restart(4)
        self.assertEqual(c.nodes[4].term,before.term)
        self.assertEqual(c.nodes[4].voted_for,before.voted_for)
        self.assertEqual(c.nodes[4].state,{})
        c.replicate(0);self.assertEqual(c.nodes[4].state,{'k':8})

    def test_previous_term_not_committed_by_count_alone(self):
        c=Cluster();c.elect(0);c.partition([0,1],[2,3,4])
        c.submit(0,('c',1,'put','x',5))
        c.heal();self.assertTrue(c.elect(1))
        c.replicate(1)
        self.assertEqual(c.nodes[1].commit,0)
        self.assertTrue(c.submit(1,None)[0])
        self.assertEqual(c.nodes[1].state,{'x':5})

    def test_partitioned_leader_cannot_serve_new_read(self):
        c=Cluster();c.elect(0);c.submit(0,('c',1,'put','x',3))
        c.partition([0],[1,2,3,4])
        self.assertFalse(c.submit(0,('c',2,'get','x',None))[0])

    def test_stale_append_rejected(self):
        n=Node(0,term=8)
        self.assertEqual(n.receive(7,0,0,[Entry(7,None)],0),(8,False))

    def test_short_append_neither_deletes_nor_commits_unconfirmed_suffix(self):
        prefix = Entry(1, ('a', 1, 'put', 'x', 1))
        extra = Entry(2, ('a', 2, 'put', 'x', 99))
        n = Node(0, term=2, log=[Entry(0, None), prefix, extra])
        self.assertEqual(n.receive(3, 1, 1, [], 2), (3, True))
        self.assertEqual(n.log, [Entry(0, None), prefix, extra])
        self.assertEqual(n.commit, 1)
        self.assertEqual(n.state, {'x': 1})

    def test_conflicting_suffix_is_replaced_after_matching_prefix(self):
        prefix = Entry(1, ('a', 1, 'put', 'x', 1))
        stale = Entry(2, ('a', 2, 'put', 'x', 99))
        replacement = Entry(3, ('b', 1, 'put', 'x', 2))
        n = Node(0, term=2, log=[Entry(0, None), prefix, stale], commit=1)
        self.assertEqual(n.receive(3, 1, 1, [replacement], 2), (3, True))
        self.assertEqual(n.log, [Entry(0, None), prefix, replacement])
        self.assertEqual(n.state, {'x': 2})

    def test_all_restart_needs_current_term_barrier_then_restores_dedup(self):
        c = Cluster(3)
        self.assertTrue(c.elect(0))
        request = ('c', 1, 'add', 'counter', 3)
        self.assertEqual(c.submit(0, request), (True, 3))
        for ident in range(3):
            c.alive[ident] = False
        for ident in range(3):
            c.restart(ident)
        self.assertTrue(c.elect(1))
        c.replicate(1)
        self.assertTrue(all(n.commit == 0 and n.state == {} for n in c.nodes))
        self.assertEqual(c.submit(1, None), (True, None))
        self.assertTrue(all(n.state == {'counter': 3} for n in c.nodes))
        self.assertEqual(c.submit(1, request), (True, 3))
        self.assertTrue(all(n.state == {'counter': 3} for n in c.nodes))

    def test_unknown_submit_can_commit_later_without_a_new_request(self):
        c = Cluster()
        self.assertTrue(c.elect(0))
        c.partition([0, 1], [2, 3, 4])
        request = ('c', 1, 'add', 'counter', 3)
        self.assertEqual(c.submit(0, request), (False, None))
        self.assertEqual(c.nodes[0].state, {})
        c.heal()
        self.assertTrue(c.replicate(0))
        self.assertTrue(all(n.state == {'counter': 3} for n in c.nodes))
        self.assertEqual(c.submit(0, request), (True, 3))

    def test_retry_of_read_returns_original_result_after_later_write(self):
        c = Cluster(3)
        c.elect(0)
        c.submit(0, ('writer', 1, 'put', 'x', 1))
        read = ('reader', 1, 'get', 'x', None)
        self.assertEqual(c.submit(0, read), (True, 1))
        c.submit(0, ('writer', 2, 'put', 'x', 2))
        self.assertTrue(c.elect(1))
        self.assertEqual(c.submit(1, read), (True, 1))
        self.assertEqual(c.submit(1, ('reader', 2, 'get', 'x', None)), (True, 2))

    def test_submitted_value_cannot_mutate_log_or_replica_state(self):
        c = Cluster(3)
        c.elect(0)
        value = {'items': [1, 2]}
        self.assertTrue(c.submit(0, ('c', 1, 'put', 'x', value))[0])
        value['items'].append(99)
        for n in c.nodes:
            self.assertEqual(n.state['x'], {'items': [1, 2]})
            self.assertEqual(n.log[1].command[-1], {'items': [1, 2]})

    def test_returned_value_cannot_mutate_state_or_cached_retry_result(self):
        c = Cluster(3)
        c.elect(0)
        request = ('c', 1, 'put', 'x', {'items': [1]})
        ok, result = c.submit(0, request)
        self.assertTrue(ok)
        result['items'].append(99)
        self.assertEqual(c.submit(0, request), (True, {'items': [1]}))
        self.assertTrue(all(n.state['x'] == {'items': [1]} for n in c.nodes))

    def test_restart_does_not_allow_a_second_vote_in_the_same_term(self):
        c = Cluster(3)
        self.assertEqual(c.nodes[0].vote(4, 1, 0, 0), (4, True))
        c.restart(0)
        self.assertEqual(c.nodes[0].vote(4, 2, 0, 0), (4, False))


if __name__=='__main__': unittest.main()