Python 3单元测试中的ResourceWarning unclosed套接字

Question

我正在修改一些代码以兼容Python 2和Python 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)

一些研究确定这也发生在流行的图书馆,如请求和boto3.

我可以忽略警告或完全过滤它.如果是我的服务,我可以connection: close在我的响应(链接)中设置标题.

这是一个展示警告的示例Python 3.6.1:

app.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())

test.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__()导致这种警告.

谢谢你的帮助.

Answer 1

如果您不太关心警告，这是最好的解决方案

只需导入警告并在您的驱动程序启动的地方添加这一行 -

import warnings

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

Answer 2

包含拆解逻辑__del__可能会使您的程序不正确或难以推理，因为无法保证何时调用该方法，有可能导致您得到警告。有两种解决方法：

1）公开一种关闭会话的方法，并在测试中调用它 tearDown

unittest的tearDown方法可让您定义一些将在每次测试后运行的代码。即使测试失败或有异常，使用此钩子来关闭会话也可以工作，这很好。

app.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()

test.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()如果上下文管理器由于某些原因不方便使用，则此代码很有用。

app.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())

test.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和/ 或上下文管理器的组合，或者摆脱该警告，并在代码中进行更明确的资源管理！