4
4
from pathlib import Path
5
5
from typing import Optional
6
6
7
+ import pexpect
7
8
from test_base import TestBase
8
9
10
+ from warnet .k8s import get_kubeconfig_value
11
+
9
12
10
13
class NamespaceAdminTest (TestBase ):
11
14
def __init__ (self ):
@@ -22,6 +25,8 @@ def run_test(self):
22
25
try :
23
26
self .setup_namespaces ()
24
27
self .setup_service_accounts ()
28
+ self .deploy_network_in_team_namespaces ()
29
+ self .authenticate_bob ()
25
30
finally :
26
31
self .cleanup ()
27
32
@@ -36,12 +41,51 @@ def setup_service_accounts(self):
36
41
self .log .info (self .warnet ("admin create-kubeconfigs" ))
37
42
self .wait_for_predicate (self .service_accounts_are_validated )
38
43
39
- def get_service_accounts (self ) -> Optional [list [dict [str , str ]]]:
44
+ def deploy_network_in_team_namespaces (self ):
45
+ network_name = "teamnet"
46
+ down = "\x1b [B"
47
+ self .log .info (f"Create a warnet network chart called '{ network_name } '" )
48
+ child = pexpect .spawn ("warnet init" )
49
+ child .expect ("Do you want to create a custom network" )
50
+ child .sendline ("y" )
51
+ child .expect ("Enter your network name" )
52
+ child .sendline (network_name )
53
+ child .expect ("How many nodes would you like" )
54
+ child .send (down )
55
+ child .send (down )
56
+ child .send (down )
57
+ child .sendline ()
58
+ child .expect ("How many connections would you like each node to have" )
59
+ child .sendline ()
60
+ child .expect ("Which version would you like nodes to run by default" )
61
+ child .sendline ()
62
+ child .expect ("Enter the number of nodes" )
63
+ child .sendline ("2" )
64
+ child .expect ("Would you like to enable fork-observer on the network" )
65
+ child .sendline ("False" )
66
+ child .expect ("Would you like to enable grafana logging on the network" )
67
+ child .sendline ("False" )
68
+ child .expect (pexpect .EOF )
69
+ self .log .info (f"Created the network chart called: { network_name } " )
70
+ self .log .info ("Deploy networks to team namespaces" )
71
+ for namespace in self .get_namespaces ():
72
+ self .log .info (f"Deploying network into: { namespace } " )
73
+ self .log .info (self .warnet (f"deploy networks/{ network_name } / --namespace { namespace } " ))
74
+ self .wait_for_all_tanks_status ()
75
+ self .wait_for_all_edges ()
76
+
77
+ def authenticate_bob (self ):
78
+ self .log .info ("Authenticating bob..." )
79
+ assert get_kubeconfig_value ("{.current-context}" ) == "minikube"
80
+ self .log .info (self .warnet ("auth kubeconfigs/bob-wargames-red-team-kubeconfig" ))
81
+ assert get_kubeconfig_value ("{.current-context}" ) == "bob-wargames-red-team"
82
+
83
+ def get_service_accounts (self ) -> Optional [dict [str , str ]]:
40
84
self .log .info ("Setting up service accounts" )
41
85
resp = self .warnet ("admin service-accounts list" )
42
86
if resp == "Could not find any matching service accounts." :
43
87
return None
44
- service_accounts = {}
88
+ service_accounts : dict [ str , [ str ]] = {}
45
89
current_namespace = ""
46
90
for line in resp .splitlines ():
47
91
if line .startswith ("Service" ):
0 commit comments