Merge pull request #42 from pomponchik/develop

0.0.31

cantok/tokens/abstract/coroutine_wrapper.py

@@ -1,11 +1,13 @@
+import sys
 import weakref
-from sys import getrefcount
 from typing import Dict, Union, Optional, Any
 from types import TracebackType
 from collections.abc import Coroutine
 from time import sleep as sync_sleep
 from asyncio import sleep as async_sleep
 
+from displayhooks import not_display
+
 
 class WaitCoroutineWrapper(Coroutine):  # type: ignore[type-arg]
     def __init__(self, step: Union[int, float], token_for_wait: 'AbstractToken', token_for_check: 'AbstractToken') -> None:  # type: ignore[name-defined]
@@ -33,7 +35,7 @@ def close(self) -> None:
     @staticmethod
     def sync_wait(step: Union[int, float], flags: Dict[str, bool], token_for_wait: 'AbstractToken', token_for_check: 'AbstractToken', wrapped_coroutine: Coroutine) -> None:  # type: ignore[type-arg, name-defined]
         if not flags.get('used', False):
-            if getrefcount(wrapped_coroutine) < 5:
+            if sys.getrefcount(wrapped_coroutine) < 5:
                 wrapped_coroutine.close()
 
                 while token_for_wait:
@@ -51,3 +53,6 @@ async def async_wait(step: Union[int, float], flags: Dict[str, bool], token_for_
         await async_sleep(0)
 
         token_for_check.check()
+
+
+not_display(WaitCoroutineWrapper)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "cantok"
-version = "0.0.30"
+version = "0.0.31"
 authors = [
   { name="Evgeniy Blinov", email="zheni-b@yandex.ru" },
 ]
@@ -13,6 +13,7 @@ readme = "README.md"
 requires-python = ">=3.7"
 dependencies = [
     'typing_extensions ; python_version <= "3.9"',
+    'displayhooks>=0.0.2',
 ]
 classifiers = [
     "Operating System :: OS Independent",

tests/units/tokens/abstract/test_coroutine_wrapper.py

@@ -0,0 +1,28 @@
+import io
+import sys
+from contextlib import redirect_stdout
+
+import pytest
+
+from cantok import SimpleToken, TimeoutToken, ConditionToken, CounterToken
+
+
+@pytest.mark.parametrize(
+    ['create_value', 'expected_string'],
+    [
+        (lambda: SimpleToken(cancelled=True).wait(), ''),
+        (lambda: TimeoutToken(0.0001).wait(), ''),
+        (lambda: ConditionToken(lambda: True).wait(), ''),
+        (lambda: CounterToken(0).wait(), ''),
+        (lambda: 1, '1\n'),
+        (lambda: 'kek', f'{repr("kek")}\n'),
+    ],
+)
+def test_displayhook_printing_coroutine_wrappers_and_other_objects(create_value, expected_string):
+    buffer = io.StringIO()
+    with redirect_stdout(buffer):
+        sys.displayhook(create_value())
+
+    output = buffer.getvalue()
+
+    assert output == expected_string