allucanget/calminer
2
0
Fork 0
Code Pull Requests Actions Packages Releases Activity

v2 init

Browse Source
This commit is contained in:
zwitschi
2025-11-09 16:49:27 +01:00
parent 22ddfb671d
commit d807a50f77
96 changed files with 3 additions and 9689 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
tests/__init__.py
View File

170
tests/e2e/conftest.py
View File

@@ -1,170 +0,0 @@
import os
import subprocess
import time
from typing import Dict, Generator
import pytest
# type: ignore[import]
from playwright.sync_api import Browser, Page, Playwright, sync_playwright
import httpx
from sqlalchemy.engine import make_url
# Use a different port for the test server to avoid conflicts
TEST_PORT = 8001
BASE_URL = f"http://localhost:{TEST_PORT}"
@pytest.fixture(scope="session", autouse=True)
def live_server() -> Generator[str, None, None]:
"""Launch a live test server in a separate process."""
env = _prepare_database_environment(os.environ.copy())
process = subprocess.Popen(
[
"uvicorn",
"main:app",
"--host",
"127.0.0.1",
f"--port={TEST_PORT}",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env=env,
)
deadline = time.perf_counter() + 30
last_error: Exception | None = None
while time.perf_counter() < deadline:
if process.poll() is not None:
raise RuntimeError("uvicorn server exited before becoming ready")
try:
response = httpx.get(BASE_URL, timeout=1.0, trust_env=False)
if response.status_code < 500:
break
except Exception as exc: # noqa: BLE001
last_error = exc
time.sleep(0.5)
else:
process.terminate()
process.wait(timeout=5)
raise TimeoutError(
"Timed out waiting for uvicorn test server to start"
) from last_error
try:
yield BASE_URL
finally:
if process.poll() is None:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=5)
@pytest.fixture(scope="session", autouse=True)
def seed_default_currencies(live_server: str) -> None:
"""Ensure a baseline set of currencies exists for UI flows."""
seeds = [
{"code": "EUR", "name": "Euro", "symbol": "EUR", "is_active": True},
{
"code": "CLP",
"name": "Chilean Peso",
"symbol": "CLP$",
"is_active": True,
},
]
with httpx.Client(
base_url=live_server, timeout=5.0, trust_env=False
) as client:
try:
response = client.get("/api/currencies/?include_inactive=true")
response.raise_for_status()
existing_codes = {
str(item.get("code"))
for item in response.json()
if isinstance(item, dict) and item.get("code")
}
except httpx.HTTPError as exc: # noqa: BLE001
raise RuntimeError("Failed to read existing currencies") from exc
for payload in seeds:
if payload["code"] in existing_codes:
continue
try:
create_response = client.post("/api/currencies/", json=payload)
except httpx.HTTPError as exc: # noqa: BLE001
raise RuntimeError("Failed to seed currencies") from exc
if create_response.status_code == 409:
continue
create_response.raise_for_status()
@pytest.fixture(scope="session")
def playwright_instance() -> Generator[Playwright, None, None]:
"""Provide a Playwright instance for the test session."""
with sync_playwright() as p:
yield p
@pytest.fixture(scope="session")
def browser(
playwright_instance: Playwright,
) -> Generator[Browser, None, None]:
"""Provide a browser instance for the test session."""
browser = playwright_instance.chromium.launch()
yield browser
browser.close()
@pytest.fixture()
def page(browser: Browser, live_server: str) -> Generator[Page, None, None]:
"""Provide a new page for each test."""
page = browser.new_page(base_url=live_server)
page.goto("/")
page.wait_for_load_state("networkidle")
yield page
page.close()
def _prepare_database_environment(env: Dict[str, str]) -> Dict[str, str]:
"""Ensure granular database env vars are available for the app under test."""
required = (
"DATABASE_HOST",
"DATABASE_USER",
"DATABASE_NAME",
"DATABASE_PASSWORD",
)
if all(env.get(key) for key in required):
return env
legacy_url = env.get("DATABASE_URL")
if not legacy_url:
return env
url = make_url(legacy_url)
env.setdefault("DATABASE_DRIVER", url.drivername)
if url.host:
env.setdefault("DATABASE_HOST", url.host)
if url.port:
env.setdefault("DATABASE_PORT", str(url.port))
if url.username:
env.setdefault("DATABASE_USER", url.username)
if url.password:
env.setdefault("DATABASE_PASSWORD", url.password)
if url.database:
env.setdefault("DATABASE_NAME", url.database)
query_options = dict(url.query) if url.query else {}
options = query_options.get("options")
if isinstance(options, str) and "search_path=" in options:
env.setdefault("DATABASE_SCHEMA", options.split("search_path=")[-1])
return env

50
tests/e2e/test_consumption.py
View File

@@ -1,50 +0,0 @@
from uuid import uuid4
from playwright.sync_api import Page, expect
def test_consumption_form_loads(page: Page):
"""Verify the consumption form page loads correctly."""
page.goto("/ui/consumption")
expect(page).to_have_title("Consumption · CalMiner")
expect(
page.locator("h2:has-text('Add Consumption Record')")
).to_be_visible()
def test_create_consumption_item(page: Page):
"""Test creating a new consumption item through the UI."""
# First, create a scenario to associate the consumption with.
page.goto("/ui/scenarios")
scenario_name = f"Consumption Test Scenario {uuid4()}"
page.fill("input[name='name']", scenario_name)
page.click("button[type='submit']")
with page.expect_response("**/api/scenarios/"):
pass # Wait for the scenario to be created
# Now, navigate to the consumption page and add an item.
page.goto("/ui/consumption")
# Create a consumption item.
consumption_desc = "Diesel for generators"
page.select_option("#consumption-form-scenario", label=scenario_name)
page.fill("textarea[name='description']", consumption_desc)
page.fill("input[name='amount']", "5000")
page.click("button[type='submit']")
with page.expect_response("**/api/consumption/") as response_info:
pass
assert response_info.value.status == 201
# Verify the new item appears in the table.
page.select_option("#consumption-scenario-filter", label=scenario_name)
expect(
page.locator("#consumption-table-body tr").filter(
has_text=consumption_desc
)
).to_be_visible()
# Verify the feedback message.
expect(page.locator("#consumption-feedback")).to_have_text(
"Consumption record saved."
)

63
tests/e2e/test_costs.py
View File

@@ -1,63 +0,0 @@
from uuid import uuid4
from playwright.sync_api import Page, expect
def test_costs_form_loads(page: Page):
"""Verify the costs form page loads correctly."""
page.goto("/ui/costs")
expect(page).to_have_title("Costs · CalMiner")
expect(page.locator("h2:has-text('Add CAPEX Entry')")).to_be_visible()
def test_create_capex_and_opex_items(page: Page):
"""Test creating new CAPEX and OPEX items through the UI."""
# First, create a scenario to associate the costs with.
page.goto("/ui/scenarios")
scenario_name = f"Cost Test Scenario {uuid4()}"
page.fill("input[name='name']", scenario_name)
page.click("button[type='submit']")
with page.expect_response("**/api/scenarios/"):
pass # Wait for the scenario to be created
# Now, navigate to the costs page and add CAPEX and OPEX items.
page.goto("/ui/costs")
# Create a CAPEX item.
capex_desc = "Initial drilling equipment"
page.select_option("#capex-form-scenario", label=scenario_name)
page.fill("#capex-form-description", capex_desc)
page.fill("#capex-form-amount", "150000")
page.click("#capex-form button[type='submit']")
with page.expect_response("**/api/costs/capex") as response_info:
pass
assert response_info.value.status == 200
# Create an OPEX item.
opex_desc = "Monthly fuel costs"
page.select_option("#opex-form-scenario", label=scenario_name)
page.fill("#opex-form-description", opex_desc)
page.fill("#opex-form-amount", "25000")
page.click("#opex-form button[type='submit']")
with page.expect_response("**/api/costs/opex") as response_info:
pass
assert response_info.value.status == 200
# Verify the new items appear in their respective tables.
page.select_option("#costs-scenario-filter", label=scenario_name)
expect(
page.locator("#capex-table-body tr").filter(has_text=capex_desc)
).to_be_visible()
expect(
page.locator("#opex-table-body tr").filter(has_text=opex_desc)
).to_be_visible()
# Verify the feedback messages.
expect(page.locator("#capex-feedback")).to_have_text(
"Entry saved successfully."
)
expect(page.locator("#opex-feedback")).to_have_text(
"Entry saved successfully."
)

135
tests/e2e/test_currencies.py
View File

@@ -1,135 +0,0 @@
import random
import string
from playwright.sync_api import Page, expect
def _unique_currency_code(existing: set[str]) -> str:
"""Generate a unique three-letter code not present in *existing*."""
alphabet = string.ascii_uppercase
for _ in range(100):
candidate = "".join(random.choices(alphabet, k=3))
if candidate not in existing and candidate != "USD":
return candidate
raise AssertionError(
"Unable to generate a unique currency code for the test run."
)
def _metric_value(page: Page, element_id: str) -> int:
locator = page.locator(f"#{element_id}")
expect(locator).to_be_visible()
return int(locator.inner_text().strip())
def _expect_feedback(page: Page, expected_text: str) -> None:
page.wait_for_function(
"expected => {"
" const el = document.getElementById('currency-form-feedback');"
" if (!el) return false;"
" const text = (el.textContent || '').trim();"
" return !el.classList.contains('hidden') && text === expected;"
"}",
arg=expected_text,
)
feedback = page.locator("#currency-form-feedback")
expect(feedback).to_have_text(expected_text)
def test_currency_workflow_create_update_toggle(page: Page) -> None:
"""Exercise create, update, and toggle flows on the currency settings page."""
page.goto("/ui/currencies")
expect(page).to_have_title("Currencies · CalMiner")
expect(page.locator("h2:has-text('Currency Overview')")).to_be_visible()
code_cells = page.locator("#currencies-table-body tr td:nth-child(1)")
existing_codes = {
text.strip().upper() for text in code_cells.all_inner_texts()
}
total_before = _metric_value(page, "currency-metric-total")
active_before = _metric_value(page, "currency-metric-active")
inactive_before = _metric_value(page, "currency-metric-inactive")
new_code = _unique_currency_code(existing_codes)
new_name = f"Test Currency {new_code}"
new_symbol = new_code[0]
page.fill("#currency-form-code", new_code)
page.fill("#currency-form-name", new_name)
page.fill("#currency-form-symbol", new_symbol)
page.select_option("#currency-form-status", "true")
with page.expect_response("**/api/currencies/") as create_info:
page.click("button[type='submit']")
create_response = create_info.value
assert create_response.status == 201
_expect_feedback(page, "Currency created successfully.")
page.wait_for_function(
"expected => Number(document.getElementById('currency-metric-total').textContent.trim()) === expected",
arg=total_before + 1,
)
page.wait_for_function(
"expected => Number(document.getElementById('currency-metric-active').textContent.trim()) === expected",
arg=active_before + 1,
)
row = page.locator("#currencies-table-body tr").filter(has_text=new_code)
expect(row).to_be_visible()
expect(row.locator("td").nth(3)).to_have_text("Active")
# Switch to update mode using the existing currency option.
page.select_option("#currency-form-existing", new_code)
updated_name = f"{new_name} Updated"
updated_symbol = f"{new_symbol}$"
page.fill("#currency-form-name", updated_name)
page.fill("#currency-form-symbol", updated_symbol)
page.select_option("#currency-form-status", "false")
with page.expect_response(f"**/api/currencies/{new_code}") as update_info:
page.click("button[type='submit']")
update_response = update_info.value
assert update_response.status == 200
_expect_feedback(page, "Currency updated successfully.")
page.wait_for_function(
"expected => Number(document.getElementById('currency-metric-active').textContent.trim()) === expected",
arg=active_before,
)
page.wait_for_function(
"expected => Number(document.getElementById('currency-metric-inactive').textContent.trim()) === expected",
arg=inactive_before + 1,
)
expect(row.locator("td").nth(1)).to_have_text(updated_name)
expect(row.locator("td").nth(2)).to_have_text(updated_symbol)
expect(row.locator("td").nth(3)).to_contain_text("Inactive")
toggle_button = row.locator("button[data-action='toggle']")
expect(toggle_button).to_have_text("Activate")
with page.expect_response(
f"**/api/currencies/{new_code}/activation"
) as toggle_info:
toggle_button.click()
toggle_response = toggle_info.value
assert toggle_response.status == 200
page.wait_for_function(
"expected => Number(document.getElementById('currency-metric-active').textContent.trim()) === expected",
arg=active_before + 1,
)
page.wait_for_function(
"expected => Number(document.getElementById('currency-metric-inactive').textContent.trim()) === expected",
arg=inactive_before,
)
_expect_feedback(page, f"Currency {new_code} activated.")
expect(row.locator("td").nth(3)).to_contain_text("Active")
expect(row.locator("button[data-action='toggle']")).to_have_text(
"Deactivate"
)

