Add unit tests

.devcontainer/devcontainer.json

@@ -9,10 +9,11 @@
             "extensions": [
                 "ms-python.black-formatter",
                 "ms-python.python",
-                "ms-python.isort"
+                "ms-python.isort",
+                "emeraldwalk.RunOnSave"
             ]
         }
     },
-    "postStartCommand": "pip3 --disable-pip-version-check --no-cache-dir install -r requirements.txt --break-system-packages",
+    "postStartCommand": "pip3 --disable-pip-version-check --no-cache-dir install -r requirements.txt --break-system-packages && python3 -m pytest tests",
     "remoteUser": "vscode"
 }

.github/workflows/pytest.yml

@@ -0,0 +1,47 @@
+name: Pytest CI
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+permissions: {}
+
+jobs:
+  test:
+    name: Pytest on ${{ matrix.os }} with Python ${{ matrix.python-version }}
+    runs-on: ${{ matrix.os }}
+
+    permissions:
+      contents: read
+      packages: read
+      statuses: write
+
+    strategy:
+      matrix:
+        os: [ubuntu-latest, macos-latest, windows-latest]
+        python-version: ["3.x"]
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: "pip"
+
+      - name: Install dependencies
+        run: |
+          pip install -r requirements.txt
+          pip install pytest pytest-cov
+
+      - name: Run Pytest (Linux/macOS)
+        if: runner.os != 'Windows'
+        run: python -m pytest --doctest-modules --junitxml=junit/test-results.xml --cov=com --cov-report=xml --cov-report=html
+
+      - name: Run Pytest (Windows)
+        if: runner.os == 'Windows'
+        run: python -m pytest --doctest-modules --junitxml=junit/test-results.xml --cov=com --cov-report=xml --cov-report=html

.gitignore

@@ -2,7 +2,7 @@
 *.pem
 
 # Upload folder
-images/
+pictures/
 
 # Environment variables file
 .env

.vscode/settings.json

@@ -5,6 +5,10 @@
     "files.insertFinalNewline": true,
     "files.trimFinalNewlines": true,
     "files.trimTrailingWhitespace": true,
+    "files.exclude": {
+        "**/__pycache__": true,
+        "**/.pytest_cache": true
+    },
     "[python]": {
         "editor.rulers": [
             88
@@ -18,5 +22,14 @@
     "isort.args": [
         "--profile",
         "black"
-    ]
+    ],
+    "emeraldwalk.runonsave": {
+        "commands": [
+            {
+                "match": "tests[/\\\\](.*[/\\\\])?test_.*\\.py$",
+                "cmd": "python3 -m pytest '${relativeFile}' -v",
+                "autoShowOutputPanel": "error"
+            }
+        ]
+    }
 }

config.py

@@ -23,22 +23,16 @@ class Config:
     JWT_REFRESH_TOKEN_EXPIRY = timedelta(days=30)
 
     # Image upload folder
-    IMAGES_FOLDER = os.path.join(os.getcwd(), "images")
+    IMAGES_FOLDER = os.path.join(os.getcwd(), "pictures")
     MAX_CONTENT_LENGTH = 1024 * 1024  # 1 MB
 
 
-# Image upload folder
-PICTURE_FOLDER = os.path.join(os.getcwd(), "picture")
-
-if not os.path.exists(PICTURE_FOLDER):
-    os.makedirs(PICTURE_FOLDER)
+if not os.path.exists(Config.IMAGES_FOLDER):
+    os.makedirs(Config.IMAGES_FOLDER)
 
 # Pagination configuration
 DEFAULT_PAGE_SIZE = 10
 
