I hope I am on the right place.
I want to write some test from my async view and I follow the documentation but it keeps failing.
this is my view:
```class Session(View):
""" Asynchronously get a session for a system."""
@classonlymethod
def as_view(cls, **initkwargs):
view = super().as_view(**initkwargs)
view._is_coroutine = asyncio.coroutines._is_coroutine # pylint: disable=protected-access
return view
async def get(self, request, **kwargs):
""" Async get list of session between two dates """
end_period: str = request.GET['end']
start_period: str = request.GET['start']
gcloud_id: str = request.GET['gcloud']
offset: int = int(request.GET.get('offset', 10))
page: int = int(request.GET.get('page', 1))
system_uuid: str = kwargs['uuid']
cache_id: str = f"session_{system_uuid}_{start_period}_{end_period}"
if cache.get(cache_id):
log_data = cache.get(cache_id)
else:
session_service: SessionService = SessionService(system_uuid, gcloud_id, start_period, end_period)
log_data: List[Dict[str, Union[str, pd.Timestamp]]] = await session_service.get_session()
cache.set(cache_id, log_data, timeout=60 * 60)
nb_page: int = math.ceil((len(log_data) / offset))
pagination: Pagination = Pagination(offset, nb_page)
pagination.page = page
return JsonResponse({'logs': log_data[pagination.start_limit:pagination.end_limit], 'nb_page': nb_page,
'page': pagination.page},
safe=False)```
""" Asynchronously get a session for a system."""
@classonlymethod
def as_view(cls, **initkwargs):
view = super().as_view(**initkwargs)
view._is_coroutine = asyncio.coroutines._is_coroutine # pylint: disable=protected-access
return view
async def get(self, request, **kwargs):
""" Async get list of session between two dates """
end_period: str = request.GET['end']
start_period: str = request.GET['start']
gcloud_id: str = request.GET['gcloud']
offset: int = int(request.GET.get('offset', 10))
page: int = int(request.GET.get('page', 1))
system_uuid: str = kwargs['uuid']
cache_id: str = f"session_{system_uuid}_{start_period}_{end_period}"
if cache.get(cache_id):
log_data = cache.get(cache_id)
else:
session_service: SessionService = SessionService(system_uuid, gcloud_id, start_period, end_period)
log_data: List[Dict[str, Union[str, pd.Timestamp]]] = await session_service.get_session()
cache.set(cache_id, log_data, timeout=60 * 60)
nb_page: int = math.ceil((len(log_data) / offset))
pagination: Pagination = Pagination(offset, nb_page)
pagination.page = page
return JsonResponse({'logs': log_data[pagination.start_limit:pagination.end_limit], 'nb_page': nb_page,
'page': pagination.page},
safe=False)```
the service called in here is actually getting some data from my postgres database.
here is my test, basically just following the doc:
```class SystemViewsTest(SystemSetUpTest):
async def test_get_session_detail(self):
uuid: str = '2265d534-5a62-426a-bc44-e17ebddabfb3'
console_gcloud_id = 'console-7ccbe2e157d6'
start_period = '2021-05-06'
end_period = '2021-05-07'
response = await self.async_client.get(
f'/v1/sessions/{uuid}?end={end_period}&start={start_period}&gcloud={console_gcloud_id}',
)
print(response)
self.assertEqual(response.status_code, 200)```
async def test_get_session_detail(self):
uuid: str = '2265d534-5a62-426a-bc44-e17ebddabfb3'
console_gcloud_id = 'console-7ccbe2e157d6'
start_period = '2021-05-06'
end_period = '2021-05-07'
response = await self.async_client.get(
f'/v1/sessions/{uuid}?end={end_period}&start={start_period}&gcloud={console_gcloud_id}',
)
print(response)
self.assertEqual(response.status_code, 200)```
SystemSetUpTest is just a class I use to create System data (using the setUp method) for my test as it is related to many other models, it inherits from django.test import TestCase
but when I want to run my test I have this infinite stack trace saying that my db connection already closed, that another session is alredy using my db and something about the generator not stopping after throw()..
======================================================================
ERROR: test_get_session_detail (systems.tests.test_views.SystemViewsTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/base.py", line 237, in _cursor
return self._prepare_cursor(self.create_cursor(name))
File "/usr/local/lib/python3.8/dist-packages/django/utils/asyncio.py", line 26, in inner
return func(*args, **kwargs)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/postgresql/base.py", line 236, in create_cursor
cursor = self.connection.cursor()
psycopg2.InterfaceError: connection already closed
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/usr/local/lib/python3.8/dist-packages/asgiref/sync.py", line 223, in __call__
return call_result.result()
File "/usr/lib/python3.8/concurrent/futures/_base.py", line 437, in result
return self.__get_result()
File "/usr/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result
raise self._exception
File "/usr/local/lib/python3.8/dist-packages/asgiref/sync.py", line 292, in main_wrap
result = await self.awaitable(*args, **kwargs)
File "/app/systems/tests/test_views.py", line 15, in test_get_session_detail
response = await self.async_client.get(
File "/usr/local/lib/python3.8/dist-packages/django/test/client.py", line 908, in request
self.check_exception(response)
File "/usr/local/lib/python3.8/dist-packages/django/test/client.py", line 580, in check_exception
raise exc_value
File "/usr/local/lib/python3.8/dist-packages/asgiref/sync.py", line 482, in thread_handler
raise exc_info[1]
File "/usr/local/lib/python3.8/dist-packages/django/core/handlers/exception.py", line 38, in inner
response = await get_response(request)
File "/usr/local/lib/python3.8/dist-packages/django/core/handlers/base.py", line 233, in _get_response_async
response = await wrapped_callback(request, *callback_args, **callback_kwargs)
File "/usr/local/lib/python3.8/dist-packages/sentry_sdk/integrations/django/asgi.py", line 94, in sentry_wrapped_callback
return await callback(request, *args, **kwargs)
File "/app/systems/views/shift.py", line 114, in get
log_data: List[Dict[str, Union[str, pd.Timestamp]]] = await session_service.get_session()
File "/app/systems/services/session.py", line 73, in get_session
await self.__vision_system_gcloud_id)
File "/usr/local/lib/python3.8/dist-packages/async_property/base.py", line 37, in get_value
return await self._fget(instance)
File "/app/systems/services/session.py", line 57, in __vision_system_gcloud_id
data = await self.__vision_data()
File "/usr/local/lib/python3.8/dist-packages/asgiref/sync.py", line 444, in __call__
ret = await asyncio.wait_for(future, timeout=None)
File "/usr/lib/python3.8/asyncio/tasks.py", line 455, in wait_for
return await fut
File "/usr/local/lib/python3.8/dist-packages/asgiref/current_thread_executor.py", line 22, in run
result = self.fn(*self.args, **self.kwargs)
File "/usr/local/lib/python3.8/dist-packages/channels/db.py", line 13, in thread_handler
return super().thread_handler(loop, *args, **kwargs)
File "/usr/local/lib/python3.8/dist-packages/asgiref/sync.py", line 486, in thread_handler
return func(*args, **kwargs)
File "/app/systems/services/session.py", line 38, in __vision_data
return System.objects.get_vision_conf_data(self.system_uuid)
File "/app/systems/models/manager/manager.py", line 130, in get_vision_conf_data
gcloud: List[str] = [data[0] for data in visions]
File "/usr/local/lib/python3.8/dist-packages/django/db/models/query.py", line 280, in __iter__
self._fetch_all()
File "/usr/local/lib/python3.8/dist-packages/django/db/models/query.py", line 1324, in _fetch_all
self._result_cache = list(self._iterable_class(self))
File "/usr/local/lib/python3.8/dist-packages/django/db/models/query.py", line 140, in __iter__
return compiler.results_iter(tuple_expected=True, chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size)
File "/usr/local/lib/python3.8/dist-packages/django/db/models/sql/compiler.py", line 1124, in results_iter
results = self.execute_sql(MULTI, chunked_fetch=chunked_fetch, chunk_size=chunk_size)
File "/usr/local/lib/python3.8/dist-packages/django/db/models/sql/compiler.py", line 1167, in execute_sql
cursor = self.connection.cursor()
File "/usr/local/lib/python3.8/dist-packages/django/utils/asyncio.py", line 26, in inner
return func(*args, **kwargs)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/base.py", line 259, in cursor
return self._cursor()
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/base.py", line 237, in _cursor
return self._prepare_cursor(self.create_cursor(name))
File "/usr/local/lib/python3.8/dist-packages/django/db/utils.py", line 90, in __exit__
raise dj_exc_value.with_traceback(traceback) from exc_value
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/base.py", line 237, in _cursor
return self._prepare_cursor(self.create_cursor(name))
File "/usr/local/lib/python3.8/dist-packages/django/utils/asyncio.py", line 26, in inner
return func(*args, **kwargs)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/postgresql/base.py", line 236, in create_cursor
cursor = self.connection.cursor()
django.db.utils.InterfaceError: connection already closed
----------------------------------------------------------------------
Ran 1 test in 0.043s
FAILED (errors=1)
Destroying test database for alias 'default'...
/usr/local/lib/python3.8/dist-packages/django/db/backends/postgresql/base.py:304: RuntimeWarning: Normally Django will use a connection to the 'postgres' database to avoid running initialization queries against the production database when it's not needed (for example, when running tests). Django was unable to create a connection to the 'postgres' database and will use the first PostgreSQL database instead.
warnings.warn(
Traceback (most recent call last):
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/utils.py", line 82, in _execute
return self.cursor.execute(sql)
psycopg2.errors.ObjectInUse: database "test_django" is being accessed by other users
DETAIL: There is 1 other session using the database.
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/postgresql/base.py", line 302, in _nodb_cursor
yield cursor
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/creation.py", line 298, in _destroy_test_db
cursor.execute("DROP DATABASE %s"
File "/usr/local/lib/python3.8/dist-packages/sentry_sdk/integrations/django/__init__.py", line 500, in execute
return real_execute(self, sql, params)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/utils.py", line 66, in execute
return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/utils.py", line 75, in _execute_with_wrappers
return executor(sql, params, many, context)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/utils.py", line 84, in _execute
return self.cursor.execute(sql, params)
File "/usr/local/lib/python3.8/dist-packages/django/db/utils.py", line 90, in __exit__
raise dj_exc_value.with_traceback(traceback) from exc_value
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/utils.py", line 82, in _execute
return self.cursor.execute(sql)
django.db.utils.OperationalError: database "test_django" is being accessed by other users
DETAIL: There is 1 other session using the database.
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "manage.py", line 17, in <module>
execute_from_command_line(sys.argv)
File "/usr/local/lib/python3.8/dist-packages/django/core/management/__init__.py", line 419, in execute_from_command_line
utility.execute()
File "/usr/local/lib/python3.8/dist-packages/django/core/management/__init__.py", line 413, in execute
self.fetch_command(subcommand).run_from_argv(self.argv)
File "/usr/local/lib/python3.8/dist-packages/django/core/management/commands/test.py", line 23, in run_from_argv
super().run_from_argv(argv)
File "/usr/local/lib/python3.8/dist-packages/django/core/management/base.py", line 354, in run_from_argv
self.execute(*args, **cmd_options)
File "/usr/local/lib/python3.8/dist-packages/django/core/management/base.py", line 398, in execute
output = self.handle(*args, **options)
File "/usr/local/lib/python3.8/dist-packages/django/core/management/commands/test.py", line 55, in handle
failures = test_runner.run_tests(test_labels)
File "/usr/local/lib/python3.8/dist-packages/django/test/runner.py", line 736, in run_tests
self.teardown_databases(old_config)
File "/usr/local/lib/python3.8/dist-packages/django/test/runner.py", line 674, in teardown_databases
_teardown_databases(
File "/usr/local/lib/python3.8/dist-packages/django/test/utils.py", line 313, in teardown_databases
connection.creation.destroy_test_db(old_name, verbosity, keepdb)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/creation.py", line 282, in destroy_test_db
self._destroy_test_db(test_database_name, verbosity)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/creation.py", line 298, in _destroy_test_db
cursor.execute("DROP DATABASE %s"
File "/usr/lib/python3.8/contextlib.py", line 162, in __exit__
raise RuntimeError("generator didn't stop after throw()")
RuntimeError: generator didn't stop after throw()
I am very blocked by this, if you could give me some help I would be so grateful.
Thanks.
Julie
You received this message because you are subscribed to the Google Groups "Django users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to django-users+unsubscribe@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/django-users/9a0754ee-e281-4363-a516-021983332acfn%40googlegroups.com.
No comments:
Post a Comment