Django用存储过程测试MSSQL遗留数据库

2024-10-02 12:27:35 发布

您现在位置:Python中文网/ 问答频道 /正文

在Django中,我希望使用单元测试来测试MS-sqlserver遗留数据库。数据库正在使用存储过程添加数据。情况如下:

  • 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的功能,比如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()
    

通过运行程序,该函数可以正常工作,但是运行'管理.py将创建一个测试数据库。存在以下问题:

  • 通过创建测试数据库缺少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')

有人能告诉我如何测试情况吗?在


Tags: importid数据库dbsqldatamodelstype

热门问题