File: bw_nodes.py

package info (click to toggle)
bundlewrap 5.0.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 3,260 kB
  • sloc: python: 20,849; makefile: 2
file content (122 lines) | stat: -rw-r--r-- 3,441 bytes parent folder | download
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
from os.path import join

from bundlewrap.utils.testing import make_repo, run


def test_empty(tmpdir):
    make_repo(tmpdir)
    stdout, stderr, rcode = run("bw nodes", path=str(tmpdir))
    assert stdout == b""
    assert stderr == b""
    assert rcode == 0


def test_single(tmpdir):
    make_repo(tmpdir, nodes={"node1": {}})
    stdout, stderr, rcode = run("bw nodes", path=str(tmpdir))
    assert stdout == b"node1\n"
    assert stderr == b""
    assert rcode == 0

def test_nonexistent(tmpdir):
    make_repo(tmpdir, nodes={"node1": {}})
    stdout, stderr, rcode = run("bw nodes node2", path=str(tmpdir))
    assert b"Target string node2 does match neither bundle, nor group, node or lambda." in stderr
    assert rcode == 1

def test_hostname(tmpdir):
    make_repo(
        tmpdir,
        groups={"all": {'member_patterns': {r".*"}}},
        nodes={"node1": {'hostname': "node1.example.com"}},
    )
    stdout, stderr, rcode = run("BW_TABLE_STYLE=grep bw nodes all -a hostname | cut -f 2", path=str(tmpdir))
    assert stdout == b"node1.example.com\n"
    assert stderr == b""
    assert rcode == 0


def test_bundles(tmpdir):
    make_repo(
        tmpdir,
        bundles={
            "bundle1": {},
            "bundle2": {},
        },
        groups={"all": {'member_patterns': {r".*"}}},
        nodes={
            "node1": {'bundles': ["bundle1", "bundle2"]},
            "node2": {'bundles': ["bundle2"]},
        },
    )
    stdout, stderr, rcode = run("BW_TABLE_STYLE=grep bw nodes all -a bundles | grep node1 | cut -f 2", path=str(tmpdir))
    assert stdout.decode().strip().split("\n") == ["bundle1", "bundle2"]
    assert stderr == b""
    assert rcode == 0


def test_bundles_via_group(tmpdir):
    make_repo(
        tmpdir,
        bundles={
            "bundle1": {},
            "bundle2": {},
            "bundle3": {},
        },
        groups={
            "group1": {
                'bundles': {"bundle2"},
                'subgroups': {"group2"},
            },
            "group2": {
                'bundles': {"bundle3"},
            }
        },
        nodes={
            "node1": {
                'bundles': {"bundle1"},
                'groups': {"group2"},
            },
        },
    )
    stdout, stderr, rcode = run("BW_TABLE_STYLE=grep bw nodes node1 -a bundles | cut -f 2", path=str(tmpdir))
    assert stdout.decode().strip().split("\n") == ["bundle1", "bundle2", "bundle3"]
    assert stderr == b""
    assert rcode == 0


def test_all_attrs(tmpdir):
    make_repo(
        tmpdir,
        nodes={
            "node1": {
                'username': "potato",
            },
        },
    )
    stdout, stderr, rcode = run("bw nodes -a all -- node1", path=str(tmpdir))
    assert "potato" in stdout.decode()
    assert stderr == b""
    assert rcode == 0


def test_dynamic_attrs(tmpdir):
    make_repo(
        tmpdir,
        groups={"all": {'member_patterns': {r".*"}}},
        nodes={
            "node1": {},
            "node2": {},
            "node3": {},
        },
    )
    with open(join(tmpdir, "nodes.py"), 'a') as f:
        f.write("""
@node_attribute
def dynamic(node):
    return node.name + "DYNAMIC"
""")
    stdout, stderr, rcode = run("BW_TABLE_STYLE=grep bw nodes all -a dynamic | cut -f 2", path=str(tmpdir))
    assert stdout.decode().strip().split("\n") == ["node1DYNAMIC", "node2DYNAMIC", "node3DYNAMIC"]
    assert stderr == b""
    assert rcode == 0