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 def parallel(*fns):
49 from threading import Thread
50 threads = [ Thread(target=fn) for fn in fns ]
51 for t in threads: t.start()
52 for t in threads: t.join()
53
54 start_all()
55 server.wait_for_unit("nats.service")
56
57 with subtest("pub sub"):
58 parallel(
59 lambda: client1.succeed(nats_cmd("sub", "--count", "1", "${topic}")),
60 lambda: client2.succeed("sleep 2 && {}".format(nats_cmd("pub", "${topic}", "hello"))),
61 )
62 '';
63})