lint

Author: san-tekart

pyproject.toml

@@ -89,8 +89,6 @@ disable_error_code = ["annotation-unchecked", "assignment"]
 warn_return_any = false
 warn_unused_configs = false
 
+exclude = "tests/|setup.py"
+
 ignore_errors = true
-omit = [
-    "tests/*",
-    "setup.py",
-]

src/tsignal/core.py

@@ -238,7 +238,6 @@ def wrapper(self, *args, **kwargs):
                 current_thread = threading.current_thread()
                 if current_thread != self._thread:
                     logger.debug("Executing regular slot from different thread")
-                    # 동기 함수는 loop.call_soon_threadsafe와 Future를 사용
                     future = concurrent.futures.Future()
 
                     def callback():

tests/unit/test_property.py

@@ -1,45 +1,60 @@
-import pytest
+"""
+Test cases for the property pattern.
+"""
+
+# pylint: disable=no-member
+# pylint: disable=unnecessary-lambda
+# pylint: disable=useless-with-lock
+
 import asyncio
 import threading
+import logging
+import pytest
 from tsignal.contrib.extensions.property import t_property
 from tsignal import t_signal, t_with_signals
-import logging
 
 
 logger = logging.getLogger(__name__)
 
 
 @t_with_signals
 class Temperature:
+    """Temperature class for testing"""
+
     def __init__(self):
         super().__init__()
         self._celsius = -273
 
     @t_signal
     def celsius_changed(self):
-        pass
+        """Signal for celsius change"""
 
     @t_property(notify=celsius_changed)
     def celsius(self) -> float:
+        """Getter for celsius"""
         return self._celsius
 
     @celsius.setter
     def celsius(self, value: float):
+        """Setter for celsius"""
         self._celsius = value
 
 
 @t_with_signals
 class ReadOnlyTemperature:
+    """ReadOnlyTemperature class for testing"""
+
     def __init__(self):
         super().__init__()
         self._celsius = 0
 
     @t_signal
     def celsius_changed(self):
-        pass
+        """Signal for celsius change"""
 
     @t_property(notify=celsius_changed)
     def celsius(self) -> float:
+        """Getter for celsius"""
         return self._celsius
 
 
@@ -96,9 +111,7 @@ async def test_property_thread_safety():
     """Test property thread safety and notifications across threads"""
     temp = Temperature()
     received_values = []
-    logger.debug(f"test_property_thread_safety #3")
     task_completed = asyncio.Event()
-    logger.debug(f"test_property_thread_safety #4")
     main_loop = asyncio.get_running_loop()
 
     def background_task():
@@ -122,7 +135,6 @@ async def set_temp():
     thread = threading.Thread(target=background_task)
     thread.start()
 
-    # asyncio.Event의 wait() 사용
     await task_completed.wait()
 
     thread.join()
@@ -138,18 +150,20 @@ async def test_property_multiple_threads():
     received_values = []
     values_lock = threading.Lock()
     threads_lock = threading.Lock()
-    NUM_THREADS = 5
+    num_threads = 5
     task_completed = asyncio.Event()
     threads_done = 0
     main_loop = asyncio.get_running_loop()
 
     def on_celsius_changed(value):
+        """Handler for celsius change"""
         with values_lock:
             received_values.append(value)
 
     temp.celsius_changed.connect(on_celsius_changed)
 
     def background_task(value):
+        """Background task for thread safety testing"""
         nonlocal threads_done
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
@@ -164,31 +178,31 @@ def background_task(value):
             with threading.Lock():
                 nonlocal threads_done
                 threads_done += 1
-                if threads_done == NUM_THREADS:
+                if threads_done == num_threads:
                     main_loop.call_soon_threadsafe(task_completed.set)
 
     threads = [
         threading.Thread(target=background_task, args=(i * 10,))
-        for i in range(NUM_THREADS)
+        for i in range(num_threads)
     ]
 
-    logger.debug("Starting threads")
+    # Start threads
     for thread in threads:
         thread.start()
 
-    logger.debug("Waiting for task_completed event")
+    # Wait for task completion
     try:
         await asyncio.wait_for(task_completed.wait(), timeout=2.0)
     except asyncio.TimeoutError:
         logger.warning("Timeout waiting for threads")
 
-    logger.debug("Joining threads")
+    # Join threads
     for thread in threads:
         thread.join()
 
     await asyncio.sleep(0.2)
 
-    expected_values = set(i * 10 for i in range(NUM_THREADS))
+    expected_values = set(i * 10 for i in range(num_threads))
     received_set = set(received_values)
 
     assert (
@@ -207,10 +221,12 @@ def test_property_exception_handling():
     received_values = []
 
     def handler_with_exception(value):
+        """Handler with exception"""
         received_values.append(value)
         raise ValueError("Test exception")
 
     def normal_handler(value):
+        """Normal handler"""
         received_values.append(value * 2)
 
     # Connect multiple handlers

tests/unit/test_slot.py

@@ -2,12 +2,14 @@
 Test cases for the slot pattern.
 """
 
+# pylint: disable=no-member
+
 import asyncio
 import threading
 import time
+import logging
 import pytest
 from tsignal import t_with_signals, t_slot
-import logging
 
 logger = logging.getLogger(__name__)
 
@@ -65,6 +67,8 @@ async def test_slot_thread_safety():
 
     @t_with_signals
     class ThreadTestReceiver:
+        """Receiver class for thread safety testing"""
+
         def __init__(self):
             super().__init__()
             self.received_value = None
@@ -73,13 +77,15 @@ def __init__(self):
 
         @t_slot
         async def async_slot(self, value):
+            """Async slot for thread safety testing"""
             self.execution_thread = threading.current_thread()
             await asyncio.sleep(0.1)  # work simulation with sleep
             self.received_value = value
             self.received_count += 1
 
         @t_slot
         def sync_slot(self, value):
+            """Sync slot for thread safety testing"""
             self.execution_thread = threading.current_thread()
             time.sleep(0.1)  # work simulation with sleep
             self.received_value = value
@@ -91,6 +97,7 @@ def sync_slot(self, value):
     initial_values = {"value": None, "count": 0}  # save initial values
 
     def background_task():
+        """Background task for thread safety testing"""
         try:
             # Before async_slot call
             initial_values["value"] = receiver.received_value
@@ -121,6 +128,7 @@ def background_task():
             task_completed.set()
 
     async def run_test():
+        """Run test for thread safety"""
         thread = threading.Thread(target=background_task)
         thread.start()
 
@@ -132,20 +140,20 @@ async def run_test():
         # Cleanup
         pending = asyncio.all_tasks(receiver._loop)
         if pending:
-            logger.debug(f"Cleaning up {len(pending)} pending tasks")
+            logger.debug("Cleaning up %d pending tasks", len(pending))
             for task in pending:
                 if "test_slot_thread_safety" in str(task.get_coro()):
-                    logger.debug(f"Skipping test function task: {task}")
+                    logger.debug("Skipping test function task: %s", task)
                 else:
-                    logger.debug(f"Found application task: {task}")
+                    logger.debug("Found application task: %s", task)
                     try:
                         await asyncio.gather(task, return_exceptions=True)
                     except Exception as e:
-                        logger.error(f"Error during cleanup: {e}")
+                        logger.error("Error during cleanup: %s", e)
         else:
             logger.debug("No pending tasks to clean up")
 
     try:
         await run_test()
     except Exception as e:
-        logger.error(f"Error in test: {e}")
+        logger.error("Error in test: %s", e)