Django测试MS-SQL遗留数据库及存储过程

0 投票
1 回答
851 浏览
提问于 2025-04-18 16:22

在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 函数。

撰写回答