build.models.EVCDeploy.prepare_push_flow() - Code Metrics - Inspection of "Split EVC class" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#67)

by macartur

created 2018-09-04 18:14 UTC

build.models.EVCDeploy.prepare_push_flow() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	33
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
cc	1
eloc	14
nop	2
dl	0
loc	33
ccs	0
cts	12
cp	0
crap	2
rs	9.7
c	0
b	0
f	0

"""Classes used in the main application."""
from datetime import datetime
from uuid import uuid4

import requests

from kytos.core import log
from kytos.core.common import GenericEntity
from kytos.core.helpers import get_time, now
from kytos.core.interface import UNI
from napps.kytos.mef_eline import settings


class EVCBase(GenericEntity):
    """"Class to represent a circuit."""

    unique_attributes = ['name', 'uni_a', 'uni_z']

    def __init__(self, **kwargs):
        """Create an EVC instance with the provided parameters.

        Args:
            id(str): EVC identifier. Whether it's None an ID will be genereted.
            name: represents an EVC name.(Required)
            uni_a (UNI): Endpoint A for User Network Interface.(Required)
            uni_z (UNI): Endpoint Z for User Network Interface.(Required)
            start_date(datetime|str): Date when the EVC was registred.
                                      Default is now().
            end_date(datetime|str): Final date that the EVC will be fineshed.
                                    Default is None.
            bandwidth(int): Bandwidth used by EVC instance. Default is 0.
            primary_links(list): Primary links used by evc. Default is []
            backup_links(list): Backups links used by evc. Default is []
            current_path(list):circuit being used at the moment if this is an
                                active circuit. Default is [].
            primary_path(list): primary circuit offered to user IF one or more
                                links were provided. Default is [].
            backup_path(list): backup circuit offered to the user IF one or
                               more links were provided. Default is [].
            dynamic_backup_path(bool): Enable computer backup path dynamically.
                                       Dafault is False.
            creation_time(datetime|str): datetime when the circuit should be
                                         activated. default is now().
            enabled(Boolean): attribute to indicate the operational state.
                              default is False.
            active(Boolean): attribute to Administrative state;
                             default is False.
            owner(str): The EVC owner. Default is None.
            priority(int): Service level provided in the request. Default is 0.

        Raises:
            ValueError: raised when object attributes are invalid.

        """
        self._validate(**kwargs)
        super().__init__()

        # required attributes
        self._id = kwargs.get('id', uuid4().hex)
        self.uni_a = kwargs.get('uni_a')
        self.uni_z = kwargs.get('uni_z')
        self.name = kwargs.get('name')

        # optional attributes
        self.start_date = get_time(kwargs.get('start_date')) or now()
        self.end_date = get_time(kwargs.get('end_date')) or None

        self.bandwidth = kwargs.get('bandwidth', 0)
        self.primary_links = kwargs.get('primary_links', [])
        self.backup_links = kwargs.get('backup_links', [])
        self.current_path = kwargs.get('current_path', [])
        self.primary_path = kwargs.get('primary_path', [])
        self.backup_path = kwargs.get('backup_path', [])
        self.dynamic_backup_path = kwargs.get('dynamic_backup_path', False)
        self.creation_time = get_time(kwargs.get('creation_time')) or now()
        self.owner = kwargs.get('owner', None)
        self.priority = kwargs.get('priority', 0)
        self.circuit_scheduler = kwargs.get('circuit_scheduler', [])

        if kwargs.get('active', False):
            self.activate()
        else:
            self.deactivate()

        if kwargs.get('enabled', False):
            self.enable()
        else:
            self.disable()

        # datetime of user request for a EVC (or datetime when object was
        # created)
        self.request_time = kwargs.get('request_time', now())
        # dict with the user original request (input)
        self._requested = kwargs

    def update(self, **kwargs):
        """Update evc attributes.

        This method will raises an error trying to change the following
        attributes: [name, uni_a and uni_z]

        Raises:
            ValueError: message with error detail.

        """
        for attribute, value in kwargs.items():
            if attribute in self.unique_attributes:
                raise ValueError(f'{attribute} can\'t be be updated.')
            if hasattr(self, attribute):
                setattr(self, attribute, value)
            else:
                raise ValueError(f'The attribute "{attribute}" is invalid.')

    def __repr__(self):
        """Repr method."""
        return f"EVC({self._id}, {self.name})"

    def _validate(self, **kwargs):
        """Do Basic validations.

        Verify required attributes: name, uni_a, uni_z
        Verify if the attributes uni_a and uni_z are valid.

        Raises:
            ValueError: message with error detail.

        """
        for attribute in self.unique_attributes:

            if attribute not in kwargs:
                raise ValueError(f'{attribute} is required.')

            if 'uni' in attribute:
                uni = kwargs.get(attribute)
                if not isinstance(uni, UNI):
                    raise ValueError(f'{attribute} is an invalid UNI.')

                elif not uni.is_valid():
                    tag = uni.user_tag.value
                    message = f'VLAN tag {tag} is not available in {attribute}'
                    raise ValueError(message)

    def __eq__(self, other):
        """Override the default implementation."""
        if not isinstance(other, EVC):
            return False

        attrs_to_compare = ['name', 'uni_a', 'uni_z', 'owner', 'bandwidth']
        for attribute in attrs_to_compare:
            if getattr(other, attribute) != getattr(self, attribute):
                return False
        return True

    def as_dict(self):
        """Return a dictionary representing an EVC object."""
        evc_dict = {"id": self.id, "name": self.name,
                    "uni_a": self.uni_a.as_dict(),
                    "uni_z": self.uni_z.as_dict()}

        time_fmt = "%Y-%m-%dT%H:%M:%S"

        def link_as_dict(links):
            """Return list comprehension of links as_dict."""
            return [link.as_dict() for link in links if link]

        evc_dict["start_date"] = self.start_date
        if isinstance(self.start_date, datetime):
            evc_dict["start_date"] = self.start_date.strftime(time_fmt)

        evc_dict["end_date"] = self.end_date
        if isinstance(self.end_date, datetime):
            evc_dict["end_date"] = self.end_date.strftime(time_fmt)

        evc_dict['bandwidth'] = self.bandwidth
        evc_dict['primary_links'] = link_as_dict(self.primary_links)
        evc_dict['backup_links'] = link_as_dict(self.backup_links)
        evc_dict['current_path'] = link_as_dict(self.current_path)
        evc_dict['primary_path'] = link_as_dict(self.primary_path)
        evc_dict['backup_path'] = link_as_dict(self.backup_path)
        evc_dict['dynamic_backup_path'] = self.dynamic_backup_path

        if self._requested:
            request_dict = self._requested.copy()
            request_dict['uni_a'] = request_dict['uni_a'].as_dict()
            request_dict['uni_z'] = request_dict['uni_z'].as_dict()
            evc_dict['_requested'] = request_dict

        evc_dict["request_time"] = self.request_time
        if isinstance(self.request_time, datetime):
            evc_dict["request_time"] = self.request_time.strftime(time_fmt)

        time = self.creation_time.strftime(time_fmt)
        evc_dict['creation_time'] = time

        evc_dict['owner'] = self.owner
        evc_dict['circuit_scheduler'] = self.circuit_scheduler
        evc_dict['active'] = self.is_active()
        evc_dict['enabled'] = self.is_enabled()
        evc_dict['priority'] = self.priority

        return evc_dict

    @property
    def id(self):  # pylint: disable=invalid-name
        """Return this EVC's ID."""
        return self._id


