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 ), 

38 ) 

39 

40 

41@pytest.fixture 

42def test_context(temp_config_dir): 

43 result = runner.invoke( 

44 app, 

45 [ 

46 "--config-path", 

47 str(temp_config_dir), 

48 "context", 

49 "create", 

50 "test-context-two", 

51 "--endpoint", 

52 "http://localhost:8001", 

53 ], 

54 ) 

55 return result 

56 

57 

58@pytest.fixture 

59def mock_enroll_response(mocker): 

60 return mocker.patch( 

61 "httpx.post", 

62 return_value=httpx.Response( 

63 status_code=200, 

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

65 ), 

66 ) 

67 

68 

69def test_host_enrollment( 

70 mock_enroll_response, 

71 temp_config_dir, 

72 sample_context, 

73 mock_download, 

74 mock_host_get_config_from_mesh, 

75): 

76 result = runner.invoke( 

77 app, 

78 [ 

79 "--config-path", 

80 str(temp_config_dir), 

81 "host", 

82 "enroll", 

83 "test-enrollment-key", 

84 "--preferred-hostname", 

85 "test-host", 

86 "--public-ip", 

87 "192.168.1.100", 

88 ], 

89 ) 

90 assert result.exit_code == 0 

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

92 auth_key_path = temp_config_dir / "auth.key" 

93 public_key_path = network_dir / "host.pub" 

94 private_key_path = network_dir / "host.key" 

95 config_path = network_dir / "config.yaml" 

96 assert network_dir.exists() 

97 assert auth_key_path.exists() 

98 assert public_key_path.exists() 

99 assert private_key_path.exists() 

100 assert config_path.exists() 

101 mock_download.assert_called() 

102 mock_host_get_config_from_mesh.assert_called() 

103 assert "enrollment finished" in result.stdout 

104 result = runner.invoke( 

105 app, 

106 [ 

107 "--config-path", 

108 str(temp_config_dir), 

109 "host", 

110 "enroll", 

111 "test-enrollment-key", 

112 ], 

113 ) 

114 assert result.exit_code == 0 

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

116 

117 

118def test_host_enrollment_shared_auth_key( 

119 mock_enroll_response, 

120 temp_config_dir, 

121 sample_context, 

122 test_context, 

123 mock_download, 

124 mock_host_get_config_from_mesh, 

125): 

126 result = runner.invoke( 

127 app, 

128 [ 

129 "--config-path", 

130 str(temp_config_dir), 

131 "--context", 

132 "test-context", 

133 "host", 

134 "enroll", 

135 "test-key", 

136 ], 

137 ) 

138 assert result.exit_code == 0 

139 assert "enrollment finished" in result.stdout 

140 auth_key_path = temp_config_dir / "auth.key" 

141 original_auth_key = auth_key_path.read_text() 

142 result = runner.invoke( 

143 app, 

144 [ 

145 "--config-path", 

146 str(temp_config_dir), 

147 "--context", 

148 "test-context-two", 

149 "host", 

150 "enroll", 

151 "test-key", 

152 ], 

153 ) 

154 assert result.exit_code == 0 

155 assert "enrollment finished" in result.stdout 

156 assert auth_key_path.read_text() == original_auth_key 

157 

158 

159def test_delete_host_success(mocker, mock_host_access_token, mock_host_context_config): 

160 mock_response = httpx.Response( 

161 status_code=200, 

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

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

164 ) 

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

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

167 assert result.exit_code == 0 

168 mock_delete.assert_called_once_with( 

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

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

171 ) 

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

173 

174 

175def test_delete_host_auth_failure(mock_host_access_token, mock_host_context_config): 

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

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

178 assert result.exit_code == 1 

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