test: use random suffix in DML test table (#114)

This should avoid flakes. Also, adds a table expiration so that
if cleanup fails, BigQuery will still eventually remove the
table.

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/python-bigquery-sqlalchemy/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #111  🦕

tests/system/conftest.py

@@ -4,7 +4,9 @@
 # license that can be found in the LICENSE file or at
 # https://opensource.org/licenses/MIT.
 
+import datetime
 import pathlib
+import random
 
 import pytest
 import google.api_core.exceptions
@@ -16,6 +18,12 @@
 DATA_DIR = pathlib.Path(__file__).parent / "data"
 
 
+def temp_suffix():
+    timestamp = datetime.datetime.utcnow().strftime("%y%m%d_%H%M%S")
+    random_string = hex(random.randrange(1000000))[2:]
+    return f"{timestamp}_{random_string}"
+
+
 def load_sample_data(
     full_table_id: str,
     bigquery_client: bigquery.Client,
@@ -84,12 +92,36 @@ def bigquery_dataset(
 
 
 @pytest.fixture(scope="session", autouse=True)
-def bigquery_empty_table(bigquery_dataset, bigquery_client, bigquery_schema):
+def bigquery_dml_dataset(bigquery_client: bigquery.Client):
+    project_id = bigquery_client.project
+    dataset_id = "test_pybigquery_dml"
+    dataset = bigquery.Dataset(f"{project_id}.{dataset_id}")
+    # Add default table expiration in case cleanup fails.
+    dataset.default_table_expiration_ms = 1000 * int(
+        datetime.timedelta(days=1).total_seconds()
+    )
+    dataset = bigquery_client.create_dataset(dataset, exists_ok=True)
+    return dataset_id
+
+
+@pytest.fixture(scope="session", autouse=True)
+def bigquery_empty_table(
+    bigquery_dataset: str,
+    bigquery_dml_dataset: str,
+    bigquery_client: bigquery.Client,
+    bigquery_schema: List[bigquery.SchemaField],
+):
     project_id = bigquery_client.project
-    dataset_id = bigquery_dataset
-    table_id = f"{project_id}.{dataset_id}.sample_dml"
+    # Cleanup the sample_dml table, if it exists.
+    old_table_id = f"{project_id}.{bigquery_dataset}.sample_dml"
+    bigquery_client.delete_table(old_table_id, not_found_ok=True)
+    # Create new table in its own dataset.
+    dataset_id = bigquery_dml_dataset
+    table_id = f"{project_id}.{dataset_id}.sample_dml_{temp_suffix()}"
     empty_table = bigquery.Table(table_id, schema=bigquery_schema)
-    bigquery_client.create_table(empty_table, exists_ok=True)
+    bigquery_client.create_table(empty_table)
+    yield table_id
+    bigquery_client.delete_table(empty_table)
 
 
 @pytest.fixture(scope="session", autouse=True)

tests/system/test_sqlalchemy_bigquery.py

@@ -174,8 +174,8 @@ def table_one_row(engine):
 
 
 @pytest.fixture(scope="session")
-def table_dml(engine):
-    return Table("test_pybigquery.sample_dml", MetaData(bind=engine), autoload=True)
+def table_dml(engine, bigquery_empty_table):
+    return Table(bigquery_empty_table, MetaData(bind=engine), autoload=True)
 
 
 @pytest.fixture(scope="session")
@@ -355,13 +355,11 @@ def test_tables_list(engine, engine_using_test_dataset):
     tables = engine.table_names()
     assert "test_pybigquery.sample" in tables
     assert "test_pybigquery.sample_one_row" in tables
-    assert "test_pybigquery.sample_dml" in tables
     assert "test_pybigquery.sample_view" not in tables
 
     tables = engine_using_test_dataset.table_names()
     assert "sample" in tables
     assert "sample_one_row" in tables
-    assert "sample_dml" in tables
     assert "sample_view" not in tables
 
 
@@ -520,7 +518,7 @@ def test_dml(engine, session, table_dml):
         {"string": "updated_row"}, synchronize_session=False
     )
     updated_result = table_dml.select().execute().fetchone()
-    assert updated_result["test_pybigquery.sample_dml_string"] == "updated_row"
+    assert updated_result[table_dml.c.string] == "updated_row"
 
     # test delete
     session.query(table_dml).filter(table_dml.c.string == "updated_row").delete(
@@ -576,16 +574,14 @@ def test_table_names_in_schema(inspector, inspector_using_test_dataset):
     tables = inspector.get_table_names("test_pybigquery")
     assert "test_pybigquery.sample" in tables
     assert "test_pybigquery.sample_one_row" in tables
-    assert "test_pybigquery.sample_dml" in tables
     assert "test_pybigquery.sample_view" not in tables
-    assert len(tables) == 3
+    assert len(tables) == 2
 
     tables = inspector_using_test_dataset.get_table_names()
     assert "sample" in tables
     assert "sample_one_row" in tables
-    assert "sample_dml" in tables
     assert "sample_view" not in tables
-    assert len(tables) == 3
+    assert len(tables) == 2
 
 
 def test_view_names(inspector, inspector_using_test_dataset):