pyrox.dev
1import ./make-test-python.nix ({ pkgs, lib, ...} :
2
3let
4 addrShared = "192.168.0.1";
5 addrHostA = "192.168.0.10";
6 addrHostB = "192.168.0.11";
7
8 mkUcarpHost = addr: { config, pkgs, lib, ... }: {
9 networking.interfaces.eth1.ipv4.addresses = lib.mkForce [
10 { address = addr; prefixLength = 24; }
11 ];
12
13 networking.ucarp = {
14 enable = true;
15 interface = "eth1";
16 srcIp = addr;
17 vhId = 1;
18 passwordFile = "${pkgs.writeText "ucarp-pass" "secure"}";
19 addr = addrShared;
20 upscript = pkgs.writeScript "upscript" ''
21 #!/bin/sh
22 ${pkgs.iproute2}/bin/ip addr add "$2"/24 dev "$1"
23 '';
24 downscript = pkgs.writeScript "downscript" ''
25 #!/bin/sh
26 ${pkgs.iproute2}/bin/ip addr del "$2"/24 dev "$1"
27 '';
28 };
29 };
30in {
31 name = "ucarp";
32 meta.maintainers = with lib.maintainers; [ oxzi ];
33
34 nodes = {
35 hostA = mkUcarpHost addrHostA;
36 hostB = mkUcarpHost addrHostB;
37 };
38
39 testScript = ''
40 def is_master(host):
41 ipOutput = host.succeed("ip addr show dev eth1")
42 return "inet ${addrShared}/24" in ipOutput
43
44
45 start_all()
46
47 # First, let both hosts start and let a master node be selected
48 for host, peer in [(hostA, "${addrHostB}"), (hostB, "${addrHostA}")]:
49 host.wait_for_unit("ucarp.service")
50 host.succeed(f"ping -c 1 {peer}")
51
52 hostA.sleep(5)
53
54 hostA_master, hostB_master = is_master(hostA), is_master(hostB)
55 assert hostA_master != hostB_master, "only one master node is allowed"
56
57 master_host = hostA if hostA_master else hostB
58 backup_host = hostB if hostA_master else hostA
59
60 # Let's crash the master host and let the backup take over
61 master_host.crash()
62
63 backup_host.sleep(5)
64 assert is_master(backup_host), "backup did not take over"
65 '';
66})