webfortilog/tests/test_app.py

import io

from app import create_app


SAMPLE_LOG = (
    'v015xxxxdate=2024-05-01 time=10:00:00 policy="Prod Policy" '
    'http_method=GET http_host=example.com http_url="/login" '
    'http_refer="https://ref.example" service=edge backend_service=api '
    'msg="SQL injection blocked" signature_subclass=SQL signature_id=942100 '
    'owasp_top10=A03 match_location=body action=blocked severity_level=high\n'
    'v015xxxxdate=2024-05-02 time=11:00:00 policy="Prod Policy" '
    'http_method=POST http_host=example.com http_url="/checkout" '
    'http_refer="https://shop.example" service=edge backend_service=orders '
    'msg="XSS blocked" signature_subclass=XSS signature_id=941100 '
    'owasp_top10=A03 match_location=query action=monitored severity_level=medium\n'
)


def test_index_page_loads(client):
    response = client.get("/")

    assert response.status_code == 200
    assert b"WAF Log Converter" in response.data


def test_convert_returns_text_preview_and_download_link(client):
    log_file = io.BytesIO(SAMPLE_LOG.encode("utf-8"))
    response = client.post(
        "/convert",
        data={
            "mode": "vendor",
            "output_format": "text",
            "sort_by": "severity",
            "order": "desc",
            "policy_cs": "",
            "policy_ci": "prod",
            "severity_cs": "",
            "severity_ci": "",
            "log_file": (log_file, "sample.log"),
        },
        content_type="multipart/form-data",
    )
    log_file.close()

    assert response.status_code == 200
    assert b"Download export" in response.data
    assert b"--- record 1 ---" in response.data
    response.close()


def test_convert_full_mode_csv_preserves_union_order(client):
    log_file = io.BytesIO(SAMPLE_LOG.encode("utf-8"))
    response = client.post(
        "/convert",
        data={
            "mode": "full",
            "output_format": "csv",
            "sort_by": "datetime",
            "order": "asc",
            "policy_cs": "",
            "policy_ci": "",
            "severity_cs": "",
            "severity_ci": "",
            "log_file": (log_file, "sample.log"),
        },
        content_type="multipart/form-data",
    )
    log_file.close()

    assert response.status_code == 200
    assert b"TEXT" not in response.data
    assert b"Download export" in response.data
    response.close()


def test_convert_rejects_mutually_exclusive_filters(client):
    log_file = io.BytesIO(SAMPLE_LOG.encode("utf-8"))
    response = client.post(
        "/convert",
        data={
            "mode": "vendor",
            "output_format": "csv",
            "sort_by": "datetime",
            "order": "asc",
            "policy_cs": "A",
            "policy_ci": "a",
            "severity_cs": "",
            "severity_ci": "",
            "log_file": (log_file, "sample.log"),
        },
        content_type="multipart/form-data",
    )
    log_file.close()

    assert response.status_code == 400
    assert b"Policy filter must use either case-sensitive or case-insensitive match" in response.data
    response.close()


def test_download_route_returns_generated_file(client):
    log_file = io.BytesIO(SAMPLE_LOG.encode("utf-8"))
    convert_response = client.post(
        "/convert",
        data={
            "mode": "vendor",
            "output_format": "csv",
            "sort_by": "datetime",
            "order": "asc",
            "policy_cs": "",
            "policy_ci": "",
            "severity_cs": "",
            "severity_ci": "",
            "log_file": (log_file, "sample.log"),
        },
        content_type="multipart/form-data",
    )
    log_file.close()

    html = convert_response.data.decode("utf-8")
    marker = '/download/'
    start = html.index(marker) + len(marker)
    end = html.index('"', start)
    result_id = html[start:end]

    download_response = client.get(f"/download/{result_id}")

    assert download_response.status_code == 200
    assert download_response.headers["Content-Type"].startswith("text/csv")
    assert b"v015xxxxdate,time,policy" in download_response.data
    convert_response.close()
    download_response.close()


def test_default_upload_limit_is_100_mib(app):
    assert app.config["MAX_CONTENT_LENGTH"] == 100 * 1024 * 1024


def test_too_large_upload_returns_friendly_message(tmp_path):
    class SmallLimitConfig:
        TESTING = True
        SECRET_KEY = "test-secret"
        MAX_CONTENT_LENGTH = 128
        PREVIEW_RECORD_LIMIT = 5
        OUTPUT_DIRECTORY = tmp_path / "tiny-limit-outputs"

    app = create_app(SmallLimitConfig)
    client = app.test_client()
    log_file = io.BytesIO(SAMPLE_LOG.encode("utf-8"))

    response = client.post(
        "/convert",
        data={
            "mode": "vendor",
            "output_format": "text",
            "sort_by": "datetime",
            "order": "asc",
            "policy_cs": "",
            "policy_ci": "",
            "severity_cs": "",
            "severity_ci": "",
            "log_file": (log_file, "sample.log"),
        },
        content_type="multipart/form-data",
    )
    log_file.close()

    assert response.status_code == 413
    assert b"Maximum allowed size is 128 bytes." in response.data
    response.close()