Reworked tests to support running on specific database

.github/workflows/pull-request.yml

@@ -43,9 +43,17 @@ jobs:
         run: |
           poetry run pytest -v --tb=short tests/unit-tests/
 
-      - name: Integration Tests
+      - name: Integration Tests on a shared cluster
         env:
           XATA_WORKSPACE_ID: ${{ secrets.INTEGRATION_TEST_WORKSPACE }}
           XATA_API_KEY: ${{ secrets.INTEGRATION_TEST_API_KEY }}
         run: |
           poetry run pytest -v --tb=short -W ignore::DeprecationWarning tests/integration-tests/
+
+      - name: Integration Tests on a dedicated cluster
+        env:
+          XATA_WORKSPACE_ID: ${{ secrets.INTEGRATION_DC_TEST_WORKSPACE }}
+          XATA_API_KEY: ${{ secrets.INTEGRATION_TEST_API_KEY }}
+          XATA_STATIC_DB_NAME: ${{ secrets.INTEGRATION_DC_STATIC_DB_NAME }}
+        run: |
+          poetry run pytest -v --tb=short -W ignore::DeprecationWarning tests/integration-tests/

test.py

@@ -0,0 +1,5 @@
+from xata.client import XataClient
+
+xata = XataClient(db_name="xata-py", workspace_id="xata-uq2d57")
+record = xata.records().get("AttachmentsIssue", "rec_cpleilidajcq52kshbpg", columns=["one_file.id"])
+print(record)

tests/integration-tests/api_request_test.py

