इस यूनिट टेस्ट में मैं यह जांचना चाहूंगा कि क्या डिवाइस बनाया गया है और भविष्य में एक्सपायरी डेट 7 दिन है।

database.py

import databases
database = databases.Database(settings.sqlalchemy_database_uri)

अध्याय परीक्षा:

from database.database import database

def test_successful_register_expiry_set_to_seven_days():
    response = client.post(
        "/register/",
        headers={},
        json={"device_id": "u1"},
    )
    assert response.status_code == 201
    query = device.select(whereclause=device.c.id == "u1")
    d = database.fetch_one(query)
    assert d.expires_at == datetime.utcnow().replace(microsecond=0) + timedelta(days=7)

क्योंकि d एक कोरटाइन ऑब्जेक्ट है, यह संदेश के साथ विफल हो जाता है:

AttributeError: 'coroutine' object has no attribute 'expires_at'

और मैं एक इकाई परीक्षण के अंदर await का उपयोग नहीं कर सकता।

d = await database.fetch_one(query)

मुझे क्या याद आ रहा है, कृपया?

2
Houman 3 फरवरी 2021, 00:41

1 उत्तर

सबसे बढ़िया उत्तर

वैसे यह कभी भी प्रतीक्षित नहीं है, आपका कोड शेड्यूलर में जाने वाले सेमाफोर से पहले कोरआउटिन लौटाता है।

यदि आप एक एसिंक्रोनस ड्राइवर का उपयोग कर रहे हैं, तो आपको इसकी प्रतीक्षा करने की आवश्यकता है। क्या इसके लिए कोई उपाय हैं? हां।

आप asyncio.run(awaitable) का उपयोग किसी ईवेंट लूप के अंदर coroutine चलाने के लिए कर सकते हैं।

import asyncio

d = asyncio.run(database.fetch_one(query))

यदि आपके पास वर्तमान में चल रहा ईवेंट लूप है, तो आप इसके बजाय उस ईवेंट लूप का उपयोग करना चाह सकते हैं। आप इसे asyncio.get_event_loop() द्वारा प्राप्त कर सकते हैं, जो चल रहे लूप के अंदर फ़ंक्शन चलाएगा।

import asyncio

asyncio.get_event_loop().run_until_complete(database.fetch_one(query))

आप @pytest.mark.asyncio डेकोरेटर का भी उपयोग कर सकते हैं (देखें दस्तावेज़ीकरण)।

@pytest.mark.asyncio
async def dummy():
    await some_awaitable()
0
Yagiz Degirmenci 3 फरवरी 2021, 12:58
दुख की बात है कि मुझे यह त्रुटि मिलती है: RuntimeError: Task <Task pending name='Task-4' coro=<Database.fetch_one() running at /Users/houmie/.pyenv/versions/v391/lib/python3.9/site-packages/databases/core.py:146> cb=[_run_until_complete_cb() at /Users/houmie/.pyenv/versions/3.9.1/lib/python3.9/asyncio/base_events.py:184]> got Future <Future pending> attached to a different loop
 – 
Houman
3 फरवरी 2021, 02:27
हम्म, मुझे लगता है कि डेटाबेस इवेंट लूप बनाता है और इसके अंदर निष्पादित करने का प्रयास करता है। क्या आप रनिंग लूप प्राप्त करने और इसके बजाय इसका उपयोग करने का प्रयास कर सकते हैं; asyncio.get_event_loop().run_until_complete(database.fetch_one(query)). यदि यह काम नहीं करता है, तो आप pytest.mark_asyncio भी आजमा सकते हैं।
 – 
Yagiz Degirmenci
3 फरवरी 2021, 02:48
@pytest.mark.asyncio अभी भी RuntimeError: This event loop is already running फेंकता है। लेकिन दूसरा उपाय काम कर गया। आप अपना उत्तर अपडेट करना चाह सकते हैं। धन्यवाद (यदि आपके पास @pytest.mark.asyncio के साथ इसे और सरल बनाने का विचार है तो मैं जानना चाहूंगा। :)
 – 
Houman
3 फरवरी 2021, 10:28
मैंने इसे दस्तावेज़ीकरण में पाया। और हाँ यह काम करता है! fastapi.tiangolo.com/advanced/async-tests/#example
 – 
Houman
3 फरवरी 2021, 12:49
महान! मुझे खुशी है कि दोनों ने काम किया। मैं वैकल्पिक समाधानों के साथ उत्तर अपडेट करूंगा।
 – 
Yagiz Degirmenci
3 फरवरी 2021, 12:58