首页
学习
活动
专区
工具
TVP
发布
社区首页 >问答首页 >Python模拟基类中的psycopg2

Python模拟基类中的psycopg2
EN

Stack Overflow用户
提问于 2017-09-07 06:46:40
回答 1查看 2.2K关注 0票数 1

嗨,我在模仿继承类中的游标对象时遇到了问题。下面是我的代码的一部分:

代码语言:javascript
复制
# classa.py
class ClassA(ClassB):
    def __init__(self, params):
        ClassB.__init__(params)
        # other params

    def get_current_attr(self):
        sql = "SELECT max(attr) FROM {};".format(self.table_name)
        self.rs_cursor.execute(sql)
        current_attr = self.rs_cursor.fetchone()[0]
        return current_attr

# classb.py
import psycopg2

class ClassB:
    def __init__(self, params):
        self.redshift_conn = self.open_redshift_connection()
        self.redshift_conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
        self.rs_cursor = self.redshift_conn.cursor()

现在,我想在ClassA的get_current_attr()函数中模拟psycopg2游标。我试着这样做:

代码语言:javascript
复制
@mock.patch("mypackage.classa.classb.psycopg2.connect")
def test_get_current_attr(self, mock_connect):
    mock_con = mock_connect.return_value
    mock_cur = mock_con.cursor.return_value
    mock_cur.fetch_one.return_value = 1
    result = self.gap_load_instance.get_current_batch_id()
    mock_cur.execute.assert_called_with(
        "SELECT max(attr) FROM tablename"
    )

我得到的是:

提升AssertionError(‘预期调用:%s\n未调用’%(预期,))

有人能帮我吗?

EN

回答 1

Stack Overflow用户

发布于 2018-07-22 08:05:21

模拟整个psycopg2是最简单的。我已经为您缺少的函数添加了存根,但这样的代码应该可以工作:

代码语言:javascript
复制
# classa.py
class ClassA(ClassB):
    def __init__(self, params):
        super().__init__(params)
        # other params

    def get_current_attr(self):
        sql = "SELECT max(attr) FROM {};".format(self.table_name)
        self.rs_cursor.execute(sql)
        current_attr = self.rs_cursor.fetchone()[0]
        return current_attr

# classb.py
import psycopg2

class ClassB:
    def __init__(self, params):
        self.redshift_conn = self.open_redshift_connection()
        self.redshift_conn.set_isolation_level(
            psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
        self.rs_cursor = self.redshift_conn.cursor()
        self.table_name = "atablename"

    def open_redshift_connection(self):
        return psycopg2.connect()

然后

代码语言:javascript
复制
from unittest import TestCase, mock
from classa import ClassA

class Tester(TestCase):
    @mock.patch("classb.psycopg2")
    def test_get_current_attr(self, mock_psycopg2):
        mock_cur = mock_psycopg2.connect().cursor()
        mock_cur.fetch_one.return_value = 1

        a = ClassA({})
        result = a.get_current_attr()

        mock_cur.execute.assert_called_with(
            "SELECT max(attr) FROM {};".format(a.table_name)
        )
票数 1
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/46085380

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档