Test fixtures

arangoasync/connection.py

@@ -172,6 +172,7 @@ async def process_request(self, request: Request) -> Response:
             ConnectionAbortedError: If it can't connect to host(s) within limit.
         """
 
+        request.endpoint = f"{self._db_endpoint}{request.endpoint}"
         host_index = self._host_resolver.get_host_index()
         for tries in range(self._host_resolver.max_tries):
             try:

tests/conftest.py

@@ -1,9 +1,13 @@
+import asyncio
 from dataclasses import dataclass
 
 import pytest
 import pytest_asyncio
 
-from arangoasync.auth import JwtToken
+from arangoasync.auth import Auth, JwtToken
+from arangoasync.client import ArangoClient
+from arangoasync.typings import UserInfo
+from tests.helpers import generate_col_name, generate_db_name, generate_username
 
 
 @dataclass
@@ -14,6 +18,7 @@ class GlobalData:
     secret: str = None
     token: str = None
     sys_db_name: str = "_system"
+    username: str = generate_username()
 
 
 global_data = GlobalData()
@@ -49,31 +54,48 @@ def pytest_configure(config):
     global_data.token = JwtToken.generate_token(global_data.secret)
 
 
-@pytest.fixture(autouse=False)
+@pytest.fixture
 def url():
     return global_data.url
 
 
-@pytest.fixture(autouse=False)
+@pytest.fixture
 def root():
     return global_data.root
 
 
-@pytest.fixture(autouse=False)
+@pytest.fixture
 def password():
     return global_data.password
 
 
-@pytest.fixture(autouse=False)
+@pytest.fixture
+def basic_auth_root(root, password):
+    return Auth(username=root, password=password)
+
+
+@pytest.fixture
+def username():
+    return global_data.username
+
+
+@pytest.fixture
 def token():
     return global_data.token
 
 
-@pytest.fixture(autouse=False)
+@pytest.fixture
 def sys_db_name():
     return global_data.sys_db_name
 
 
+@pytest.fixture(scope="session")
+def event_loop():
+    loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()
+
+
 @pytest_asyncio.fixture
 async def client_session():
     """Make sure we close all sessions after the test is done."""
@@ -88,3 +110,89 @@ def get_client_session(client, url):
 
     for session in sessions:
         await session.close()
+
+
+@pytest_asyncio.fixture
+async def arango_client(url):
+    async with ArangoClient(hosts=url) as client:
+        yield client
+
+
+@pytest_asyncio.fixture
+async def sys_db(arango_client, sys_db_name, basic_auth_root):
+    return await arango_client.db(
+        sys_db_name, auth_method="basic", auth=basic_auth_root, verify=False
+    )
+
+
+@pytest_asyncio.fixture
+async def test_db(arango_client, sys_db, password):
+    tst_db_name = generate_db_name()
+    tst_user = UserInfo(
+        user=generate_username(),
+        password=password,
+        active=True,
+    )
+    await sys_db.create_database(tst_db_name, users=[tst_user])
+    yield await arango_client.db(
+        tst_db_name,
+        auth_method="basic",
+        auth=Auth(username=tst_user.user, password=password),
+        verify=False,
+    )
+    await sys_db.delete_database(tst_db_name)
+
+
+@pytest_asyncio.fixture
+async def bad_db(arango_client):
+    return await arango_client.db(
+        generate_db_name(),
+        auth_method="basic",
+        auth=Auth(username="bad_user", password="bad_password"),
+        verify=False,
+    )
+
+
+@pytest_asyncio.fixture
+async def test_doc_col(test_db):
+    col_name = generate_col_name()
+    yield await test_db.create_collection(col_name)
+    await test_db.delete_collection(col_name)
+
+
+@pytest_asyncio.fixture(scope="session", autouse=True)
+async def teardown():
+    yield
+    async with ArangoClient(hosts=global_data.url) as client:
+        sys_db = await client.db(
+            global_data.sys_db_name,
+            auth_method="basic",
+            auth=Auth(username=global_data.root, password=global_data.password),
+            verify=False,
+        )
+
+        # Remove all test users.
+        tst_users = [
+            user["user"]
+            for user in await sys_db.users()
+            if user["user"].startswith("test_user")
+        ]
+        await asyncio.gather(*(sys_db.delete_user(user) for user in tst_users))
+
+        # Remove all test databases.
+        tst_dbs = [
+            db_name
+            for db_name in await sys_db.databases()
+            if db_name.startswith("test_database")
+        ]
+        await asyncio.gather(*(sys_db.delete_database(db_name) for db_name in tst_dbs))
+
+        # Remove all test collections.
+        tst_cols = [
+            col_info.name
+            for col_info in await sys_db.collections()
+            if col_info.name.startswith("test_collection")
+        ]
+        await asyncio.gather(
+            *(sys_db.delete_collection(col_name) for col_name in tst_cols)
+        )

tests/test_client.py

@@ -1,6 +1,6 @@
 import pytest
 
-from arangoasync.auth import Auth, JwtToken
+from arangoasync.auth import JwtToken
 from arangoasync.client import ArangoClient
 from arangoasync.compression import DefaultCompressionManager
 from arangoasync.http import DefaultHTTPClient
@@ -62,12 +62,15 @@ async def test_client_bad_auth_method(url, sys_db_name):
 
 
 @pytest.mark.asyncio
-async def test_client_basic_auth(url, sys_db_name, root, password):
-    auth = Auth(username=root, password=password)
-
+async def test_client_basic_auth(url, sys_db_name, basic_auth_root):
     # successful authentication
     async with ArangoClient(hosts=url) as client:
-        await client.db(sys_db_name, auth_method="basic", auth=auth, verify=True)
+        await client.db(
+            sys_db_name,
+            auth_method="basic",
+            auth=basic_auth_root,
+            verify=True,
+        )
 
     # auth missing
     async with ArangoClient(hosts=url) as client:
@@ -82,13 +85,17 @@ async def test_client_basic_auth(url, sys_db_name, root, password):
 
 
 @pytest.mark.asyncio
-async def test_client_jwt_auth(url, sys_db_name, root, password):
-    auth = Auth(username=root, password=password)
+async def test_client_jwt_auth(url, sys_db_name, basic_auth_root):
     token: JwtToken
 
     # successful authentication with auth only
     async with ArangoClient(hosts=url) as client:
-        db = await client.db(sys_db_name, auth_method="jwt", auth=auth, verify=True)
+        db = await client.db(
+            sys_db_name,
+            auth_method="jwt",
+            auth=basic_auth_root,
+            verify=True,
+        )
         token = db.connection.token
 
     # successful authentication with token only
@@ -98,7 +105,11 @@ async def test_client_jwt_auth(url, sys_db_name, root, password):
         # successful authentication with both
         async with ArangoClient(hosts=url) as client:
             await client.db(
-                sys_db_name, auth_method="jwt", auth=auth, token=token, verify=True
+                sys_db_name,
+                auth_method="jwt",
+                auth=basic_auth_root,
+                token=token,
+                verify=True,
             )
 
     # auth and token missing
@@ -108,9 +119,7 @@ async def test_client_jwt_auth(url, sys_db_name, root, password):
 
 
 @pytest.mark.asyncio
-async def test_client_jwt_superuser_auth(url, sys_db_name, root, password, token):
-    auth = Auth(username=root, password=password)
-
+async def test_client_jwt_superuser_auth(url, sys_db_name, basic_auth_root, token):
     # successful authentication
     async with ArangoClient(hosts=url) as client:
         await client.db(sys_db_name, auth_method="superuser", token=token, verify=True)
@@ -119,5 +128,5 @@ async def test_client_jwt_superuser_auth(url, sys_db_name, root, password, token
     async with ArangoClient(hosts=url) as client:
         with pytest.raises(ValueError):
             await client.db(
-                sys_db_name, auth_method="superuser", auth=auth, verify=True
+                sys_db_name, auth_method="superuser", auth=basic_auth_root, verify=True
             )

tests/test_database.py

@@ -1,81 +1,64 @@
 import pytest
 
-from arangoasync.auth import Auth
-from arangoasync.client import ArangoClient
 from arangoasync.collection import StandardCollection
 from tests.helpers import generate_col_name, generate_db_name
 
 
 @pytest.mark.asyncio
-async def test_database_misc_methods(url, sys_db_name, root, password):
-    auth = Auth(username=root, password=password)
-
-    # TODO also handle exceptions
-    async with ArangoClient(hosts=url) as client:
-        db = await client.db(sys_db_name, auth_method="basic", auth=auth, verify=True)
-        status = await db.status()
-        assert status["server"] == "arango"
+async def test_database_misc_methods(sys_db):
+    status = await sys_db.status()
+    assert status["server"] == "arango"
 
 
 @pytest.mark.asyncio
-async def test_create_drop_database(url, sys_db_name, root, password):
-    auth = Auth(username=root, password=password)
-
+async def test_create_drop_database(arango_client, sys_db, basic_auth_root):
     # TODO also handle exceptions
     # TODO use more options (cluster must be enabled for that)
-    async with ArangoClient(hosts=url) as client:
-        sys_db = await client.db(
-            sys_db_name, auth_method="basic", auth=auth, verify=True
-        )
-
-        # Create a new database
-        db_name = generate_db_name()
-        assert await sys_db.create_database(db_name) is True
-        new_db = await client.db(db_name, auth_method="basic", auth=auth, verify=True)
-        assert await sys_db.has_database(db_name) is True
 
-        # List available databases
-        dbs = await sys_db.databases()
-        assert db_name in dbs
-        assert "_system" in dbs
-
-        # TODO move this to a separate test
-        col_name = generate_col_name()
-        col = await new_db.create_collection(col_name)
-        await col.insert({"_key": "1", "a": 1})
-        doc = await col.get("1")
-        assert doc["_key"] == "1"
-
-        # Drop the newly created database
-        assert await sys_db.delete_database(db_name) is True
-        non_existent_db = generate_db_name()
-        assert await sys_db.has_database(non_existent_db) is False
-        assert (
-            await sys_db.delete_database(non_existent_db, ignore_missing=True) is False
-        )
+    # Create a new database
+    db_name = generate_db_name()
+    assert await sys_db.create_database(db_name) is True
+    new_db = await arango_client.db(
+        db_name, auth_method="basic", auth=basic_auth_root, verify=True
+    )
+    assert await sys_db.has_database(db_name) is True
+
+    # List available databases
+    dbs = await sys_db.databases()
+    assert db_name in dbs
+    assert "_system" in dbs
+
+    # TODO move this to a separate test for documents
+    col_name = generate_col_name()
+    col = await new_db.create_collection(col_name)
+    await col.insert({"_key": "1", "a": 1})
+    doc = await col.get("1")
+    assert doc["_key"] == "1"
+
+    # Drop the newly created database
+    assert await sys_db.delete_database(db_name) is True
+    non_existent_db = generate_db_name()
+    assert await sys_db.has_database(non_existent_db) is False
+    assert await sys_db.delete_database(non_existent_db, ignore_missing=True) is False
 
 
 @pytest.mark.asyncio
-async def test_create_drop_collection(url, sys_db_name, root, password):
-    auth = Auth(username=root, password=password)
-
+async def test_create_drop_collection(test_db):
     # TODO also handle exceptions
-    async with ArangoClient(hosts=url) as client:
-        db = await client.db(sys_db_name, auth_method="basic", auth=auth, verify=True)
-
-        # Create a new collection
-        col_name = generate_col_name()
-        col = await db.create_collection(col_name)
-        assert isinstance(col, StandardCollection)
-        assert await db.has_collection(col_name)
-        cols = await db.collections()
-        assert any(c.name == col_name for c in cols)
 
-        # Drop the newly created collection
-        assert await db.delete_collection(col_name) is True
-        assert not await db.has_collection(col_name)
-        non_existent_col = generate_col_name()
-        assert await db.has_collection(non_existent_col) is False
-        assert (
-            await db.delete_collection(non_existent_col, ignore_missing=True) is False
-        )
+    # Create a new collection
+    col_name = generate_col_name()
+    col = await test_db.create_collection(col_name)
+    assert isinstance(col, StandardCollection)
+    assert await test_db.has_collection(col_name)
+    cols = await test_db.collections()
+    assert any(c.name == col_name for c in cols)
+
+    # Drop the newly created collection
+    assert await test_db.delete_collection(col_name) is True
+    assert not await test_db.has_collection(col_name)
+    non_existent_col = generate_col_name()
+    assert await test_db.has_collection(non_existent_col) is False
+    assert (
+        await test_db.delete_collection(non_existent_col, ignore_missing=True) is False
+    )