Coverage for src/meshadmin/cli/tests/test_host.py: 100%

1import httpx 

2import pytest 

3from typer.testing import CliRunner 

4 

5from meshadmin.cli.main import app 

6 

7runner = CliRunner() 

8 

9 

10@pytest.fixture 

11def mock_host_access_token(mocker): 

12 return mocker.patch( 

13 "meshadmin.cli.commands.host.get_access_token", return_value="fake-token" 

14 ) 

15 

16 

17@pytest.fixture 

18def mock_host_context_config(mocker): 

19 return mocker.patch( 

20 "meshadmin.cli.commands.host.get_context_config", 

21 return_value={"endpoint": "http://testserver"}, 

22 ) 

23 

24 

25@pytest.fixture 

26def mock_download(mocker): 

27 return mocker.patch("meshadmin.cli.commands.host.download") 

28 

29 

30@pytest.fixture 

31def mock_host_get_config_from_mesh(mocker): 

32 return mocker.patch( 

33 "meshadmin.cli.commands.host.get_config_from_mesh", 

34 return_value=( 

35 "pki: ca: test-ca-cert cert: test-host-cert key: host.key lighthouse: am_lighthouse: false hosts: - 10.0.0.1 static_host_map: '10.0.0.1': ['lighthouse.example.com:4242']", 

36 5, 

37 "false", 

38 ), 

39 ) 

40 

41 

42@pytest.fixture 

43def test_context(temp_config_dir): 

44 result = runner.invoke( 

45 app, 

46 [ 

47 "--config-path", 

48 str(temp_config_dir), 

49 "context", 

50 "create", 

51 "test-context-two", 

52 "--endpoint", 

53 "http://localhost:8001", 

54 ], 

55 ) 

56 return result 

57 

58 

59@pytest.fixture 

60def mock_enroll_response(mocker): 

61 return mocker.patch( 

62 "httpx.post", 

63 return_value=httpx.Response( 

64 status_code=200, 

65 request=httpx.Request("POST", "http://testserver/api/v1/enroll"), 

66 ), 

67 ) 

68 

69 

70def test_host_enrollment( 

71 mock_enroll_response, 

72 temp_config_dir, 

73 sample_context, 

74 mock_download, 

75 mock_host_get_config_from_mesh, 

76): 

77 result = runner.invoke( 

78 app, 

79 [ 

80 "--config-path", 

81 str(temp_config_dir), 

82 "host", 

83 "enroll", 

84 "test-enrollment-key", 

85 "--preferred-hostname", 

86 "test-host", 

87 "--public-ip", 

88 "192.168.1.100", 

89 ], 

90 ) 

91 assert result.exit_code == 0 

92 network_dir = temp_config_dir / "networks" / "test-context" 

93 auth_key_path = temp_config_dir / "auth.key" 

94 public_key_path = network_dir / "host.pub" 

95 private_key_path = network_dir / "host.key" 

96 config_path = network_dir / "config.yaml" 

97 assert network_dir.exists() 

98 assert auth_key_path.exists() 

99 assert public_key_path.exists() 

100 assert private_key_path.exists() 

101 assert config_path.exists() 

102 mock_download.assert_called() 

103 mock_host_get_config_from_mesh.assert_called() 

104 assert "enrollment finished" in result.stdout 

105 result = runner.invoke( 

106 app, 

107 [ 

108 "--config-path", 

109 str(temp_config_dir), 

110 "host", 

111 "enroll", 

112 "test-enrollment-key", 

113 ], 

114 ) 

115 assert result.exit_code == 0 

116 assert "private and public nebula key already exists" in result.stdout 

117 

118 

119def test_host_enrollment_shared_auth_key( 

120 mock_enroll_response, 

121 temp_config_dir, 

122 sample_context, 

123 test_context, 

124 mock_download, 

125 mock_host_get_config_from_mesh, 

126): 

127 result = runner.invoke( 

128 app, 

129 [ 

130 "--config-path", 

131 str(temp_config_dir), 

132 "--context", 

133 "test-context", 

134 "host", 

135 "enroll", 

136 "test-key", 

137 ], 

138 ) 

139 assert result.exit_code == 0 

140 assert "enrollment finished" in result.stdout 

141 auth_key_path = temp_config_dir / "auth.key" 

142 original_auth_key = auth_key_path.read_text() 

143 result = runner.invoke( 

144 app, 

145 [ 

146 "--config-path", 

147 str(temp_config_dir), 

148 "--context", 

149 "test-context-two", 

150 "host", 

151 "enroll", 

152 "test-key", 

153 ], 

154 ) 

155 assert result.exit_code == 0 

156 assert "enrollment finished" in result.stdout 

157 assert auth_key_path.read_text() == original_auth_key 

158 

159 

160def test_delete_host_success(mocker, mock_host_access_token, mock_host_context_config): 

161 mock_response = httpx.Response( 

162 status_code=200, 

163 json={"message": "Host test-host deleted"}, 

164 request=httpx.Request("DELETE", "http://testserver/api/v1/hosts/test-host"), 

165 ) 

166 mock_delete = mocker.patch("httpx.delete", return_value=mock_response) 

167 result = runner.invoke(app, ["host", "delete", "test-host"]) 

168 assert result.exit_code == 0 

169 mock_delete.assert_called_once_with( 

170 "http://testserver/api/v1/hosts/test-host", 

171 headers={"Authorization": "Bearer fake-token"}, 

172 ) 

173 assert "deleted" in result.stdout.lower() 

174 

175 

176def test_delete_host_auth_failure(mock_host_access_token, mock_host_context_config): 

177 mock_host_access_token.side_effect = Exception("Auth failed") 

178 result = runner.invoke(app, ["host", "delete", "test-host"]) 

179 assert result.exit_code == 1 

180 assert "failed to get access token" in result.stdout