Merge pull request #417 from multinet-app/user-test-context

test/conftest.py

@@ -1,15 +1,40 @@
 """Pytest configurations for multinet tests."""
 
 import pytest
+import dacite
 from uuid import uuid4
+from dataclasses import asdict, dataclass
+from contextlib import contextmanager
 from pathlib import Path
-from dataclasses import asdict
 
 from multinet import create_app
 from multinet.db import create_workspace, delete_workspace
-from multinet.user import register_user, set_user_cookie, user_collection, UserInfo
+from multinet.user import (
+    register_user,
+    set_user_cookie,
+    user_collection,
+    UserInfo,
+    User,
+    MULTINET_COOKIE,
+)
 
-from typing import Tuple
+from typing import Generator, Tuple
+
+
+@dataclass
+class ContextUser(User):
+    """Inherits User to provide testing login context."""
+
+    @contextmanager
+    def login(self, server) -> Generator[None, None, None]:
+        """Ensure the user is logged in during this context."""
+        with server.session_transaction() as session:
+            session[MULTINET_COOKIE] = self.multinet.session
+
+        yield None
+
+        with server.session_transaction() as session:
+            del session[MULTINET_COOKIE]
 
 
 @pytest.fixture
@@ -33,7 +58,7 @@ def data_directory() -> Path:
 
 
 @pytest.fixture
-def handled_user():
+def managed_user():
     """
     Create a user, and yield that user.
 
@@ -52,24 +77,24 @@ def handled_user():
         )
     )
 
-    yield user
+    yield dacite.from_dict(data_class=ContextUser, data=asdict(user))
 
     # TODO: Once the function exists, use `delete_user` here instead
     user_collection().delete(asdict(user))
 
 
 @pytest.fixture
-def generated_workspace(handled_user):
+def generated_workspace(managed_user):
     """Create a workspace, and yield the name of the workspace."""
 
     workspace_name = uuid4().hex
 
-    create_workspace(workspace_name, handled_user)
+    create_workspace(workspace_name, managed_user)
     return workspace_name
 
 
 @pytest.fixture
-def handled_workspace(generated_workspace):
+def managed_workspace(generated_workspace):
     """
     Create a workspace, and yield the name of the workspace.
 
@@ -82,7 +107,7 @@ def handled_workspace(generated_workspace):
 
 @pytest.fixture
 def populated_workspace(
-    handled_workspace, data_directory, server
+    managed_workspace, data_directory, server
 ) -> Tuple[str, str, str, str]:
     """
     Populate a workspace with some data.
@@ -94,7 +119,7 @@ def populated_workspace(
     with open(Path(data_directory) / "miserables.json") as miserables:
         data = miserables.read()
 
-    resp = server.post(f"/api/d3_json/{handled_workspace}/miserables", data=data)
+    resp = server.post(f"/api/d3_json/{managed_workspace}/miserables", data=data)
     assert resp.status_code == 200
 
-    return (handled_workspace, "miserables", "miserables_nodes", "miserables_links")
+    return (managed_workspace, "miserables", "miserables_nodes", "miserables_links")

test/test_aql_table_creation.py

@@ -1,14 +1,14 @@
 """Tests for creating a table from an AQL query."""
 
 
-def test_malformed_aql(handled_workspace, server):
+def test_malformed_aql(managed_workspace, server):
     """Test that invalid/malformed AQL results in an error."""
     table_name = "malformed_table"
     malformed_aql = "FOR members RETURN member"
     malformed_aql_error = "unexpected RETURN declaration"
 
     resp = server.post(
-        f"/api/workspaces/{handled_workspace}/tables",
+        f"/api/workspaces/{managed_workspace}/tables",
         data=malformed_aql,
         query_string={"table": table_name},
     )

test/test_meta.py

@@ -1,38 +1,42 @@
 """Tests that ensure fixtures act properly."""
 
-from multinet.user import MULTINET_COOKIE
 
-
-def test_generated_workspace(handled_workspace, handled_user, server):
+def test_generated_workspace(managed_workspace, managed_user, server):
     """Test that a generated workspace exists when querying the API."""
 
-    with server.session_transaction() as session:
-        session[MULTINET_COOKIE] = handled_user.multinet.session
+    with managed_user.login(server):
+        resp = server.get(f"/api/workspaces/{managed_workspace}")
+        assert resp.status_code == 200
+
 
-    resp = server.get(f"/api/workspaces/{handled_workspace}")
-    assert resp.status_code == 200
+def test_user_context(managed_workspace, managed_user, server):
+    """Test that the user context properly controls login."""
+    with managed_user.login(server):
+        resp = server.get(f"/api/workspaces/{managed_workspace}/tables")
+        assert resp.status_code == 200
 
+    resp = server.get(f"/api/workspaces/{managed_workspace}/tables")
+    assert resp.status_code == 401
 
-def test_populated_workspace(populated_workspace, handled_user, server):
+
+def test_populated_workspace(populated_workspace, managed_user, server):
     """Test that the populated workspace has a graph, an edge and node table."""
     workspace, graphs, *tables = populated_workspace
 
-    with server.session_transaction() as session:
-        session[MULTINET_COOKIE] = handled_user.multinet.session
-
-    # Graphs
-    resp = server.get(f"/api/workspaces/{workspace}/graphs")
-    assert resp.status_code == 200
+    with managed_user.login(server):
+        # Graphs
+        resp = server.get(f"/api/workspaces/{workspace}/graphs")
+        assert resp.status_code == 200
 
-    graphs = resp.json
-    assert len(graphs) == 1
-    assert graphs[0] == "miserables"
+        graphs = resp.json
+        assert len(graphs) == 1
+        assert graphs[0] == "miserables"
 
-    # Tables
-    resp = server.get(f"/api/workspaces/{workspace}/tables")
-    assert resp.status_code == 200
+        # Tables
+        resp = server.get(f"/api/workspaces/{workspace}/tables")
+        assert resp.status_code == 200
 
-    tables = resp.json
-    assert len(tables) == 2
-    assert "miserables_nodes" in tables
-    assert "miserables_links" in tables
+        tables = resp.json
+        assert len(tables) == 2
+        assert "miserables_nodes" in tables
+        assert "miserables_links" in tables

test/test_workspace.py

@@ -9,17 +9,17 @@
 )
 
 
-def test_present_workspace(handled_workspace):
+def test_present_workspace(managed_workspace):
     """Test that workspace caching works as expected on present workspaces."""
 
     # Assert that the cached response matches the actual response
-    assert workspace_mapping.__wrapped__(handled_workspace) == workspace_mapping(
-        handled_workspace
+    assert workspace_mapping.__wrapped__(managed_workspace) == workspace_mapping(
+        managed_workspace
     )
     workspace_mapping.cache_clear()
 
-    first_resp = workspace_mapping(handled_workspace)
-    second_resp = workspace_mapping(handled_workspace)
+    first_resp = workspace_mapping(managed_workspace)
+    second_resp = workspace_mapping(managed_workspace)
 
     # Assert that cached response is idempotent
     assert first_resp == second_resp
@@ -42,12 +42,12 @@ def test_absent_workspace():
     assert second_resp is None
 
 
-def test_workspace_create(handled_user):
+def test_workspace_create(managed_user):
     """Test that creating a workspace doesn't result in invalid caching."""
     workspace_name = uuid4().hex
 
     pre_create = workspace_mapping(workspace_name)
-    create_workspace(workspace_name, handled_user)
+    create_workspace(workspace_name, managed_user)
     post_create = workspace_mapping(workspace_name)
     post_create_exists = workspace_exists(workspace_name)