kawa
d1de8395a3
- Add _make_driver_zip_with_cat() helper for .inf + .cat fixture ZIPs - Add _make_bom_driver_zip() helper for UTF-16 LE BOM encoded INF edge case - Add test_upload_500_regression (parametrized: plain UTF-8 + BOM variant) - Add test_upload_returns_oob_when_called_from_form (RED: handler lacks caller param) - Add test_upload_oob_autoselects_new_driver (RED: no OOB fragment emitted) - Add test_upload_no_oob_from_standalone_drivers_page OOB tests fail: handler returns driver_list.html always regardless of caller field. 500 regression tests pass: existing synthetic ZIPs work fine against current handler.
259 lines
9.0 KiB
Python
259 lines
9.0 KiB
Python
"""Integration tests for driver upload endpoint and drivers page."""
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import zipfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SAMPLE_INF = """\
|
|
[Version]
|
|
Signature="$Windows NT$"
|
|
Class=Printer
|
|
Provider=%MFG%
|
|
|
|
[Manufacturer]
|
|
%MFG%=Models,NTamd64
|
|
|
|
[Models.NTamd64]
|
|
%DRIVER_NAME%=Install,{12345678-1234-1234-1234-123456789012}
|
|
|
|
[Strings]
|
|
MFG="Test Manufacturer"
|
|
DRIVER_NAME="Test LaserJet Pro"
|
|
"""
|
|
|
|
|
|
def _make_driver_zip(
|
|
inf_content: str = SAMPLE_INF,
|
|
inf_name: str = "sample.inf",
|
|
extra_files: dict[str, bytes] | None = None,
|
|
) -> bytes:
|
|
"""Build an in-memory ZIP with one .inf file and optional extra files."""
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr(inf_name, inf_content.encode("utf-8"))
|
|
if extra_files:
|
|
for name, data in extra_files.items():
|
|
zf.writestr(name, data)
|
|
return buf.getvalue()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_drivers_page(client: TestClient) -> None:
|
|
"""GET /drivers returns 200 with an upload form targeting /drivers/upload."""
|
|
resp = client.get("/drivers")
|
|
assert resp.status_code == 200
|
|
html = resp.text
|
|
assert 'type="file"' in html
|
|
assert "/drivers/upload" in html
|
|
assert "hx-post" in html
|
|
|
|
|
|
def test_upload_valid_zip(client: TestClient) -> None:
|
|
"""POST /drivers/upload with a valid ZIP containing .inf returns 200 with driver name."""
|
|
zip_bytes = _make_driver_zip()
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "Test LaserJet Pro" in resp.text
|
|
|
|
|
|
def test_upload_non_zip(client: TestClient) -> None:
|
|
"""POST /drivers/upload with a .txt file (not a ZIP) returns 400."""
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", b"this is not a zip", "application/zip")},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_upload_no_inf(client: TestClient) -> None:
|
|
"""POST /drivers/upload with a ZIP containing no .inf returns 400."""
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr("readme.txt", b"no driver here")
|
|
zip_bytes = buf.getvalue()
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_upload_returns_select(client: TestClient) -> None:
|
|
"""POST /drivers/upload with valid ZIP returns HTML containing a <select> element."""
|
|
zip_bytes = _make_driver_zip()
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "<select" in resp.text
|
|
assert "Test LaserJet Pro" in resp.text
|
|
|
|
|
|
def test_driver_persisted(client: TestClient, tmp_data_dir) -> None:
|
|
"""After upload, Driver record exists in DB and file exists in DriverStore."""
|
|
from imptune.db.models import Driver
|
|
|
|
zip_bytes = _make_driver_zip()
|
|
expected_sha = hashlib.sha256(zip_bytes).hexdigest()
|
|
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
count = Driver.select().where(Driver.sha256 == expected_sha).count()
|
|
assert count == 1
|
|
|
|
driver_file = tmp_data_dir / "drivers" / f"{expected_sha}.zip"
|
|
assert driver_file.exists()
|
|
|
|
|
|
def test_dedup_upload(client: TestClient) -> None:
|
|
"""Uploading the same ZIP twice creates only one Driver record."""
|
|
from imptune.db.models import Driver
|
|
|
|
zip_bytes = _make_driver_zip()
|
|
|
|
resp1 = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp1.status_code == 200
|
|
|
|
resp2 = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp2.status_code == 200
|
|
|
|
sha = hashlib.sha256(zip_bytes).hexdigest()
|
|
count = Driver.select().where(Driver.sha256 == sha).count()
|
|
assert count == 1
|
|
|
|
|
|
def test_unused_files_in_response(client: TestClient) -> None:
|
|
"""Upload a ZIP with an extra file not in INF; response HTML mentions 'unused'."""
|
|
zip_bytes = _make_driver_zip(
|
|
extra_files={"readme.txt": b"This file is not referenced by the INF"}
|
|
)
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp.status_code == 200
|
|
# Response should indicate unused files (count or the word "unused")
|
|
assert "unused" in resp.text.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Regression + OOB contract tests (Wave 0 additions -- Task 1 of 09-01)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_driver_zip_with_cat(
|
|
inf_content: str = SAMPLE_INF,
|
|
inf_name: str = "sample.inf",
|
|
) -> bytes:
|
|
"""Build a ZIP with an .inf and a .cat file (has_cat_file=True)."""
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr(inf_name, inf_content.encode("utf-8"))
|
|
zf.writestr(inf_name.replace(".inf", ".cat"), b"fake-cat-content")
|
|
return buf.getvalue()
|
|
|
|
|
|
def _make_bom_driver_zip() -> bytes:
|
|
"""Build a ZIP with a UTF-16 LE BOM-encoded .inf (encoding edge case)."""
|
|
bom_inf_text = (
|
|
"[Version]\r\nSignature=\"$Windows NT$\"\r\nClass=Printer\r\n\r\n"
|
|
"[Manufacturer]\r\n%MFG%=Models,NTamd64\r\n\r\n"
|
|
"[Models.NTamd64]\r\n%DRIVER_NAME%=Install,{ABCD1234-0000-0000-0000-000000000001}\r\n\r\n"
|
|
"[Strings]\r\nMFG=\"BOM Manufacturer\"\r\nDRIVER_NAME=\"BOM LaserJet 9000\"\r\n"
|
|
)
|
|
bom_inf_bytes = b"\xff\xfe" + bom_inf_text.encode("utf-16-le")
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("driver.inf", bom_inf_bytes)
|
|
zf.writestr("driver.cat", b"fake-catalog")
|
|
return buf.getvalue()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"zip_bytes_fn, label",
|
|
[
|
|
(lambda: _make_driver_zip(extra_files={"sample.cat": b"cat"}), "plain_utf8_inf"),
|
|
(_make_bom_driver_zip, "bom_utf16le_inf"),
|
|
],
|
|
)
|
|
def test_upload_500_regression(client: TestClient, zip_bytes_fn, label: str) -> None:
|
|
"""POST /drivers/upload with a valid driver ZIP MUST return 200, never 500."""
|
|
zip_bytes = zip_bytes_fn()
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp.status_code != 500, f"[{label}] Upload returned HTTP 500:\n{resp.text}"
|
|
assert resp.status_code == 200, f"[{label}] Expected 200, got {resp.status_code}:\n{resp.text}"
|
|
|
|
|
|
def test_upload_returns_oob_when_called_from_form(client: TestClient) -> None:
|
|
"""POST /drivers/upload with caller=printer_form must return OOB swap markup."""
|
|
zip_bytes = _make_driver_zip_with_cat()
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
data={"caller": "printer_form"},
|
|
)
|
|
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}:\n{resp.text}"
|
|
assert 'hx-swap-oob="true"' in resp.text, "Response missing hx-swap-oob attribute"
|
|
assert 'id="printer-form-driver-select"' in resp.text, "Response missing OOB select id"
|
|
|
|
|
|
def test_upload_oob_autoselects_new_driver(client: TestClient) -> None:
|
|
"""POST /drivers/upload with caller=printer_form must auto-select the new driver."""
|
|
import re
|
|
|
|
from imptune.db.models import Driver
|
|
|
|
zip_bytes = _make_driver_zip_with_cat()
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
data={"caller": "printer_form"},
|
|
)
|
|
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}:\n{resp.text}"
|
|
sha = hashlib.sha256(zip_bytes).hexdigest()
|
|
driver = Driver.get(Driver.sha256 == sha)
|
|
new_id = driver.id
|
|
assert f'value="{new_id}"' in resp.text, f"Driver id={new_id} not found in OOB response"
|
|
pattern = rf'<option\s+value="{new_id}"\s+selected'
|
|
assert re.search(pattern, resp.text), f"New driver (id={new_id}) not marked as selected"
|
|
|
|
|
|
def test_upload_no_oob_from_standalone_drivers_page(client: TestClient) -> None:
|
|
"""POST /drivers/upload WITHOUT caller field must NOT contain hx-swap-oob."""
|
|
zip_bytes = _make_driver_zip_with_cat()
|
|
resp = client.post(
|
|
"/drivers/upload",
|
|
files={"file": ("driver.zip", zip_bytes, "application/zip")},
|
|
)
|
|
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}:\n{resp.text}"
|
|
assert "hx-swap-oob" not in resp.text, "Standalone upload should NOT return OOB fragments"
|