add tests

Author: harminius

server/mergin/sync/public_api_v2_controller.py

@@ -10,7 +10,8 @@
 from mergin.sync.forms import project_name_validation
 from .files import ProjectFileSchema
 
-from .schemas import ProjectMemberSchema, ProjectSchemaV2
+from .schemas import ProjectMemberSchema
+from .schemas_v2 import ProjectSchema
 from .workspace import WorkspaceRole
 from ..app import db
 from ..auth import auth_required
@@ -134,7 +135,7 @@ def remove_project_collaborator(id, user_id):
 def get_project(id, files_at_version=None):
     """Get project info. Include list of files at specific version if requested."""
     project = require_project_by_uuid(id, ProjectPermissions.Read)
-    data = ProjectSchemaV2().dump(project)
+    data = ProjectSchema().dump(project)
     if files_at_version:
         pv = ProjectVersion.query.filter_by(
             project_id=project.id, name=ProjectVersion.from_v_name(files_at_version)

server/mergin/sync/schemas.py

@@ -405,40 +405,3 @@ class ProjectMemberSchema(Schema):
     project_role = fields.Enum(enum=ProjectRole, by_value=True)
     workspace_role = fields.Enum(enum=WorkspaceRole, by_value=True)
     role = fields.Enum(enum=ProjectRole, by_value=True)
-
-
-class ProjectSchemaV2(ma.SQLAlchemyAutoSchema):
-    id = fields.UUID()
-    name = fields.String()
-    version = fields.Function(lambda obj: ProjectVersion.to_v_name(obj.latest_version))
-    public = fields.Boolean()
-    size = fields.Integer(attribute="disk_usage")
-
-    created_at = DateTimeWithZ(attribute="created")
-    updated_at = DateTimeWithZ(attribute="updated")
-
-    workspace = fields.Function(
-        lambda obj: {"id": obj.workspace.id, "name": obj.workspace.name}
-    )
-    role = fields.Method("_role")
-
-    def _role(self, obj):
-        role = ProjectPermissions.get_user_project_role(obj, current_user)
-        if not role:
-            return None
-        return role.value
-
-    class Meta:
-        model = Project
-        load_instance = True
-        fields = (
-            "id",
-            "name",
-            "version",
-            "public",
-            "size",
-            "created_at",
-            "updated_at",
-            "workspace",
-            "role",
-        )

server/mergin/sync/schemas_v2.py

@@ -0,0 +1,50 @@
+# Copyright (C) Lutra Consulting Limited
+#
+# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
+
+from marshmallow import fields
+from flask_login import current_user
+
+from ..app import DateTimeWithZ, ma
+from .permissions import ProjectPermissions
+from .models import (
+    Project,
+    ProjectVersion,
+)
+
+
+class ProjectSchema(ma.SQLAlchemyAutoSchema):
+    id = fields.UUID()
+    name = fields.String()
+    version = fields.Function(lambda obj: ProjectVersion.to_v_name(obj.latest_version))
+    public = fields.Boolean()
+    size = fields.Integer(attribute="disk_usage")
+
+    created_at = DateTimeWithZ(attribute="created")
+    updated_at = DateTimeWithZ(attribute="updated")
+
+    workspace = fields.Function(
+        lambda obj: {"id": obj.workspace.id, "name": obj.workspace.name}
+    )
+    role = fields.Method("_role")
+
+    def _role(self, obj):
+        role = ProjectPermissions.get_user_project_role(obj, current_user)
+        if not role:
+            return None
+        return role.value
+
+    class Meta:
+        model = Project
+        load_instance = True
+        fields = (
+            "id",
+            "name",
+            "version",
+            "public",
+            "size",
+            "created_at",
+            "updated_at",
+            "workspace",
+            "role",
+        )

server/mergin/tests/test_public_api_v2.py

@@ -1,11 +1,22 @@
 # Copyright (C) Lutra Consulting Limited
 #
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
-from .utils import add_user
+from datetime import datetime
+
+from . import DEFAULT_USER
+from .utils import (
+    add_user,
+    login,
+    login_as_admin,
+    create_workspace,
+    create_project,
+    upload_file_to_project,
+)
 from ..app import db
 from mergin.sync.models import Project
 from tests import test_project, test_workspace_id
 
+from ..auth.models import User
 from ..config import Configuration
 from ..sync.models import ProjectRole
 
@@ -126,3 +137,69 @@ def test_project_members(client):
     # access provided by workspace role cannot be removed directly
     response = client.delete(url + f"/{user.id}")
     assert response.status_code == 404
+
+
+def test_get_project(client):
+    """Test get project info endpoint"""
+    admin = User.query.filter_by(username=DEFAULT_USER[0]).first()
+    test_workspace = create_workspace()
+    project = create_project("new_project", test_workspace, admin)
+    add_user("test_user", "ilovemergin")
+    login(client, "test_user", "ilovemergin")
+    # lack of permissions
+    response = client.get(f"v2/projects/{project.id}")
+    assert response.status_code == 403
+    # access public project
+    project.public = True
+    db.session.commit()
+    response = client.get(f"v2/projects/{project.id}")
+    assert response.status_code == 200
+    assert response.json["public"] is True
+    # project scheduled for deletion
+    login_as_admin(client)
+    project.public = False
+    project.removed_at = datetime.utcnow()
+    db.session.commit()
+    response = client.get(f"v2/projects/{project.id}")
+    assert response.status_code == 404
+    # success
+    project.removed_at = None
+    db.session.commit()
+    response = client.get(f"v2/projects/{project.id}")
+    assert response.status_code == 200
+    expected_keys = {
+        "id",
+        "name",
+        "workspace",
+        "role",
+        "version",
+        "created_at",
+        "updated_at",
+        "public",
+        "size",
+    }
+    assert expected_keys == response.json.keys()
+    # create new versions
+    files = ["test.txt", "test3.txt", "test.qgs"]
+    for file in files:
+        upload_file_to_project(project, file, client)
+    # project version does not exist
+    response = client.get(
+        f"v2/projects/{project.id}?files_at_version=v{project.latest_version+1}"
+    )
+    assert response.status_code == 404
+    # files
+    response = client.get(
+        f"v2/projects/{project.id}?files_at_version=v{project.latest_version-2}"
+    )
+    assert response.status_code == 200
+    assert len(response.json["files"]) == 1
+    assert any(resp_files["path"] == files[0] for resp_files in response.json["files"])
+    assert not any(
+        resp_files["path"] == files[1] for resp_files in response.json["files"]
+    )
+    response = client.get(
+        f"v2/projects/{project.id}?files_at_version=v{project.latest_version}"
+    )
+    assert len(response.json["files"]) == 3
+    assert {f["path"] for f in response.json["files"]} == set(files)