Coverage for src/meshadmin/cli/tests/test_integration.py: 23%

1import json 

2import os 

3import subprocess 

4import time 

5from pathlib import Path 

6 

7import pytest 

8import requests 

9from requests.exceptions import ConnectionError 

10 

11 

12@pytest.fixture(scope="session") 

13def docker_compose_runtime(): 

14 root_dir = Path(__file__).resolve().parent.parent.parent.parent.parent 

15 compose_file = root_dir / "docker-compose.test.yml" 

16 subprocess.run( 

17 ["docker", "compose", "-f", str(compose_file), "up", "--build", "-d"], 

18 check=True, 

19 ) 

20 

21 # Wait for server to be ready 

22 max_retries = 5 

23 retry_interval = 1 

24 for attempt in range(max_retries): 

25 try: 

26 response = requests.get("http://localhost:8000/api/v1/test") 

27 if response.status_code == 200: 

28 print(f"Server ready after {attempt + 1} attempts") 

29 break 

30 except ConnectionError: 

31 if attempt + 1 < max_retries: 

32 print(f"Waiting for server (attempt {attempt + 1}/{max_retries})...") 

33 time.sleep(retry_interval) 

34 else: 

35 subprocess.run( 

36 ["docker", "compose", "-f", str(compose_file), "logs"], 

37 check=True, 

38 ) 

39 raise Exception("Server failed to become ready") 

40 

41 yield 

42 

43 # Cleanup 

44 subprocess.run( 

45 ["docker", "compose", "-f", str(compose_file), "down", "-v"], check=True 

46 ) 

47 

48 

49@pytest.mark.runtime 

50def test_complete_workflow(docker_compose_runtime, temp_config_dir): 

51 # 1. Create a context 

52 result = subprocess.run( 

53 [ 

54 "meshadmin", 

55 "--config-path", 

56 str(temp_config_dir), 

57 "context", 

58 "create", 

59 "integration-test", 

60 "--endpoint", 

61 "http://localhost:8000", 

62 ], 

63 check=True, 

64 capture_output=True, 

65 text=True, 

66 ) 

67 assert "Set 'integration-test' as active context" in result.stdout 

68 context_dir = temp_config_dir / "contexts.yaml" 

69 assert context_dir.exists(), "Context file not created" 

70 

71 test_env = { 

72 "MESHADMIN_TEST_MODE": "true", 

73 "MESHADMIN_CONFIG_PATH": str(temp_config_dir), 

74 "MESH_CONTEXT": "integration-test", 

75 **os.environ, 

76 } 

77 

78 # 2. Create a network 

79 result = subprocess.run( 

80 [ 

81 "meshadmin", 

82 "network", 

83 "create", 

84 "test-network", 

85 "100.100.100.0/24", 

86 ], 

87 check=True, 

88 capture_output=True, 

89 text=True, 

90 env=test_env, 

91 ) 

92 assert "test-network" in result.stdout 

93 

94 # 3. Create a template 

95 result = subprocess.run( 

96 [ 

97 "meshadmin", 

98 "template", 

99 "create", 

100 "test-template", 

101 "test-network", 

102 "true", 

103 "false", 

104 "false", 

105 ], 

106 check=True, 

107 capture_output=True, 

108 text=True, 

109 env=test_env, 

110 ) 

111 assert "test-template" in result.stdout 

112 

113 # 4. Get the template token 

114 result = subprocess.run( 

115 [ 

116 "meshadmin", 

117 "template", 

118 "get-token", 

119 "test-template", 

120 ], 

121 check=True, 

122 capture_output=True, 

123 text=True, 

124 env=test_env, 

125 ) 

126 token = json.loads(result.stdout.strip())["token"] 

127 

128 # 5. Enroll a host 

129 result = subprocess.run( 

130 [ 

131 "meshadmin", 

132 "host", 

133 "enroll", 

134 token, 

135 ], 

136 check=True, 

137 capture_output=True, 

138 text=True, 

139 env=test_env, 

140 ) 

141 assert "enrollment finished" in result.stdout 

142 network_dir = temp_config_dir / "networks" / "integration-test" 

143 assert network_dir.exists(), "Network directory not created" 

144 private_key_path = network_dir / "host.key" 

145 public_key_path = network_dir / "host.pub" 

146 assert private_key_path.exists(), "Private key not created" 

147 assert public_key_path.exists(), "Public key not created" 

148 config_path = network_dir / "config.yaml" 

149 assert config_path.exists(), "Config file not created"