Stirling-PDF/testing/cucumber/features/steps/folders_step_definitions.py

"""
Step definitions for the server-side folders feature
(folders_and_files.feature).

Covers:
  - POST/PATCH/DELETE/GET /api/v1/storage/folders[/{id}]
  - POST /api/v1/storage/files  (multipart upload)
  - PATCH /api/v1/storage/files/{id}/folder
  - GET /api/v1/storage/files

Authentication: uses the JWT token stashed in context.jwt_token by the
existing auth step "Given I am logged in as admin".

State stashing per scenario:
  context.folders_by_name  -> { display_name: uuid }
  context.uploaded_files   -> { filename: numeric file id }
  context.response         -> the most recent requests.Response

The "Name.id" placeholder convention in the Gherkin lets us write
human-readable scenarios while resolving to real UUIDs at step time.
"""

import io

import requests
from behave import given, then, when
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas

BASE_URL = "http://localhost:8080"
FOLDERS_URL = f"{BASE_URL}/api/v1/storage/folders"
FILES_URL = f"{BASE_URL}/api/v1/storage/files"
HTTP_TIMEOUT = 30


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _jwt_headers(context):
    token = getattr(context, "jwt_token", None)
    assert token, (
        "No JWT token in context. Use 'Given I am logged in as admin' first."
    )
    return {"Authorization": f"Bearer {token}"}


def _json_jwt_headers(context):
    return {**_jwt_headers(context), "Content-Type": "application/json"}


def _ensure_folders_dict(context):
    if not hasattr(context, "folders_by_name"):
        context.folders_by_name = {}


def _ensure_files_dict(context):
    if not hasattr(context, "uploaded_files"):
        context.uploaded_files = {}


def _resolve_folder_id(context, ref):
    """
    Resolve a Gherkin placeholder to a folder id.

    Conventions:
      'null'    -> Python None  (means "no folder" / root)
      'Name.id' -> uuid stashed under 'Name' in folders_by_name
      anything else -> returned unchanged (raw UUID string)
    """
    if ref is None or ref == "null":
        return None
    if ref.endswith(".id"):
        name = ref[:-3]
        _ensure_folders_dict(context)
        assert name in context.folders_by_name, (
            f"No folder named {name!r} stashed; available: "
            f"{list(context.folders_by_name)}"
        )
        return context.folders_by_name[name]
    return ref


def _generate_pdf_bytes():
    buffer = io.BytesIO()
    c = canvas.Canvas(buffer, pagesize=letter)
    c.drawString(100, 700, "behave folder fixture")
    c.showPage()
    c.save()
    return buffer.getvalue()


def _create_folder(context, name, parent_id=None):
    payload = {"name": name}
    if parent_id is not None:
        payload["parentFolderId"] = parent_id
    response = requests.post(
        FOLDERS_URL,
        headers=_json_jwt_headers(context),
        json=payload,
        timeout=HTTP_TIMEOUT,
    )
    return response


# ---------------------------------------------------------------------------
# GIVEN - setup
# ---------------------------------------------------------------------------


@given("I clear all my folders")
def step_clear_all_folders(context):
    """
    Tear-down for prior runs. Folder DELETE cascades to children, so
    a single pass over the top-level list is enough.

    A 404 from the GET means the folder endpoint isn't deployed on this
    build, which would render every scenario unrunnable - surface that
    early as a clear assertion rather than letting individual steps fail
    with cryptic errors.
    """
    response = requests.get(
        FOLDERS_URL, headers=_jwt_headers(context), timeout=HTTP_TIMEOUT
    )
    assert response.status_code in (200, 204), (
        f"Folder list returned {response.status_code} during teardown - "
        f"is the proprietary storage-folders module deployed? Body: "
        f"{response.text[:300]}"
    )
    for folder in response.json() or []:
        # Top-level folders only; cascade does the rest. Skipping children
        # avoids 404s on the second pass when their parent has already gone.
        if folder.get("parentFolderId") is None:
            requests.delete(
                f"{FOLDERS_URL}/{folder['id']}",
                headers=_jwt_headers(context),
                timeout=HTTP_TIMEOUT,
            )
    _ensure_folders_dict(context)
    context.folders_by_name.clear()


