资源警告Python 3单元测试中未关闭的套接字

zqdjd7g9  于 2022-12-24  发布在  Python
关注(0)|答案(2)|浏览(127)

我正在修改一些代码,以便在Python 2Python 3之间兼容,但在单元测试输出中观察到一个警告。

/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/unittest/case.py:601:
    ResourceWarning: unclosed socket.socket fd=4,
    family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6,
    laddr=('1.1.2.3', 65087), raddr=('5.8.13.21', 8080)

一项小的研究表明,requestsboto3等流行库也发生了这种情况。
我可以完全忽略警告或filter it。如果是我的服务,我可以在响应(link)中设置connection: close头。
下面的示例展示了Python 3.6.1中的警告:

    • 应用程序. py**
import requests

class Service(object):
    def __init__(self):
        self.session = requests.Session()

    def get_info(self):
        uri = 'http://api.stackexchange.com/2.2/info?site=stackoverflow'
        response = self.session.get(uri)
        if response.status_code == 200:
            return response.json()
        else:
            response.raise_for_status()

    def __del__(self):
        self.session.close()

if __name__ == '__main__':
    service = Service()
    print(service.get_info())
    • 测试. py**
import unittest

class TestService(unittest.TestCase):
    def test_growing(self):
        import app
        service = app.Service()
        res = service.get_info()
        self.assertTrue(res['items'][0]['new_active_users'] > 1)

if __name__ == '__main__':
    unittest.main()

是否有更好/正确的方法来管理会话,使其显式关闭,而不依赖于__del__()来导致此类警告。
谢谢你的帮助。

hfsqlsce

hfsqlsce1#

__del__中使用拆卸逻辑可能会使程序不正确或难以推理,因为无法保证何时调用该方法,从而可能导致您收到警告。有几种方法可以解决此问题:

1)公开一个方法以关闭会话,并在测试tearDown中调用它

unittesttearDown方法允许你定义一些在每次测试后运行的代码,使用这个钩子关闭会话即使测试失败或者有异常也会起作用,这很不错。

    • 应用程序. py**
import requests

class Service(object):

    def __init__(self):
        self.session = requests.Session()

    def get_info(self):
        uri = 'http://api.stackexchange.com/2.2/info?site=stackoverflow'
        response = self.session.get(uri)
        if response.status_code == 200:
            return response.json()
        else:
            response.raise_for_status()

    def close(self):
        self.session.close()

if __name__ == '__main__':
    service = Service()
    print(service.get_info())
    service.close()
    • 测试. py**
import unittest
import app

class TestService(unittest.TestCase):

    def setUp(self):
        self.service = app.Service()
        super().setUp()

    def tearDown(self):
        self.service.close()

    def test_growing(self):
        res = self.service.get_info()
        self.assertTrue(res['items'][0]['new_active_users'] > 1)

if __name__ == '__main__':
    unittest.main()

2)使用上下文管理器

上下文管理器也是一个非常有用的方式来显式地定义某个对象的作用域。在前面的例子中,你必须确保.close()在每个调用点都被正确地调用,否则你的资源会泄漏。使用上下文管理器,即使在上下文管理器的作用域中有异常,这也会被自动处理。
在解决方案1)的基础上,您可以定义额外的魔术方法(__enter____exit__),以便您的类可以使用with语句。

  • 注意:这里的好处是,这段代码还支持解决方案1)中的用法,显式使用.close(),如果上下文管理器由于某种原因而不方便使用,这会很有用。*
    • 应用程序. py**
import requests

class Service(object):

    def __init__(self):
        self.session = requests.Session()

    def __enter__(self):
        return self

    def get_info(self):
        uri = 'http://api.stackexchange.com/2.2/info?site=stackoverflow'
        response = self.session.get(uri)
        if response.status_code == 200:
            return response.json()
        else:
            response.raise_for_status()

    def close(self):
        self.session.close()

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

if __name__ == '__main__':
    with Service() as service:
        print(service.get_info())
    • 测试. py**
import unittest

import app

class TestService(unittest.TestCase):

    def test_growing(self):
        with app.Service() as service:
            res = service.get_info()
        self.assertTrue(res['items'][0]['new_active_users'] > 1)

if __name__ == '__main__':
    unittest.main()

根据您的需要,您可以使用setUp/tearDown和上下文管理器中的一个或组合,并消除该警告,同时在代码中拥有更显式的资源管理!

pb3s4cty

pb3s4cty2#

如果您不太关心警告,则这是最佳解决方案
只需导入警告并在驱动程序启动的位置添加此行-

import warnings

warnings.filterwarnings(action="ignore", message="unclosed", category=ResourceWarning)

相关问题