1import json 

2import pytest 

3from contextlib import nullcontext as does_not_raise 

4from pdugateway import app, get_pdu_or_fail, populate_static_pdus 

5from pdu import PDU 

6from unittest import mock 

7 

8 

9@pytest.fixture 

10def client(): 

11 app.config['TESTING'] = True 

12 

13 with app.test_client() as client: 

14 yield client 

15 

16 

17def test_routes(client): 

18 rv = client.get('/') 

19 assert set([r for r in rv.json.get('routes', {})]) == set(["/v1/pdus", "/pdu_gateway", "/"]) 

20 

21 

22def test_pdu_gateway(client): 

23 rv = client.get('/pdu_gateway') 

24 ret = rv.json 

25 

26 assert ret['version'] == 1 

27 assert "dummy" in ret['supported_pdus'] 

28 

29 

30def test_get_pdu_or_fail__non_existing_pdu(): 

31 with pytest.raises(ValueError): 

32 get_pdu_or_fail("invalid") 

33 

34 

35def test_non_json_post_request(client): 

36 for url in ['/v1/pdus', '/v1/pdus/MyPDU/ports/1/state']: 

37 ret = client.post(url, data="Hello") 

38 assert ret.status_code == 400 

39 

40 

41def test_full_integration_test(client): 

42 # Make sure all the state is gone 

43 client.delete('/v1/pdus') 

44 

45 # Check that we do not have any PDUs 

46 ret = client.get('/v1/pdus').json 

47 assert ret['pdus'] == [] 

48 

49 # Add a PDU 

50 rv = client.post('/v1/pdus', json={ 

51 "model": "dummy", 

52 "pdu_name": "MyPDU", 

53 "config": { 

54 "ports": ['P1', 'P2', 'P3'] 

55 } 

56 }) 

57 assert rv.status_code == 200 

58 

59 # Check that listing the ports 

60 ret = client.get('/v1/pdus/MyPDU/ports').json 

61 assert ret == { 

62 '0': {'label': 'P1', 'state': 'ON'}, 

63 '1': {'label': 'P2', 'state': 'ON'}, 

64 '2': {'label': 'P3', 'state': 'ON'} 

65 } 

66 

67 # Check that the PDU is now visible in the list of PDUs 

68 ret = client.get('/v1/pdus').json 

69 assert ret['pdus'] == ["MyPDU"] 

70 

71 # Read the port 

72 ret = client.get('/v1/pdus/MyPDU/ports/1/state') 

73 assert ret.data == b'ON\n' 

74 

75 # Change the port's state 

76 ret = client.post('/v1/pdus/MyPDU/ports/1/state', json={"state": "REBOOT"}) 

77 assert ret.status_code == 200 

78 assert ret.data == b'REBOOT\n' 

79 

80 # Try to set the state, but without specifying it 

81 ret = client.post('/v1/pdus/MyPDU/ports/1/state', json={}) 

82 assert ret.status_code == 400 

83 

84 # Try to set an invalid state 

85 ret = client.post('/v1/pdus/MyPDU/ports/1/state', json={"state": "INVALID"}) 

86 assert ret.status_code == 400 

87 

88 # Read back the state 

89 ret = client.get('/v1/pdus/MyPDU/ports/1/state') 

90 assert ret.data == b'REBOOT\n' 

91 

92 # Check that the change got reflected to the list of ports 

93 ret = client.get('/v1/pdus/MyPDU/ports').json 

94 assert ret == { 

95 '0': {'label': 'P1', 'state': 'ON'}, 

96 '1': {'label': 'P2', 'state': 'REBOOT'}, 

97 '2': {'label': 'P3', 'state': 'ON'} 

98 } 

99 

100 

101@pytest.mark.parametrize( 

102 "static_spec,calls,expectation", 

103 [ 

104 ("""[{"model": "dummy", "config":{"hostname": "10.42.0.2"}}]""", 

105 [], 

106 pytest.raises(ValueError)), 

107 ("""[{"pdu_name": "MyPDU", "model": "dummy"}]""", 

108 [], 

109 pytest.raises(ValueError)), 

110 ("""[{"pdu_name": "MyPDU", "config":{"hostname": "10.42.0.2"}}]""", 

111 [], 

112 pytest.raises(ValueError)), 

113 ("""[{"pdu_name": "MyPDU", "model": "dummy", "config":{"hostname": "10.42.0.2"}}]""", 

114 [ 

115 mock.call(model_name="dummy", 

116 pdu_name="MyPDU", 

117 config={"hostname": "10.42.0.2"}), 

118 ], 

119 does_not_raise()), 

120 ("""[{"pdu_name": "MyPDU", "model": "dummy", "config":{"hostname": "10.42.0.2"}}, {"pdu_name": "MyPDU2", 

121 "model": "cyberpower_pdu41004", "config":{ "hostname": "10.42.0.3"}}]""", 

122 [ 

123 mock.call(model_name="dummy", 

124 pdu_name="MyPDU", 

125 config={"hostname": "10.42.0.2"}), 

126 mock.call(model_name="cyberpower_pdu41004", 

127 pdu_name="MyPDU2", 

128 config={"hostname": "10.42.0.3"}) 

129 ], 

130 does_not_raise()), 

131 ("", 

132 [], 

133 pytest.raises(json.decoder.JSONDecodeError)), 

134 ], 

135) 

136def test_populate_static_pdus(monkeypatch, static_spec, calls, expectation): 

137 m = mock.Mock() 

138 with expectation: 

139 monkeypatch.setenv("PDUGATEWAY_STATIC_PDUS", 

140 static_spec) 

141 monkeypatch.setattr(PDU, "register", m) 

142 populate_static_pdus() 

143 if len(calls): 

144 m.assert_has_calls(calls) 

145 else: 

146 m.assert_not_called()