1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
# Copyright (C) 2014 Nippon Telegraph and Telephone Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import time
import sys
import nose
import quagga_access as qaccess
import docker_control as fab
from gobgp_test import GoBGPTestBase
from gobgp_test import ADJ_RIB_IN, ADJ_RIB_OUT, LOCAL_RIB, GLOBAL_RIB
from gobgp_test import NEIGHBOR
from noseplugin import OptionParser
from noseplugin import parser_option
class GoBGPIPv6Test(GoBGPTestBase):
quagga_num = 2
append_quagga = 10
remove_quagga = 10
append_quagga_best = 20
def __init__(self, *args, **kwargs):
super(GoBGPIPv6Test, self).__init__(*args, **kwargs)
# test each neighbor state is turned establish
def test_01_ipv4_ipv6_neighbor_established(self):
print "test_ipv4_ipv6_neighbor_established"
use_local = parser_option.use_local
go_path = parser_option.go_path
log_debug = parser_option.gobgp_log_debug
fab.init_ipv6_test_env_executor(self.quagga_num, use_local, go_path, log_debug)
print "please wait (" + str(self.initial_wait_time) + " second)"
time.sleep(self.initial_wait_time)
fab.docker_container_ipv6_quagga_append_executor([3, 4], go_path)
print "please wait (" + str(self.initial_wait_time) + " second)"
time.sleep(self.initial_wait_time)
if self.check_load_config() is False:
return
addresses = self.get_neighbor_address(self.gobgp_config)
self.retry_routine_for_state(addresses, "BGP_FSM_ESTABLISHED")
for address in addresses:
# get neighbor state and remote ip from gobgp connections
print "check of [ " + address + " ]"
neighbor = self.ask_gobgp(NEIGHBOR, address)
state = neighbor['info']['bgp_state']
remote_ip = neighbor['conf']['remote_ip']
self.assertEqual(address, remote_ip)
self.assertEqual(state, "BGP_FSM_ESTABLISHED")
print "state" + state
def test_02_ipv4_ipv6_received_route(self):
print "test_ipv4_ipv6_received_route"
if self.check_load_config() is False:
return
for address in self.get_neighbor_address(self.gobgp_config):
print "check of [ " + address + " ]"
af = fab.IPv6 if ":" in address else fab.IPv4
local_rib = self.ask_gobgp(LOCAL_RIB, address, af)
for quagga_config in self.quagga_configs:
if quagga_config.peer_ip == address or quagga_config.ip_version != af:
for c_dest in quagga_config.destinations.itervalues():
# print "config : ", c_dest.prefix, "my ip or different ip version!!!"
g_dests = local_rib['Destinations']
exist_n = 0
for g_dest in g_dests:
# print "gobgp : ", g_dest['Prefix']
if c_dest.prefix == g_dest['Prefix']:
exist_n += 1
self.assertEqual(exist_n, 0)
else:
for c_dest in quagga_config.destinations.itervalues():
# print "config : ", c_dest.prefix"
g_dests = local_rib['Destinations']
exist_n = 0
for g_dest in g_dests:
# print "gobgp : ", g_dest['Prefix']
if c_dest.prefix == g_dest['Prefix']:
exist_n += 1
self.assertEqual(exist_n, 1)
def test_03_advertising_route(self):
print "test_advertising_route"
if self.check_load_config() is False:
return
for address in self.get_neighbor_address(self.gobgp_config):
print "check of [ " + address + " ]"
af = fab.IPv6 if ":" in address else fab.IPv4
tn = qaccess.login(address)
q_rib = qaccess.show_rib(tn, af)
for quagga_config in self.quagga_configs:
if quagga_config.peer_ip == address or quagga_config.ip_version != af:
for c_dest in quagga_config.destinations.itervalues():
exist_n = 0
for c_path in c_dest.paths:
# print "conf : ", c_path.network, c_path.nexthop, "my ip or different ip version!!!"
for q_path in q_rib:
# print "quag : ", q_path['Network'], q_path['Next Hop']
if "0.0.0.0" == q_path['Next Hop'] or "::" == q_path['Next Hop']:
continue
if c_path.network.split("/")[0] == q_path['Network']:
exist_n += 1
self.assertEqual(exist_n, 0)
else:
for c_dest in quagga_config.destinations.itervalues():
exist_n = 0
for c_path in c_dest.paths:
# print "conf : ", c_path.network, c_path.nexthop
for q_path in q_rib:
# print "quag : ", q_path['Network'], q_path['Next Hop']
if quagga_config.ip_version != fab.IPv6:
c_path.network = c_path.network.split("/")[0]
if c_path.network == q_path['Network'] and c_path.nexthop == q_path['Next Hop']:
exist_n += 1
self.assertEqual(exist_n, 1)
if __name__ == '__main__':
if fab.test_user_check() is False:
print "you are not root."
sys.exit(1)
if fab.docker_pkg_check() is False:
print "not install docker package."
sys.exit(1)
nose.main(argv=sys.argv, addplugins=[OptionParser()], defaultTest=sys.argv[0])
|
