Merges pull request #3

Off of Amazon!!!

1 files changed, 69 insertions, 0 deletions

diff --git a/tests/test_postgres.py b/tests/test_postgres.py
new file mode 100644
index 0000000..244c132
--- /dev/null
+++ b/tests/test_postgres.py
@@ -0,0 +1,69 @@
+import socket
+import subprocess
+from collections.abc import Callable
+from typing import TYPE_CHECKING
+
+from generalresearch.pg_helper import PostgresConfig
+from pydantic import PostgresDsn
+
+if TYPE_CHECKING:
+    from generalresearch.models.custom_types import InternalHostname, PostgresDict
+
+
+def is_port_open(host: InternalHostname, port: int = 5432, timeout: int = 3):
+    try:
+        with socket.create_connection((host, port), timeout=timeout):
+            return True
+    except (TimeoutError, ConnectionRefusedError, OSError):
+        return False
+
+
+def can_ping(host: InternalHostname):
+    return (
+        subprocess.call(
+            ["ping", "-c", "1", str(host)],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        == 0
+    )
+
+
+class TestPostgresDSN:
+
+    def test_ping(self, postgres_instance_host: InternalHostname):
+        assert can_ping(host=postgres_instance_host)
+
+    def test_port(self, postgres_instance_host: InternalHostname):
+        assert is_port_open(host=postgres_instance_host)
+
+    def test_conn(self, postgres_instance: PostgresDsn):
+        config = PostgresConfig(
+            dsn=postgres_instance,
+            connect_timeout=1,
+            statement_timeout=1,
+        )
+        res = config.execute_sql_query(query="SELECT 1;")
+        assert len(res) == 1
+
+
+class TestPostgresDjangoCreation:
+
+    def test_ping(self, postgres_instance_dict: PostgresDict):
+        assert can_ping(host=postgres_instance_dict["host"])
+
+    def test_django_creation(
+        self,
+        django_db_factory: Callable[..., None],
+    ):
+        dsn = django_db_factory()
+        assert isinstance(dsn, PostgresDsn)
+
+    def test_django_tables(self, pg_config: PostgresConfig):
+        res = pg_config.execute_sql_query(query="""
+            SELECT COUNT(*)
+            FROM information_schema.tables
+            WHERE table_schema = 'public';
+        """)
+        assert len(res) == 1
+        assert res[0]["count"] == 7