Commit initial

This commit is contained in:
2026-04-15 17:57:12 +02:00
parent 005d8e797e
commit 55516ee10f
269 changed files with 26854 additions and 0 deletions
View File
Binary file not shown.
Binary file not shown.
+36
View File
@@ -0,0 +1,36 @@
import os
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client(tmp_data_dir):
from imptune.main import app
with TestClient(app) as c:
yield c
@pytest.fixture
def tmp_data_dir(tmp_path, monkeypatch):
"""Set DATA_DIR to a temp directory so tests don't write to /data."""
data_dir = tmp_path / "data"
data_dir.mkdir()
monkeypatch.setenv("DATA_DIR", str(data_dir))
# Patch config module so the app uses the temp dir
import imptune.config as cfg
cfg.DATA_DIR = str(data_dir)
cfg.DB_PATH = str(data_dir / "imptune.db")
cfg.DRIVERS_DIR = str(data_dir / "drivers")
cfg.ICONS_DIR = str(data_dir / "icons")
yield data_dir
# Close the test-thread's DB connection so the next test gets a fresh one
# pointing to its own tmp DB (Peewee connections are thread-local).
from imptune.db.database import db
if not db.is_closed():
db.close()
View File
Binary file not shown.
+70
View File
@@ -0,0 +1,70 @@
"""E2E test fixtures: live uvicorn server for Playwright."""
from __future__ import annotations
import os
import socket
import threading
import time
import httpx
import pytest
import uvicorn
def _free_port() -> int:
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@pytest.fixture(scope="session")
def live_server(tmp_path_factory):
"""Start the FastAPI app on a random port in a background thread."""
# Isolated data dir for E2E session
data_dir = tmp_path_factory.mktemp("imptune_e2e_data")
# Patch config module so the app uses the temp dir
import imptune.config as cfg
cfg.DATA_DIR = str(data_dir)
cfg.DB_PATH = str(data_dir / "imptune.db")
cfg.DRIVERS_DIR = str(data_dir / "drivers")
cfg.ICONS_DIR = str(data_dir / "icons")
os.makedirs(cfg.DRIVERS_DIR, exist_ok=True)
os.makedirs(cfg.ICONS_DIR, exist_ok=True)
# Also set env var so lifespan handler picks up correct dirs
os.environ["DATA_DIR"] = str(data_dir)
# Initialize DB against the tmp dir
from imptune.db.database import init_db
init_db()
from imptune.main import app
port = _free_port()
config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")
server = uvicorn.Server(config)
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
base_url = f"http://127.0.0.1:{port}"
# Readiness poll via /health (up to 5 s)
deadline = time.time() + 5.0
while time.time() < deadline:
try:
r = httpx.get(f"{base_url}/health", timeout=0.5)
if r.status_code == 200:
break
except Exception:
pass
time.sleep(0.1)
else:
raise RuntimeError("live_server did not become ready within 5 s")
yield base_url
server.should_exit = True
thread.join(timeout=2.0)
+132
View File
@@ -0,0 +1,132 @@
"""UIE-05: E2E tests for language toggle — FR/EN switching and localStorage persistence."""
from __future__ import annotations
import pytest
# ---------------------------------------------------------------------------
# Task 1 (Phase 12-01): navigator.language auto-detection
# ---------------------------------------------------------------------------
def test_navigator_language_en_sets_lang_en(browser, live_server: str) -> None:
"""When localStorage is empty and navigator.language is 'en-US', lang defaults to 'en'."""
context = browser.new_context(locale="en-US")
page = context.new_page()
# Clear any saved preference so localStorage fallback is not used
page.goto(f"{live_server}/", wait_until="domcontentloaded")
page.evaluate("localStorage.removeItem('imptune_lang')")
# Reload so the alpine:init IIFE runs with no localStorage and en-US locale
page.reload(wait_until="domcontentloaded")
page.wait_for_function(
"document.querySelector('nav a[href=\"/printers\"]').textContent.trim() !== ''",
timeout=3000,
)
lang = page.evaluate("Alpine.store('i18n').lang")
assert lang == "en", f"Expected 'en' when navigator.language is 'en-US', got '{lang}'"
context.close()
def test_navigator_language_fr_sets_lang_fr(browser, live_server: str) -> None:
"""When localStorage is empty and navigator.language is 'fr-FR', lang defaults to 'fr'."""
context = browser.new_context(locale="fr-FR")
page = context.new_page()
page.goto(f"{live_server}/", wait_until="domcontentloaded")
page.evaluate("localStorage.removeItem('imptune_lang')")
page.reload(wait_until="domcontentloaded")
page.wait_for_function(
"document.querySelector('nav a[href=\"/printers\"]').textContent.trim() !== ''",
timeout=3000,
)
lang = page.evaluate("Alpine.store('i18n').lang")
assert lang == "fr", f"Expected 'fr' when navigator.language is 'fr-FR', got '{lang}'"
context.close()
def test_localstorage_wins_over_navigator_language(browser, live_server: str) -> None:
"""localStorage preference wins over navigator.language when a value is saved."""
# Browser locale is 'en-US' but localStorage says 'fr' → should use 'fr'
context = browser.new_context(locale="en-US")
page = context.new_page()
page.goto(f"{live_server}/", wait_until="domcontentloaded")
page.evaluate("localStorage.setItem('imptune_lang', 'fr')")
page.reload(wait_until="domcontentloaded")
page.wait_for_function(
"document.querySelector('nav a[href=\"/printers\"]').textContent.trim() !== ''",
timeout=3000,
)
lang = page.evaluate("Alpine.store('i18n').lang")
assert lang == "fr", f"Expected localStorage 'fr' to win over navigator.language 'en-US', got '{lang}'"
# Cleanup
page.evaluate("localStorage.removeItem('imptune_lang')")
context.close()
# ---------------------------------------------------------------------------
# Original toggle and persistence tests
# ---------------------------------------------------------------------------
def test_language_toggle_switches_nav_label(page, live_server: str) -> None:
"""Clicking FR/EN button switches nav label from French to English."""
page.goto(f"{live_server}/", wait_until="domcontentloaded")
# Default lang is 'fr' — nav should show French labels
# Wait for Alpine to hydrate
page.wait_for_function(
"document.querySelector('nav a[href=\"/printers\"]').textContent.trim() !== ''",
timeout=3000,
)
# In French, printers nav label = 'Imprimantes'
printers_label_fr = page.text_content("nav a[href='/printers']").strip()
assert printers_label_fr == "Imprimantes", f"Expected 'Imprimantes', got '{printers_label_fr}'"
# Click the language toggle button
page.click("button[title='Toggle language']")
# Wait for label to update
page.wait_for_function(
"document.querySelector('nav a[href=\"/printers\"]').textContent.trim() === 'Printers'",
timeout=2000,
)
printers_label_en = page.text_content("nav a[href='/printers']").strip()
assert printers_label_en == "Printers"
# Cleanup: reset to fr
page.evaluate("localStorage.setItem('imptune_lang', 'fr')")
def test_language_persists_across_reload(page, live_server: str) -> None:
"""After switching to EN, language is preserved on page reload."""
page.goto(f"{live_server}/", wait_until="domcontentloaded")
# Switch to English
page.click("button[title='Toggle language']")
page.wait_for_function(
"document.querySelector('nav a[href=\"/printers\"]').textContent.trim() === 'Printers'",
timeout=2000,
)
# Reload
page.reload(wait_until="domcontentloaded")
page.wait_for_function(
"document.querySelector('nav a[href=\"/printers\"]').textContent.trim() !== ''",
timeout=3000,
)
label_after_reload = page.text_content("nav a[href='/printers']").strip()
assert label_after_reload == "Printers"
# Cleanup: reset to fr
page.evaluate("localStorage.setItem('imptune_lang', 'fr')")
+29
View File
@@ -0,0 +1,29 @@
"""UX-02: live-browser verification of PRNT-03 Alpine IP->port auto-derivation."""
from __future__ import annotations
import pytest
def test_port_autofill(page, live_server: str) -> None:
"""Typing an IP address into the printer form auto-populates port_name.
Evidence for UX-02: tests/e2e/test_port_autofill.py
Closes: PRNT-03 Alpine.js port auto-derivation requirement.
"""
# /printers/new renders a full page (extends base.html) with Alpine.js loaded
# and the add-printer form (moved from /printers to /printers/new in Plan 11-01)
page.goto(f"{live_server}/printers/new", wait_until="domcontentloaded")
# Wait for Alpine to initialise (x-data hydration on the form wrapper)
page.wait_for_selector("input[name='ip_address']")
# Fill triggers the 'input' event that Alpine @input listens to
page.fill("input[name='ip_address']", "192.168.1.100")
# Alpine @input reacts synchronously; wait_for_function keeps the test stable
page.wait_for_function(
"document.querySelector(\"input[name='port_name']\").value === 'IP_192_168_1_100'",
timeout=2000,
)
assert page.input_value("input[name='port_name']") == "IP_192_168_1_100"
+62
View File
@@ -0,0 +1,62 @@
"""UIE-01: E2E test for printer edit modal — open, pre-fill, submit, list update."""
from __future__ import annotations
import pytest
def test_printer_edit_modal_open_and_prefill(page, live_server: str) -> None:
"""Edit button opens modal with printer's current name pre-filled."""
import httpx
# Create a printer via API
with httpx.Client(base_url=live_server, follow_redirects=True) as api:
api.post(
"/printers",
data={
"name": "EditTest Printer",
"ip_address": "10.0.5.1",
"port_name": "IP_10_0_5_1",
},
)
page.goto(f"{live_server}/printers", wait_until="domcontentloaded")
page.wait_for_selector("button:has-text('Edit')")
page.click("button:has-text('Edit')")
# Dialog should be open
page.wait_for_selector("dialog[open]")
# Name input should be pre-filled
name_val = page.input_value("dialog[open] input[name='name']")
assert name_val == "EditTest Printer"
def test_printer_edit_submit_updates_list(page, live_server: str) -> None:
"""Submitting the edit form updates the printer name in the list (no page reload)."""
import httpx
with httpx.Client(base_url=live_server, follow_redirects=True) as api:
resp = api.post(
"/printers",
data={
"name": "OriginalName",
"ip_address": "10.0.5.2",
"port_name": "IP_10_0_5_2",
},
)
# Find the printer ID from the DB via API — target the specific Edit button by printer ID
page.goto(f"{live_server}/printers", wait_until="domcontentloaded")
# Find the row containing "OriginalName" and click its Edit button
row = page.locator("tr", has=page.locator("a", has_text="OriginalName"))
row.wait_for()
row.locator("button:has-text('Edit')").click()
page.wait_for_selector("dialog[open]")
# Clear and update the name field
page.fill("dialog[open] input[name='name']", "UpdatedName")
page.click("dialog[open] button[type='submit']")
# Wait for HTMX to swap the updated list — the anchor for the printer should show new name
page.wait_for_selector("a:has-text('UpdatedName')")
assert page.locator("a:has-text('UpdatedName')").is_visible()
assert page.locator("a:has-text('OriginalName')").count() == 0
+50
View File
@@ -0,0 +1,50 @@
"""UIE-04: E2E tests for theme toggle — data-theme cycling and localStorage persistence."""
from __future__ import annotations
import pytest
def test_theme_cycles_on_click(page, live_server: str) -> None:
"""Clicking theme button cycles data-theme attribute: auto -> light -> dark -> auto."""
page.goto(f"{live_server}/", wait_until="domcontentloaded")
# Initial state: auto (default from base.html)
initial_theme = page.evaluate("document.documentElement.getAttribute('data-theme')")
assert initial_theme == "auto"
# Click once -> light
page.click("button[aria-label='auto']")
page.wait_for_function(
"document.documentElement.getAttribute('data-theme') === 'light'",
timeout=2000,
)
assert page.evaluate("document.documentElement.getAttribute('data-theme')") == "light"
# Click again -> dark
page.click("button[aria-label='light']")
page.wait_for_function(
"document.documentElement.getAttribute('data-theme') === 'dark'",
timeout=2000,
)
assert page.evaluate("document.documentElement.getAttribute('data-theme')") == "dark"
def test_theme_persists_across_reload(page, live_server: str) -> None:
"""After clicking theme toggle, the chosen theme is restored on reload."""
page.goto(f"{live_server}/", wait_until="domcontentloaded")
# Switch to light mode
page.click("button[aria-label='auto']")
page.wait_for_function(
"document.documentElement.getAttribute('data-theme') === 'light'",
timeout=2000,
)
# Reload the page
page.reload(wait_until="domcontentloaded")
# Theme should still be light (from localStorage)
theme_after_reload = page.evaluate("document.documentElement.getAttribute('data-theme')")
assert theme_after_reload == "light"
# Cleanup: reset to auto
page.evaluate("localStorage.setItem('imptune_theme', 'auto')")
+14
View File
@@ -0,0 +1,14 @@
[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"
+18
View File
@@ -0,0 +1,18 @@
[Version]
Signature="$Windows NT$"
Class=Printer
[Manufacturer]
%MFG%=Models,Models.NTamd64
[Models]
%DRIVER_A%=InstallA,{11111111-1111-1111-1111-111111111111}
[Models.NTamd64]
%DRIVER_A%=InstallA,{11111111-1111-1111-1111-111111111111}
%DRIVER_B%=InstallB,{22222222-2222-2222-2222-222222222222}
[Strings]
MFG="Multi Corp"
DRIVER_A="Multi Printer 1000"
DRIVER_B="Multi Printer 2000"
BIN
View File
Binary file not shown.
+141
View File
@@ -0,0 +1,141 @@
"""Tests for database initialization and driver storage."""
import hashlib
from pathlib import Path
import pytest
@pytest.fixture
def db_env(tmp_path, monkeypatch):
"""Set up temp DATA_DIR and configure db to use temp paths."""
data_dir = tmp_path / "data"
data_dir.mkdir()
drivers_dir = data_dir / "drivers"
drivers_dir.mkdir()
db_path = str(data_dir / "imptune.db")
monkeypatch.setenv("DATA_DIR", str(data_dir))
import imptune.config as cfg
cfg.DATA_DIR = str(data_dir)
cfg.DB_PATH = db_path
cfg.DRIVERS_DIR = str(drivers_dir)
# Close and re-init the db with the temp path
from imptune.db.database import db
if not db.is_closed():
db.close()
return {
"data_dir": data_dir,
"drivers_dir": drivers_dir,
"db_path": db_path,
}
def test_create_tables(db_env):
"""init_db() creates all 4 tables in a fresh SQLite file."""
from imptune.db.database import db, init_db
if not db.is_closed():
db.close()
init_db()
tables = db.get_tables()
assert "client" in tables
assert "driver" in tables
assert "printer" in tables
assert "icon" in tables
db.close()
def test_wal_mode(db_env):
"""After init_db(), PRAGMA journal_mode returns 'wal'."""
from imptune.db.database import db, init_db
if not db.is_closed():
db.close()
init_db()
cursor = db.execute_sql("PRAGMA journal_mode;")
mode = cursor.fetchone()[0]
assert mode == "wal"
db.close()
def test_foreign_keys(db_env):
"""After init_db(), PRAGMA foreign_keys returns 1."""
from imptune.db.database import db, init_db
if not db.is_closed():
db.close()
init_db()
cursor = db.execute_sql("PRAGMA foreign_keys;")
value = cursor.fetchone()[0]
assert value == 1
db.close()
def test_idempotent(db_env):
"""Calling init_db() twice does not raise an error."""
from imptune.db.database import db, init_db
if not db.is_closed():
db.close()
init_db()
db.close()
init_db() # second call — must not raise
db.close()
def test_driver_store_save(db_env, tmp_path):
"""Saving bytes returns their SHA256 hex digest and creates the file."""
from imptune.storage.driver_store import DriverStore
drivers_dir = db_env["drivers_dir"]
store = DriverStore(str(drivers_dir))
data = b"test driver package content"
expected_sha256 = hashlib.sha256(data).hexdigest()
result = store.save(data)
assert result == expected_sha256
assert (drivers_dir / f"{expected_sha256}.zip").exists()
def test_driver_store_dedup(db_env):
"""Saving the same bytes twice results in one file on disk."""
from imptune.storage.driver_store import DriverStore
drivers_dir = db_env["drivers_dir"]
store = DriverStore(str(drivers_dir))
data = b"duplicate driver data"
store.save(data)
store.save(data)
files = list(drivers_dir.iterdir())
assert len(files) == 1
def test_driver_store_get_path(db_env):
"""get_path(sha256) returns the correct file path."""
from imptune.storage.driver_store import DriverStore
drivers_dir = db_env["drivers_dir"]
store = DriverStore(str(drivers_dir))
sha256 = "abcdef1234567890" * 4 # 64 hex chars
path = store.get_path(sha256)
assert path == Path(str(drivers_dir)) / f"{sha256}.zip"
+258
View File
@@ -0,0 +1,258 @@
"""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"
+5
View File
@@ -0,0 +1,5 @@
def test_health_returns_200(client):
"""GET /health returns 200 with {"status": "ok"}."""
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
+140
View File
@@ -0,0 +1,140 @@
"""Integration tests for icon upload endpoint — PKG-04."""
from __future__ import annotations
import io
import pytest
from PIL import Image
def _make_png(width: int = 256, height: int = 256, size_bytes: int | None = None) -> bytes:
"""Create a valid PNG image with given dimensions in memory."""
img = Image.new("RGBA", (width, height), color="red")
buf = io.BytesIO()
img.save(buf, format="PNG")
data = buf.getvalue()
if size_bytes is not None and size_bytes > len(data):
# Pad the PNG by embedding extra data (won't affect PIL open, but will exceed size limit)
# Instead, use a raw bytes approach: return oversized raw content
data = data + b"\x00" * (size_bytes - len(data))
return data
def _make_jpeg(width: int = 256, height: int = 256) -> bytes:
"""Create a valid JPEG image with given dimensions in memory."""
img = Image.new("RGB", (width, height), color="blue")
buf = io.BytesIO()
img.save(buf, format="JPEG")
return buf.getvalue()
def _create_printer(client):
"""Create a test Printer record and return it."""
from imptune.db.models import Printer
return Printer.create(
name="Test Printer",
ip_address="10.0.0.1",
port_name="IP_10.0.0.1",
)
class TestIconUpload:
def test_upload_valid_png(self, client, tmp_data_dir):
"""POST a valid 256x256 PNG returns 200 and Icon record created."""
from imptune.db.models import Icon
printer = _create_printer(client)
png_data = _make_png(256, 256)
response = client.post(
f"/printers/{printer.id}/icon",
files={"file": ("icon.png", io.BytesIO(png_data), "image/png")},
)
assert response.status_code == 200
assert "Icon uploaded successfully" in response.text
icons = list(Icon.select().where(Icon.printer == printer.id))
assert len(icons) == 1
assert icons[0].original_filename == "icon.png"
# File must be stored on disk
import hashlib
from pathlib import Path
sha256 = hashlib.sha256(png_data).hexdigest()
icon_file = Path(tmp_data_dir) / "icons" / sha256
assert icon_file.exists()
def test_reject_non_png(self, client, tmp_data_dir):
"""POST with a JPEG file returns 422 with PNG format error."""
printer = _create_printer(client)
jpeg_data = _make_jpeg(256, 256)
response = client.post(
f"/printers/{printer.id}/icon",
files={"file": ("icon.jpg", io.BytesIO(jpeg_data), "image/jpeg")},
)
assert response.status_code == 422
assert "PNG" in response.text
def test_reject_oversized(self, client, tmp_data_dir):
"""POST with PNG > 750KB returns 422 with 750 KB error."""
printer = _create_printer(client)
# Craft oversized data: valid PNG bytes followed by padding
png_bytes = _make_png(256, 256)
oversized = png_bytes + b"\x00" * (750 * 1024 + 1 - len(png_bytes))
response = client.post(
f"/printers/{printer.id}/icon",
files={"file": ("big.png", io.BytesIO(oversized), "image/png")},
)
assert response.status_code == 422
assert "750" in response.text
def test_reject_wrong_dimensions(self, client, tmp_data_dir):
"""POST with 128x128 PNG returns 422 with 256x256 error."""
printer = _create_printer(client)
png_data = _make_png(128, 128)
response = client.post(
f"/printers/{printer.id}/icon",
files={"file": ("small.png", io.BytesIO(png_data), "image/png")},
)
assert response.status_code == 422
assert "256x256" in response.text
def test_replace_existing_icon(self, client, tmp_data_dir):
"""Second upload for same printer replaces the Icon record."""
from imptune.db.models import Icon
printer = _create_printer(client)
# First upload
png1 = _make_png(256, 256)
client.post(
f"/printers/{printer.id}/icon",
files={"file": ("icon1.png", io.BytesIO(png1), "image/png")},
)
# Second upload (different color to get different sha256)
img2 = Image.new("RGBA", (256, 256), color="green")
buf2 = io.BytesIO()
img2.save(buf2, format="PNG")
png2 = buf2.getvalue()
response = client.post(
f"/printers/{printer.id}/icon",
files={"file": ("icon2.png", io.BytesIO(png2), "image/png")},
)
assert response.status_code == 200
# Still only one Icon record for this printer
icons = list(Icon.select().where(Icon.printer == printer.id))
assert len(icons) == 1
assert icons[0].original_filename == "icon2.png"
def test_404_missing_printer(self, client, tmp_data_dir):
"""POST to nonexistent printer_id returns 404."""
png_data = _make_png(256, 256)
response = client.post(
"/printers/99999/icon",
files={"file": ("icon.png", io.BytesIO(png_data), "image/png")},
)
assert response.status_code == 404
+318
View File
@@ -0,0 +1,318 @@
"""
Unit tests for imptune.services.inf_parser.
Covers DRV-02 (INF parsing, token resolution, encoding detection,
multi-model) and DRV-05 (unused file detection).
"""
from pathlib import Path
import pytest
from imptune.services.inf_parser import ParsedInf, _detect_encoding, parse_inf
FIXTURES = Path(__file__).parent / "fixtures"
# ---------------------------------------------------------------------------
# _detect_encoding tests
# ---------------------------------------------------------------------------
def test_detect_encoding_utf16le():
"""UTF-16 LE BOM (\xff\xfe) is detected as 'utf-16'."""
raw = b"\xff\xfe" + "[Version]\r\n".encode("utf-16-le")
assert _detect_encoding(raw) == "utf-16"
def test_detect_encoding_utf16be():
"""UTF-16 BE BOM (\xfe\xff) is also detected as 'utf-16'."""
raw = b"\xfe\xff" + "[Version]\r\n".encode("utf-16-be")
assert _detect_encoding(raw) == "utf-16"
def test_detect_encoding_utf8bom():
"""UTF-8 BOM (\xef\xbb\xbf) is detected as 'utf-8-sig'."""
raw = b"\xef\xbb\xbf" + b"[Version]\r\n"
assert _detect_encoding(raw) == "utf-8-sig"
def test_detect_encoding_ansi():
"""Byte sequence with no BOM falls back to cp1252 (ANSI)."""
raw = b"[Version]\r\nSignature=\"$Windows NT$\"\r\n"
assert _detect_encoding(raw) == "cp1252"
# ---------------------------------------------------------------------------
# parse_inf simple INF with literal driver description
# ---------------------------------------------------------------------------
SIMPLE_INF = """
[Version]
Signature="$Windows NT$"
Class=Printer
[Manufacturer]
Acme=AcmeModels,NTamd64
[AcmeModels.NTamd64]
Acme SuperPrint 9000=Install,{AAAAAAAA-AAAA-AAAA-AAAA-AAAAAAAAAAAA}
[Strings]
"""
def test_simple_driver_desc():
"""parse_inf extracts literal DriverDesc from a Models section."""
result = parse_inf(SIMPLE_INF, "acme.inf", ["acme.inf", "acme.dll"])
assert "Acme SuperPrint 9000" in result.driver_names
# ---------------------------------------------------------------------------
# parse_inf %TOKEN% resolution from [Strings]
# ---------------------------------------------------------------------------
TOKEN_INF = """
[Version]
Signature="$Windows NT$"
[Manufacturer]
%MFG%=Models,NTamd64
[Models.NTamd64]
%HP_DRIVER%=Install,{BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB}
[Strings]
MFG="HP"
HP_DRIVER="HP LaserJet"
"""
def test_token_resolution():
"""parse_inf resolves %TOKEN% references via the [Strings] section."""
result = parse_inf(TOKEN_INF, "hp.inf", ["hp.inf"])
assert "HP LaserJet" in result.driver_names
# Raw token should NOT appear in the output
assert "%HP_DRIVER%" not in result.driver_names
# ---------------------------------------------------------------------------
# parse_inf UTF-16 LE fixture
# ---------------------------------------------------------------------------
def test_utf16_encoding():
"""Reading a UTF-16 LE BOM fixture and passing decoded text to parse_inf produces correct driver_names."""
raw = (FIXTURES / "sample_utf16.inf").read_bytes()
encoding = _detect_encoding(raw)
assert encoding == "utf-16"
text = raw.decode(encoding, errors="replace")
result = parse_inf(text, "sample_utf16.inf", ["sample_utf16.inf"])
assert "Test LaserJet Pro" in result.driver_names
# ---------------------------------------------------------------------------
# parse_inf multi-model INF (NTamd64 + undecorated)
# ---------------------------------------------------------------------------
def test_multi_model_inf():
"""INF with both NTamd64 and undecorated sections returns deduplicated driver_names."""
text = (FIXTURES / "sample_multi_model.inf").read_text(encoding="cp1252")
result = parse_inf(text, "multi.inf", ["multi.inf"])
# Both driver names must appear exactly once
assert "Multi Printer 1000" in result.driver_names
assert "Multi Printer 2000" in result.driver_names
assert result.driver_names.count("Multi Printer 1000") == 1
assert result.driver_names.count("Multi Printer 2000") == 1
# ---------------------------------------------------------------------------
# parse_inf architecture detection
# ---------------------------------------------------------------------------
AMD64_INF = """
[Version]
Signature="$Windows NT$"
[Manufacturer]
Mfg=Mfg.Models,NTamd64
[Mfg.Models.NTamd64]
Some Driver=Install,{CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC}
[Strings]
"""
ARM64_INF = """
[Version]
Signature="$Windows NT$"
[Manufacturer]
Mfg=Mfg.Models,NTarm64
[Mfg.Models.NTarm64]
Some Driver=Install,{CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC}
[Strings]
"""
UNDECORATED_INF = """
[Version]
Signature="$Windows NT$"
[Manufacturer]
Mfg=Mfg.Models
[Mfg.Models]
Some Driver=Install,{CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC}
[Strings]
"""
MIXED_INF = """
[Version]
Signature="$Windows NT$"
[Manufacturer]
Mfg=Mfg.Models,Mfg.Models.NTamd64
[Mfg.Models]
Some Driver=Install,{CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC}
[Mfg.Models.NTamd64]
Some Driver=Install,{CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC}
[Strings]
"""
def test_architecture_detection_amd64():
"""NTamd64 decoration -> architecture='x64'."""
result = parse_inf(AMD64_INF, "amd64.inf", [])
assert result.architecture == "x64"
def test_architecture_detection_arm64():
"""NTarm64 decoration -> architecture='arm64'."""
result = parse_inf(ARM64_INF, "arm64.inf", [])
assert result.architecture == "arm64"
def test_architecture_detection_undecorated():
"""Undecorated section only -> architecture='x86'."""
result = parse_inf(UNDECORATED_INF, "x86.inf", [])
assert result.architecture == "x86"
def test_architecture_detection_mixed():
"""Mixed architectures -> architecture=None (ambiguous)."""
result = parse_inf(MIXED_INF, "mixed.inf", [])
assert result.architecture is None
# ---------------------------------------------------------------------------
# parse_inf .cat file detection
# ---------------------------------------------------------------------------
def test_cat_file_detection_present():
"""zip_names containing a .cat file -> has_cat_file=True."""
result = parse_inf(SIMPLE_INF, "acme.inf", ["acme.inf", "driver.cat", "acme.dll"])
assert result.has_cat_file is True
def test_cat_file_detection_absent():
"""zip_names without a .cat file -> has_cat_file=False."""
result = parse_inf(SIMPLE_INF, "acme.inf", ["acme.inf", "acme.dll"])
assert result.has_cat_file is False
# ---------------------------------------------------------------------------
# parse_inf unused files detection
# ---------------------------------------------------------------------------
def test_unused_files():
"""ZIP members not referenced in INF text appear in unused_files."""
inf_text = """
[Version]
Signature="$Windows NT$"
[Manufacturer]
Mfg=Mfg.Models
[Mfg.Models]
Some Driver=Install,{DDDDDDDD-DDDD-DDDD-DDDD-DDDDDDDDDDDD}
[SourceDisksFiles]
driver.inf=1
driver.dll=1
[Strings]
"""
zip_names = ["driver.inf", "driver.dll", "readme.txt"]
result = parse_inf(inf_text, "driver.inf", zip_names)
# readme.txt is not mentioned anywhere in inf_text
assert "readme.txt" in result.unused_files
# driver.inf and driver.dll ARE referenced in inf_text
assert "driver.inf" not in result.unused_files
assert "driver.dll" not in result.unused_files
# ---------------------------------------------------------------------------
# parse_inf empty models section
# ---------------------------------------------------------------------------
EMPTY_MODELS_INF = """
[Version]
Signature="$Windows NT$"
[Manufacturer]
Mfg=MfgModels
[MfgModels]
[Strings]
"""
def test_empty_models_section():
"""INF with empty Models section returns empty driver_names list."""
result = parse_inf(EMPTY_MODELS_INF, "empty.inf", [])
assert result.driver_names == []
# ---------------------------------------------------------------------------
# parse_inf tolerates bare-line sections (real-world Ricoh oemsetup.inf)
# ---------------------------------------------------------------------------
BARE_LINE_INF = """\
[Version]
Signature="$Windows NT$"
[Manufacturer]
%MFG%=Models,NTamd64
[Models.NTamd64]
%RICOH_DRIVER%=Install,{CCCCCCCC-CCCC-CCCC-CCCC-CCCCCCCCCCCC}
[SourceDisksFiles]
ricu18ui.dll,ricu18ui.dl_
ricu18ui.irj
ricu18ui.rdj
ricu18gl.dll,ricu18gl.dl_
RD01Kd64.dll,RD01Kd64.dl_,,0x00000020
[Strings]
MFG="Ricoh"
RICOH_DRIVER="Ricoh PCL6 Universal"
"""
def test_bare_line_sections_do_not_raise():
"""Real INFs (e.g. Ricoh oemsetup.inf) include [SourceDisksFiles] entries
with bare filename lines and no '=' — parse_inf must not raise and must
still extract DriverDesc from the Models section."""
result = parse_inf(BARE_LINE_INF, "oemsetup.inf", ["oemsetup.inf"])
assert "Ricoh PCL6 Universal" in result.driver_names
# Synthetic __bare_N keys must not leak into driver_names
assert not any(n.startswith("__bare_") for n in result.driver_names)
+283
View File
@@ -0,0 +1,283 @@
"""
Byte-level validation tests for the .intunewin file format.
These tests serve as the format specification: if they pass, the byte layout is correct.
The only remaining validation is a real Intune upload (manual, Phase 10 gate).
Detection.xml format follows the IntuneWinAppUtil.exe reference exactly:
- ToolVersion is an XML *attribute* on <ApplicationInfo> (not a child element)
- No xmlns namespace (reference uses [XmlRoot("ApplicationInfo")] with no Namespace param)
- No <?xml?> declaration header (reference uses OmitXmlDeclaration=true)
- No MacAlgorithm element (not present in reference FileEncryptionInfo model)
"""
import base64
import hashlib
import hmac
import io
import os
import xml.etree.ElementTree as ET
import zipfile
import pytest
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from imptune.generators.intunewin_builder import _TOOL_VERSION, build_intunewin
@pytest.fixture
def source_dir(tmp_path):
"""Create a small test source directory with a few files."""
src = tmp_path / "source"
src.mkdir()
(src / "install.ps1").write_text("Write-Host 'Installing printer...'")
(src / "config.json").write_text('{"printer": "HP LaserJet"}')
(src / "readme.txt").write_text("Printer deployment package")
return str(src)
@pytest.fixture
def built_package(source_dir, tmp_path):
"""Build an .intunewin package and return the path."""
output = str(tmp_path / "package.intunewin")
build_intunewin(source_dir, "install.ps1", output)
return output
@pytest.fixture
def package_contents(built_package):
"""Extract outer ZIP contents and parsed Detection.xml."""
with zipfile.ZipFile(built_package, "r") as outer:
names = outer.namelist()
blob = outer.read("IntuneWinPackage/Contents/IntunePackage.intunewin")
detection_xml_bytes = outer.read("IntuneWinPackage/Metadata/Detection.xml")
tree = ET.fromstring(detection_xml_bytes.decode("utf-8"))
return {
"names": names,
"blob": blob,
"detection_xml_bytes": detection_xml_bytes,
"tree": tree,
}
def _get_xml_text(tree, tag):
"""Get text of a direct child element (no namespace — reference omits xmlns)."""
elem = tree.find(tag)
return elem.text if elem is not None else None
def _get_encryption_info(tree):
"""Get EncryptionInfo sub-element (no namespace — reference omits xmlns)."""
return tree.find("EncryptionInfo")
def _get_enc_text(tree, tag):
"""Get text of a child of EncryptionInfo (no namespace)."""
enc = _get_encryption_info(tree)
if enc is None:
return None
elem = enc.find(tag)
return elem.text if elem is not None else None
class TestOuterZipStructure:
def test_output_is_valid_zip(self, built_package):
"""build_intunewin() output file is a valid ZIP archive."""
assert zipfile.is_zipfile(built_package), "Output file must be a valid ZIP archive"
def test_outer_zip_structure(self, package_contents):
"""Outer ZIP contains exactly the two required entries."""
names = set(package_contents["names"])
assert "IntuneWinPackage/Contents/IntunePackage.intunewin" in names
assert "IntuneWinPackage/Metadata/Detection.xml" in names
def test_outer_zip_stored(self, built_package):
"""Outer ZIP entries use ZIP_STORED compression (no extra compression on encrypted content)."""
with zipfile.ZipFile(built_package, "r") as outer:
for info in outer.infolist():
assert info.compress_type == zipfile.ZIP_STORED, (
f"Entry {info.filename} must use ZIP_STORED, got compress_type={info.compress_type}"
)
class TestDetectionXml:
def test_detection_xml_valid(self, package_contents):
"""Detection.xml is valid XML with ApplicationInfo root element and ToolVersion attribute.
Reference format: <ApplicationInfo ToolVersion="1.8.6.0"> with NO xmlns namespace.
Presence of xmlns would change element identity for Intune's XML parser, causing
silent metadata-parse failure in the upload wizard.
"""
tree = package_contents["tree"]
assert tree.tag == "ApplicationInfo", (
f"Root element must be plain 'ApplicationInfo' (no xmlns namespace), got {tree.tag!r}"
)
assert tree.get("ToolVersion") == _TOOL_VERSION, (
f"ApplicationInfo must have ToolVersion attribute = {_TOOL_VERSION!r}, "
f"got {tree.get('ToolVersion')!r}"
)
def test_detection_xml_fields(self, package_contents):
"""Detection.xml contains all required elements matching the reference FileEncryptionInfo model.
The reference model has 7 EncryptionInfo sub-elements (MacAlgorithm is NOT present).
"""
tree = package_contents["tree"]
# Direct children
for field in ("Name", "UnencryptedContentSize", "FileName", "SetupFile"):
assert _get_xml_text(tree, field) is not None, f"Missing field: {field}"
# EncryptionInfo sub-elements — 7 required (MacAlgorithm absent per reference schema)
for field in (
"EncryptionKey",
"MacKey",
"InitializationVector",
"Mac",
"ProfileIdentifier",
"FileDigest",
"FileDigestAlgorithm",
):
assert _get_enc_text(tree, field) is not None, f"Missing EncryptionInfo/{field}"
# MacAlgorithm must NOT be present (not in reference FileEncryptionInfo model)
assert _get_enc_text(tree, "MacAlgorithm") is None, (
"EncryptionInfo/MacAlgorithm must NOT be present — not in reference schema"
)
def test_setup_file_in_detection_xml(self, package_contents):
"""SetupFile element matches the setup_file argument passed to build_intunewin."""
tree = package_contents["tree"]
assert _get_xml_text(tree, "SetupFile") == "install.ps1"
class TestEncryptedBlobLayout:
def test_encrypted_blob_layout(self, package_contents):
"""Encrypted blob starts with 32 bytes (HMAC) + 16 bytes (IV) + remainder (ciphertext)."""
blob = package_contents["blob"]
# Must be at least 48 bytes (HMAC + IV) plus at least one AES block (16 bytes)
assert len(blob) >= 64, f"Blob too short: {len(blob)} bytes"
# Total length = 48 header + ciphertext length; ciphertext length is a multiple of 16
ciphertext_len = len(blob) - 48
assert ciphertext_len > 0, "Blob has no ciphertext after header"
assert ciphertext_len % 16 == 0, (
f"Ciphertext length {ciphertext_len} must be a multiple of AES block size 16"
)
def test_iv_is_16_bytes(self, package_contents):
"""IV extracted from Detection.xml base64-decodes to exactly 16 bytes (NOT 32 — critical per RESEARCH.md)."""
tree = package_contents["tree"]
iv_b64 = _get_enc_text(tree, "InitializationVector")
assert iv_b64 is not None, "InitializationVector missing from Detection.xml"
iv = base64.b64decode(iv_b64)
assert len(iv) == 16, f"IV must be exactly 16 bytes, got {len(iv)}"
def test_encryption_key_is_32_bytes(self, package_contents):
"""EncryptionKey from Detection.xml base64-decodes to exactly 32 bytes."""
tree = package_contents["tree"]
key_b64 = _get_enc_text(tree, "EncryptionKey")
assert key_b64 is not None, "EncryptionKey missing from Detection.xml"
key = base64.b64decode(key_b64)
assert len(key) == 32, f"EncryptionKey must be exactly 32 bytes, got {len(key)}"
def test_mac_key_is_32_bytes(self, package_contents):
"""MacKey from Detection.xml base64-decodes to exactly 32 bytes."""
tree = package_contents["tree"]
mac_key_b64 = _get_enc_text(tree, "MacKey")
assert mac_key_b64 is not None, "MacKey missing from Detection.xml"
mac_key = base64.b64decode(mac_key_b64)
assert len(mac_key) == 32, f"MacKey must be exactly 32 bytes, got {len(mac_key)}"
class TestCryptographicVerification:
def test_hmac_matches(self, package_contents):
"""HMAC-SHA256 of (IV || ciphertext) matches first 32 bytes of blob AND Mac in Detection.xml.
The reference implementation (svrooij/ContentPrep Zipper.cs DecryptFileAsync) reads
the first 32 bytes as the stored HMAC, then computes the hash of the *remaining* bytes
(bytes[32:] = IV || ciphertext) to verify integrity. The IV MUST be included in the
HMAC so that a forged IV cannot redirect decryption without being detected.
"""
blob = package_contents["blob"]
tree = package_contents["tree"]
blob_hmac = blob[:32]
iv_and_ciphertext = blob[32:] # IV (16 bytes) + ciphertext — what the reference hashes
mac_key_b64 = _get_enc_text(tree, "MacKey")
mac_key = base64.b64decode(mac_key_b64)
computed_hmac = hmac.new(mac_key, iv_and_ciphertext, hashlib.sha256).digest()
# Must match the blob header
assert computed_hmac == blob_hmac, (
"HMAC-SHA256 of (IV || ciphertext) does not match the first 32 bytes of the blob"
)
# Must also match Detection.xml Mac field
xml_mac = base64.b64decode(_get_enc_text(tree, "Mac"))
assert computed_hmac == xml_mac, (
"HMAC-SHA256 of (IV || ciphertext) does not match the Mac value in Detection.xml"
)
def test_decryption_roundtrip(self, source_dir, package_contents):
"""Decrypt the ciphertext and verify it is a valid ZIP containing the original source files."""
blob = package_contents["blob"]
tree = package_contents["tree"]
aes_key = base64.b64decode(_get_enc_text(tree, "EncryptionKey"))
iv = base64.b64decode(_get_enc_text(tree, "InitializationVector"))
ciphertext = blob[48:]
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
plaintext = unpad(cipher.decrypt(ciphertext), AES.block_size)
# Must be a valid ZIP
assert zipfile.is_zipfile(io.BytesIO(plaintext)), (
"Decrypted plaintext is not a valid ZIP file"
)
# Must contain the original source files
with zipfile.ZipFile(io.BytesIO(plaintext), "r") as inner_zip:
inner_names = set(inner_zip.namelist())
for filename in ("install.ps1", "config.json", "readme.txt"):
assert filename in inner_names, (
f"Original file {filename} not found in decrypted inner ZIP. Found: {inner_names}"
)
def test_file_digest_matches(self, package_contents):
"""FileDigest in Detection.xml matches SHA256 of the decrypted plaintext ZIP."""
blob = package_contents["blob"]
tree = package_contents["tree"]
aes_key = base64.b64decode(_get_enc_text(tree, "EncryptionKey"))
iv = base64.b64decode(_get_enc_text(tree, "InitializationVector"))
ciphertext = blob[48:]
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
plaintext = unpad(cipher.decrypt(ciphertext), AES.block_size)
computed_digest = hashlib.sha256(plaintext).digest()
xml_digest = base64.b64decode(_get_enc_text(tree, "FileDigest"))
assert computed_digest == xml_digest, (
"FileDigest in Detection.xml does not match SHA256 of decrypted plaintext"
)
def test_unencrypted_content_size(self, package_contents):
"""UnencryptedContentSize in Detection.xml matches byte length of decrypted plaintext ZIP."""
blob = package_contents["blob"]
tree = package_contents["tree"]
aes_key = base64.b64decode(_get_enc_text(tree, "EncryptionKey"))
iv = base64.b64decode(_get_enc_text(tree, "InitializationVector"))
ciphertext = blob[48:]
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
plaintext = unpad(cipher.decrypt(ciphertext), AES.block_size)
xml_size = int(_get_xml_text(tree, "UnencryptedContentSize"))
assert xml_size == len(plaintext), (
f"UnencryptedContentSize {xml_size} does not match actual plaintext size {len(plaintext)}"
)
+256
View File
@@ -0,0 +1,256 @@
"""Integration tests for package export endpoints — NinjaRMM ZIP and .intunewin."""
import io
import json
import zipfile
import pytest
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def driver_zip_bytes():
"""Create a minimal valid driver ZIP with a fake INF and CAT file."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("printer.inf", "[Version]\nSignature=$WINDOWS NT$\n")
zf.writestr("printer.cat", "FAKE_CAT")
return buf.getvalue()
@pytest.fixture
def setup_printer_with_driver(tmp_data_dir, driver_zip_bytes):
"""Create Driver record (with ZIP on disk) and Printer record linked to it."""
import hashlib
import os
import imptune.config as cfg
from imptune.db.models import Driver, Printer
# Write driver ZIP to DRIVERS_DIR
sha = hashlib.sha256(driver_zip_bytes).hexdigest()
drivers_dir = cfg.DRIVERS_DIR
os.makedirs(drivers_dir, exist_ok=True)
zip_path = os.path.join(drivers_dir, f"{sha}.zip")
with open(zip_path, "wb") as f:
f.write(driver_zip_bytes)
driver = Driver.create(
sha256=sha,
original_filename="printer_driver.zip",
size_bytes=len(driver_zip_bytes),
driver_desc=json.dumps(["HP LaserJet Pro"]),
inf_filename="printer.inf",
)
printer = Printer.create(
name="Test Printer",
ip_address="192.168.1.100",
port_name="IP_192.168.1.100",
driver=driver,
duplex_mode="OneSided",
color_mode=True,
paper_size="A4",
collate=True,
)
return printer, driver
@pytest.fixture
def printer_no_driver(tmp_data_dir):
"""Create Printer record with no driver assigned."""
from imptune.db.models import Printer
return Printer.create(
name="No Driver Printer",
ip_address="10.0.0.1",
port_name="IP_10.0.0.1",
driver=None,
)
# ---------------------------------------------------------------------------
# NinjaRMM ZIP endpoint
# ---------------------------------------------------------------------------
class TestNinjaDownload:
def test_returns_zip(self, client, setup_printer_with_driver):
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/packages/ninja")
assert resp.status_code == 200
assert resp.headers["content-type"] == "application/zip"
assert "attachment" in resp.headers["content-disposition"]
assert ".zip" in resp.headers["content-disposition"]
def test_zip_contains_install_script(self, client, setup_printer_with_driver):
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/packages/ninja")
assert resp.status_code == 200
safe_name = printer.name.replace(" ", "_")
with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
names = zf.namelist()
assert f"{safe_name}/install.ps1" in names
def test_zip_contains_driver_files(self, client, setup_printer_with_driver):
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/packages/ninja")
assert resp.status_code == 200
safe_name = printer.name.replace(" ", "_")
with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
names = zf.namelist()
# Driver ZIP contained printer.inf and printer.cat
assert f"{safe_name}/drivers/printer.inf" in names
assert f"{safe_name}/drivers/printer.cat" in names
def test_404_missing_printer(self, client, tmp_data_dir):
resp = client.get("/printers/9999/packages/ninja")
assert resp.status_code == 404
def test_422_no_driver(self, client, printer_no_driver):
resp = client.get(f"/printers/{printer_no_driver.id}/packages/ninja")
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# .intunewin endpoint
# ---------------------------------------------------------------------------
class TestIntunewinDownload:
def test_returns_intunewin(self, client, setup_printer_with_driver):
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/packages/intunewin")
assert resp.status_code == 200
assert resp.headers["content-type"] == "application/octet-stream"
assert "attachment" in resp.headers["content-disposition"]
assert ".intunewin" in resp.headers["content-disposition"]
def test_intunewin_is_valid_zip(self, client, setup_printer_with_driver):
"""Outer .intunewin file must be a valid ZIP with IntuneWinPackage/ structure."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/packages/intunewin")
assert resp.status_code == 200
with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
names = zf.namelist()
assert any(n.startswith("IntuneWinPackage/") for n in names)
def test_404_missing_printer(self, client, tmp_data_dir):
resp = client.get("/printers/9999/packages/intunewin")
assert resp.status_code == 404
def test_422_no_driver(self, client, printer_no_driver):
resp = client.get(f"/printers/{printer_no_driver.id}/packages/intunewin")
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# Printer detail page — command preview and export links
# ---------------------------------------------------------------------------
class TestCommandPreview:
def test_detail_page_shows_commands(self, client, setup_printer_with_driver):
"""Printer detail page shows install and uninstall command strings."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}")
assert resp.status_code == 200
html = resp.text
assert "install-cmd" in html
assert "uninstall-cmd" in html
assert "install.ps1" in html
assert "uninstall.ps1" in html
def test_detail_page_shows_export_links(self, client, setup_printer_with_driver):
"""Printer detail page shows NinjaRMM ZIP and .intunewin download links."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}")
assert resp.status_code == 200
html = resp.text
assert f"/printers/{printer.id}/packages/ninja" in html
assert f"/printers/{printer.id}/packages/intunewin" in html
def test_detail_page_hides_commands_without_driver(self, client, printer_no_driver):
"""Detail page hides command section and export links when no driver assigned."""
resp = client.get(f"/printers/{printer_no_driver.id}")
assert resp.status_code == 200
html = resp.text
assert "install-cmd" not in html
assert "packages/ninja" not in html
assert "packages/intunewin" not in html
def test_detail_page_shows_icon_upload_form(self, client, setup_printer_with_driver):
"""Printer detail page always shows icon upload form."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}")
assert resp.status_code == 200
html = resp.text
assert f"/printers/{printer.id}/icon" in html
assert "icon-status" in html
def test_detail_page_shows_script_links(self, client, setup_printer_with_driver):
"""Printer detail page shows 3 direct .ps1 script download links when driver assigned."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}")
assert resp.status_code == 200
html = resp.text
assert f"/printers/{printer.id}/scripts/install.ps1" in html
assert f"/printers/{printer.id}/scripts/uninstall.ps1" in html
assert f"/printers/{printer.id}/scripts/detect.ps1" in html
# ---------------------------------------------------------------------------
# Icon inclusion in .intunewin export
# ---------------------------------------------------------------------------
class TestIntunewinIconInclusion:
def test_intunewin_includes_icon(self, client, setup_printer_with_driver, tmp_data_dir, monkeypatch):
"""When a printer has an uploaded icon, icon.png must appear in the .intunewin staging directory."""
import io as _io
import os
from PIL import Image
printer, _ = setup_printer_with_driver
# Upload a 256x256 PNG icon for the printer
buf = _io.BytesIO()
Image.new("RGBA", (256, 256), color="red").save(buf, format="PNG")
buf.seek(0)
upload_resp = client.post(
f"/printers/{printer.id}/icon",
files={"file": ("icon.png", buf, "image/png")},
)
assert upload_resp.status_code == 200
# Monkeypatch build_intunewin to capture staged files and write a fake output
staged_files: list[str] = []
def fake_build(source_dir, setup_file, output_path):
staged_files.extend(os.listdir(source_dir))
with open(output_path, "wb") as f:
f.write(b"FAKE")
monkeypatch.setattr("imptune.api.packages.build_intunewin", fake_build)
resp = client.get(f"/printers/{printer.id}/packages/intunewin")
assert resp.status_code == 200
assert "icon.png" in staged_files
def test_intunewin_without_icon_succeeds(self, client, setup_printer_with_driver, monkeypatch):
"""When a printer has no icon, .intunewin export must succeed without error."""
import os
# Monkeypatch build_intunewin to write a fake output
def fake_build(source_dir, setup_file, output_path):
with open(output_path, "wb") as f:
f.write(b"FAKE")
monkeypatch.setattr("imptune.api.packages.build_intunewin", fake_build)
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/packages/intunewin")
assert resp.status_code == 200
+389
View File
@@ -0,0 +1,389 @@
"""Integration tests for printer and client CRUD endpoints."""
import json
import pytest
from fastapi.testclient import TestClient
# ---------------------------------------------------------------------------
# Task 1: RED state — routes do not exist yet, tests should fail
# ---------------------------------------------------------------------------
def test_create_printer_persisted(client: TestClient) -> None:
"""POST /printers with required fields returns 303; GET /printers contains printer name."""
from imptune.db.models import Printer
resp = client.post(
"/printers",
data={
"name": "Test Printer",
"ip_address": "192.168.1.100",
"port_name": "IP_192_168_1_100",
},
follow_redirects=False,
)
assert resp.status_code == 303
# GET /printers page contains the printer name
page = client.get("/printers")
assert page.status_code == 200
assert "Test Printer" in page.text
# Verify DB persistence
count = Printer.select().where(Printer.name == "Test Printer").count()
assert count == 1
def test_create_printer_duplex(client: TestClient) -> None:
"""POST /printers with duplex_mode=LongEdge persists correctly."""
from imptune.db.models import Printer
resp = client.post(
"/printers",
data={
"name": "Duplex Printer",
"ip_address": "192.168.1.101",
"port_name": "IP_192_168_1_101",
"duplex_mode": "LongEdge",
},
follow_redirects=False,
)
assert resp.status_code == 303
printers = list(Printer.select().where(Printer.name == "Duplex Printer"))
assert len(printers) == 1
assert printers[0].duplex_mode == "LongEdge"
def test_create_printer_color_mode(client: TestClient) -> None:
"""POST /printers with color_mode not sent (unchecked) sets color_mode=False."""
from imptune.db.models import Printer
resp = client.post(
"/printers",
data={
"name": "Mono Printer",
"ip_address": "192.168.1.102",
"port_name": "IP_192_168_1_102",
# color_mode intentionally omitted (unchecked checkbox)
},
follow_redirects=False,
)
assert resp.status_code == 303
printers = list(Printer.select().where(Printer.name == "Mono Printer"))
assert len(printers) == 1
assert printers[0].color_mode is False
def test_create_printer_paper_size(client: TestClient) -> None:
"""POST /printers with paper_size=Letter persists correctly."""
from imptune.db.models import Printer
resp = client.post(
"/printers",
data={
"name": "Letter Printer",
"ip_address": "192.168.1.103",
"port_name": "IP_192_168_1_103",
"paper_size": "Letter",
},
follow_redirects=False,
)
assert resp.status_code == 303
printers = list(Printer.select().where(Printer.name == "Letter Printer"))
assert len(printers) == 1
assert printers[0].paper_size == "Letter"
def test_create_printer_collate(client: TestClient) -> None:
"""POST /printers with collate not sent (unchecked) sets collate=False."""
from imptune.db.models import Printer
resp = client.post(
"/printers",
data={
"name": "No Collate Printer",
"ip_address": "192.168.1.104",
"port_name": "IP_192_168_1_104",
# collate intentionally omitted (unchecked checkbox)
},
follow_redirects=False,
)
assert resp.status_code == 303
printers = list(Printer.select().where(Printer.name == "No Collate Printer"))
assert len(printers) == 1
assert printers[0].collate is False
def test_create_client(client: TestClient) -> None:
"""POST /clients with name=Contoso returns 200; GET /clients contains Contoso."""
resp = client.post("/clients", data={"name": "Contoso"})
assert resp.status_code == 200
page = client.get("/clients")
assert page.status_code == 200
assert "Contoso" in page.text
def test_printer_grouped_by_client(client: TestClient) -> None:
"""Printer assigned to a client appears under that client group heading."""
from imptune.db.models import Client, Printer
# Create client
resp = client.post("/clients", data={"name": "Contoso"})
assert resp.status_code == 200
contoso_list = list(Client.select().where(Client.name == "Contoso"))
assert len(contoso_list) == 1
contoso = contoso_list[0]
# Create printer assigned to that client
resp = client.post(
"/printers",
data={
"name": "Contoso Printer",
"ip_address": "10.0.0.1",
"port_name": "IP_10_0_0_1",
"client_id": str(contoso.id),
},
follow_redirects=False,
)
assert resp.status_code == 303
# GET /printers should show "Contoso" as a group header
page = client.get("/printers")
assert page.status_code == 200
html = page.text
assert "Contoso" in html
# Client name should appear in a heading element
assert "<h3" in html
def test_create_printer_missing_name(client: TestClient) -> None:
"""POST /printers with empty name returns 400."""
resp = client.post(
"/printers",
data={
"name": "",
"ip_address": "192.168.1.100",
"port_name": "IP_192_168_1_100",
},
)
assert resp.status_code == 400
def test_create_printer_invalid_ip(client: TestClient) -> None:
"""POST /printers with empty ip_address returns 400."""
resp = client.post(
"/printers",
data={
"name": "Valid Name",
"ip_address": "",
"port_name": "SOME_PORT",
},
)
assert resp.status_code == 400
def test_printer_detail_shows_driver(client: TestClient) -> None:
"""GET /printers/{id} returns 200 with all printer fields and driver name."""
from imptune.db.models import Driver, Printer
driver_obj = Driver.create(
sha256="abc123",
original_filename="hp_universal.zip",
size_bytes=1000,
driver_desc=json.dumps(["HP Universal"]),
)
printer = Printer.create(
name="HP Office Printer",
ip_address="10.0.1.1",
port_name="IP_10_0_1_1",
driver=driver_obj,
)
resp = client.get(f"/printers/{printer.id}")
assert resp.status_code == 200
html = resp.text
assert "HP Office Printer" in html
assert "10.0.1.1" in html
assert "HP Universal" in html
def test_printer_detail_not_found(client: TestClient) -> None:
"""GET /printers/9999 returns 404."""
resp = client.get("/printers/9999")
assert resp.status_code == 404
def test_printer_detail_no_driver(client: TestClient) -> None:
"""GET /printers/{id} for printer with no driver returns 200 with 'No driver assigned'."""
from imptune.db.models import Printer
printer = Printer.create(
name="Driverless Printer",
ip_address="10.0.1.2",
port_name="IP_10_0_1_2",
driver=None,
)
resp = client.get(f"/printers/{printer.id}")
assert resp.status_code == 200
html = resp.text
assert "No driver assigned" in html
def test_delete_printer(client: TestClient) -> None:
"""DELETE /printers/{id} removes the printer; GET /printers no longer shows it."""
from imptune.db.models import Printer
# Create a printer
resp = client.post(
"/printers",
data={
"name": "To Delete",
"ip_address": "192.168.1.200",
"port_name": "IP_192_168_1_200",
},
follow_redirects=False,
)
assert resp.status_code == 303
# Find its ID
printers = list(Printer.select().where(Printer.name == "To Delete"))
assert len(printers) == 1
printer_id = printers[0].id
# Delete it
resp = client.delete(f"/printers/{printer_id}")
assert resp.status_code == 200
# Confirm it's gone from the DB
count = Printer.select().where(Printer.id == printer_id).count()
assert count == 0
# GET /printers should no longer show it
page = client.get("/printers")
assert "To Delete" not in page.text
# ---------------------------------------------------------------------------
# Wave 0 scaffolds — UIE-01 (PATCH), UIE-02 (separated form/list), UIE-03 (client detail)
# These tests are RED until plan 01 task 2 and plans 02-03 implement the routes.
# ---------------------------------------------------------------------------
def test_printers_new_returns_200(client: TestClient) -> None:
"""GET /printers/new returns 200 with the add printer form markup."""
resp = client.get("/printers/new")
assert resp.status_code == 200
html = resp.text
# Must contain form fields
assert 'name="name"' in html or "Printer Name" in html
def test_create_printer_redirects(client: TestClient) -> None:
"""POST /printers (no HX-Request header) returns 303 redirect to /printers."""
resp = client.post(
"/printers",
data={
"name": "Redirect Printer",
"ip_address": "192.168.2.1",
"port_name": "IP_192_168_2_1",
},
follow_redirects=False,
)
assert resp.status_code == 303
assert resp.headers["location"] == "/printers"
def test_printers_library_no_form(client: TestClient) -> None:
"""GET /printers does NOT contain the inline add-printer form markup."""
resp = client.get("/printers")
assert resp.status_code == 200
html = resp.text
# The inline form must be gone — check for the actual form element, not UI label strings
assert 'hx-post="/printers"' not in html
assert 'action="/printers" method="post"' not in html
def test_patch_printer(client: TestClient) -> None:
"""PATCH /printers/{id} with updated name returns 200, updated name in response, DB updated."""
from imptune.db.models import Printer
printer = Printer.create(
name="Original Name",
ip_address="10.0.2.1",
port_name="IP_10_0_2_1",
)
resp = client.patch(
f"/printers/{printer.id}",
data={"name": "Updated Name"},
)
assert resp.status_code == 200
assert "Updated Name" in resp.text
# Verify DB update
updated = Printer.get_by_id(printer.id)
assert updated.name == "Updated Name"
def test_patch_printer_not_found(client: TestClient) -> None:
"""PATCH /printers/9999 returns 404."""
resp = client.patch("/printers/9999", data={"name": "Ghost"})
assert resp.status_code == 404
def test_client_detail_returns_200(client: TestClient) -> None:
"""GET /clients/{id} returns 200 with client name and assigned printer name."""
from imptune.db.models import Client, Printer
# Create client
cl = Client.create(name="Detail Client")
# Create printer assigned to that client
Printer.create(
name="Client Printer",
ip_address="10.0.3.1",
port_name="IP_10_0_3_1",
client=cl,
)
resp = client.get(f"/clients/{cl.id}")
assert resp.status_code == 200
html = resp.text
assert "Detail Client" in html
assert "Client Printer" in html
def test_client_detail_not_found(client: TestClient) -> None:
"""GET /clients/9999 returns 404."""
resp = client.get("/clients/9999")
assert resp.status_code == 404
def test_client_links_in_printer_list(client: TestClient) -> None:
"""GET /printers with a printer assigned to a client contains href to client detail."""
from imptune.db.models import Client, Printer
# Create client via POST /clients
resp = client.post("/clients", data={"name": "Link Client"})
assert resp.status_code == 200
cl = Client.get(Client.name == "Link Client")
# Create printer assigned to that client
Printer.create(
name="Linked Printer",
ip_address="10.0.4.1",
port_name="IP_10_0_4_1",
client=cl,
)
resp = client.get("/printers")
assert resp.status_code == 200
assert f'href="/clients/{cl.id}"' in resp.text
+41
View File
@@ -0,0 +1,41 @@
"""Tests for printer form template wiring — inline driver upload + OOB select."""
from fastapi.testclient import TestClient
def test_printer_form_has_inline_driver_upload(client: TestClient) -> None:
"""GET /printers/new renders printer form with:
- Driver <select> with stable id="printer-form-driver-select"
- Inline upload form with caller=printer_form hidden field
- hx-post="/drivers/upload" for the driver upload sub-form
- Plain <form action="/printers" method="post"> for the printer (no HTMX on main form)
Updated in 11-01: form moved from /printers to /printers/new (UIE-02).
"""
resp = client.get("/printers/new")
assert resp.status_code == 200
html = resp.text
# Driver select has stable id for OOB swap target
assert 'id="printer-form-driver-select"' in html, (
'Printer form driver <select> must have id="printer-form-driver-select" '
"for HTMX OOB swap to work"
)
# Inline upload form posts caller=printer_form sentinel
assert 'name="caller"' in html, 'Inline upload form missing name="caller" field'
assert 'value="printer_form"' in html, (
'Inline upload form missing value="printer_form" sentinel'
)
# Inline upload form targets /drivers/upload
assert 'hx-post="/drivers/upload"' in html, (
'Inline upload form missing hx-post="/drivers/upload"'
)
# Main printer form uses plain POST (no HTMX) so browser follows 303 redirect
assert 'action="/printers"' in html, (
'Printer form must use action="/printers" (plain HTML form, not hx-post)'
)
assert 'method="post"' in html, (
'Printer form must use method="post"'
)
+120
View File
@@ -0,0 +1,120 @@
"""Integration tests for the .ps1-suffixed script download routes."""
import io
import json
import zipfile
import pytest
# ---------------------------------------------------------------------------
# Fixtures (same pattern as tests/test_packages.py)
# ---------------------------------------------------------------------------
@pytest.fixture
def driver_zip_bytes():
"""Create a minimal valid driver ZIP with a fake INF and CAT file."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("printer.inf", "[Version]\nSignature=$WINDOWS NT$\n")
zf.writestr("printer.cat", "FAKE_CAT")
return buf.getvalue()
@pytest.fixture
def setup_printer_with_driver(tmp_data_dir, driver_zip_bytes):
"""Create Driver record (with ZIP on disk) and Printer record linked to it."""
import hashlib
import os
import imptune.config as cfg
from imptune.db.models import Driver, Printer
sha = hashlib.sha256(driver_zip_bytes).hexdigest()
drivers_dir = cfg.DRIVERS_DIR
os.makedirs(drivers_dir, exist_ok=True)
zip_path = os.path.join(drivers_dir, f"{sha}.zip")
with open(zip_path, "wb") as f:
f.write(driver_zip_bytes)
driver = Driver.create(
sha256=sha,
original_filename="printer_driver.zip",
size_bytes=len(driver_zip_bytes),
driver_desc=json.dumps(["HP LaserJet Pro"]),
inf_filename="printer.inf",
)
printer = Printer.create(
name="Test Printer",
ip_address="192.168.1.100",
port_name="IP_192.168.1.100",
driver=driver,
duplex_mode="OneSided",
color_mode=True,
paper_size="A4",
collate=True,
)
return printer, driver
@pytest.fixture
def printer_no_driver(tmp_data_dir):
"""Create Printer record with no driver assigned."""
from imptune.db.models import Printer
return Printer.create(
name="No Driver Printer",
ip_address="10.0.0.1",
port_name="IP_10.0.0.1",
driver=None,
)
# ---------------------------------------------------------------------------
# .ps1 route tests
# ---------------------------------------------------------------------------
class TestPs1Routes:
def test_install_ps1_route(self, client, setup_printer_with_driver):
"""GET /printers/{id}/scripts/install.ps1 returns 200 with attachment and PowerShell content."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/scripts/install.ps1")
assert resp.status_code == 200
assert 'attachment' in resp.headers["content-disposition"]
assert 'filename="install.ps1"' in resp.headers["content-disposition"]
body = resp.text
assert len(body) > 0
assert "Add-Printer" in body or "$PSScriptRoot" in body
def test_uninstall_ps1_route(self, client, setup_printer_with_driver):
"""GET /printers/{id}/scripts/uninstall.ps1 returns 200 with attachment and uninstall content."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/scripts/uninstall.ps1")
assert resp.status_code == 200
assert 'attachment' in resp.headers["content-disposition"]
assert 'filename="uninstall.ps1"' in resp.headers["content-disposition"]
body = resp.text
assert len(body) > 0
assert "Remove-Printer" in body
def test_detect_ps1_route(self, client, setup_printer_with_driver):
"""GET /printers/{id}/scripts/detect.ps1 returns 200 with attachment and detect content."""
printer, _ = setup_printer_with_driver
resp = client.get(f"/printers/{printer.id}/scripts/detect.ps1")
assert resp.status_code == 200
assert 'attachment' in resp.headers["content-disposition"]
assert 'filename="detect.ps1"' in resp.headers["content-disposition"]
body = resp.text
assert len(body) > 0
assert "Get-Printer" in body
def test_ps1_routes_missing_printer(self, client, tmp_data_dir):
"""GET for non-existent printer returns 404."""
resp = client.get("/printers/99999/scripts/install.ps1")
assert resp.status_code == 404
def test_ps1_routes_no_driver(self, client, printer_no_driver):
"""GET for printer without driver returns 422."""
resp = client.get(f"/printers/{printer_no_driver.id}/scripts/install.ps1")
assert resp.status_code == 422
+197
View File
@@ -0,0 +1,197 @@
"""Unit tests for script_generator.py — SCRPT-01, SCRPT-02, SCRPT-03, SCRPT-04, SCRPT-05."""
import pytest
from imptune.generators.script_generator import render_detect, render_install, render_uninstall
def _create_test_driver_and_printer():
"""Helper: create a Driver + Printer for integration tests."""
from imptune.db.models import Driver, Printer
driver = Driver.create(
sha256="abc123",
original_filename="test.zip",
size_bytes=100,
driver_desc='["Test Driver"]',
inf_filename="test.inf",
)
printer = Printer.create(
name="Test Printer",
ip_address="10.0.0.1",
port_name="IP_10.0.0.1",
driver=driver,
duplex_mode="LongEdge",
color_mode=True,
paper_size="A4",
collate=True,
)
return driver, printer
SAMPLE_ARGS = dict(
printer_name="HP LaserJet 4000",
ip_address="192.168.1.10",
port_name="IP_192.168.1.10",
driver_name="HP Universal Printing PCL 6",
inf_filename="hpcu270u.inf",
duplex_mode="LongEdge",
color_mode=True,
paper_size="A4",
collate=True,
)
def test_render_install_returns_string():
"""render_install returns a non-empty string."""
result = render_install(**SAMPLE_ARGS)
assert isinstance(result, str)
assert len(result) > 0
def test_render_install_contains_pnputil():
"""SCRPT-01: Output contains pnputil /add-driver and Add-PrinterDriver."""
result = render_install(**SAMPLE_ARGS)
assert "pnputil.exe /add-driver" in result
assert "Add-PrinterDriver" in result
assert "hpcu270u.inf" in result
assert "HP Universal Printing PCL 6" in result
def test_render_install_print_config():
"""SCRPT-01: Set-PrintConfiguration with translated duplex mode."""
# LongEdge -> TwoSidedLongEdge
result = render_install(**SAMPLE_ARGS)
assert "Set-PrintConfiguration" in result
assert "TwoSidedLongEdge" in result
assert "A4" in result
# ShortEdge -> TwoSidedShortEdge
result_short = render_install(**{**SAMPLE_ARGS, "duplex_mode": "ShortEdge"})
assert "TwoSidedShortEdge" in result_short
# OneSided -> OneSided (no change)
result_one = render_install(**{**SAMPLE_ARGS, "duplex_mode": "OneSided"})
assert "OneSided" in result_one
# Must NOT contain the two-sided variants when OneSided
assert "TwoSided" not in result_one
def test_render_install_wow64_guard():
"""SCRPT-05: WOW64 relaunch guard is present in the output."""
result = render_install(**SAMPLE_ARGS)
assert "PROCESSOR_ARCHITECTURE" in result
assert "SysNative" in result
# Guard must appear near the top (before pnputil)
wow64_pos = result.find("PROCESSOR_ARCHITECTURE")
pnputil_pos = result.find("pnputil.exe")
assert wow64_pos < pnputil_pos, "WOW64 guard must appear before pnputil"
def test_render_install_uac_guard():
"""SCRPT-04: SYSTEM vs user detection with UAC self-elevation."""
result = render_install(**SAMPLE_ARGS)
assert "IsSystem" in result
assert "Start-Process" in result
assert "-Verb Runas" in result or "Verb Runas" in result
def test_render_install_idempotency():
"""SCRPT-01: All add operations wrapped in idempotency checks."""
result = render_install(**SAMPLE_ARGS)
# Port idempotency
assert "Get-PrinterPort" in result
assert "Add-PrinterPort" in result
# Printer idempotency
assert "Get-Printer" in result
assert "Add-Printer" in result
# Port check must come before Add-PrinterPort
get_port_pos = result.find("Get-PrinterPort")
add_port_pos = result.find("Add-PrinterPort")
assert get_port_pos < add_port_pos, "Get-PrinterPort check must precede Add-PrinterPort"
def test_render_install_booleans():
"""Color and collate are rendered as $true / $false in the script."""
result_true = render_install(**SAMPLE_ARGS) # color_mode=True, collate=True
assert "$true" in result_true
result_false = render_install(**{**SAMPLE_ARGS, "color_mode": False, "collate": False})
assert "$false" in result_false
def test_render_uninstall():
"""SCRPT-02: render_uninstall produces correct removal order with SilentlyContinue."""
result = render_uninstall("Test Printer", "HP Driver", "IP_10.0.0.1")
# All three Remove-* commands must be present
assert 'Remove-Printer -Name "Test Printer"' in result
assert 'Remove-PrinterDriver -Name "HP Driver"' in result
assert 'Remove-PrinterPort -Name "IP_10.0.0.1"' in result
# All must have -ErrorAction SilentlyContinue
assert result.count("-ErrorAction SilentlyContinue") >= 3
# Order: Remove-Printer BEFORE Remove-PrinterDriver (driver removal fails if printer still active)
printer_pos = result.find("Remove-Printer -Name")
driver_pos = result.find("Remove-PrinterDriver -Name")
port_pos = result.find("Remove-PrinterPort -Name")
assert printer_pos < driver_pos, "Remove-Printer must appear before Remove-PrinterDriver"
assert driver_pos < port_pos, "Remove-PrinterDriver must appear before Remove-PrinterPort"
def test_render_detect():
"""SCRPT-03: render_detect follows Intune detection contract (Write-Output + exit 0/1)."""
result = render_detect("Test Printer")
assert 'Get-Printer -Name "Test Printer"' in result
assert "Write-Output" in result
assert "exit 0" in result
assert "exit 1" in result
# ---------------------------------------------------------------------------
# Integration tests — script API endpoints
# ---------------------------------------------------------------------------
def test_install_endpoint(client):
"""GET /printers/{id}/scripts/install returns 200 with pnputil in content."""
_driver, printer = _create_test_driver_and_printer()
response = client.get(f"/printers/{printer.id}/scripts/install")
assert response.status_code == 200
assert "pnputil" in response.text
def test_uninstall_endpoint(client):
"""GET /printers/{id}/scripts/uninstall returns 200 with Remove-Printer in content."""
_driver, printer = _create_test_driver_and_printer()
response = client.get(f"/printers/{printer.id}/scripts/uninstall")
assert response.status_code == 200
assert "Remove-Printer" in response.text
def test_detect_endpoint(client):
"""GET /printers/{id}/scripts/detect returns 200 with Write-Output in content."""
_driver, printer = _create_test_driver_and_printer()
response = client.get(f"/printers/{printer.id}/scripts/detect")
assert response.status_code == 200
assert "Write-Output" in response.text
def test_script_endpoint_missing_printer(client):
"""GET /printers/{id}/scripts/install returns 404 for nonexistent printer."""
response = client.get("/printers/9999/scripts/install")
assert response.status_code == 404
def test_script_endpoint_no_driver(client):
"""GET /printers/{id}/scripts/install returns 422 when no driver assigned."""
from imptune.db.models import Printer
printer = Printer.create(
name="No Driver Printer",
ip_address="10.0.0.2",
port_name="IP_10.0.0.2",
driver=None,
duplex_mode="OneSided",
color_mode=True,
paper_size="A4",
collate=True,
)
response = client.get(f"/printers/{printer.id}/scripts/install")
assert response.status_code == 422
+113
View File
@@ -0,0 +1,113 @@
import re
from pathlib import Path
CDN_PATTERNS = re.compile(
r"(cdn\.jsdelivr\.net|unpkg\.com|cdnjs\.com)",
re.IGNORECASE,
)
# Matches href="https://..." or src="https://..."
EXTERNAL_URL_PATTERN = re.compile(
r'(?:href|src)=["\']https?://',
re.IGNORECASE,
)
def test_no_cdn_urls_in_templates():
"""All .html templates use /static/ paths only — no CDN URLs in href/src attributes."""
templates_dir = Path(__file__).parent.parent / "imptune" / "templates"
html_files = list(templates_dir.glob("**/*.html"))
assert html_files, "No HTML templates found — check templates directory path"
violations = []
for html_file in html_files:
content = html_file.read_text(encoding="utf-8")
if CDN_PATTERNS.search(content):
violations.append(f"{html_file.name}: contains CDN domain reference")
if EXTERNAL_URL_PATTERN.search(content):
violations.append(f"{html_file.name}: contains external https:// in href/src")
assert not violations, "CDN/external URL violations found:\n" + "\n".join(violations)
def test_dashboard_returns_200(client):
"""GET / returns 200 (dashboard page)."""
response = client.get("/")
assert response.status_code == 200
def test_packages_returns_200(client):
"""GET /packages returns 200 (packages listing page)."""
response = client.get("/packages")
assert response.status_code == 200
def test_theme_toggle_present(client):
"""GET / contains a theme toggle button (data-theme cycling control)."""
response = client.get("/")
assert response.status_code == 200
# The button's @click should reference $store.theme.cycle
assert "theme" in response.text
assert "cycle" in response.text or "store.theme" in response.text
def test_dashboard_shows_recent_printers(client):
"""Dashboard renders names of recently-created printers from DB."""
from imptune.db.models import Printer
Printer.create(
name="TestPrinter-Alpha",
ip_address="10.0.0.1",
port_name="IP_10.0.0.1",
)
Printer.create(
name="TestPrinter-Beta",
ip_address="10.0.0.2",
port_name="IP_10.0.0.2",
)
response = client.get("/")
assert response.status_code == 200
assert "TestPrinter-Alpha" in response.text
assert "TestPrinter-Beta" in response.text
# Use class-specific check: the string also appears in the i18n store JS
assert 'class="empty-state">No printers configured yet' not in response.text
def test_theme_toggle_present(client):
"""GET / contains a theme toggle button (data-theme cycling control)."""
response = client.get("/")
assert response.status_code == 200
# The button's @click should reference $store.theme.cycle
assert "theme" in response.text
assert "cycle" in response.text or "store.theme" in response.text
def test_dashboard_shows_recent_packages(client):
"""Dashboard recent-packages section shows only printers with a driver assigned."""
from imptune.db.models import Driver, Printer
driver = Driver.create(
sha256="abc123",
original_filename="test.zip",
size_bytes=1000,
driver_desc='["TestDriver"]',
)
Printer.create(
name="PkgPrinter-Assigned",
ip_address="10.0.0.2",
port_name="IP_10.0.0.2",
driver=driver,
)
Printer.create(
name="PkgPrinter-NoDriver",
ip_address="10.0.0.3",
port_name="IP_10.0.0.3",
)
response = client.get("/")
assert response.status_code == 200
assert "PkgPrinter-Assigned" in response.text
assert 'class="empty-state">No packages exported yet' not in response.text
+52
View File
@@ -0,0 +1,52 @@
"""End-to-end guard: upload a driver through the HTTP endpoint, then export
a NinjaRMM package from the resulting Driver record. This prevents regressions
where DriverStore save path and packages.py driver lookup path diverge."""
import io
import json
import zipfile
import pytest
@pytest.fixture
def driver_zip_bytes():
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("printer.inf", "[Version]\nSignature=$WINDOWS NT$\n")
zf.writestr("printer.cat", "FAKE_CAT")
return buf.getvalue()
def test_upload_then_ninja_export_finds_driver_on_disk(
client, tmp_data_dir, driver_zip_bytes
):
from imptune.db.models import Client, Driver, Printer
upload_resp = client.post(
"/drivers/upload",
files={"file": ("printer_driver.zip", driver_zip_bytes, "application/zip")},
)
assert upload_resp.status_code == 200
driver = Driver.select().order_by(Driver.id.desc()).first()
assert driver is not None
driver.driver_desc = json.dumps(["HP LaserJet Pro"])
driver.save()
tenant = Client.create(name="Acme Corp")
printer = Printer.create(
name="Round Trip Printer",
ip_address="192.168.1.50",
port_name="IP_192.168.1.50",
client=tenant,
driver=driver,
)
resp = client.get(f"/printers/{printer.id}/packages/ninja")
assert resp.status_code == 200, resp.text
assert resp.headers["content-type"] == "application/zip"
out = zipfile.ZipFile(io.BytesIO(resp.content))
names = out.namelist()
assert any(n.endswith("install.ps1") for n in names)
assert any("drivers/printer.inf" in n for n in names)