@@ -43,7 +43,7 @@ def test_direct_instance_in_namespace_wit_client_dependencies(self):
         Direct namespace invocation with dependency on internal client config
         :link https://github.com/xataio/xata-py/issues/57
         """
-        db_name = utils.get_db_name()
+        db_name = utils.get_db_name(True)
         databases = XataClient().databases()
 
         # workspace_id should be provided by the the client implicitly

tests/integration-tests/api_response_test.py

@@ -17,6 +17,8 @@
 # under the License.
 #
 
+import os
+
 import utils
 
 from xata.client import XataClient
@@ -26,9 +28,11 @@ class TestApiResponse(object):
     def setup_class(self):
         self.db_name = utils.get_db_name()
         self.client = XataClient(db_name=self.db_name)
-        assert self.client.databases().create(self.db_name).is_success()
-        assert self.client.table().create("Posts").is_success()
-        assert self.client.table().set_schema("Posts", utils.get_posts_schema()).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().create(self.db_name).is_success()
+        if not self.client.table().get_schema("Posts").is_success():
+            assert self.client.table().create("Posts").is_success()
+            assert self.client.table().set_schema("Posts", utils.get_posts_schema()).is_success()
 
         payload = {
             "operations": [
@@ -47,7 +51,9 @@ def setup_class(self):
         assert self.client.records().transaction(payload).is_success()
 
     def teardown_class(self):
-        assert self.client.databases().delete(self.db_name).is_success()
+        assert self.client.table().delete("Posts").is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().delete(self.db_name).is_success()
 
     def test_is_success_true(self):
         user = XataClient().users().get()

tests/integration-tests/branch_test.py

@@ -17,6 +17,8 @@
 # under the License.
 #
 
+import os
+
 import pytest
 import utils
 
@@ -29,28 +31,19 @@ def setup_class(self):
         self.branch_name = "main"
         self.client = XataClient(db_name=self.db_name, branch_name=self.branch_name)
 
-        # create database
-        assert self.client.databases().create(self.db_name).is_success()
-        assert self.client.table().create("Posts").is_success
-        assert (
-            self.client.table()
-            .set_schema(
-                "Posts",
-                {
-                    "columns": [
-                        {"name": "title", "type": "string"},
-                        {"name": "labels", "type": "multiple"},
-                        {"name": "slug", "type": "string"},
-                        {"name": "text", "type": "text"},
-                    ]
-                },
-            )
-            .is_success()
-        )
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().create(self.db_name).is_success()
+        assert self.client.table().create("Posts").is_success()
+        assert self.client.table().set_schema("Posts", utils.get_posts_schema()).is_success()
 
     def teardown_class(self):
-        r = self.client.databases().delete(self.db_name)
-        assert r.is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().delete(self.db_name).is_success()
+        else:
+            assert self.client.table().delete("Posts").is_success()
+            for b in self.client.branch().list(self.db_name).get("branches"):
+                if b["name"] != "main":
+                    self.client.branch().delete(branch_name=b["name"])
 
     def test_get_branch_list(self):
         r = self.client.branch().list(self.db_name)

tests/integration-tests/databases_test.py

@@ -26,7 +26,7 @@
 
 class TestDatabasesNamespace(object):
     def setup_class(self):
-        self.db_name = utils.get_db_name()
+        self.db_name = utils.get_db_name(True)
         self.branch_name = "main"
         self.client = XataClient(db_name=self.db_name, branch_name=self.branch_name)
 

tests/integration-tests/files_access.py

@@ -17,6 +17,7 @@
 # under the License.
 #
 
+import os
 import time
 
 import utils
@@ -28,25 +29,19 @@
 class TestFilesAccess(object):
     def setup_class(self):
         self.db_name = utils.get_db_name()
-        self.branch_name = "main"
-        self.client = XataClient(db_name=self.db_name, branch_name=self.branch_name)
+        self.client = XataClient(db_name=self.db_name)
         self.fake = utils.get_faker()
 
-        assert self.client.databases().create(self.db_name).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().create(self.db_name).is_success()
         assert self.client.table().create("Attachments").is_success()
-        assert (
-            self.client.table()
-            .set_schema(
-                "Attachments",
-                utils.get_attachments_schema(),
-                db_name=self.db_name,
-                branch_name=self.branch_name,
-            )
-            .is_success()
-        )
+        assert self.client.table().set_schema("Attachments", utils.get_attachments_schema()).is_success()
 
     def teardown_class(self):
-        assert self.client.databases().delete(self.db_name).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().delete(self.db_name).is_success()
+        else:
+            assert self.client.table().delete("Attachments").is_success()
 
     def test_public_flag_true(self):
         payload = {

tests/integration-tests/files_multiple_files_test.py

@@ -17,6 +17,8 @@
 # under the License.
 #
 
+import os
+
 import utils
 from requests import request
 
@@ -26,29 +28,22 @@
 class TestFilesMultipleFiles(object):
     def setup_class(self):
         self.db_name = utils.get_db_name()
-        self.branch_name = "main"
-        self.client = XataClient(db_name=self.db_name, branch_name=self.branch_name)
+        self.client = XataClient(db_name=self.db_name)
         self.fake = utils.get_faker()
 
-        assert self.client.databases().create(self.db_name).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().create(self.db_name).is_success()
         assert self.client.table().create("Attachments").is_success()
-        assert (
-            self.client.table()
-            .set_schema(
-                "Attachments",
-                utils.get_attachments_schema(),
-                db_name=self.db_name,
-                branch_name=self.branch_name,
-            )
-            .is_success()
-        )
+        assert self.client.table().set_schema("Attachments", utils.get_attachments_schema()).is_success()
 
     def teardown_class(self):
-        assert self.client.databases().delete(self.db_name).is_success()
+        assert self.client.table().delete("Attachments").is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().delete(self.db_name).is_success()
 
     def test_put_file_item(self):
         payload = {
-            "title": self.fake.catch_phrase(),
+            "title": "test_put_file",
             "many_files": [
                 utils.get_file("images/01.gif", public_url=True),
                 utils.get_file("images/02.gif", public_url=True),
@@ -74,7 +69,12 @@ def test_put_file_item(self):
         assert img_2 == proof_2.content
 
         # overwrite item 1 with image 2
+        assert record["many_files"][0] == ""
         file_1 = self.client.files().put_item("Attachments", rid, "many_files", record["many_files"][0]["id"], img_2)
+
+        # assert file_1.status_code == 0 # extra
+        # assert file_1.json() == ""     # extra
+
         assert file_1.is_success()
         assert "attributes" in file_1
         assert "mediaType" in file_1
@@ -91,7 +91,7 @@ def test_put_file_item(self):
 
     def test_delete_file(self):
         payload = {
-            "title": self.fake.catch_phrase(),
+            "title": "test_delete_file",
             "many_files": [
                 utils.get_file("images/01.gif", public_url=True),
                 utils.get_file("images/02.gif", public_url=True),
@@ -118,7 +118,7 @@ def test_delete_file(self):
 
     def test_get_item(self):
         payload = {
-            "title": self.fake.catch_phrase(),
+            "title": "test_get_item",
             "many_files": [
                 utils.get_file("images/01.gif", public_url=True),
                 utils.get_file("images/02.gif", public_url=True),

tests/integration-tests/files_single_file_test.py

@@ -17,6 +17,8 @@
 # under the License.
 #
 
+import os
+
 import utils
 from faker import Faker
 
@@ -26,25 +28,19 @@
 class TestFilesSingleFile(object):
     def setup_class(self):
         self.db_name = utils.get_db_name()
-        self.branch_name = "main"
-        self.client = XataClient(db_name=self.db_name, branch_name=self.branch_name)
-        self.fake = Faker()
+        self.client = XataClient(db_name=self.db_name)
+        self.fake = utils.get_faker()
 
-        assert self.client.databases().create(self.db_name).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().create(self.db_name).is_success()
         assert self.client.table().create("Attachments").is_success()
-        assert (
-            self.client.table()
-            .set_schema(
-                "Attachments",
-                utils.get_attachments_schema(),
-                db_name=self.db_name,
-                branch_name=self.branch_name,
-            )
-            .is_success()
-        )
+        assert self.client.table().set_schema("Attachments", utils.get_attachments_schema()).is_success()
 
     def teardown_class(self):
-        assert self.client.databases().delete(self.db_name).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().delete(self.db_name).is_success()
+        else:
+            assert self.client.table().delete("Attachments").is_success()
 
     def test_put_csv_file(self):
         payload = {"title": self.fake.catch_phrase()}

tests/integration-tests/files_transormations_test.py

@@ -19,6 +19,7 @@
 
 import io
 import json
+import os
 
 import pytest
 import utils
@@ -33,22 +34,17 @@ class TestFilesTransformations(object):
     def setup_class(self):
         self.db_name = utils.get_db_name()
         self.client = XataClient(db_name=self.db_name)
-        self.fake = Faker()
+        self.fake = utils.get_faker()
 
-        assert self.client.databases().create(self.db_name).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().create(self.db_name).is_success()
         assert self.client.table().create("Attachments").is_success()
-        assert (
-            self.client.table()
-            .set_schema(
-                "Attachments",
-                utils.get_attachments_schema(),
-                db_name=self.db_name,
-            )
-            .is_success()
-        )
+        assert self.client.table().set_schema("Attachments", utils.get_attachments_schema()).is_success()
 
     def teardown_class(self):
-        assert self.client.databases().delete(self.db_name).is_success()
+        assert self.client.table().delete("Attachments").is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().delete(self.db_name).is_success()
 
     def test_rotate_public_file(self):
         payload = {

tests/integration-tests/files_upload_url_test.py

@@ -17,33 +17,29 @@
 # under the License.
 #
 
-import utils
+import os
 
+import utils
 from requests import request
+
 from xata.client import XataClient
 
+
 class TestFilesSingleFile(object):
     def setup_class(self):
         self.db_name = utils.get_db_name()
-        self.branch_name = "main"
-        self.client = XataClient(db_name=self.db_name, branch_name=self.branch_name)
+        self.client = XataClient(db_name=self.db_name)
         self.fake = utils.get_faker()
 
-        assert self.client.databases().create(self.db_name).is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().create(self.db_name).is_success()
         assert self.client.table().create("Attachments").is_success()
-        assert (
-            self.client.table()
-            .set_schema(
-                "Attachments",
-                utils.get_attachments_schema(),
-                db_name=self.db_name,
-                branch_name=self.branch_name,
-            )
-            .is_success()
-        )
+        assert self.client.table().set_schema("Attachments", utils.get_attachments_schema()).is_success()
 
     def teardown_class(self):
-        assert self.client.databases().delete(self.db_name).is_success()
+        assert self.client.table().delete("Attachments").is_success()
+        if not os.environ.get("XATA_STATIC_DB_NAME"):
+            assert self.client.databases().delete(self.db_name).is_success()
 
     def test_upload_file(self):
         payload = {"title": self.fake.catch_phrase()}