feat(02-02): add driver upload endpoint with INF parsing and dedup

- POST /drivers/upload: validates ZIP, parses INF, persists via DriverStore + Peewee
- SHA256 dedup: get_or_create prevents duplicate Driver records
- GET /drivers route added to pages.py with driver_data context
- drivers.router registered in main.py
- drivers.html template with HTMX upload form
- partials/driver_list.html HTMX target with select dropdown and unused-file notice
- Fixed conftest client fixture to use context manager (triggers lifespan/init_db)
- [Rule 3] conftest: TestClient context manager required for lifespan trigger
- [Rule 1] drivers.py: dynamic DRIVERS_DIR read so monkeypatch works in tests
This commit is contained in:
2026-04-10 12:02:00 +02:00
parent 8ecfbf25a7
commit c648fc5793
6 changed files with 212 additions and 2 deletions
+115
View File
@@ -0,0 +1,115 @@
"""Driver upload API — POST /drivers/upload."""
from __future__ import annotations
import io
import json
import zipfile
from pathlib import Path
from fastapi import APIRouter, Request, UploadFile
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
import imptune.config as _cfg
from imptune.db.models import Driver
from imptune.services.inf_parser import _detect_encoding, parse_inf
from imptune.storage.driver_store import DriverStore
router = APIRouter(prefix="/drivers")
templates = Jinja2Templates(
directory=str(Path(__file__).parent.parent / "templates")
)
MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 MB
def _error_response(message: str, status_code: int = 400) -> HTMLResponse:
"""Return an HTMX-friendly error fragment swapped into #driver-list."""
return HTMLResponse(
content=f"<div id='driver-list' class='error'><p>{message}</p></div>",
status_code=status_code,
)
@router.post("/upload", response_class=HTMLResponse)
def upload_driver(request: Request, file: UploadFile) -> HTMLResponse:
"""Accept a driver ZIP, parse its INF, persist via DriverStore + Peewee ORM.
Returns an HTMX partial (partials/driver_list.html) on success, or an
inline error fragment with HTTP 400 on validation failure.
"""
data = file.file.read(MAX_UPLOAD_BYTES + 1)
if len(data) > MAX_UPLOAD_BYTES:
return _error_response("File exceeds 100 MB limit.")
# Must end with .zip
filename = file.filename or ""
if not filename.lower().endswith(".zip"):
return _error_response("Only .zip files are accepted.")
# Must be a valid ZIP archive
if not zipfile.is_zipfile(io.BytesIO(data)):
return _error_response("Uploaded file is not a valid ZIP archive.")
with zipfile.ZipFile(io.BytesIO(data)) as zf:
zip_names = zf.namelist()
# Reject zip-slip paths
for name in zip_names:
if ".." in name or name.startswith("/"):
return _error_response("ZIP contains unsafe paths.")
# Find .inf files
inf_names = [n for n in zip_names if n.lower().endswith(".inf")]
if not inf_names:
return _error_response("No .inf file found in the uploaded ZIP.")
# Prefer amd64/x64 INF when multiple exist; fall back to alphabetical first
preferred = [
n for n in inf_names if "amd64" in n.lower() or "x64" in n.lower()
]
chosen_inf = preferred[0] if preferred else sorted(inf_names)[0]
raw_inf = zf.read(chosen_inf)
# Decode INF
encoding = _detect_encoding(raw_inf)
inf_text = raw_inf.decode(encoding)
# Parse INF
parsed = parse_inf(inf_text, inf_filename=chosen_inf, zip_names=zip_names)
# Persist file (content-addressed, dedup automatic)
# Read DRIVERS_DIR at call time so tests can monkeypatch imptune.config.DRIVERS_DIR
store = DriverStore(_cfg.DRIVERS_DIR)
sha256 = store.save(data)
# Upsert Driver record (no duplicate if same SHA256)
Driver.get_or_create(
sha256=sha256,
defaults={
"original_filename": filename,
"size_bytes": len(data),
"driver_desc": json.dumps(parsed.driver_names),
"inf_filename": parsed.inf_filename,
"architecture": parsed.architecture,
"has_cat_file": parsed.has_cat_file,
},
)
# Build driver_data for template
drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
driver_data = []
for d in drivers:
names = json.loads(d.driver_desc) if d.driver_desc else []
driver_data.append({"driver": d, "names": names})
return templates.TemplateResponse(
request=request,
name="partials/driver_list.html",
context={
"driver_data": driver_data,
"parsed": parsed,
},
)
+18
View File
@@ -1,3 +1,5 @@
import json
from fastapi import APIRouter, Request from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
@@ -18,3 +20,19 @@ def dashboard(request: Request):
"recent_packages": [], "recent_packages": [],
}, },
) )
@router.get("/drivers", response_class=HTMLResponse)
def drivers_page(request: Request):
from imptune.db.models import Driver
drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
driver_data = []
for d in drivers:
names = json.loads(d.driver_desc) if d.driver_desc else []
driver_data.append({"driver": d, "names": names})
return templates.TemplateResponse(
request=request,
name="drivers.html",
context={"driver_data": driver_data},
)
+2 -1
View File
@@ -5,7 +5,7 @@ from pathlib import Path
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
from imptune.api import health, pages from imptune.api import drivers, health, pages
from imptune.config import DATA_DIR, DRIVERS_DIR from imptune.config import DATA_DIR, DRIVERS_DIR
from imptune.db.database import init_db from imptune.db.database import init_db
@@ -27,3 +27,4 @@ app.mount("/static", StaticFiles(directory=str(_static_dir)), name="static")
# Register routers # Register routers
app.include_router(health.router) app.include_router(health.router)
app.include_router(pages.router) app.include_router(pages.router)
app.include_router(drivers.router)
+25
View File
@@ -0,0 +1,25 @@
{% extends "base.html" %}
{% block content %}
<h1>Drivers</h1>
<section>
<h2>Upload Driver Package</h2>
<form
hx-post="/drivers/upload"
hx-encoding="multipart/form-data"
hx-target="#driver-list"
hx-swap="outerHTML"
hx-indicator="#upload-spinner"
>
<label for="driver-file">Driver Package (ZIP containing .inf + driver files)</label>
<input type="file" id="driver-file" name="file" accept=".zip" required>
<button type="submit">Upload</button>
<span id="upload-spinner" class="htmx-indicator" aria-busy="true">Uploading...</span>
</form>
</section>
<section>
<h2>Driver Library</h2>
{% include "partials/driver_list.html" %}
</section>
{% endblock %}
@@ -0,0 +1,50 @@
<div id="driver-list">
{% if parsed is defined and parsed.unused_files %}
<p class="notice">
{{ parsed.unused_files | length }} file(s) may be unused (not referenced by the INF):
<details>
<summary>Show unused files</summary>
<ul>
{% for f in parsed.unused_files %}
<li>{{ f }}</li>
{% endfor %}
</ul>
</details>
</p>
{% endif %}
{% if driver_data %}
<table>
<thead>
<tr>
<th>Filename</th>
<th>Driver Name(s)</th>
<th>Architecture</th>
<th>Uploaded</th>
</tr>
</thead>
<tbody>
{% for item in driver_data %}
<tr>
<td>{{ item.driver.original_filename }}</td>
<td>
{% if item.names %}
<select aria-label="Driver names">
{% for name in item.names %}
<option>{{ name }}</option>
{% endfor %}
</select>
{% else %}
<em>Unknown</em>
{% endif %}
</td>
<td>{{ item.driver.architecture or "Unknown" }}</td>
<td>{{ item.driver.uploaded_at }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p>No drivers uploaded yet.</p>
{% endif %}
</div>
+2 -1
View File
@@ -8,7 +8,8 @@ from fastapi.testclient import TestClient
def client(tmp_data_dir): def client(tmp_data_dir):
from imptune.main import app from imptune.main import app
return TestClient(app) with TestClient(app) as c:
yield c
@pytest.fixture @pytest.fixture