taskbuffer/tests/wrapped_conn_tests.py

0001 # Description: Unit tests for the wrappers of the Oracle and Postgres connections. No database is used
0002 import unittest
0003 from typing import Any
0004
0005 from pandaserver.taskbuffer.wrapped_oracle_conn import WrappedOracleConn
0006 from pandaserver.taskbuffer.WrappedPostgresConn import WrappedPostgresConn
0007
0008
0009 class RawConnection:
0010     closed = 0
0011
0012     def commit(self) -> str:
0013         return "committed"
0014
0015
0016 def counting(wrapper_class: type[Any]) -> tuple[type[Any], list[str]]:
0017     # a subclass recording each name its __getattribute__ is asked for, which is also what a
0018     # read of an attribute through self inside the wrapper's own __getattribute__ would call
0019     names: list[str] = []
0020
0021     class Counting(wrapper_class):  # type: ignore[misc, valid-type]
0022         def __getattribute__(self, item: str) -> Any:
0023             names.append(item)
0024             return super().__getattribute__(item)
0025
0026     return Counting, names
0027
0028
0029 class TestWrappedConnections(unittest.TestCase):
0030     def test_one_lookup_per_read(self) -> None:
0031         for wrapper_class in (WrappedOracleConn, WrappedPostgresConn):
0032             counting_class, names = counting(wrapper_class)
0033             wrapped = counting_class(RawConnection())
0034             self.assertEqual(wrapped.commit(), "committed")
0035             self.assertEqual(names, ["commit"], wrapper_class.__name__)
0036
0037     def test_names_of_the_connection_then_of_the_wrapper(self) -> None:
0038         raw = RawConnection()
0039         for wrapper_class in (WrappedOracleConn, WrappedPostgresConn):
0040             wrapped = wrapper_class(raw)
0041             self.assertIs(wrapped.orig_conn, raw)
0042             self.assertEqual(wrapped.commit(), "committed")
0043             with self.assertRaises(AttributeError):
0044                 wrapped.no_such_attribute
0045         # the methods the Postgres connection lacks come from the wrapper. Any, since what is read
0046         # here is what __getattribute__ finds, which the static types do not tell
0047         postgres: Any = WrappedPostgresConn(raw)
0048         self.assertIs(postgres.begin.__func__, WrappedPostgresConn.begin)
0049         self.assertIs(postgres.ping.__func__, WrappedPostgresConn.ping)
0050         # reads closed of the connection through the wrapper, and passes since it is not closed
0051         postgres.ping()
0052
0053
0054 if __name__ == "__main__":
0055     unittest.main()