问题描述

我正在修改一些代码以兼容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这样的流行图书馆也出现了这种情况。 我可以忽略警告或完全忽略filter it。如果是我的服务，我可以在我的响应中设置connection: close头(link)。

下面的示例显示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__()导致此类警告。

谢谢您的帮助。

推荐答案

在__del__中使用tearDown逻辑可能会使您的程序不正确或更难推理，因为无法保证何时调用该方法，这可能会导致您收到警告。有几种方法可以解决这个问题：

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)使用上下文管理器

在解决方案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和上下文管理器中的任何一个或组合，并消除该警告，并在代码中进行更显式的资源管理！

这篇关于在Python3单元测试中警告未关闭的套接字的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！