I'm modifying some code to be compatible between Python 2
and Python 3
, but have observed a warning in unit test output.
/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)
A little research determined this was also happening from popular libraries like requests and boto3.
I could ignore the warning or filter it completely. If was my service, I could set the connection: close
header in my response (link).
Here's an example that exhibits the warning in 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()
Is there a better / correct way to manage the session so that it gets explicitly closed and not rely on __del__()
to result in this sort of warning.
Thanks for any help.
Having the teardown logic in __del__
can make your program incorrect or harder to reason about, because there is no guarantee on when that method will get called, potentially leading to the warning you got. There are a couple of ways to address this:
tearDown
unittest
's tearDown
method allows you to define some code that will be run after each test. Using this hook to close the session will work even if the test fails or has an exception, which is nice.
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()
A context manager is also a very useful way to explicitly define the scope of something. In the previous example, you have to make sure .close()
is called correctly at every call site, otherwise your resources will leak. With a context manager, this is handled automatically even if there is an exception within the scope of the context manager.
Building on top of solution 1), you can define extra magic methods (__enter__
and __exit__
) so that your class works with the with
statement.
Note: The nice thing here is that this code also supports the usage in solution 1), with explicit .close()
, which can be useful if a context manager was inconvenient for some reason.
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()
Depending on what you need, you can use either, or a combination of setUp
/tearDown
and context manager, and get rid of that warning, plus having more explicit resource management in your code!
This is the best solution if you are not much concern about warnings
Just import warnings and add this line where your driver is initiating -
import warnings
warnings.filterwarnings(action="ignore", message="unclosed", category=ResourceWarning)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With