Django测试MS-SQL遗留数据库及存储过程
在Django中,我想用单元测试来测试一个MS-SQL Server的旧数据库。这个数据库使用存储过程来添加数据。具体情况如下:
这个MS-SQL数据库在Django中的设置如下:
DATABASES['vadain_import'] = { 'ENGINE': 'sql_server.pyodbc', 'USER': 'xx', 'PASSWORD': 'xxx', 'HOST': '192.168.103.102', 'PORT': '', 'NAME': 'Vadain_import', 'OPTIONS': { 'driver': 'SQL Server Native Client 11.0', 'MARS_Connection': True, } }
数据库的模型是通过inspectdb生成的,比如:
class WaOrders(models.Model): order_id = models.IntegerField(primary_key=True, db_column='intOrderId') type = models.TextField(db_column='chvType', blank=True) class Meta: db_table = 'WA_ORDERS' database_name = 'vadain_import' managed = False # (There's a lot more of properties and models)
在模型中执行存储过程。我不能使用Django的save功能,比如WAOrders.save(),因为在MS-SQL数据库中,主键是通过存储过程生成的。
@classmethod def export(cls, **kwargs): # Stored procedure for adding data into the db sql = "declare @id int \ set @id=0\ declare @error varchar(1000)\ set @error=''\ exec UspWA_OrderImport\ @intOrderId=@id out\ ,@chvType=N'%s'" % kwargs['type'] + " \ ,@chvErrorMsg=@error output\ select @id as id, @error as 'error' \ " # Connection Vadain db cursor = connections['vadain_import'].cursor() # Execute sql stored procedure, no_count is needed otherwise it's returning an error # Return value primary key of WAOrders try: cursor.execute(no_count + sql) result = cursor.fetchone() # Check if primary key is set and if there are no errors: if result[0] > 1 and result[1] == '': # Commit SP cursor.execute('COMMIT') return result[0]
创建模型时需要进行映射,因为MS-SQL数据库期望的数据格式与普通对象(比如‘order’)不同。
def adding_data_into_vadain(self, order): for curtain in order.curtains.all(): order_id = WaOrders.export( type=format(curtain.type) ) # relation with default and vadain db. order.exported_id = order_id order.save()
这个功能在运行程序时工作正常,但当我运行‘manage.py test’时,会创建一个测试数据库。这就带来了以下问题:
- 创建测试数据库时缺少south表(在旧数据库中也不需要这些表)
- 把SOUTH_TESTS_MIGRATE改为False后,我收到错误信息,提示默认数据库中的表已经存在。
我的测试代码如下:
class AddDataServiceTest(TestCase):
fixtures = ['order']
def test_data_service(self):
# add data into vadain_import data
for order in Order.objects.all():
AddingDataService.adding_data_into_vadain(order)
# test the exported values
for order in Order.objects.all():
exported_order = WaOrders.objects.get(order_exported_id=order.exported_id)
self.assertEqual(exported_order.type, 'Pleat curtain')
有没有人能给我建议,如何测试这种情况?
1 个回答
0
也许在你的 WaOrders 里面,你可以扩展 save() 方法,然后在里面调用 export 函数。