Add enhanced + missing test cases

Author: subhashb

README.md

@@ -33,4 +33,8 @@ mdb.write("your_stream_name", "your_message_type", {"data": "value"})
 # Read a message
 message = mdb.read_last_message("your_stream_name")
 print(message)
-```
+```
+
+## License
+
+The Postgres Message Store is released under the [MIT License](https://github.com/subhashb/message-db-py/blob/main/LICENSE).

src/message_db/client.py

@@ -55,7 +55,7 @@ def _write(
         data: Dict[str, Any],
         metadata: Dict[str, Any] | None = None,
         expected_version: int | None = None,
-    ) -> int:
+    ) -> int | None:
         try:
             with connection.cursor(cursor_factory=RealDictCursor) as cursor:
                 cursor.execute(
@@ -74,8 +74,6 @@ def _write(
                 )
 
                 result = cursor.fetchone()
-                if result is None:
-                    raise ValueError("No result returned from the database operation.")
         except Exception as exc:
             raise ValueError(
                 f"{getattr(exc, 'pgcode')}-{getattr(exc, 'pgerror').splitlines()[0]}"
@@ -90,7 +88,7 @@ def write(
         data: Dict,
         metadata: Dict | None = None,
         expected_version: int | None = None,
-    ) -> int:
+    ) -> int | None:
         conn = self.connection_pool.get_connection()
 
         try:
@@ -105,7 +103,7 @@ def write(
 
     def write_batch(
         self, stream_name, data, expected_version: int | None = None
-    ) -> int:
+    ) -> int | None:
         conn = self.connection_pool.get_connection()
 
         try:

tests/test_client.py

@@ -1,3 +1,7 @@
+from unittest.mock import patch
+
+from psycopg2 import OperationalError
+
 from message_db.client import MessageDB
 from message_db.connection import ConnectionPool
 
@@ -18,3 +22,14 @@ def test_client_construction_from_args(self):
 
         assert isinstance(store, MessageDB)
         assert isinstance(store.connection_pool, ConnectionPool)
+
+    def test_reconnection_after_failure(self, client):
+        with patch(
+            "psycopg2.connect", side_effect=[OperationalError("Connection lost"), None]
+        ):
+            try:
+                client.read("testStream-123")
+            except OperationalError:
+                # Retry reading after catching the first failure
+                messages = client.read("testStream-123")
+                assert messages is not None

tests/test_connection.py

@@ -1,9 +1,14 @@
+from unittest.mock import patch
+
 import pytest
 from psycopg2 import OperationalError, ProgrammingError
 from psycopg2.extensions import TRANSACTION_STATUS_ACTIVE
+from psycopg2.pool import PoolError
 
 from message_db.connection import ConnectionPool
 
+CONNECT_URL = "postgresql://message_store@localhost:5432/message_store"
+
 
 def test_constructing_connection_pool_from_url(pool):
     assert pool is not None
@@ -19,14 +24,14 @@ def test_error_on_invalid_url():
 
 def test_error_on_invalid_role():
     with pytest.raises(OperationalError) as exc:
-        ConnectionPool("postgresql://foo@localhost:5432/postgres")
+        ConnectionPool("postgresql://foo@localhost:5432/message_store")
 
     assert 'role "foo" does not exist' in exc.value.args[0]
 
 
 def test_error_on_invalid_max_connections():
     with pytest.raises(ValueError):
-        ConnectionPool("postgresql://foo@localhost:5432/postgres", max_connections=-1)
+        ConnectionPool(CONNECT_URL, max_connections=-1)
 
 
 def test_retrieving_connection_from_pool(pool):
@@ -46,3 +51,59 @@ def test_releasing_a_connection(pool):
 
     pool.release(conn)
     assert len(pool._connection_pool._used) == used_count - 1
+
+
+def test_connection_pool_initialization():
+    """Test the connection pool initialization with different parameters."""
+    pool = ConnectionPool(CONNECT_URL, max_connections=10)
+    assert pool.max_connections == 10
+    assert pool._connection_pool.maxconn == 10
+
+
+def test_multiple_simultaneous_connections():
+    """Test handling multiple simultaneous connections within max limit."""
+    pool = ConnectionPool(CONNECT_URL, max_connections=5)
+    connections = [pool.get_connection() for _ in range(5)]
+    assert len(pool._connection_pool._used) == 5
+    # Clean up
+    for conn in connections:
+        pool.release(conn)
+
+
+def test_exceeding_connection_limit():
+    """Test behavior when more connections than max are requested."""
+    pool = ConnectionPool(CONNECT_URL, max_connections=2)
+    conn1 = pool.get_connection()
+    conn2 = pool.get_connection()
+    with pytest.raises(Exception) as exc:
+        pool.get_connection()
+    assert "connection pool exhausted" in str(exc.value)
+    # Clean up
+    pool.release(conn1)
+    pool.release(conn2)
+
+
+def test_release_invalid_connection():
+    """Test releasing a connection that was never retrieved."""
+    pool = ConnectionPool(CONNECT_URL, max_connections=5)
+    fake_conn = None  # Simulating an invalid connection
+    with pytest.raises(PoolError) as exc:
+        pool.release(fake_conn)
+    assert "trying to put unkeyed connection" in str(exc.value)
+
+
+def test_close_all_connections():
+    """Test closing all connections."""
+    pool = ConnectionPool(CONNECT_URL, max_connections=5)
+    connections = [pool.get_connection() for _ in range(5)]
+    pool.closeall()
+    assert all(conn.closed for conn in connections)
+
+
+def test_network_error_on_connection():
+    """Test handling network errors during connection retrieval."""
+    with patch("psycopg2.connect", side_effect=OSError("Network Error")):
+        with pytest.raises(OSError) as exc:
+            pool = ConnectionPool(CONNECT_URL, max_connections=5)
+            pool.get_connection()
+        assert "Network Error" in str(exc.value)

tests/test_read.py

@@ -34,6 +34,14 @@ def test_read_stream_last_message(self, client):
         assert message["position"] == 4
         assert message["data"] == {"foo": "bar4"}
 
+    def test_last_message_empty_stream(self, client):
+        result = client.read_last_message("emptyStream-123")
+        assert result is None
+
+    def test_read_from_non_existing_stream(self, client):
+        messages = client.read("nonExistingStream-123")
+        assert messages == []
+
     def test_read_specific_stream_message(self, client):
         for i in range(5):
             client.write("testStream-123", "Event1", {"foo": f"bar{i}"})

tests/test_write.py

@@ -68,3 +68,18 @@ def test_that_write_fails_on_expected_version_mismatch(self, client):
             exc.value.args[0]
             == "P0001-ERROR:  Wrong expected version: 1 (Stream: testStream-123, Stream Version: 2)"
         )
+
+    def test_concurrent_writes(self, client):
+        from threading import Thread
+
+        def write_msg():
+            client.write("concurrentStream-123", "Event", {"thread": "value"})
+
+        threads = [Thread(target=write_msg) for _ in range(10)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        messages = client.read("concurrentStream-123")
+        assert len(messages) == 10