fix sql alchemy test

Author: gangtao

python/test/conftest.py

@@ -83,13 +83,21 @@ def engine():
     password = os.environ.get("TIMEPLUS_PASSWORD")
     api_address = os.environ.get("TIMEPLUS_HOST")
     parsed_url = urlparse(api_address)
-    api_netloc = parsed_url.netloc
-    api_schema = parsed_url.scheme
 
     workspace = os.environ.get("TIMEPLUS_WORKSPACE") or "tp-demo"
 
+    schema = parsed_url.scheme
+    host = parsed_url.hostname
+    port = parsed_url.port
+
+    if schema == "https" and port is None:
+        port = 443
+
+    if schema == "http" and port is None:
+        port = 80
+
     if api_key is not None:
-        engine_connection_string = f"timeplus://:{api_key}@{api_netloc}/{workspace}"
+        engine_connection_string = f"timeplus://:{api_key}@{host}:{port}/{workspace}"
         print(f"create engine with connection {engine_connection_string}")
         engine = create_engine(engine_connection_string)
         return engine

python/test/test_sqlalchemy_query.py

@@ -4,38 +4,33 @@
 from timeplus import View
 
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_driver_sql(engine):
     with engine.connect() as conn:
         result = conn.exec_driver_sql(
             "select cid from table(car_live_data) limit 5")
         assert len(result.fetchall()) == 5
 
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_driver_sql_live(engine):
     with engine.connect() as conn:
         result = conn.exec_driver_sql(
             "select cid from car_live_data limit 5")
         assert len(result.fetchall()) == 5
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_text_sql(engine):
     with engine.connect() as connection:
         result = connection.execute(
             text("select * from table(car_live_data) limit 3"))
         rows = [row for row in result]
         assert len(rows) == 3
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_text_sql_live(engine):
     with engine.connect() as connection:
         result = connection.execute(
             text("select * from car_live_data limit 3"))
         rows = [row for row in result]
         assert len(rows) == 3
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_text_streaming_sql(engine):
     with engine.connect() as connection:
         result = connection.execute(text("select * from car_live_data"))
@@ -48,20 +43,17 @@ def test_text_streaming_sql(engine):
                 break
         assert count == max
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_check_stream_existence(engine):
     table_name = "car_live_data"
     with engine.connect() as conn:
         table_exists = engine.dialect.has_table(conn, table_name)
         assert table_exists
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_table_names(engine):
     with engine.connect() as conn:
         tables = engine.dialect.get_table_names(conn)
         assert "car_live_data" in tables
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_view_names(test_environment, engine):
     view_name = "example_mv"
     view = View(env=test_environment).name(view_name)
@@ -87,7 +79,6 @@ def test_view_names(test_environment, engine):
 
     view.delete()
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_materialized_view_names(engine,test_environment,test_stream):
     view_name = "example_mv"
     view = View(env=test_environment).name(view_name)
@@ -111,23 +102,21 @@ def test_materialized_view_names(engine,test_environment,test_stream):
 
     view.delete()
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_view_reflection(engine):
     metadata_obj = MetaData()
     car_view = Table("car_live_data", metadata_obj, autoload_with=engine)
     assert car_view is not None
     column_names = [c.name for c in car_view.columns]
     assert "speed_kmh" in column_names
 
-@pytest.mark.skip(reason="Skipping this test for now")
+
 def test_table_reflection(engine):
     metadata_obj = MetaData()
     car_table = Table("car_live_data", metadata_obj, autoload_with=engine)
     assert car_table is not None
     column_names = [c.name for c in car_table.columns]
     assert "cid" in column_names
 
-@pytest.mark.skip(reason="Skipping this test for now")
 def test_select_query(engine):
     metadata_obj = MetaData()
     car_table = Table("car_live_data", metadata_obj, autoload_with=engine)

python/test/test_sqlalchemy_types.py

@@ -1,7 +1,6 @@
-import pytest
 from sqlalchemy import text
 
-@pytest.mark.skip(reason="Skipping this test for now")
+
 def test_select_basic_types(engine):
     with engine.connect() as conn:
         result = conn.execute(text(
@@ -20,7 +19,7 @@ def test_select_basic_types(engine):
             assert isinstance(row[9], bool)
             assert isinstance(row[10], str)
 
-@pytest.mark.skip(reason="Skipping this test for now")
+
 def test_select_composite_array_tuple(engine):
     with engine.connect() as conn:
         result = conn.execute(text(
@@ -34,7 +33,7 @@ def test_select_composite_array_tuple(engine):
             assert isinstance(row[1][1], str)
             assert isinstance(row[1][2], float)
 
-@pytest.mark.skip(reason="Skipping this test for now")
+
 def test_select_composite_map(engine):
     with engine.connect() as conn:
         result = conn.execute(text(