Originally hosted at : https://github.com/erlepereira/thrift-experimentation (now deleted) Have pulled in inside here, owing to recent acquisition rumors ;) Date: 4 June 2018.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

157 lines
4.8 KiB

#!/usr/bin/env python
import sys, unittest
sys.path.append('../gen-py')
from atmservice import atmService
from atmservice.ttypes import *
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from thrift.server import TServer
import socket
if len(sys.argv) <= 2 or sys.argv[1] == '--help':
print ''
print 'Usage: ' + sys.argv[0] + ' <number of 20$ notes> <number of 50$ notes>'
print ''
sys.exit(0)
amt20 = int(sys.argv[1])
amt50 = int(sys.argv[2])
class atmHandler:
def __init__(self):
self.denom = [ 20, 50 ];
self.denom_amts = [ amt20, amt50 ];
self.vault = self.denom[0] * self.denom_amts[0] + self.denom[1] * self.denom_amts[1]
self.note20 = 0
self.note50 = 0
self.log = {}
def active(self):
print "active()"
def amountquery(self):
return self.vault
def statusquery(self):
return str(self.vault) + " - " + str(self.denom_amts[0]) + " in 20$ bills," + str(self.denom_amts[1]) + " in 50$ bills"
def atmloader(self, loadamount):
self.vault += loadamount
return "active: amount reserve is $" + str(loadamount)
def atmserve(self, number):
if self.validcombo(number) != None:
print "withdraw " + str(number)
self.vault -= number
print "left is " + str(self.vault)
else:
return "Invalid Amount"
return str(number) + " - " + str(self.note20) + " in 20$ bills," + str(self.note50) + " in 50$ bills"
return str(number)
def commandfailure(self, ouch):
ouch = "Something bad happended, amount not loaded"
return ouch
def transactionfailure (self, uhoh):
uhoh = "Sorry, we cannot process this amount"
return uhoh
def validcombo(self, number):
self.note50 = 0
self.note20 = 0
if number > 0 and number <= self.vault and len(str(number)) <= 3:
print str(number) +" requested"
while number >= 50:
self.note50 += 1
number -= 50
#print number
while number >= 20:
self.note20 += 1
number -= 20
#print number
if number != 0:
return None
self.denom_amts[0] -= self.note20
self.denom_amts[1] -= self.note50
return number
else:
return None
def selftesting(atmtesting):
case = unittest.TestLoader().loadTestsFromTestCase(atmtesting)
result = unittest.TestResult()
case(result)
if result.wasSuccessful():
return True
else:
print("Some tests failed!")
for test, err in result.failures + result.errors:
print(test)
print(err)
return False
class selfTest(unittest.TestCase):
def setUp(self):
self.testhandler = atmHandler()
def tearDown(self):
self.testhandler = None
def test_loader(self):
print self.testhandler.statusquery()
pass
def test_statusquery(self):
print "for Vault Amount :"
self.assertEqual(self.testhandler.vault, self.testhandler.amountquery())
print " ok "
def test_withdraw_inlimit(self):
print "for Withdrawl :"
self.testhandler.atmserve(20)
self.assertEqual((self.testhandler.vault ), self.testhandler.amountquery(), "First Withdrawal Failure")
def test_withdraw_outlimit(self):
print "for Exceeded Withdrawl Limit :"
value = self.testhandler.vault
self.testhandler.atmserve(self.testhandler.vault * 10)
self.assertEqual(value, self.testhandler.amountquery(), "Withdrawl Failure: Amount too large")
print " ok "
def test_invalidamount(self):
print "for Withdrawl Invalid Amount:"
value = self.testhandler.vault
print "value" + str(self.testhandler.vault)
self.testhandler.atmserve(21)
self.assertEqual(value, self.testhandler.amountquery(), "Wrong Amount Withdrawal Failure")
print " ok "
print "Running self test..."
if not(selftesting(selfTest)):
print "Init Tests Failed"
sys.exit()
print "Tests OK: resetting for use"
handler = atmHandler()
#re-init
handler.denom_amts = [ amt20, amt50 ];
handler.vault = handler.denom[0] * handler.denom_amts[0] + handler.denom[1] * handler.denom_amts[1]
print handler.statusquery()
processor = atmService.Processor(handler)
transport = TSocket.TServerSocket(port=8082)
tfactory = TTransport.TBufferedTransportFactory()
pfactory = TBinaryProtocol.TBinaryProtocolFactory()
server = TServer.TSimpleServer(processor, transport, tfactory, pfactory)
print "Starting atmServer......"
server.serve()
print "done!"