demo-backend/document-service/tests/test_document_router.py

from io import BytesIO

import pytest

from .test_payloads.payload_document_router import *


# upload file for consumer
@pytest.mark.parametrize("file, form_data", upload_document_payload)
def test_upload_document(file, form_data, test_client, consumer_header):
    file_data = BytesIO(file["file_content"])
    file_name = file["file_name"]
    content_type = file["content_type"]

    file = {"file": (file_name, file_data, content_type)}

    data = {
        "userId": form_data["user_id"],
        "applicationId": form_data["application_id"],
        "userProfileId": form_data["user_profile_id"],
        "documentTypeId": form_data["document_type_id"],
        "documentSubTypeId": form_data["document_sub_type_id"],
        "documentTitle": form_data["document_title"],
    }

    url = "/file/upload_file"
    response = test_client.post(url, data=data, files=file, headers=consumer_header)
    assert response.status_code == 200

    response = response.json()
    assert response["status"] == 200
    assert response["message"] == "File uploaded succesfully!"


# update existing file for consumer
@pytest.mark.parametrize("file, form_data", update_document_payload)
def test_update_document(file, form_data, test_client, consumer_header):
    document_id = 495
    file_data = BytesIO(file["file_content"])
    file_name = file["file_name"]
    content_type = file["content_type"]

    file = {"file": (file_name, file_data, content_type)}

    data = {
        "userId": form_data["user_id"],
        "applicationId": form_data["application_id"],
        "documentTitle": form_data["document_title"],
    }

    url = f"/file/update_file/{document_id}"
    response = test_client.put(url, data=data, files=file, headers=consumer_header)
    assert response.status_code == 200

    response = response.json()
    assert response["status"] == 200
    assert response["message"] == "File updated succesfully!"


# read file
def test_read_file(test_client, consumer_header):
    document_id = 495
    url = f"/file/read_file/{document_id}"
    response = test_client.get(url, headers=consumer_header)
    if response.status_code == 200:
        assert response.status_code == 200
    elif response.status_code == 500:
        response = response.json()
        assert response["message"] == "404: Document does not exist!"


# fetch users document list
def test_list_documents(test_client, consumer_header, application_id):
    url = f"/file/documents/{application_id}"
    response = test_client.get(url, headers=consumer_header)
    assert response.status_code == 200

    response = response.json()
    assert response["status"] == 200
    assert response["message"] == "Document types retrieved successfully"


# fetch users document list
def test_list_internal_documents(test_client, consumer_header, application_id):
    user_id = 425
    url = f"/file/documents/internal/{application_id}?user_id={user_id}"
    response = test_client.get(url, headers=consumer_header)
    assert response.status_code == 200

    response = response.json()
    assert response["status"] == 200
    assert response["message"] == "Document types retrieved successfully"


# delete existing document
def test_delete_document(test_client, consumer_header):
    document_id = 495
    url = f"/file/document/{document_id}"
    response = test_client.delete(url, headers=consumer_header)
    if response.status_code == 200:
        response = response.json()
        assert response["status"] == 200
        assert response["message"] == "User Document deleted successfully"
    elif response.status_code == 404:
        response = response.json()
        assert response["status"] == 200
        assert response["message"] == "No documents found"


# fetch logged-in user information
def test_get_user_info(test_client, consumer_header):
    url = "/file/user_info"
    response = test_client.get(url, headers=consumer_header)
    assert response.status_code == 200

    response = response.json()
    assert response["status"] == 200
    assert response["message"] == "User info fetched successfully"


# fetch document status
def test_get_document_status(test_client, consumer_header):
    url = "/file/document_status"
    response = test_client.get(url, headers=consumer_header)
    assert response.status_code == 200

    response = response.json()
    assert response["status"] == 200
    assert response["message"] == "Document status fetched successfully!"