代码之家  ›  专栏  ›  技术社区  ›  j12y

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

  •  65
  • j12y  · 技术社区  · 7 年前

    我正在修改一些代码,以便在 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 :

    应用程序。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__() 导致此类警告。

    谢谢你的帮助。

    2 回复  |  直到 6 年前
        1
  •  28
  •   Samuel Dion-Girardeau    6 年前

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

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

    unittest tearDown 方法允许您定义一些将在每次测试后运行的代码。即使测试失败或出现异常,使用这个钩子关闭会话也会起作用,这很好。

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

    A. context manager 也是明确定义某事物范围的非常有用的方法。在上一个示例中,您必须确保 .close() 在每个呼叫站点上正确调用,否则您的资源将泄漏。使用上下文管理器,即使在上下文管理器的范围内存在异常,也会自动处理此问题。

    基于解决方案1),您可以定义额外的魔术方法( __enter__ __exit__ )这样你的班级就可以 with 陈述

    注意:这里的好处是,这段代码还支持在解决方案1)中使用显式 .关闭() ,如果上下文管理器由于某种原因而不方便,这可能很有用。

    应用程序。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 / 拆卸 和上下文管理器,消除该警告,并在代码中进行更明确的资源管理!

        2
  •  21
  •   Uri    4 年前

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

    仅导入 警告 并将这一行添加到驱动程序启动的位置-

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