planetstack/core/tests.py

   1 #!/usr/bin/python
   2 from django.test import TestCase
   3 from core.models import *
   4 from rest_framework.test import *
   5 from genapi import *
   6 import json
   7 from datetime import datetime
   8
   9 FIXTURES_FILE = 'core/fixtures/initial_data.json'
  10 MODELS = ['Deployment','Image','Node','Reservation','Slice','Sliver','User']
  11
  12 def is_dynamic_type(x):
  13         t = type(x)
  14         return t in [datetime]
  15
  16 class APITestCase(TestCase):
  17         def setUp(self):
  18                 self.init_data=json.loads(open(FIXTURES_FILE).read())
  19                 self.data_dict={}
  20                 self.hidden_keys={}
  21
  22                 for d in self.init_data:
  23                         model_tag = d['model']
  24                         model_name = model_tag.split('.')[1]
  25
  26                         try:
  27                                 self.data_dict[model_name].append(d)
  28                         except:
  29                                 self.data_dict[model_name]=[d]
  30
  31                 # Any admin user would do
  32                 self.calling_user = User('sapan@onlab.us')
  33                 self.client = APIClient()
  34                 self.client.force_authenticate(user=self.calling_user)
  35
  36
  37         def check_items(self, response, data_list):
  38                 rdict = {}
  39                 for r in response:
  40                         rdict['%d'%r['id']]=r
  41
  42                 for d in data_list:
  43                         match = True
  44                         try:
  45                                 item = rdict['%d'%d['pk']]
  46                         except Exception,e:
  47                                 print 'API missing item %d / %r'%(d['pk'],rdict.keys())
  48                                 raise e
  49
  50                         fields=d['fields']
  51                         fields['id']=d['pk']
  52
  53                         for k in item.keys():
  54                                 try:
  55                                         resp_val = fields[k]
  56                                 except KeyError:
  57                                         if (not self.hidden_keys.has_key(k)):
  58                                                 print 'Hidden key %s'%k
  59                                                 self.hidden_keys[k]=True
  60
  61                                         continue
  62
  63                                 if (item[k]!=resp_val and not is_dynamic_type(item[k])):
  64                                         if (type(resp_val)==type(item[k])):
  65                                                 print 'Key %s did not match: 1. %r 2. %r'%(k,item[k],resp_val)
  66                                                 print fields
  67                                                 match = False
  68
  69
  70
  71         def create(self, model, mplural, record):
  72                 request = self.client.put('/plstackapi/%s/'%mplural,record['fields'])
  73
  74                 #if (len2==len1):
  75                 #       raise Exception('Could not delete %s/%d'%(model,pk))
  76
  77                 return
  78
  79         def update(self, model, mplural, pk):
  80                 src_record = self.data_dict[model.lower()][0]
  81                 record_to_update = src_record['fields']
  82                 now = datetime.now()
  83                 record_to_update['enacted']=now
  84                 response = self.client.put('/plstackapi/%s/%d/'%(mplural,pk),record_to_update)
  85                 self.assertEqual(response.data['enacted'],now)
  86
  87                 return
  88
  89         def delete(self, model, mplural, pk):
  90                 mclass = globals()[model]
  91                 len1 = len(mclass.objects.all())
  92                 response = self.client.delete('/plstackapi/%s/%d/'%(mplural,pk))
  93                 len2 = len(mclass.objects.all())
  94                 self.assertNotEqual(len1,len2)
  95
  96                 return
  97
  98         def retrieve(self, m, mplural, mlower):
  99                 response = self.client.get('/plstackapi/%s/'%mplural)
 100                 #force_authenticate(request,user=self.calling_user)
 101                 self.check_items(response.data,self.data_dict[mlower])
 102
 103                 return
 104
 105         def test_initial_retrieve(self):
 106                 for m in MODELS:
 107                         print 'Checking retrieve on %s...'%m
 108                         self.retrieve(m, m.lower()+'s',m.lower())
 109
 110
 111         def test_update(self):
 112                 for m in MODELS:
 113                         print 'Checking update on %s...'%m
 114                         first = self.data_dict[m.lower()][0]['pk']
 115                         self.update(m, m.lower()+'s',int(first))
 116
 117         def test_delete(self):
 118                 for m in MODELS:
 119                         print 'Checking delete on %s...'%m
 120                         first = self.data_dict[m.lower()][0]['pk']
 121                         self.delete(m, m.lower()+'s',int(first))
 122
 123         def test_create(self):
 124                 for m in MODELS:
 125                         print 'Checking create on %s...'%m
 126                         first = self.data_dict[m.lower()][0]
 127                         self.create(m, m.lower()+'s',first)
 128