17
tests/e2e/test_dashboard.py
View File

@@ -1,17 +0,0 @@
from playwright.sync_api import Page, expect
def test_dashboard_loads_and_has_title(page: Page):
"""Verify the dashboard page loads and the title is correct."""
expect(page).to_have_title("Dashboard · CalMiner")
def test_dashboard_shows_summary_metrics_panel(page: Page):
"""Check that the summary metrics panel is visible."""
expect(page.locator("h2:has-text('Operations Overview')")).to_be_visible()
def test_dashboard_renders_cost_chart(page: Page):
"""Ensure the scenario cost chart canvas is present."""
expect(page.locator("#cost-chart")).to_be_attached()
expect(page.locator("#cost-chart-empty")).to_be_visible()

45
tests/e2e/test_equipment.py
View File

@@ -1,45 +0,0 @@
from uuid import uuid4
from playwright.sync_api import Page, expect
def test_equipment_form_loads(page: Page):
"""Verify the equipment form page loads correctly."""
page.goto("/ui/equipment")
expect(page).to_have_title("Equipment · CalMiner")
expect(page.locator("h2:has-text('Add Equipment')")).to_be_visible()
def test_create_equipment_item(page: Page):
"""Test creating a new equipment item through the UI."""
# First, create a scenario to associate the equipment with.
page.goto("/ui/scenarios")
scenario_name = f"Equipment Test Scenario {uuid4()}"
page.fill("input[name='name']", scenario_name)
page.click("button[type='submit']")
with page.expect_response("**/api/scenarios/"):
pass # Wait for the scenario to be created
# Now, navigate to the equipment page and add an item.
page.goto("/ui/equipment")
# Create an equipment item.
equipment_name = "Haul Truck HT-05"
equipment_desc = "Primary haul truck for ore transport."
page.select_option("#equipment-form-scenario", label=scenario_name)
page.fill("#equipment-form-name", equipment_name)
page.fill("#equipment-form-description", equipment_desc)
page.click("button[type='submit']")
with page.expect_response("**/api/equipment/") as response_info:
pass
assert response_info.value.status == 200
# Verify the new item appears in the table.
page.select_option("#equipment-scenario-filter", label=scenario_name)
expect(
page.locator("#equipment-table-body tr").filter(has_text=equipment_name)
).to_be_visible()
# Verify the feedback message.
expect(page.locator("#equipment-feedback")).to_have_text("Equipment saved.")

58
tests/e2e/test_maintenance.py
View File

@@ -1,58 +0,0 @@
from uuid import uuid4
from playwright.sync_api import Page, expect
def test_maintenance_form_loads(page: Page):
"""Verify the maintenance form page loads correctly."""
page.goto("/ui/maintenance")
expect(page).to_have_title("Maintenance · CalMiner")
expect(page.locator("h2:has-text('Add Maintenance Entry')")).to_be_visible()
def test_create_maintenance_item(page: Page):
"""Test creating a new maintenance item through the UI."""
# First, create a scenario and an equipment item.
page.goto("/ui/scenarios")
scenario_name = f"Maintenance Test Scenario {uuid4()}"
page.fill("input[name='name']", scenario_name)
page.click("button[type='submit']")
with page.expect_response("**/api/scenarios/"):
pass
page.goto("/ui/equipment")
equipment_name = f"Excavator EX-12 {uuid4()}"
page.select_option("#equipment-form-scenario", label=scenario_name)
page.fill("#equipment-form-name", equipment_name)
page.click("button[type='submit']")
with page.expect_response("**/api/equipment/"):
pass
# Now, navigate to the maintenance page and add an item.
page.goto("/ui/maintenance")
# Create a maintenance item.
maintenance_desc = "Scheduled engine overhaul"
page.select_option("#maintenance-form-scenario", label=scenario_name)
page.select_option("#maintenance-form-equipment", label=equipment_name)
page.fill("#maintenance-form-date", "2025-12-01")
page.fill("#maintenance-form-description", maintenance_desc)
page.fill("#maintenance-form-cost", "12000")
page.click("button[type='submit']")
with page.expect_response("**/api/maintenance/") as response_info:
pass
assert response_info.value.status == 201
# Verify the new item appears in the table.
page.select_option("#maintenance-scenario-filter", label=scenario_name)
expect(
page.locator("#maintenance-table-body tr").filter(
has_text=maintenance_desc
)
).to_be_visible()
# Verify the feedback message.
expect(page.locator("#maintenance-feedback")).to_have_text(
"Maintenance entry saved."
)

48
tests/e2e/test_production.py
View File

@@ -1,48 +0,0 @@
from uuid import uuid4
from playwright.sync_api import Page, expect
def test_production_form_loads(page: Page):
"""Verify the production form page loads correctly."""
page.goto("/ui/production")
expect(page).to_have_title("Production · CalMiner")
expect(page.locator("h2:has-text('Add Production Output')")).to_be_visible()
def test_create_production_item(page: Page):
"""Test creating a new production item through the UI."""
# First, create a scenario to associate the production with.
page.goto("/ui/scenarios")
scenario_name = f"Production Test Scenario {uuid4()}"
page.fill("input[name='name']", scenario_name)
page.click("button[type='submit']")
with page.expect_response("**/api/scenarios/"):
pass # Wait for the scenario to be created
# Now, navigate to the production page and add an item.
page.goto("/ui/production")
# Create a production item.
production_desc = "Ore extracted - Grade A"
page.select_option("#production-form-scenario", label=scenario_name)
page.fill("#production-form-description", production_desc)
page.fill("#production-form-amount", "1500")
page.click("button[type='submit']")
with page.expect_response("**/api/production/") as response_info:
pass
assert response_info.value.status == 201
# Verify the new item appears in the table.
page.select_option("#production-scenario-filter", label=scenario_name)
expect(
page.locator("#production-table-body tr").filter(
has_text=production_desc
)
).to_be_visible()
# Verify the feedback message.
expect(page.locator("#production-feedback")).to_have_text(
"Production output saved."
)

9
tests/e2e/test_reporting.py
View File

@@ -1,9 +0,0 @@
from playwright.sync_api import Page, expect
def test_reporting_view_loads(page: Page):
"""Verify the reporting view page loads correctly."""
page.get_by_role("link", name="Reporting").click()
expect(page).to_have_url("http://localhost:8001/ui/reporting")
expect(page).to_have_title("Reporting · CalMiner")
expect(page.locator("h2:has-text('Scenario KPI Summary')")).to_be_visible()

43
tests/e2e/test_scenarios.py
View File

@@ -1,43 +0,0 @@
from uuid import uuid4
from playwright.sync_api import Page, expect
def test_scenario_form_loads(page: Page):
"""Verify the scenario form page loads correctly."""
page.goto("/ui/scenarios")
expect(page).to_have_url(
"http://localhost:8001/ui/scenarios"
) # Updated port
expect(page.locator("h2:has-text('Create a New Scenario')")).to_be_visible()
def test_create_new_scenario(page: Page):
"""Test creating a new scenario via the UI form."""
page.goto("/ui/scenarios")
scenario_name = f"E2E Test Scenario {uuid4()}"
scenario_desc = "A scenario created during an end-to-end test."
page.fill("input[name='name']", scenario_name)
page.fill("input[name='description']", scenario_desc)
# Expect a network response from the POST request after clicking the submit button.
with page.expect_response("**/api/scenarios/") as response_info:
page.click("button[type='submit']")
response = response_info.value
assert response.status == 200
# After a successful submission, the new scenario should be visible in the table.
# The table is dynamically updated, so we might need to wait for it to appear.
new_row = page.locator(f"tr:has-text('{scenario_name}')")
expect(new_row).to_be_visible()
expect(new_row.locator("td").nth(1)).to_have_text(scenario_desc)
# Verify the feedback message.
feedback = page.locator("#feedback")
expect(feedback).to_be_visible()
expect(feedback).to_have_text(
f'Scenario "{scenario_name}" created successfully.'
)

85
tests/e2e/test_smoke.py
View File

@@ -1,85 +0,0 @@
import pytest
from playwright.sync_api import Page, expect
# A list of UI routes to check, with their URL, expected title, and a key heading text.
UI_ROUTES = [
("/", "Dashboard · CalMiner", "Operations Overview"),
("/ui/dashboard", "Dashboard · CalMiner", "Operations Overview"),
(
"/ui/scenarios",
"Scenario Management · CalMiner",
"Create a New Scenario",
),
("/ui/parameters", "Process Parameters · CalMiner", "Scenario Parameters"),
("/ui/settings", "Settings · CalMiner", "Settings"),
("/ui/costs", "Costs · CalMiner", "Cost Overview"),
("/ui/consumption", "Consumption · CalMiner", "Consumption Tracking"),
("/ui/production", "Production · CalMiner", "Production Output"),
("/ui/equipment", "Equipment · CalMiner", "Equipment Inventory"),
("/ui/maintenance", "Maintenance · CalMiner", "Maintenance Schedule"),
("/ui/simulations", "Simulations · CalMiner", "Monte Carlo Simulations"),
("/ui/reporting", "Reporting · CalMiner", "Scenario KPI Summary"),
("/ui/currencies", "Currencies · CalMiner", "Currency Overview"),
]
@pytest.mark.parametrize("url, title, heading", UI_ROUTES)
def test_ui_pages_load_correctly(
page: Page, url: str, title: str, heading: str
):
"""Verify that all UI pages load with the correct title and a visible heading."""
page.goto(url)
expect(page).to_have_title(title)
# The app uses a mix of h1 and h2 for main page headings.
heading_locator = page.locator(
f"h1:has-text('{heading}'), h2:has-text('{heading}')"
)
expect(heading_locator.first).to_be_visible()
def test_settings_theme_form_interaction(page: Page):
page.goto("/theme-settings")
expect(page).to_have_title("Theme Settings · CalMiner")
env_rows = page.locator("#theme-env-overrides tbody tr")
disabled_inputs = page.locator(
"#theme-settings-form input.color-value-input[disabled]"
)
env_row_count = env_rows.count()
disabled_count = disabled_inputs.count()
assert disabled_count == env_row_count
color_input = page.locator(
"#theme-settings-form input[name='--color-primary']"
)
expect(color_input).to_be_visible()
expect(color_input).to_be_enabled()
original_value = color_input.input_value()
candidate_values = ("#114455", "#225566")
new_value = (
candidate_values[0]
if original_value != candidate_values[0]
else candidate_values[1]
)
color_input.fill(new_value)
page.click("#theme-settings-form button[type='submit']")
feedback = page.locator("#theme-settings-feedback")
expect(feedback).to_contain_text("updated successfully")
computed_color = page.evaluate(
"() => getComputedStyle(document.documentElement).getPropertyValue('--color-primary').trim()"
)
assert computed_color.lower() == new_value.lower()
page.reload()
expect(color_input).to_have_value(new_value)
color_input.fill(original_value)
page.click("#theme-settings-form button[type='submit']")
expect(feedback).to_contain_text("updated successfully")
page.reload()
expect(color_input).to_have_value(original_value)

0
tests/unit/__init__.py
View File

266
tests/unit/conftest.py
View File