class EVCDeploy(EVCBase):
    """Class to handle the deploy procedures."""

    def create(self):
        """Create a EVC."""
        pass

    def discover_new_path(self):
        """Discover a new path to satisfy this circuit and deploy."""
        pass

    def change_path(self, path):
        """Change EVC path."""
        pass

    def reprovision(self):
        """Force the EVC (re-)provisioning."""
        pass

    def remove(self):
        """Remove EVC path."""
        pass

    @staticmethod
    def choose_vlans(path=None):
        """Choose the VLANs to be used for the circuit."""
        for link in path:
            tag = link.get_next_available_tag()
            link.use_tag(tag)
            link.add_metadata('s_vlan', tag)

    @staticmethod
    def links_zipped(path=None):
        """Return an iterator which yields pairs of links in order."""
        if not path:
            return []
        return zip(path[:-1], path[1:])

    def should_deploy(self, path=None):
        """Verify if the circuit should be deployed."""
        if not path:
            log.debug("Path is empty.")
            return False

        if not self.is_enabled():
            log.debug(f'{self} is disabled.')
            return False

        if not self.is_active():
            log.debug(f'{self} will be deployed.')
            return True

        return False

    def deploy(self, path=None):
        """Install the flows for this circuit.

        Procedures to deploy:

        1. Decide if will deploy "path" or discover a new path
        2. Choose vlan
        3. Install NNI flows
        4. Install UNI flows
        5. Activate
        6. Update current_path
        7. Update links caches(primary, current, backup)

        """
        if not self.should_deploy(path):
            return

        self.choose_vlans(path)
        self.install_nni_flows(path)
        self.install_uni_flows(path)
        self.activate()
        log.info(f"{self} was deployed.")

    def install_nni_flows(self, path=None):
        """Install NNI flows."""
        for incoming, outcoming in self.links_zipped(path):
            in_vlan = incoming.get_metadata('s_vlan').value
            out_vlan = outcoming.get_metadata('s_vlan').value

            flows = []
            # Flow for one direction
            flows.append(self.prepare_nni_flow(incoming.endpoint_b,
                                               outcoming.endpoint_a,
                                               in_vlan, out_vlan))

            # Flow for the other direction
            flows.append(self.prepare_nni_flow(outcoming.endpoint_a,
                                               incoming.endpoint_b,
                                               out_vlan, in_vlan))
            self.send_flow_mods(incoming.endpoint_b.switch, flows)

    def install_uni_flows(self, path=None):
        """Install UNI flows."""
        if not path:
            log.info('install uni flows without path.')
            return

        # Determine VLANs
        in_vlan_a = self.uni_a.user_tag.value if self.uni_a.user_tag else None
        out_vlan_a = path[0].get_metadata('s_vlan').value

        in_vlan_z = self.uni_z.user_tag.value if self.uni_z.user_tag else None
        out_vlan_z = path[-1].get_metadata('s_vlan').value

        # Flows for the first UNI
        flows_a = []

        # Flow for one direction, pushing the service tag
        push_flow = self.prepare_push_flow(self.uni_a.interface,
                                           path[0].endpoint_a,
                                           in_vlan_a, out_vlan_a, in_vlan_z)
        flows_a.append(push_flow)

        # Flow for the other direction, popping the service tag
        pop_flow = self.prepare_pop_flow(path[0].endpoint_a,
                                         self.uni_a.interface, out_vlan_a)
        flows_a.append(pop_flow)

        self.send_flow_mods(self.uni_a.interface.switch, flows_a)

        # Flows for the second UNI
        flows_z = []

        # Flow for one direction, pushing the service tag
        push_flow = self.prepare_push_flow(self.uni_z.interface,
                                           path[-1].endpoint_b,
                                           in_vlan_z, out_vlan_z, in_vlan_a)
        flows_z.append(push_flow)

        # Flow for the other direction, popping the service tag
        pop_flow = self.prepare_pop_flow(path[-1].endpoint_b,
                                         self.uni_z.interface, out_vlan_z)
        flows_z.append(pop_flow)

        self.send_flow_mods(self.uni_z.interface.switch, flows_z)

    @staticmethod
    def send_flow_mods(switch, flow_mods):
        """Send a flow_mod list to a specific switch."""
        endpoint = "%s/flows/%s" % (settings.MANAGER_URL, switch.id)

        data = {"flows": flow_mods}
        requests.post(endpoint, json=data)

    @staticmethod
    def prepare_flow_mod(in_interface, out_interface):
        """Prepare a common flow mod."""
        default_action = {"action_type": "output",
                          "port": out_interface.port_number}

        flow_mod = {"match": {"in_port": in_interface.port_number},
                    "actions": [default_action]}

        return flow_mod

    def prepare_nni_flow(self, in_interface, out_interface, in_vlan, out_vlan):
        """Create NNI flows."""
        flow_mod = self.prepare_flow_mod(in_interface, out_interface)
        flow_mod['match']['dl_vlan'] = in_vlan

        new_action = {"action_type": "set_vlan",
                      "vlan_id": out_vlan}
        flow_mod["actions"].insert(0, new_action)

        return flow_mod

    def prepare_push_flow(self, *args):
        """Prepare push flow.

        Arguments:
            in_interface(str): Interface input.
            out_interface(str): Interface output.
            in_vlan(str): Vlan input.
            out_vlan(str): Vlan output.
            new_in_vlan(str): Interface input.

        Return:
            dict: An python dictionary representing a FlowMod

        """
        # assign all arguments
        in_interface, out_interface, in_vlan, out_vlan, new_in_vlan = args

        flow_mod = self.prepare_flow_mod(in_interface, out_interface)
        flow_mod['match']['dl_vlan'] = in_vlan

        new_action = {"action_type": "set_vlan",
                      "vlan_id": out_vlan}
        flow_mod["actions"].insert(0, new_action)

        new_action = {"action_type": "push_vlan",
                      "tag_type": "s"}
        flow_mod["actions"].insert(0, new_action)

        new_action = {"action_type": "set_vlan",
                      "vlan_id": new_in_vlan}
        flow_mod["actions"].insert(0, new_action)

        return flow_mod

    def prepare_pop_flow(self, in_interface, out_interface, in_vlan):
        """Prepare pop flow."""
        flow_mod = self.prepare_flow_mod(in_interface, out_interface)
        flow_mod['match']['dl_vlan'] = in_vlan
        new_action = {"action_type": "pop_vlan"}
        flow_mod["actions"].insert(0, new_action)
        return flow_mod


