Add unit tests for generating the bucket policy statements

#9

.gitlab-ci.yml

@@ -89,6 +89,18 @@ e2e-test-job:   # This job runs in the test stage.
     reports:
       junit: $BASE_DIR/e2e-report.xml
 
+unit-test-job:   # This job runs in the test stage.
+  stage: test    # It only starts when the job in the build stage completes successfully.
+  script:
+    - pytest --junitxml=unit-report.xml --noconftest --cov=app --cov-report=term-missing app/tests/unit
+    - mkdir coverage-unit
+    - mv .coverage coverage-unit
+  artifacts:
+    paths:
+      - $BASE_DIR/coverage-unit/.coverage
+    reports:
+      junit: $BASE_DIR/unit-report.xml
+
 combine-test-coverage-job:   # Combine coverage reports from different test jobs
   stage: test
   needs:
@@ -96,8 +108,10 @@ combine-test-coverage-job:   # Combine coverage reports from different test jobs
       artifacts: true
     - job: "integration-test-job"
       artifacts: true
+    - job: "unit-test-job"
+      artifacts: true
   script:
-    - coverage combine coverage-e2e/.coverage coverage-integration/.coverage
+    - coverage combine coverage-e2e/.coverage coverage-integration/.coverage coverage-unit
     - coverage report
     - cd ..
     - coverage xml --data-file=$BASE_DIR/.coverage -o coverage.xml

ProxyAPI/app/tests/unit/test_bucket_permission_scheme.py

