pyrox.dev
1let
2
3 port = 4222;
4 username = "client";
5 password = "password";
6 topic = "foo.bar";
7
8in import ./make-test-python.nix ({ pkgs, lib, ... }: {
9 name = "nats";
10 meta = with pkgs.lib; { maintainers = with maintainers; [ c0deaddict ]; };
11
12 nodes = let
13 client = { pkgs, ... }: {
14 environment.systemPackages = with pkgs; [ natscli ];
15 };
16 in {
17 server = { pkgs, ... }: {
18 networking.firewall.allowedTCPPorts = [ port ];
19 services.nats = {
20 inherit port;
21 enable = true;
22 settings = {
23 authorization = {
24 users = [{
25 user = username;
26 inherit password;
27 }];
28 };
29 };
30 };
31 };
32
33 client1 = client;
34 client2 = client;
35 };
36
37 testScript = let file = "/tmp/msg";
38 in ''
39 def nats_cmd(*args):
40 return (
41 "nats "
42 "--server=nats://server:${toString port} "
43 "--user=${username} "
44 "--password=${password} "
45 "{}"
46 ).format(" ".join(args))
47
48 start_all()
49 server.wait_for_unit("nats.service")
50
51 client1.fail("test -f ${file}")
52
53 # Subscribe on topic on client1 and echo messages to file.
54 client1.execute("({} | tee ${file} &)".format(nats_cmd("sub", "--raw", "${topic}")))
55
56 # Give client1 some time to subscribe.
57 client1.execute("sleep 2")
58
59 # Publish message on client2.
60 client2.execute(nats_cmd("pub", "${topic}", "hello"))
61
62 # Check if message has been received.
63 client1.succeed("grep -q hello ${file}")
64 '';
65})