-if not os.path.exists(Config.IMAGES_FOLDER):
-    os.makedirs(Config.IMAGES_FOLDER)
-
 limiter = Limiter(
     key_func=get_remote_address,
     default_limits=["1000 per day", "200 per hour", "30 per minute", "3 per second"],

jwt_helper.py

@@ -11,9 +11,7 @@
 
 
 class TokenError(Exception):
-    """
-    Custom exception for token-related errors.
-    """
+    """Custom exception for token-related errors."""
 
     def __init__(self, message, status_code):
         super().__init__(message)
@@ -22,9 +20,7 @@ def __init__(self, message, status_code):
 
 
 def generate_access_token(person_id: int) -> str:
-    """
-    Generate a short-lived JWT access token for a user.
-    """
+    """Generate a short-lived JWT access token for a user."""
     payload = {
         "person_id": person_id,
         "exp": datetime.now(timezone.utc) + JWT_ACCESS_TOKEN_EXPIRY,  # Expiration
@@ -35,9 +31,7 @@ def generate_access_token(person_id: int) -> str:
 
 
 def generate_refresh_token(person_id: int) -> str:
-    """
-    Generate a long-lived refresh token for a user.
-    """
+    """Generate a long-lived refresh token for a user."""
     payload = {
         "person_id": person_id,
         "exp": datetime.now(timezone.utc) + JWT_REFRESH_TOKEN_EXPIRY,
@@ -48,19 +42,15 @@ def generate_refresh_token(person_id: int) -> str:
 
 
 def extract_token_from_header() -> str:
-    """
-    Extract the Bearer token from the Authorization header.
-    """
+    """Extract the Bearer token from the Authorization header."""
     auth_header = request.headers.get("Authorization")
     if not auth_header or not auth_header.startswith("Bearer "):
         raise TokenError("Token is missing or improperly formatted", 401)
     return auth_header.split("Bearer ")[1]
 
 
 def verify_token(token: str, required_type: str) -> dict:
-    """
-    Verify and decode a JWT token.
-    """
+    """Verify and decode a JWT token."""
     try:
         decoded = jwt.decode(token, JWT_SECRET_KEY, algorithms=["HS256"])
         if decoded.get("token_type") != required_type:
@@ -73,9 +63,7 @@ def verify_token(token: str, required_type: str) -> dict:
 
 
 def token_required(f):
-    """
-    Decorator to protect routes by requiring a valid token.
-    """
+    """Decorator to protect routes by requiring a valid token."""
 
     @wraps(f)
     def decorated(*args, **kwargs):

requirements.txt

@@ -7,4 +7,5 @@ flask-cors>=4.0.1
 PyMySQL>=1.1.1
 requests>=2.32.3
 waitress>=3.0.0
+pytest>=8.3.5
 python-dotenv>=1.0.1

routes/authentication.py

@@ -15,6 +15,7 @@
     encrypt_email,
     hash_email,
     hash_password,
+    validate_email,
     validate_password,
     verify_password,
 )
@@ -47,6 +48,9 @@ def register():
     if not name or not email or not password:
         return jsonify(message="Username, email, and password are required"), 400
 
+    if not validate_email(email):
+        return jsonify(message="Invalid email address"), 400
+
     if not validate_password(password):
         return jsonify(message="Password does not meet security requirements"), 400
 

routes/picture.py

@@ -2,7 +2,7 @@
 
 from flask import Blueprint, jsonify, request, send_from_directory
 
-from config import PICTURE_FOLDER
+from config import Config
 from jwt_helper import token_required
 from utility import database_cursor
 
@@ -50,7 +50,7 @@ def get_pictures_by_author(author_id):
 
 @picture_blueprint.route("/<path:filename>", methods=["GET"])
 def get_picture(filename):
-    return send_from_directory(PICTURE_FOLDER, filename)
+    return send_from_directory(Config.IMAGES_FOLDER, filename)
 
 
 @picture_blueprint.route("", methods=["POST"])
@@ -80,8 +80,8 @@ def upload_picture():
     with database_cursor() as cursor:
         cursor.callproc(procedure, (hexname, request.person_id))
 
-    fullpath = os.path.normpath(os.path.join(PICTURE_FOLDER, hexname))
-    if not fullpath.startswith(PICTURE_FOLDER):
+    fullpath = os.path.abspath(os.path.join(Config.IMAGES_FOLDER, hexname))
+    if os.path.commonpath([fullpath, Config.IMAGES_FOLDER]) != Config.IMAGES_FOLDER:
         return jsonify({"error": "Invalid file path"}), 400
     file.save(fullpath)
 

tests/test_jwt/conftest.py

@@ -0,0 +1,29 @@
+import pytest
+from flask import Flask
+
+from jwt_helper import generate_access_token
+
+
+@pytest.fixture
+def client(app: Flask):
+    """Create a test client for the Flask application"""
+    with app.test_client() as client:
+        yield client
+
+
+@pytest.fixture
+def sample_person_id() -> int:
+    """Provide a sample person ID for testing"""
+    return 12345
+
+
+@pytest.fixture
+def sample_token():
+    """Provide a sample token for testing"""
+    return "mock_token_123"
+
+
+@pytest.fixture
+def sample_access_token(sample_person_id):
+    """Provide a sample access token for testing"""
+    return generate_access_token(sample_person_id)

tests/test_jwt/test_extract_token_from_header.py

@@ -0,0 +1,42 @@
+import pytest
+from flask import Flask
+
+from jwt_helper import TokenError, extract_token_from_header
+
+app = Flask(__name__)
+
+
+def test_extract_token_valid(sample_token):
+    """Test extracting a valid Bearer token."""
+    with app.test_request_context(headers={"Authorization": f"Bearer {sample_token}"}):
+        assert extract_token_from_header() == sample_token
+
+
+def test_extract_token_missing():
+    """Test missing Authorization header."""
+    with app.test_request_context(headers={}):
+        with pytest.raises(
+            TokenError, match="Token is missing or improperly formatted"
+        ) as excinfo:
+            extract_token_from_header()
+        assert excinfo.value.status_code == 401
+
+
+def test_extract_token_invalid_format():
+    """Test an improperly formatted Authorization header."""
+    with app.test_request_context(headers={"Authorization": "InvalidTokenFormat"}):
+        with pytest.raises(
+            TokenError, match="Token is missing or improperly formatted"
+        ) as excinfo:
+            extract_token_from_header()
+        assert excinfo.value.status_code == 401
+
+
+def test_extract_token_no_bearer(sample_token):
+    """Test Authorization header without 'Bearer ' prefix."""
+    with app.test_request_context(headers={"Authorization": f"Basic {sample_token}"}):
+        with pytest.raises(
+            TokenError, match="Token is missing or improperly formatted"
+        ) as excinfo:
+            extract_token_from_header()
+        assert excinfo.value.status_code == 401

tests/test_jwt/test_generate_access_token.py

@@ -0,0 +1,40 @@
+import jwt
+
+from jwt_helper import JWT_ACCESS_TOKEN_EXPIRY, JWT_SECRET_KEY
+
+
+def test_access_token_type(sample_access_token):
+    """Ensure generate_access_token returns a string"""
+    assert isinstance(sample_access_token, str)
+
+
+def test_decoded_access_token(sample_person_id, sample_access_token):
+    """
+    Ensure the generated access token can be decoded and contains the correct payload
+    - Check if the payload contains the correct person ID
+    - Check if the token has an expiration time
+    - Check if the token type is 'access'
+    """
+    decoded_payload = jwt.decode(
+        sample_access_token, JWT_SECRET_KEY, algorithms=["HS256"]
+    )
+
+    assert decoded_payload["person_id"] == sample_person_id
+    assert "exp" in decoded_payload
+    assert decoded_payload["token_type"] == "access"
+
+
+def test_access_token_expiration(sample_access_token):
+    """
+    Ensure the generated access token has a valid expiration time
+    - Check if the expiration time is greater than 0
+    - Check if the expiration time is greater than the issued at time
+    - Check if the token is not expired
+    """
+    decoded_payload = jwt.decode(
+        sample_access_token, JWT_SECRET_KEY, algorithms=["HS256"]
+    )
+
+    assert decoded_payload["exp"] > 0
+    assert decoded_payload["exp"] > decoded_payload["iat"]
+    assert decoded_payload["exp"] > JWT_ACCESS_TOKEN_EXPIRY.total_seconds()