@@ -1,266 +0,0 @@
from datetime import date
from typing import Any, Dict, Generator
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
from config.database import Base
from main import app
from models.capex import Capex
from models.consumption import Consumption
from models.equipment import Equipment
from models.maintenance import Maintenance
from models.opex import Opex
from models.parameters import Parameter
from models.production_output import ProductionOutput
from models.scenario import Scenario
from models.simulation_result import SimulationResult
SQLALCHEMY_TEST_URL = "sqlite:///:memory:"
engine = create_engine(
SQLALCHEMY_TEST_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(
autocommit=False, autoflush=False, bind=engine
)
@pytest.fixture(scope="session", autouse=True)
def setup_database() -> Generator[None, None, None]:
# Ensure all model metadata is registered before creating tables
from models import (
application_setting,
capex,
consumption,
currency,
distribution,
equipment,
maintenance,
opex,
parameters,
production_output,
role,
scenario,
simulation_result,
theme_setting,
user,
) # noqa: F401 - imported for side effects
_ = (
capex,
consumption,
currency,
distribution,
equipment,
maintenance,
application_setting,
opex,
parameters,
production_output,
role,
scenario,
simulation_result,
theme_setting,
user,
)
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
@pytest.fixture()
def db_session() -> Generator[Session, None, None]:
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
session = TestingSessionLocal()
try:
yield session
finally:
session.rollback()
session.close()
@pytest.fixture()
def api_client(db_session: Session) -> Generator[TestClient, None, None]:
def override_get_db():
try:
yield db_session
finally:
pass
from routes.dependencies import get_db
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as client:
yield client
app.dependency_overrides.pop(get_db, None)
@pytest.fixture()
def seeded_ui_data(
db_session: Session,
) -> Generator[Dict[str, Any], None, None]:
"""Populate a scenario with representative related records for UI tests."""
scenario_name = f"Scenario Alpha {uuid4()}"
scenario = Scenario(name=scenario_name, description="Seeded UI scenario")
db_session.add(scenario)
db_session.flush()
parameter = Parameter(
scenario_id=scenario.id,
name="Ore Grade",
value=1.5,
distribution_type="normal",
distribution_parameters={"mean": 1.5, "std_dev": 0.1},
)
capex = Capex(
scenario_id=scenario.id,
amount=1_000_000.0,
description="Drill purchase",
currency_code="USD",
)
opex = Opex(
scenario_id=scenario.id,
amount=250_000.0,
description="Fuel spend",
currency_code="USD",
)
consumption = Consumption(
scenario_id=scenario.id,
amount=1_200.0,
description="Diesel (L)",
unit_name="Liters",
unit_symbol="L",
)
production = ProductionOutput(
scenario_id=scenario.id,
amount=800.0,
description="Ore (tonnes)",
unit_name="Tonnes",
unit_symbol="t",
)
equipment = Equipment(
scenario_id=scenario.id,
name="Excavator 42",
description="Primary loader",
)
db_session.add_all(
[parameter, capex, opex, consumption, production, equipment]
)
db_session.flush()
maintenance = Maintenance(
scenario_id=scenario.id,
equipment_id=equipment.id,
maintenance_date=date(2025, 1, 15),
description="Hydraulic service",
cost=15_000.0,
)
simulation_results = [
SimulationResult(
scenario_id=scenario.id,
iteration=index,
result=value,
)
for index, value in enumerate(
(950_000.0, 975_000.0, 990_000.0), start=1
)
]
db_session.add(maintenance)
db_session.add_all(simulation_results)
db_session.commit()
try:
yield {
"scenario": scenario,
"equipment": equipment,
"simulation_results": simulation_results,
}
finally:
db_session.query(SimulationResult).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Maintenance).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Equipment).filter_by(id=equipment.id).delete()
db_session.query(ProductionOutput).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Consumption).filter_by(
scenario_id=scenario.id
).delete()
db_session.query(Opex).filter_by(scenario_id=scenario.id).delete()
db_session.query(Capex).filter_by(scenario_id=scenario.id).delete()
db_session.query(Parameter).filter_by(scenario_id=scenario.id).delete()
db_session.query(Scenario).filter_by(id=scenario.id).delete()
db_session.commit()
@pytest.fixture()
def invalid_request_payloads(
db_session: Session,
) -> Generator[Dict[str, Any], None, None]:
"""Provide reusable invalid request bodies for exercising validation branches."""
duplicate_name = f"Scenario Duplicate {uuid4()}"
existing = Scenario(
name=duplicate_name,
description="Existing scenario for duplicate checks",
)
db_session.add(existing)
db_session.commit()
payloads: Dict[str, Any] = {
"existing_scenario": existing,
"scenario_duplicate": {
"name": duplicate_name,
"description": "Second scenario should fail with duplicate name",
},
"parameter_missing_scenario": {
"scenario_id": existing.id + 99,
"name": "Invalid Parameter",
"value": 1.0,
},
"parameter_invalid_distribution": {
"scenario_id": existing.id,
"name": "Weird Dist",
"value": 2.5,
"distribution_type": "invalid",
},
"simulation_unknown_scenario": {
"scenario_id": existing.id + 99,
"iterations": 10,
"parameters": [
{"name": "grade", "value": 1.2, "distribution": "normal"}
],
},
"simulation_missing_parameters": {
"scenario_id": existing.id,
"iterations": 5,
"parameters": [],
},
"reporting_non_list_payload": {"result": 10.0},
"reporting_missing_result": [{"value": 12.0}],
"maintenance_negative_cost": {
"equipment_id": 1,
"scenario_id": existing.id,
"maintenance_date": "2025-01-15",
"cost": -500.0,
},
}
try:
yield payloads
finally:
db_session.query(Scenario).filter_by(id=existing.id).delete()
db_session.commit()

231
tests/unit/test_auth.py
View File

