v0.1.27
Fix: Throughput API v1 available - Details posted to Irving. POWERBI_KEY was missing from the .ENV file, so was not live. Add: Editor now supports editing a mix's resolved formula directly, with % and kg dual entry on ingredient rows Fix: Mix Editor should bring through correct ingredients. New resolved formula (same logic we use in Mix Calculator). Fix: Security headers on all API responses (hardening) Add: New mix button available on the Mix Editor. Add: New ingredient button available on the Ingredient Editor
This commit is contained in:
@@ -0,0 +1,102 @@
|
||||
"""The Mix Editor must resolve and edit the SAME formula the Mix Calculator reads.
|
||||
|
||||
These cover `resolve_editor_mix_formula` / `resolve_representative_product`: a mix
|
||||
whose product carries its own formula resolves to that product (not the shared
|
||||
mix master), percentages are computed against the total, and the representative
|
||||
product is chosen the way the calculator chooses it.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from app.db.session import Base
|
||||
from app.models.mix import Mix, MixIngredient
|
||||
from app.models.product import Product, ProductIngredient
|
||||
from app.models.raw_material import RawMaterial
|
||||
from app.services.mix_calculator_service import (
|
||||
resolve_editor_mix_formula,
|
||||
resolve_representative_product,
|
||||
)
|
||||
|
||||
TENANT = "hunter-premium-produce"
|
||||
|
||||
|
||||
def _session() -> Session:
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(bind=engine)
|
||||
return sessionmaker(bind=engine, expire_on_commit=False)()
|
||||
|
||||
|
||||
def _raw(db: Session, name: str) -> RawMaterial:
|
||||
material = RawMaterial(tenant_id=TENANT, name=name, unit_of_measure="kg", kg_per_unit=1, status="active")
|
||||
db.add(material)
|
||||
db.flush()
|
||||
return material
|
||||
|
||||
|
||||
def test_resolves_product_formula_not_mix_master():
|
||||
db = _session()
|
||||
bayley = _raw(db, "Bayley")
|
||||
filler = _raw(db, "Filler")
|
||||
|
||||
mix = Mix(tenant_id=TENANT, client_name="Hunter", name="Pigeon Mix")
|
||||
db.add(mix)
|
||||
db.flush()
|
||||
|
||||
# Shared mix master says something different from the product formula.
|
||||
db.add(MixIngredient(tenant_id=TENANT, mix_id=mix.id, raw_material_id=bayley.id, quantity_kg=100))
|
||||
db.add(MixIngredient(tenant_id=TENANT, mix_id=mix.id, raw_material_id=filler.id, quantity_kg=100))
|
||||
|
||||
product = Product(tenant_id=TENANT, client_name="Hunter", name="Pigeon 20kg", mix_id=mix.id, unit_of_measure="20kg bag", visible=True)
|
||||
db.add(product)
|
||||
db.flush()
|
||||
# The calculator's real numbers live here: 787.5 / 1320.41 ~ 59.6%.
|
||||
db.add(ProductIngredient(tenant_id=TENANT, product_id=product.id, raw_material_id=bayley.id, quantity_kg=787.5, sort_order=1))
|
||||
db.add(ProductIngredient(tenant_id=TENANT, product_id=product.id, raw_material_id=filler.id, quantity_kg=532.91, sort_order=2))
|
||||
db.commit()
|
||||
|
||||
formula = resolve_editor_mix_formula(db, tenant_id=TENANT, mix=mix)
|
||||
|
||||
assert formula["source"] == "product"
|
||||
assert formula["product_id"] == product.id
|
||||
assert formula["total_kg"] == 1320.41
|
||||
by_name = {row["raw_material_name"]: row for row in formula["ingredients"]}
|
||||
assert by_name["Bayley"]["quantity_kg"] == 787.5
|
||||
# Percentage matches the worked example (787.5 / 1320.41 * 100).
|
||||
assert abs(by_name["Bayley"]["mix_percentage"] - 59.6406) < 0.001
|
||||
|
||||
|
||||
def test_falls_back_to_mix_master_when_no_product_formula():
|
||||
db = _session()
|
||||
maize = _raw(db, "Maize")
|
||||
mix = Mix(tenant_id=TENANT, client_name="Hunter", name="Plain Mix")
|
||||
db.add(mix)
|
||||
db.flush()
|
||||
db.add(MixIngredient(tenant_id=TENANT, mix_id=mix.id, raw_material_id=maize.id, quantity_kg=50))
|
||||
db.commit()
|
||||
|
||||
formula = resolve_editor_mix_formula(db, tenant_id=TENANT, mix=mix)
|
||||
assert formula["source"] == "mix"
|
||||
assert formula["total_kg"] == 50
|
||||
assert formula["ingredients"][0]["mix_percentage"] == 100.0
|
||||
|
||||
|
||||
def test_representative_product_prefers_20kg_bag():
|
||||
db = _session()
|
||||
maize = _raw(db, "Maize")
|
||||
mix = Mix(tenant_id=TENANT, client_name="Hunter", name="Dual Mix")
|
||||
db.add(mix)
|
||||
db.flush()
|
||||
|
||||
bulka = Product(tenant_id=TENANT, client_name="Hunter", name="Dual Bulka", mix_id=mix.id, unit_of_measure="500kg bulka", visible=True)
|
||||
bag = Product(tenant_id=TENANT, client_name="Hunter", name="Dual 20kg", mix_id=mix.id, unit_of_measure="20kg bag", visible=True)
|
||||
db.add_all([bulka, bag])
|
||||
db.flush()
|
||||
for product in (bulka, bag):
|
||||
db.add(ProductIngredient(tenant_id=TENANT, product_id=product.id, raw_material_id=maize.id, quantity_kg=20, sort_order=1))
|
||||
db.commit()
|
||||
|
||||
representative = resolve_representative_product(db, tenant_id=TENANT, mix_id=mix.id)
|
||||
assert representative is not None
|
||||
assert representative.unit_of_measure == "20kg bag"
|
||||
@@ -0,0 +1,115 @@
|
||||
"""Tests for the read-only external data API (`/api/v1`) used by Power BI.
|
||||
|
||||
Drives the real FastAPI app via TestClient against an in-memory database, and
|
||||
swaps the configured API key in by replacing the module-level `settings` object
|
||||
(the real one is a frozen dataclass).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import replace
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.api import public_v1
|
||||
from app.core.config import settings
|
||||
from app.db.session import Base, get_db
|
||||
from app.main import app
|
||||
from app.models.throughput import ProductionThroughput
|
||||
|
||||
API_KEY = "test-powerbi-key"
|
||||
TENANT = "hunter-premium-produce"
|
||||
|
||||
|
||||
def _seed_entry(db, *, tenant_id: str, product: str, quantity: float, when: date) -> None:
|
||||
db.add(
|
||||
ProductionThroughput(
|
||||
tenant_id=tenant_id,
|
||||
production_date=when,
|
||||
product_name_snapshot=product,
|
||||
bag_size=20,
|
||||
quantity=quantity,
|
||||
quantity_type="bags",
|
||||
calculated_kg=quantity * 20,
|
||||
created_by="test",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def db_factory():
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
TestingSession = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
|
||||
|
||||
def override_get_db():
|
||||
db = TestingSession()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
yield TestingSession
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def _make_client(monkeypatch, *, api_key: str) -> TestClient:
|
||||
# Replace the whole settings object the router reads (frozen dataclass).
|
||||
monkeypatch.setattr(public_v1, "settings", replace(settings, powerbi_api_key=api_key, powerbi_tenant_id=TENANT))
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_disabled_when_key_unset(monkeypatch, db_factory):
|
||||
client = _make_client(monkeypatch, api_key="")
|
||||
response = client.get("/api/v1/throughput", headers={"X-API-Key": "anything"})
|
||||
assert response.status_code == 503
|
||||
|
||||
|
||||
def test_rejects_missing_and_wrong_key(monkeypatch, db_factory):
|
||||
client = _make_client(monkeypatch, api_key=API_KEY)
|
||||
assert client.get("/api/v1/throughput").status_code == 401
|
||||
assert client.get("/api/v1/throughput", headers={"X-API-Key": "nope"}).status_code == 401
|
||||
|
||||
|
||||
def test_returns_entries_for_tenant(monkeypatch, db_factory):
|
||||
db = db_factory()
|
||||
_seed_entry(db, tenant_id=TENANT, product="Maize", quantity=10, when=date(2026, 1, 5))
|
||||
_seed_entry(db, tenant_id=TENANT, product="Barley", quantity=5, when=date(2026, 1, 6))
|
||||
# An entry in another tenant must never leak through.
|
||||
_seed_entry(db, tenant_id="someone-else", product="Secret", quantity=99, when=date(2026, 1, 7))
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
client = _make_client(monkeypatch, api_key=API_KEY)
|
||||
|
||||
# Header auth.
|
||||
response = client.get("/api/v1/throughput", headers={"X-API-Key": API_KEY})
|
||||
assert response.status_code == 200
|
||||
rows = response.json()
|
||||
names = [row["product_name_snapshot"] for row in rows]
|
||||
assert names == ["Maize", "Barley"] # oldest first, other tenant excluded
|
||||
assert rows[0]["calculated_kg"] == 200.0
|
||||
|
||||
# Query-string auth works too (Power BI Web connector convenience).
|
||||
assert client.get(f"/api/v1/throughput?api_key={API_KEY}").status_code == 200
|
||||
|
||||
|
||||
def test_date_filter(monkeypatch, db_factory):
|
||||
db = db_factory()
|
||||
_seed_entry(db, tenant_id=TENANT, product="Old", quantity=1, when=date(2026, 1, 1))
|
||||
_seed_entry(db, tenant_id=TENANT, product="New", quantity=1, when=date(2026, 2, 1))
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
client = _make_client(monkeypatch, api_key=API_KEY)
|
||||
response = client.get("/api/v1/throughput?date_from=2026-01-15", headers={"X-API-Key": API_KEY})
|
||||
assert [row["product_name_snapshot"] for row in response.json()] == ["New"]
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Regression guards for the print/PDF Content-Security-Policy.
|
||||
|
||||
The in-app print dialog loads a generated PDF as a same-origin ``blob:`` URL into
|
||||
an iframe and calls ``contentWindow.print()``. If the CSP omits ``frame-src`` /
|
||||
``child-src`` for ``blob:`` the directive falls back to ``default-src 'self'``,
|
||||
which silently blocks the frame and breaks printing for every user (regardless of
|
||||
role). These tests pin the policy on both layers that emit it:
|
||||
|
||||
* the FastAPI security middleware (covers every API response), and
|
||||
* the production nginx config (the source of the *document* CSP that actually
|
||||
governs ``frame-src`` in the browser).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.security import issue_token
|
||||
from app.main import app
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
NGINX_CONFIGS = [REPO_ROOT / "deploy" / "nginx" / "clients.lean-101.conf"]
|
||||
|
||||
# Directives the print flow depends on. blob: must be framable, and that must not
|
||||
# come at the cost of dropping the same-origin baseline.
|
||||
REQUIRED_FRAME_SOURCES = {"'self'", "blob:"}
|
||||
|
||||
|
||||
def _parse_csp(header: str) -> dict[str, set[str]]:
|
||||
"""Parse a CSP header string into ``{directive: {sources}}``."""
|
||||
directives: dict[str, set[str]] = {}
|
||||
for part in header.split(";"):
|
||||
tokens = part.split()
|
||||
if not tokens:
|
||||
continue
|
||||
directives[tokens[0].lower()] = set(tokens[1:])
|
||||
return directives
|
||||
|
||||
|
||||
def _assert_blob_framing(header: str) -> None:
|
||||
csp = _parse_csp(header)
|
||||
# frame-src must exist and allow self + blob (no falling back to default-src).
|
||||
assert "frame-src" in csp, f"frame-src missing from CSP: {header!r}"
|
||||
assert REQUIRED_FRAME_SOURCES <= csp["frame-src"], (
|
||||
f"frame-src must allow {REQUIRED_FRAME_SOURCES}, got {csp['frame-src']}"
|
||||
)
|
||||
# child-src is the Safari fallback for frame-src; keep it aligned.
|
||||
assert "child-src" in csp, f"child-src missing from CSP: {header!r}"
|
||||
assert "blob:" in csp["child-src"], f"child-src must allow blob:, got {csp['child-src']}"
|
||||
# We only widened framing: the same-origin default must stay intact.
|
||||
assert csp.get("default-src") == {"'self'"}, f"default-src weakened: {csp.get('default-src')}"
|
||||
|
||||
|
||||
# --- Backend middleware policy ------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client() -> TestClient:
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
def test_backend_csp_allows_blob_frames(client: TestClient) -> None:
|
||||
response = client.get("/health")
|
||||
assert "content-security-policy" in response.headers
|
||||
_assert_blob_framing(response.headers["content-security-policy"])
|
||||
|
||||
|
||||
def test_backend_csp_present_for_all_users(client: TestClient) -> None:
|
||||
"""The policy is identical for anonymous, authenticated, and rejected (401)
|
||||
requests, so printing can never depend on who is signed in."""
|
||||
admin_token = issue_token({"name": "Admin", "email": settings.admin_email, "role": "admin"})
|
||||
|
||||
responses = [
|
||||
client.get("/health"), # anonymous
|
||||
client.get("/api/access/me"), # the endpoint that 401s for warehouse users
|
||||
client.get("/api/access/me", headers={"Authorization": f"Bearer {admin_token}"}),
|
||||
]
|
||||
|
||||
policies = set()
|
||||
for response in responses:
|
||||
header = response.headers.get("content-security-policy")
|
||||
assert header is not None, f"CSP missing on {response.request.url} ({response.status_code})"
|
||||
_assert_blob_framing(header)
|
||||
policies.add(header)
|
||||
|
||||
assert len(policies) == 1, "CSP must not vary by authentication state"
|
||||
|
||||
|
||||
# --- Production document policy (nginx) ---------------------------------------
|
||||
|
||||
|
||||
def test_nginx_csp_allows_blob_frames() -> None:
|
||||
"""Every CSP the production nginx emits must allow blob framing. This guards
|
||||
the *document* policy, which is what the browser enforces for the print iframe."""
|
||||
csp_line = re.compile(r'Content-Security-Policy\s+"([^"]+)"', re.IGNORECASE)
|
||||
|
||||
for config_path in NGINX_CONFIGS:
|
||||
assert config_path.exists(), f"missing nginx config: {config_path}"
|
||||
text = config_path.read_text(encoding="utf-8")
|
||||
policies = csp_line.findall(text)
|
||||
assert policies, f"no Content-Security-Policy header found in {config_path}"
|
||||
for policy in policies:
|
||||
_assert_blob_framing(policy)
|
||||
Reference in New Issue
Block a user