Update: Adding integration test initial
TAMDeveloper13
2 months ago
parent
2dfa783a58
commit
54fe0e1c12
@ -0,0 +1,905 @@
|
||||
"""
|
||||
Integration tests for Optimum OH backend.
|
||||
All endpoints are sourced from frontend API calls using OPTIMUM_OH_API_URL.
|
||||
|
||||
Run with:
|
||||
pytest src/testing/integration_test.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
BASE_URL = "http://100.125.115.116:8000/optimumoh"
|
||||
AUTH_BASE_URL = "http://100.125.115.116:8000/auth"
|
||||
|
||||
# Replace with a valid token before running
|
||||
AUTH_TOKEN = "Bearer <your_token_here>"
|
||||
|
||||
HEADERS = {
|
||||
"Authorization": AUTH_TOKEN,
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
# Placeholder IDs — replace with real values from your database before running
|
||||
OVERHAUL_SESSION_ID = 1
|
||||
OVERHAUL_SCHEDULE_ID = 1
|
||||
OVERHAUL_EQUIPMENT_ID = 1
|
||||
SCOPE_JOB_ID = 1
|
||||
CALCULATION_ID = 1
|
||||
BUDGET_THRESHOLD = 100000
|
||||
RISK_COST = 1
|
||||
LOCATION_TAG = "LOCATION-001"
|
||||
SCOPE_NAME = "scope-name-example"
|
||||
OVERHAUL_SESSION_HISTORY_ID = 1
|
||||
ASSET_NUM = "ASSET-001"
|
||||
SCOPE_EQUIPMENT_JOB_ID = 1
|
||||
EQUIPMENT_ACTIVITY_ID = 1
|
||||
EAF_INPUT = 0.85
|
||||
DURATION = 8760
|
||||
SCOPE_EQUIPMENT_ACTIVITY_ID = 1
|
||||
ITEMNUM = "ITEM-001"
|
||||
SPREADSHEET_LINK = "https://docs.google.com/spreadsheets/d/example_spreadsheet_id/edit"
|
||||
# UUID-format placeholder — replace with a real UUID from your database
|
||||
OVERHAUL_SESSION_UUID = "00000000-0000-0000-0000-000000000001"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Users
|
||||
# ---------------------------------------------------------------------------
|
||||
USER_LIST = [
|
||||
{"id": 1, "name": "devmustbeadmin", "pass": "", "email": "devmustbeadmin@mail.com", "authorization_token": ""},
|
||||
# {"id": 2, "name": "devmustbeengineer", "pass": "", "email": "devmustbeengineer@mail.com", "authorization_token": ""},
|
||||
# {"id": 3, "name": "devmustbeapplicationadmin", "pass": "", "email": "devmustbeapplicationadmin@mail.com", "authorization_token": ""},
|
||||
# {"id": 4, "name": "devmustbemanagement", "pass": "", "email": "devmustbemanagement@mail.com", "authorization_token": ""},
|
||||
]
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Role-based access constants
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoints guarded by require_any_role(*ALLOWED_ROLES) block Management entirely.
|
||||
# Endpoints with only required_permission(READ) allow Management (has READ in ACL).
|
||||
# Endpoints with no guard at all allow every role.
|
||||
NON_MANAGEMENT = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin"}
|
||||
ALL_ALLOWED = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin", "devmustbemanagement"}
|
||||
|
||||
|
||||
def user_auth_headers(token: str) -> dict:
|
||||
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||||
|
||||
|
||||
def get_csrf_token(path: str, auth_token: str) -> str:
|
||||
"""Fetch a one-time CSRF token from be-auth for the given endpoint path.
|
||||
|
||||
Args:
|
||||
path: The endpoint path as seen by be-optimumoh (e.g. "/spareparts").
|
||||
Do NOT include the "/optimumoh" Kong prefix.
|
||||
auth_token: The raw JWT token string (without "Bearer " prefix).
|
||||
|
||||
Returns:
|
||||
The plain-text CSRF token to send in the X-CSRF-Token header.
|
||||
"""
|
||||
res = requests.post(
|
||||
f"{AUTH_BASE_URL}/csrf-token",
|
||||
headers={
|
||||
"Authorization": f"Bearer {auth_token}",
|
||||
"Content-Type": "application/json",
|
||||
"X-Target-Path": path,
|
||||
},
|
||||
json={
|
||||
"_csrf_request": True,
|
||||
}
|
||||
)
|
||||
print(f"Requested CSRF token for {path}, got status {res.status_code}, response: {res.text}")
|
||||
assert res.status_code == 200, f"Failed to get CSRF token for {path}: {res.status_code} {res.text}"
|
||||
data = res.json()
|
||||
# Response shape: {"data": {"csrf_token": "..."}, ...}
|
||||
return data["data"]["csrf_token"]
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhaul Session
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
# # Note: Source code got commented
|
||||
# def test_get_overhaul_session():
|
||||
# """GET /overhaul-session?all=1 — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(f"{BASE_URL}/overhaul-session?all=1", headers=user_auth_headers(user["authorization_token"]))
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# print(f"[{user['name']}] expected {expected}, got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
|
||||
# # Note: Source code got commented
|
||||
# def test_post_overhaul_session():
|
||||
# """POST /overhaul-session — require_any_role + CREATE blocks Management"""
|
||||
# payload = {
|
||||
# "type": "Major",
|
||||
# "start_date": "2025-01-01",
|
||||
# "end_date": "2025-12-31",
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(f"{BASE_URL}/overhaul-session", json=payload, headers=user_auth_headers(user["authorization_token"]))
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Overhaul Schedules
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
|
||||
# def test_post_overhaul_schedules():
|
||||
# """POST /overhaul-schedules — require_any_role + CREATE blocks Management"""
|
||||
# payload = {
|
||||
# "start_date": "2024-06-01T00:00:00Z",
|
||||
# "end_date": "2024-06-30T23:59:59Z",
|
||||
# "duration_oh": 720,
|
||||
# "crew_number": 10,
|
||||
# "status": "Upcoming"
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(f"{BASE_URL}/overhaul-schedules", json=payload, headers=user_auth_headers(user["authorization_token"]))
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_delete_overhaul_schedule():
|
||||
# """DELETE /overhaul-schedules/{id} — require_any_role + DELETE blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.delete(
|
||||
# f"{BASE_URL}/overhaul-schedules/{OVERHAUL_SCHEDULE_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhaul Activity
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
# def test_post_overhaul_activity_current():
|
||||
# """POST /overhaul-activity/{session_id} — require_any_role blocks Management"""
|
||||
# payload = {
|
||||
# "location_tag": LOCATION_TAG,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_delete_overhaul_activity():
|
||||
# """DELETE /overhaul-activity/{overhaul_session}/delete/{location_tag} — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.delete(
|
||||
# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}/delete/{LOCATION_TAG}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhaul Jobs
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
|
||||
# def test_post_overhaul_jobs():
|
||||
# """POST /overhaul-jobs/{scope_job_id} — no role guard; all roles allowed"""
|
||||
# payload = {
|
||||
# "overhaul_equipment_id": OVERHAUL_EQUIPMENT_ID,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/overhaul-jobs/{SCOPE_JOB_ID}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Scope Equipments
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
|
||||
# def test_post_scope_equipments():
|
||||
# """POST /scope-equipments — require_any_role blocks Management"""
|
||||
# payload = {
|
||||
# "assetnums": ["A26312", "A26313", "A26314"],
|
||||
# "scope_name": "A"
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(f"{BASE_URL}/scope-equipments", json=payload, headers=user_auth_headers(user["authorization_token"]))
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_delete_scope_equipment():
|
||||
# """DELETE /scope-equipments/delete/{id} — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.delete(
|
||||
# f"{BASE_URL}/scope-equipments/delete/{OVERHAUL_EQUIPMENT_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Scope Equipment Jobs
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
# def test_post_scope_equipment_jobs():
|
||||
# """POST /scope-equipment-jobs/{assetnum} — no role guard; all roles allowed"""
|
||||
# payload = {
|
||||
# "job_id": SCOPE_JOB_ID,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/scope-equipment-jobs/{ASSET_NUM}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_delete_scope_equipment_job():
|
||||
# """DELETE /scope-equipment-jobs/delete/{id} — no role guard; all roles allowed"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.delete(
|
||||
# f"{BASE_URL}/scope-equipment-jobs/delete/{SCOPE_EQUIPMENT_JOB_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Calculation — Budget Constraint
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_post_calculation_time_constraint():
|
||||
# """POST /calculation/time-constraint — required_permission(CREATE); Management has no CREATE"""
|
||||
# payload = {
|
||||
# # "overhaulCost": 500000000,
|
||||
# # "overhaul_session_id": "86d6610e-1d44-4221-a521-e1dff2ea150f",
|
||||
# # "costPerFailure": 0
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/calculation/time-constraint",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_post_calculation_time_constraint_select_equipment():
|
||||
# """POST /calculation/time-constraint?scope_calculation_id=... — required_permission(CREATE); Management has no CREATE"""
|
||||
# payload = {}
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/calculation/time-constraint?scope_calculation_id={CALCULATION_ID}&with_results=1",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_put_calculation_time_constraint_update():
|
||||
# """PUT /calculation/time-constraint/update/{calculationId} — required_permission(EDIT); Management has no EDIT"""
|
||||
# payload = {
|
||||
# "risk_cost": RISK_COST,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.put(
|
||||
# f"{BASE_URL}/calculation/time-constraint/update/{CALCULATION_ID}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Spareparts
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
|
||||
# def test_post_spareparts():
|
||||
# """POST /spareparts — require_any_role blocks Management"""
|
||||
# payload = {
|
||||
# "name": "Test Part",
|
||||
# "part_number": "TP-001",
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(f"{BASE_URL}/spareparts", json=payload, headers=user_auth_headers(user["authorization_token"]))
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Equipment Activities / Workscopes
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_equipment_workscopes():
|
||||
# """GET /equipment-workscopes/{location_tag} — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/equipment-workscopes/{LOCATION_TAG}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_post_equipment_activities():
|
||||
# """POST /equipment-activities — no role guard; all roles allowed"""
|
||||
# payload = {
|
||||
# "location_tag": LOCATION_TAG,
|
||||
# "workscope": "Inspection",
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/equipment-activities",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_delete_equipment_activity():
|
||||
# """DELETE /equipment-activities/delete/{id} — no role guard; all roles allowed"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.delete(
|
||||
# f"{BASE_URL}/equipment-activities/delete/{EQUIPMENT_ACTIVITY_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Jobs
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_jobs():
|
||||
# """GET /jobs — no role guard; all roles allowed"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(f"{BASE_URL}/jobs", headers=user_auth_headers(user["authorization_token"]))
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Overhaul Jobs
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_overhaul_jobs():
|
||||
# """GET /overhaul-jobs/{overhaul_equipment_id} — no role guard; all roles allowed"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/overhaul-jobs/{OVERHAUL_EQUIPMENT_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Calculation — Time Constraint
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_calculation_time_constraint_by_id():
|
||||
# """GET /calculation/time-constraint/{calculationId} — required_permission(READ) only; Management has READ"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}?risk_cost={RISK_COST}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# def test_get_calculation_time_constraint_parameters():
|
||||
# """GET /calculation/time-constraint/parameters — required_permission(READ) only; Management has READ"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/time-constraint/parameters",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Calculation — Target Reliability
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_calculation_target_reliability():
|
||||
# """GET /calculation/target-reliability — required_permission(READ) only; Management has READ"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/target-reliability"
|
||||
# f"?eaf_input={EAF_INPUT}&oh_session_id={OVERHAUL_SESSION_ID}&duration={DURATION}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# def test_get_calculation_target_reliability_with_cut_hours():
|
||||
# """GET /calculation/target-reliability?...&cut_hours=... — required_permission(READ) only; Management has READ"""
|
||||
# cut_hours = 500
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/target-reliability"
|
||||
# f"?oh_session_id={OVERHAUL_SESSION_ID}&eaf_input={EAF_INPUT}"
|
||||
# f"&duration={DURATION}&cut_hours={cut_hours}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Overhaul Activity
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_overhaul_activity():
|
||||
# """GET /overhaul-activity/{overhaul_session} — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
|
||||
# # =============================================================================
|
||||
# # NEW TESTS — endpoints missing from the original test file
|
||||
# # =============================================================================
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Overhauls sub-routes
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_overhauls_schedules():
|
||||
# """GET /overhauls/schedules — required_permission(READ); Management has READ"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(f"{BASE_URL}/overhauls/schedules", headers=user_auth_headers(user["authorization_token"]))
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_get_overhauls_critical_parts():
|
||||
# """GET /overhauls/critical-parts — required_permission(READ); Management has READ"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(f"{BASE_URL}/overhauls/critical-parts", headers=user_auth_headers(user["authorization_token"]))
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Scope Equipments
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# # Could not test endpoint due to unknown oh_session_id
|
||||
# def test_get_scope_equipments_history():
|
||||
# """GET /scope-equipments/history/{oh_session_id} — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/scope-equipments/history/{OVERHAUL_SESSION_UUID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_post_scope_equipments():
|
||||
# """POST /scope-equipments — require_any_role blocks Management
|
||||
# Payload: assetnums (List[str]), scope_name (str), removal_date (optional datetime), type (optional str)
|
||||
# """
|
||||
# payload = {
|
||||
# "assetnums": [ASSET_NUM],
|
||||
# "scope_name": SCOPE_NAME,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(f"{BASE_URL}/scope-equipments", json=payload, headers=user_auth_headers(user["authorization_token"]))
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Overhaul Activity
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_overhaul_activity():
|
||||
# """GET /overhaul-activity/{overhaul_session} — require_any_role blocks Management
|
||||
# Query params: location_tag (optional), scope_name (optional)
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_UUID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_post_overhaul_activity():
|
||||
# """POST /overhaul-activity/{overhaul_session_id} — require_any_role blocks Management
|
||||
# Payload: location_tags (List[str])
|
||||
# Path: overhaul_session_id must be a valid UUID
|
||||
# """
|
||||
# payload = {
|
||||
# "location_tags": [LOCATION_TAG],
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_UUID}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_get_overhaul_activity_by_assetnum():
|
||||
# """GET /overhaul-activity/{overhaul_session}/{assetnum} — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/overhaul-activity/{OVERHAUL_SESSION_UUID}/{ASSET_NUM}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_post_overhaul_activity_delete():
|
||||
# """POST /overhaul-activity/delete/{overhaul_session}/{location_tag} — require_any_role blocks Management
|
||||
# No body required. CSRF token required in production.
|
||||
# Path: overhaul_session must be a valid UUID
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/overhaul-activity/delete/{OVERHAUL_SESSION_UUID}/{LOCATION_TAG}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Workscopes
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_post_workscopes():
|
||||
# """POST /workscopes — no role guard; all roles allowed
|
||||
# Payload: workscope (str)
|
||||
# """
|
||||
# payload = {
|
||||
# # "workscope": "Test workscope description",
|
||||
# # "system": "Test system",
|
||||
# # "subsystem": "Test subsystem",
|
||||
# # "tasks": "Test task",
|
||||
# # "equipment_workscope_groups": "Test group",
|
||||
# # "oh_types": "Test OH type",
|
||||
# "description": "Test description",
|
||||
# # "job_id": SCOPE_JOB_ID,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(f"{BASE_URL}/workscopes", json=payload, headers=user_auth_headers(user["authorization_token"]))
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_get_workscopes_by_id():
|
||||
# """GET /workscopes/{scope_equipment_activity_id} — no role guard; all roles allowed"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/workscopes/{SCOPE_EQUIPMENT_ACTIVITY_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_post_workscopes_update():
|
||||
# """POST /workscopes/update/{scope_equipment_activity_id} — no role guard; all roles allowed
|
||||
# Payload: description (str)
|
||||
# """
|
||||
# payload = {
|
||||
# "description": "Updated workscope description",
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/workscopes/update/{SCOPE_EQUIPMENT_ACTIVITY_ID}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_post_workscopes_delete():
|
||||
# """POST /workscopes/delete/{scope_equipment_activity_id} — no role guard; all roles allowed
|
||||
# No body required.
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/workscopes/delete/{SCOPE_EQUIPMENT_ACTIVITY_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Equipment Workscopes
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_equipment_workscopes():
|
||||
# """GET /equipment-workscopes/{location_tag} — no role guard; all roles allowed
|
||||
# Query params: scope (optional str)
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/equipment-workscopes/{LOCATION_TAG}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_post_equipment_workscopes():
|
||||
# """POST /equipment-workscopes/{assetnum} — no role guard; all roles allowed
|
||||
# Payload: job_ids (List[UUID])
|
||||
# """
|
||||
# payload = {
|
||||
# "job_ids": [],
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/equipment-workscopes/{ASSET_NUM}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_post_equipment_workscopes_delete():
|
||||
# """POST /equipment-workscopes/delete/{scope_job_id} — no role guard; all roles allowed
|
||||
# No body required.
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/equipment-workscopes/delete/{SCOPE_JOB_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Overhaul Schedules (additional)
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_overhaul_schedules_by_id():
|
||||
# """GET /overhaul-schedules/{overhaul_session_id} — require_any_role blocks Management"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/overhaul-schedules/{OVERHAUL_SESSION_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_post_overhaul_schedules_update():
|
||||
# """POST /overhaul-schedules/update/{scope_id} — require_any_role + EDIT blocks Management
|
||||
# Payload: duration_oh (optional int), crew_number (optional int), status (optional str)
|
||||
# """
|
||||
# payload = {
|
||||
# "duration_oh": 720,
|
||||
# "crew_number": 10,
|
||||
# "status": "Upcoming",
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/overhaul-schedules/update/{OVERHAUL_SESSION_ID}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_post_overhaul_schedules_delete():
|
||||
# """POST /overhaul-schedules/delete/{scope_id} — require_any_role + DELETE blocks Management
|
||||
# No body required. CSRF token required in production.
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/overhaul-schedules/delete/{OVERHAUL_SESSION_ID}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Calculation — Time Constraint
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_post_calculation_time_constraint():
|
||||
# """POST /calculation/time-constraint — required_permission(CREATE); Management has no CREATE
|
||||
# Payload: overhaulCost (float), ohSessionId (UUID), costPerFailure (float)
|
||||
# Query params: scope_calculation_id (optional), with_results (optional int), simulation_id (optional)
|
||||
# """
|
||||
# payload = {
|
||||
# "overhaulCost": 0,
|
||||
# "ohSessionId": OVERHAUL_SESSION_UUID,
|
||||
# "costPerFailure": 0,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/calculation/time-constraint",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_get_calculation_time_constraint():
|
||||
# """GET /calculation/time-constraint — required_permission(READ); Management has READ"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/time-constraint",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_get_calculation_time_constraint_parameters():
|
||||
# """GET /calculation/time-constraint/parameters — required_permission(READ); Management has READ
|
||||
# Query params: calculation_id (optional)
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/time-constraint/parameters",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_get_calculation_time_constraint_by_id():
|
||||
# """GET /calculation/time-constraint/{calculation_id} — required_permission(READ); Management has READ
|
||||
# Query params: risk_cost (optional int, default 1)
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}?risk_cost={RISK_COST}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_get_calculation_time_constraint_by_id_and_assetnum():
|
||||
# """GET /calculation/time-constraint/{calculation_id}/{assetnum} — required_permission(READ); Management has READ"""
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}/{ASSET_NUM}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 404), f"[{user['name']}] expected 200/404, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# def test_post_calculation_time_constraint_simulation():
|
||||
# """POST /calculation/time-constraint/{calculation_id}/simulation — required_permission(READ); Management has READ
|
||||
# Payload: intervalDays (int)
|
||||
# """
|
||||
# payload = {
|
||||
# "intervalDays": 365,
|
||||
# }
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/calculation/time-constraint/{CALCULATION_ID}/simulation",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# assert res.status_code in (200, 201, 404), f"[{user['name']}] expected 200/201/404, got {res.status_code}"
|
||||
|
||||
|
||||
# def test_post_calculation_time_constraint_update():
|
||||
# """POST /calculation/time-constraint/update/{calculation_id} — required_permission(EDIT); Management has no EDIT
|
||||
# Payload: List of { id (UUID), is_included (bool) }. CSRF token required in production.
|
||||
# """
|
||||
# payload = []
|
||||
# for user in USER_LIST:
|
||||
# res = requests.post(
|
||||
# f"{BASE_URL}/calculation/time-constraint/update/{CALCULATION_ID}",
|
||||
# json=payload,
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# print(f"[{user['name']}], got {res.status_code}, response: {res.text}")
|
||||
# if user["name"] in NON_MANAGEMENT:
|
||||
# assert res.status_code in (200, 204, 404), f"[{user['name']}] expected 200/204/404, got {res.status_code}"
|
||||
# else:
|
||||
# assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Calculation — Target Reliability
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_calculation_target_reliability():
|
||||
# """GET /calculation/target-reliability — required_permission(READ); Management has READ
|
||||
# Query params: oh_session_id (required), eaf_input (float), duration (int), simulation_id (optional), cut_hours (optional)
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/target-reliability"
|
||||
# f"?oh_session_id={OVERHAUL_SESSION_UUID}&eaf_input={EAF_INPUT}&duration={DURATION}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# assert res.status_code in (200, 400, 404, 425), f"[{user['name']}] expected 200/400/404/425, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# # ─────────────────────────────────────────────
|
||||
# # Calculation — Budget Constraint
|
||||
# # ─────────────────────────────────────────────
|
||||
|
||||
# def test_get_calculation_budget_constraint():
|
||||
# """GET /calculation/budget-constraint/{session_id} — require_any_role blocks Management
|
||||
# Query params: cost_threshold (float, default 100)
|
||||
# """
|
||||
# for user in USER_LIST:
|
||||
# res = requests.get(
|
||||
# f"{BASE_URL}/calculation/budget-constraint/{OVERHAUL_SESSION_UUID}?cost_threshold={BUDGET_THRESHOLD}",
|
||||
# headers=user_auth_headers(user["authorization_token"]),
|
||||
# )
|
||||
# expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
# assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}"
|
||||
@ -0,0 +1,285 @@
|
||||
"""
|
||||
Integration tests for Optimum OH backend.
|
||||
All endpoints are sourced from frontend API calls using OPTIMUM_OH_API_URL.
|
||||
|
||||
Run with:
|
||||
pytest src/testing/integration_test.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
BASE_URL = "http://100.125.115.116:8000/optimumoh"
|
||||
AUTH_BASE_URL = "http://100.125.115.116:8000/auth"
|
||||
|
||||
# Replace with a valid token before running
|
||||
AUTH_TOKEN = "Bearer <your_token_here>"
|
||||
|
||||
HEADERS = {
|
||||
"Authorization": AUTH_TOKEN,
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
# Placeholder IDs — replace with real values from your database before running
|
||||
OVERHAUL_SESSION_ID = 1
|
||||
OVERHAUL_SCHEDULE_ID = 1
|
||||
OVERHAUL_EQUIPMENT_ID = 1
|
||||
SCOPE_JOB_ID = 1
|
||||
CALCULATION_ID = 1
|
||||
BUDGET_THRESHOLD = 100000
|
||||
RISK_COST = 1
|
||||
LOCATION_TAG = "LOCATION-001"
|
||||
SCOPE_NAME = "scope-name-example"
|
||||
OVERHAUL_SESSION_HISTORY_ID = 1
|
||||
ASSET_NUM = "ASSET-001"
|
||||
SCOPE_EQUIPMENT_JOB_ID = 1
|
||||
EQUIPMENT_ACTIVITY_ID = 1
|
||||
EAF_INPUT = 0.85
|
||||
DURATION = 8760
|
||||
SCOPE_EQUIPMENT_ACTIVITY_ID = 1
|
||||
ITEMNUM = "ITEM-001"
|
||||
SPREADSHEET_LINK = "https://docs.google.com/spreadsheets/d/example_spreadsheet_id/edit"
|
||||
# UUID-format placeholder — replace with a real UUID from your database
|
||||
OVERHAUL_SESSION_UUID = "00000000-0000-0000-0000-000000000001"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Users
|
||||
# ---------------------------------------------------------------------------
|
||||
USER_LIST = [
|
||||
{"id": 1, "name": "devmustbeadmin", "pass": "", "email": "devmustbeadmin@mail.com", "authorization_token": ""},
|
||||
# {"id": 2, "name": "devmustbeengineer", "pass": "", "email": "devmustbeengineer@mail.com", "authorization_token": ""},
|
||||
# {"id": 3, "name": "devmustbeapplicationadmin", "pass": "", "email": "devmustbeapplicationadmin@mail.com", "authorization_token": ""},
|
||||
# {"id": 4, "name": "devmustbemanagement", "pass": "", "email": "devmustbemanagement@mail.com", "authorization_token": ""},
|
||||
]
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Role-based access constants
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoints guarded by require_any_role(*ALLOWED_ROLES) block Management entirely.
|
||||
# Endpoints with only required_permission(READ) allow Management (has READ in ACL).
|
||||
# Endpoints with no guard at all allow every role.
|
||||
NON_MANAGEMENT = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin"}
|
||||
ALL_ALLOWED = {"devmustbeadmin", "devmustbeengineer", "devmustbeapplicationadmin", "devmustbemanagement"}
|
||||
|
||||
|
||||
def user_auth_headers(token: str) -> dict:
|
||||
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||||
|
||||
|
||||
def get_csrf_token(path: str, auth_token: str) -> str:
|
||||
"""Fetch a one-time CSRF token from be-auth for the given endpoint path.
|
||||
|
||||
Args:
|
||||
path: The endpoint path as seen by be-optimumoh (e.g. "/spareparts").
|
||||
Do NOT include the "/optimumoh" Kong prefix.
|
||||
auth_token: The raw JWT token string (without "Bearer " prefix).
|
||||
|
||||
Returns:
|
||||
The plain-text CSRF token to send in the X-CSRF-Token header.
|
||||
"""
|
||||
res = requests.post(
|
||||
f"{AUTH_BASE_URL}/csrf-token",
|
||||
headers={
|
||||
"Authorization": f"Bearer {auth_token}",
|
||||
"Content-Type": "application/json",
|
||||
"X-Target-Path": path,
|
||||
},
|
||||
json={
|
||||
"_csrf_request": True,
|
||||
}
|
||||
)
|
||||
assert res.status_code == 200, f"Failed to get CSRF token for {path}: {res.status_code} {res.text}"
|
||||
data = res.json()
|
||||
return data["data"]["csrf_token"]
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhaul Schedules
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_overhaul_schedules():
|
||||
"""GET /overhaul-schedules — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/overhaul-schedules", headers=user_auth_headers(user["authorization_token"]))
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
|
||||
def test_get_overhaul_schedules_history():
|
||||
"""GET /overhaul-schedules/history — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/overhaul-schedules/history", headers=user_auth_headers(user["authorization_token"]))
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhauls
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_overhauls():
|
||||
"""GET /overhauls — required_permission(READ) only; Management has READ in ACL"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/overhauls", headers=user_auth_headers(user["authorization_token"]))
|
||||
assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhaul Gantt
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_overhaul_gantt():
|
||||
"""GET /overhaul-gantt — no role guard; all roles allowed"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/overhaul-gantt", headers=user_auth_headers(user["authorization_token"]))
|
||||
assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
def test_get_overhaul_gantt_spreadsheet():
|
||||
"""GET /overhaul-gantt/spreadsheet — no role guard; all roles allowed"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(
|
||||
f"{BASE_URL}/overhaul-gantt/spreadsheet",
|
||||
headers=user_auth_headers(user["authorization_token"]),
|
||||
)
|
||||
assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Spareparts
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_spareparts():
|
||||
"""GET /spareparts — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/spareparts", headers=user_auth_headers(user["authorization_token"]))
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
def test_get_equipment_spareparts():
|
||||
"""GET /equipment-spareparts/{location_tag} — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(
|
||||
f"{BASE_URL}/equipment-spareparts/{LOCATION_TAG}",
|
||||
headers=user_auth_headers(user["authorization_token"]),
|
||||
)
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
def test_post_spareparts():
|
||||
"""POST /spareparts — require_any_role blocks Management
|
||||
Payload: itemnum (str), remark (str). CSRF token required.
|
||||
"""
|
||||
payload = {
|
||||
"itemnum": ITEMNUM,
|
||||
"remark": "Test remark for integration test",
|
||||
}
|
||||
for user in USER_LIST:
|
||||
csrf_token = get_csrf_token("/spareparts", user["authorization_token"])
|
||||
headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token}
|
||||
res = requests.post(f"{BASE_URL}/spareparts", json=payload, headers=headers)
|
||||
if user["name"] in NON_MANAGEMENT:
|
||||
assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
else:
|
||||
assert res.status_code == 403, f"[{user['name']}] expected 403, got {res.status_code}"
|
||||
|
||||
def test_post_overhaul_gantt_spreadsheet():
|
||||
"""POST /overhaul-gantt/spreadsheet — no role guard; all roles allowed
|
||||
Payload: spreadsheet_link (str). CSRF token required in production.
|
||||
"""
|
||||
payload = {
|
||||
"spreadsheet_link": SPREADSHEET_LINK,
|
||||
}
|
||||
for user in USER_LIST:
|
||||
csrf_token = get_csrf_token("/overhaul-gantt/spreadsheet", user["authorization_token"])
|
||||
headers = {**user_auth_headers(user["authorization_token"]), "X-CSRF-Token": csrf_token}
|
||||
res = requests.post(
|
||||
f"{BASE_URL}/overhaul-gantt/spreadsheet",
|
||||
json=payload,
|
||||
headers=headers,
|
||||
)
|
||||
assert res.status_code in (200, 201), f"[{user['name']}] expected 200/201, got {res.status_code}"
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Scope Equipments
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_scope_equipments():
|
||||
"""GET /scope-equipments?page=1 — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/scope-equipments?page=1", headers=user_auth_headers(user["authorization_token"]))
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
|
||||
def test_get_scope_equipments_available():
|
||||
"""GET /scope-equipments/available/{scope_name} — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(
|
||||
f"{BASE_URL}/scope-equipments/available/{SCOPE_NAME}",
|
||||
headers=user_auth_headers(user["authorization_token"]),
|
||||
)
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}"
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Calculation — Time Constraint
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_calculation_time_constraint_history():
|
||||
"""GET /calculation/time-constraint — required_permission(READ) only; Management has READ"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(
|
||||
f"{BASE_URL}/calculation/time-constraint",
|
||||
headers=user_auth_headers(user["authorization_token"]),
|
||||
)
|
||||
assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhaul Schedules (additional)
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_overhaul_schedules():
|
||||
"""GET /overhaul-schedules — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/overhaul-schedules", headers=user_auth_headers(user["authorization_token"]))
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
def test_get_overhaul_schedules_history():
|
||||
"""GET /overhaul-schedules/history — require_any_role blocks Management"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/overhaul-schedules/history", headers=user_auth_headers(user["authorization_token"]))
|
||||
expected = 200 if user["name"] in NON_MANAGEMENT else 403
|
||||
assert res.status_code == expected, f"[{user['name']}] expected {expected}, got {res.status_code}: {res.text}"
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Overhauls sub-routes
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_overhauls_system_components():
|
||||
"""GET /overhauls/system-components — required_permission(READ); Management has READ"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/overhauls/system-components", headers=user_auth_headers(user["authorization_token"]))
|
||||
assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Scope Equipments
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_scope_equipments():
|
||||
"""GET /scope-equipments — required_permission(READ) only; Management has READ in ACL"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/scope-equipments", headers=user_auth_headers(user["authorization_token"]))
|
||||
assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Workscopes
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def test_get_workscopes():
|
||||
"""GET /workscopes — no role guard; all roles allowed"""
|
||||
for user in USER_LIST:
|
||||
res = requests.get(f"{BASE_URL}/workscopes", headers=user_auth_headers(user["authorization_token"]))
|
||||
assert res.status_code == 200, f"[{user['name']}] expected 200, got {res.status_code}: {res.text}"
|
||||
Loading…
Reference in New Issue