@given('a folder "{name}" exists')
def step_folder_exists(context, name):
    response = _create_folder(context, name)
    assert response.status_code == 201, (
        f"Could not create folder {name!r} during Given step: "
        f"{response.status_code} {response.text}"
    )
    _ensure_folders_dict(context)
    context.folders_by_name[name] = response.json()["id"]
    context.response = response


@given('a folder "{name}" exists under "{parent}"')
def step_folder_exists_under(context, name, parent):
    _ensure_folders_dict(context)
    parent_id = context.folders_by_name.get(parent)
    assert parent_id, (
        f"Parent folder {parent!r} not created yet; available: "
        f"{list(context.folders_by_name)}"
    )
    response = _create_folder(context, name, parent_id=parent_id)
    assert response.status_code == 201, (
        f"Could not create child folder {name!r}: "
        f"{response.status_code} {response.text}"
    )
    context.folders_by_name[name] = response.json()["id"]
    context.response = response


@given('I upload a file as "{filename}"')
@when('I upload a file as "{filename}"')
def step_upload_file(context, filename):
    pdf_bytes = _generate_pdf_bytes()
    response = requests.post(
        FILES_URL,
        headers=_jwt_headers(context),  # multipart sets its own Content-Type
        files={"file": (filename, pdf_bytes, "application/pdf")},
        timeout=HTTP_TIMEOUT,
    )
    context.response = response
    if response.status_code == 200:
        _ensure_files_dict(context)
        context.uploaded_files[filename] = response.json()["id"]


# ---------------------------------------------------------------------------
# WHEN - folder mutations
# ---------------------------------------------------------------------------


@when('I POST a folder with name "{name}"')
def step_post_folder(context, name):
    response = _create_folder(context, name)
    context.response = response
    if response.status_code == 201:
        _ensure_folders_dict(context)
        context.folders_by_name[name] = response.json()["id"]


@when("I POST a folder with a 256-character name")
def step_post_oversize_folder(context):
    """256 chars trips the @Size(max = 255) validator on CreateFolderRequest."""
    long_name = "x" * 256
    response = _create_folder(context, long_name)
    context.response = response


@when('I PATCH folder "{name}" with name "{new_name}"')
def step_patch_folder_name(context, name, new_name):
    folder_id = _resolve_folder_id(context, f"{name}.id")
    response = requests.patch(
        f"{FOLDERS_URL}/{folder_id}",
        headers=_json_jwt_headers(context),
        json={"name": new_name},
        timeout=HTTP_TIMEOUT,
    )
    context.response = response
    if response.status_code == 200:
        # Update the stash so later steps can still refer to it by either
        # the old or new name in the same scenario.
        context.folders_by_name[new_name] = folder_id


@when('I PATCH folder "{name}" with parentFolderId = "{parent_ref}"')
def step_patch_folder_parent(context, name, parent_ref):
    folder_id = _resolve_folder_id(context, f"{name}.id")
    parent_id = _resolve_folder_id(context, parent_ref)
    # `reparent: true` is required - the UpdateFolderRequest treats a
    # missing reparent flag as "do not change parent" even when
    # parentFolderId is present. Sending the flag explicitly is what the
    # service contract expects.
    response = requests.patch(
        f"{FOLDERS_URL}/{folder_id}",
        headers=_json_jwt_headers(context),
        json={"reparent": True, "parentFolderId": parent_id},
        timeout=HTTP_TIMEOUT,
    )
    context.response = response


@when('I DELETE folder "{name}"')
def step_delete_folder(context, name):
    folder_id = _resolve_folder_id(context, f"{name}.id")
    response = requests.delete(
        f"{FOLDERS_URL}/{folder_id}",
        headers=_jwt_headers(context),
        timeout=HTTP_TIMEOUT,
    )
    context.response = response


@when("I list folders")
def step_list_folders(context):
    response = requests.get(
        FOLDERS_URL, headers=_jwt_headers(context), timeout=HTTP_TIMEOUT
    )
    context.response = response


# ---------------------------------------------------------------------------
# WHEN - file - folder placement
# ---------------------------------------------------------------------------


