Revision c6ef66ba snf-astakos-app/astakos/quotaholder/test/config.py

b/snf-astakos-app/astakos/quotaholder/test/config.py
1 
# Copyright 2012 GRNET S.A. All rights reserved.
1 
# Copyright 2012, 2013 GRNET S.A. All rights reserved.
2 2 
#
3 3 
# Redistribution and use in source and binary forms, with or
4 4 
# without modification, are permitted provided that the following
......
36 36 
import os
37 37 
import subprocess
38 38 
import time
39 
from socket import socket, AF_INET, SOCK_STREAM, IPPROTO_TCP, error as socket_error
40 39 
from errno import ECONNREFUSED
41 40 
from os.path import dirname
42 41 

  		





  
  42
  
    
path = os.path.dirname(os.path.realpath(__file__))


  		





  
  43
  
    
os.environ['SYNNEFO_SETTINGS_DIR'] = path + '/settings'


  		





  
  44
  
    
os.environ['DJANGO_SETTINGS_MODULE'] = 'synnefo.settings'


  		





  
  45
  
    

  		





  43
  46
  
    
# The following import is copied from snf-tools/syneffo_tools/burnin.py


  		





  44
  47
  
    
# Thank you John Giannelos <johngian@grnet.gr>


  		





  45
  48
  
    
# Use backported unittest functionality if Python < 2.7


  		





  ......




  50
  53
  
    
        raise Exception("The unittest2 package is required for Python < 2.7")


  		





  51
  54
  
    
    import unittest


  		





  52
  55
  
    

  		





  53
  
  
    
from synnefo.lib.quotaholder import QuotaholderClient


  		





  54
  
  
    
from synnefo.lib.quotaholder.api import (InvalidKeyError, NoEntityError,


  		





  55
  
  
    
                                         NoQuantityError, NoCapacityError,


  		





  56
  
  
    
                                         ExportLimitError, ImportLimitError)


  		





  
  56
  
    
from astakos.quotaholder.callpoint import QuotaholderDjangoDBCallpoint


  		





  57
  57
  
    

  		





  58
  
  
    
import random 


  		





  
  58
  
    
import random


  		





  59
  59
  
    

  		





  60
  60
  
    
def printf(fmt, *args):


  		





  61
  61
  
    
    print(fmt.format(*args))


  		





  62
  62
  
    

  		





  63
  
  
    
def environ_get(key, default_value = ''):


  		





  64
  
  
    
    if os.environ.has_key(key):


  		





  65
  
  
    
        return os.environ.get(key)


  		





  66
  
  
    
    else:


  		





  67
  
  
    
        return default_value


  		





  68
  
  
    

  		





  69
  
  
    
# Use environ vars [TEST_]QH_{HOST, PORT}


  		





  70
  
  
    
QH_HOST = environ_get("TEST_QH_HOST", environ_get("QH_HOST", "127.0.0.1"))


  		





  71
  
  
    
QH_PORT = environ_get("TEST_QH_PORT", environ_get("QH_PORT", "8888"))


  		





  72
  
  
    

  		





  73
  
  
    
assert QH_HOST != None


  		





  74
  
  
    
assert QH_PORT != None


  		





  75
  
  
    

  		





  76
  
  
    
printf("Will connect to QH_HOST = {0}", QH_HOST)


  		





  77
  
  
    
printf("            and QH_PORT = {0}", QH_PORT)


  		





  78
  
  
    

  		





  79
  
  
    
QH_SERVER = '{0}:{1}'.format(QH_HOST, QH_PORT)


  		





  80
  
  
    
QH_URL = "http://{0}/quotaholder/v".format(QH_SERVER)


  		





  81
  63
  
    

  		





  82
  64
  
    
### DEFS ###


  		





  83
  65
  
    
def new_quota_holder_client():


  		





  84
  66
  
    
    """


  		





  85
  67
  
    
    Create a new quota holder api client


  		





  86
  68
  
    
    """


  		





  87
  
  
    
    return QuotaholderClient(QH_URL, 'test')


  		





  
  69
  
    
    return QuotaholderDjangoDBCallpoint()


  		





  88
  70
  
    

  		





  89
  71
  
    
def run_test_case(test_case):


  		





  90
  72
  
    
    """


  		





  ......




  114
  96
  
    

  		





  115
  97
  
    
HERE = dirname(__file__)


  		





  116
  98
  
    

  		





  117
  
  
    
def init_server():


  		





  118
  
  
    
    p = subprocess.Popen(['setsid', HERE+'/qh_init', QH_SERVER])


  		





  119
  
  
    
    s = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)


  		





  120
  
  
    
    while True:


  		





  121
  
  
    
        try:


  		





  122
  
  
    
            s.connect((QH_HOST, int(QH_PORT)))


  		





  123
  
  
    
            break


  		





  124
  
  
    
        except socket_error, e:


  		





  125
  
  
    
            if e.errno != ECONNREFUSED:


  		





  126
  
  
    
                raise


  		





  127
  
  
    
            time.sleep(0.1)


  		





  128
  
  
    
    return p.pid


  		





  129
  
  
    

  		





  130
  99
  
    
### CLASSES ###


  		





  131
  100
  
    
class QHTestCase(unittest.TestCase):


  		





  132
  101
  
    
    @classmethod


  		





  133
  102
  
    
    def setUpClass(self):


  		





  134
  
  
    
        self.server = init_server()


  		





  
  103
  
    
        subprocess.call([HERE+'/qh_init'])


  		





  135
  104
  
    
        self.qh = new_quota_holder_client()


  		





  136
  105
  
    

  		





  137
  106
  
    
    def setUp(self):


  		





  ......




  139
  108
  
    

  		





  140
  109
  
    
    @classmethod


  		





  141
  110
  
    
    def tearDownClass(self):


  		





  142
  
  
    
        from signal import SIGTERM


  		





  143
  
  
    
        os.kill(-self.server, SIGTERM)


  		





  144
  
  
    

  		





  145
  
  
    

  		





  146
  
  
    
### VARS ###


  		





  147
  
  
    
DefaultOrCustom = {


  		





  148
  
  
    
    True: "default",


  		





  149
  
  
    
    False: "custom"


  		





  150
  
  
    
}


  		





  151
  
  
    

  		





  
  111
  
    
        pass


  		












Also available in:
  Unified diff