Django 1.6.1 в Windows с Postgres
Я использую unittest для тестирования представления, которое загружает файл .cfg на сервер. Я столкнулся с ошибкой 2, называемой
(1) TransactionManagementError: в текущей транзакции произошла ошибка. Вы не можете выполнять запросы до конца атомарного блока. (2) OSError: [Errno 22] Недопустимый аргумент
Я записываю свое мнение и тест ниже.
Views.py
def upload(request, device_id):
owner=account.objects.get(user=request.user)
dev = get_object_or_404(device, pk=device_id, owner=owner)
state = 0
# Handle file upload
if request.method == 'POST':
form = UploadConfigForm(request.POST, request.FILES)
if form.is_valid():
new_file = form.save(commit=False)
#now ask the device to download the config file
sess, action, action_opt = add_device_action(device=dev, action=Action.AC_CONFIG_FILE_SET)
#get the action related to this file
if action_opt:
new_file.act = action_opt
else:
new_file.act = action
new_file.save()
form.save_m2m()
#add the payload to the action
if action_opt:
action_opt.payload = new_file.payload()
action_opt.save ()
else:
action.payload = new_file.payload()
action.save ()
if settings.RENDER_JSON_INSTEAD_OF_HTML:
# Render as an json response
resp = action.descriptor()
return HttpResponse(json.dumps(resp), content_type="application/json")
else:
# Redirect to the document list after POST
# return HttpResponseRedirect('..')
state = 1
else:
form = UploadConfigForm() # A empty, unbound form
тесты.py
class DevAPISimulDevTests(TestCase):
"""
Class that test the device API
"""
fixtures = ['test_auth.json', 'test_users.json', 'test_device.json']
def setUp(self):
"""
"""
ret = self.client.login(username='cws', password='cws123')
self.assertEqual(ret, True)
def test_api_sim_usr_upload_config_file(self):
"""
simulate the upload of the config file ba the user. submit with a file that is ok
"""
logger.debug('test_upload_config_file')
dev_id = 1
dev = device.objects.get(pk=dev_id)
url = reverse('device-upload', kwargs={'device_id':dev_id})
myfile = open('device/fixtures/cfg_ok.csr','r')
# with transaction.atomic():
response = self.client.post(url, {'config_file':myfile})
logger.debug('POST[%s]:[%s]', url, response.status_code)
self.assertEqual(response.status_code, 200)
ad = json.loads(response.content)
# Check the ad we think we should receive.
self.assertEqual(ad[Action.AD_EP_URL_BASE], settings.API_BASE_URL)
self.assertEqual(ad[Action.AD_EP_ACTION], settings.API_EP_ACK)
self.assertEqual(ad[Action.AD_EP_ERROR], settings.API_EP_ERR)
self.assertEqual(ad[Action.AD_DEVICE_CODE], dev.access_code)
self.assertEqual(ad[Action.AD_ACTION_CODE], Action.AC_ACKNOWLEDGE)
self.assertIsNone(ad.get(Action.AD_ACTION_PAYLOAD))
self.assertIsNone(ad.get(Action.AD_ACTION_WAIT))
# Check the DB
# We should have 1 session
self.assertEqual(dev.session_set.count(), 1)
sess = Session.objects.first()
self.assertEqual(sess.state, Session.STATE_SMS_SENT)
# We should have 2 actions more
self.assertEqual(sess.action_set.count(), 2)
# The first action should have the same token as in the SMS and the status should be in progress
first_act = sess.action_set.first()
self.assertEqual(first_act.next_token, ad[Action.AD_EP_TOKEN])
self.assertEqual(first_act.status, Action.STATUS_IN_PROGRESS)
# The second action should have the same token as in the SMS and the status should be queued
next_act = sess.action_set.last()
self.assertEqual(next_act.current_token, ad[Action.AD_EP_TOKEN])
self.assertEqual(next_act.status, Action.STATUS_QUEUED)
# test the status
status_should_be = {u'0': {u'state': u'in progress', u'state_label': u'processing', u'label': u'contacting device'},
u'1': {u'state': u'queued', u'state_label': u'not started', u'label': u'uploading config file to device'}}
self.test_sessionstatus(status_should_be)
return ad
# except IntegrityError:
# pass
def test_api_sim_dev_init_session (self):
"""
API : simulate the device receiving the first AD trhough SMS and initiating the session
"""
# try:
logger.debug('test_api_sim_dev_init_session')
# with transaction.atomic():
ad = self.test_api_sim_usr_upload_config_file()
# validate that the url is correct
url = reverse('device-api_ack', kwargs={'token':ad [Action.AD_EP_TOKEN]})
url2 = ''.join (['/device/', ad [Action.AD_EP_TOKEN], '/', ad[Action.AD_EP_ACTION]])
self.assertEqual(url, url2)
# Simulate the device processing the first ad of the session
# retrieve the next command from the server
response = self.client.get(url)
logger.debug('GET[%s]:[%s]', url, response.status_code)
# check the answer
self.assertEqual(response.status_code, 200)
ad2 = json.loads(response.content)
self.assertTrue(type(ad2) is dict)
# Check the ad we think we should receive.
self.assertEqual(ad2[Action.AD_ACTION_CODE], Action.AC_CONFIG_FILE_SET)
self.assertEqual(ad2[Action.AD_EP_ACTION], settings.API_EP_CFG)
self.assertTrue(type(ad2[Action.AD_ACTION_PAYLOAD]) is dict)
#TODO: will need on time to understand why this asssrt fail.
# self.assertContains(ad2[Action.AD_ACTION_PAYLOAD]['url'], '.csr')
self.assertIsNone(ad2.get(Action.AD_DEVICE_CODE))
self.assertIsNone(ad2.get(Action.AD_EP_URL_BASE))
self.assertIsNone(ad2.get(Action.AD_EP_ERROR))
# Check the DB
# We should have 1 session in STATE_WAIT_DEV_REQ
dev_id = 1
dev = device.objects.get(pk=dev_id)
self.assertEqual(dev.session_set.count(), 1)
sess = Session.objects.first()
self.assertEqual(sess.state, Session.STATE_WAIT_DEV_REQ)
# We should still have 2 actions
self.assertEqual(sess.action_set.count(), 2)
# The first action should have a status done
first_act = sess.action_set.first()
self.assertEqual(first_act.status, Action.STATUS_DONE)
self.assertEqual(first_act.next_token, ad[Action.AD_EP_TOKEN])
# The second action should have the same token as in the SMS and the status should be queued
next_act = sess.action_set.last()
self.assertEqual(next_act.current_token, ad[Action.AD_EP_TOKEN])
self.assertEqual(next_act.next_token, ad2[Action.AD_EP_TOKEN])
self.assertEqual(next_act.status, Action.STATUS_IN_PROGRESS)
# test the status
status_should_be = {u'0': {u'state': u'done', u'state_label': u'done', u'label': u'contacting device'},
u'1': {u'state': u'in progress', u'state_label': u'processing', u'label': u'uploading config file to device'}}
self.test_sessionstatus(status_should_be)
return ad2
Пожалуйста, предложите, что я здесь делаю неправильно. Если код, вставляемый сюда, не касается ошибок, укажите, какой код вызывает эту ошибку, так как я не могу понять происхождение этих ошибок.
