Python 3 单元测试中的 ResourceWarning 未关闭套接字
ResourceWarning unclosed socket in Python 3 Unit Test
我正在修改一些代码以在 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)
等热门图书馆也发生了这种情况
我可以忽略警告或 filter it completely. If was my service, I could set the connection: close
header in my response (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=Whosebug'
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__
中使用拆卸逻辑可能会使您的程序不正确或更难推理,因为无法保证何时调用该方法,可能会导致您收到警告。有几种方法可以解决这个问题:
1) 暴露一个关闭session的方法,在测试中调用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=Whosebug'
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) 使用上下文管理器
A context manager 也是一种非常有用的明确定义范围的方法。在前面的示例中,您必须确保 .close()
在每个调用站点都被正确调用,否则您的资源将泄漏。使用上下文管理器,即使在上下文管理器的范围内出现异常,也会自动处理。
在解决方案 1) 的基础上构建,您可以定义额外的魔术方法(__enter__
和 __exit__
),以便您的 class 与 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=Whosebug'
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
和上下文管理器中的任一个或组合使用,并摆脱该警告,并在您的资源中进行更明确的资源管理代码!
如果您不太在意警告,这是最好的解决方案
只需导入 warnings 并在您的驱动程序启动的位置添加此行 -
import warnings
warnings.filterwarnings(action="ignore", message="unclosed", category=ResourceWarning)
我正在修改一些代码以在 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)
等热门图书馆也发生了这种情况
我可以忽略警告或 filter it completely. If was my service, I could set the connection: close
header in my response (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=Whosebug'
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__
中使用拆卸逻辑可能会使您的程序不正确或更难推理,因为无法保证何时调用该方法,可能会导致您收到警告。有几种方法可以解决这个问题:
1) 暴露一个关闭session的方法,在测试中调用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=Whosebug'
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) 使用上下文管理器
A context manager 也是一种非常有用的明确定义范围的方法。在前面的示例中,您必须确保 .close()
在每个调用站点都被正确调用,否则您的资源将泄漏。使用上下文管理器,即使在上下文管理器的范围内出现异常,也会自动处理。
在解决方案 1) 的基础上构建,您可以定义额外的魔术方法(__enter__
和 __exit__
),以便您的 class 与 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=Whosebug'
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
和上下文管理器中的任一个或组合使用,并摆脱该警告,并在您的资源中进行更明确的资源管理代码!
如果您不太在意警告,这是最好的解决方案
只需导入 warnings 并在您的驱动程序启动的位置添加此行 -
import warnings
warnings.filterwarnings(action="ignore", message="unclosed", category=ResourceWarning)