Fix wrong pip install and path that pg_es_fdw is installed into; …

…add a test that instantiates the ES FDW and checks the table schema.

engine/Dockerfile

@@ -183,8 +183,8 @@ COPY ./bin /splitgraph/bin
 # "Install" elasticsearch_fdw
 RUN --mount=type=cache,id=pip-cache,target=/root/.cache/pip \
     mkdir /pg_es_fdw && \
-    pip install elasticsearch>=7.7.0
-COPY ./engine/src/postgres-elasticsearch-fdw/pg_es_fdw /pg_es_fdw/
+    pip install "elasticsearch>=7.7.0"
+COPY ./engine/src/postgres-elasticsearch-fdw/pg_es_fdw /pg_es_fdw/pg_es_fdw
 
 ENV PATH "${PATH}:/splitgraph/bin"
 ENV PYTHONPATH "${PYTHONPATH}:/splitgraph:/pg_es_fdw"

test/splitgraph/commands/test_mounting.py

@@ -6,6 +6,7 @@
 from splitgraph.core.repository import Repository
 from splitgraph.core.types import TableColumn
 from splitgraph.engine import get_engine
+from splitgraph.hooks.mount_handlers import mount
 
 PG_MNT = Repository.from_schema("test/pg_mount")
 MG_MNT = Repository.from_schema("test_mg_mount")
@@ -65,3 +66,51 @@ def test_cross_joins(local_engine_empty):
         )
         == [(2, "James", "orange")]
     )
+
+
+@pytest.mark.mounting
+def test_mount_elasticsearch(local_engine_empty):
+    # No ES running in this stack: this is just a test that we can instantiate the FDW.
+    repo = Repository("test", "es_mount")
+    try:
+        mount(
+            repo.to_schema(),
+            "elasticsearch",
+            dict(
+                username=None,
+                password=None,
+                server="elasticsearch",
+                port=9200,
+                table_spec={
+                    "table_1": {
+                        "schema": {
+                            "id": "text",
+                            "@timestamp": "timestamp",
+                            "query": "text",
+                            "col_1": "text",
+                            "col_2": "boolean",
+                        },
+                        "index": "index-pattern*",
+                        "rowid_column": "id",
+                        "query_column": "query",
+                    }
+                },
+            ),
+        )
+
+        assert get_engine().get_full_table_schema(repo.to_schema(), "table_1") == [
+            TableColumn(ordinal=1, name="id", pg_type="text", is_pk=False, comment=None),
+            TableColumn(
+                ordinal=2,
+                name="@timestamp",
+                pg_type="timestamp without time zone",
+                is_pk=False,
+                comment=None,
+            ),
+            TableColumn(ordinal=3, name="query", pg_type="text", is_pk=False, comment=None),
+            TableColumn(ordinal=4, name="col_1", pg_type="text", is_pk=False, comment=None),
+            TableColumn(ordinal=5, name="col_2", pg_type="boolean", is_pk=False, comment=None),
+        ]
+
+    finally:
+        repo.delete()