Hi,
I am newbie to Python,
I am trying to use unittest and python. My python script is like this:
#! /usr/bin/env python
__author__ = 'Milson Munakami'
__revision__ = '0.0.2'
import json
import urllib
import httplib
from scapy.all import *
import unittest
import os, sys, socket, struct, select, time
from threading import Thread
import logging
import traceback
from mininet.net import Mininet
from mininet.node import OVSSwitch, OVSKernelSwitch, Controller,
RemoteController
from mininet.log import setLogLevel, info
from mininet.cli import CLI
class testFirewallS1( unittest.TestCase ):
#I am trying to set net variable to global
global net
######################
def setUp(self):
self.controllerIp="127.0.0.1"
self.switch = "00:00:00:00:00:00:00:01"
self.destinationIp = "10.0.0.1"
self.startTime_ = time.time()
self.failed = False
self.reportStatus_ = True
self.name_ = "Firewall"
self.log = logging.getLogger("unittest")
self.CreateNet()
self.SetPrecondition()
def CreateNet(self):
"Create an empty network and add nodes to it."
net = Mininet( controller=RemoteController )
info( '*** Adding controller\n' )
net.addController( 'c0' , controller=RemoteController,ip=
"127.0.0.1", port=6633)
info( '*** Adding hosts\n' )
h1 = net.addHost( 'h1', ip='10.0.0.1' )
h2 = net.addHost( 'h2', ip='10.0.0.2' )
h3 = net.addHost( 'h3', ip='10.0.0.3' )
info( '*** Adding switch\n' )
s1 = net.addSwitch( 's1' )
info( '*** Creating links\n' )
net.addLink( h1, s1 )
net.addLink( h2, s1 )
net.addLink( h3, s1 )
info( '*** Starting network\n')
net.start()
def tearDown(self):
if self.failed:
return
duration = time.time() - self.startTime_
self.cleanup(True)
if self.reportStatus_:
self.log.info("=== Test %s completed normally (%d
sec)", self.name_, duration)
def cleanup(self, success):
sys.excepthook = sys.__excepthook__
self.SetFinalcondition()
try:
return
except NameError:
self.log.error("Exception hit during cleanup,
bypassing:\n%s\n\n" % traceback.format_exc())
pass
else:
fail("Expected a NameError")
def StatusFirewall(self):
command = "http://%s:8080/wm/firewall/module/status/json"; %
self.controllerIp
x = urllib.urlopen(command).read()
parsedResult = json.loads(x)
return parsedResult['result']
def CountFirewallRules(self):
command = "http://%s:8080/wm/firewall/rules/json"; %
self.controllerIp
x = urllib.urlopen(command).read()
return x
def CountFlowRules(self):
command = "http://%s:8080/wm/core/switch/%s/flow/json"; %
(self.controllerIp, self.switch)
x = urllib.urlopen(command).read()
parsedResult = json.loads(x)
content = parsedResult['00:00:00:00:00:00:00:01']
if content is None:
return "[]"
else:
return str(content)
def SetPrecondition(self):
command = "http://%s:8080/wm/firewall/module/enable/json"; %
self.controllerIp
urllib.urlopen(command).read()
# cleanup all Firewall rules
command = "http://%s:8080/wm/firewall/rules/json"; %
self.controllerIp
x = urllib.urlopen(command).read()
parsedResult = json.loads(x)
for i in range(len(parsedResult)):
params = "{\"ruleid\":\"%s\"}" %
parsedResult[i]['ruleid']
command = "/wm/firewall/rules/json"
url = "%s:8080" % self.controllerIp
connection = httplib.HTTPConnection(url)
connection.request("DELETE", command, params)
connection.getresponse().read()
# sleep for REST command to get processed to avoid racing
time.sleep(5)
def SetFinalcondition(self):
command = "http://%s:8080/wm/firewall/module/disable/json"; %
self.controllerIp
urllib.urlopen(command).read()
# cleanup all Firewall rules
command = "http://%s:8080/wm/firewall/rules/json"; %
self.controllerIp
x = urllib.urlopen(command).read()
parsedResult = json.loads(x)
for i in range(len(parsedResult)):
params = "{\"ruleid\":\"%s\"}" %
parsedResult[i]['ruleid']
command = "/wm/firewall/rules/json"
url = "%s:8080" % self.controllerIp
connection = httplib.HTTPConnection(url)
connection.request("DELETE", command, params)
connection.getresponse().read()
# sleep for REST command to get processed to avoid racing
time.sleep(5)
info( '*** Stopping network' )
net.stop()
#Precondition Test
def testPreConditionFirewall(self):
self.assertTrue("enabled" in self.StatusFirewall())
self.assertTrue("[]" in self.CountFirewallRules())
self.assertEqual("[]",self.CountFlowRules(), "should be empty")
#7
def testCreateFlow(self):
info( '*** Testing network connecivity\n')
net.pingAll()
#Using Scapy
#send(IP(dst="10.0.0.3")/ICMP()/"Hello World")
#info( '*** Running CLI\n' )
#CLI( net )
# Post Conditions Validation
self.assertTrue("enabled" in self.StatusFirewall())
self.assertTrue("[]" in self.CountFirewallRules())
self.assertNotEqual("[]",self.CountFlowRules(), "should not be
empty rules cause it is added from Firewall with Action = []")
#8
def testCountFlow(self):
command = "http://%s:8080/wm/core/switch/all/flow/json"; %
self.controllerIp
x = urllib.urlopen(command).read()
self.assertEqual("[]",self.CountFlowRules(), "should be empty")
# Post Conditions Validation
self.assertTrue("enabled" in self.StatusFirewall())
self.assertTrue("[]" in self.CountFirewallRules())
self.assertEqual("[]",self.CountFlowRules(), "should be empty")
#10
def testDeleteFlow(self):
# sleep for REST command to get processed to avoid racing
time.sleep(5)
command = "http://%s:8080/wm/core/switch/all/flow/json"; %
self.controllerIp
x = urllib.urlopen(command).read()
self.assertEqual("[]",self.CountFlowRules(), "should be empty")
# Post Conditions Validation
self.assertTrue("enabled" in self.StatusFirewall())
self.assertTrue("[]" in self.CountFirewallRules())
self.assertEqual("[]",self.CountFlowRules(), "should be empty")
def suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(testFirewallS1))
return suite
if __name__ == '__main__':
logging.basicConfig(filename='/tmp/testfirewall.log',
level=logging.DEBUG,
format='%(asctime)s %(levelname)s %(name)s %(message)s')
logger=logging.getLogger(__name__)
suiteFew = unittest.TestSuite()
# Preconditions:
suiteFew.addTest(testFirewallS1("testPreConditionFirewall"))
# 7. Add one Flow Entry
suiteFew.addTest(testFirewallS1("testCreateFlow"))
# 8. Count Flow rules - empty
suiteFew.addTest(testFirewallS1("testCountFlow"))
# 9. Update a Flow Rule - Not Implemented due to Controller's
Constraint
# 10. wait till the time-stamp period to validate the flow is
deleted with that time.
suiteFew.addTest(testFirewallS1("testDeleteFlow"))
#Testing the Test Cases Begins Here
unittest.TextTestRunner(verbosity=2).run(suiteFew)
#unittest.main()
#unittest.TextTestRunner(verbosity=2).run(suite())
I tired to set a globla net variable inside the class and tring to access it
from another methods like from def SetFinalcondition(self): and def
testCreateFlow(self): but the error says:\
ERROR: testPreConditionFirewall (__main__.testFirewallS1)
----------------------------------------------------------------------
Traceback (most recent call last):
File "mm.py", line 68, in tearDown
self.cleanup(True)
File "mm.py", line 74, in cleanup
self.SetFinalcondition()
File "mm.py", line 144, in SetFinalcondition
net.stop()
NameError: global name 'net' is not defined
======================================================================
ERROR: testCreateFlow (__main__.testFirewallS1)
----------------------------------------------------------------------
Traceback (most recent call last):
File "mm.py", line 156, in testCreateFlow
net.pingAll()
NameError: global name 'net' is not defined
======================================================================
ERROR: testCreateFlow (__main__.testFirewallS1)
----------------------------------------------------------------------
Traceback (most recent call last):
File "mm.py", line 68, in tearDown
self.cleanup(True)
File "mm.py", line 74, in cleanup
self.SetFinalcondition()
File "mm.py", line 144, in SetFinalcondition
net.stop()
NameError: global name 'net' is not defined
======================================================================
ERROR: testCountFlow (__main__.testFirewallS1)
----------------------------------------------------------------------
Traceback (most recent call last):
File "mm.py", line 68, in tearDown
self.cleanup(True)
File "mm.py", line 74, in cleanup
self.SetFinalcondition()
File "mm.py", line 144, in SetFinalcondition
net.stop()
NameError: global name 'net' is not defined
======================================================================
ERROR: testDeleteFlow (__main__.testFirewallS1)
----------------------------------------------------------------------
Traceback (most recent call last):
File "mm.py", line 68, in tearDown
self.cleanup(True)
File "mm.py", line 74, in cleanup
self.SetFinalcondition()
File "mm.py", line 144, in SetFinalcondition
net.stop()
NameError: global name 'net' is not defined
How to fix it!
Thanks for any help
- Milson
--
https://mail.python.org/mailman/listinfo/python-list