@@ -1,231 +0,0 @@
from services.security import get_password_hash, verify_password
def test_password_hashing():
password = "testpassword"
hashed_password = get_password_hash(password)
assert verify_password(password, hashed_password)
assert not verify_password("wrongpassword", hashed_password)
def test_register_user(api_client):
response = api_client.post(
"/users/register",
json={
"username": "testuser",
"email": "test@example.com",
"password": "testpassword",
},
)
assert response.status_code == 201
data = response.json()
assert data["username"] == "testuser"
assert data["email"] == "test@example.com"
assert "id" in data
assert "role_id" in data
response = api_client.post(
"/users/register",
json={
"username": "testuser",
"email": "another@example.com",
"password": "testpassword",
},
)
assert response.status_code == 400
assert response.json() == {"detail": "Username already registered"}
response = api_client.post(
"/users/register",
json={
"username": "anotheruser",
"email": "test@example.com",
"password": "testpassword",
},
)
assert response.status_code == 400
assert response.json() == {"detail": "Email already registered"}
def test_login_user(api_client):
# Register a user first
api_client.post(
"/users/register",
json={
"username": "loginuser",
"email": "login@example.com",
"password": "loginpassword",
},
)
response = api_client.post(
"/users/login",
json={"username": "loginuser", "password": "loginpassword"},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
response = api_client.post(
"/users/login",
json={"username": "loginuser", "password": "wrongpassword"},
)
assert response.status_code == 401
assert response.json() == {"detail": "Incorrect username or password"}
response = api_client.post(
"/users/login",
json={"username": "nonexistent", "password": "password"},
)
assert response.status_code == 401
assert response.json() == {"detail": "Incorrect username or password"}
def test_read_users_me(api_client):
# Register a user first
api_client.post(
"/users/register",
json={
"username": "profileuser",
"email": "profile@example.com",
"password": "profilepassword",
},
)
# Login to get a token
login_response = api_client.post(
"/users/login",
json={"username": "profileuser", "password": "profilepassword"},
)
token = login_response.json()["access_token"]
response = api_client.get(
"/users/me", headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "profileuser"
assert data["email"] == "profile@example.com"
def test_update_users_me(api_client):
# Register a user first
api_client.post(
"/users/register",
json={
"username": "updateuser",
"email": "update@example.com",
"password": "updatepassword",
},
)
# Login to get a token
login_response = api_client.post(
"/users/login",
json={"username": "updateuser", "password": "updatepassword"},
)
token = login_response.json()["access_token"]
response = api_client.put(
"/users/me",
headers={"Authorization": f"Bearer {token}"},
json={
"username": "updateduser",
"email": "updated@example.com",
"password": "newpassword",
},
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "updateduser"
assert data["email"] == "updated@example.com"
# Verify password change
response = api_client.post(
"/users/login",
json={"username": "updateduser", "password": "newpassword"},
)
assert response.status_code == 200
token = response.json()["access_token"]
# Test username already taken
api_client.post(
"/users/register",
json={
"username": "anotherupdateuser",
"email": "anotherupdate@example.com",
"password": "password",
},
)
response = api_client.put(
"/users/me",
headers={"Authorization": f"Bearer {token}"},
json={
"username": "anotherupdateuser",
},
)
assert response.status_code == 400
assert response.json() == {"detail": "Username already taken"}
# Test email already registered
api_client.post(
"/users/register",
json={
"username": "yetanotheruser",
"email": "yetanother@example.com",
"password": "password",
},
)
response = api_client.put(
"/users/me",
headers={"Authorization": f"Bearer {token}"},
json={
"email": "yetanother@example.com",
},
)
assert response.status_code == 400
assert response.json() == {"detail": "Email already registered"}
def test_forgot_password(api_client):
response = api_client.post(
"/users/forgot-password", json={"email": "nonexistent@example.com"}
)
assert response.status_code == 200
assert response.json() == {
"message": "Password reset email sent (not really)"}
def test_reset_password(api_client):
# Register a user first
api_client.post(
"/users/register",
json={
"username": "resetuser",
"email": "reset@example.com",
"password": "oldpassword",
},
)
response = api_client.post(
"/users/reset-password",
json={
"token": "resetuser", # Use username as token for test
"new_password": "newpassword",
},
)
assert response.status_code == 200
assert response.json() == {
"message": "Password has been reset successfully"}
# Verify password change
response = api_client.post(
"/users/login",
json={"username": "resetuser", "password": "newpassword"},
)
assert response.status_code == 200
response = api_client.post(
"/users/login",
json={"username": "resetuser", "password": "oldpassword"},
)
assert response.status_code == 401

77
tests/unit/test_consumption.py
View File

@@ -1,77 +0,0 @@
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client(api_client: TestClient) -> TestClient:
return api_client
def _create_scenario(client: TestClient) -> int:
payload = {
"name": f"Consumption Scenario {uuid4()}",
"description": "Scenario for consumption tests",
}
response = client.post("/api/scenarios/", json=payload)
assert response.status_code == 200
return response.json()["id"]
def test_create_consumption(client: TestClient) -> None:
scenario_id = _create_scenario(client)
payload = {
"scenario_id": scenario_id,
"amount": 125.5,
"description": "Fuel usage baseline",
"unit_name": "Liters",
"unit_symbol": "L",
}
response = client.post("/api/consumption/", json=payload)
assert response.status_code == 201
body = response.json()
assert body["id"] > 0
assert body["scenario_id"] == scenario_id
assert body["amount"] == pytest.approx(125.5)
assert body["description"] == "Fuel usage baseline"
assert body["unit_symbol"] == "L"
def test_list_consumption_returns_created_items(client: TestClient) -> None:
scenario_id = _create_scenario(client)
values = [50.0, 80.75]
for amount in values:
response = client.post(
"/api/consumption/",
json={
"scenario_id": scenario_id,
"amount": amount,
"description": f"Consumption {amount}",
"unit_name": "Tonnes",
"unit_symbol": "t",
},
)
assert response.status_code == 201
list_response = client.get("/api/consumption/")
assert list_response.status_code == 200
items = [
item
for item in list_response.json()
if item["scenario_id"] == scenario_id
]
assert {item["amount"] for item in items} == set(values)
def test_create_consumption_rejects_negative_amount(client: TestClient) -> None:
scenario_id = _create_scenario(client)
payload = {
"scenario_id": scenario_id,
"amount": -10,
"description": "Invalid negative amount",
}
response = client.post("/api/consumption/", json=payload)
assert response.status_code == 422

123
tests/unit/test_costs.py
View File

@@ -1,123 +0,0 @@
from uuid import uuid4
from fastapi.testclient import TestClient
from config.database import Base, engine
from main import app
def setup_module(module):
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
def teardown_module(module):
Base.metadata.drop_all(bind=engine)
client = TestClient(app)
def _create_scenario() -> int:
payload = {
"name": f"CostScenario-{uuid4()}",
"description": "Cost tracking test scenario",
}
response = client.post("/api/scenarios/", json=payload)
assert response.status_code == 200
return response.json()["id"]
def test_create_and_list_capex_and_opex():
sid = _create_scenario()
capex_payload = {
"scenario_id": sid,
"amount": 1000.0,
"description": "Initial capex",
"currency_code": "USD",
}
resp2 = client.post("/api/costs/capex", json=capex_payload)
assert resp2.status_code == 200
capex = resp2.json()
assert capex["scenario_id"] == sid
assert capex["amount"] == 1000.0
assert capex["currency_code"] == "USD"
resp3 = client.get("/api/costs/capex")
assert resp3.status_code == 200
data = resp3.json()
assert any(
item["amount"] == 1000.0 and item["scenario_id"] == sid for item in data
)
opex_payload = {
"scenario_id": sid,
"amount": 500.0,
"description": "Recurring opex",
"currency_code": "USD",
}
resp4 = client.post("/api/costs/opex", json=opex_payload)
assert resp4.status_code == 200
opex = resp4.json()
assert opex["scenario_id"] == sid
assert opex["amount"] == 500.0
assert opex["currency_code"] == "USD"
resp5 = client.get("/api/costs/opex")
assert resp5.status_code == 200
data_o = resp5.json()
assert any(
item["amount"] == 500.0 and item["scenario_id"] == sid
for item in data_o
)
def test_multiple_capex_entries():
sid = _create_scenario()
amounts = [250.0, 750.0]
for amount in amounts:
resp = client.post(
"/api/costs/capex",
json={
"scenario_id": sid,
"amount": amount,
"description": f"Capex {amount}",
"currency_code": "EUR",
},
)
assert resp.status_code == 200
resp = client.get("/api/costs/capex")
assert resp.status_code == 200
data = resp.json()
retrieved_amounts = [
item["amount"] for item in data if item["scenario_id"] == sid
]
for amount in amounts:
assert amount in retrieved_amounts
def test_multiple_opex_entries():
sid = _create_scenario()
amounts = [120.0, 340.0]
for amount in amounts:
resp = client.post(
"/api/costs/opex",
json={
"scenario_id": sid,
"amount": amount,
"description": f"Opex {amount}",
"currency_code": "CAD",
},
)
assert resp.status_code == 200
resp = client.get("/api/costs/opex")
assert resp.status_code == 200
data = resp.json()
retrieved_amounts = [
item["amount"] for item in data if item["scenario_id"] == sid
]
for amount in amounts:
assert amount in retrieved_amounts

125
tests/unit/test_currencies.py
View File

@@ -1,125 +0,0 @@
from typing import Dict
import pytest
from models.currency import Currency
@pytest.fixture(autouse=True)
def _cleanup_currencies(db_session):
db_session.query(Currency).delete()
db_session.commit()
yield
db_session.query(Currency).delete()
db_session.commit()
def _assert_currency(
payload: Dict[str, object],
code: str,
name: str,
symbol: str | None,
is_active: bool,
) -> None:
assert payload["code"] == code
assert payload["name"] == name
assert payload["is_active"] is is_active
if symbol is None:
assert payload["symbol"] is None
else:
assert payload["symbol"] == symbol
def test_list_returns_default_currency(api_client, db_session):
response = api_client.get("/api/currencies/")
assert response.status_code == 200
data = response.json()
assert any(item["code"] == "USD" for item in data)
def test_create_currency_success(api_client, db_session):
payload = {"code": "EUR", "name": "Euro", "symbol": "€", "is_active": True}
response = api_client.post("/api/currencies/", json=payload)
assert response.status_code == 201
data = response.json()
_assert_currency(data, "EUR", "Euro", "€", True)
stored = db_session.query(Currency).filter_by(code="EUR").one()
assert stored.name == "Euro"
assert stored.symbol == "€"
assert stored.is_active is True
def test_create_currency_conflict(api_client, db_session):
api_client.post(
"/api/currencies/",
json={
"code": "CAD",
"name": "Canadian Dollar",
"symbol": "$",
"is_active": True,
},
)
duplicate = api_client.post(
"/api/currencies/",
json={
"code": "CAD",
"name": "Canadian Dollar",
"symbol": "$",
"is_active": True,
},
)
assert duplicate.status_code == 409
def test_update_currency_fields(api_client, db_session):
api_client.post(
"/api/currencies/",
json={
"code": "GBP",
"name": "British Pound",
"symbol": "£",
"is_active": True,
},
)
response = api_client.put(
"/api/currencies/GBP",
json={"name": "Pound Sterling", "symbol": "£", "is_active": False},
)
assert response.status_code == 200
data = response.json()
_assert_currency(data, "GBP", "Pound Sterling", "£", False)
def test_toggle_currency_activation(api_client, db_session):
api_client.post(
"/api/currencies/",
json={
"code": "AUD",
"name": "Australian Dollar",
"symbol": "A$",
"is_active": True,
},
)
response = api_client.patch(
"/api/currencies/AUD/activation",
json={"is_active": False},
)
assert response.status_code == 200
data = response.json()
_assert_currency(data, "AUD", "Australian Dollar", "A$", False)
def test_default_currency_cannot_be_deactivated(api_client, db_session):
api_client.get("/api/currencies/")
response = api_client.patch(
"/api/currencies/USD/activation",
json={"is_active": False},
)
assert response.status_code == 400
assert (
response.json()["detail"]
== "The default currency cannot be deactivated."
)

75
tests/unit/test_currency_workflow.py
View File

@@ -1,75 +0,0 @@
from uuid import uuid4
import pytest
from models.currency import Currency
@pytest.fixture
def seeded_currency(db_session):
currency = Currency(code="GBP", name="British Pound", symbol="GBP")
db_session.add(currency)
db_session.commit()
db_session.refresh(currency)
try:
yield currency
finally:
db_session.delete(currency)
db_session.commit()
def _create_scenario(api_client):
payload = {
"name": f"CurrencyScenario-{uuid4()}",
"description": "Currency workflow scenario",
}
resp = api_client.post("/api/scenarios/", json=payload)
assert resp.status_code == 200
return resp.json()["id"]
def test_create_capex_with_currency_code_and_list(api_client, seeded_currency):
sid = _create_scenario(api_client)
payload = {
"scenario_id": sid,
"amount": 500.0,
"description": "Capex with GBP",
"currency_code": seeded_currency.code,
}
resp = api_client.post("/api/costs/capex", json=payload)
assert resp.status_code == 200
data = resp.json()
assert (
data.get("currency_code") == seeded_currency.code
or data.get("currency", {}).get("code") == seeded_currency.code
)
def test_create_opex_with_currency_id(api_client, seeded_currency):
sid = _create_scenario(api_client)
resp = api_client.get("/api/currencies/")
assert resp.status_code == 200
currencies = resp.json()
assert any(c["id"] == seeded_currency.id for c in currencies)
payload = {
"scenario_id": sid,
"amount": 120.0,
"description": "Opex with explicit id",
"currency_id": seeded_currency.id,
}
resp = api_client.post("/api/costs/opex", json=payload)
assert resp.status_code == 200
data = resp.json()
assert data["currency_id"] == seeded_currency.id
def test_list_currencies_endpoint(api_client, seeded_currency):
resp = api_client.get("/api/currencies/")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data, list)
assert any(c["id"] == seeded_currency.id for c in data)

71
tests/unit/test_distribution.py
View File

@@ -1,71 +0,0 @@
from uuid import uuid4
from fastapi.testclient import TestClient
from config.database import Base, engine
from main import app
def setup_module(module):
Base.metadata.create_all(bind=engine)
def teardown_module(module):
Base.metadata.drop_all(bind=engine)
client = TestClient(app)
def test_create_and_list_distribution():
dist_name = f"NormalDist-{uuid4()}"
payload = {
"name": dist_name,
"distribution_type": "normal",
"parameters": {"mu": 0, "sigma": 1},
}
resp = client.post("/api/distributions/", json=payload)
assert resp.status_code == 200
data = resp.json()
assert data["name"] == dist_name
resp2 = client.get("/api/distributions/")
assert resp2.status_code == 200
data2 = resp2.json()
assert any(d["name"] == dist_name for d in data2)
def test_duplicate_distribution_name_allowed():
dist_name = f"DupDist-{uuid4()}"
payload = {
"name": dist_name,
"distribution_type": "uniform",
"parameters": {"min": 0, "max": 1},
}
first = client.post("/api/distributions/", json=payload)
assert first.status_code == 200
duplicate = client.post("/api/distributions/", json=payload)
assert duplicate.status_code == 200
resp = client.get("/api/distributions/")
assert resp.status_code == 200
matching = [item for item in resp.json() if item["name"] == dist_name]
assert len(matching) >= 2
def test_list_distributions_returns_all():
names = {f"ListDist-{uuid4()}" for _ in range(2)}
for name in names:
payload = {
"name": name,
"distribution_type": "triangular",
"parameters": {"min": 0, "max": 10, "mode": 5},
}
resp = client.post("/api/distributions/", json=payload)
assert resp.status_code == 200
resp = client.get("/api/distributions/")
assert resp.status_code == 200
found_names = {item["name"] for item in resp.json()}
assert names.issubset(found_names)

77
tests/unit/test_equipment.py
View File

@@ -1,77 +0,0 @@
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client(api_client: TestClient) -> TestClient:
return api_client
def _create_scenario(client: TestClient) -> int:
payload = {
"name": f"Equipment Scenario {uuid4()}",
"description": "Scenario for equipment tests",
}
response = client.post("/api/scenarios/", json=payload)
assert response.status_code == 200
return response.json()["id"]
def test_create_equipment(client: TestClient) -> None:
scenario_id = _create_scenario(client)
payload = {
"scenario_id": scenario_id,
"name": "Excavator",
"description": "Heavy machinery",
}
response = client.post("/api/equipment/", json=payload)
assert response.status_code == 200
created = response.json()
assert created["id"] > 0
assert created["scenario_id"] == scenario_id
assert created["name"] == "Excavator"
assert created["description"] == "Heavy machinery"
def test_list_equipment_filters_by_scenario(client: TestClient) -> None:
target_scenario = _create_scenario(client)
other_scenario = _create_scenario(client)
for scenario_id, name in [
(target_scenario, "Bulldozer"),
(target_scenario, "Loader"),
(other_scenario, "Conveyor"),
]:
response = client.post(
"/api/equipment/",
json={
"scenario_id": scenario_id,
"name": name,
"description": f"Equipment {name}",
},
)
assert response.status_code == 200
list_response = client.get("/api/equipment/")
assert list_response.status_code == 200
items = [
item
for item in list_response.json()
if item["scenario_id"] == target_scenario
]
assert {item["name"] for item in items} == {"Bulldozer", "Loader"}
def test_create_equipment_requires_name(client: TestClient) -> None:
scenario_id = _create_scenario(client)
response = client.post(
"/api/equipment/",
json={
"scenario_id": scenario_id,
"description": "Missing name",
},
)
assert response.status_code == 422

125
tests/unit/test_maintenance.py
View File

@@ -1,125 +0,0 @@
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client(api_client: TestClient) -> TestClient:
return api_client
def _create_scenario_and_equipment(client: TestClient):
scenario_payload = {
"name": f"Test Scenario {uuid4()}",
"description": "Scenario for maintenance tests",
}
scenario_response = client.post("/api/scenarios/", json=scenario_payload)
assert scenario_response.status_code == 200
scenario_id = scenario_response.json()["id"]
equipment_payload = {
"scenario_id": scenario_id,
"name": f"Test Equipment {uuid4()}",
"description": "Equipment linked to maintenance",
}
equipment_response = client.post("/api/equipment/", json=equipment_payload)
assert equipment_response.status_code == 200
equipment_id = equipment_response.json()["id"]
return scenario_id, equipment_id
def _create_maintenance_payload(
equipment_id: int, scenario_id: int, description: str
):
return {
"equipment_id": equipment_id,
"scenario_id": scenario_id,
"maintenance_date": "2025-10-20",
"description": description,
"cost": 100.0,
}
def test_create_and_list_maintenance(client: TestClient):
scenario_id, equipment_id = _create_scenario_and_equipment(client)
payload = _create_maintenance_payload(
equipment_id, scenario_id, "Create maintenance"
)
response = client.post("/api/maintenance/", json=payload)
assert response.status_code == 201
created = response.json()
assert created["equipment_id"] == equipment_id
assert created["scenario_id"] == scenario_id
assert created["description"] == "Create maintenance"
list_response = client.get("/api/maintenance/")
assert list_response.status_code == 200
items = list_response.json()
assert any(item["id"] == created["id"] for item in items)
def test_get_maintenance(client: TestClient):
scenario_id, equipment_id = _create_scenario_and_equipment(client)
payload = _create_maintenance_payload(
equipment_id, scenario_id, "Retrieve maintenance"
)
create_response = client.post("/api/maintenance/", json=payload)
assert create_response.status_code == 201
maintenance_id = create_response.json()["id"]
response = client.get(f"/api/maintenance/{maintenance_id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == maintenance_id
assert data["equipment_id"] == equipment_id
assert data["description"] == "Retrieve maintenance"
def test_update_maintenance(client: TestClient):
scenario_id, equipment_id = _create_scenario_and_equipment(client)
create_response = client.post(
"/api/maintenance/",
json=_create_maintenance_payload(
equipment_id, scenario_id, "Maintenance before update"
),
)
assert create_response.status_code == 201
maintenance_id = create_response.json()["id"]
update_payload = {
"equipment_id": equipment_id,
"scenario_id": scenario_id,
"maintenance_date": "2025-11-01",
"description": "Maintenance after update",
"cost": 250.0,
}
response = client.put(
f"/api/maintenance/{maintenance_id}", json=update_payload
)
assert response.status_code == 200
updated = response.json()
assert updated["maintenance_date"] == "2025-11-01"
assert updated["description"] == "Maintenance after update"
assert updated["cost"] == 250.0
def test_delete_maintenance(client: TestClient):
scenario_id, equipment_id = _create_scenario_and_equipment(client)
create_response = client.post(
"/api/maintenance/",
json=_create_maintenance_payload(
equipment_id, scenario_id, "Delete maintenance"
),
)
assert create_response.status_code == 201
maintenance_id = create_response.json()["id"]
delete_response = client.delete(f"/api/maintenance/{maintenance_id}")
assert delete_response.status_code == 204
get_response = client.get(f"/api/maintenance/{maintenance_id}")
assert get_response.status_code == 404

126
tests/unit/test_parameters.py
View File

@@ -1,126 +0,0 @@
from typing import Any, Dict, List
from uuid import uuid4
from fastapi.testclient import TestClient
from config.database import Base, engine
from main import app
def setup_module(module: object) -> None:
Base.metadata.create_all(bind=engine)
def teardown_module(module: object) -> None:
Base.metadata.drop_all(bind=engine)
def _create_scenario(name: str | None = None) -> int:
payload: Dict[str, Any] = {
"name": name or f"ParamScenario-{uuid4()}",
"description": "Parameter test scenario",
}
response = TestClient(app).post("/api/scenarios/", json=payload)
assert response.status_code == 200
return response.json()["id"]
def _create_distribution() -> int:
payload: Dict[str, Any] = {
"name": f"NormalDist-{uuid4()}",
"distribution_type": "normal",
"parameters": {"mu": 10, "sigma": 2},
}
response = TestClient(app).post("/api/distributions/", json=payload)
assert response.status_code == 200
return response.json()["id"]
client = TestClient(app)
def test_create_and_list_parameter():
scenario_id = _create_scenario()
distribution_id = _create_distribution()
parameter_payload: Dict[str, Any] = {
"scenario_id": scenario_id,
"name": f"param-{uuid4()}",
"value": 3.14,
"distribution_id": distribution_id,
}
create_response = client.post("/api/parameters/", json=parameter_payload)
assert create_response.status_code == 200
created = create_response.json()
assert created["scenario_id"] == scenario_id
assert created["name"] == parameter_payload["name"]
assert created["value"] == parameter_payload["value"]
assert created["distribution_id"] == distribution_id
assert created["distribution_type"] == "normal"
assert created["distribution_parameters"] == {"mu": 10, "sigma": 2}
list_response = client.get("/api/parameters/")
assert list_response.status_code == 200
params = list_response.json()
assert any(p["id"] == created["id"] for p in params)
def test_create_parameter_for_missing_scenario():
payload: Dict[str, Any] = {
"scenario_id": 0,
"name": "invalid",
"value": 1.0,
}
response = client.post("/api/parameters/", json=payload)
assert response.status_code == 404
assert response.json()["detail"] == "Scenario not found"
def test_multiple_parameters_listed():
scenario_id = _create_scenario()
payloads: List[Dict[str, Any]] = [
{"scenario_id": scenario_id, "name": f"alpha-{i}", "value": float(i)}
for i in range(2)
]
for payload in payloads:
resp = client.post("/api/parameters/", json=payload)
assert resp.status_code == 200
list_response = client.get("/api/parameters/")
assert list_response.status_code == 200
names = {item["name"] for item in list_response.json()}
for payload in payloads:
assert payload["name"] in names
def test_parameter_inline_distribution_metadata():
scenario_id = _create_scenario()
payload: Dict[str, Any] = {
"scenario_id": scenario_id,
"name": "inline-param",
"value": 7.5,
"distribution_type": "uniform",
"distribution_parameters": {"min": 5, "max": 10},
}
response = client.post("/api/parameters/", json=payload)
assert response.status_code == 200
created = response.json()
assert created["distribution_id"] is None
assert created["distribution_type"] == "uniform"
assert created["distribution_parameters"] == {"min": 5, "max": 10}
def test_parameter_with_missing_distribution_reference():
scenario_id = _create_scenario()
payload: Dict[str, Any] = {
"scenario_id": scenario_id,
"name": "missing-dist",
"value": 1.0,
"distribution_id": 9999,
}
response = client.post("/api/parameters/", json=payload)
assert response.status_code == 404
assert response.json()["detail"] == "Distribution not found"

82
tests/unit/test_production.py
View File

@@ -1,82 +0,0 @@
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client(api_client: TestClient) -> TestClient:
return api_client
def _create_scenario(client: TestClient) -> int:
payload = {
"name": f"Production Scenario {uuid4()}",
"description": "Scenario for production tests",
}
response = client.post("/api/scenarios/", json=payload)
assert response.status_code == 200
return response.json()["id"]
def test_create_production_record(client: TestClient) -> None:
scenario_id = _create_scenario(client)
payload: dict[str, any] = {
"scenario_id": scenario_id,
"amount": 475.25,
"description": "Daily output",
"unit_name": "Tonnes",
"unit_symbol": "t",
}
response = client.post("/api/production/", json=payload)
assert response.status_code == 201
created = response.json()
assert created["scenario_id"] == scenario_id
assert created["amount"] == pytest.approx(475.25)
assert created["description"] == "Daily output"
assert created["unit_symbol"] == "t"
def test_list_production_filters_by_scenario(client: TestClient) -> None:
target_scenario = _create_scenario(client)
other_scenario = _create_scenario(client)
for scenario_id, amount in [
(target_scenario, 100.0),
(target_scenario, 150.0),
(other_scenario, 200.0),
]:
response = client.post(
"/api/production/",
json={
"scenario_id": scenario_id,
"amount": amount,
"description": f"Output {amount}",
"unit_name": "Kilograms",
"unit_symbol": "kg",
},
)
assert response.status_code == 201
list_response = client.get("/api/production/")
assert list_response.status_code == 200
items = [
item
for item in list_response.json()
if item["scenario_id"] == target_scenario
]
assert {item["amount"] for item in items} == {100.0, 150.0}
def test_create_production_rejects_negative_amount(client: TestClient) -> None:
scenario_id = _create_scenario(client)
response = client.post(
"/api/production/",
json={
"scenario_id": scenario_id,
"amount": -5,
"description": "Invalid output",
},
)
assert response.status_code == 422

123
tests/unit/test_reporting.py
View File

@@ -1,123 +0,0 @@
import math
from typing import Any, Dict, List
import pytest
from fastapi.testclient import TestClient
from services.reporting import generate_report
def test_generate_report_empty():
report = generate_report([])
assert report == {
"count": 0,
"mean": 0.0,
"median": 0.0,
"min": 0.0,
"max": 0.0,
"std_dev": 0.0,
"variance": 0.0,
"percentile_10": 0.0,
"percentile_90": 0.0,
"percentile_5": 0.0,
"percentile_95": 0.0,
"value_at_risk_95": 0.0,
"expected_shortfall_95": 0.0,
}
def test_generate_report_with_values():
values: List[Dict[str, float]] = [
{"iteration": 1, "result": 10.0},
{"iteration": 2, "result": 20.0},
{"iteration": 3, "result": 30.0},
]
report = generate_report(values)
assert report["count"] == 3
assert math.isclose(float(report["mean"]), 20.0)
assert math.isclose(float(report["median"]), 20.0)
assert math.isclose(float(report["min"]), 10.0)
assert math.isclose(float(report["max"]), 30.0)
assert math.isclose(float(report["std_dev"]), 8.1649658, rel_tol=1e-6)
assert math.isclose(float(report["variance"]), 66.6666666, rel_tol=1e-6)
assert math.isclose(float(report["percentile_10"]), 12.0)
assert math.isclose(float(report["percentile_90"]), 28.0)
assert math.isclose(float(report["percentile_5"]), 11.0)
assert math.isclose(float(report["percentile_95"]), 29.0)
assert math.isclose(float(report["value_at_risk_95"]), 11.0)
assert math.isclose(float(report["expected_shortfall_95"]), 10.0)
def test_generate_report_single_value():
report = generate_report(
[
{"iteration": 1, "result": 42.0},
]
)
assert report["count"] == 1
assert report["std_dev"] == 0.0
assert report["variance"] == 0.0
assert report["percentile_10"] == 42.0
assert report["expected_shortfall_95"] == 42.0
def test_generate_report_ignores_invalid_entries():
raw_values: List[Any] = [
{"iteration": 1, "result": 10.0},
"not-a-mapping",
{"iteration": 2},
{"iteration": 3, "result": None},
{"iteration": 4, "result": 20},
]
report = generate_report(raw_values)
assert report["count"] == 2
assert math.isclose(float(report["mean"]), 15.0)
assert math.isclose(float(report["min"]), 10.0)
assert math.isclose(float(report["max"]), 20.0)
@pytest.fixture
def client(api_client: TestClient) -> TestClient:
return api_client
def test_reporting_endpoint_invalid_input(client: TestClient):
resp = client.post("/api/reporting/summary", json={})
assert resp.status_code == 400
assert resp.json()["detail"] == "Invalid input format"
def test_reporting_endpoint_success(client: TestClient):
input_data: List[Dict[str, float]] = [
{"iteration": 1, "result": 10.0},
{"iteration": 2, "result": 20.0},
{"iteration": 3, "result": 30.0},
]
resp = client.post("/api/reporting/summary", json=input_data)
assert resp.status_code == 200
data: Dict[str, Any] = resp.json()
assert data["count"] == 3
assert math.isclose(float(data["mean"]), 20.0)
assert math.isclose(float(data["variance"]), 66.6666666, rel_tol=1e-6)
assert math.isclose(float(data["value_at_risk_95"]), 11.0)
assert math.isclose(float(data["expected_shortfall_95"]), 10.0)
validation_error_cases: List[tuple[List[Any], str]] = [
(["not-a-dict"], "Entry at index 0 must be an object"),
([{"iteration": 1}], "Entry at index 0 must include numeric 'result'"),
(
[{"iteration": 1, "result": "bad"}],
"Entry at index 0 must include numeric 'result'",
),
]
@pytest.mark.parametrize("payload,expected_detail", validation_error_cases)
def test_reporting_endpoint_validation_errors(
client: TestClient, payload: List[Any], expected_detail: str
):
resp = client.post("/api/reporting/summary", json=payload)
assert resp.status_code == 400
assert resp.json()["detail"] == expected_detail

94
tests/unit/test_router_validation.py
View File

@@ -1,94 +0,0 @@
from typing import Any, Dict
import pytest
from fastapi.testclient import TestClient
@pytest.mark.usefixtures("invalid_request_payloads")
def test_duplicate_scenario_returns_400(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["scenario_duplicate"]
response = api_client.post("/api/scenarios/", json=payload)
assert response.status_code == 400
body = response.json()
assert body["detail"] == "Scenario already exists"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_parameter_create_missing_scenario_returns_404(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["parameter_missing_scenario"]
response = api_client.post("/api/parameters/", json=payload)
assert response.status_code == 404
assert response.json()["detail"] == "Scenario not found"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_parameter_create_invalid_distribution_is_422(
api_client: TestClient,
) -> None:
response = api_client.post(
"/api/parameters/",
json={
"scenario_id": 1,
"name": "Bad Dist",
"value": 2.0,
"distribution_type": "invalid",
},
)
assert response.status_code == 422
errors = response.json()["detail"]
assert any("distribution_type" in err["loc"] for err in errors)
@pytest.mark.usefixtures("invalid_request_payloads")
def test_simulation_unknown_scenario_returns_404(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["simulation_unknown_scenario"]
response = api_client.post("/api/simulations/run", json=payload)
assert response.status_code == 404
assert response.json()["detail"] == "Scenario not found"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_simulation_missing_parameters_returns_400(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["simulation_missing_parameters"]
response = api_client.post("/api/simulations/run", json=payload)
assert response.status_code == 400
assert response.json()["detail"] == "No parameters provided"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_reporting_summary_rejects_non_list_payload(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["reporting_non_list_payload"]
response = api_client.post("/api/reporting/summary", json=payload)
assert response.status_code == 400
assert response.json()["detail"] == "Invalid input format"
@pytest.mark.usefixtures("invalid_request_payloads")
def test_reporting_summary_requires_result_field(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["reporting_missing_result"]
response = api_client.post("/api/reporting/summary", json=payload)
assert response.status_code == 400
assert "must include numeric 'result'" in response.json()["detail"]
@pytest.mark.usefixtures("invalid_request_payloads")
def test_maintenance_negative_cost_rejected_by_schema(
api_client: TestClient, invalid_request_payloads: Dict[str, Any]
) -> None:
payload = invalid_request_payloads["maintenance_negative_cost"]
response = api_client.post("/api/maintenance/", json=payload)
assert response.status_code == 422
error_locations = [tuple(item["loc"]) for item in response.json()["detail"]]
assert ("body", "cost") in error_locations

45
tests/unit/test_scenario.py
View File

@@ -1,45 +0,0 @@
from uuid import uuid4
from fastapi.testclient import TestClient
from config.database import Base, engine
from main import app
def setup_module(module):
Base.metadata.create_all(bind=engine)
def teardown_module(module):
Base.metadata.drop_all(bind=engine)
client = TestClient(app)
def test_create_and_list_scenario():
scenario_name = f"Scenario-{uuid4()}"
response = client.post(
"/api/scenarios/",
json={"name": scenario_name, "description": "Integration test"},
)
assert response.status_code == 200
data = response.json()
assert data["name"] == scenario_name
response2 = client.get("/api/scenarios/")
assert response2.status_code == 200
data2 = response2.json()
assert any(s["name"] == scenario_name for s in data2)
def test_create_duplicate_scenario_rejected():
scenario_name = f"Duplicate-{uuid4()}"
payload = {"name": scenario_name, "description": "Primary"}
first_resp = client.post("/api/scenarios/", json=payload)
assert first_resp.status_code == 200
second_resp = client.post("/api/scenarios/", json=payload)
assert second_resp.status_code == 400
assert second_resp.json()["detail"] == "Scenario already exists"

46
tests/unit/test_seed_data.py
View File

@@ -1,46 +0,0 @@
import argparse
from unittest import mock
import scripts.seed_data as seed_data
from scripts.seed_data import DatabaseConfig
def test_run_with_namespace_handles_missing_theme_flag_without_actions() -> None:
args = argparse.Namespace(currencies=False, units=False, defaults=False)
config = mock.create_autospec(DatabaseConfig)
config.application_dsn.return_value = "postgresql://example"
with (
mock.patch("scripts.seed_data._configure_logging") as configure_logging,
mock.patch("scripts.seed_data.psycopg2.connect") as connect_mock,
mock.patch.object(seed_data.logger, "info") as info_mock,
):
seed_data.run_with_namespace(args, config=config)
configure_logging.assert_called_once()
connect_mock.assert_not_called()
info_mock.assert_called_with("No seeding options provided; exiting")
def test_run_with_namespace_seeds_defaults_without_theme_flag() -> None:
args = argparse.Namespace(
currencies=False, units=False, defaults=True, dry_run=False)
config = mock.create_autospec(DatabaseConfig)
config.application_dsn.return_value = "postgresql://example"
connection_mock = mock.MagicMock()
cursor_context = mock.MagicMock()
cursor_mock = mock.MagicMock()
connection_mock.__enter__.return_value = connection_mock
connection_mock.cursor.return_value = cursor_context
cursor_context.__enter__.return_value = cursor_mock
with (
mock.patch("scripts.seed_data._configure_logging"),
mock.patch("scripts.seed_data.psycopg2.connect", return_value=connection_mock) as connect_mock,
mock.patch("scripts.seed_data._seed_defaults") as seed_defaults,
):
seed_data.run_with_namespace(args, config=config)
connect_mock.assert_called_once_with(config.application_dsn())
seed_defaults.assert_called_once_with(cursor_mock, dry_run=False)

53
tests/unit/test_settings_routes.py
View File

@@ -1,53 +0,0 @@
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from services import settings as settings_service
@pytest.mark.usefixtures("db_session")
def test_read_css_settings_reflects_env_overrides(
api_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
env_var = settings_service.css_key_to_env_var("--color-background")
monkeypatch.setenv(env_var, "#123456")
response = api_client.get("/api/settings/css")
assert response.status_code == 200
body = response.json()
assert body["variables"]["--color-background"] == "#123456"
assert body["env_overrides"]["--color-background"] == "#123456"
assert any(
source["env_var"] == env_var and source["value"] == "#123456"
for source in body["env_sources"]
)
@pytest.mark.usefixtures("db_session")
def test_update_css_settings_persists_changes(
api_client: TestClient, db_session: Session
) -> None:
payload = {"variables": {"--color-primary": "#112233"}}
response = api_client.put("/api/settings/css", json=payload)
assert response.status_code == 200
body = response.json()
assert body["variables"]["--color-primary"] == "#112233"
persisted = settings_service.get_css_color_settings(db_session)
assert persisted["--color-primary"] == "#112233"
@pytest.mark.usefixtures("db_session")
def test_update_css_settings_invalid_value_returns_422(
api_client: TestClient,
) -> None:
response = api_client.put(
"/api/settings/css",
json={"variables": {"--color-primary": "not-a-color"}},
)
assert response.status_code == 422
body = response.json()
assert "color" in body["detail"].lower()

149
tests/unit/test_settings_service.py
View File

@@ -1,149 +0,0 @@
from types import SimpleNamespace
from typing import Dict
import pytest
from sqlalchemy.orm import Session
from models.application_setting import ApplicationSetting
from services import settings as settings_service
from services.settings import CSS_COLOR_DEFAULTS
@pytest.fixture(name="clean_env")
def fixture_clean_env(monkeypatch: pytest.MonkeyPatch) -> Dict[str, str]:
"""Provide an isolated environment mapping for tests."""
env: Dict[str, str] = {}
monkeypatch.setattr(settings_service, "os", SimpleNamespace(environ=env))
return env
def test_css_key_to_env_var_formatting():
assert (
settings_service.css_key_to_env_var("--color-background")
== "CALMINER_THEME_COLOR_BACKGROUND"
)
assert (
settings_service.css_key_to_env_var("--color-primary-stronger")
== "CALMINER_THEME_COLOR_PRIMARY_STRONGER"
)
@pytest.mark.parametrize(
"env_key,env_value",
[
("--color-background", "#ffffff"),
("--color-primary", "rgb(10, 20, 30)"),
("--color-accent", "rgba(1,2,3,0.5)"),
("--color-text-secondary", "hsla(210, 40%, 40%, 1)"),
],
)
def test_read_css_color_env_overrides_valid_values(
clean_env, env_key, env_value
):
env_var = settings_service.css_key_to_env_var(env_key)
clean_env[env_var] = env_value
overrides = settings_service.read_css_color_env_overrides(clean_env)
assert overrides[env_key] == env_value
@pytest.mark.parametrize(
"invalid_value",
[
"", # empty
"not-a-color", # arbitrary string
"#12", # short hex
"rgb(1,2)", # malformed rgb
],
)
def test_read_css_color_env_overrides_invalid_values_raise(
clean_env, invalid_value
):
env_var = settings_service.css_key_to_env_var("--color-background")
clean_env[env_var] = invalid_value
with pytest.raises(ValueError):
settings_service.read_css_color_env_overrides(clean_env)
def test_read_css_color_env_overrides_ignores_missing(clean_env):
overrides = settings_service.read_css_color_env_overrides(clean_env)
assert overrides == {}
def test_list_css_env_override_rows_returns_structured_data(clean_env):
clean_env[settings_service.css_key_to_env_var("--color-primary")] = (
"#123456"
)
rows = settings_service.list_css_env_override_rows(clean_env)
assert rows == [
{
"css_key": "--color-primary",
"env_var": settings_service.css_key_to_env_var("--color-primary"),
"value": "#123456",
}
]
def test_normalize_color_value_strips_and_validates():
assert settings_service._normalize_color_value(" #abcdef ") == "#abcdef"
with pytest.raises(ValueError):
settings_service._normalize_color_value(123) # type: ignore[arg-type]
with pytest.raises(ValueError):
settings_service._normalize_color_value(" ")
with pytest.raises(ValueError):
settings_service._normalize_color_value("#12")
def test_ensure_css_color_settings_creates_defaults(db_session: Session):
settings_service.ensure_css_color_settings(db_session)
stored = {
record.key: record.value
for record in db_session.query(ApplicationSetting).all()
}
assert set(stored.keys()) == set(CSS_COLOR_DEFAULTS.keys())
assert stored == CSS_COLOR_DEFAULTS
def test_update_css_color_settings_persists_changes(db_session: Session):
settings_service.ensure_css_color_settings(db_session)
updated = settings_service.update_css_color_settings(
db_session,
{"--color-background": "#000000", "--color-accent": "#abcdef"},
)
assert updated["--color-background"] == "#000000"
assert updated["--color-accent"] == "#abcdef"
stored = {
record.key: record.value
for record in db_session.query(ApplicationSetting).all()
}
assert stored["--color-background"] == "#000000"
assert stored["--color-accent"] == "#abcdef"
def test_get_css_color_settings_respects_env_overrides(
db_session: Session, clean_env: Dict[str, str]
):
settings_service.ensure_css_color_settings(db_session)
override_value = "#112233"
clean_env[settings_service.css_key_to_env_var("--color-background")] = (
override_value
)
values = settings_service.get_css_color_settings(db_session)
assert values["--color-background"] == override_value
db_value = (
db_session.query(ApplicationSetting)
.filter_by(key="--color-background")
.one()
.value
)
assert db_value != override_value

547
tests/unit/test_setup_database.py
View File

@@ -1,547 +0,0 @@
import argparse
from unittest import mock
import psycopg2
import pytest
from psycopg2 import errors as psycopg_errors
import scripts.setup_database as setup_db_module
from scripts import seed_data
from scripts.setup_database import DatabaseConfig, DatabaseSetup
@pytest.fixture()
def mock_config() -> DatabaseConfig:
return DatabaseConfig(
driver="postgresql",
host="localhost",
port=5432,
database="calminer_test",
user="calminer",
password="secret",
schema="public",
admin_user="postgres",
admin_password="secret",
)
@pytest.fixture()
def setup_instance(mock_config: DatabaseConfig) -> DatabaseSetup:
return DatabaseSetup(mock_config, dry_run=True)
def test_seed_baseline_data_dry_run_skips_verification(
setup_instance: DatabaseSetup,
) -> None:
with (
mock.patch("scripts.seed_data.run_with_namespace") as seed_run,
mock.patch.object(setup_instance, "_verify_seeded_data") as verify_mock,
):
setup_instance.seed_baseline_data(dry_run=True)
seed_run.assert_called_once()
namespace_arg = seed_run.call_args[0][0]
assert isinstance(namespace_arg, argparse.Namespace)
assert namespace_arg.dry_run is True
assert namespace_arg.currencies is True
assert namespace_arg.units is True
assert namespace_arg.theme is True
assert seed_run.call_args.kwargs["config"] is setup_instance.config
verify_mock.assert_not_called()
def test_seed_baseline_data_invokes_verification(
setup_instance: DatabaseSetup,
) -> None:
expected_currencies = {code for code, *_ in seed_data.CURRENCY_SEEDS}
expected_units = {code for code, *_ in seed_data.MEASUREMENT_UNIT_SEEDS}
with (
mock.patch("scripts.seed_data.run_with_namespace") as seed_run,
mock.patch.object(setup_instance, "_verify_seeded_data") as verify_mock,
):
setup_instance.seed_baseline_data(dry_run=False)
seed_run.assert_called_once()
namespace_arg = seed_run.call_args[0][0]
assert isinstance(namespace_arg, argparse.Namespace)
assert namespace_arg.dry_run is False
assert seed_run.call_args.kwargs["config"] is setup_instance.config
assert namespace_arg.theme is True
verify_mock.assert_called_once_with(
expected_currency_codes=expected_currencies,
expected_unit_codes=expected_units,
)
def test_run_migrations_applies_baseline_when_missing(
mock_config: DatabaseConfig, tmp_path
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
baseline = tmp_path / "000_base.sql"
baseline.write_text("SELECT 1;", encoding="utf-8")
other_migration = tmp_path / "20251022_add_other.sql"
other_migration.write_text("SELECT 2;", encoding="utf-8")
migration_calls: list[str] = []
def capture_migration(cursor, schema_name: str, path):
migration_calls.append(path.name)
return path.name
connection_mock = mock.MagicMock()
connection_mock.__enter__.return_value = connection_mock
cursor_context = mock.MagicMock()
cursor_mock = mock.MagicMock()
cursor_context.__enter__.return_value = cursor_mock
connection_mock.cursor.return_value = cursor_context
with (
mock.patch.object(
setup_instance,
"_application_connection",
return_value=connection_mock,
),
mock.patch.object(
setup_instance, "_migrations_table_exists", return_value=True
),
mock.patch.object(
setup_instance, "_fetch_applied_migrations", return_value=set()
),
mock.patch.object(
setup_instance,
"_apply_migration_file",
side_effect=capture_migration,
) as apply_mock,
):
setup_instance.run_migrations(tmp_path)
assert apply_mock.call_count == 1
assert migration_calls == ["000_base.sql"]
legacy_marked = any(
call.args[1] == ("20251022_add_other.sql",)
for call in cursor_mock.execute.call_args_list
if len(call.args) == 2
)
assert legacy_marked
def test_run_migrations_noop_when_all_files_already_applied(
mock_config: DatabaseConfig, tmp_path
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
baseline = tmp_path / "000_base.sql"
baseline.write_text("SELECT 1;", encoding="utf-8")
other_migration = tmp_path / "20251022_add_other.sql"
other_migration.write_text("SELECT 2;", encoding="utf-8")
connection_mock, cursor_mock = _connection_with_cursor()
with (
mock.patch.object(
setup_instance,
"_application_connection",
return_value=connection_mock,
),
mock.patch.object(
setup_instance, "_migrations_table_exists", return_value=True
),
mock.patch.object(
setup_instance,
"_fetch_applied_migrations",
return_value={"000_base.sql", "20251022_add_other.sql"},
),
mock.patch.object(
setup_instance, "_apply_migration_file"
) as apply_mock,
):
setup_instance.run_migrations(tmp_path)
apply_mock.assert_not_called()
cursor_mock.execute.assert_not_called()
def _connection_with_cursor() -> tuple[mock.MagicMock, mock.MagicMock]:
connection_mock = mock.MagicMock()
connection_mock.__enter__.return_value = connection_mock
cursor_context = mock.MagicMock()
cursor_mock = mock.MagicMock()
cursor_context.__enter__.return_value = cursor_mock
connection_mock.cursor.return_value = cursor_context
return connection_mock, cursor_mock
def test_verify_seeded_data_raises_when_currency_missing(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
connection_mock, cursor_mock = _connection_with_cursor()
cursor_mock.fetchall.return_value = [("USD", True)]
with mock.patch.object(
setup_instance, "_application_connection", return_value=connection_mock
):
with pytest.raises(RuntimeError) as exc:
setup_instance._verify_seeded_data(
expected_currency_codes={"USD", "EUR"},
expected_unit_codes=set(),
)
assert "EUR" in str(exc.value)
def test_verify_seeded_data_raises_when_default_currency_inactive(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
connection_mock, cursor_mock = _connection_with_cursor()
cursor_mock.fetchall.return_value = [("USD", False)]
with mock.patch.object(
setup_instance, "_application_connection", return_value=connection_mock
):
with pytest.raises(RuntimeError) as exc:
setup_instance._verify_seeded_data(
expected_currency_codes={"USD"},
expected_unit_codes=set(),
)
assert "inactive" in str(exc.value)
def test_verify_seeded_data_raises_when_units_missing(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
connection_mock, cursor_mock = _connection_with_cursor()
cursor_mock.fetchall.return_value = [("tonnes", True)]
with mock.patch.object(
setup_instance, "_application_connection", return_value=connection_mock
):
with pytest.raises(RuntimeError) as exc:
setup_instance._verify_seeded_data(
expected_currency_codes=set(),
expected_unit_codes={"tonnes", "liters"},
)
assert "liters" in str(exc.value)
def test_verify_seeded_data_raises_when_measurement_table_missing(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
connection_mock, cursor_mock = _connection_with_cursor()
cursor_mock.execute.side_effect = psycopg_errors.UndefinedTable(
"relation does not exist"
)
with mock.patch.object(
setup_instance, "_application_connection", return_value=connection_mock
):
with pytest.raises(RuntimeError) as exc:
setup_instance._verify_seeded_data(
expected_currency_codes=set(),
expected_unit_codes={"tonnes"},
)
assert "measurement_unit" in str(exc.value)
connection_mock.rollback.assert_called_once()
def test_seed_baseline_data_rerun_uses_existing_records(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
connection_mock, cursor_mock = _connection_with_cursor()
currency_rows = [(code, True) for code, *_ in seed_data.CURRENCY_SEEDS]
unit_rows = [(code, True) for code, *_ in seed_data.MEASUREMENT_UNIT_SEEDS]
cursor_mock.fetchall.side_effect = [
currency_rows,
unit_rows,
currency_rows,
unit_rows,
]
with (
mock.patch.object(
setup_instance,
"_application_connection",
return_value=connection_mock,
),
mock.patch("scripts.seed_data.run_with_namespace") as seed_run,
):
setup_instance.seed_baseline_data(dry_run=False)
setup_instance.seed_baseline_data(dry_run=False)
assert seed_run.call_count == 2
first_namespace = seed_run.call_args_list[0].args[0]
assert isinstance(first_namespace, argparse.Namespace)
assert first_namespace.dry_run is False
assert seed_run.call_args_list[0].kwargs["config"] is setup_instance.config
assert cursor_mock.execute.call_count == 4
def test_ensure_database_raises_with_context(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
connection_mock = mock.MagicMock()
cursor_mock = mock.MagicMock()
cursor_mock.fetchone.return_value = None
cursor_mock.execute.side_effect = [None, psycopg2.Error("create_fail")]
connection_mock.cursor.return_value = cursor_mock
with mock.patch.object(
setup_instance, "_admin_connection", return_value=connection_mock
):
with pytest.raises(RuntimeError) as exc:
setup_instance.ensure_database()
assert "Failed to create database" in str(exc.value)
def test_ensure_role_raises_with_context_during_creation(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
admin_conn, admin_cursor = _connection_with_cursor()
admin_cursor.fetchone.return_value = None
admin_cursor.execute.side_effect = [None, psycopg2.Error("role_fail")]
with mock.patch.object(
setup_instance,
"_admin_connection",
side_effect=[admin_conn],
):
with pytest.raises(RuntimeError) as exc:
setup_instance.ensure_role()
assert "Failed to create role" in str(exc.value)
def test_ensure_role_raises_with_context_during_privilege_grants(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
admin_conn, admin_cursor = _connection_with_cursor()
admin_cursor.fetchone.return_value = (1,)
privilege_conn, privilege_cursor = _connection_with_cursor()
privilege_cursor.execute.side_effect = [psycopg2.Error("grant_fail")]
with mock.patch.object(
setup_instance,
"_admin_connection",
side_effect=[admin_conn, privilege_conn],
):
with pytest.raises(RuntimeError) as exc:
setup_instance.ensure_role()
assert "Failed to grant privileges" in str(exc.value)
def test_ensure_database_dry_run_skips_creation(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=True)
connection_mock = mock.MagicMock()
cursor_mock = mock.MagicMock()
cursor_mock.fetchone.return_value = None
connection_mock.cursor.return_value = cursor_mock
with (
mock.patch.object(
setup_instance, "_admin_connection", return_value=connection_mock
),
mock.patch("scripts.setup_database.logger") as logger_mock,
):
setup_instance.ensure_database()
# expect only existence check, no create attempt
cursor_mock.execute.assert_called_once()
logger_mock.info.assert_any_call(
"Dry run: would create database '%s'. Run without --dry-run to proceed.",
mock_config.database,
)
def test_ensure_role_dry_run_skips_creation_and_grants(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=True)
admin_conn, admin_cursor = _connection_with_cursor()
admin_cursor.fetchone.return_value = None
with (
mock.patch.object(
setup_instance,
"_admin_connection",
side_effect=[admin_conn],
) as conn_mock,
mock.patch("scripts.setup_database.logger") as logger_mock,
):
setup_instance.ensure_role()
assert conn_mock.call_count == 1
admin_cursor.execute.assert_called_once()
logger_mock.info.assert_any_call(
"Dry run: would create role '%s'. Run without --dry-run to apply.",
mock_config.user,
)
def test_register_rollback_skipped_when_dry_run(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=True)
setup_instance._register_rollback("noop", lambda: None)
assert setup_instance._rollback_actions == []
def test_execute_rollbacks_runs_in_reverse_order(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
calls: list[str] = []
def first_action() -> None:
calls.append("first")
def second_action() -> None:
calls.append("second")
setup_instance._register_rollback("first", first_action)
setup_instance._register_rollback("second", second_action)
with mock.patch("scripts.setup_database.logger"):
setup_instance.execute_rollbacks()
assert calls == ["second", "first"]
assert setup_instance._rollback_actions == []
def test_ensure_database_registers_rollback_action(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
connection_mock = mock.MagicMock()
cursor_mock = mock.MagicMock()
cursor_mock.fetchone.return_value = None
connection_mock.cursor.return_value = cursor_mock
with (
mock.patch.object(
setup_instance, "_admin_connection", return_value=connection_mock
),
mock.patch.object(
setup_instance, "_register_rollback"
) as register_mock,
mock.patch.object(setup_instance, "_drop_database") as drop_mock,
):
setup_instance.ensure_database()
register_mock.assert_called_once()
label, action = register_mock.call_args[0]
assert "drop database" in label
action()
drop_mock.assert_called_once_with(mock_config.database)
def test_ensure_role_registers_rollback_actions(
mock_config: DatabaseConfig,
) -> None:
setup_instance = DatabaseSetup(mock_config, dry_run=False)
admin_conn, admin_cursor = _connection_with_cursor()
admin_cursor.fetchone.return_value = None
privilege_conn, privilege_cursor = _connection_with_cursor()
with (
mock.patch.object(
setup_instance,
"_admin_connection",
side_effect=[admin_conn, privilege_conn],
),
mock.patch.object(
setup_instance, "_register_rollback"
) as register_mock,
mock.patch.object(setup_instance, "_drop_role") as drop_mock,
mock.patch.object(
setup_instance, "_revoke_role_privileges"
) as revoke_mock,
):
setup_instance.ensure_role()
assert register_mock.call_count == 2
drop_label, drop_action = register_mock.call_args_list[0][0]
revoke_label, revoke_action = register_mock.call_args_list[1][0]
assert "drop role" in drop_label
assert "revoke privileges" in revoke_label
drop_action()
drop_mock.assert_called_once_with(mock_config.user)
revoke_action()
revoke_mock.assert_called_once()
def test_main_triggers_rollbacks_on_failure(
mock_config: DatabaseConfig,
) -> None:
args = argparse.Namespace(
ensure_database=True,
ensure_role=True,
ensure_schema=False,
initialize_schema=False,
run_migrations=False,
seed_data=False,
migrations_dir=None,
db_driver=None,
db_host=None,
db_port=None,
db_name=None,
db_user=None,
db_password=None,
db_schema=None,
admin_url=None,
admin_user=None,
admin_password=None,
admin_db=None,
dry_run=False,
verbose=0,
)
with (
mock.patch.object(setup_db_module, "parse_args", return_value=args),
mock.patch.object(
setup_db_module.DatabaseConfig, "from_env", return_value=mock_config
),
mock.patch.object(setup_db_module, "DatabaseSetup") as setup_cls,
):
setup_instance = mock.MagicMock()
setup_instance.dry_run = False
setup_instance._rollback_actions = [
("drop role", mock.MagicMock()),
]
setup_instance.ensure_database.side_effect = RuntimeError("boom")
setup_instance.execute_rollbacks = mock.MagicMock()
setup_instance.clear_rollbacks = mock.MagicMock()
setup_cls.return_value = setup_instance
with pytest.raises(RuntimeError):
setup_db_module.main()
setup_instance.execute_rollbacks.assert_called_once()
setup_instance.clear_rollbacks.assert_called_once()

232
tests/unit/test_simulation.py
View File

@@ -1,232 +0,0 @@
from math import isclose
from random import Random
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from typing import Any, Dict, List
from models.simulation_result import SimulationResult
from services.simulation import DEFAULT_UNIFORM_SPAN_RATIO, run_simulation
@pytest.fixture
def client(api_client: TestClient) -> TestClient:
return api_client
def test_run_simulation_function_generates_samples():
params: List[Dict[str, Any]] = [
{
"name": "grade",
"value": 1.8,
"distribution": "normal",
"std_dev": 0.2,
},
{
"name": "recovery",
"value": 0.9,
"distribution": "uniform",
"min": 0.8,
"max": 0.95,
},
]
results = run_simulation(params, iterations=5, seed=123)
assert isinstance(results, list)
assert len(results) == 5
assert results[0]["iteration"] == 1
def test_run_simulation_with_zero_iterations_returns_empty():
params: List[Dict[str, Any]] = [
{"name": "grade", "value": 1.2, "distribution": "normal"}
]
results = run_simulation(params, iterations=0)
assert results == []
@pytest.mark.parametrize(
"parameter_payload,error_message",
[
(
{"name": "missing-value"},
"Parameter at index 0 must include 'value'",
),
(
{
"name": "bad-dist",
"value": 1.0,
"distribution": "unsupported",
},
"Parameter 'bad-dist' has unsupported distribution 'unsupported'",
),
(
{
"name": "uniform-range",
"value": 1.0,
"distribution": "uniform",
"min": 5,
"max": 5,
},
"Parameter 'uniform-range' requires 'min' < 'max' for uniform distribution",
),
(
{
"name": "triangular-mode",
"value": 5.0,
"distribution": "triangular",
"min": 1,
"max": 3,
"mode": 5,
},
"Parameter 'triangular-mode' mode must be within min/max bounds for triangular distribution",
),
],
)
def test_run_simulation_parameter_validation_errors(
parameter_payload: Dict[str, Any], error_message: str
) -> None:
with pytest.raises(ValueError) as exc:
run_simulation([parameter_payload])
assert str(exc.value) == error_message
def test_run_simulation_normal_std_dev_fallback():
params: List[Dict[str, Any]] = [
{
"name": "std-dev-fallback",
"value": 10.0,
"distribution": "normal",
"std_dev": 0,
}
]
results = run_simulation(params, iterations=3, seed=99)
assert len(results) == 3
assert all("result" in entry for entry in results)
def test_run_simulation_triangular_sampling_path():
params: List[Dict[str, Any]] = [
{"name": "tri", "value": 10.0, "distribution": "triangular"}
]
seed = 21
iterations = 4
results = run_simulation(params, iterations=iterations, seed=seed)
assert len(results) == iterations
span = 10.0 * DEFAULT_UNIFORM_SPAN_RATIO
rng = Random(seed)
expected_samples = [
rng.triangular(10.0 - span, 10.0 + span, 10.0)
for _ in range(iterations)
]
actual_samples = [entry["result"] for entry in results]
for actual, expected in zip(actual_samples, expected_samples):
assert isclose(actual, expected, rel_tol=1e-9)
def test_run_simulation_uniform_defaults_apply_bounds():
params: List[Dict[str, Any]] = [
{"name": "uniform-auto", "value": 200.0, "distribution": "uniform"}
]
seed = 17
iterations = 3
results = run_simulation(params, iterations=iterations, seed=seed)
assert len(results) == iterations
span = 200.0 * DEFAULT_UNIFORM_SPAN_RATIO
rng = Random(seed)
expected_samples = [
rng.uniform(200.0 - span, 200.0 + span) for _ in range(iterations)
]
actual_samples = [entry["result"] for entry in results]
for actual, expected in zip(actual_samples, expected_samples):
assert isclose(actual, expected, rel_tol=1e-9)
def test_run_simulation_without_parameters_returns_empty():
assert run_simulation([], iterations=5) == []
def test_simulation_endpoint_no_params(client: TestClient):
scenario_payload: Dict[str, Any] = {
"name": f"NoParamScenario-{uuid4()}",
"description": "No parameters run",
}
scenario_resp = client.post("/api/scenarios/", json=scenario_payload)
assert scenario_resp.status_code == 200
scenario_id = scenario_resp.json()["id"]
resp = client.post(
"/api/simulations/run",
json={"scenario_id": scenario_id, "parameters": [], "iterations": 10},
)
assert resp.status_code == 400
assert resp.json()["detail"] == "No parameters provided"
def test_simulation_endpoint_success(client: TestClient, db_session: Session):
scenario_payload: Dict[str, Any] = {
"name": f"SimScenario-{uuid4()}",
"description": "Simulation test",
}
scenario_resp = client.post("/api/scenarios/", json=scenario_payload)
assert scenario_resp.status_code == 200
scenario_id = scenario_resp.json()["id"]
params: List[Dict[str, Any]] = [
{
"name": "param1",
"value": 2.5,
"distribution": "normal",
"std_dev": 0.5,
}
]
payload: Dict[str, Any] = {
"scenario_id": scenario_id,
"parameters": params,
"iterations": 10,
"seed": 42,
}
resp = client.post("/api/simulations/run", json=payload)
assert resp.status_code == 200
data = resp.json()
assert data["scenario_id"] == scenario_id
assert len(data["results"]) == 10
assert data["summary"]["count"] == 10
db_session.expire_all()
persisted = (
db_session.query(SimulationResult)
.filter(SimulationResult.scenario_id == scenario_id)
.all()
)
assert len(persisted) == 10
def test_simulation_endpoint_uses_stored_parameters(client: TestClient):
scenario_payload: Dict[str, Any] = {
"name": f"StoredParams-{uuid4()}",
"description": "Stored parameter simulation",
}
scenario_resp = client.post("/api/scenarios/", json=scenario_payload)
assert scenario_resp.status_code == 200
scenario_id = scenario_resp.json()["id"]
parameter_payload: Dict[str, Any] = {
"scenario_id": scenario_id,
"name": "grade",
"value": 1.5,
}
param_resp = client.post("/api/parameters/", json=parameter_payload)
assert param_resp.status_code == 200
resp = client.post(
"/api/simulations/run",
json={"scenario_id": scenario_id, "iterations": 3, "seed": 7},
)
assert resp.status_code == 200
data = resp.json()
assert data["summary"]["count"] == 3
assert len(data["results"]) == 3

56
tests/unit/test_theme_settings.py
View File

@@ -1,56 +0,0 @@
from sqlalchemy.orm import Session
from services.settings import save_theme_settings, get_theme_settings
def test_save_theme_settings(db_session: Session):
theme_data = {
"theme_name": "dark",
"primary_color": "#000000",
"secondary_color": "#333333",
"accent_color": "#ff0000",
"background_color": "#1a1a1a",
"text_color": "#ffffff"
}
saved_setting = save_theme_settings(db_session, theme_data)
assert str(saved_setting.theme_name) == "dark"
assert str(saved_setting.primary_color) == "#000000"
def test_get_theme_settings(db_session: Session):
# Create a theme setting first
theme_data = {
"theme_name": "light",
"primary_color": "#ffffff",
"secondary_color": "#cccccc",
"accent_color": "#0000ff",
"background_color": "#f0f0f0",
"text_color": "#000000"
}
save_theme_settings(db_session, theme_data)
settings = get_theme_settings(db_session)
assert settings["theme_name"] == "light"
assert settings["primary_color"] == "#ffffff"
def test_theme_settings_api(api_client):
# Test API endpoint for saving theme settings
theme_data = {
"theme_name": "test_theme",
"primary_color": "#123456",
"secondary_color": "#789abc",
"accent_color": "#def012",
"background_color": "#345678",
"text_color": "#9abcde"
}
response = api_client.post("/api/settings/theme", json=theme_data)
assert response.status_code == 200
assert response.json()["theme"]["theme_name"] == "test_theme"
# Test API endpoint for getting theme settings
response = api_client.get("/api/settings/theme")
assert response.status_code == 200
assert response.json()["theme_name"] == "test_theme"

179
tests/unit/test_ui_routes.py
View File

@@ -1,179 +0,0 @@
from typing import Any, Dict, cast
import pytest
from fastapi.testclient import TestClient
from models.scenario import Scenario
from services import settings as settings_service
def test_dashboard_route_provides_summary(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/dashboard")
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == "Dashboard.html"
context = cast(Dict[str, Any], getattr(response, "context", {}))
assert context.get("report_available") is True
metric_labels = {item["label"] for item in context["summary_metrics"]}
assert {
"CAPEX Total",
"OPEX Total",
"Production",
"Simulation Iterations",
}.issubset(metric_labels)
scenario = cast(Scenario, seeded_ui_data["scenario"])
scenario_row = next(
row
for row in context["scenario_rows"]
if row["scenario_name"] == scenario.name
)
assert scenario_row["iterations"] == 3
assert scenario_row["simulation_mean_display"] == "971,666.67"
assert scenario_row["capex_display"] == "$1,000,000.00"
assert scenario_row["opex_display"] == "$250,000.00"
assert scenario_row["production_display"] == "800.00"
assert scenario_row["consumption_display"] == "1,200.00"
def test_scenarios_route_lists_seeded_scenario(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/scenarios")
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == "ScenarioForm.html"
context = cast(Dict[str, Any], getattr(response, "context", {}))
names = [item["name"] for item in context["scenarios"]]
scenario = cast(Scenario, seeded_ui_data["scenario"])
assert scenario.name in names
def test_reporting_route_includes_summary(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/reporting")
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == "reporting.html"
context = cast(Dict[str, Any], getattr(response, "context", {}))
summaries = context["report_summaries"]
scenario = cast(Scenario, seeded_ui_data["scenario"])
scenario_summary = next(
item for item in summaries if item["scenario_id"] == scenario.id
)
assert scenario_summary["iterations"] == 3
mean_value = float(scenario_summary["summary"]["mean"])
assert abs(mean_value - 971_666.6666666666) < 1e-6
def test_dashboard_data_endpoint_returns_aggregates(
api_client: TestClient, seeded_ui_data: Dict[str, Any]
) -> None:
response = api_client.get("/ui/dashboard/data")
assert response.status_code == 200
payload = response.json()
assert payload["report_available"] is True
metric_map = {
item["label"]: item["value"] for item in payload["summary_metrics"]
}
assert metric_map["CAPEX Total"].startswith("$")
assert metric_map["Maintenance Cost"].startswith("$")
scenario = cast(Scenario, seeded_ui_data["scenario"])
scenario_rows = payload["scenario_rows"]
scenario_entry = next(
row for row in scenario_rows if row["scenario_name"] == scenario.name
)
assert scenario_entry["capex_display"] == "$1,000,000.00"
assert scenario_entry["production_display"] == "800.00"
labels = payload["scenario_cost_chart"]["labels"]
idx = labels.index(scenario.name)
assert payload["scenario_cost_chart"]["capex"][idx] == 1_000_000.0
activity_labels = payload["scenario_activity_chart"]["labels"]
activity_idx = activity_labels.index(scenario.name)
assert (
payload["scenario_activity_chart"]["production"][activity_idx] == 800.0
)
@pytest.mark.parametrize(
("path", "template_name"),
[
("/", "Dashboard.html"),
("/ui/parameters", "ParameterInput.html"),
("/ui/costs", "costs.html"),
("/ui/consumption", "consumption.html"),
("/ui/production", "production.html"),
("/ui/equipment", "equipment.html"),
("/ui/maintenance", "maintenance.html"),
("/ui/simulations", "simulations.html"),
],
)
def test_additional_ui_routes_render_templates(
api_client: TestClient,
seeded_ui_data: Dict[str, Any],
path: str,
template_name: str,
) -> None:
response = api_client.get(path)
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == template_name
context = cast(Dict[str, Any], getattr(response, "context", {}))
assert context
def test_settings_route_provides_css_context(
api_client: TestClient,
monkeypatch: pytest.MonkeyPatch,
) -> None:
env_var = settings_service.css_key_to_env_var("--color-accent")
monkeypatch.setenv(env_var, "#abcdef")
response = api_client.get("/ui/settings")
assert response.status_code == 200
template = getattr(response, "template", None)
assert template is not None
assert template.name == "settings.html"
context = cast(Dict[str, Any], getattr(response, "context", {}))
assert "css_variables" in context
assert "css_defaults" in context
assert "css_env_overrides" in context
assert "css_env_override_rows" in context
assert "css_env_override_meta" in context
assert context["css_variables"]["--color-accent"] == "#abcdef"
assert (
context["css_defaults"]["--color-accent"]
== settings_service.CSS_COLOR_DEFAULTS["--color-accent"]
)
assert context["css_env_overrides"]["--color-accent"] == "#abcdef"
override_rows = context["css_env_override_rows"]
assert any(row["env_var"] == env_var for row in override_rows)
meta = context["css_env_override_meta"]["--color-accent"]
assert meta["value"] == "#abcdef"
assert meta["env_var"] == env_var

28
tests/unit/test_validation.py
View File

@@ -1,28 +0,0 @@
from uuid import uuid4
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
def test_validate_json_allows_valid_payload(api_client: TestClient) -> None:
payload = {
"name": f"ValidJSON-{uuid4()}",
"description": "Middleware should allow valid JSON.",
}
response = api_client.post("/api/scenarios/", json=payload)
assert response.status_code == 200
data = response.json()
assert data["name"] == payload["name"]
def test_validate_json_rejects_invalid_payload(api_client: TestClient) -> None:
with pytest.raises(HTTPException) as exc_info:
api_client.post(
"/api/scenarios/",
content=b"{not valid json",
headers={"Content-Type": "application/json"},
)
assert exc_info.value.status_code == 400
assert exc_info.value.detail == "Invalid JSON payload"