<commit_before>"""
Tests for the opennsa service.
"""

import os
import uuid
import time
import datetime
import StringIO

from twisted.trial import unittest
from twisted.internet import defer, reactor

from opennsa import nsa, setup, registry
from opennsa.backends import dud

from . import topology as testtopology


class ServiceTest(unittest.TestCase):

    def setUp(self):

        self.iports = []

        HOST = 'localhost'
        WSDL_DIR = os.path.join(os.getcwd(), '..', 'wsdl')

        # service

        SERVICES = [ ('Aruba', 9080), ('Bonaire', 9081), ('Curacao',9082) ]

        for network, port in SERVICES:

            topo_source = StringIO.StringIO(testtopology.TEST_TOPOLOGY)
            backend = dud.DUDNSIBackend(network)
            sr = registry.ServiceRegistry()
            factory = setup.createService(network, [ topo_source ], backend, sr, HOST, port, WSDL_DIR)

            iport = reactor.listenTCP(port, factory, interface='localhost')
            self.iports.append(iport)

        # client

        CLIENT_PORT = 7080

        self.client, client_factory  = setup.createClient(HOST, CLIENT_PORT, WSDL_DIR)
        self.client_nsa = nsa.NetworkServiceAgent('OpenNSA-Test-Client', 'nsa://localhost:%i' % CLIENT_PORT)

        client_iport = reactor.listenTCP(CLIENT_PORT, client_factory)
        self.iports.append(client_iport)


    def tearDown(self):
        for iport in self.iports:
            iport.stopListening()


    @defer.inlineCallbacks
    def testBasicConnectionLifeCycle(self):

        provider = nsa.Network('Aruba', nsa.NetworkServiceAgent('Aruba-OpenNSA', 'http://localhost:9080/NSI/services/ConnectionService'))

        source_stp      = nsa.STP('Aruba', 'A1' )
        dest_stp        = nsa.STP('Aruba', 'A2')
        #dest_stp        = nsa.STP('Bonaire', 'B3')
        #dest_stp        = nsa.STP('Curacao', 'C3')

        start_time = datetime.datetime.utcfromtimestamp(time.time() + 1.5 )
        end_time   = datetime.datetime.utcfromtimestamp(time.time() + 120 )

        bwp = nsa.BandwidthParameters(200)
        service_params  = nsa.ServiceParameters(start_time, end_time, source_stp, dest_stp, bandwidth=bwp)
        global_reservation_id = 'urn:uuid:' + str(uuid.uuid1())
        connection_id         = 'conn-id1'

        yield self.client.reserve(self.client_nsa, provider.nsa, None, global_reservation_id, 'Test Connection', connection_id, service_params)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Reserved')

        yield self.client.provision(self.client_nsa, provider.nsa, None, connection_id)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Provisioned')

        yield self.client.release(self.client_nsa, provider.nsa, None, connection_id)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Scheduled')

        yield self.client.terminate(self.client_nsa, provider.nsa, None, connection_id)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Terminated')

        # give the service side time to dump its connections
        # this prevents finishing the test with a dirty reactor
        from twisted.internet import task
        d = task.deferLater(reactor, 0.01, lambda : None)
        yield d

<commit_msg>add error for testing invalid network name<commit_after>"""
Tests for the opennsa service.
"""

import os
import uuid
import time
import datetime
import StringIO

from twisted.trial import unittest
from twisted.internet import defer, reactor

from opennsa import nsa, error, setup, registry
from opennsa.backends import dud

from . import topology as testtopology