class EVC(EVCDeploy):
    """Class that represents a E-Line Virtual Connection."""
    pass


1			"""Classes used in the main application."""
2			from datetime import datetime
3			from uuid import uuid4
4
5			import requests
6
7			from kytos.core import log
8			from kytos.core.common import GenericEntity
9			from kytos.core.helpers import get_time, now
10			from kytos.core.interface import UNI
11			from napps.kytos.mef_eline import settings
12
13
14			class EVCBase(GenericEntity):
15			""""Class to represent a circuit."""
16
17			unique_attributes = ['name', 'uni_a', 'uni_z']
18
19			def __init__(self, **kwargs):
20			"""Create an EVC instance with the provided parameters.
21
22			Args:
23			id(str): EVC identifier. Whether it's None an ID will be genereted.
24			name: represents an EVC name.(Required)
25			uni_a (UNI): Endpoint A for User Network Interface.(Required)
26			uni_z (UNI): Endpoint Z for User Network Interface.(Required)
27			start_date(datetime\|str): Date when the EVC was registred.
28			Default is now().
29			end_date(datetime\|str): Final date that the EVC will be fineshed.
30			Default is None.
31			bandwidth(int): Bandwidth used by EVC instance. Default is 0.
32			primary_links(list): Primary links used by evc. Default is []
33			backup_links(list): Backups links used by evc. Default is []
34			current_path(list):circuit being used at the moment if this is an
35			active circuit. Default is [].
36			primary_path(list): primary circuit offered to user IF one or more
37			links were provided. Default is [].
38			backup_path(list): backup circuit offered to the user IF one or
39			more links were provided. Default is [].
40			dynamic_backup_path(bool): Enable computer backup path dynamically.
41			Dafault is False.
42			creation_time(datetime\|str): datetime when the circuit should be
43			activated. default is now().
44			enabled(Boolean): attribute to indicate the operational state.
45			default is False.
46			active(Boolean): attribute to Administrative state;
47			default is False.
48			owner(str): The EVC owner. Default is None.
49			priority(int): Service level provided in the request. Default is 0.
50
51			Raises:
52			ValueError: raised when object attributes are invalid.
53
54			"""
55			self._validate(**kwargs)
56			super().__init__()
57
58			# required attributes
59			self._id = kwargs.get('id', uuid4().hex)
60			self.uni_a = kwargs.get('uni_a')
61			self.uni_z = kwargs.get('uni_z')
62			self.name = kwargs.get('name')
63
64			# optional attributes
65			self.start_date = get_time(kwargs.get('start_date')) or now()
66			self.end_date = get_time(kwargs.get('end_date')) or None
67
68			self.bandwidth = kwargs.get('bandwidth', 0)
69			self.primary_links = kwargs.get('primary_links', [])
70			self.backup_links = kwargs.get('backup_links', [])
71			self.current_path = kwargs.get('current_path', [])
72			self.primary_path = kwargs.get('primary_path', [])
73			self.backup_path = kwargs.get('backup_path', [])
74			self.dynamic_backup_path = kwargs.get('dynamic_backup_path', False)
75			self.creation_time = get_time(kwargs.get('creation_time')) or now()
76			self.owner = kwargs.get('owner', None)
77			self.priority = kwargs.get('priority', 0)
78			self.circuit_scheduler = kwargs.get('circuit_scheduler', [])
79
80			if kwargs.get('active', False):
81			self.activate()
82			else:
83			self.deactivate()
84
85			if kwargs.get('enabled', False):
86			self.enable()
87			else:
88			self.disable()
89
90			# datetime of user request for a EVC (or datetime when object was
91			# created)
92			self.request_time = kwargs.get('request_time', now())
93			# dict with the user original request (input)
94			self._requested = kwargs
95
96			def update(self, **kwargs):
97			"""Update evc attributes.
98
99			This method will raises an error trying to change the following
100			attributes: [name, uni_a and uni_z]
101
102			Raises:
103			ValueError: message with error detail.
104
105			"""
106			for attribute, value in kwargs.items():
107			if attribute in self.unique_attributes:
108			raise ValueError(f'{attribute} can\'t be be updated.')
109			if hasattr(self, attribute):
110			setattr(self, attribute, value)
111			else:
112			raise ValueError(f'The attribute "{attribute}" is invalid.')
113
114			def __repr__(self):
115			"""Repr method."""
116			return f"EVC({self._id}, {self.name})"
117
118			def _validate(self, **kwargs):
119			"""Do Basic validations.
120
121			Verify required attributes: name, uni_a, uni_z
122			Verify if the attributes uni_a and uni_z are valid.
123
124			Raises:
125			ValueError: message with error detail.
126
127			"""
128			for attribute in self.unique_attributes:
129
130			if attribute not in kwargs:
131			raise ValueError(f'{attribute} is required.')
132
133			if 'uni' in attribute:
134			uni = kwargs.get(attribute)
135			if not isinstance(uni, UNI):
136			raise ValueError(f'{attribute} is an invalid UNI.')
137
138			elif not uni.is_valid():
139			tag = uni.user_tag.value
140			message = f'VLAN tag {tag} is not available in {attribute}'
141			raise ValueError(message)
142
143			def __eq__(self, other):
144			"""Override the default implementation."""
145			if not isinstance(other, EVC):
146			return False
147
148			attrs_to_compare = ['name', 'uni_a', 'uni_z', 'owner', 'bandwidth']
149			for attribute in attrs_to_compare:
150			if getattr(other, attribute) != getattr(self, attribute):
151			return False
152			return True
153
154			def as_dict(self):
155			"""Return a dictionary representing an EVC object."""
156			evc_dict = {"id": self.id, "name": self.name,
157			"uni_a": self.uni_a.as_dict(),
158			"uni_z": self.uni_z.as_dict()}
159
160			time_fmt = "%Y-%m-%dT%H:%M:%S"
161
162			def link_as_dict(links):
163			"""Return list comprehension of links as_dict."""
164			return [link.as_dict() for link in links if link]
165
166			evc_dict["start_date"] = self.start_date
167			if isinstance(self.start_date, datetime):
168			evc_dict["start_date"] = self.start_date.strftime(time_fmt)
169
170			evc_dict["end_date"] = self.end_date
171			if isinstance(self.end_date, datetime):
172			evc_dict["end_date"] = self.end_date.strftime(time_fmt)
173
174			evc_dict['bandwidth'] = self.bandwidth
175			evc_dict['primary_links'] = link_as_dict(self.primary_links)
176			evc_dict['backup_links'] = link_as_dict(self.backup_links)
177			evc_dict['current_path'] = link_as_dict(self.current_path)
178			evc_dict['primary_path'] = link_as_dict(self.primary_path)
179			evc_dict['backup_path'] = link_as_dict(self.backup_path)
180			evc_dict['dynamic_backup_path'] = self.dynamic_backup_path
181
182			if self._requested:
183			request_dict = self._requested.copy()
184			request_dict['uni_a'] = request_dict['uni_a'].as_dict()
185			request_dict['uni_z'] = request_dict['uni_z'].as_dict()
186			evc_dict['_requested'] = request_dict
187
188			evc_dict["request_time"] = self.request_time
189			if isinstance(self.request_time, datetime):
190			evc_dict["request_time"] = self.request_time.strftime(time_fmt)
191
192			time = self.creation_time.strftime(time_fmt)
193			evc_dict['creation_time'] = time
194
195			evc_dict['owner'] = self.owner
196			evc_dict['circuit_scheduler'] = self.circuit_scheduler
197			evc_dict['active'] = self.is_active()
198			evc_dict['enabled'] = self.is_enabled()
199			evc_dict['priority'] = self.priority
200
201			return evc_dict
202
203			@property
204			def id(self): # pylint: disable=invalid-name
205			"""Return this EVC's ID."""
206			return self._id
207
208
209			class EVCDeploy(EVCBase):
210			"""Class to handle the deploy procedures."""
211
212			def create(self):
213			"""Create a EVC."""
214			pass
215
216			def discover_new_path(self):
217			"""Discover a new path to satisfy this circuit and deploy."""
218			pass
219
220			def change_path(self, path):
221			"""Change EVC path."""
222			pass
223
224			def reprovision(self):
225			"""Force the EVC (re-)provisioning."""
226			pass
227
228			def remove(self):
229			"""Remove EVC path."""
230			pass
231
232			@staticmethod
233			def choose_vlans(path=None):
234			"""Choose the VLANs to be used for the circuit."""
235			for link in path:
236			tag = link.get_next_available_tag()
237			link.use_tag(tag)
238			link.add_metadata('s_vlan', tag)
239
240			@staticmethod
241			def links_zipped(path=None):
242			"""Return an iterator which yields pairs of links in order."""
243			if not path:
244			return []
245			return zip(path[:-1], path[1:])
246
247			def should_deploy(self, path=None):
248			"""Verify if the circuit should be deployed."""
249			if not path:
250			log.debug("Path is empty.")
251			return False
252
253			if not self.is_enabled():
254			log.debug(f'{self} is disabled.')
255			return False
256
257			if not self.is_active():
258			log.debug(f'{self} will be deployed.')
259			return True
260
261			return False
262
263			def deploy(self, path=None):
264			"""Install the flows for this circuit.
265
266			Procedures to deploy:
267
268			1. Decide if will deploy "path" or discover a new path
269			2. Choose vlan
270			3. Install NNI flows
271			4. Install UNI flows
272			5. Activate
273			6. Update current_path
274			7. Update links caches(primary, current, backup)
275
276			"""
277			if not self.should_deploy(path):
278			return
279
280			self.choose_vlans(path)
281			self.install_nni_flows(path)
282			self.install_uni_flows(path)
283			self.activate()
284			log.info(f"{self} was deployed.")
285
286			def install_nni_flows(self, path=None):
287			"""Install NNI flows."""
288			for incoming, outcoming in self.links_zipped(path):
289			in_vlan = incoming.get_metadata('s_vlan').value
290			out_vlan = outcoming.get_metadata('s_vlan').value
291
292			flows = []
293			# Flow for one direction
294			flows.append(self.prepare_nni_flow(incoming.endpoint_b,
295			outcoming.endpoint_a,
296			in_vlan, out_vlan))
297
298			# Flow for the other direction
299			flows.append(self.prepare_nni_flow(outcoming.endpoint_a,
300			incoming.endpoint_b,
301			out_vlan, in_vlan))
302			self.send_flow_mods(incoming.endpoint_b.switch, flows)
303
304			def install_uni_flows(self, path=None):
305			"""Install UNI flows."""
306			if not path:
307			log.info('install uni flows without path.')
308			return
309
310			# Determine VLANs
311			in_vlan_a = self.uni_a.user_tag.value if self.uni_a.user_tag else None
312			out_vlan_a = path[0].get_metadata('s_vlan').value
313
314			in_vlan_z = self.uni_z.user_tag.value if self.uni_z.user_tag else None
315			out_vlan_z = path[-1].get_metadata('s_vlan').value
316
317			# Flows for the first UNI
318			flows_a = []
319
320			# Flow for one direction, pushing the service tag
321			push_flow = self.prepare_push_flow(self.uni_a.interface,
322			path[0].endpoint_a,
323			in_vlan_a, out_vlan_a, in_vlan_z)
324			flows_a.append(push_flow)
325
326			# Flow for the other direction, popping the service tag
327			pop_flow = self.prepare_pop_flow(path[0].endpoint_a,
328			self.uni_a.interface, out_vlan_a)
329			flows_a.append(pop_flow)
330
331			self.send_flow_mods(self.uni_a.interface.switch, flows_a)
332
333			# Flows for the second UNI
334			flows_z = []
335
336			# Flow for one direction, pushing the service tag
337			push_flow = self.prepare_push_flow(self.uni_z.interface,
338			path[-1].endpoint_b,
339			in_vlan_z, out_vlan_z, in_vlan_a)
340			flows_z.append(push_flow)
341
342			# Flow for the other direction, popping the service tag
343			pop_flow = self.prepare_pop_flow(path[-1].endpoint_b,
344			self.uni_z.interface, out_vlan_z)
345			flows_z.append(pop_flow)
346
347			self.send_flow_mods(self.uni_z.interface.switch, flows_z)
348
349			@staticmethod
350			def send_flow_mods(switch, flow_mods):
351			"""Send a flow_mod list to a specific switch."""
352			endpoint = "%s/flows/%s" % (settings.MANAGER_URL, switch.id)
353
354			data = {"flows": flow_mods}
355			requests.post(endpoint, json=data)
356
357			@staticmethod
358			def prepare_flow_mod(in_interface, out_interface):
359			"""Prepare a common flow mod."""
360			default_action = {"action_type": "output",
361			"port": out_interface.port_number}
362
363			flow_mod = {"match": {"in_port": in_interface.port_number},
364			"actions": [default_action]}
365
366			return flow_mod
367
368			def prepare_nni_flow(self, in_interface, out_interface, in_vlan, out_vlan):
369			"""Create NNI flows."""
370			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
371			flow_mod['match']['dl_vlan'] = in_vlan
372
373			new_action = {"action_type": "set_vlan",
374			"vlan_id": out_vlan}
375			flow_mod["actions"].insert(0, new_action)
376
377			return flow_mod
378
379			def prepare_push_flow(self, *args):
380			"""Prepare push flow.
381
382			Arguments:
383			in_interface(str): Interface input.
384			out_interface(str): Interface output.
385			in_vlan(str): Vlan input.
386			out_vlan(str): Vlan output.
387			new_in_vlan(str): Interface input.
388
389			Return:
390			dict: An python dictionary representing a FlowMod
391
392			"""
393			# assign all arguments
394			in_interface, out_interface, in_vlan, out_vlan, new_in_vlan = args
395
396			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
397			flow_mod['match']['dl_vlan'] = in_vlan
398
399			new_action = {"action_type": "set_vlan",
400			"vlan_id": out_vlan}
401			flow_mod["actions"].insert(0, new_action)
402
403			new_action = {"action_type": "push_vlan",
404			"tag_type": "s"}
405			flow_mod["actions"].insert(0, new_action)
406
407			new_action = {"action_type": "set_vlan",
408			"vlan_id": new_in_vlan}
409			flow_mod["actions"].insert(0, new_action)
410
411			return flow_mod
412
413			def prepare_pop_flow(self, in_interface, out_interface, in_vlan):
414			"""Prepare pop flow."""
415			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
416			flow_mod['match']['dl_vlan'] = in_vlan
417			new_action = {"action_type": "pop_vlan"}
418			flow_mod["actions"].insert(0, new_action)
419			return flow_mod
420
421
422			class EVC(EVCDeploy):
423			"""Class that represents a E-Line Virtual Connection."""
424			pass
425

kytos / mef_eline

Pull Request — master (#67)

build.models.EVCDeploy.prepare_push_flow() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like