Python 3单元测试中的ResourceWarning未关闭套接字

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)

经过一番研究，发现这个问题也出现在流行的库中，比如requests和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__()来导致这种警告。

感谢任何帮助。

- j12y

---

为您的类添加一个close方法，该方法将委托给底层资源的close方法。更好的做法是让您的类实现上下文管理器协议，然后在with语句中使用它（__exit__方法只需调用您的close方法，而__enter__方法可以简单地返回self，因此额外的工作不多）。 - ShadowRanger

不要使用 __del__，因为在关闭时无法保证以正确的顺序调用它。这就是上下文管理器被发明的原因。@ShadowRanger 是正确的 - 添加 enter 和 exit 方法并使用 with 语法。 - Major Eccles

在收到DeprecationWarning后，我遇到了这个错误（我使用的是assertEquals而不是assertEqual）。 - Dave Russell

Answer 1

在__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()

测试.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和上下文管理器的组合，从而消除该警告并在代码中实现更明确的资源管理！

- Samuel Dion-Girardeau

---

我是唯一一个即使使用上下文管理器仍然遇到这个错误的人吗？如果在setUp()中打开连接并在tearDown()中关闭，则情况相同。两者都使用也是如此... - 555Russich

@555Russich，你有一个最小可重现的示例吗？也许警告来自对request.Session的另一个使用？ - Samuel Dion-Girardeau

我有点撒谎，因为我正在使用类似的异步代码来复现同样的错误。具体来说，我使用了 aiohttp、unittest.IsolatedAsyncioTestCase、async with 上下文管理器等。刚刚创建了一个新问题。 - 555Russich

Answer 2

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

只需导入warnings，并在初始化驱动程序的地方添加此行 -

import warnings

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

- Shivam Bharadwaj

---

在我的情况下，在 if __name__ == '__main__': 之后添加 warnings.filterwarnings(... 行解决了问题。 - Comrade Che

我需要在每个类的第一个违规函数顶部放置“警告”行。 - JJones

对我来说，如果“warnings”行位于文件顶部，它将在类中起作用，否则它不起作用。 - Uri

对我来说，只有在出现问题的读取语句之前将其放置在我的单元测试函数中才有效。 - Michael Behrens