class ServiceTest(unittest.TestCase):

    def setUp(self):

        self.iports = []

        HOST = 'localhost'
        WSDL_DIR = os.path.join(os.getcwd(), '..', 'wsdl')

        # service

        SERVICES = [ ('Aruba', 9080), ('Bonaire', 9081), ('Curacao',9082) ]

        for network, port in SERVICES:

            topo_source = StringIO.StringIO(testtopology.TEST_TOPOLOGY)
            backend = dud.DUDNSIBackend(network)
            sr = registry.ServiceRegistry()
            factory = setup.createService(network, [ topo_source ], backend, sr, HOST, port, WSDL_DIR)

            iport = reactor.listenTCP(port, factory, interface='localhost')
            self.iports.append(iport)

        # client

        CLIENT_PORT = 7080

        self.client, client_factory  = setup.createClient(HOST, CLIENT_PORT, WSDL_DIR)
        self.client_nsa = nsa.NetworkServiceAgent('OpenNSA-Test-Client', 'nsa://localhost:%i' % CLIENT_PORT)

        client_iport = reactor.listenTCP(CLIENT_PORT, client_factory)
        self.iports.append(client_iport)


    def tearDown(self):
        for iport in self.iports:
            iport.stopListening()


    @defer.inlineCallbacks
    def testBasicConnectionLifeCycle(self):

        provider = nsa.Network('Aruba', nsa.NetworkServiceAgent('Aruba-OpenNSA', 'http://localhost:9080/NSI/services/ConnectionService'))

        source_stp      = nsa.STP('Aruba', 'A1' )
        dest_stp        = nsa.STP('Aruba', 'A2')
        #dest_stp        = nsa.STP('Bonaire', 'B3')
        #dest_stp        = nsa.STP('Curacao', 'C3')

        start_time = datetime.datetime.utcfromtimestamp(time.time() + 1.5 )
        end_time   = datetime.datetime.utcfromtimestamp(time.time() + 120 )

        bwp = nsa.BandwidthParameters(200)
        service_params  = nsa.ServiceParameters(start_time, end_time, source_stp, dest_stp, bandwidth=bwp)
        global_reservation_id = 'urn:uuid:' + str(uuid.uuid1())
        connection_id         = 'conn-id1'

        yield self.client.reserve(self.client_nsa, provider.nsa, None, global_reservation_id, 'Test Connection', connection_id, service_params)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Reserved')

        yield self.client.provision(self.client_nsa, provider.nsa, None, connection_id)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Provisioned')

        yield self.client.release(self.client_nsa, provider.nsa, None, connection_id)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Scheduled')

        yield self.client.terminate(self.client_nsa, provider.nsa, None, connection_id)

        qr = yield self.client.query(self.client_nsa, provider.nsa, None, "Summary", connection_ids = [ connection_id ] )
        self.assertEquals(qr.reservationSummary[0].connectionState, 'Terminated')

        # give the service side time to dump its connections
        # this prevents finishing the test with a dirty reactor
        from twisted.internet import task
        d = task.deferLater(reactor, 0.01, lambda : None)
        yield d


    @defer.inlineCallbacks
    def testInvalidNetwork(self):

        provider = nsa.Network('Aruba', nsa.NetworkServiceAgent('Aruba-OpenNSA', 'http://localhost:9080/NSI/services/ConnectionService'))

        source_stp      = nsa.STP('NoSuchNetwork', 'PS' )
        dest_stp        = nsa.STP('Aruba', 'A2')

        start_time = datetime.datetime.utcfromtimestamp(time.time() + 1.5 )
        end_time   = datetime.datetime.utcfromtimestamp(time.time() + 120 )

        bwp = nsa.BandwidthParameters(200)
        service_params  = nsa.ServiceParameters(start_time, end_time, source_stp, dest_stp, bandwidth=bwp)
        connection_id         = 'conn-id1'

        try:
            yield self.client.reserve(self.client_nsa, provider.nsa, None, None, '', connection_id, service_params)
            self.fail('Reserve call should have failed')
        except error.ReserveError:
            pass


{'fraction_non_alphanumeric': 0.11201605288007555, 'fraction_numerical': 0.0113314447592068, 'mean_word_length': 3.113106796116505, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3473849', 'n_tokens_mistral': 2759, 'n_tokens_neox': 2583, 'n_words': 522}