Feat: add tests to remaining useful funcs and make read_last_n_lines pass tests

Author: rf20008

helpful_modules/threads_or_useful_funcs.py

@@ -341,16 +341,18 @@ async def read_last_n_lines(filename, n):
         position = await f.tell()
         lines = []
         current_line = []
-
         # Read backwards until we find the last n lines
         while position >= 0 and len(lines) < n:
             await f.seek(position)
-            char = await f.read(1)
-
+            char = (await f.read(1)).decode()
+            #print(char, end='')
             # Check for newline character
-            if char == b"\n" and current_line:
-                # Store the completed line
-                lines.append(current_line[::-1].decode())
+            #print(f"{current_line=} {char=} {lines=} ")
+            if char == "\n":
+                if len("".join(current_line)) > 0:
+                    # Store the completed line
+
+                    lines.append("".join(current_line[::-1]))
                 current_line = []
             else:
                 current_line.append(char)
@@ -359,7 +361,7 @@ async def read_last_n_lines(filename, n):
 
         # Capture the last line if it does not end with a newline
         if current_line:
-            lines.append(current_line[::-1].decode())
+            lines.append("".join(current_line[::-1]))
 
         # Reverse the lines to get them in the correct order
         return lines[::-1][:n]

tests/test_helpful_modules/test_threads_or_useful_funcs.py

@@ -28,6 +28,9 @@
 import aiofiles
 import pyfakefs.fake_filesystem_unittest
 
+from pyfakefs.fake_filesystem_unittest import TestCase
+
+from helpful_modules.threads_or_useful_funcs import read_last_n_lines
 from helpful_modules import threads_or_useful_funcs
 from tests.mockable_aiofiles import MockableAioFiles
 from tests.utils import generate_many_randoms
@@ -354,7 +357,94 @@ def test_secure_fisher_yates_shuffle(self, mock_randbelow):
         self.assertNotEqual(my_list, shuffled_list)
         self.assertCountEqual(my_list, shuffled_list)
         mock_randbelow.assert_called_with(2)
+import asyncio
+import concurrent.futures
+import unittest
+
+from helpful_modules.threads_or_useful_funcs import async_wait_for_future
+
+
+class TestAsyncWaitForFuture(unittest.IsolatedAsyncioTestCase):
+
+    async def test_asyncio_future_returns_result(self):
+        future = asyncio.Future()
+        future.set_result("hello")
+
+        result = await async_wait_for_future(future)
+
+        self.assertEqual(result, "hello")
+
+    async def test_asyncio_future_timeout(self):
+        future = asyncio.Future()
+
+        with self.assertRaises(asyncio.TimeoutError):
+            await async_wait_for_future(future, timeout=0.01)
+
+    async def test_concurrent_future_returns_result(self):
+        future = concurrent.futures.Future()
+        future.set_result(123)
+
+        result = await async_wait_for_future(future, interval=0.001)
+
+        self.assertEqual(result, 123)
+
+    async def test_invalid_future_type(self):
+        with self.assertRaises(TypeError):
+            await async_wait_for_future("not a future")
+import asyncio
+import concurrent.futures
+import unittest
+
+from helpful_modules.threads_or_useful_funcs import async_wait_for_future
+
+
+class TestAsyncWaitForFuture(unittest.IsolatedAsyncioTestCase):
 
+    async def test_asyncio_future_returns_result(self):
+        future = asyncio.Future()
+        future.set_result("hello")
+
+        result = await async_wait_for_future(future)
+
+        self.assertEqual(result, "hello")
+
+    async def test_asyncio_future_timeout(self):
+        future = asyncio.Future()
+
+        with self.assertRaises(asyncio.TimeoutError):
+            await async_wait_for_future(future, timeout=0.01)
+
+    async def test_concurrent_future_returns_result(self):
+        future = concurrent.futures.Future()
+        future.set_result(123)
+
+        result = await async_wait_for_future(future, interval=0.001)
+
+        self.assertEqual(result, 123)
+
+    async def test_invalid_future_type(self):
+        with self.assertRaises(TypeError):
+            await async_wait_for_future("not a future")
+
+
+
+class TestReadLastNLines(TestCase, unittest.IsolatedAsyncioTestCase):
+
+    def setUp(self):
+        self.setUpPyfakefs()
+
+    async def test_reads_last_lines(self):
+        self.fs.create_file(
+            "/tmp/test.txt",
+            contents="line1\nline2\nline3\nline4\n"
+        )
+
+        result = await read_last_n_lines("/tmp/test.txt", 2)
+
+        self.assertEqual(
+            result,
+            ["line3", "line4"]
+        )
 
 if __name__ == "__main__":
     unittest.main()