+from datetime import datetime
+
+import pytest
+
+from app.models.bucket_permission import PermissionEnum
+from app.schemas.bucket_permission import BucketPermission
+from app.tests.utils.utils import random_lower_string
+
+
+class _TestPermissionPolicy:
+    @pytest.fixture(scope="function")
+    def random_base_permission(self) -> BucketPermission:
+        """
+        Generate a base READ bucket permission schema.
+        """
+        return BucketPermission(
+            username=random_lower_string(), bucket_name=random_lower_string(), permission=PermissionEnum.READ
+        )
+
+
+class TestPermissionPolicyPermissionType(_TestPermissionPolicy):
+    def test_READ_permission(self, random_base_permission: BucketPermission) -> None:
+        """
+        Test for converting a READ Permission into a bucket policy statement.
+
+        Parameters
+        ----------
+        random_base_permission : app.schemas.bucket_permission.BucketPermission
+            Random base bucket permission for testing. pytest fixture.
+        """
+        uid = random_lower_string()
+        stmts = random_base_permission.map_to_bucket_policy_statement(user_id=uid)
+        assert len(stmts) == 2
+
+        object_stmt = stmts[0]
+        assert object_stmt["Effect"] == "Allow"
+        assert object_stmt["Sid"] == random_base_permission.to_hash(uid)
+        assert object_stmt["Principal"]["AWS"] == f"arn:aws:iam:::user/{random_base_permission.username}"
+        assert object_stmt["Resource"] == f"arn:aws:s3:::{random_base_permission.bucket_name}/*"
+        with pytest.raises(KeyError):
+            assert object_stmt["Condition"]
+        assert len(object_stmt["Action"]) == 1
+        assert object_stmt["Action"][0] == "s3:GetObject"
+
+        bucket_stmt = stmts[1]
+        assert bucket_stmt["Sid"] == random_base_permission.to_hash(uid)
+        assert bucket_stmt["Effect"] == "Allow"
+        assert bucket_stmt["Principal"]["AWS"] == f"arn:aws:iam:::user/{random_base_permission.username}"
+        assert bucket_stmt["Resource"] == f"arn:aws:s3:::{random_base_permission.bucket_name}"
+        with pytest.raises(KeyError):
+            assert bucket_stmt["Condition"]
+        assert len(bucket_stmt["Action"]) == 1
+        assert bucket_stmt["Action"][0] == "s3:ListBucket"
+
+    def test_WRITE_permission(self, random_base_permission: BucketPermission) -> None:
+        """
+        Test for converting a WRITE Permission into a bucket policy statement.
+
+        Parameters
+        ----------
+        random_base_permission : app.schemas.bucket_permission.BucketPermission
+            Random base bucket permission for testing. pytest fixture.
+        """
+        random_base_permission.permission = PermissionEnum.WRITE
+        stmts = random_base_permission.map_to_bucket_policy_statement(user_id=random_lower_string())
+        assert len(stmts) == 1
+
+        object_stmt = stmts[0]
+        with pytest.raises(KeyError):
+            assert object_stmt["Condition"]
+        assert len(object_stmt["Action"]) == 2
+        assert "s3:PutObject" in object_stmt["Action"]
+        assert "s3:DeleteObject" in object_stmt["Action"]
+
+    def test_READWRITE_permission(self, random_base_permission: BucketPermission) -> None:
+        """
+        Test for converting a READWRITE Permission into a bucket policy statement.
+
+        Parameters
+        ----------
+        random_base_permission : app.schemas.bucket_permission.BucketPermission
+            Random base bucket permission for testing. pytest fixture.
+        """
+        random_base_permission.permission = PermissionEnum.READWRITE
+        stmts = random_base_permission.map_to_bucket_policy_statement(user_id=random_lower_string())
+        assert len(stmts) == 2
+
+        object_stmt = stmts[0]
+        with pytest.raises(KeyError):
+            assert object_stmt["Condition"]
+        assert len(object_stmt["Action"]) == 3
+        assert "s3:PutObject" in object_stmt["Action"]
+        assert "s3:DeleteObject" in object_stmt["Action"]
+        assert "s3:GetObject" in object_stmt["Action"]
+
+        bucket_stmt = stmts[1]
+        with pytest.raises(KeyError):
+            assert bucket_stmt["Condition"]
+        assert len(bucket_stmt["Action"]) == 1
+        assert bucket_stmt["Action"][0] == "s3:ListBucket"
+
+
+class TestPermissionPolicyCondition(_TestPermissionPolicy):
+    def test_to_timestamp_condition(self, random_base_permission: BucketPermission) -> None:
+        """
+        Test for converting a READ Permission with end time condition into a bucket policy statement.
+
+        Parameters
+        ----------
+        random_base_permission : app.schemas.bucket_permission.BucketPermission
+            Random base bucket permission for testing. pytest fixture.
+        """
+        time = datetime.now()
+        random_base_permission.to_timestamp = time
+
+        stmts = random_base_permission.map_to_bucket_policy_statement(user_id=random_lower_string())
+        assert len(stmts) == 2
+
+        object_stmt = stmts[0]
+        assert object_stmt["Condition"]
+        assert object_stmt["Condition"]["DateLessThan"]["aws:CurrentTime"] == time.strftime("%Y-%m-%dT%H:%M:%SZ")
+        with pytest.raises(KeyError):
+            assert object_stmt["Condition"]["DateGreaterThan"]
+
+        bucket_stmt = stmts[1]
+        assert bucket_stmt["Condition"]
+        assert bucket_stmt["Condition"]["DateLessThan"]["aws:CurrentTime"] == time.strftime("%Y-%m-%dT%H:%M:%SZ")
+        with pytest.raises(KeyError):
+            assert bucket_stmt["Condition"]["DateGreaterThan"]
+
+    def test_from_timestamp_condition(self, random_base_permission: BucketPermission) -> None:
+        """
+        Test for converting a READ Permission with start time condition into a bucket policy statement.
+
+        Parameters
+        ----------
+        random_base_permission : app.schemas.bucket_permission.BucketPermission
+            Random base bucket permission for testing. pytest fixture.
+        """
+        time = datetime.now()
+        random_base_permission.from_timestamp = time
+
+        stmts = random_base_permission.map_to_bucket_policy_statement(user_id=random_lower_string())
+        assert len(stmts) == 2
+
+        object_stmt = stmts[0]
+        assert object_stmt["Condition"]
+        assert object_stmt["Condition"]["DateGreaterThan"]["aws:CurrentTime"] == time.strftime("%Y-%m-%dT%H:%M:%SZ")
+        with pytest.raises(KeyError):
+            assert object_stmt["Condition"]["DateLessThan"]
+
+        bucket_stmt = stmts[1]
+        assert bucket_stmt["Condition"]
+        assert bucket_stmt["Condition"]["DateGreaterThan"]["aws:CurrentTime"] == time.strftime("%Y-%m-%dT%H:%M:%SZ")
+        with pytest.raises(KeyError):
+            assert bucket_stmt["Condition"]["DateLessThan"]
+
+    def test_file_prefix_condition(self, random_base_permission: BucketPermission) -> None:
+        """
+        Test for converting a READ Permission with file prefix condition into a bucket policy statement.
+
+        Parameters
+        ----------
+        random_base_permission : app.schemas.bucket_permission.BucketPermission
+            Random base bucket permission for testing. pytest fixture.
+        """
+        random_base_permission.file_prefix = random_lower_string(length=8) + "/" + random_lower_string(length=8) + "/"
+
+        stmts = random_base_permission.map_to_bucket_policy_statement(user_id=random_lower_string())
+        assert len(stmts) == 2
+
+        object_stmt = stmts[0]
+        assert (
+            object_stmt["Resource"]
+            == f"arn:aws:s3:::{random_base_permission.bucket_name}/{random_base_permission.file_prefix}*"
+        )
+        with pytest.raises(KeyError):
+            assert object_stmt["Condition"]
+
+        bucket_stmt = stmts[1]
+        assert bucket_stmt["Condition"]
+        assert bucket_stmt["Condition"]["StringLike"]["s3:prefix"] == random_base_permission.file_prefix + "*"