Reputation: 452
Consider the function in a module.
async def _get_user_mapping():
async with aiofiles.open('/etc/subuid') as f:
async for line in f:
print(line)
And a test class
from atestfile import _get_user_mapping
class TestFS(fake_filesystem_unittest.TestCase):
def setUp(self):
self.a_required_class_parameter = True
self.setUpPyfakefs()
for file_path in ['/etc/subuid']:
self.fs.create_file(file_path, contents = """testuser:100000:65536
testuser2:165536:65536""")
def test(self):
if self.a_required_class_parameter:
asyncio.get_event_loop().run_until_complete(_get_user_mapping())
In this example, the test function should set pyfakefs up to provide a fake file. Unfortunately aiofiles doesn't have access to the file and the printed output is the real file on the system.
Does anyone know how I can patch the aiofiles event loop to use pyfakefs as a fake filesystem? In testing I've found the following snippet using a library called pytest-aiofiles
(sounds like what I need, right?) but the example they show:
@pytest.mark.asyncio
async def test_stuff(self):
filename = 'test'
etc.....
If I add the mark.asyncio
decorator to the test
class method, the imported function doesn't have access to the generated fake file in the setUp
method.
I assume I'm missing something simple so really this can all be broken down into the simple question: How the heck do I test this?
Thanks!
Upvotes: 1
Views: 1126
Reputation: 1
Here is how I managed to mock aiofiles used in a file let's say called fofo.py:
with patch("fofo.aiofiles") as patched_aiofiles:
instance = AsyncMock()
attrs = {
"write": AsyncMock(),
"flush": AsyncMock(),
"close": AsyncMock(),
}
instance.configure_mock(**attrs)
attrs = {"open": AsyncMock(return_value=instance)}
patched_aiofiles.configure_mock(**attrs)
Upvotes: 0
Reputation: 1305
You could create a class to patch 'aiofiles' altogether?
class FakeAiofiles
def open():
# return the open fake file you created in the test
then in your test, use:
def test(self):
with mock.patch('atestfile.aiofiles', return_value=FakeAiofiles)
if self.a_required_class_parameter:
asyncio.get_event_loop().run_until_complete(_get_user_mapping())
Upvotes: 0