This repository has been archived by the owner on May 22, 2018. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 24
/
test.py
98 lines (80 loc) · 2.87 KB
/
test.py
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
"""Test kytos.core.controller module."""
import json
import logging
import warnings
from copy import copy
from unittest import TestCase
from unittest.mock import Mock, patch
from unittest.mock import MagicMock
from kytos.core.controller import Controller
from kytos.core.switch import Switch
# from .main import CircuitManager
from napps.amlight.mef_eline.models import Endpoint, Circuit, Path, Tag
# from .pathfinder import Pathfinder
from .flowmanager import FlowManager
#
#
# class TestCircuitManager(TestCase):
#
# def test_circuit(self):
# circuitManager = CircuitManager()
# _id = circuitManager._new_circuit_id()
# self.assertEqual(1, _id)
#
# for num in range(2, 10):
# _id = circuitManager._new_circuit_id()
# self.assertEqual(num, _id)
#
#
# class TestPathfinder(TestCase):
# def test_get_paths(self):
# pathfinder = Pathfinder()
# endpoint_a = '00:00:00:00:00:00:00:01:1'
# endpoint_b = '00:00:00:00:00:00:00:02:1'
#
# pathfinder._get_paths(endpoint_a, endpoint_b)
#
# for h in pathfinder._paths:
# self.assertEqual(h['hops'][0], '00:00:00:00:00:00:00:01:1')
# self.assertEqual(h['hops'][ len(h['hops'])-1 ], '00:00:00:00:00:00:00:02:1')
class TestFlowManager(TestCase):
# def test_install_flow(self):
# def get_switch_by_dpid(*args, **kwargs):
# return Switch('00:00:00:00:00:00:00:01')
#
# c = Controller()
# c.get_switch_by_dpid = get_switch_by_dpid
#
# flowmanager = FlowManager(c)
#
# endpoint_a = Endpoint('00:00:00:00:00:00:00:01', 3)
# endpoint_b = Endpoint('00:00:00:00:00:00:00:01', 4)
#
# flow = flowmanager.install_flow(endpoint_a, endpoint_b)
#
# self.assertIsNotNone(flow)
def test_install_circuit(self):
print ('test_install_circuit')
def get_switch_by_dpid(*args, **kwargs):
return Switch(args[0])
c = Controller()
c.get_switch_by_dpid = get_switch_by_dpid
flowmanager = FlowManager(c)
endpoint_a = Endpoint('00:00:00:00:00:00:00:01', 1)
endpoint_a._tag = Tag('ctag', 200)
endpoint_b = Endpoint('00:00:00:00:00:00:00:01', 2)
endpoint_c = Endpoint('00:00:00:00:00:00:00:02', 1)
endpoint_d = Endpoint('00:00:00:00:00:00:00:02', 2)
endpoint_e = Endpoint('00:00:00:00:00:00:00:04', 1)
endpoint_f = Endpoint('00:00:00:00:00:00:00:04', 2)
endpoint_f._tag = Tag('ctag', 200)
#FIXME path should be an object
path = Path([endpoint_a,
endpoint_b,
endpoint_c,
endpoint_d,
endpoint_e,
endpoint_f])
circuit = Circuit(0, 'circuitA', path)
flow = flowmanager.install_circuit(circuit)
self.assertIsNotNone(flow)