Repository: list-family/ansq (language: python)

Issue:
Unhandled exception on invalid `connection.fin()`
#### Environment Versions
1. Python version: `Python 3.8.10`
2. ansq version: `0.1.0`

#### Steps to reproduce

1. Run the following snipped:
```python
import ansq
import asyncio


async def main():
    nsq = await ansq.open_connection()
    await nsq.fin('invalid')
    await asyncio.sleep(0.2)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
```

2. Exception:
```python
2022-09-12 18:18:37,770 - ERROR - asyncio: Task exception was never retrieved
future: <Task finished name='Task-2' coro=<NSQConnection._read_data_task() done, defined at /Users/albert/Projects/ansq/ansq/tcp/connection.py:355> exception=IndexError('pop from an empty deque')>
Traceback (most recent call last):
  File "/Users/albert/Projects/ansq/ansq/tcp/connection.py", line 373, in _read_data_task
    await self._read_buffer()
  File "/Users/albert/Projects/ansq/ansq/tcp/connection.py", line 448, in _read_buffer
    is_continue = await self._parse_data()
  File "/Users/albert/Projects/ansq/ansq/tcp/connection.py", line 408, in _parse_data
    future, callback = self._cmd_waiters.popleft()
IndexError: pop from an empty deque
```


Discussion / hints:
Additional context. It looks like `ansq` doesn't handle properly  error responses from fin/req/touch/rdy commands which [do not have successful responses](https://github.com/nsqio/nsq/issues/320#issuecomment-36290652) by design. That's why it doesn't append a command waiter on 244-245:

https://github.com/list-family/ansq/blob/1b9d26558fc023b0ed69fb6b3da146ec150c0351/ansq/tcp/connection.py#L236-L247


Fix patch:
diff --git a/ansq/tcp/connection.py b/ansq/tcp/connection.py
index 80ba9c2..b489cd6 100644
--- a/ansq/tcp/connection.py
+++ b/ansq/tcp/connection.py
@@ -411,8 +411,18 @@ async def _parse_data(self) -> bool:
             await self._on_message_hook(response)
             return True
 
-        future: asyncio.Future
-        callback: Optional[Callable[[TCPResponse], Any]]
+        # commands like RDY/FIN/REQ/TOUCH do not return a success response, however,
+        # they might return an error
+        if response.is_error and not self._cmd_waiters:
+            self.logger.error(response.text)
+            return False
+
+        # non-error responses must have a command waiter, otherwise,
+        # it's more likely a bug
+        if not self._cmd_waiters:  # pragma: no cover
+            self.logger.error("Unexpected response: %s", truncate_text(response.text))
+            return False
+
         future, callback = self._cmd_waiters.popleft()
 
         if response.is_response:
diff --git a/ansq/tcp/types/response_schemas.py b/ansq/tcp/types/response_schemas.py
index b0e8b34..ee67aa1 100644
--- a/ansq/tcp/types/response_schemas.py
+++ b/ansq/tcp/types/response_schemas.py
@@ -93,3 +93,7 @@ def __repr__(self) -> str:
 
     def __bool__(self) -> bool:
         return False
+
+    @property
+    def text(self) -> str:
+        return f"[{self.code}] {super().text}"


Test patch:
diff --git a/tests/test_connection.py b/tests/test_connection.py
index cf7678e..cf165b5 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -1,6 +1,7 @@
 import pytest
 
 from ansq import ConnectionFeatures, ConnectionOptions, open_connection
+from ansq.tcp.types import NSQCommands
 
 
 @pytest.mark.asyncio
@@ -86,3 +87,19 @@ async def test_invalid_kwarg(nsqd):
         TypeError, match="got an unexpected keyword argument: 'invalid_kwarg'"
     ):
         await open_connection(invalid_kwarg=1)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "cmd", (NSQCommands.RDY, NSQCommands.FIN, NSQCommands.TOUCH, NSQCommands.REQ)
+)
+async def test_errors_from_commands_without_responses(nsqd, wait_for, cmd, caplog):
+    nsq = await open_connection()
+
+    response = await nsq.execute(cmd)
+    await wait_for(lambda: caplog.messages)
+    await nsq.close()
+
+    expected_log = f"[E_INVALID] cannot {cmd.decode('utf8')} in current state"
+    assert expected_log in caplog.text
+    assert response is None