jss-devsecops-challenge/tests/test_app.py

import pytest
import json
from unittest.mock import patch, MagicMock
from app.server import app


@pytest.fixture
def client():
    app.config["TESTING"] = True
    with app.test_client() as c:
        yield c


class TestLogAuth:
    def test_missing_body(self, client):
        resp = client.post("/log/auth", content_type="application/json")
        assert resp.status_code == 400
        assert resp.get_json()["message"] == "bad_request"

    def test_invalid_device_type(self, client):
        resp = client.post(
            "/log/auth",
            data=json.dumps({"userKey": "user1", "deviceType": "Nokia"}),
            content_type="application/json",
        )
        assert resp.status_code == 400

    def test_invalid_user_key(self, client):
        resp = client.post(
            "/log/auth",
            data=json.dumps({"userKey": "", "deviceType": "iOS"}),
            content_type="application/json",
        )
        assert resp.status_code == 400

    def test_user_key_injection(self, client):
        resp = client.post(
            "/log/auth",
            data=json.dumps({"userKey": "'; DROP TABLE--", "deviceType": "iOS"}),
            content_type="application/json",
        )
        assert resp.status_code == 400

    @patch("app.server.get_db_connection")
    def test_success(self, mock_conn, client):
        mock_cursor = MagicMock()
        mock_conn.return_value.cursor.return_value = mock_cursor

        resp = client.post(
            "/log/auth",
            data=json.dumps({"userKey": "user1", "deviceType": "iOS"}),
            content_type="application/json",
        )
        assert resp.status_code == 200
        assert resp.get_json()["message"] == "success"

    @patch("app.server.get_db_connection")
    def test_db_failure(self, mock_conn, client):
        mock_conn.side_effect = Exception("db error")

        resp = client.post(
            "/log/auth",
            data=json.dumps({"userKey": "user1", "deviceType": "Android"}),
            content_type="application/json",
        )
        assert resp.status_code == 400


class TestLogAuthStatistics:
    def test_invalid_device_type(self, client):
        resp = client.get("/log/auth/statistics?deviceType=Nokia")
        assert resp.status_code == 400
        assert resp.get_json()["count"] == -1

    def test_missing_device_type(self, client):
        resp = client.get("/log/auth/statistics")
        assert resp.status_code == 400

    @patch("app.server.get_db_connection")
    def test_success(self, mock_conn, client):
        mock_cursor = MagicMock()
        mock_cursor.fetchone.return_value = (5,)
        mock_conn.return_value.cursor.return_value = mock_cursor

        resp = client.get("/log/auth/statistics?deviceType=iOS")
        assert resp.status_code == 200
        data = resp.get_json()
        assert data["deviceType"] == "iOS"
        assert data["count"] == 5

    @patch("app.server.get_db_connection")
    def test_db_failure(self, mock_conn, client):
        mock_conn.side_effect = Exception("connection refused")

        resp = client.get("/log/auth/statistics?deviceType=Android")
        assert resp.status_code == 500
        assert resp.get_json()["count"] == -1


class TestDeviceRegister:
    def test_missing_body(self, client):
        resp = client.post("/device/register", content_type="application/json")
        assert resp.status_code == 400
        assert resp.get_json()["statusCode"] == 400

    def test_invalid_device_type(self, client):
        resp = client.post(
            "/device/register",
            data=json.dumps({"userKey": "user1", "deviceType": "Fridge"}),
            content_type="application/json",
        )
        assert resp.status_code == 400

    def test_empty_user_key(self, client):
        resp = client.post(
            "/device/register",
            data=json.dumps({"userKey": "", "deviceType": "iOS"}),
            content_type="application/json",
        )
        assert resp.status_code == 400

    def test_special_chars_rejected(self, client):
        resp = client.post(
            "/device/register",
            data=json.dumps({"userKey": "<script>alert(1)</script>", "deviceType": "iOS"}),
            content_type="application/json",
        )
        assert resp.status_code == 400

    @patch("app.server.get_db_connection")
    def test_success(self, mock_conn, client):
        mock_cursor = MagicMock()
        mock_conn.return_value.cursor.return_value = mock_cursor

        resp = client.post(
            "/device/register",
            data=json.dumps({"userKey": "user1", "deviceType": "Android"}),
            content_type="application/json",
        )
        assert resp.status_code == 200
        assert resp.get_json()["statusCode"] == 200

    @patch("app.server.get_db_connection")
    def test_db_failure(self, mock_conn, client):
        mock_conn.side_effect = Exception("db down")

        resp = client.post(
            "/device/register",
            data=json.dumps({"userKey": "user1", "deviceType": "Watch"}),
            content_type="application/json",
        )
        assert resp.status_code == 400

    @patch("app.server.get_db_connection")
    def test_all_valid_device_types(self, mock_conn, client):
        mock_cursor = MagicMock()
        mock_conn.return_value.cursor.return_value = mock_cursor
        for dt in ["iOS", "Android", "Watch", "TV", "Tablet", "Desktop", "IoT"]:
            resp = client.post(
                "/device/register",
                data=json.dumps({"userKey": "user1", "deviceType": dt}),
                content_type="application/json",
            )
            assert resp.status_code == 200