@when('I PATCH file "{filename}" with folderId = "{folder_ref}"')
@given('I PATCH file "{filename}" with folderId = "{folder_ref}"')
def step_patch_file_folder(context, filename, folder_ref):
    _ensure_files_dict(context)
    file_id = context.uploaded_files.get(filename)
    assert file_id, (
        f"File {filename!r} not uploaded yet; available: "
        f"{list(context.uploaded_files)}"
    )
    folder_id = _resolve_folder_id(context, folder_ref)
    response = requests.patch(
        f"{FILES_URL}/{file_id}/folder",
        headers=_json_jwt_headers(context),
        json={"folderId": folder_id},
        timeout=HTTP_TIMEOUT,
    )
    context.response = response


@when("I list files")
def step_list_files(context):
    response = requests.get(
        FILES_URL, headers=_jwt_headers(context), timeout=HTTP_TIMEOUT
    )
    context.response = response


# ---------------------------------------------------------------------------
# WHEN - unauthenticated probes
# ---------------------------------------------------------------------------


@when("I GET the folder list with no authentication")
def step_get_folders_no_auth(context):
    response = requests.get(FOLDERS_URL, timeout=HTTP_TIMEOUT)
    context.response = response


@when('I POST a folder named "{name}" with no authentication')
def step_post_folder_no_auth(context, name):
    response = requests.post(
        FOLDERS_URL,
        headers={"Content-Type": "application/json"},
        json={"name": name},
        timeout=HTTP_TIMEOUT,
    )
    context.response = response


# ---------------------------------------------------------------------------
# THEN - JSON shape assertions
# ---------------------------------------------------------------------------


@then('the response JSON folder should have name "{name}"')
def step_response_folder_name(context, name):
    data = context.response.json()
    assert data.get("name") == name, (
        f"Expected name={name!r}, got {data.get('name')!r}. Body: {data}"
    )


@then("the response JSON folder.parentFolderId should be null")
def step_response_folder_parent_null(context):
    data = context.response.json()
    assert data.get("parentFolderId") is None, (
        f"Expected parentFolderId=null, got {data.get('parentFolderId')!r}"
    )


@then('the response JSON folder.parentFolderId should equal "{ref}"')
def step_response_folder_parent_equal(context, ref):
    data = context.response.json()
    expected = _resolve_folder_id(context, ref)
    assert data.get("parentFolderId") == expected, (
        f"Expected parentFolderId={expected!r}, "
        f"got {data.get('parentFolderId')!r}"
    )


@then("the response JSON folder.createdAt should not be empty")
def step_response_folder_createdat_not_empty(context):
    data = context.response.json()
    assert data.get("createdAt"), (
        f"Expected non-empty createdAt; body: {data}"
    )


@then("the response JSON file.folderId should be null")
def step_response_file_folderid_null(context):
    data = context.response.json()
    assert data.get("folderId") is None, (
        f"Expected folderId=null, got {data.get('folderId')!r}"
    )


@then('the response JSON file.folderId should equal "{ref}"')
def step_response_file_folderid_equal(context, ref):
    data = context.response.json()
    expected = _resolve_folder_id(context, ref)
    assert data.get("folderId") == expected, (
        f"Expected folderId={expected!r}, got {data.get('folderId')!r}"
    )


@then('the folder list should contain "{name}"')
def step_folder_list_contains(context, name):
    folders = context.response.json()
    names = [f.get("name") for f in folders]
    assert name in names, f"Folder {name!r} not in list: {names}"


@then('the folder list should not contain "{name}"')
def step_folder_list_not_contains(context, name):
    folders = context.response.json()
    names = [f.get("name") for f in folders]
    assert name not in names, (
        f"Folder {name!r} should not be in list: {names}"
    )


@then('the file list should contain a file named "{filename}" with folderId null')
def step_file_list_contains_root_file(context, filename):
    files = context.response.json()
    matches = [
        f
        for f in files
        if f.get("fileName") == filename and f.get("folderId") is None
    ]
    assert matches, (
        f"No file named {filename!r} with folderId=null in list. "
        f"Files seen: "
        f"{[(f.get('fileName'), f.get('folderId')) for f in files]}"
    )